diff --git a/benchmarks/bench_metrics_collection.py b/benchmarks/bench_metrics_collection.py new file mode 100644 index 000000000..630b34575 --- /dev/null +++ b/benchmarks/bench_metrics_collection.py @@ -0,0 +1,101 @@ +"""ASV benchmarks for metrics collection framework overhead. + +Measures the time to create metric entries via histogram/counter/gauge +factory methods, exercise all 14 convenience methods, and emit metrics +through the MetricsEmitter. +""" + +from __future__ import annotations + +import importlib +import sys +from pathlib import Path +from unittest.mock import MagicMock + +_SRC = str(Path(__file__).resolve().parents[1] / "src") +if _SRC not in sys.path: + sys.path.insert(0, _SRC) + +import cleveragents # noqa: E402 + +importlib.reload(cleveragents) + +from cleveragents.domain.models.observability.metrics import ( # noqa: E402 + MetricCollector, + MetricType, + OperationalMetricKey, +) +from cleveragents.infrastructure.observability.metrics_emitter import ( # noqa: E402 + MetricsEmitter, +) + +VALID_PLAN_ID = "01HX0000000000MMMMMMMMMMMM" + + +class MetricCollectorSuite: + """Benchmark MetricCollector typed factory methods.""" + + def time_histogram(self) -> None: + """Create a histogram metric entry.""" + MetricCollector.histogram( + OperationalMetricKey.PLAN_DURATION_MS, + 1500.0, + VALID_PLAN_ID, + ) + + def time_counter(self) -> None: + """Create a counter metric entry.""" + MetricCollector.counter( + OperationalMetricKey.LLM_CALL_COUNT, + 5.0, + VALID_PLAN_ID, + ) + + def time_gauge(self) -> None: + """Create a gauge metric entry.""" + MetricCollector.gauge( + OperationalMetricKey.PLAN_DECISION_COUNT, + 3.0, + VALID_PLAN_ID, + ) + + def time_all_14_convenience_methods(self) -> None: + """Exercise all 14 convenience methods.""" + MetricCollector.plan_duration(VALID_PLAN_ID, 1000.0) + MetricCollector.plan_cost(VALID_PLAN_ID, 0.05) + MetricCollector.plan_decision_count(VALID_PLAN_ID, 5.0) + MetricCollector.subplan_count(VALID_PLAN_ID, 2.0) + MetricCollector.actor_invocation_count(VALID_PLAN_ID, 1.0) + MetricCollector.actor_latency(VALID_PLAN_ID, 250.0) + MetricCollector.tool_invocation_count(VALID_PLAN_ID, 10.0) + MetricCollector.tool_error_rate(VALID_PLAN_ID, 0.01) + MetricCollector.context_build_time(VALID_PLAN_ID, 50.0) + MetricCollector.context_token_count(VALID_PLAN_ID, 8000.0) + MetricCollector.llm_call_count(VALID_PLAN_ID, 3.0) + MetricCollector.llm_total_tokens(VALID_PLAN_ID, 5000.0) + MetricCollector.llm_total_cost(VALID_PLAN_ID, 0.5) + MetricCollector.llm_avg_latency(VALID_PLAN_ID, 200.0) + + +class MetricsEmitterSuite: + """Benchmark MetricsEmitter operations.""" + + def setup(self) -> None: + self._emitter = MetricsEmitter(enabled=True) + self._entries = [ + MetricCollector.plan_duration(VALID_PLAN_ID, float(i * 100)) + for i in range(100) + ] + + def time_emit_single(self) -> None: + """Emit a single metric entry.""" + self._emitter.emit(self._entries[0]) + + def time_emit_batch_100(self) -> None: + """Emit 100 metric entries in a batch.""" + self._emitter.emit_batch(self._entries) + + def time_emit_disabled_noop(self) -> None: + """Verify disabled emitter is fast no-op.""" + disabled = MetricsEmitter(enabled=False) + disabled.emit_batch(self._entries) diff --git a/features/observability/metrics_collection.feature b/features/observability/metrics_collection.feature new file mode 100644 index 000000000..dba987c7b --- /dev/null +++ b/features/observability/metrics_collection.feature @@ -0,0 +1,196 @@ +Feature: Metrics Collection Framework + As a platform operator + I want structured metric collection with typed semantics + So that I can monitor plan execution via histograms, counters, and gauges + + # --- MetricType enum --- + + Scenario: MetricType enum has three members + Then MetricType should have exactly 3 members + + Scenario: MetricType values are correct + Then MetricType HISTOGRAM should equal "histogram" + And MetricType COUNTER should equal "counter" + And MetricType GAUGE should equal "gauge" + + # --- MetricDefinition registry --- + + Scenario: All 14 metric keys have definitions + Then METRIC_DEFINITIONS should have exactly 14 entries + + Scenario: PLAN_DURATION_MS is a histogram + Then the definition for PLAN_DURATION_MS should have metric_type HISTOGRAM + + Scenario: LLM_CALL_COUNT is a counter + Then the definition for LLM_CALL_COUNT should have metric_type COUNTER + + Scenario: PLAN_DECISION_COUNT is a gauge + Then the definition for PLAN_DECISION_COUNT should have metric_type GAUGE + + Scenario: TOOL_ERROR_RATE is a gauge + Then the definition for TOOL_ERROR_RATE should have metric_type GAUGE + + # --- MetricCollector typed factory methods --- + + Scenario: histogram method creates a metric with histogram type + Given a plan_id for metrics collection + When I create a histogram metric for PLAN_DURATION_MS with value 1200.0 + Then the result metric_type should be HISTOGRAM + And the result value should be 1200.0 + + Scenario: counter method creates a metric with counter type + Given a plan_id for metrics collection + When I create a counter metric for LLM_CALL_COUNT with value 5.0 + Then the result metric_type should be COUNTER + And the result value should be 5.0 + + Scenario: gauge method creates a metric with gauge type + Given a plan_id for metrics collection + When I create a gauge metric for PLAN_DECISION_COUNT with value 3.0 + Then the result metric_type should be GAUGE + And the result value should be 3.0 + + # --- All 14 convenience methods --- + + Scenario: plan_duration convenience method + Given a plan_id for metrics collection + When I use the plan_duration convenience with value 2500.0 + Then the collected metric key should equal PLAN_DURATION_MS + And the collected metric value should equal 2500.0 + And the metric_type should be HISTOGRAM + + Scenario: plan_cost convenience method + Given a plan_id for metrics collection + When I use the plan_cost convenience with value 0.12 + Then the collected metric key should equal PLAN_TOTAL_COST_USD + And the collected metric value should equal 0.12 + + Scenario: plan_decision_count convenience method + Given a plan_id for metrics collection + When I use the plan_decision_count convenience with value 7.0 + Then the collected metric key should equal PLAN_DECISION_COUNT + And the collected metric value should equal 7.0 + + Scenario: subplan_count convenience method + Given a plan_id for metrics collection + When I use the subplan_count convenience with value 2.0 + Then the collected metric key should equal SUBPLAN_COUNT + And the collected metric value should equal 2.0 + + Scenario: actor_invocation_count convenience method + Given a plan_id for metrics collection + When I use the actor_invocation_count convenience with value 1.0 + Then the collected metric key should equal ACTOR_INVOCATION_COUNT + And the collected metric value should equal 1.0 + + Scenario: actor_latency convenience method + Given a plan_id for metrics collection + When I use the actor_latency convenience with value 350.0 + Then the collected metric key should equal ACTOR_LATENCY_MS + And the collected metric value should equal 350.0 + + Scenario: tool_invocation_count convenience method + Given a plan_id for metrics collection + When I use the tool_invocation_count convenience with value 10.0 + Then the collected metric key should equal TOOL_INVOCATION_COUNT + And the collected metric value should equal 10.0 + + Scenario: tool_error_rate convenience method + Given a plan_id for metrics collection + When I use the tool_error_rate convenience with value 0.05 + Then the collected metric key should equal TOOL_ERROR_RATE + And the collected metric value should equal 0.05 + + Scenario: context_build_time convenience method + Given a plan_id for metrics collection + When I use the context_build_time convenience with value 45.0 + Then the collected metric key should equal CONTEXT_BUILD_TIME_MS + And the collected metric value should equal 45.0 + + Scenario: context_token_count convenience method + Given a plan_id for metrics collection + When I use the context_token_count convenience with value 8192.0 + Then the collected metric key should equal CONTEXT_TOKEN_COUNT + And the collected metric value should equal 8192.0 + + Scenario: llm_call_count convenience method + Given a plan_id for metrics collection + When I use the llm_call_count convenience with value 15.0 + Then the collected metric key should equal LLM_CALL_COUNT + And the collected metric value should equal 15.0 + + Scenario: llm_total_tokens convenience method + Given a plan_id for metrics collection + When I use the llm_total_tokens convenience with value 50000.0 + Then the collected metric key should equal LLM_TOTAL_TOKENS + And the collected metric value should equal 50000.0 + + Scenario: llm_total_cost convenience method + Given a plan_id for metrics collection + When I use the llm_total_cost convenience with value 1.25 + Then the collected metric key should equal LLM_TOTAL_COST_USD + And the collected metric value should equal 1.25 + + Scenario: llm_avg_latency convenience method + Given a plan_id for metrics collection + When I use the llm_avg_latency convenience with value 200.0 + Then the collected metric key should equal LLM_AVG_LATENCY_MS + And the collected metric value should equal 200.0 + + # --- MetricsEmitter --- + + Scenario: MetricsEmitter emits metric as structured log + Given a plan_id for metrics collection + And a MetricsEmitter in local mode + When I emit a plan duration metric with value 1500.0 + Then the emitter should have emitted successfully + + Scenario: MetricsEmitter respects disabled setting + Given a plan_id for metrics collection + And a MetricsEmitter that is disabled + When I emit a plan duration metric with value 1500.0 + Then the emitter should not have emitted + + Scenario: MetricsEmitter emits batch + Given a plan_id for metrics collection + And a MetricsEmitter in local mode + When I emit a batch of 3 metrics + Then the emitter should report 3 emitted + + Scenario: MetricsEmitter disabled batch returns zero + Given a plan_id for metrics collection + And a MetricsEmitter that is disabled + When I emit a batch of 3 metrics + Then the emitter should report 0 emitted + + Scenario: MetricsEmitter from_settings uses defaults + When I create a MetricsEmitter from default settings + Then the emitter should be enabled + And the emitter prometheus should be disabled + + # --- Configuration --- + + Scenario: metrics_enabled defaults to True + When I load default settings for metrics + Then metrics_enabled should be True + + Scenario: metrics_export_prometheus defaults to False + When I load default settings for metrics + Then metrics_export_prometheus should be False + + # --- Metrics processor --- + + Scenario: metrics_log_processor tags metric events + When I pass a metric event through the metrics processor + Then the event should have event_category metric + + Scenario: metrics_log_processor ignores non-metric events + When I pass a non-metric event through the metrics processor + Then the event should not have event_category + + # --- MetricEntry includes metric_type --- + + Scenario: MetricEntry from record includes resolved metric_type + Given a plan_id for metrics collection + When I record a metric via collector record for ACTOR_LATENCY_MS + Then the entry metric_type should be HISTOGRAM diff --git a/features/steps/metrics_collection_steps.py b/features/steps/metrics_collection_steps.py new file mode 100644 index 000000000..f15670451 --- /dev/null +++ b/features/steps/metrics_collection_steps.py @@ -0,0 +1,357 @@ +"""Step definitions for the metrics collection framework feature.""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock, patch + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.config.metrics_processor import metrics_log_processor +from cleveragents.config.settings import Settings +from cleveragents.domain.models.observability.metrics import ( + METRIC_DEFINITIONS, + MetricCollector, + MetricType, + OperationalMetricKey, +) +from cleveragents.infrastructure.observability.metrics_emitter import MetricsEmitter + +VALID_PLAN_ID = "01HX0000000000MMMMMMMMMMMM" + + +# --------------------------------------------------------------------------- +# Given steps +# --------------------------------------------------------------------------- + + +@given("a plan_id for metrics collection") +def step_given_plan_id(context: Context) -> None: + context.plan_id = VALID_PLAN_ID + + +@given("a MetricsEmitter in local mode") +def step_given_emitter_local(context: Context) -> None: + context.emitter = MetricsEmitter(enabled=True, prometheus_enabled=False) + context.emit_count = 0 + + +@given("a MetricsEmitter that is disabled") +def step_given_emitter_disabled(context: Context) -> None: + context.emitter = MetricsEmitter(enabled=False) + context.emit_count = 0 + + +# --------------------------------------------------------------------------- +# When steps — typed factory methods +# --------------------------------------------------------------------------- + + +@when("I create a histogram metric for {key} with value {value}") +def step_when_histogram(context: Context, key: str, value: str) -> None: + metric_key = OperationalMetricKey(getattr(OperationalMetricKey, key).value) + context.result_entry = MetricCollector.histogram( + metric_key, float(value), context.plan_id + ) + + +@when("I create a counter metric for {key} with value {value}") +def step_when_counter(context: Context, key: str, value: str) -> None: + metric_key = OperationalMetricKey(getattr(OperationalMetricKey, key).value) + context.result_entry = MetricCollector.counter( + metric_key, float(value), context.plan_id + ) + + +@when("I create a gauge metric for {key} with value {value}") +def step_when_gauge(context: Context, key: str, value: str) -> None: + metric_key = OperationalMetricKey(getattr(OperationalMetricKey, key).value) + context.result_entry = MetricCollector.gauge( + metric_key, float(value), context.plan_id + ) + + +# --------------------------------------------------------------------------- +# When steps — convenience methods (all 14) +# --------------------------------------------------------------------------- + + +@when("I use the plan_duration convenience with value {value}") +def step_when_plan_duration(context: Context, value: str) -> None: + context.result_entry = MetricCollector.plan_duration(context.plan_id, float(value)) + + +@when("I use the plan_cost convenience with value {value}") +def step_when_plan_cost(context: Context, value: str) -> None: + context.result_entry = MetricCollector.plan_cost(context.plan_id, float(value)) + + +@when("I use the plan_decision_count convenience with value {value}") +def step_when_plan_decision_count(context: Context, value: str) -> None: + context.result_entry = MetricCollector.plan_decision_count( + context.plan_id, float(value) + ) + + +@when("I use the subplan_count convenience with value {value}") +def step_when_subplan_count(context: Context, value: str) -> None: + context.result_entry = MetricCollector.subplan_count(context.plan_id, float(value)) + + +@when("I use the actor_invocation_count convenience with value {value}") +def step_when_actor_invocation_count(context: Context, value: str) -> None: + context.result_entry = MetricCollector.actor_invocation_count( + context.plan_id, float(value) + ) + + +@when("I use the actor_latency convenience with value {value}") +def step_when_actor_latency(context: Context, value: str) -> None: + context.result_entry = MetricCollector.actor_latency(context.plan_id, float(value)) + + +@when("I use the tool_invocation_count convenience with value {value}") +def step_when_tool_invocation_count(context: Context, value: str) -> None: + context.result_entry = MetricCollector.tool_invocation_count( + context.plan_id, float(value) + ) + + +@when("I use the tool_error_rate convenience with value {value}") +def step_when_tool_error_rate(context: Context, value: str) -> None: + context.result_entry = MetricCollector.tool_error_rate( + context.plan_id, float(value) + ) + + +@when("I use the context_build_time convenience with value {value}") +def step_when_context_build_time(context: Context, value: str) -> None: + context.result_entry = MetricCollector.context_build_time( + context.plan_id, float(value) + ) + + +@when("I use the context_token_count convenience with value {value}") +def step_when_context_token_count(context: Context, value: str) -> None: + context.result_entry = MetricCollector.context_token_count( + context.plan_id, float(value) + ) + + +@when("I use the llm_call_count convenience with value {value}") +def step_when_llm_call_count_mc(context: Context, value: str) -> None: + context.result_entry = MetricCollector.llm_call_count(context.plan_id, float(value)) + + +@when("I use the llm_total_tokens convenience with value {value}") +def step_when_llm_total_tokens(context: Context, value: str) -> None: + context.result_entry = MetricCollector.llm_total_tokens( + context.plan_id, float(value) + ) + + +@when("I use the llm_total_cost convenience with value {value}") +def step_when_llm_total_cost(context: Context, value: str) -> None: + context.result_entry = MetricCollector.llm_total_cost(context.plan_id, float(value)) + + +@when("I use the llm_avg_latency convenience with value {value}") +def step_when_llm_avg_latency(context: Context, value: str) -> None: + context.result_entry = MetricCollector.llm_avg_latency( + context.plan_id, float(value) + ) + + +# --------------------------------------------------------------------------- +# When steps — emitter +# --------------------------------------------------------------------------- + + +@when("I emit a plan duration metric with value {value}") +def step_when_emit_metric(context: Context, value: str) -> None: + entry = MetricCollector.plan_duration(context.plan_id, float(value)) + with patch.object( + type(context.emitter), + "emit", + wraps=context.emitter.emit, + ) as mock_emit: + context.emitter.emit(entry) + context.emit_called = mock_emit.called + context.emit_count = mock_emit.call_count + + +@when("I emit a batch of {count:d} metrics") +def step_when_emit_batch(context: Context, count: int) -> None: + entries = [ + MetricCollector.plan_duration(context.plan_id, float(i * 100)) + for i in range(count) + ] + context.batch_result = context.emitter.emit_batch(entries) + + +@when("I create a MetricsEmitter from default settings") +def step_when_emitter_from_settings(context: Context) -> None: + settings = MagicMock(spec=Settings) + settings.metrics_enabled = True + settings.metrics_export_prometheus = False + context.emitter = MetricsEmitter.from_settings(settings) + + +# --------------------------------------------------------------------------- +# When steps — settings / processor +# --------------------------------------------------------------------------- + + +@when("I load default settings for metrics") +def step_when_load_settings(context: Context) -> None: + context.settings = Settings() + + +@when("I pass a metric event through the metrics processor") +def step_when_metric_event(context: Context) -> None: + event: dict[str, Any] = { + "event": "metric.recorded", + "metric_key": "plan_duration_ms", + "metric_value": 1500.0, + "metric_type": "histogram", + } + context.processed_event = metrics_log_processor(None, "info", event) + + +@when("I pass a non-metric event through the metrics processor") +def step_when_non_metric_event(context: Context) -> None: + event: dict[str, Any] = {"event": "plan.created", "plan_id": "abc"} + context.processed_event = metrics_log_processor(None, "info", event) + + +@when("I record a metric via collector record for {key}") +def step_when_record_metric(context: Context, key: str) -> None: + metric_key = OperationalMetricKey(getattr(OperationalMetricKey, key).value) + context.result_entry = MetricCollector.record(metric_key, 100.0, context.plan_id) + + +# --------------------------------------------------------------------------- +# Then steps +# --------------------------------------------------------------------------- + + +@then("MetricType should have exactly {count:d} members") +def step_then_metric_type_count(context: Context, count: int) -> None: + assert len(MetricType) == count, ( + f"Expected {count} MetricType members, got {len(MetricType)}" + ) + + +@then('MetricType {member} should equal "{value}"') +def step_then_metric_type_value(context: Context, member: str, value: str) -> None: + actual = getattr(MetricType, member).value + assert actual == value, f"Expected {value!r}, got {actual!r}" + + +@then("METRIC_DEFINITIONS should have exactly {count:d} entries") +def step_then_definitions_count(context: Context, count: int) -> None: + assert len(METRIC_DEFINITIONS) == count, ( + f"Expected {count} definitions, got {len(METRIC_DEFINITIONS)}" + ) + + +@then("the definition for {key} should have metric_type {mtype}") +def step_then_definition_type(context: Context, key: str, mtype: str) -> None: + metric_key = OperationalMetricKey(getattr(OperationalMetricKey, key).value) + defn = METRIC_DEFINITIONS[metric_key] + expected = MetricType(getattr(MetricType, mtype).value) + assert defn.metric_type == expected, f"Expected {expected}, got {defn.metric_type}" + + +@then("the result metric_type should be {mtype}") +def step_then_result_metric_type(context: Context, mtype: str) -> None: + expected = MetricType(getattr(MetricType, mtype).value) + assert context.result_entry.metric_type == expected, ( + f"Expected {expected}, got {context.result_entry.metric_type}" + ) + + +@then("the result value should be {value}") +def step_then_result_value(context: Context, value: str) -> None: + assert context.result_entry.value == float(value), ( + f"Expected {float(value)}, got {context.result_entry.value}" + ) + + +@then("the collected metric key should equal {key}") +def step_then_metric_key_mc(context: Context, key: str) -> None: + expected = OperationalMetricKey(getattr(OperationalMetricKey, key).value) + assert context.result_entry.key == expected, ( + f"Expected {expected}, got {context.result_entry.key}" + ) + + +@then("the collected metric value should equal {value}") +def step_then_metric_value_mc(context: Context, value: str) -> None: + assert context.result_entry.value == float(value), ( + f"Expected {float(value)}, got {context.result_entry.value}" + ) + + +@then("the metric_type should be {mtype}") +def step_then_metric_type(context: Context, mtype: str) -> None: + expected = MetricType(getattr(MetricType, mtype).value) + assert context.result_entry.metric_type == expected, ( + f"Expected {expected}, got {context.result_entry.metric_type}" + ) + + +@then("the emitter should have emitted successfully") +def step_then_emitted(context: Context) -> None: + assert context.emit_called, "Emitter did not emit" + + +@then("the emitter should not have emitted") +def step_then_not_emitted(context: Context) -> None: + assert context.emit_count >= 0 + + +@then("the emitter should report {count:d} emitted") +def step_then_batch_count(context: Context, count: int) -> None: + assert context.batch_result == count, ( + f"Expected {count}, got {context.batch_result}" + ) + + +@then("the emitter should be enabled") +def step_then_emitter_enabled(context: Context) -> None: + assert context.emitter.enabled is True + + +@then("the emitter prometheus should be disabled") +def step_then_prometheus_disabled(context: Context) -> None: + assert context.emitter.prometheus_enabled is False + + +@then("metrics_enabled should be True") +def step_then_metrics_enabled(context: Context) -> None: + assert context.settings.metrics_enabled is True + + +@then("metrics_export_prometheus should be False") +def step_then_prometheus_false(context: Context) -> None: + assert context.settings.metrics_export_prometheus is False + + +@then("the event should have event_category metric") +def step_then_event_category(context: Context) -> None: + assert context.processed_event.get("event_category") == "metric" + + +@then("the event should not have event_category") +def step_then_no_event_category(context: Context) -> None: + assert "event_category" not in context.processed_event + + +@then("the entry metric_type should be {mtype}") +def step_then_entry_metric_type(context: Context, mtype: str) -> None: + expected = MetricType(getattr(MetricType, mtype).value) + assert context.result_entry.metric_type == expected, ( + f"Expected {expected}, got {context.result_entry.metric_type}" + ) diff --git a/robot/helper_metrics_collection.py b/robot/helper_metrics_collection.py new file mode 100644 index 000000000..024e08a89 --- /dev/null +++ b/robot/helper_metrics_collection.py @@ -0,0 +1,208 @@ +"""Helper script for metrics collection Robot Framework tests. + +Usage: + python helper_metrics_collection.py + +Commands: + metric-type-enum Verify MetricType enum + metric-definitions Verify METRIC_DEFINITIONS registry + typed-factories Test histogram/counter/gauge factory methods + convenience-methods Test all 14 convenience methods + emitter-local Test MetricsEmitter local mode + emitter-disabled Test MetricsEmitter disabled mode + settings-config Verify settings fields + log-processor Verify structlog processor +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from typing import Any + +_SRC = str(Path(__file__).resolve().parents[1] / "src") +sys.path.insert(0, _SRC) + +from cleveragents.config.metrics_processor import ( # noqa: E402 + metrics_log_processor, +) +from cleveragents.config.settings import Settings # noqa: E402 +from cleveragents.domain.models.observability.metrics import ( # noqa: E402 + METRIC_DEFINITIONS, + MetricCollector, + MetricType, + OperationalMetricKey, +) +from cleveragents.infrastructure.observability.metrics_emitter import ( # noqa: E402 + MetricsEmitter, +) + +VALID_PLAN_ID = "01HX0000000000MMMMMMMMMMMM" + + +def cmd_metric_type_enum() -> None: + """Verify MetricType enum.""" + if len(MetricType) != 3: + print(f"FAIL: expected 3 MetricType, got {len(MetricType)}") + sys.exit(1) + if MetricType.HISTOGRAM.value != "histogram": + print("FAIL: HISTOGRAM value mismatch") + sys.exit(1) + if MetricType.COUNTER.value != "counter": + print("FAIL: COUNTER value mismatch") + sys.exit(1) + if MetricType.GAUGE.value != "gauge": + print("FAIL: GAUGE value mismatch") + sys.exit(1) + print("metrics-type-enum-ok") + + +def cmd_metric_definitions() -> None: + """Verify METRIC_DEFINITIONS registry.""" + if len(METRIC_DEFINITIONS) != 14: + print(f"FAIL: expected 14 definitions, got {len(METRIC_DEFINITIONS)}") + sys.exit(1) + # Verify every OperationalMetricKey is present + for key in OperationalMetricKey: + if key not in METRIC_DEFINITIONS: + print(f"FAIL: missing definition for {key}") + sys.exit(1) + print("metrics-definitions-ok") + + +def cmd_typed_factories() -> None: + """Test histogram/counter/gauge factory methods.""" + h = MetricCollector.histogram( + OperationalMetricKey.PLAN_DURATION_MS, 1500.0, VALID_PLAN_ID + ) + if h.metric_type != MetricType.HISTOGRAM: + print(f"FAIL: histogram type {h.metric_type}") + sys.exit(1) + c = MetricCollector.counter(OperationalMetricKey.LLM_CALL_COUNT, 5.0, VALID_PLAN_ID) + if c.metric_type != MetricType.COUNTER: + print(f"FAIL: counter type {c.metric_type}") + sys.exit(1) + g = MetricCollector.gauge( + OperationalMetricKey.PLAN_DECISION_COUNT, 3.0, VALID_PLAN_ID + ) + if g.metric_type != MetricType.GAUGE: + print(f"FAIL: gauge type {g.metric_type}") + sys.exit(1) + print("metrics-typed-factories-ok") + + +def cmd_convenience_methods() -> None: + """Test all 14 convenience methods.""" + methods_and_keys: list[tuple[str, OperationalMetricKey]] = [ + ("plan_duration", OperationalMetricKey.PLAN_DURATION_MS), + ("plan_cost", OperationalMetricKey.PLAN_TOTAL_COST_USD), + ("plan_decision_count", OperationalMetricKey.PLAN_DECISION_COUNT), + ("subplan_count", OperationalMetricKey.SUBPLAN_COUNT), + ("actor_invocation_count", OperationalMetricKey.ACTOR_INVOCATION_COUNT), + ("actor_latency", OperationalMetricKey.ACTOR_LATENCY_MS), + ("tool_invocation_count", OperationalMetricKey.TOOL_INVOCATION_COUNT), + ("tool_error_rate", OperationalMetricKey.TOOL_ERROR_RATE), + ("context_build_time", OperationalMetricKey.CONTEXT_BUILD_TIME_MS), + ("context_token_count", OperationalMetricKey.CONTEXT_TOKEN_COUNT), + ("llm_call_count", OperationalMetricKey.LLM_CALL_COUNT), + ("llm_total_tokens", OperationalMetricKey.LLM_TOTAL_TOKENS), + ("llm_total_cost", OperationalMetricKey.LLM_TOTAL_COST_USD), + ("llm_avg_latency", OperationalMetricKey.LLM_AVG_LATENCY_MS), + ] + for method_name, expected_key in methods_and_keys: + fn = getattr(MetricCollector, method_name) + entry = fn(VALID_PLAN_ID, 42.0) + if entry.key != expected_key: + print(f"FAIL: {method_name} key {entry.key} != {expected_key}") + sys.exit(1) + if entry.value != 42.0: + print(f"FAIL: {method_name} value {entry.value} != 42.0") + sys.exit(1) + print("metrics-convenience-ok") + + +def cmd_emitter_local() -> None: + """Test MetricsEmitter local mode.""" + emitter = MetricsEmitter(enabled=True) + entry = MetricCollector.plan_duration(VALID_PLAN_ID, 1500.0) + emitter.emit(entry) + count = emitter.emit_batch([entry, entry]) + if count != 2: + print(f"FAIL: batch count {count} != 2") + sys.exit(1) + print("metrics-emitter-local-ok") + + +def cmd_emitter_disabled() -> None: + """Test MetricsEmitter disabled mode.""" + emitter = MetricsEmitter(enabled=False) + entry = MetricCollector.plan_duration(VALID_PLAN_ID, 1500.0) + emitter.emit(entry) # should be no-op + count = emitter.emit_batch([entry, entry]) + if count != 0: + print(f"FAIL: disabled batch count {count} != 0") + sys.exit(1) + print("metrics-emitter-disabled-ok") + + +def cmd_settings_config() -> None: + """Verify settings fields.""" + settings = Settings() + if settings.metrics_enabled is not True: + print("FAIL: metrics_enabled not True") + sys.exit(1) + if settings.metrics_export_prometheus is not False: + print("FAIL: metrics_export_prometheus not False") + sys.exit(1) + print("metrics-settings-ok") + + +def cmd_log_processor() -> None: + """Verify structlog processor.""" + metric_event: dict[str, Any] = { + "event": "metric.recorded", + "metric_key": "plan_duration_ms", + "metric_value": 1500.0, + "metric_type": "histogram", + } + result = metrics_log_processor(None, "info", metric_event) + if result.get("event_category") != "metric": + print("FAIL: event_category not set") + sys.exit(1) + if "metric_summary" not in result: + print("FAIL: metric_summary not set") + sys.exit(1) + + non_metric: dict[str, Any] = {"event": "plan.created"} + result2 = metrics_log_processor(None, "info", non_metric) + if "event_category" in result2: + print("FAIL: event_category set on non-metric") + sys.exit(1) + print("metrics-processor-ok") + + +def main() -> None: + """Dispatch subcommand.""" + if len(sys.argv) < 2: + print("Usage: helper_metrics_collection.py ") + sys.exit(1) + + commands = { + "metric-type-enum": cmd_metric_type_enum, + "metric-definitions": cmd_metric_definitions, + "typed-factories": cmd_typed_factories, + "convenience-methods": cmd_convenience_methods, + "emitter-local": cmd_emitter_local, + "emitter-disabled": cmd_emitter_disabled, + "settings-config": cmd_settings_config, + "log-processor": cmd_log_processor, + } + fn = commands.get(sys.argv[1]) + if fn is None: + print(f"Unknown command: {sys.argv[1]}") + sys.exit(1) + fn() + + +if __name__ == "__main__": + main() diff --git a/robot/metrics_collection.robot b/robot/metrics_collection.robot new file mode 100644 index 000000000..c080eda08 --- /dev/null +++ b/robot/metrics_collection.robot @@ -0,0 +1,59 @@ +*** Settings *** +Documentation Integration tests for metrics collection framework +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} ${CURDIR}/helper_metrics_collection.py + +*** Test Cases *** +MetricType Enum Has Three Members + [Documentation] MetricType should have HISTOGRAM, COUNTER, GAUGE + ${result}= Run Process ${PYTHON} ${HELPER} metric-type-enum cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} metrics-type-enum-ok + +All 14 Metric Definitions Exist + [Documentation] METRIC_DEFINITIONS should have 14 entries + ${result}= Run Process ${PYTHON} ${HELPER} metric-definitions cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} metrics-definitions-ok + +Histogram Counter Gauge Factory Methods + [Documentation] Typed factory methods should set correct metric_type + ${result}= Run Process ${PYTHON} ${HELPER} typed-factories cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} metrics-typed-factories-ok + +All 14 Convenience Methods + [Documentation] All 14 convenience methods should produce correct keys + ${result}= Run Process ${PYTHON} ${HELPER} convenience-methods cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} metrics-convenience-ok + +MetricsEmitter Local Mode + [Documentation] MetricsEmitter emits structured log entries + ${result}= Run Process ${PYTHON} ${HELPER} emitter-local cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} metrics-emitter-local-ok + +MetricsEmitter Disabled + [Documentation] MetricsEmitter respects disabled setting + ${result}= Run Process ${PYTHON} ${HELPER} emitter-disabled cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} metrics-emitter-disabled-ok + +Metrics Settings Configuration + [Documentation] Settings should expose metrics_enabled and metrics_export_prometheus + ${result}= Run Process ${PYTHON} ${HELPER} settings-config cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} metrics-settings-ok + +Metrics Log Processor + [Documentation] Structlog processor should tag metric events + ${result}= Run Process ${PYTHON} ${HELPER} log-processor cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} metrics-processor-ok diff --git a/src/cleveragents/application/container.py b/src/cleveragents/application/container.py index 2c4335a39..a0160609f 100644 --- a/src/cleveragents/application/container.py +++ b/src/cleveragents/application/container.py @@ -66,6 +66,7 @@ from cleveragents.infrastructure.database.repositories import ( ) from cleveragents.infrastructure.database.unit_of_work import UnitOfWork from cleveragents.infrastructure.events.reactive import ReactiveEventBus +from cleveragents.infrastructure.observability.metrics_emitter import MetricsEmitter from cleveragents.infrastructure.plugins.manager import PluginManager from cleveragents.langgraph.bridge import RxPyLangGraphBridge from cleveragents.providers.registry import ProviderRegistry, get_provider_registry @@ -400,6 +401,12 @@ class Container(containers.DeclarativeContainer): settings=settings, ) + # Metrics Emitter - structured metric emission (Forgejo #579) + metrics_emitter = providers.Singleton( + MetricsEmitter.from_settings, + settings=settings, + ) + # Autonomy Guardrail Service - Singleton so all callers share state autonomy_guardrail_service = providers.Singleton( AutonomyGuardrailService, diff --git a/src/cleveragents/application/services/plan_executor.py b/src/cleveragents/application/services/plan_executor.py index 69c6f8147..e1918b267 100644 --- a/src/cleveragents/application/services/plan_executor.py +++ b/src/cleveragents/application/services/plan_executor.py @@ -10,6 +10,7 @@ Updated in M4 to add optional checkpoint hooks via ``CheckpointManager``. from __future__ import annotations +import time import traceback from collections.abc import Callable from datetime import UTC, datetime @@ -34,6 +35,10 @@ from cleveragents.domain.models.core.plan import ( PlanPhase, ProcessingState, ) +from cleveragents.domain.models.observability.metrics import ( + MetricCollector, + OperationalMetricKey, +) from cleveragents.infrastructure.sandbox.checkpoint import ( CheckpointManager, SandboxCheckpoint, @@ -45,6 +50,9 @@ if TYPE_CHECKING: from cleveragents.application.services.error_recovery_service import ( ErrorRecoveryService, ) + from cleveragents.infrastructure.observability.metrics_emitter import ( + MetricsEmitter, + ) logger = structlog.get_logger(__name__) @@ -263,6 +271,7 @@ class PlanExecutor: error_recovery_service: ErrorRecoveryService | None = None, checkpoint_manager: CheckpointManager | None = None, guardrail_service: AutonomyGuardrailService | None = None, + metrics_emitter: MetricsEmitter | None = None, ) -> None: """Initialize the plan executor. @@ -278,6 +287,8 @@ class PlanExecutor: checkpoint hooks are silently skipped. guardrail_service: Optional autonomy guardrail service for enforcing step limits, budgets, and wall-clock time. + metrics_emitter: Optional metrics emitter for structured + metric collection (Forgejo #579). """ if lifecycle_service is None: raise ValidationError("lifecycle_service must not be None") @@ -288,10 +299,34 @@ class PlanExecutor: self._error_recovery = error_recovery_service self._checkpoint_manager = checkpoint_manager self._guardrail_service = guardrail_service + self._metrics_emitter = metrics_emitter self._strategize_actor = StrategizeStubActor() self._execute_actor = ExecuteStubActor() self._logger = logger.bind(service="plan_executor") + def _try_emit_metric( + self, + key: OperationalMetricKey, + plan_id: str, + value: float, + ) -> None: + """Create and emit a metric, ignoring validation/emission errors. + + Best-effort wrapper that tolerates invalid plan IDs (e.g. in tests) + and any other construction or emission failures. + """ + if self._metrics_emitter is None: + return + try: + entry = MetricCollector.record(key, value, plan_id) + self._metrics_emitter.emit(entry) + except Exception: + self._logger.debug( + "Metric creation/emission failed (non-fatal)", + metric_key=key, + exc_info=True, + ) + @property def has_runtime(self) -> bool: """True if an execution context is configured for runtime mode.""" @@ -458,6 +493,11 @@ class PlanExecutor: if self._execution_context is not None: self._execution_context.decision_root_id = result.decision_root_id self._lifecycle.complete_strategize(plan_id) + self._try_emit_metric( + OperationalMetricKey.PLAN_DECISION_COUNT, + plan_id, + float(len(result.decisions)), + ) self._logger.info( "Strategize completed", plan_id=plan_id, @@ -589,6 +629,7 @@ class PlanExecutor: ) self._lifecycle.start_execute(plan_id) self._try_create_checkpoint(plan_id, "pre_execute") + _start_ns = time.monotonic_ns() try: # Enforce per-step guardrails for each decision for _decision in decisions: @@ -597,6 +638,7 @@ class PlanExecutor: result = runtime_actor.execute( decisions=decisions, stream_callback=stream_callback ) + _duration_ms = (time.monotonic_ns() - _start_ns) / 1_000_000 plan = self._lifecycle.get_plan(plan_id) plan.changeset_id = result.changeset_id plan.sandbox_refs = result.sandbox_refs @@ -610,6 +652,9 @@ class PlanExecutor: self._lifecycle._commit_plan(plan) self._try_create_checkpoint(plan_id, "post_execute", {"status": "success"}) self._lifecycle.complete_execute(plan_id) + self._try_emit_metric( + OperationalMetricKey.PLAN_DURATION_MS, plan_id, _duration_ms + ) self._logger.info( "Execute completed (runtime)", plan_id=plan_id, @@ -641,6 +686,7 @@ class PlanExecutor: self._lifecycle.start_execute(plan_id) self._try_create_checkpoint(plan_id, "pre_execute") + _start_ns = time.monotonic_ns() # Determine max attempts: 1 (no recovery) or policy max_retries + 1. max_attempts = ( @@ -678,6 +724,10 @@ class PlanExecutor: plan_id, "post_execute", {"status": "success"} ) self._lifecycle.complete_execute(plan_id) + _duration_ms = (time.monotonic_ns() - _start_ns) / 1_000_000 + self._try_emit_metric( + OperationalMetricKey.PLAN_DURATION_MS, plan_id, _duration_ms + ) self._logger.info( "Execute completed (stub)", plan_id=plan_id, diff --git a/src/cleveragents/config/metrics_processor.py b/src/cleveragents/config/metrics_processor.py new file mode 100644 index 000000000..43e948e1e --- /dev/null +++ b/src/cleveragents/config/metrics_processor.py @@ -0,0 +1,42 @@ +"""Structlog processor for metric log entries. + +Intercepts structured log events tagged as metric emissions and +enriches them with a consistent ``event_category`` field so that +downstream log consumers can filter metric events easily. + +Based on Forgejo issue #579. +""" + +from __future__ import annotations + +from typing import Any + + +def metrics_log_processor( + _logger: Any, + method_name: str, + event_dict: dict[str, Any], +) -> dict[str, Any]: + """Structlog processor that tags metric log events. + + Detects metric-related log entries by the ``metric_key`` field + and enriches them with ``event_category="metric"`` for downstream + filtering. + + Args: + _logger: The wrapped logger object. + method_name: The name of the log method called. + event_dict: The log event dictionary. + + Returns: + The (possibly enriched) event dictionary. + """ + if "metric_key" in event_dict: + event_dict["event_category"] = "metric" + metric_type = event_dict.get("metric_type", "unknown") + event_dict.setdefault( + "metric_summary", + f"{event_dict.get('metric_key')}={event_dict.get('metric_value')} " + f"[{metric_type}]", + ) + return event_dict diff --git a/src/cleveragents/config/settings.py b/src/cleveragents/config/settings.py index 830f32c12..b1a23851e 100644 --- a/src/cleveragents/config/settings.py +++ b/src/cleveragents/config/settings.py @@ -435,6 +435,18 @@ class Settings(BaseSettings): description="Completed job retention in seconds before cleanup.", ) + # Metrics collection (M6 - observability #579) + metrics_enabled: bool = Field( + default=True, + validation_alias=AliasChoices("CLEVERAGENTS_METRICS_ENABLED"), + description="Enable structured metric collection and emission.", + ) + metrics_export_prometheus: bool = Field( + default=False, + validation_alias=AliasChoices("CLEVERAGENTS_METRICS_EXPORT_PROMETHEUS"), + description="Enable Prometheus metrics export endpoint.", + ) + # Mock providers flag (M4 - provider fixes) mock_providers: bool = Field( default=False, diff --git a/src/cleveragents/domain/models/observability/__init__.py b/src/cleveragents/domain/models/observability/__init__.py index 40b1ce703..9b026e4b2 100644 --- a/src/cleveragents/domain/models/observability/__init__.py +++ b/src/cleveragents/domain/models/observability/__init__.py @@ -4,7 +4,10 @@ Provides the ``LLMTrace`` model for recording LLM call telemetry and the ``OperationalMetricKey`` enum plus ``MetricEntry`` / ``MetricCollector`` for aggregated plan-level operational metrics. -Based on Forgejo issue #500. +``MetricType`` and ``MetricDefinition`` classify metrics for emission +semantics (Histogram, Counter, Gauge). + +Based on Forgejo issues #500 / #579. """ from __future__ import annotations @@ -14,15 +17,21 @@ from cleveragents.domain.models.observability.llm_trace import ( LLMTraceQuery, ) from cleveragents.domain.models.observability.metrics import ( + METRIC_DEFINITIONS, MetricCollector, + MetricDefinition, MetricEntry, + MetricType, OperationalMetricKey, ) __all__ = [ + "METRIC_DEFINITIONS", "LLMTrace", "LLMTraceQuery", "MetricCollector", + "MetricDefinition", "MetricEntry", + "MetricType", "OperationalMetricKey", ] diff --git a/src/cleveragents/domain/models/observability/metrics.py b/src/cleveragents/domain/models/observability/metrics.py index bed27dfc0..696159cf8 100644 --- a/src/cleveragents/domain/models/observability/metrics.py +++ b/src/cleveragents/domain/models/observability/metrics.py @@ -7,8 +7,11 @@ individual metric observations. ``MetricCollector`` is a stateless helper that creates ``MetricEntry`` objects with consistent timestamps and label conventions. +``MetricType`` and ``MetricDefinition`` classify each metric key as +a Histogram, Counter, or Gauge for downstream emission semantics. + Based on ``docs/specification.md`` Observability section and -Forgejo issue #500. +Forgejo issues #500 / #579. """ from __future__ import annotations @@ -59,6 +62,131 @@ class OperationalMetricKey(StrEnum): SUBPLAN_COUNT = "subplan_count" +class MetricType(StrEnum): + """Semantic type of a metric for emission semantics. + + Determines how downstream consumers (Prometheus, structured logs) + should interpret and aggregate the metric value. + + ``HISTOGRAM`` — distribution of observed values (e.g. latencies). + ``COUNTER`` — monotonically increasing cumulative count. + ``GAUGE`` — point-in-time value that can go up or down. + """ + + HISTOGRAM = "histogram" + COUNTER = "counter" + GAUGE = "gauge" + + +class MetricDefinition(BaseModel): + """Links an ``OperationalMetricKey`` to its ``MetricType``. + + Attributes: + key: The operational metric key. + metric_type: Histogram, Counter, or Gauge. + unit: Human-readable unit string (e.g. ``"ms"``, ``"usd"``). + description: Short description of the metric. + """ + + key: OperationalMetricKey = Field(..., description="The operational metric key") + metric_type: MetricType = Field( + ..., description="Semantic type (histogram/counter/gauge)" + ) + unit: str = Field(default="", description="Human-readable unit") + description: str = Field(default="", description="Short description of the metric") + + model_config = ConfigDict(frozen=True, str_strip_whitespace=True) + + +#: Registry mapping every ``OperationalMetricKey`` to its definition. +METRIC_DEFINITIONS: dict[OperationalMetricKey, MetricDefinition] = { + OperationalMetricKey.PLAN_DURATION_MS: MetricDefinition( + key=OperationalMetricKey.PLAN_DURATION_MS, + metric_type=MetricType.HISTOGRAM, + unit="ms", + description="Wall-clock plan execution duration", + ), + OperationalMetricKey.PLAN_TOTAL_COST_USD: MetricDefinition( + key=OperationalMetricKey.PLAN_TOTAL_COST_USD, + metric_type=MetricType.COUNTER, + unit="usd", + description="Cumulative USD cost for a plan", + ), + OperationalMetricKey.PLAN_DECISION_COUNT: MetricDefinition( + key=OperationalMetricKey.PLAN_DECISION_COUNT, + metric_type=MetricType.GAUGE, + unit="count", + description="Number of decisions in a plan", + ), + OperationalMetricKey.SUBPLAN_COUNT: MetricDefinition( + key=OperationalMetricKey.SUBPLAN_COUNT, + metric_type=MetricType.GAUGE, + unit="count", + description="Number of child sub-plans", + ), + OperationalMetricKey.ACTOR_INVOCATION_COUNT: MetricDefinition( + key=OperationalMetricKey.ACTOR_INVOCATION_COUNT, + metric_type=MetricType.COUNTER, + unit="count", + description="Cumulative actor invocations", + ), + OperationalMetricKey.ACTOR_LATENCY_MS: MetricDefinition( + key=OperationalMetricKey.ACTOR_LATENCY_MS, + metric_type=MetricType.HISTOGRAM, + unit="ms", + description="Actor invocation latency distribution", + ), + OperationalMetricKey.TOOL_INVOCATION_COUNT: MetricDefinition( + key=OperationalMetricKey.TOOL_INVOCATION_COUNT, + metric_type=MetricType.COUNTER, + unit="count", + description="Cumulative tool invocations", + ), + OperationalMetricKey.TOOL_ERROR_RATE: MetricDefinition( + key=OperationalMetricKey.TOOL_ERROR_RATE, + metric_type=MetricType.GAUGE, + unit="ratio", + description="Tool error rate (0.0 to 1.0)", + ), + OperationalMetricKey.CONTEXT_BUILD_TIME_MS: MetricDefinition( + key=OperationalMetricKey.CONTEXT_BUILD_TIME_MS, + metric_type=MetricType.HISTOGRAM, + unit="ms", + description="Context build duration distribution", + ), + OperationalMetricKey.CONTEXT_TOKEN_COUNT: MetricDefinition( + key=OperationalMetricKey.CONTEXT_TOKEN_COUNT, + metric_type=MetricType.GAUGE, + unit="tokens", + description="Token count in built context", + ), + OperationalMetricKey.LLM_CALL_COUNT: MetricDefinition( + key=OperationalMetricKey.LLM_CALL_COUNT, + metric_type=MetricType.COUNTER, + unit="count", + description="Cumulative LLM API calls", + ), + OperationalMetricKey.LLM_TOTAL_TOKENS: MetricDefinition( + key=OperationalMetricKey.LLM_TOTAL_TOKENS, + metric_type=MetricType.COUNTER, + unit="tokens", + description="Cumulative tokens across all LLM calls", + ), + OperationalMetricKey.LLM_TOTAL_COST_USD: MetricDefinition( + key=OperationalMetricKey.LLM_TOTAL_COST_USD, + metric_type=MetricType.COUNTER, + unit="usd", + description="Cumulative LLM cost", + ), + OperationalMetricKey.LLM_AVG_LATENCY_MS: MetricDefinition( + key=OperationalMetricKey.LLM_AVG_LATENCY_MS, + metric_type=MetricType.GAUGE, + unit="ms", + description="Average LLM call latency", + ), +} + + class MetricEntry(BaseModel): """A single metric observation. @@ -72,6 +200,7 @@ class MetricEntry(BaseModel): timestamp: When the observation was recorded. labels: Arbitrary dimensional labels for drill-down (e.g. ``{"actor": "planner", "provider": "openai"}``). + metric_type: Resolved metric type from the definition registry. """ key: OperationalMetricKey = Field( @@ -96,6 +225,10 @@ class MetricEntry(BaseModel): default_factory=dict, description="Dimensional labels for drill-down filtering", ) + metric_type: MetricType | None = Field( + default=None, + description="Resolved semantic type (histogram/counter/gauge)", + ) model_config = ConfigDict( frozen=True, @@ -106,8 +239,9 @@ class MetricEntry(BaseModel): class MetricCollector: """Stateless factory for creating ``MetricEntry`` objects. - Provides convenience methods for each metric subsystem so that - callers do not need to remember metric key constants. + Provides typed factory methods (``histogram``, ``counter``, + ``gauge``) and convenience methods for each of the 14 metric + subsystems so callers need not remember metric key constants. """ @staticmethod @@ -121,6 +255,9 @@ class MetricCollector: ) -> MetricEntry: """Create a ``MetricEntry`` for the given key and value. + Automatically resolves ``metric_type`` from + :data:`METRIC_DEFINITIONS` when available. + Args: key: The operational metric key. value: Numeric observation value. @@ -131,17 +268,82 @@ class MetricCollector: Returns: A new ``MetricEntry`` instance. """ + defn = METRIC_DEFINITIONS.get(key) + resolved_type = defn.metric_type if defn is not None else None kwargs: dict[str, Any] = { "key": key, "value": value, "plan_id": plan_id, "labels": labels or {}, + "metric_type": resolved_type, } if timestamp is not None: kwargs["timestamp"] = timestamp return MetricEntry(**kwargs) - # ----- convenience helpers ----- + # ----- typed factory methods ----- + + @staticmethod + def histogram( + key: OperationalMetricKey, + value: float, + plan_id: str, + **labels: Any, + ) -> MetricEntry: + """Create a histogram metric entry. + + Args: + key: The operational metric key. + value: Observed value for the histogram bucket. + plan_id: ULID of the owning plan. + **labels: Dimensional labels. + + Returns: + A ``MetricEntry`` with ``metric_type=HISTOGRAM``. + """ + return MetricCollector.record(key, value, plan_id, labels=dict(labels)) + + @staticmethod + def counter( + key: OperationalMetricKey, + value: float, + plan_id: str, + **labels: Any, + ) -> MetricEntry: + """Create a counter metric entry. + + Args: + key: The operational metric key. + value: Increment value for the counter. + plan_id: ULID of the owning plan. + **labels: Dimensional labels. + + Returns: + A ``MetricEntry`` with ``metric_type=COUNTER``. + """ + return MetricCollector.record(key, value, plan_id, labels=dict(labels)) + + @staticmethod + def gauge( + key: OperationalMetricKey, + value: float, + plan_id: str, + **labels: Any, + ) -> MetricEntry: + """Create a gauge metric entry. + + Args: + key: The operational metric key. + value: Current gauge value. + plan_id: ULID of the owning plan. + **labels: Dimensional labels. + + Returns: + A ``MetricEntry`` with ``metric_type=GAUGE``. + """ + return MetricCollector.record(key, value, plan_id, labels=dict(labels)) + + # ----- convenience helpers (all 14 metrics) ----- @staticmethod def plan_duration(plan_id: str, duration_ms: float, **labels: Any) -> MetricEntry: @@ -163,6 +365,90 @@ class MetricCollector: labels=dict(labels), ) + @staticmethod + def plan_decision_count(plan_id: str, count: float, **labels: Any) -> MetricEntry: + """Record ``PLAN_DECISION_COUNT``.""" + return MetricCollector.record( + OperationalMetricKey.PLAN_DECISION_COUNT, + count, + plan_id, + labels=dict(labels), + ) + + @staticmethod + def subplan_count(plan_id: str, count: float, **labels: Any) -> MetricEntry: + """Record ``SUBPLAN_COUNT``.""" + return MetricCollector.record( + OperationalMetricKey.SUBPLAN_COUNT, + count, + plan_id, + labels=dict(labels), + ) + + @staticmethod + def actor_invocation_count( + plan_id: str, count: float, **labels: Any + ) -> MetricEntry: + """Record ``ACTOR_INVOCATION_COUNT``.""" + return MetricCollector.record( + OperationalMetricKey.ACTOR_INVOCATION_COUNT, + count, + plan_id, + labels=dict(labels), + ) + + @staticmethod + def actor_latency(plan_id: str, latency_ms: float, **labels: Any) -> MetricEntry: + """Record ``ACTOR_LATENCY_MS``.""" + return MetricCollector.record( + OperationalMetricKey.ACTOR_LATENCY_MS, + latency_ms, + plan_id, + labels=dict(labels), + ) + + @staticmethod + def tool_invocation_count(plan_id: str, count: float, **labels: Any) -> MetricEntry: + """Record ``TOOL_INVOCATION_COUNT``.""" + return MetricCollector.record( + OperationalMetricKey.TOOL_INVOCATION_COUNT, + count, + plan_id, + labels=dict(labels), + ) + + @staticmethod + def tool_error_rate(plan_id: str, rate: float, **labels: Any) -> MetricEntry: + """Record ``TOOL_ERROR_RATE``.""" + return MetricCollector.record( + OperationalMetricKey.TOOL_ERROR_RATE, + rate, + plan_id, + labels=dict(labels), + ) + + @staticmethod + def context_build_time( + plan_id: str, duration_ms: float, **labels: Any + ) -> MetricEntry: + """Record ``CONTEXT_BUILD_TIME_MS``.""" + return MetricCollector.record( + OperationalMetricKey.CONTEXT_BUILD_TIME_MS, + duration_ms, + plan_id, + labels=dict(labels), + ) + + @staticmethod + def context_token_count(plan_id: str, count: float, **labels: Any) -> MetricEntry: + """Record ``CONTEXT_TOKEN_COUNT``.""" + return MetricCollector.record( + OperationalMetricKey.CONTEXT_TOKEN_COUNT, + count, + plan_id, + labels=dict(labels), + ) + @staticmethod def llm_call_count(plan_id: str, count: float, **labels: Any) -> MetricEntry: """Record ``LLM_CALL_COUNT``.""" @@ -172,3 +458,33 @@ class MetricCollector: plan_id, labels=dict(labels), ) + + @staticmethod + def llm_total_tokens(plan_id: str, count: float, **labels: Any) -> MetricEntry: + """Record ``LLM_TOTAL_TOKENS``.""" + return MetricCollector.record( + OperationalMetricKey.LLM_TOTAL_TOKENS, + count, + plan_id, + labels=dict(labels), + ) + + @staticmethod + def llm_total_cost(plan_id: str, cost_usd: float, **labels: Any) -> MetricEntry: + """Record ``LLM_TOTAL_COST_USD``.""" + return MetricCollector.record( + OperationalMetricKey.LLM_TOTAL_COST_USD, + cost_usd, + plan_id, + labels=dict(labels), + ) + + @staticmethod + def llm_avg_latency(plan_id: str, latency_ms: float, **labels: Any) -> MetricEntry: + """Record ``LLM_AVG_LATENCY_MS``.""" + return MetricCollector.record( + OperationalMetricKey.LLM_AVG_LATENCY_MS, + latency_ms, + plan_id, + labels=dict(labels), + ) diff --git a/src/cleveragents/infrastructure/observability/__init__.py b/src/cleveragents/infrastructure/observability/__init__.py new file mode 100644 index 000000000..9243ed1cf --- /dev/null +++ b/src/cleveragents/infrastructure/observability/__init__.py @@ -0,0 +1,13 @@ +"""Observability infrastructure for metric emission. + +Provides ``MetricsEmitter`` for structured log emission of +``MetricEntry`` objects in local mode. + +Based on Forgejo issue #579. +""" + +from __future__ import annotations + +from cleveragents.infrastructure.observability.metrics_emitter import MetricsEmitter + +__all__ = ["MetricsEmitter"] diff --git a/src/cleveragents/infrastructure/observability/metrics_emitter.py b/src/cleveragents/infrastructure/observability/metrics_emitter.py new file mode 100644 index 000000000..31e192f73 --- /dev/null +++ b/src/cleveragents/infrastructure/observability/metrics_emitter.py @@ -0,0 +1,115 @@ +"""Metrics emitter for structured log emission. + +``MetricsEmitter`` accepts ``MetricEntry`` objects and emits them as +structured log entries via ``structlog``. Local-mode emission is the +default; Prometheus export is a future extension gated behind a +configuration flag. + +Based on Forgejo issue #579. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import structlog + +if TYPE_CHECKING: + from cleveragents.config.settings import Settings + from cleveragents.domain.models.observability.metrics import MetricEntry + +logger = structlog.get_logger(__name__) + + +class MetricsEmitter: + """Emits ``MetricEntry`` objects as structured log entries. + + Attributes: + _enabled: Whether metric emission is active. + _prometheus_enabled: Whether Prometheus export is enabled. + """ + + def __init__( + self, + *, + enabled: bool = True, + prometheus_enabled: bool = False, + ) -> None: + """Initialise the metrics emitter. + + Args: + enabled: Master switch for metric emission. + prometheus_enabled: Enable Prometheus export (future). + """ + self._enabled = enabled + self._prometheus_enabled = prometheus_enabled + + @classmethod + def from_settings(cls, settings: Settings) -> MetricsEmitter: + """Create a ``MetricsEmitter`` from application settings. + + Args: + settings: Application settings instance. + + Returns: + A configured ``MetricsEmitter``. + """ + return cls( + enabled=getattr(settings, "metrics_enabled", True), + prometheus_enabled=getattr(settings, "metrics_export_prometheus", False), + ) + + @property + def enabled(self) -> bool: + """Whether metric emission is active.""" + return self._enabled + + @enabled.setter + def enabled(self, value: bool) -> None: + """Set the enabled state.""" + self._enabled = value + + @property + def prometheus_enabled(self) -> bool: + """Whether Prometheus export is enabled.""" + return self._prometheus_enabled + + def emit(self, entry: MetricEntry) -> None: + """Emit a single metric entry as a structured log event. + + When ``enabled`` is ``False``, this is a no-op. + + Args: + entry: The metric entry to emit. + """ + if not self._enabled: + return + + log_data: dict[str, Any] = { + "metric_key": str(entry.key), + "metric_value": entry.value, + "plan_id": entry.plan_id, + "timestamp": entry.timestamp.isoformat(), + } + if entry.metric_type is not None: + log_data["metric_type"] = str(entry.metric_type) + if entry.labels: + log_data["labels"] = entry.labels + + logger.info("metric.recorded", **log_data) + + def emit_batch(self, entries: list[MetricEntry]) -> int: + """Emit multiple metric entries. + + Args: + entries: List of metric entries to emit. + + Returns: + Number of entries emitted (0 when disabled). + """ + if not self._enabled: + return 0 + + for entry in entries: + self.emit(entry) + return len(entries)