cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat(observability): implement Prometheus metrics export and event-to-audit bridge #1308

Closed
freemo wants to merge 1 commits from feature/observability-prometheus-audit into master
pull from: feature/observability-prometheus-audit
merge into: cleveragents:master
Conversation 5 Commits 1 Files Changed 13
+1937 -5
13 changed files with 1937 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -172,3 +172,4 @@ src/cleveragents/acp/
# Git worktrees for parallel task branches
worktrees/
*.bak
ca-cow-backup-*
features/observability/audit_bridge.feature
+156
View File
@@ -0,0 +1,156 @@
Feature: Automatic Audit Bridge
As a security-conscious system
I want domain events with audit semantics to be automatically written to audit.jsonl
So that all security-relevant operations are tracked with structured records
# --- AuditBridge creation ---
Scenario: AuditBridge can be instantiated with an event bus
Given a ReactiveEventBus for audit bridge
When I create an AuditBridge with the event bus
Then the audit bridge should be created successfully
Scenario: AuditBridge subscribes to all audit event types
Given a ReactiveEventBus for audit bridge
When I create an AuditBridge with the event bus
Then the audit bridge should cover at least 10 event types
# --- Audit record schema ---
Scenario: AuditRecord has all required fields
When I build an audit record from a plan_applied event
Then the audit record should have a "timestamp" field
And the audit record should have a "event_type" field
And the audit record should have a "actor" field
And the audit record should have a "action" field
And the audit record should have a "resource" field
And the audit record should have a "outcome" field
And the audit record should have a "metadata" field
And the audit record should have a "correlation_id" field
And the audit record should have a "plan_id" field
And the audit record should have a "session_id" field
Scenario: AuditRecord timestamp is ISO-8601 UTC
When I build an audit record from a plan_applied event
Then the audit record timestamp should be a valid ISO-8601 string
Scenario: AuditRecord event_type is the dot-separated string
When I build an audit record from a plan_applied event
Then the audit record event_type should equal "plan.applied"
Scenario: AuditRecord outcome is success for plan_applied
When I build an audit record from a plan_applied event
Then the audit record outcome should equal "success"
Scenario: AuditRecord outcome is failure for plan_errored
When I build an audit record from a plan_errored event
Then the audit record outcome should equal "failure"
Scenario: AuditRecord outcome is failure for auth_failure
When I build an audit record from an auth_failure event
Then the audit record outcome should equal "failure"
Scenario: AuditRecord outcome is failure for invariant_violated
When I build an audit record from an invariant_violated event
Then the audit record outcome should equal "failure"
Scenario: AuditRecord actor is populated from event actor_name
When I build an audit record from a plan_applied event with actor "admin"
Then the audit record actor should equal "admin"
Scenario: AuditRecord plan_id is propagated from event
When I build an audit record from a plan_applied event with plan_id "PLAN-001"
Then the audit record plan_id should equal "PLAN-001"
Scenario: AuditRecord correlation_id is propagated from event
When I build an audit record from a plan_applied event with correlation_id "CORR-XYZ"
Then the audit record correlation_id should equal "CORR-XYZ"
# --- Event routing ---
Scenario: plan_applied event is written to audit log
Given an AuditBridge with in-memory capture
When a plan_applied event is emitted on the bus
Then the audit bridge should have 1 record
And the first record event_type should equal "plan.applied"
Scenario: plan_cancelled event is written to audit log
Given an AuditBridge with in-memory capture
When a plan_cancelled event is emitted on the bus
Then the audit bridge should have 1 record
Scenario: config_changed event is written to audit log
Given an AuditBridge with in-memory capture
When a config_changed event is emitted on the bus
Then the audit bridge should have 1 record
Scenario: auth_success event is written to audit log
Given an AuditBridge with in-memory capture
When an auth_success event is emitted on the bus
Then the audit bridge should have 1 record
Scenario: auth_failure event is written to audit log
Given an AuditBridge with in-memory capture
When an auth_failure event is emitted on the bus
Then the audit bridge should have 1 record
Scenario: plan_created event is NOT written to audit log
Given an AuditBridge with in-memory capture
When a plan_created event is emitted on the bus
Then the audit bridge should have 0 records
Scenario: actor_invoked event is NOT written to audit log
Given an AuditBridge with in-memory capture
When an actor_invoked event is emitted on the bus
Then the audit bridge should have 0 records
Scenario: Multiple audit events are all captured
Given an AuditBridge with in-memory capture
When a plan_applied event is emitted on the bus
And a config_changed event is emitted on the bus
And an auth_failure event is emitted on the bus
Then the audit bridge should have 3 records
# --- JSONL file output ---
Scenario: AuditBridge writes records to audit.jsonl file
Given an AuditBridge writing to a temp file
When a plan_applied event is emitted on the bus
Then the audit jsonl file should exist
And the audit jsonl file should contain 1 line
Scenario: Each line in audit.jsonl is valid JSON
Given an AuditBridge writing to a temp file
When a plan_applied event is emitted on the bus
And a config_changed event is emitted on the bus
Then each line in the audit jsonl file should be valid JSON
Scenario: audit.jsonl records contain all required fields
Given an AuditBridge writing to a temp file
When a plan_applied event is emitted on the bus
Then the first jsonl record should have field "timestamp"
And the first jsonl record should have field "event_type"
And the first jsonl record should have field "actor"
And the first jsonl record should have field "action"
And the first jsonl record should have field "outcome"
And the first jsonl record should have field "metadata"
And the first jsonl record should have field "correlation_id"
# --- Error resilience ---
Scenario: AuditBridge handles file write errors gracefully
Given an AuditBridge with an invalid file path
When a plan_applied event is emitted on the bus
Then no exception should propagate from the audit bridge write
# --- to_dict and to_json ---
Scenario: AuditRecord to_dict returns a plain dict
When I build an audit record from a plan_applied event
Then to_dict should return a dict
And the dict should have key "timestamp"
And the dict should have key "event_type"
Scenario: AuditRecord to_json returns a JSON string
When I build an audit record from a plan_applied event
Then to_json should return a valid JSON string
features/observability/prometheus_metrics.feature
+105
View File
@@ -0,0 +1,105 @@
Feature: Prometheus Metrics Registry
As a platform operator
I want Prometheus-compatible metrics export
So that I can monitor plan execution, tool calls, LLM usage, and context assembly
# --- Registry creation ---
Scenario: PrometheusRegistry can be instantiated
When I create a PrometheusRegistry
Then the registry should be created successfully
Scenario: PrometheusRegistry registers instruments for all 14 metric keys
When I create a PrometheusRegistry
Then the registry should have 14 instruments registered
Scenario: PrometheusRegistry registers histogram for PLAN_DURATION_MS
When I create a PrometheusRegistry
Then the instrument for plan_duration_ms should be a Histogram
Scenario: PrometheusRegistry registers counter for LLM_CALL_COUNT
When I create a PrometheusRegistry
Then the instrument for llm_call_count should be a Counter
Scenario: PrometheusRegistry registers gauge for PLAN_DECISION_COUNT
When I create a PrometheusRegistry
Then the instrument for plan_decision_count should be a Gauge
Scenario: PrometheusRegistry registers counter for PLAN_TOTAL_COST_USD
When I create a PrometheusRegistry
Then the instrument for plan_total_cost_usd should be a Counter
Scenario: PrometheusRegistry registers histogram for ACTOR_LATENCY_MS
When I create a PrometheusRegistry
Then the instrument for actor_latency_ms should be a Histogram
Scenario: PrometheusRegistry registers counter for TOOL_INVOCATION_COUNT
When I create a PrometheusRegistry
Then the instrument for tool_invocation_count should be a Counter
Scenario: PrometheusRegistry registers gauge for TOOL_ERROR_RATE
When I create a PrometheusRegistry
Then the instrument for tool_error_rate should be a Gauge
Scenario: PrometheusRegistry registers histogram for CONTEXT_BUILD_TIME_MS
When I create a PrometheusRegistry
Then the instrument for context_build_time_ms should be a Histogram
# --- Recording metrics ---
Scenario: Recording a histogram MetricEntry observes the value
Given a PrometheusRegistry
And a plan_id "01HX0000000000MMMMMMMMMMMM"
When I record a PLAN_DURATION_MS histogram entry with value 1500.0
Then the Prometheus text output should contain "cleveragents_plan_duration_ms"
Scenario: Recording a counter MetricEntry increments the counter
Given a PrometheusRegistry
And a plan_id "01HX0000000000MMMMMMMMMMMM"
When I record a LLM_CALL_COUNT counter entry with value 3.0
Then the Prometheus text output should contain "cleveragents_llm_call_count"
Scenario: Recording a gauge MetricEntry sets the gauge value
Given a PrometheusRegistry
And a plan_id "01HX0000000000MMMMMMMMMMMM"
When I record a PLAN_DECISION_COUNT gauge entry with value 7.0
Then the Prometheus text output should contain "cleveragents_plan_decision_count"
Scenario: generate_text returns bytes
Given a PrometheusRegistry
When I call generate_text
Then the generate_text result should be bytes
Scenario: generate_text output is non-empty after recording a metric
Given a PrometheusRegistry
And a plan_id "01HX0000000000MMMMMMMMMMMM"
When I record a LLM_TOTAL_TOKENS counter entry with value 1000.0
And I call generate_text
Then the generate_text result should be non-empty bytes
# --- File export ---
Scenario: export_to_file writes metrics to a file
Given a PrometheusRegistry with a temp metrics file
And a plan_id "01HX0000000000MMMMMMMMMMMM"
When I record a ACTOR_INVOCATION_COUNT counter entry with value 5.0
And I export metrics to file
Then the metrics file should exist
And the metrics file should contain "cleveragents_actor_invocation_count"
Scenario: export_to_file raises RuntimeError when no path configured
Given a PrometheusRegistry
When I call export_to_file without a path
Then a RuntimeError should be raised
# --- Settings integration ---
Scenario: PrometheusRegistry respects metrics_export_prometheus setting
Given settings with metrics_export_prometheus enabled
When I create a MetricsEmitter from settings
Then the emitter prometheus_enabled should be True
Scenario: PrometheusRegistry is disabled by default in settings
Given default settings for prometheus
When I create a MetricsEmitter from settings
Then the emitter prometheus_enabled should be False
features/observability/structlog_config.feature
+81
View File
@@ -0,0 +1,81 @@
Feature: JSON Structlog Configuration
As a platform operator
I want configurable structured logging with JSON output and correlation IDs
So that I can parse and correlate log records across plan executions
# --- configure_structlog ---
Scenario: configure_structlog can be called with json format
When I call configure_structlog with log_format "json"
Then no exception should be raised from configure_structlog
Scenario: configure_structlog can be called with text format
When I call configure_structlog with log_format "text"
Then no exception should be raised from configure_structlog
Scenario: configure_structlog can be called with INFO log level
When I call configure_structlog with log_level "INFO"
Then no exception should be raised from configure_structlog
Scenario: configure_structlog can be called with DEBUG log level
When I call configure_structlog with log_level "DEBUG"
Then no exception should be raised from configure_structlog
# --- Correlation ID context variable ---
Scenario: set_correlation_id stores a value retrievable by get_correlation_id
When I set the correlation ID to "CORR-ABC-123"
Then get_correlation_id should return "CORR-ABC-123"
Scenario: get_correlation_id returns None when not set
Given the correlation ID is cleared
Then get_correlation_id should return None
Scenario: set_correlation_id can be updated
When I set the correlation ID to "CORR-FIRST"
And I set the correlation ID to "CORR-SECOND"
Then get_correlation_id should return "CORR-SECOND"
Scenario: set_correlation_id can be cleared with None
When I set the correlation ID to "CORR-TO-CLEAR"
And I set the correlation ID to None
Then get_correlation_id should return None
# --- inject_correlation_id processor ---
Scenario: inject_correlation_id adds correlation_id to event dict when set
Given the correlation ID is set to "PROC-CORR-001"
When I run the inject_correlation_id processor on an empty event dict
Then the event dict should contain key "correlation_id" with value "PROC-CORR-001"
Scenario: inject_correlation_id does not overwrite existing correlation_id
Given the correlation ID is set to "PROC-CORR-002"
When I run the inject_correlation_id processor on an event dict with correlation_id "EXISTING"
Then the event dict should contain key "correlation_id" with value "EXISTING"
Scenario: inject_correlation_id does nothing when correlation ID is None
Given the correlation ID is cleared
When I run the inject_correlation_id processor on an empty event dict
Then the event dict should not contain key "correlation_id"
# --- Settings integration ---
Scenario: Settings has log_format field defaulting to text
When I create default settings for structlog
Then settings.log_format should equal "text"
Scenario: Settings log_format can be set to json
When I create settings with log_format "json"
Then settings.log_format should equal "json"
Scenario: Settings has metrics_prometheus_port field
When I create default settings for structlog
Then settings.metrics_prometheus_port should be an integer
Scenario: Settings has audit_jsonl_path field
When I create default settings for structlog
Then settings.audit_jsonl_path should be a string
Scenario: Settings has metrics_file_path field
When I create default settings for structlog
Then settings.metrics_file_path should be a string
features/steps/audit_bridge_steps.py
+403
View File
@@ -0,0 +1,403 @@
"""Step definitions for the automatic audit bridge feature (Forgejo #940)."""
from __future__ import annotations
import json
import tempfile
from pathlib import Path
from typing import Any
from behave import given, then, when
from behave.runner import Context
from cleveragents.infrastructure.events.models import DomainEvent
from cleveragents.infrastructure.events.reactive import ReactiveEventBus
from cleveragents.infrastructure.events.types import EventType
from cleveragents.infrastructure.observability.audit_bridge import (
AUDIT_EVENT_TYPES,
AuditBridge,
_build_audit_record,
)
__all__: list[str] = []
VALID_PLAN_ID = "01HX0000000000MMMMMMMMMMMM"
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_event(
event_type: EventType,
*,
plan_id: str | None = VALID_PLAN_ID,
actor_name: str | None = None,
correlation_id: str | None = None,
session_id: str | None = None,
details: dict[str, Any] | None = None,
) -> DomainEvent:
kwargs: dict[str, Any] = {
"event_type": event_type,
"plan_id": plan_id,
"actor_name": actor_name,
"session_id": session_id,
"details": details or {},
}
if correlation_id is not None:
kwargs["correlation_id"] = correlation_id
return DomainEvent(**kwargs)
# ---------------------------------------------------------------------------
# Given — event bus
# ---------------------------------------------------------------------------
@given("a ReactiveEventBus for audit bridge")
def step_given_reactive_event_bus(context: Context) -> None:
context.event_bus = ReactiveEventBus()
# ---------------------------------------------------------------------------
# When — AuditBridge creation
# ---------------------------------------------------------------------------
@when("I create an AuditBridge with the event bus")
def step_create_audit_bridge(context: Context) -> None:
context.temp_dir = tempfile.mkdtemp()
context.audit_path = Path(context.temp_dir) / "audit.jsonl"
context.audit_bridge = AuditBridge(
event_bus=context.event_bus,
audit_path=context.audit_path,
capture_in_memory=True,
)
# ---------------------------------------------------------------------------
# Then — AuditBridge creation assertions
# ---------------------------------------------------------------------------
@then("the audit bridge should be created successfully")
def step_audit_bridge_created(context: Context) -> None:
assert context.audit_bridge is not None
assert isinstance(context.audit_bridge, AuditBridge)
@then("the audit bridge should cover at least {count:d} event types")
def step_audit_bridge_covers_event_types(context: Context, count: int) -> None:
assert len(AUDIT_EVENT_TYPES) >= count, (
f"Expected at least {count} audit event types, got {len(AUDIT_EVENT_TYPES)}"
)
# ---------------------------------------------------------------------------
# When — building audit records
# ---------------------------------------------------------------------------
@when("I build an audit record from a plan_applied event")
def step_build_record_plan_applied(context: Context) -> None:
event = _make_event(EventType.PLAN_APPLIED)
context.audit_record = _build_audit_record(event)
@when("I build an audit record from a plan_errored event")
def step_build_record_plan_errored(context: Context) -> None:
event = _make_event(EventType.PLAN_ERRORED)
context.audit_record = _build_audit_record(event)
@when("I build an audit record from an auth_failure event")
def step_build_record_auth_failure(context: Context) -> None:
event = _make_event(EventType.AUTH_FAILURE)
context.audit_record = _build_audit_record(event)
@when("I build an audit record from an invariant_violated event")
def step_build_record_invariant_violated(context: Context) -> None:
event = _make_event(EventType.INVARIANT_VIOLATED)
context.audit_record = _build_audit_record(event)
@when('I build an audit record from a plan_applied event with actor "{actor}"')
def step_build_record_with_actor(context: Context, actor: str) -> None:
event = _make_event(EventType.PLAN_APPLIED, actor_name=actor)
context.audit_record = _build_audit_record(event)
@when('I build an audit record from a plan_applied event with plan_id "{plan_id}"')
def step_build_record_with_plan_id(context: Context, plan_id: str) -> None:
event = _make_event(EventType.PLAN_APPLIED, plan_id=plan_id)
context.audit_record = _build_audit_record(event)
@when('I build an audit record from a plan_applied event with correlation_id "{cid}"')
def step_build_record_with_correlation_id(context: Context, cid: str) -> None:
event = _make_event(EventType.PLAN_APPLIED, correlation_id=cid)
context.audit_record = _build_audit_record(event)
# ---------------------------------------------------------------------------
# Then — AuditRecord field assertions
# ---------------------------------------------------------------------------
@then('the audit record should have a "{field}" field')
def step_record_has_field(context: Context, field: str) -> None:
assert hasattr(context.audit_record, field), f"AuditRecord missing field '{field}'"
@then("the audit record timestamp should be a valid ISO-8601 string")
def step_record_timestamp_iso8601(context: Context) -> None:
ts = context.audit_record.timestamp
assert isinstance(ts, str), f"Expected str timestamp, got {type(ts).__name__}"
assert "T" in ts, f"Timestamp '{ts}' does not look like ISO-8601"
@then('the audit record event_type should equal "{expected}"')
def step_record_event_type_equals(context: Context, expected: str) -> None:
assert context.audit_record.event_type == expected, (
f"Expected event_type '{expected}', got '{context.audit_record.event_type}'"
)
@then('the audit record outcome should equal "{expected}"')
def step_record_outcome_equals(context: Context, expected: str) -> None:
assert context.audit_record.outcome == expected, (
f"Expected outcome '{expected}', got '{context.audit_record.outcome}'"
)
@then('the audit record actor should equal "{expected}"')
def step_record_actor_equals(context: Context, expected: str) -> None:
assert context.audit_record.actor == expected, (
f"Expected actor '{expected}', got '{context.audit_record.actor}'"
)
@then('the audit record plan_id should equal "{expected}"')
def step_record_plan_id_equals(context: Context, expected: str) -> None:
assert context.audit_record.plan_id == expected, (
f"Expected plan_id '{expected}', got '{context.audit_record.plan_id}'"
)
@then('the audit record correlation_id should equal "{expected}"')
def step_record_correlation_id_equals(context: Context, expected: str) -> None:
assert context.audit_record.correlation_id == expected, (
f"Expected correlation_id '{expected}', got "
f"'{context.audit_record.correlation_id}'"
)
# ---------------------------------------------------------------------------
# Given — AuditBridge with in-memory capture
# ---------------------------------------------------------------------------
@given("an AuditBridge with in-memory capture")
def step_given_audit_bridge_in_memory(context: Context) -> None:
context.temp_dir = tempfile.mkdtemp()
context.audit_path = Path(context.temp_dir) / "audit.jsonl"
context.event_bus = ReactiveEventBus()
context.audit_bridge = AuditBridge(
event_bus=context.event_bus,
audit_path=context.audit_path,
capture_in_memory=True,
)
@given("an AuditBridge writing to a temp file")
def step_given_audit_bridge_temp_file(context: Context) -> None:
context.temp_dir = tempfile.mkdtemp()
context.audit_path = Path(context.temp_dir) / "audit.jsonl"
context.event_bus = ReactiveEventBus()
context.audit_bridge = AuditBridge(
event_bus=context.event_bus,
audit_path=context.audit_path,
capture_in_memory=False,
)
@given("an AuditBridge with an invalid file path")
def step_given_audit_bridge_invalid_path(context: Context) -> None:
context.event_bus = ReactiveEventBus()
# Use a path that cannot be written to (root-owned directory)
context.audit_bridge = AuditBridge(
event_bus=context.event_bus,
audit_path=Path("/proc/invalid_audit_test_path/audit.jsonl"),
capture_in_memory=True,
)
# ---------------------------------------------------------------------------
# When — emitting events
# ---------------------------------------------------------------------------
@when("a plan_applied event is emitted on the bus")
def step_emit_plan_applied_on_bus(context: Context) -> None:
context.event_bus.emit(_make_event(EventType.PLAN_APPLIED))
@when("a plan_cancelled event is emitted on the bus")
def step_emit_plan_cancelled_on_bus(context: Context) -> None:
context.event_bus.emit(_make_event(EventType.PLAN_CANCELLED))
@when("a config_changed event is emitted on the bus")
def step_emit_config_changed_on_bus(context: Context) -> None:
context.event_bus.emit(_make_event(EventType.CONFIG_CHANGED))
@when("an auth_success event is emitted on the bus")
def step_emit_auth_success_on_bus(context: Context) -> None:
context.event_bus.emit(_make_event(EventType.AUTH_SUCCESS))
@when("an auth_failure event is emitted on the bus")
def step_emit_auth_failure_on_bus(context: Context) -> None:
context.event_bus.emit(_make_event(EventType.AUTH_FAILURE))
@when("a plan_created event is emitted on the bus")
def step_emit_plan_created_on_bus(context: Context) -> None:
context.event_bus.emit(_make_event(EventType.PLAN_CREATED))
@when("an actor_invoked event is emitted on the bus")
def step_emit_actor_invoked_on_bus(context: Context) -> None:
context.event_bus.emit(_make_event(EventType.ACTOR_INVOKED))
# ---------------------------------------------------------------------------
# Then — in-memory record assertions
# ---------------------------------------------------------------------------
@then("the audit bridge should have {count:d} record")
def step_audit_bridge_has_n_records_singular(context: Context, count: int) -> None:
actual = len(context.audit_bridge.records)
assert actual == count, f"Expected {count} record(s), got {actual}"
@then("the audit bridge should have {count:d} records")
def step_audit_bridge_has_n_records(context: Context, count: int) -> None:
actual = len(context.audit_bridge.records)
assert actual == count, f"Expected {count} record(s), got {actual}"
@then('the first record event_type should equal "{expected}"')
def step_first_record_event_type(context: Context, expected: str) -> None:
records = context.audit_bridge.records
assert len(records) >= 1, "No records in audit bridge"
assert records[0].event_type == expected, (
f"Expected event_type '{expected}', got '{records[0].event_type}'"
)
# ---------------------------------------------------------------------------
# Then — JSONL file assertions
# ---------------------------------------------------------------------------
@then("the audit jsonl file should exist")
def step_audit_jsonl_exists(context: Context) -> None:
assert context.audit_path.exists(), (
f"Audit JSONL file {context.audit_path} does not exist"
)
@then("the audit jsonl file should contain {count:d} line")
def step_audit_jsonl_has_n_lines_singular(context: Context, count: int) -> None:
lines = [
line
for line in context.audit_path.read_text(encoding="utf-8").splitlines()
if line.strip()
]
assert len(lines) == count, f"Expected {count} line(s), got {len(lines)}"
@then("the audit jsonl file should contain {count:d} lines")
def step_audit_jsonl_has_n_lines(context: Context, count: int) -> None:
lines = [
line
for line in context.audit_path.read_text(encoding="utf-8").splitlines()
if line.strip()
]
assert len(lines) == count, f"Expected {count} line(s), got {len(lines)}"
@then("each line in the audit jsonl file should be valid JSON")
def step_audit_jsonl_valid_json(context: Context) -> None:
lines = [
line
for line in context.audit_path.read_text(encoding="utf-8").splitlines()
if line.strip()
]
assert len(lines) > 0, "No lines in audit JSONL file"
for i, line in enumerate(lines):
try:
json.loads(line)
except json.JSONDecodeError as exc:
raise AssertionError(
f"Line {i + 1} is not valid JSON: {exc}\nLine: {line}"
) from exc
@then('the first jsonl record should have field "{field}"')
def step_first_jsonl_record_has_field(context: Context, field: str) -> None:
lines = [
line
for line in context.audit_path.read_text(encoding="utf-8").splitlines()
if line.strip()
]
assert len(lines) >= 1, "No lines in audit JSONL file"
record = json.loads(lines[0])
assert field in record, (
f"Expected field '{field}' in JSONL record, got keys: {list(record.keys())}"
)
# ---------------------------------------------------------------------------
# Then — error resilience
# ---------------------------------------------------------------------------
@then("no exception should propagate from the audit bridge write")
def step_no_exception_from_audit_bridge(context: Context) -> None:
# If we got here without an exception, the test passes
pass
# ---------------------------------------------------------------------------
# Then — to_dict / to_json
# ---------------------------------------------------------------------------
@then("to_dict should return a dict")
def step_to_dict_returns_dict(context: Context) -> None:
result = context.audit_record.to_dict()
assert isinstance(result, dict), f"Expected dict, got {type(result).__name__}"
@then('the dict should have key "{key}"')
def step_dict_has_key(context: Context, key: str) -> None:
result = context.audit_record.to_dict()
assert key in result, (
f"Expected key '{key}' in dict, got keys: {list(result.keys())}"
)
@then("to_json should return a valid JSON string")
def step_to_json_returns_valid_json(context: Context) -> None:
result = context.audit_record.to_json()
assert isinstance(result, str), f"Expected str, got {type(result).__name__}"
try:
json.loads(result)
except json.JSONDecodeError as exc:
raise AssertionError(f"to_json() returned invalid JSON: {exc}") from exc
features/steps/prometheus_metrics_steps.py
+261
View File
@@ -0,0 +1,261 @@
"""Step definitions for Prometheus metrics registry feature (Forgejo #940)."""
from __future__ import annotations
import tempfile
from pathlib import Path
from behave import given, then, when
from behave.runner import Context
from cleveragents.config.settings import Settings
from cleveragents.domain.models.observability.metrics import MetricCollector
from cleveragents.infrastructure.observability.metrics_emitter import MetricsEmitter
from cleveragents.infrastructure.observability.prometheus_registry import (
PrometheusRegistry,
)
__all__: list[str] = []
VALID_PLAN_ID = "01HX0000000000MMMMMMMMMMMM"
# ---------------------------------------------------------------------------
# When — registry creation
# ---------------------------------------------------------------------------
@when("I create a PrometheusRegistry")
def step_create_registry(context: Context) -> None:
context.registry = PrometheusRegistry()
context.registry_error = None
# ---------------------------------------------------------------------------
# Then — registry creation assertions
# ---------------------------------------------------------------------------
@then("the registry should be created successfully")
def step_registry_created(context: Context) -> None:
assert context.registry is not None
assert isinstance(context.registry, PrometheusRegistry)
@then("the registry should have 14 instruments registered")
def step_registry_14_instruments(context: Context) -> None:
assert len(context.registry.instruments) == 14, (
f"Expected 14 instruments, got {len(context.registry.instruments)}: "
f"{list(context.registry.instruments.keys())}"
)
@then("the instrument for {key} should be a Histogram")
def step_instrument_is_histogram(context: Context, key: str) -> None:
try:
from prometheus_client import Histogram # type: ignore[import-untyped]
except ImportError:
context.scenario.skip("prometheus_client not available")
return
instrument = context.registry.instruments.get(key)
assert instrument is not None, f"No instrument found for key '{key}'"
assert isinstance(instrument, Histogram), (
f"Expected Histogram for '{key}', got {type(instrument).__name__}"
)
@then("the instrument for {key} should be a Counter")
def step_instrument_is_counter(context: Context, key: str) -> None:
try:
from prometheus_client import Counter # type: ignore[import-untyped]
except ImportError:
context.scenario.skip("prometheus_client not available")
return
instrument = context.registry.instruments.get(key)
assert instrument is not None, f"No instrument found for key '{key}'"
assert isinstance(instrument, Counter), (
f"Expected Counter for '{key}', got {type(instrument).__name__}"
)
@then("the instrument for {key} should be a Gauge")
def step_instrument_is_gauge(context: Context, key: str) -> None:
try:
from prometheus_client import Gauge # type: ignore[import-untyped]
except ImportError:
context.scenario.skip("prometheus_client not available")
return
instrument = context.registry.instruments.get(key)
assert instrument is not None, f"No instrument found for key '{key}'"
assert isinstance(instrument, Gauge), (
f"Expected Gauge for '{key}', got {type(instrument).__name__}"
)
# ---------------------------------------------------------------------------
# Given — registry with plan_id
# ---------------------------------------------------------------------------
@given("a PrometheusRegistry")
def step_given_registry(context: Context) -> None:
context.registry = PrometheusRegistry()
@given('a plan_id "{plan_id}"')
def step_given_plan_id_quoted(context: Context, plan_id: str) -> None:
context.plan_id = plan_id
# ---------------------------------------------------------------------------
# When — recording metrics
# ---------------------------------------------------------------------------
@when("I record a PLAN_DURATION_MS histogram entry with value {value:f}")
def step_record_plan_duration(context: Context, value: float) -> None:
entry = MetricCollector.plan_duration(context.plan_id, value)
context.registry.record(entry)
@when("I record a LLM_CALL_COUNT counter entry with value {value:f}")
def step_record_llm_call_count(context: Context, value: float) -> None:
entry = MetricCollector.llm_call_count(context.plan_id, value)
context.registry.record(entry)
@when("I record a PLAN_DECISION_COUNT gauge entry with value {value:f}")
def step_record_plan_decision_count(context: Context, value: float) -> None:
entry = MetricCollector.plan_decision_count(context.plan_id, value)
context.registry.record(entry)
@when("I record a LLM_TOTAL_TOKENS counter entry with value {value:f}")
def step_record_llm_total_tokens(context: Context, value: float) -> None:
entry = MetricCollector.llm_total_tokens(context.plan_id, value)
context.registry.record(entry)
@when("I record a ACTOR_INVOCATION_COUNT counter entry with value {value:f}")
def step_record_actor_invocation_count(context: Context, value: float) -> None:
entry = MetricCollector.actor_invocation_count(context.plan_id, value)
context.registry.record(entry)
# ---------------------------------------------------------------------------
# When — generate_text
# ---------------------------------------------------------------------------
@when("I call generate_text")
def step_call_generate_text(context: Context) -> None:
context.generate_text_result = context.registry.generate_text()
# ---------------------------------------------------------------------------
# Then — generate_text assertions
# ---------------------------------------------------------------------------
@then("the generate_text result should be bytes")
def step_result_is_bytes(context: Context) -> None:
assert isinstance(context.generate_text_result, bytes), (
f"Expected bytes, got {type(context.generate_text_result).__name__}"
)
@then("the generate_text result should be non-empty bytes")
def step_result_is_non_empty_bytes(context: Context) -> None:
assert isinstance(context.generate_text_result, bytes)
assert len(context.generate_text_result) > 0
@then('the Prometheus text output should contain "{text}"')
def step_prometheus_output_contains(context: Context, text: str) -> None:
output = context.registry.generate_text().decode("utf-8")
assert text in output, (
f"Expected '{text}' in Prometheus output, but got:\n{output[:500]}"
)
# ---------------------------------------------------------------------------
# Given / When / Then — file export
# ---------------------------------------------------------------------------
@given("a PrometheusRegistry with a temp metrics file")
def step_given_registry_with_temp_file(context: Context) -> None:
context.temp_dir = tempfile.mkdtemp()
context.metrics_file = Path(context.temp_dir) / "metrics.prom"
context.registry = PrometheusRegistry(metrics_file=context.metrics_file)
@when("I export metrics to file")
def step_export_to_file(context: Context) -> None:
context.registry.export_to_file()
@then("the metrics file should exist")
def step_metrics_file_exists(context: Context) -> None:
assert context.metrics_file.exists(), (
f"Metrics file {context.metrics_file} does not exist"
)
@then('the metrics file should contain "{text}"')
def step_metrics_file_contains(context: Context, text: str) -> None:
content = context.metrics_file.read_text(encoding="utf-8")
assert text in content, f"Expected '{text}' in metrics file, got:\n{content[:500]}"
@when("I call export_to_file without a path")
def step_export_to_file_no_path(context: Context) -> None:
try:
context.registry.export_to_file()
context.export_error = None
except RuntimeError as exc:
context.export_error = exc
@then("a RuntimeError should be raised")
def step_runtime_error_raised(context: Context) -> None:
assert context.export_error is not None, "Expected RuntimeError but none was raised"
assert isinstance(context.export_error, RuntimeError)
# ---------------------------------------------------------------------------
# Settings integration
# ---------------------------------------------------------------------------
@given("settings with metrics_export_prometheus enabled")
def step_settings_prometheus_enabled(context: Context) -> None:
import os
Settings._instance = None
os.environ["CLEVERAGENTS_METRICS_EXPORT_PROMETHEUS"] = "true"
try:
context.settings = Settings(database_url="sqlite:///:memory:")
finally:
del os.environ["CLEVERAGENTS_METRICS_EXPORT_PROMETHEUS"]
@given("default settings for prometheus")
def step_default_settings(context: Context) -> None:
Settings._instance = None
context.settings = Settings(database_url="sqlite:///:memory:")
@when("I create a MetricsEmitter from settings")
def step_create_emitter_from_settings(context: Context) -> None:
context.emitter = MetricsEmitter.from_settings(context.settings)
@then("the emitter prometheus_enabled should be True")
def step_emitter_prometheus_enabled_true(context: Context) -> None:
assert context.emitter.prometheus_enabled is True
@then("the emitter prometheus_enabled should be False")
def step_emitter_prometheus_enabled_false(context: Context) -> None:
assert context.emitter.prometheus_enabled is False
features/steps/structlog_config_steps.py
+170
View File
@@ -0,0 +1,170 @@
"""Step definitions for JSON structlog configuration feature (Forgejo #940)."""
from __future__ import annotations
from typing import Any
from behave import given, then, when
from behave.runner import Context
from cleveragents.config.settings import Settings
from cleveragents.infrastructure.observability.structlog_config import (
configure_structlog,
get_correlation_id,
inject_correlation_id,
set_correlation_id,
)
__all__: list[str] = []
# ---------------------------------------------------------------------------
# When — configure_structlog
# ---------------------------------------------------------------------------
@when('I call configure_structlog with log_format "{log_format}"')
def step_configure_structlog_format(context: Context, log_format: str) -> None:
context.configure_error = None
try:
configure_structlog(log_format=log_format)
except Exception as exc:
context.configure_error = exc
@when('I call configure_structlog with log_level "{log_level}"')
def step_configure_structlog_level(context: Context, log_level: str) -> None:
context.configure_error = None
try:
configure_structlog(log_level=log_level)
except Exception as exc:
context.configure_error = exc
@then("no exception should be raised from configure_structlog")
def step_no_exception_raised(context: Context) -> None:
error = getattr(context, "configure_error", None)
assert error is None, f"Unexpected exception: {error}"
# ---------------------------------------------------------------------------
# When / Then — correlation ID context variable
# ---------------------------------------------------------------------------
@when('I set the correlation ID to "{cid}"')
def step_set_correlation_id(context: Context, cid: str) -> None:
set_correlation_id(cid)
@when("I set the correlation ID to None")
def step_set_correlation_id_none(context: Context) -> None:
set_correlation_id(None)
@given("the correlation ID is cleared")
def step_clear_correlation_id(context: Context) -> None:
set_correlation_id(None)
@given('the correlation ID is set to "{cid}"')
def step_given_correlation_id_set(context: Context, cid: str) -> None:
set_correlation_id(cid)
@then('get_correlation_id should return "{expected}"')
def step_get_correlation_id_returns(context: Context, expected: str) -> None:
actual = get_correlation_id()
assert actual == expected, f"Expected '{expected}', got '{actual}'"
@then("get_correlation_id should return None")
def step_get_correlation_id_returns_none(context: Context) -> None:
assert get_correlation_id() is None, f"Expected None, got '{get_correlation_id()}'"
# ---------------------------------------------------------------------------
# When / Then — inject_correlation_id processor
# ---------------------------------------------------------------------------
@when("I run the inject_correlation_id processor on an empty event dict")
def step_run_processor_empty_dict(context: Context) -> None:
context.event_dict: dict[str, Any] = {}
context.event_dict = inject_correlation_id(None, "info", context.event_dict)
@when(
'I run the inject_correlation_id processor on an event dict with correlation_id "{existing}"'
)
def step_run_processor_existing_cid(context: Context, existing: str) -> None:
context.event_dict = {"correlation_id": existing}
context.event_dict = inject_correlation_id(None, "info", context.event_dict)
@then('the event dict should contain key "{key}" with value "{value}"')
def step_event_dict_has_key_value(context: Context, key: str, value: str) -> None:
assert key in context.event_dict, (
f"Key '{key}' not found in event dict: {context.event_dict}"
)
assert context.event_dict[key] == value, (
f"Expected event_dict['{key}'] == '{value}', got '{context.event_dict[key]}'"
)
@then('the event dict should not contain key "{key}"')
def step_event_dict_missing_key(context: Context, key: str) -> None:
assert key not in context.event_dict, (
f"Key '{key}' unexpectedly found in event dict: {context.event_dict}"
)
# ---------------------------------------------------------------------------
# Settings integration
# ---------------------------------------------------------------------------
@when("I create default settings for structlog")
def step_create_default_settings(context: Context) -> None:
Settings._instance = None
context.settings = Settings(database_url="sqlite:///:memory:")
@when('I create settings with log_format "{log_format}"')
def step_create_settings_with_log_format(context: Context, log_format: str) -> None:
import os
Settings._instance = None
os.environ["CLEVERAGENTS_LOG_FORMAT"] = log_format
try:
context.settings = Settings(database_url="sqlite:///:memory:")
finally:
del os.environ["CLEVERAGENTS_LOG_FORMAT"]
@then('settings.log_format should equal "{expected}"')
def step_settings_log_format_equals(context: Context, expected: str) -> None:
assert context.settings.log_format == expected, (
f"Expected log_format '{expected}', got '{context.settings.log_format}'"
)
@then("settings.metrics_prometheus_port should be an integer")
def step_settings_prometheus_port_is_int(context: Context) -> None:
assert isinstance(context.settings.metrics_prometheus_port, int), (
f"Expected int, got {type(context.settings.metrics_prometheus_port).__name__}"
)
@then("settings.audit_jsonl_path should be a string")
def step_settings_audit_jsonl_path_is_str(context: Context) -> None:
assert isinstance(context.settings.audit_jsonl_path, str), (
f"Expected str, got {type(context.settings.audit_jsonl_path).__name__}"
)
@then("settings.metrics_file_path should be a string")
def step_settings_metrics_file_path_is_str(context: Context) -> None:
assert isinstance(context.settings.metrics_file_path, str), (
f"Expected str, got {type(context.settings.metrics_file_path).__name__}"
)
pyproject.toml
+1
View File
@@ -47,6 +47,7 @@ dependencies = [
"jsonschema>=4.20.0", # JSON Schema validation for tool inputs/outputs
"tomlkit>=0.13.0", # TOML writing with comment preservation for config CLI
"tenacity>=8.2.0", # Retry framework for service layer resilience
"prometheus-client>=0.20.0", # Prometheus metrics export (#940)
]
[project.optional-dependencies]
src/cleveragents/config/settings.py
+27
View File
@@ -505,6 +505,33 @@ class Settings(BaseSettings):
description="Enable Prometheus metrics export endpoint.",
)
# Observability configuration (#940)
metrics_prometheus_port: int = Field(
default=9090,
ge=1,
le=65535,
validation_alias=AliasChoices("CLEVERAGENTS_METRICS_PROMETHEUS_PORT"),
description="HTTP port for Prometheus /metrics endpoint (server mode).",
)
metrics_file_path: str = Field(
default="metrics.prom",
validation_alias=AliasChoices("CLEVERAGENTS_METRICS_FILE_PATH"),
description="File path for Prometheus text-format export (local mode).",
)
audit_jsonl_path: str = Field(
default="audit.jsonl",
validation_alias=AliasChoices("CLEVERAGENTS_AUDIT_JSONL_PATH"),
description="File path for the automatic audit bridge JSONL output.",
)
log_format: str = Field(
default="text",
validation_alias=AliasChoices("CLEVERAGENTS_LOG_FORMAT"),
description=(
"Log output format: 'json' for structured JSON lines, "
"'text' for human-readable console output."
),
)
# Retry policy configuration (M6 - #313)
retry_max_attempts: int = Field(
default=3,
src/cleveragents/infrastructure/observability/__init__.py
+31 -5
View File
@@ -1,13 +1,39 @@
"""Observability infrastructure for metric emission.
"""Observability infrastructure package.
Provides ``MetricsEmitter`` for structured log emission of
``MetricEntry`` objects in local mode.
Provides:
Based on Forgejo issue #579.
* ``MetricsEmitter`` — structured log emission of ``MetricEntry`` objects.
* ``PrometheusRegistry`` — Prometheus metrics registry with counters/histograms/gauges.
* ``AuditBridge`` — automatic audit bridge writing security events to ``audit.jsonl``.
* ``configure_structlog`` — JSON structlog formatter configuration.
Based on Forgejo issues #579 and #940.
"""
from __future__ import annotations
from cleveragents.infrastructure.observability.audit_bridge import (
AUDIT_EVENT_TYPES,
AuditBridge,
AuditRecord,
)
from cleveragents.infrastructure.observability.metrics_emitter import MetricsEmitter
from cleveragents.infrastructure.observability.prometheus_registry import (
PrometheusRegistry,
)
from cleveragents.infrastructure.observability.structlog_config import (
configure_structlog,
get_correlation_id,
set_correlation_id,
)
__all__ = ["MetricsEmitter"]
__all__ = [
"AUDIT_EVENT_TYPES",
"AuditBridge",
"AuditRecord",
"MetricsEmitter",
"PrometheusRegistry",
"configure_structlog",
"get_correlation_id",
"set_correlation_id",
]
src/cleveragents/infrastructure/observability/audit_bridge.py
+304
View File
@@ -0,0 +1,304 @@
"""Automatic audit bridge: event subscriber writing to ``audit.jsonl``.
The ``AuditBridge`` subscribes to all domain events that carry
``audit=True`` semantics (the :data:`AUDIT_EVENT_TYPES` set) and writes
a structured JSON line to ``audit.jsonl`` for each matching event.
Each audit record conforms to the :class:`AuditRecord` schema:
* ``timestamp`` — ISO-8601 UTC string
* ``event_type`` — dot-separated event type string
* ``actor`` — actor name or ``None``
* ``action`` — action extracted from event details or event type
* ``resource`` — resource identifier from event details or ``None``
* ``outcome`` — ``"success"`` or ``"failure"`` based on event type
* ``metadata`` — sanitised copy of event details (secrets redacted)
* ``correlation_id`` — ULID linking related events
* ``plan_id`` — associated plan ULID or ``None``
* ``session_id`` — associated session ULID or ``None``
Based on Forgejo issue #940.
"""
from __future__ import annotations
import json
import logging
from datetime import UTC
from pathlib import Path
from typing import Any
from cleveragents.infrastructure.events.models import DomainEvent
from cleveragents.infrastructure.events.protocol import EventBus
from cleveragents.infrastructure.events.types import EventType
from cleveragents.shared.redaction import redact_dict
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Audit event set — events that are automatically written to audit.jsonl
# ---------------------------------------------------------------------------
#: Domain event types that are automatically bridged to the audit log.
#: These correspond to security-relevant and compliance-critical operations.
AUDIT_EVENT_TYPES: frozenset[EventType] = frozenset(
{
EventType.PLAN_APPLIED,
EventType.PLAN_CANCELLED,
EventType.PLAN_ERRORED,
EventType.RESOURCE_MODIFIED,
EventType.CORRECTION_APPLIED,
EventType.CONFIG_CHANGED,
EventType.ENTITY_DELETED,
EventType.SESSION_CREATED,
EventType.AUTH_SUCCESS,
EventType.AUTH_FAILURE,
EventType.INVARIANT_VIOLATED,
EventType.INVARIANT_ENFORCED,
EventType.DECISION_APPROVED,
EventType.DECISION_CORRECTED,
}
)
# Event types whose outcome is classified as "failure"
_FAILURE_EVENT_TYPES: frozenset[EventType] = frozenset(
{
EventType.PLAN_ERRORED,
EventType.AUTH_FAILURE,
EventType.INVARIANT_VIOLATED,
}
)
class AuditRecord:
"""Structured audit log record.
Attributes:
timestamp: ISO-8601 UTC timestamp string.
event_type: Dot-separated event type string.
actor: Actor name or ``None``.
action: Action description derived from event details or type.
resource: Resource identifier or ``None``.
outcome: ``"success"`` or ``"failure"``.
metadata: Sanitised event details dict.
correlation_id: ULID linking related events.
plan_id: Associated plan ULID or ``None``.
session_id: Associated session ULID or ``None``.
"""
__slots__ = (
"action",
"actor",
"correlation_id",
"event_type",
"metadata",
"outcome",
"plan_id",
"resource",
"session_id",
"timestamp",
)
def __init__(
self,
*,
timestamp: str,
event_type: str,
actor: str | None,
action: str,
resource: str | None,
outcome: str,
metadata: dict[str, Any],
correlation_id: str,
plan_id: str | None,
session_id: str | None,
) -> None:
self.timestamp = timestamp
self.event_type = event_type
self.actor = actor
self.action = action
self.resource = resource
self.outcome = outcome
self.metadata = metadata
self.correlation_id = correlation_id
self.plan_id = plan_id
self.session_id = session_id
def to_dict(self) -> dict[str, Any]:
"""Serialise the record to a plain dict suitable for JSON output."""
return {
"timestamp": self.timestamp,
"event_type": self.event_type,
"actor": self.actor,
"action": self.action,
"resource": self.resource,
"outcome": self.outcome,
"metadata": self.metadata,
"correlation_id": self.correlation_id,
"plan_id": self.plan_id,
"session_id": self.session_id,
}
def to_json(self) -> str:
"""Serialise the record to a JSON string (single line)."""
return json.dumps(self.to_dict(), default=str)
def _build_audit_record(event: DomainEvent) -> AuditRecord:
"""Build an :class:`AuditRecord` from a :class:`DomainEvent`.
Args:
event: The domain event to convert.
Returns:
A structured audit record.
"""
details = dict(event.details) if event.details else {}
sanitised = redact_dict(details)
# Derive action from details or fall back to event type
action: str = (
str(details.get("action_name", ""))
or str(details.get("action", ""))
or str(event.event_type)
)
# Derive resource from details
resource: str | None = (
str(details.get("resource_id", "")) or str(details.get("resource", "")) or None
)
if resource == "":
resource = None
outcome = "failure" if event.event_type in _FAILURE_EVENT_TYPES else "success"
return AuditRecord(
timestamp=event.timestamp.astimezone(UTC).isoformat(),
event_type=str(event.event_type),
actor=event.actor_name,
action=action,
resource=resource,
outcome=outcome,
metadata=sanitised,
correlation_id=event.correlation_id,
plan_id=event.plan_id,
session_id=event.session_id,
)
class AuditBridge:
"""Event subscriber that writes audit-flagged events to ``audit.jsonl``.
On construction the bridge subscribes to every event type in
:data:`AUDIT_EVENT_TYPES` on the provided *event_bus*. When a
matching event is received it is converted to an :class:`AuditRecord`
and appended as a JSON line to the configured audit file.
Errors during file I/O are logged but never propagated so that a
failing audit write does not interrupt normal event processing.
Attributes:
_audit_path: Path to the ``audit.jsonl`` file.
_event_bus: The event bus this bridge is subscribed to.
_records: In-memory list of records written (for testing).
"""
def __init__(
self,
*,
event_bus: EventBus,
audit_path: Path | str = Path("audit.jsonl"),
capture_in_memory: bool = False,
) -> None:
"""Initialise the audit bridge and subscribe to audit event types.
Args:
event_bus: The event bus to subscribe to.
audit_path: Path to the audit JSONL file.
capture_in_memory: If ``True``, also keep records in
``self.records`` (useful for testing without file I/O).
"""
self._audit_path = Path(audit_path)
self._event_bus = event_bus
self._capture_in_memory = capture_in_memory
self._records: list[AuditRecord] = []
for event_type in AUDIT_EVENT_TYPES:
event_bus.subscribe(event_type, self._handle_event)
# ------------------------------------------------------------------
# Internal handler
# ------------------------------------------------------------------
def _handle_event(self, event: DomainEvent) -> None:
"""Handle an incoming audit-flagged domain event.
Args:
event: The domain event to audit.
"""
try:
record = _build_audit_record(event)
if self._capture_in_memory:
self._records.append(record)
self._write_record(record)
except Exception as exc:
logger.warning(
"audit_bridge.handle_error",
extra={
"event_type": str(event.event_type),
"error": str(exc),
},
)
def _write_record(self, record: AuditRecord) -> None:
"""Append *record* as a JSON line to the audit file.
Args:
record: The audit record to persist.
"""
try:
self._audit_path.parent.mkdir(parents=True, exist_ok=True)
with self._audit_path.open("a", encoding="utf-8") as fh:
fh.write(record.to_json() + "\n")
except OSError as exc:
logger.error(
"audit_bridge.write_error",
extra={
"path": str(self._audit_path),
"error": str(exc),
},
)
# ------------------------------------------------------------------
# Public helpers
# ------------------------------------------------------------------
@property
def records(self) -> list[AuditRecord]:
"""In-memory audit records (only populated when ``capture_in_memory=True``)."""
return list(self._records)
def read_records(self) -> list[dict[str, Any]]:
"""Read all records from the audit JSONL file.
Returns:
List of parsed audit record dicts.
"""
if not self._audit_path.exists():
return []
records: list[dict[str, Any]] = []
with self._audit_path.open("r", encoding="utf-8") as fh:
for line in fh:
line = line.strip()
if line:
try:
records.append(json.loads(line))
except json.JSONDecodeError:
logger.warning(
"audit_bridge.malformed_line",
extra={"line": line[:100]},
)
return records
__all__ = ["AUDIT_EVENT_TYPES", "AuditBridge", "AuditRecord"]
src/cleveragents/infrastructure/observability/prometheus_registry.py
+245
View File
@@ -0,0 +1,245 @@
"""Prometheus metrics registry for CleverAgents observability.
Provides a ``PrometheusRegistry`` that wraps ``prometheus_client`` to
expose counters, histograms, and gauges for the 14 operational metric
keys defined in :mod:`cleveragents.domain.models.observability.metrics`.
The registry supports two export modes:
* **Server mode** — exposes a ``/metrics`` HTTP endpoint via
``prometheus_client.start_http_server``.
* **Local mode** — writes metrics to a file via
``prometheus_client.write_to_textfile``.
Based on Forgejo issue #940.
"""
from __future__ import annotations
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Any
import prometheus_client # type: ignore[import-untyped]
from prometheus_client import ( # type: ignore[import-untyped]
CollectorRegistry,
Counter,
Gauge,
Histogram,
generate_latest,
start_http_server,
write_to_textfile,
)
if TYPE_CHECKING:
from cleveragents.domain.models.observability.metrics import MetricEntry
logger = logging.getLogger(__name__)
# Histogram bucket boundaries for latency metrics (milliseconds)
_LATENCY_BUCKETS = (
10.0,
50.0,
100.0,
250.0,
500.0,
1000.0,
2500.0,
5000.0,
10000.0,
float("inf"),
)
# Histogram bucket boundaries for cost metrics (USD)
_COST_BUCKETS = (
0.001,
0.005,
0.01,
0.05,
0.1,
0.5,
1.0,
5.0,
float("inf"),
)
class PrometheusRegistry:
"""Prometheus metrics registry for CleverAgents operational metrics.
Registers one Prometheus instrument per ``OperationalMetricKey``:
* ``HISTOGRAM`` keys → :class:`prometheus_client.Histogram`
* ``COUNTER`` keys → :class:`prometheus_client.Counter`
* ``GAUGE`` keys → :class:`prometheus_client.Gauge`
All instruments are registered on an isolated
:class:`prometheus_client.CollectorRegistry` so that tests can create
multiple instances without collisions.
Attributes:
_registry: The underlying Prometheus collector registry.
_instruments: Mapping from metric key string to Prometheus instrument.
_http_port: Port for the HTTP metrics server (server mode).
_metrics_file: Path for file-based export (local mode).
_server_started: Whether the HTTP server has been started.
"""
def __init__(
self,
*,
http_port: int | None = None,
metrics_file: Path | str | None = None,
) -> None:
"""Initialise the registry and register all instruments.
Args:
http_port: If set, ``start_server()`` will expose metrics on
this port (server mode).
metrics_file: If set, ``export_to_file()`` will write metrics
to this path (local mode).
"""
self._http_port = http_port
self._metrics_file = Path(metrics_file) if metrics_file else None
self._server_started = False
self._instruments: dict[str, Any] = {}
self._registry: CollectorRegistry = CollectorRegistry()
self._register_instruments()
# ------------------------------------------------------------------
# Instrument registration
# ------------------------------------------------------------------
def _register_instruments(self) -> None:
"""Register all 14 operational metric instruments."""
from cleveragents.domain.models.observability.metrics import (
METRIC_DEFINITIONS,
MetricType,
)
for key, defn in METRIC_DEFINITIONS.items():
metric_name = f"cleveragents_{key.value.replace('.', '_')}"
labels = ["plan_id"]
if defn.metric_type == MetricType.HISTOGRAM:
buckets = (
_COST_BUCKETS
if "cost" in key.value or "usd" in key.value
else _LATENCY_BUCKETS
)
self._instruments[key.value] = Histogram(
metric_name,
defn.description or key.value,
labels,
buckets=buckets,
registry=self._registry,
)
elif defn.metric_type == MetricType.COUNTER:
self._instruments[key.value] = Counter(
metric_name,
defn.description or key.value,
labels,
registry=self._registry,
)
elif defn.metric_type == MetricType.GAUGE:
self._instruments[key.value] = Gauge(
metric_name,
defn.description or key.value,
labels,
registry=self._registry,
)
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def record(self, entry: MetricEntry) -> None:
"""Record a single ``MetricEntry`` in the Prometheus registry.
Dispatches to the appropriate Prometheus instrument based on the
entry's ``metric_type``. Unknown keys are silently ignored.
Args:
entry: The metric entry to record.
"""
instrument = self._instruments.get(str(entry.key))
if instrument is None:
logger.debug(
"prometheus_registry.unknown_key",
extra={"key": str(entry.key)},
)
return
plan_id = entry.plan_id or "unknown"
from cleveragents.domain.models.observability.metrics import MetricType
if entry.metric_type == MetricType.HISTOGRAM:
instrument.labels(plan_id=plan_id).observe(entry.value)
elif entry.metric_type == MetricType.COUNTER:
instrument.labels(plan_id=plan_id).inc(entry.value)
elif entry.metric_type == MetricType.GAUGE:
instrument.labels(plan_id=plan_id).set(entry.value)
def generate_text(self) -> bytes:
"""Generate Prometheus text-format metrics output.
Returns:
UTF-8 encoded Prometheus text exposition format.
"""
return generate_latest(self._registry)
def start_server(self, port: int | None = None) -> None:
"""Start the Prometheus HTTP metrics server.
Args:
port: Override the port configured at construction time.
Raises:
RuntimeError: If no port is configured.
RuntimeError: If the server has already been started.
"""
effective_port = port or self._http_port
if effective_port is None:
raise RuntimeError("No HTTP port configured for Prometheus metrics server.")
if self._server_started:
raise RuntimeError("Prometheus metrics server is already running.")
start_http_server(effective_port, registry=self._registry)
self._server_started = True
logger.info(
"prometheus_registry.server_started",
extra={"port": effective_port},
)
def export_to_file(self, path: Path | str | None = None) -> None:
"""Write metrics to a text file in Prometheus format.
Args:
path: Override the file path configured at construction time.
Raises:
RuntimeError: If no file path is configured.
"""
effective_path = Path(path) if path else self._metrics_file
if effective_path is None:
raise RuntimeError("No metrics file path configured for file-based export.")
effective_path.parent.mkdir(parents=True, exist_ok=True)
write_to_textfile(str(effective_path), self._registry)
logger.debug(
"prometheus_registry.exported_to_file",
extra={"path": str(effective_path)},
)
@property
def instruments(self) -> dict[str, Any]:
"""Read-only view of registered Prometheus instruments."""
return dict(self._instruments)
@property
def server_started(self) -> bool:
"""Whether the HTTP metrics server has been started."""
return self._server_started
__all__ = ["PrometheusRegistry"]
src/cleveragents/infrastructure/observability/structlog_config.py
+152
View File
@@ -0,0 +1,152 @@
"""JSON structlog formatter and configuration.
Provides ``configure_structlog`` to set up :mod:`structlog` with either
JSON output (for production / machine-readable logs) or coloured
console output (for development).
The format is controlled by the ``log_format`` setting:
* ``"json"`` — structured JSON lines, one per log record.
* ``"text"`` (default) — human-readable coloured console output.
Log correlation IDs are injected into every log record via the
``inject_correlation_id`` processor when a correlation ID is present in
the current context.
Based on Forgejo issue #940.
"""
from __future__ import annotations
import logging
import sys
from contextvars import ContextVar
from typing import Any
import structlog
# ---------------------------------------------------------------------------
# Correlation ID context variable
# ---------------------------------------------------------------------------
#: Thread/async-safe storage for the current log correlation ID.
_correlation_id_var: ContextVar[str | None] = ContextVar(
"log_correlation_id", default=None
)
def set_correlation_id(correlation_id: str | None) -> None:
"""Set the current log correlation ID in the context.
Args:
correlation_id: The correlation ID to inject into log records,
or ``None`` to clear it.
"""
_correlation_id_var.set(correlation_id)
def get_correlation_id() -> str | None:
"""Return the current log correlation ID from the context.
Returns:
The current correlation ID, or ``None`` if not set.
"""
return _correlation_id_var.get()
# ---------------------------------------------------------------------------
# Custom structlog processors
# ---------------------------------------------------------------------------
def inject_correlation_id(
logger: Any,
method: str,
event_dict: dict[str, Any],
) -> dict[str, Any]:
"""Structlog processor that injects the current correlation ID.
If a correlation ID is set in the current context it is added to the
log record under the key ``"correlation_id"``. Existing values are
not overwritten.
Args:
logger: The structlog logger (unused).
method: The log method name (unused).
event_dict: The current log record dict.
Returns:
The (possibly mutated) event dict.
"""
cid = _correlation_id_var.get()
if cid is not None and "correlation_id" not in event_dict:
event_dict["correlation_id"] = cid
return event_dict
# ---------------------------------------------------------------------------
# Configuration entry point
# ---------------------------------------------------------------------------
def configure_structlog(*, log_format: str = "text", log_level: str = "INFO") -> None:
"""Configure :mod:`structlog` for the application.
Sets up the shared processors, renderer, and stdlib integration.
Safe to call multiple times — subsequent calls reconfigure structlog.
Args:
log_format: ``"json"`` for JSON output, ``"text"`` for console.
log_level: Standard log level string (e.g. ``"INFO"``).
"""
level = getattr(logging, log_level.upper(), logging.INFO)
# Shared processors applied to every log record
shared_processors: list[Any] = [
structlog.contextvars.merge_contextvars,
inject_correlation_id,
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.processors.TimeStamper(fmt="iso", utc=True),
structlog.processors.StackInfoRenderer(),
]
if log_format == "json":
renderer: Any = structlog.processors.JSONRenderer()
else:
renderer = structlog.dev.ConsoleRenderer(colors=sys.stderr.isatty())
structlog.configure(
processors=[
*shared_processors,
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
],
wrapper_class=structlog.make_filtering_bound_logger(level),
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
cache_logger_on_first_use=True,
)
formatter = structlog.stdlib.ProcessorFormatter(
foreign_pre_chain=shared_processors,
processors=[
structlog.stdlib.ProcessorFormatter.remove_processors_meta,
renderer,
],
)
handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(formatter)
root_logger = logging.getLogger()
root_logger.handlers.clear()
root_logger.addHandler(handler)
root_logger.setLevel(level)
__all__ = [
"configure_structlog",
"get_correlation_id",
"inject_correlation_id",
"set_correlation_id",
]