diff --git a/helm/kagent/values.yaml b/helm/kagent/values.yaml index b30e9338b4..277c74f362 100644 --- a/helm/kagent/values.yaml +++ b/helm/kagent/values.yaml @@ -854,6 +854,14 @@ otel: endpoint: "" timeout: 15000 # milliseconds insecure: true + metrics: + enabled: false + exporter: + otlp: + endpoint: "" + protocol: "grpc" + timeout: 15000 # milliseconds + insecure: true # ============================================================================== # EXTRA OBJECTS diff --git a/python/packages/kagent-core/src/kagent/core/tracing/_utils.py b/python/packages/kagent-core/src/kagent/core/tracing/_utils.py index 2b2f78f791..89f37f6f2c 100644 --- a/python/packages/kagent-core/src/kagent/core/tracing/_utils.py +++ b/python/packages/kagent-core/src/kagent/core/tracing/_utils.py @@ -3,12 +3,14 @@ import os from fastapi import FastAPI -from opentelemetry import _logs, trace +from opentelemetry import _logs, metrics, trace from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor from opentelemetry.instrumentation.openai import OpenAIInstrumentor from opentelemetry.sdk._logs import LoggerProvider from opentelemetry.sdk._logs.export import BatchLogRecordProcessor +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor @@ -48,6 +50,17 @@ def _create_log_exporter(**kwargs): return OTLPLogExporter(**kwargs) +def _create_metric_exporter(**kwargs): + """Create an OTLPMetricExporter using the protocol from env vars.""" + protocol = _resolve_otlp_protocol("METRICS") + if protocol == "http/protobuf": + from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter + else: + from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter + logging.info("Using %s protocol for metric exporter", protocol) + return OTLPMetricExporter(**kwargs) + + def _resolve_otlp_timeout_seconds(signal: str) -> float: """ Resolve OTLP timeout env vars (milliseconds) into seconds for exporters. @@ -125,26 +138,35 @@ def _resolve_flush_timeout_millis() -> int: def force_flush(timeout_millis: int | None = None) -> None: - """Export any spans still buffered in the tracer provider's batch processor. + """Export any buffered telemetry (traces and metrics) before suspension. Call before a response completes when the process may be suspended right afterwards: Agent Substrate checkpoints the actor as soon as the A2A - response body closes, so unexported spans stay frozen in the snapshot - until the session's next resume (or forever, for a session's last - message). No-op when the provider has no force_flush (tracing disabled). - The timeout defaults to 3000ms, configurable via - KAGENT_TRACE_FLUSH_TIMEOUT_MS. + response body closes, so unexported recording stays frozen in the snapshot + until the session's next resume (or forever, for a session's last message). + No-op for each provider that has no force_flush (signal disabled). The + timeout defaults to 3000ms, configurable via KAGENT_TRACE_FLUSH_TIMEOUT_MS. """ if timeout_millis is None: timeout_millis = _resolve_flush_timeout_millis() + provider = trace.get_tracer_provider() flush = getattr(provider, "force_flush", None) - if flush is None: - return - try: - flush(timeout_millis) - except Exception: - logging.warning("Failed to flush pending spans", exc_info=True) + if flush is not None: + try: + flush(timeout_millis) + except Exception: + logging.warning("Failed to flush pending spans", exc_info=True) + + # A periodic metric reader may hold buffered points that are never exported + # before suspension; flush it whenever traces are flushed. + metric_provider = metrics.get_meter_provider() + flush = getattr(metric_provider, "force_flush", None) + if flush is not None: + try: + flush(timeout_millis) + except Exception: + logging.warning("Failed to flush pending metrics", exc_info=True) # High-frequency probe endpoints with nothing worth flushing. @@ -218,10 +240,12 @@ def configure( fastapi_app: FastAPI | None = None, instrument_openai_client: bool = True, ): - """Configure OpenTelemetry tracing and logging for this service. + """Configure OpenTelemetry tracing, logging, and metrics for this service. - This sets up OpenTelemetry providers and exporters for tracing and logging, - using environment variables to determine whether each is enabled. + This sets up OpenTelemetry providers and exporters for tracing, logging, + and metrics, using environment variables to determine whether each is + enabled (OTEL_TRACING_ENABLED, OTEL_LOGGING_ENABLED, OTEL_METRICS_ENABLED). + Metrics are off by default. Args: name: service name to report to OpenTelemetry (used as ``service.name``). Default is "kagent". @@ -235,6 +259,7 @@ def configure( """ tracing_enabled = os.getenv("OTEL_TRACING_ENABLED", "false").lower() == "true" logging_enabled = os.getenv("OTEL_LOGGING_ENABLED", "false").lower() == "true" + metrics_enabled = os.getenv("OTEL_METRICS_ENABLED", "false").lower() == "true" # Resource.create merges in OTEL_RESOURCE_ATTRIBUTES and the telemetry.sdk.* # attributes; the bare constructor drops both, so deployment.environment.name, @@ -329,3 +354,24 @@ def configure( _instrument_anthropic() _instrument_google_generativeai() # Neither signal enabled: skip GenAI instrumentation so telemetry has no runtime side effects. + # Configure metrics if enabled (independent of tracing/logging: builds a + # MeterProvider so google-adk's built-in GenAI metrics under the + # gcp.vertex.agent scope are exported instead of silently discarded). + if metrics_enabled: + logging.info("Enabling metrics") + # Check standard OTEL env vars: signal-specific endpoint first, then general endpoint + metric_endpoint = os.getenv("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT") or os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT") + metric_timeout_seconds = _resolve_otlp_timeout_seconds("METRICS") + logging.info("Metric endpoint: %s", metric_endpoint or "") + + if metric_endpoint: + metric_reader = PeriodicExportingMetricReader( + _create_metric_exporter(endpoint=metric_endpoint, timeout=metric_timeout_seconds) + ) + else: + metric_reader = PeriodicExportingMetricReader( + _create_metric_exporter(timeout=metric_timeout_seconds) + ) + meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader]) + metrics.set_meter_provider(meter_provider) + logging.info("Metric provider configured with OTLP") diff --git a/python/packages/kagent-core/tests/test_tracing_configure.py b/python/packages/kagent-core/tests/test_tracing_configure.py index 3ad0b3e3f4..ac8029d992 100644 --- a/python/packages/kagent-core/tests/test_tracing_configure.py +++ b/python/packages/kagent-core/tests/test_tracing_configure.py @@ -417,3 +417,135 @@ def test_post_response_flush_exports_server_span(monkeypatch): names = [span.name for span in exporter.get_finished_spans()] assert any("POST" in name for name in names), f"server span not exported by flush, got {names}" provider.shutdown() + + +def test_configure_metrics_disabled_defaults_to_no_meter_provider(monkeypatch): + # Metrics must be default-off and byte-identical to current behavior: no + # MeterProvider is set and no gauge/histogram recording call sites exist. + monkeypatch.setenv("OTEL_TRACING_ENABLED", "false") + monkeypatch.setenv("OTEL_LOGGING_ENABLED", "false") + monkeypatch.delenv("OTEL_METRICS_ENABLED", raising=False) + + set_calls = [] + monkeypatch.setattr(_utils.metrics, "set_meter_provider", lambda provider: set_calls.append(provider)) + monkeypatch.setattr(_utils, "_create_metric_exporter", lambda **kwargs: object()) + monkeypatch.setattr(_utils, "PeriodicExportingMetricReader", lambda exporter=None: object()) + monkeypatch.setattr(_utils, "MeterProvider", lambda **kwargs: object()) + + _utils.configure(name="test", namespace="test") + + assert set_calls == [] + + +@pytest.mark.parametrize("env_value", [("true", True), ("True", True), ("false", False), ("garbage", False)]) +def test_configure_metrics_gates_on_env(monkeypatch, env_value): + value, expect_enabled = env_value + monkeypatch.setenv("OTEL_TRACING_ENABLED", "false") + monkeypatch.setenv("OTEL_LOGGING_ENABLED", "false") + monkeypatch.setenv("OTEL_METRICS_ENABLED", value) + + set_calls = [] + monkeypatch.setattr(_utils.metrics, "set_meter_provider", lambda provider: set_calls.append(provider)) + monkeypatch.setattr(_utils, "_create_metric_exporter", lambda **kwargs: object()) + monkeypatch.setattr(_utils, "PeriodicExportingMetricReader", lambda exporter=None, **kw: object()) + monkeypatch.setattr(_utils, "MeterProvider", lambda **kwargs: object()) + + _utils.configure(name="test", namespace="test") + + assert (len(set_calls) == 1) is expect_enabled + + +def test_configure_metrics_enabled_builds_otlp_meter_provider(monkeypatch): + monkeypatch.setenv("OTEL_TRACING_ENABLED", "false") + monkeypatch.setenv("OTEL_LOGGING_ENABLED", "false") + monkeypatch.setenv("OTEL_METRICS_ENABLED", "true") + + captured = {} + + def fake_exporter(**kwargs): + captured["exporter_kwargs"] = kwargs + return object() + + class FakeReader: + def __init__(self, exporter): + captured["reader_exporter"] = exporter + + class FakeMeterProvider: + def __init__(self, resource, metric_readers): + captured["resource"] = resource + captured["metric_readers"] = metric_readers + self.metric_readers = metric_readers + + monkeypatch.setattr(_utils, "_create_metric_exporter", fake_exporter) + monkeypatch.setattr(_utils, "PeriodicExportingMetricReader", FakeReader) + monkeypatch.setattr(_utils, "MeterProvider", FakeMeterProvider) + + set_calls = [] + monkeypatch.setattr(_utils.metrics, "set_meter_provider", lambda provider: set_calls.append(provider)) + + _utils.configure(name="test-agent", namespace="test-ns") + + # A single MeterProvider wired to a periodic reader is registered globally. + assert len(set_calls) == 1 + provider = set_calls[0] + assert isinstance(provider, FakeMeterProvider) + assert len(provider.metric_readers) == 1 + # The reader carries the OTLP exporter built via the shared protocol helper. + assert captured["reader_exporter"] is not None + # Resource metadata is threaded through so attributes merge correctly. + assert captured["resource"].attributes["service.name"] == "test-agent" + + +def test_configure_metrics_uses_signal_specific_endpoint(monkeypatch): + monkeypatch.setenv("OTEL_TRACING_ENABLED", "false") + monkeypatch.setenv("OTEL_LOGGING_ENABLED", "false") + monkeypatch.setenv("OTEL_METRICS_ENABLED", "true") + monkeypatch.setenv("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", "http://collector:4317") + + exporter_kwargs = {} + + def fake_exporter(**kwargs): + exporter_kwargs.update(kwargs) + return object() + + monkeypatch.setattr(_utils, "_create_metric_exporter", fake_exporter) + monkeypatch.setattr(_utils, "PeriodicExportingMetricReader", lambda exporter=None, **kw: object()) + monkeypatch.setattr(_utils, "MeterProvider", lambda **kwargs: object()) + monkeypatch.setattr(_utils.metrics, "set_meter_provider", lambda provider: None) + + _utils.configure(name="test", namespace="test") + + assert exporter_kwargs["endpoint"] == "http://collector:4317" + + +def test_force_flush_flushes_traces_and_metrics(monkeypatch): + calls = {"trace": [], "metric": []} + trace_provider = SimpleNamespace(force_flush=lambda timeout: calls["trace"].append(timeout)) + metric_provider = SimpleNamespace(force_flush=lambda timeout: calls["metric"].append(timeout)) + monkeypatch.setattr(_utils.trace, "get_tracer_provider", lambda: trace_provider) + monkeypatch.setattr(_utils.metrics, "get_meter_provider", lambda: metric_provider) + + _utils.force_flush() + + assert calls == {"trace": [3000], "metric": [3000]} + + +def test_force_flush_noop_when_metric_provider_lacks_force_flush(monkeypatch): + # A default (no-op) meter provider exposes no force_flush; must not raise. + monkeypatch.setattr(_utils.trace, "get_tracer_provider", lambda: SimpleNamespace()) + monkeypatch.setattr(_utils.metrics, "get_meter_provider", lambda: SimpleNamespace()) + + _utils.force_flush() + + +def test_force_flush_swallows_metric_exporter_errors(monkeypatch): + trace_provider = SimpleNamespace(force_flush=lambda timeout: None) + + def boom(timeout): + raise RuntimeError("collector down") + + metric_provider = SimpleNamespace(force_flush=boom) + monkeypatch.setattr(_utils.trace, "get_tracer_provider", lambda: trace_provider) + monkeypatch.setattr(_utils.metrics, "get_meter_provider", lambda: metric_provider) + + _utils.force_flush()