marko/ansible-role-aide
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Write roles fingerprints to /var/log/sysroles.jsonl [citest_skip]

Browse Source 
Feature: Write roles fingerprints to /var/log/sysroles.jsonl [citest_skip]

Reason: By default logs are printed to rsyslog. This change adds a possibility to write logs to a file on the system for the downstream users.

Result: For the upstream, this makes rsyslog log message more detailed. For the downstream - also writes logs to /var/log/sysroles.jsonl
Signed-off-by: Sergei Petrosian <spetrosi@redhat.com>
This commit is contained in:
Sergei Petrosian
2026-08-06 16:11:09 -06:00
committed by Richard Megginson
parent c04dc49ded
commit d9ba105ef3
2 changed files with 717 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
library/sr_fingerprint.py
+303 -25
View File
@@ -7,29 +7,148 @@ __metaclass__ = type
DOCUMENTATION = """
---
module: sr_fingerprint
short_description: Write a message string to syslog using Ansible C(module.log) function.
short_description: Write role fingerprint data to syslog and optionally to a JSONL log file
description:
- Writes the given string to the system log using Ansible C(module.log) function.
- Collects role fingerprint data into a canonical record and writes it to
syslog using Ansible C(module.log) as C(key=value) pairs.
- Optionally appends the same record as a JSON line to a log file
(one JSON object per line, JSONL format), by default
C(/var/log/sysroles.jsonl).
- Playbook variables are not available inside modules automatically. Roles
pass C(role_name), C(role_path), C(ansible_play_hosts_all),
C(distribution), and C(distribution_version) from the task.
- C(ansible_check_mode) is collected from the module execution context.
- Intended for role-internal or diagnostic use.
author: Rich Megginson (@richm)
options:
sr_message:
description: Text to record in syslog.
status:
description: Role execution status.
type: str
required: true
choices:
- begin
- success
write_log_file:
description: >-
If C(true), append fingerprint data to the JSONL log file.
Defaults to C(false).
type: bool
default: false
log_file:
description: >-
Path to the JSONL log file. A lock sidecar (C(<log_file>.lock))
is created next to the log file for cross-process safety.
type: path
default: /var/log/sysroles.jsonl
max_log_size:
description: >-
Maximum log file size in bytes. When appending a new record
would exceed this limit, the oldest records are removed first.
Set to C(0) to disable trimming.
type: int
default: 2000000
role_name:
description: Name of the role, typically C({{ role_name }}).
type: str
required: true
role_path:
description: Path to the role, typically C({{ role_path }}).
type: path
required: true
ansible_play_hosts_all:
description: >-
All hosts in the play, typically C({{ ansible_play_hosts_all }}).
Used to derive C(play_hosts_number).
type: list
elements: str
required: true
distribution:
description: >-
OS distribution name, typically
C({{ ansible_facts["distribution"] }}).
type: str
default: ""
distribution_version:
description: >-
OS distribution version, typically
C({{ ansible_facts["distribution_version"] }}).
type: str
default: ""
"""
EXAMPLES = """
- name: Record a fingerprint message in syslog
- name: Record role begin fingerprint to syslog only (not log file)
sr_fingerprint:
sr_message: "system_role:ROLENAME"
status: begin
role_name: bootloader
role_path: "{{ role_path }}"
ansible_play_hosts_all: "{{ ansible_play_hosts_all }}"
distribution: "{{ ansible_facts['distribution'] }}"
distribution_version: "{{ ansible_facts['distribution_version'] }}"
write_log_file: false
- name: Record role success fingerprint
sr_fingerprint:
status: success
role_name: bootloader
role_path: "{{ role_path }}"
ansible_play_hosts_all: "{{ ansible_play_hosts_all }}"
distribution: "{{ ansible_facts['distribution'] }}"
distribution_version: "{{ ansible_facts['distribution_version'] }}"
write_log_file: true
"""
RETURN = r""" # """
RETURN = r"""
fingerprint:
description: The fingerprint record written to syslog and optionally to the log file.
returned: always
type: dict
sample:
date: "2026-08-03T10:15:00+02:00"
role_name: network
role_path: /usr/share/ansible/roles/linux-system-roles.network
status: success
ansible_version: "2.16.3"
managed_node_distro: RedHat-9.4
play_hosts_number: 3
ansible_check_mode: false
message:
description: Informational message shown in check mode.
returned: check mode
type: str
sample: "Check mode: message not logged - [date=... role_name=...]"
jsonl_row:
description: The JSON line that would be appended to the log file.
returned: check mode and O(write_log_file=true)
type: str
log_file:
description: Path to the log file that would be written.
returned: check mode and O(write_log_file=true)
type: str
"""
from ansible.module_utils.basic import AnsibleModule
import datetime
import errno
import fcntl
import json
import os
import stat
import tempfile
FINGERPRINT_FIELDS = (
"date",
"role_name",
"role_path",
"status",
"ansible_version",
"managed_node_distro",
"play_hosts_number",
"ansible_check_mode",
)
FINGERPRINT_SYSLOG_SEPARATOR = " "
def _local_iso8601_no_microseconds():
@@ -51,9 +170,184 @@ def _local_iso8601_no_microseconds():
return datetime.datetime.now(utc).astimezone().replace(microsecond=0).isoformat()
def _ensure_parent_dir(path):
parent = os.path.dirname(path)
if not parent:
return
if os.path.isdir(parent):
return
try:
os.makedirs(parent)
except OSError as exc:
# another process may have created the directory
if exc.errno != errno.EEXIST or not os.path.isdir(parent):
raise
def _format_fingerprint_jsonl(record):
"""Format the canonical fingerprint record as a single JSON line."""
return json.dumps(record, separators=(",", ":"), sort_keys=False)
def _trim_log_file(log_file, size_needed):
"""Remove oldest records until the file can accommodate size_needed bytes."""
with open(log_file, "r") as log_fd:
lines = log_fd.readlines()
size_removed = 0
while lines and size_removed < size_needed:
size_removed += len(lines.pop(0))
orig_stat = os.stat(log_file)
dir_name = os.path.dirname(log_file) or "."
fd, tmp_path = tempfile.mkstemp(dir=dir_name, suffix=".tmp")
try:
os.fchmod(fd, stat.S_IMODE(orig_stat.st_mode))
try:
os.fchown(fd, orig_stat.st_uid, orig_stat.st_gid)
except OSError:
# not running as root; keep default ownership
pass
with os.fdopen(fd, "w") as tmp_fd:
tmp_fd.writelines(lines)
tmp_fd.flush()
os.fsync(tmp_fd.fileno())
os.rename(tmp_path, log_file)
except BaseException:
try:
os.unlink(tmp_path)
except OSError:
# already removed or never created
pass
raise
def _write_jsonl_log(log_file, record, max_size=0):
_ensure_parent_dir(log_file)
new_line = _format_fingerprint_jsonl(record) + "\n"
lock_path = log_file + ".lock"
lock_fd = open(lock_path, "w")
try:
fcntl.flock(lock_fd, fcntl.LOCK_EX)
try:
cur_size = os.path.getsize(log_file)
except OSError:
# file does not exist yet
cur_size = 0
if max_size > 0 and cur_size + len(new_line) > max_size and cur_size > 0:
_trim_log_file(log_file, len(new_line))
with open(log_file, "a") as log_fd:
log_fd.write(new_line)
finally:
fcntl.flock(lock_fd, fcntl.LOCK_UN)
lock_fd.close()
def _get_managed_node_distro(distribution, distribution_version):
if distribution and distribution_version:
return "%s-%s" % (distribution, distribution_version)
return "unknown"
def _get_play_hosts_number(play_hosts_all):
return len(play_hosts_all)
def _get_ansible_version(module):
version = getattr(module, "ansible_version", None)
if version:
return version
return "unknown"
def _get_check_mode(module):
return bool(getattr(module, "check_mode", False))
def _collect_fingerprint_record(module, status):
"""Build the canonical fingerprint record used by all output formatters."""
return {
"date": _local_iso8601_no_microseconds(),
"role_name": module.params["role_name"],
"role_path": module.params["role_path"],
"status": status,
"ansible_version": _get_ansible_version(module),
"managed_node_distro": _get_managed_node_distro(
module.params["distribution"], module.params["distribution_version"]
),
"play_hosts_number": _get_play_hosts_number(
module.params["ansible_play_hosts_all"]
),
"ansible_check_mode": _get_check_mode(module),
}
def _fingerprint_record_items(record):
return [(field, record[field]) for field in FINGERPRINT_FIELDS]
def _format_fingerprint_key_value(field, value):
text = "" if value is None else str(value)
if any(char in text for char in ' "='):
return '%s="%s"' % (field, text.replace('"', '""'))
return "%s=%s" % (field, text)
def _format_fingerprint_syslog(record):
"""Format the canonical fingerprint record as key=value syslog text."""
pairs = [
_format_fingerprint_key_value(field, value)
for field, value in _fingerprint_record_items(record)
]
return FINGERPRINT_SYSLOG_SEPARATOR.join(pairs)
def _handle_fingerprint(module):
max_log_size = module.params["max_log_size"]
if max_log_size < 0:
module.fail_json(
msg="max_log_size must be 0 or a positive integer, got %d" % max_log_size
)
fingerprint_record = _collect_fingerprint_record(module, module.params["status"])
log_message = _format_fingerprint_syslog(fingerprint_record)
if module.check_mode:
result = dict(
changed=False,
message="Check mode: message not logged - [%s]" % log_message,
fingerprint=fingerprint_record,
)
if module.params["write_log_file"]:
result["jsonl_row"] = _format_fingerprint_jsonl(fingerprint_record)
result["log_file"] = module.params["log_file"]
module.exit_json(**result)
module.log(log_message)
if module.params["write_log_file"]:
log_file = module.params["log_file"]
try:
_write_jsonl_log(
log_file, fingerprint_record, module.params["max_log_size"]
)
except (IOError, OSError) as exc:
module.fail_json(
msg="Failed to write fingerprint log file %s: %s" % (log_file, exc)
)
module.exit_json(changed=False, fingerprint=fingerprint_record)
def run_module():
module_args = dict(
sr_message=dict(type="str", required=True),
status=dict(type="str", required=True, choices=["begin", "success"]),
write_log_file=dict(type="bool", default=False),
log_file=dict(type="path", default="/var/log/sysroles.jsonl"),
max_log_size=dict(type="int", default=2000000),
role_name=dict(type="str", required=True),
role_path=dict(type="path", required=True),
ansible_play_hosts_all=dict(type="list", elements="str", required=True),
distribution=dict(type="str", default=""),
distribution_version=dict(type="str", default=""),
)
module = AnsibleModule(
@@ -61,23 +355,7 @@ def run_module():
supports_check_mode=True,
)
log_message = "%s %s" % (
module.params["sr_message"],
_local_iso8601_no_microseconds(),
)
if module.check_mode:
module.exit_json(
changed=False,
message="Check mode: message not logged - [%s]" % log_message,
)
module.log(log_message)
# we don't actually change anything, so we're not changed - writing a log message
# is not considered a change
# also, we don't want to report changed every time the role runs
module.exit_json(changed=False)
_handle_fingerprint(module)
def main():
tests/unit/test_sr_fingerprint.py
+414
View File
@@ -0,0 +1,414 @@
# -*- coding: utf-8 -*-
# Copyright: (c) 2026, Red Hat, Inc.
# SPDX-License-Identifier: MIT
"""Unit tests for sr_fingerprint module helpers."""
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import json
import os
import re
import tempfile
import unittest
import sr_fingerprint
class _ExitJsonException(Exception):
def __init__(self, kwargs):
self.kwargs = kwargs
class _FailJsonException(Exception):
def __init__(self, kwargs):
self.kwargs = kwargs
class _FakeModule(object):
ansible_version = "2.16.3"
def __init__(self, params=None, check_mode=False):
self.params = params or {}
self.check_mode = check_mode
self.logged = []
def log(self, msg):
self.logged.append(msg)
def exit_json(self, **kwargs):
raise _ExitJsonException(kwargs)
def fail_json(self, **kwargs):
raise _FailJsonException(kwargs)
def _cleanup_log(log_file):
for path in (log_file, log_file + ".lock"):
try:
os.unlink(path)
except OSError:
# file may not exist
pass
def _sample_fingerprint_record():
return {
"date": "2026-06-10T12:00:00+00:00",
"role_name": "systemd",
"role_path": "/usr/share/ansible/roles/linux-system-roles.systemd",
"status": "begin",
"ansible_version": "2.16.3",
"managed_node_distro": "RedHat-9.4",
"play_hosts_number": 3,
"ansible_check_mode": False,
}
class TestSrFingerprint(unittest.TestCase):
def test_fingerprint_fields_match_record_keys(self):
record = _sample_fingerprint_record()
self.assertEqual(set(sr_fingerprint.FINGERPRINT_FIELDS), set(record.keys()))
def test_format_fingerprint_syslog(self):
record = _sample_fingerprint_record()
message = sr_fingerprint._format_fingerprint_syslog(record)
self.assertEqual(
message,
"date=2026-06-10T12:00:00+00:00 role_name=systemd "
"role_path=/usr/share/ansible/roles/linux-system-roles.systemd status=begin "
"ansible_version=2.16.3 managed_node_distro=RedHat-9.4 "
"play_hosts_number=3 ansible_check_mode=False",
)
for field in sr_fingerprint.FINGERPRINT_FIELDS:
self.assertIn("%s=" % field, message)
def test_format_fingerprint_jsonl(self):
record = _sample_fingerprint_record()
line = sr_fingerprint._format_fingerprint_jsonl(record)
parsed = json.loads(line)
self.assertEqual(parsed, record)
def test_collect_fingerprint_record_from_passed_inputs(self):
module = _FakeModule(
{
"role_name": "systemd",
"role_path": "/usr/share/ansible/roles/linux-system-roles.systemd",
"ansible_play_hosts_all": ["host1", "host2", "host3"],
"distribution": "RedHat",
"distribution_version": "9.4",
},
check_mode=True,
)
record = sr_fingerprint._collect_fingerprint_record(module, "begin")
self.assertEqual(record["role_name"], "systemd")
self.assertEqual(
record["role_path"], "/usr/share/ansible/roles/linux-system-roles.systemd"
)
self.assertEqual(record["managed_node_distro"], "RedHat-9.4")
self.assertEqual(record["play_hosts_number"], 3)
self.assertTrue(record["ansible_check_mode"])
self.assertEqual(
set(record.keys()),
set(sr_fingerprint.FINGERPRINT_FIELDS),
)
def test_get_managed_node_distro_from_params(self):
distro = sr_fingerprint._get_managed_node_distro("Fedora", "42")
self.assertEqual(distro, "Fedora-42")
def test_get_managed_node_distro_missing(self):
self.assertEqual(sr_fingerprint._get_managed_node_distro("", ""), "unknown")
def test_get_play_hosts_number(self):
self.assertEqual(
sr_fingerprint._get_play_hosts_number(["a", "b"]),
2,
)
self.assertEqual(sr_fingerprint._get_play_hosts_number([]), 0)
def test_format_fingerprint_syslog_quotes_values_with_spaces(self):
record = _sample_fingerprint_record()
record["role_path"] = (
"/usr/share/ansible/roles/linux-system-roles.systemd extra"
)
message = sr_fingerprint._format_fingerprint_syslog(record)
self.assertIn(
'role_path="/usr/share/ansible/roles/linux-system-roles.systemd extra"',
message,
)
def test_write_jsonl_log_appends_valid_json_lines(self):
with tempfile.NamedTemporaryFile(delete=False, suffix=".jsonl") as tmp:
log_file = tmp.name
try:
record = _sample_fingerprint_record()
sr_fingerprint._write_jsonl_log(log_file, record)
sr_fingerprint._write_jsonl_log(log_file, record)
with open(log_file, "r") as log_fd:
lines = log_fd.read().splitlines()
self.assertEqual(len(lines), 2)
for line in lines:
parsed = json.loads(line)
self.assertEqual(parsed, record)
finally:
_cleanup_log(log_file)
def test_write_jsonl_log_creates_parent_dir(self):
tmpdir = tempfile.mkdtemp()
log_file = os.path.join(tmpdir, "subdir", "fingerprint.jsonl")
try:
record = _sample_fingerprint_record()
sr_fingerprint._write_jsonl_log(log_file, record)
with open(log_file, "r") as log_fd:
parsed = json.loads(log_fd.readline())
self.assertEqual(parsed["role_name"], "systemd")
finally:
subdir = os.path.dirname(log_file)
for name in os.listdir(subdir):
os.unlink(os.path.join(subdir, name))
os.rmdir(subdir)
os.rmdir(tmpdir)
def test_write_jsonl_log_preserves_types(self):
with tempfile.NamedTemporaryFile(delete=False, suffix=".jsonl") as tmp:
log_file = tmp.name
try:
record = _sample_fingerprint_record()
sr_fingerprint._write_jsonl_log(log_file, record)
with open(log_file, "r") as log_fd:
parsed = json.loads(log_fd.readline())
self.assertIsInstance(parsed["play_hosts_number"], int)
self.assertIsInstance(parsed["ansible_check_mode"], bool)
finally:
_cleanup_log(log_file)
def test_trim_removes_oldest_lines(self):
with tempfile.NamedTemporaryFile(delete=False, suffix=".jsonl") as tmp:
log_file = tmp.name
try:
record = _sample_fingerprint_record()
sample = dict(record, role_name="role_0")
line_size = len(sr_fingerprint._format_fingerprint_jsonl(sample) + "\n")
max_size = line_size * 5
for _i in range(10):
record_copy = dict(record, role_name="role_%d" % _i)
sr_fingerprint._write_jsonl_log(
log_file, record_copy, max_size=max_size
)
with open(log_file, "r") as log_fd:
lines = log_fd.read().splitlines()
self.assertEqual(len(lines), 5)
first = json.loads(lines[0])
last = json.loads(lines[-1])
self.assertEqual(first["role_name"], "role_5")
self.assertEqual(last["role_name"], "role_9")
finally:
_cleanup_log(log_file)
def test_trim_multiple_lines(self):
with tempfile.NamedTemporaryFile(delete=False, suffix=".jsonl") as tmp:
log_file = tmp.name
try:
record = _sample_fingerprint_record()
sample = dict(record, role_name="role_0")
line_size = len(sr_fingerprint._format_fingerprint_jsonl(sample) + "\n")
# Log contains more than 3 lines.
n_initial = 4
max_size = line_size * n_initial
for _i in range(n_initial):
sr_fingerprint._write_jsonl_log(
log_file,
dict(record, role_name="role_%d" % _i),
max_size=max_size,
)
with open(log_file, "r") as log_fd:
initial_lines = log_fd.read().splitlines()
self.assertEqual(len(initial_lines), n_initial)
# New record larger than one existing line (up to two) via a very
# long role_path, so trim removes exactly two oldest records.
base_record = dict(record, role_name="role_long", role_path="")
base_size = len(
sr_fingerprint._format_fingerprint_jsonl(base_record) + "\n"
)
path_len = 2 * line_size - base_size
self.assertGreater(path_len, 0)
long_path = "x" * path_len
long_record = dict(record, role_name="role_long", role_path=long_path)
new_line = sr_fingerprint._format_fingerprint_jsonl(long_record) + "\n"
self.assertGreater(len(new_line), line_size)
self.assertLessEqual(len(new_line), 2 * line_size)
sr_fingerprint._write_jsonl_log(log_file, long_record, max_size=max_size)
with open(log_file, "r") as log_fd:
lines = log_fd.read().splitlines()
# Exactly two oldest records removed; new record appended.
self.assertEqual(len(lines), n_initial - 2 + 1)
parsed = [json.loads(line) for line in lines]
role_names = [entry["role_name"] for entry in parsed]
self.assertEqual(role_names, ["role_2", "role_3", "role_long"])
self.assertEqual(parsed[-1], long_record)
self.assertEqual(parsed[-1]["role_path"], long_path)
self.assertNotIn("role_0", role_names)
self.assertNotIn("role_1", role_names)
finally:
_cleanup_log(log_file)
def test_trim_disabled_when_zero(self):
with tempfile.NamedTemporaryFile(delete=False, suffix=".jsonl") as tmp:
log_file = tmp.name
try:
record = _sample_fingerprint_record()
for _i in range(20):
sr_fingerprint._write_jsonl_log(log_file, record, max_size=0)
with open(log_file, "r") as log_fd:
lines = log_fd.read().splitlines()
self.assertEqual(len(lines), 20)
finally:
_cleanup_log(log_file)
def test_trim_no_op_when_under_limit(self):
with tempfile.NamedTemporaryFile(delete=False, suffix=".jsonl") as tmp:
log_file = tmp.name
try:
record = _sample_fingerprint_record()
for _i in range(3):
sr_fingerprint._write_jsonl_log(log_file, record, max_size=2000000)
with open(log_file, "r") as log_fd:
lines = log_fd.read().splitlines()
self.assertEqual(len(lines), 3)
finally:
_cleanup_log(log_file)
def test_handle_fingerprint_check_mode_without_log_file(self):
module = _FakeModule(
{
"status": "begin",
"write_log_file": False,
"max_log_size": 2000000,
"role_name": "systemd",
"role_path": "/usr/share/ansible/roles/linux-system-roles.systemd",
"ansible_play_hosts_all": ["host1"],
"distribution": "RedHat",
"distribution_version": "9.4",
},
check_mode=True,
)
with self.assertRaises(_ExitJsonException) as ctx:
sr_fingerprint._handle_fingerprint(module)
result = ctx.exception.kwargs
self.assertFalse(result["changed"])
self.assertIn("Check mode", result["message"])
self.assertIn("fingerprint", result)
self.assertNotIn("jsonl_row", result)
def test_handle_fingerprint_check_mode_with_log_file(self):
log_path = os.path.join(tempfile.gettempdir(), "test_sr_fingerprint.jsonl")
module = _FakeModule(
{
"status": "success",
"write_log_file": True,
"log_file": log_path,
"max_log_size": 2000000,
"role_name": "systemd",
"role_path": "/usr/share/ansible/roles/linux-system-roles.systemd",
"ansible_play_hosts_all": ["host1"],
"distribution": "RedHat",
"distribution_version": "9.4",
},
check_mode=True,
)
with self.assertRaises(_ExitJsonException) as ctx:
sr_fingerprint._handle_fingerprint(module)
result = ctx.exception.kwargs
self.assertIn("jsonl_row", result)
self.assertEqual(result["log_file"], log_path)
parsed = json.loads(result["jsonl_row"])
self.assertEqual(parsed["role_name"], "systemd")
def test_handle_fingerprint_write_failure_calls_fail_json(self):
log_path = os.path.join(tempfile.gettempdir(), "test_write_fail.jsonl")
module = _FakeModule(
{
"status": "success",
"write_log_file": True,
"log_file": log_path,
"max_log_size": 2000000,
"role_name": "systemd",
"role_path": "/usr/share/ansible/roles/linux-system-roles.systemd",
"ansible_play_hosts_all": ["host1"],
"distribution": "RedHat",
"distribution_version": "9.4",
},
check_mode=False,
)
original = sr_fingerprint._write_jsonl_log
def _raise_ioerror(*args, **kwargs):
raise IOError("disk full")
sr_fingerprint._write_jsonl_log = _raise_ioerror
try:
with self.assertRaises(_FailJsonException) as ctx:
sr_fingerprint._handle_fingerprint(module)
self.assertIn(
"Failed to write fingerprint log file", ctx.exception.kwargs["msg"]
)
finally:
sr_fingerprint._write_jsonl_log = original
def test_handle_fingerprint_rejects_negative_max_log_size(self):
module = _FakeModule(
{
"status": "begin",
"write_log_file": False,
"max_log_size": -1,
"role_name": "systemd",
"role_path": "/usr/share/ansible/roles/linux-system-roles.systemd",
"ansible_play_hosts_all": ["host1"],
"distribution": "RedHat",
"distribution_version": "9.4",
},
check_mode=False,
)
with self.assertRaises(_FailJsonException) as ctx:
sr_fingerprint._handle_fingerprint(module)
self.assertIn(
"max_log_size must be 0 or a positive integer",
ctx.exception.kwargs["msg"],
)
def test_local_iso8601_no_microseconds_has_no_fraction(self):
timestamp = sr_fingerprint._local_iso8601_no_microseconds()
match = re.match(
r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}[+-]\d{2}:?\d{2}$", timestamp
)
self.assertIsNotNone(match)
if __name__ == "__main__":
unittest.main()