logger + callback plugins

ansible.cfg

@@ -76,6 +76,8 @@ host_key_checking = False
 # change the default callback, you can only have one 'stdout' type  enabled at a time.
 #stdout_callback = skippy
 stdout_callback = minimal
+callback_plugins   = ./callback_plugins
+callbacks_enabled  = json_logger
 
 
 ## Ansible ships with some plugins that require whitelisting,

ansible_runs.jsonl

@@ -0,0 +1,2 @@
+{"ts": "2026-04-10T21:09:25", "playbook": "test_logger.yml", "user": "seba", "duration": 1.8, "status": "success", "hosts": {"localhost": {"ok": 2, "changed": 0, "failed": 0, "unreachable": 0, "skipped": 0}}, "tasks": ["Gathering Facts", "test task"]}
+{"ts": "2026-04-10T21:21:32", "playbook": "crons-only.yml", "user": "seba", "duration": 4.6, "status": "success", "hosts": {"acemagic": {"ok": 5, "changed": 0, "failed": 0, "unreachable": 0, "skipped": 0}}, "tasks": ["Gathering Facts", "../roles/crons.acemagic : Restart Frigate Docker container", "../roles/crons.acemagic : Backup Bitwarden database", "../roles/crons.acemagic : Cron: Keep-alive hack for stubborn HDDs (Seagate 5T & 6T)", "../roles/crons.acemagic : Configure hdparm to prevent spindown for supported drives (SDE & SDB)"]}

callback_plugins/json_logger.py

@@ -0,0 +1,75 @@
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+DOCUMENTATION = '''
+    name: json_logger
+    type: notification
+    short_description: Appends one JSON line per playbook run to ansible_runs.jsonl
+    description:
+      - Records playbook execution results as newline-delimited JSON.
+'''
+
+import json
+import os
+import time
+from datetime import datetime
+
+from ansible.plugins.callback import CallbackBase
+
+CALLBACK_TYPE = 'notification'
+CALLBACK_NAME = 'json_logger'
+CALLBACK_NEEDS_WHITELIST = False
+
+_LOG_PATH = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'ansible_runs.jsonl')
+
+
+class CallbackModule(CallbackBase):
+
+    def __init__(self):
+        super(CallbackModule, self).__init__()
+        self._playbook_name = None
+        self._start_time = None
+        self._tasks = []
+
+    def v2_playbook_on_start(self, playbook):
+        self._playbook_name = os.path.basename(playbook._file_name)
+        self._start_time = time.time()
+        self._tasks = []
+
+    def v2_playbook_on_task_start(self, task, is_conditional):
+        name = task.get_name()
+        if name and name not in self._tasks:
+            self._tasks.append(name)
+
+    def v2_playbook_on_stats(self, stats):
+        duration = round(time.time() - self._start_time, 1) if self._start_time else 0.0
+
+        hosts_data = {}
+        overall_failed = False
+        for host in sorted(stats.processed.keys()):
+            s = stats.summarize(host)
+            hosts_data[host] = {
+                'ok':          s['ok'],
+                'changed':     s['changed'],
+                'failed':      s['failures'],
+                'unreachable': s['unreachable'],
+                'skipped':     s['skipped'],
+            }
+            if s['failures'] > 0 or s['unreachable'] > 0:
+                overall_failed = True
+
+        record = {
+            'ts':       datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S'),
+            'playbook': self._playbook_name or 'unknown',
+            'user':     os.environ.get('USER', os.environ.get('LOGNAME', 'unknown')),
+            'duration': duration,
+            'status':   'failed' if overall_failed else 'success',
+            'hosts':    hosts_data,
+            'tasks':    self._tasks,
+        }
+
+        try:
+            with open(_LOG_PATH, 'a') as f:
+                f.write(json.dumps(record) + '\n')
+        except Exception as e:
+            self._display.warning(f'json_logger: could not write to {_LOG_PATH}: {e}')