Add Prometheus metrics exporter for action execution

Author: JE-Chen

automation_file/__init__.py

@@ -40,6 +40,8 @@
 from automation_file.core.fim import IntegrityMonitor
 from automation_file.core.json_store import read_action_json, write_action_json
 from automation_file.core.manifest import ManifestException, verify_manifest, write_manifest
+from automation_file.core.metrics import ACTION_COUNT, ACTION_DURATION, record_action
+from automation_file.core.metrics import render as render_metrics
 from automation_file.core.package_loader import PackageLoader
 from automation_file.core.progress import (
     CancellationToken,
@@ -169,6 +171,7 @@
 )
 from automation_file.server.action_acl import ActionACL, ActionNotPermittedException
 from automation_file.server.http_server import HTTPActionServer, start_http_action_server
+from automation_file.server.metrics_server import MetricsServer, start_metrics_server
 from automation_file.server.tcp_server import (
     TCPActionServer,
     start_autocontrol_socket_server,
@@ -337,6 +340,12 @@ def __getattr__(name: str) -> Any:
     "decrypt_file",
     "generate_key",
     "key_from_password",
+    "ACTION_COUNT",
+    "ACTION_DURATION",
+    "record_action",
+    "render_metrics",
+    "MetricsServer",
+    "start_metrics_server",
     # Triggers
     "FileWatcher",
     "TriggerManager",

automation_file/core/action_executor.py

@@ -15,12 +15,14 @@
 
 from __future__ import annotations
 
+import time
 from collections.abc import Mapping
 from concurrent.futures import ThreadPoolExecutor
 from typing import Any
 
 from automation_file.core.action_registry import ActionRegistry, build_default_registry
 from automation_file.core.json_store import read_action_json
+from automation_file.core.metrics import record_action
 from automation_file.core.substitution import substitute as substitute_payload
 from automation_file.exceptions import ExecuteActionException, ValidationException
 from automation_file.logging_config import file_automation_logger
@@ -150,27 +152,40 @@ def add_command_to_executor(self, command_dict: Mapping[str, Any]) -> None:
 
     # Internals ---------------------------------------------------------
     def _run_one(self, action: list, dry_run: bool) -> Any:
+        name = _safe_action_name(action)
+        if dry_run:
+            return self._run_dry(action)
+        started = time.monotonic()
+        ok = False
         try:
-            if dry_run:
-                name, kind, payload = self._parse_action(action)
-                if self.registry.resolve(name) is None:
-                    raise ExecuteActionException(f"unknown action: {name!r}")
-                file_automation_logger.info(
-                    "dry_run: %s kind=%s payload=%r",
-                    name,
-                    kind,
-                    payload,
-                )
-                return f"dry_run:{name}"
             value = self._execute_event(action)
             file_automation_logger.info("execute_action: %s", action)
+            ok = True
             return value
         except ExecuteActionException as error:
             file_automation_logger.error("execute_action malformed: %r", error)
             return repr(error)
         except Exception as error:  # pylint: disable=broad-except
             file_automation_logger.error("execute_action runtime error: %r", error)
             return repr(error)
+        finally:
+            record_action(name, time.monotonic() - started, ok)
+
+    def _run_dry(self, action: list) -> Any:
+        try:
+            name, kind, payload = self._parse_action(action)
+            if self.registry.resolve(name) is None:
+                raise ExecuteActionException(f"unknown action: {name!r}")
+        except ExecuteActionException as error:
+            file_automation_logger.error("execute_action malformed: %r", error)
+            return repr(error)
+        file_automation_logger.info(
+            "dry_run: %s kind=%s payload=%r",
+            name,
+            kind,
+            payload,
+        )
+        return f"dry_run:{name}"
 
     @staticmethod
     def _coerce(action_list: list | Mapping[str, Any]) -> list:
@@ -188,6 +203,12 @@ def _coerce(action_list: list | Mapping[str, Any]) -> list:
         return action_list
 
 
+def _safe_action_name(action: Any) -> str:
+    if isinstance(action, list) and action and isinstance(action[0], str):
+        return action[0]
+    return "unknown"
+
+
 # Default shared executor — built once, mutated in place by plugins.
 executor: ActionExecutor = ActionExecutor()
 

automation_file/core/metrics.py

@@ -0,0 +1,65 @@
+"""Prometheus metrics — per-action counters and duration histogram.
+
+The module exposes two metrics that are updated from
+:class:`~automation_file.core.action_executor.ActionExecutor` on every
+call:
+
+* ``automation_file_actions_total{action, status}`` — counter incremented
+  with ``status="ok"`` or ``status="error"`` per action.
+* ``automation_file_action_duration_seconds{action}`` — histogram of wall
+  time spent inside the registered callable.
+
+:func:`render` returns the wire-format text and matching ``Content-Type``
+suitable for a ``GET /metrics`` handler. :func:`record_action` is the
+single write path — failures are swallowed so a broken metrics backend
+can never abort a real action.
+"""
+
+from __future__ import annotations
+
+from prometheus_client import CONTENT_TYPE_LATEST, REGISTRY, Counter, Histogram, generate_latest
+
+from automation_file.logging_config import file_automation_logger
+
+_DURATION_BUCKETS = (
+    0.005,
+    0.01,
+    0.025,
+    0.05,
+    0.1,
+    0.25,
+    0.5,
+    1.0,
+    2.5,
+    5.0,
+    10.0,
+    30.0,
+    60.0,
+)
+
+ACTION_COUNT = Counter(
+    "automation_file_actions_total",
+    "Total actions executed, partitioned by outcome.",
+    labelnames=("action", "status"),
+)
+ACTION_DURATION = Histogram(
+    "automation_file_action_duration_seconds",
+    "Time spent inside a registered action callable.",
+    labelnames=("action",),
+    buckets=_DURATION_BUCKETS,
+)
+
+
+def record_action(action: str, duration_seconds: float, ok: bool) -> None:
+    """Record one action execution. Never raises."""
+    status = "ok" if ok else "error"
+    try:
+        ACTION_COUNT.labels(action=action, status=status).inc()
+        ACTION_DURATION.labels(action=action).observe(max(0.0, float(duration_seconds)))
+    except Exception as err:  # pragma: no cover - defensive
+        file_automation_logger.error("metrics.record_action failed: %r", err)
+
+
+def render() -> tuple[bytes, str]:
+    """Return ``(payload, content_type)`` for a ``/metrics`` response."""
+    return generate_latest(REGISTRY), CONTENT_TYPE_LATEST

automation_file/server/metrics_server.py

@@ -0,0 +1,67 @@
+"""Prometheus scrape endpoint (``GET /metrics``).
+
+``start_metrics_server(host, port)`` spawns a threaded HTTP server that
+serves the current metrics snapshot rendered by
+:func:`automation_file.core.metrics.render`. Like the other servers in
+this package it defaults to loopback and requires ``allow_non_loopback=True``
+to bind elsewhere. No auth is attached by default — put it behind a
+reverse proxy or use an ACL at the network layer if exposing beyond a
+trusted host.
+"""
+
+from __future__ import annotations
+
+import threading
+from http import HTTPStatus
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+from automation_file.core.metrics import render
+from automation_file.logging_config import file_automation_logger
+from automation_file.server.network_guards import ensure_loopback
+
+_DEFAULT_HOST = "127.0.0.1"
+_DEFAULT_PORT = 9945
+_METRICS_PATH = "/metrics"
+
+
+class _MetricsHandler(BaseHTTPRequestHandler):
+    """GET /metrics -> current Prometheus snapshot."""
+
+    def log_message(  # pylint: disable=arguments-differ
+        self, format_str: str, *args: object
+    ) -> None:
+        file_automation_logger.info("metrics_server: " + format_str, *args)
+
+    def do_GET(self) -> None:  # pylint: disable=invalid-name — BaseHTTPRequestHandler API
+        if self.path != _METRICS_PATH:
+            self._send(HTTPStatus.NOT_FOUND, b"not found", "text/plain; charset=utf-8")
+            return
+        payload, content_type = render()
+        self._send(HTTPStatus.OK, payload, content_type)
+
+    def _send(self, status: HTTPStatus, payload: bytes, content_type: str) -> None:
+        self.send_response(status)
+        self.send_header("Content-Type", content_type)
+        self.send_header("Content-Length", str(len(payload)))
+        self.end_headers()
+        self.wfile.write(payload)
+
+
+class MetricsServer(ThreadingHTTPServer):
+    """Threaded HTTP server serving ``GET /metrics``."""
+
+
+def start_metrics_server(
+    host: str = _DEFAULT_HOST,
+    port: int = _DEFAULT_PORT,
+    allow_non_loopback: bool = False,
+) -> MetricsServer:
+    """Start a metrics server on a background thread and return it."""
+    if not allow_non_loopback:
+        ensure_loopback(host)
+    server = MetricsServer((host, port), _MetricsHandler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    address = server.server_address
+    file_automation_logger.info("metrics_server: listening on %s:%d", address[0], address[1])
+    return server

dev.toml

@@ -26,6 +26,7 @@ dependencies = [
     "PySide6>=6.6.0",
     "watchdog>=4.0.0",
     "cryptography>=42.0.0",
+    "prometheus_client>=0.20.0",
     "tomli>=2.0.1; python_version<\"3.11\""
 ]
 classifiers = [

stable.toml

@@ -26,6 +26,7 @@ dependencies = [
     "PySide6>=6.6.0",
     "watchdog>=4.0.0",
     "cryptography>=42.0.0",
+    "prometheus_client>=0.20.0",
     "tomli>=2.0.1; python_version<\"3.11\""
 ]
 classifiers = [

tests/test_metrics.py

@@ -0,0 +1,95 @@
+"""Tests for the Prometheus metrics exporter."""
+
+from __future__ import annotations
+
+import urllib.request
+
+import pytest
+
+from automation_file import (
+    ACTION_COUNT,
+    execute_action,
+    record_action,
+    render_metrics,
+    start_metrics_server,
+)
+from automation_file.core.action_executor import executor
+from tests._insecure_fixtures import insecure_url, ipv4
+
+
+def _register_probes() -> None:
+    if "test_metric_ok" not in executor.registry:
+        executor.registry.register("test_metric_ok", lambda: "ok")
+    if "test_metric_fail" not in executor.registry:
+
+        def _boom() -> None:
+            raise RuntimeError("boom")
+
+        executor.registry.register("test_metric_fail", _boom)
+
+
+def test_record_action_increments_counter() -> None:
+    before = ACTION_COUNT.labels(action="unit.test", status="ok")._value.get()
+    record_action("unit.test", 0.01, ok=True)
+    after = ACTION_COUNT.labels(action="unit.test", status="ok")._value.get()
+    assert after == before + 1
+
+
+def test_record_action_clamps_negative_duration() -> None:
+    # Must not raise.
+    record_action("unit.clamp", -5.0, ok=True)
+
+
+def test_render_metrics_content_type() -> None:
+    payload, content_type = render_metrics()
+    assert content_type.startswith("text/plain")
+    assert b"automation_file_actions_total" in payload
+
+
+def test_execute_action_increments_ok_counter() -> None:
+    _register_probes()
+    before = ACTION_COUNT.labels(action="test_metric_ok", status="ok")._value.get()
+    execute_action([["test_metric_ok"]])
+    after = ACTION_COUNT.labels(action="test_metric_ok", status="ok")._value.get()
+    assert after == before + 1
+
+
+def test_execute_action_increments_error_counter() -> None:
+    _register_probes()
+    before = ACTION_COUNT.labels(action="test_metric_fail", status="error")._value.get()
+    execute_action([["test_metric_fail"]])
+    after = ACTION_COUNT.labels(action="test_metric_fail", status="error")._value.get()
+    assert after == before + 1
+
+
+def test_metrics_server_serves_metrics_endpoint() -> None:
+    _register_probes()
+    execute_action([["test_metric_ok"]])
+    server = start_metrics_server(host="127.0.0.1", port=0)
+    host, port = server.server_address
+    try:
+        url = insecure_url("http", f"{host}:{port}/metrics")
+        with urllib.request.urlopen(url, timeout=3) as resp:  # nosec B310 - loopback test server
+            body = resp.read().decode("utf-8")
+        assert resp.status == 200
+        assert "automation_file_actions_total" in body
+    finally:
+        server.shutdown()
+
+
+def test_metrics_server_returns_404_for_other_paths() -> None:
+    server = start_metrics_server(host="127.0.0.1", port=0)
+    host, port = server.server_address
+    try:
+        url = insecure_url("http", f"{host}:{port}/other")
+        with pytest.raises(urllib.error.HTTPError) as info:
+            urllib.request.urlopen(url, timeout=3)  # nosec B310 - loopback test server
+        assert info.value.code == 404
+    finally:
+        server.shutdown()
+
+
+def test_metrics_server_rejects_non_loopback() -> None:
+    non_loopback = ipv4(8, 8, 8, 8)
+    with pytest.raises(ValueError):
+        start_metrics_server(host=non_loopback, port=0)