diff --git a/helm/kagent/templates/controller-configmap.yaml b/helm/kagent/templates/controller-configmap.yaml index a8ebb6ea02..f9602a9af5 100644 --- a/helm/kagent/templates/controller-configmap.yaml +++ b/helm/kagent/templates/controller-configmap.yaml @@ -19,8 +19,10 @@ data: # OpenTelemetry Configuration OTEL_TRACING_ENABLED: {{ .Values.otel.tracing.enabled | quote }} OTEL_LOGGING_ENABLED: {{ .Values.otel.logging.enabled | quote }} + OTEL_METRICS_ENABLED: {{ .Values.otel.metrics.enabled | quote }} {{- $tracesEndpoint := .Values.otel.tracing.exporter.otlp.endpoint }} {{- $logsEndpoint := .Values.otel.logging.exporter.otlp.endpoint }} + {{- $metricsEndpoint := .Values.otel.metrics.exporter.otlp.endpoint }} {{- if and $tracesEndpoint $logsEndpoint (eq $tracesEndpoint $logsEndpoint) }} # Using unified OTEL endpoint (same for traces and logs) OTEL_EXPORTER_OTLP_ENDPOINT: {{ $tracesEndpoint | quote }} @@ -28,6 +30,8 @@ data: OTEL_EXPORTER_OTLP_TRACES_TIMEOUT: {{ .Values.otel.tracing.exporter.otlp.timeout | quote }} OTEL_EXPORTER_OTLP_LOGS_INSECURE: {{ .Values.otel.logging.exporter.otlp.insecure | quote }} OTEL_EXPORTER_OTLP_LOGS_TIMEOUT: {{ .Values.otel.logging.exporter.otlp.timeout | quote }} + OTEL_EXPORTER_OTLP_METRICS_INSECURE: {{ .Values.otel.metrics.exporter.otlp.insecure | quote }} + OTEL_EXPORTER_OTLP_METRICS_TIMEOUT: {{ .Values.otel.metrics.exporter.otlp.timeout | quote }} OTEL_EXPORTER_OTLP_PROTOCOL: {{ .Values.otel.tracing.exporter.otlp.protocol | quote }} {{- else }} # Using separate endpoints for traces and logs @@ -42,6 +46,11 @@ data: OTEL_EXPORTER_OTLP_LOGS_INSECURE: {{ .Values.otel.logging.exporter.otlp.insecure | quote }} OTEL_EXPORTER_OTLP_LOGS_TIMEOUT: {{ .Values.otel.logging.exporter.otlp.timeout | quote }} {{- end }} + {{- if $metricsEndpoint }} + OTEL_EXPORTER_OTLP_METRICS_ENDPOINT: {{ $metricsEndpoint | quote }} + OTEL_EXPORTER_OTLP_METRICS_INSECURE: {{ .Values.otel.metrics.exporter.otlp.insecure | quote }} + OTEL_EXPORTER_OTLP_METRICS_TIMEOUT: {{ .Values.otel.metrics.exporter.otlp.timeout | quote }} + {{- end }} {{- end }} {{- if .Values.proxy.url }} PROXY_URL: {{ .Values.proxy.url | quote }} diff --git a/helm/kagent/values.yaml b/helm/kagent/values.yaml index b30e9338b4..fc4dc1e8b9 100644 --- a/helm/kagent/values.yaml +++ b/helm/kagent/values.yaml @@ -854,6 +854,13 @@ otel: endpoint: "" timeout: 15000 # milliseconds insecure: true + metrics: + enabled: false + exporter: + otlp: + endpoint: "" + 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..b9dfb7af46 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,13 +138,14 @@ 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 telemetry still buffered by the tracer and meter providers. 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). + response body closes, so unexported spans (and metric points from a + periodic reader that may never fire before suspension) stay frozen in the + snapshot until the session's next resume (or forever, for a session's last + message). No-op when a provider has no force_flush (signal disabled). The timeout defaults to 3000ms, configurable via KAGENT_TRACE_FLUSH_TIMEOUT_MS. """ @@ -139,12 +153,18 @@ def force_flush(timeout_millis: int | None = None) -> 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) + meter_provider = metrics.get_meter_provider() + meter_flush = getattr(meter_provider, "force_flush", None) + if meter_flush is not None: + try: + meter_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 +238,10 @@ 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. Args: name: service name to report to OpenTelemetry (used as ``service.name``). Default is "kagent". @@ -235,6 +255,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, @@ -321,6 +342,32 @@ def configure( OpenAIInstrumentor(use_legacy_attributes=False).instrument(logger_provider=logger_provider) _instrument_anthropic(logger_provider) _instrument_google_generativeai(logger_provider) + # Configure metrics if enabled. google-adk already defines and records its + # GenAI metric instruments (duration histograms, inference/tool-call counts, + # gen_ai.client.token.usage) under meter scope gcp.vertex.agent; installing a + # MeterProvider here is what turns those recorded data points into exports. + if metrics_enabled: + logging.info("Enabling metrics") + metric_endpoint = ( + os.getenv("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT") + or os.getenv("OTEL_METRICS_EXPORTER_OTLP_ENDPOINT") # Backward compatibility + or os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT") + ) + metric_timeout_seconds = _resolve_otlp_timeout_seconds("METRICS") + logging.info("Metrics endpoint: %s", metric_endpoint or "") + # A periodic reader exports on an interval, exactly like the batch + # processors for traces and logs. On Agent Substrate the actor can be + # checkpointed before the interval elapses, so KAGENT_PRE_RESPONSE_* + # flushing (see force_flush) also drains the metric reader. + 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)) + metric_provider = MeterProvider(resource=resource, metric_readers=[metric_reader]) + metrics.set_meter_provider(metric_provider) + logging.info("Meter provider configured with OTLP") elif tracing_enabled: # Use legacy attributes (input/output as GenAI span attributes) logging.info("OpenAI instrumentation configured with legacy GenAI span attributes") diff --git a/python/packages/kagent-core/tests/test_tracing_configure.py b/python/packages/kagent-core/tests/test_tracing_configure.py index 3ad0b3e3f4..2d6087786c 100644 --- a/python/packages/kagent-core/tests/test_tracing_configure.py +++ b/python/packages/kagent-core/tests/test_tracing_configure.py @@ -417,3 +417,113 @@ 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_is_default_off(monkeypatch): + # OTEL_METRICS_ENABLED is unset: no MeterProvider may be installed, so the + # default gate keeps behavior byte-identical to before metrics existed. + monkeypatch.setenv("OTEL_LOGGING_ENABLED", "false") + monkeypatch.setenv("OTEL_TRACING_ENABLED", "false") + + instrument_calls = {} + + def set_meter_provider(provider): + instrument_calls["meter_provider"] = provider + + monkeypatch.setattr(_utils, "metrics", SimpleNamespace(set_meter_provider=set_meter_provider)) + monkeypatch.setattr(_utils, "OpenAIInstrumentor", lambda **kwargs: SimpleNamespace(instrument=lambda **kw: None)) + monkeypatch.setattr(_utils, "_instrument_anthropic", lambda *a, **kw: None) + monkeypatch.setattr(_utils, "_instrument_google_generativeai", lambda *a, **kw: None) + + _utils.configure(name="test", namespace="test") + + assert "meter_provider" not in instrument_calls + + +def test_configure_metrics_enabled_installs_meter_provider(monkeypatch): + monkeypatch.setenv("OTEL_METRICS_ENABLED", "true") + monkeypatch.setenv("OTEL_LOGGING_ENABLED", "false") + monkeypatch.setenv("OTEL_TRACING_ENABLED", "false") + + instrument_calls = {} + + class FakeMeterProvider: + def __init__(self, resource=None, metric_readers=None): + instrument_calls["resource"] = resource + instrument_calls["metric_readers"] = metric_readers + + def fake_create_metric_exporter(**kwargs): + instrument_calls["exporter_kwargs"] = kwargs + return object() + + def fake_periodic_reader(exporter=None): + instrument_calls["reader_exporter"] = exporter + return SimpleNamespace() + + monkeypatch.setattr(_utils, "MeterProvider", FakeMeterProvider) + monkeypatch.setattr(_utils, "_create_metric_exporter", fake_create_metric_exporter) + monkeypatch.setattr(_utils, "PeriodicExportingMetricReader", fake_periodic_reader) + monkeypatch.setattr( + _utils, + "metrics", + SimpleNamespace(set_meter_provider=lambda provider: instrument_calls.setdefault("meter_provider", provider)), + ) + + _utils.configure(name="test", namespace="test") + + assert "meter_provider" in instrument_calls + assert isinstance(instrument_calls["metric_readers"], list) + assert instrument_calls["reader_exporter"] is not None + # No endpoint set: only the default 10s timeout is passed to the exporter. + assert instrument_calls["exporter_kwargs"] == {"timeout": 10.0} + + +def test_configure_metrics_uses_endpoint_and_timeout(monkeypatch): + monkeypatch.setenv("OTEL_METRICS_ENABLED", "true") + monkeypatch.setenv("OTEL_LOGGING_ENABLED", "false") + monkeypatch.setenv("OTEL_TRACING_ENABLED", "false") + monkeypatch.setenv("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", "http://collector:4317") + monkeypatch.setenv("OTEL_EXPORTER_OTLP_METRICS_TIMEOUT", "2000") + + instrument_calls = {} + + def fake_create_metric_exporter(**kwargs): + instrument_calls["exporter_kwargs"] = kwargs + return object() + + monkeypatch.setattr(_utils, "_create_metric_exporter", fake_create_metric_exporter) + monkeypatch.setattr(_utils, "PeriodicExportingMetricReader", lambda exporter=None: SimpleNamespace()) + monkeypatch.setattr(_utils, "MeterProvider", lambda resource=None, metric_readers=None: SimpleNamespace()) + monkeypatch.setattr(_utils, "metrics", SimpleNamespace(set_meter_provider=lambda provider: None)) + monkeypatch.setattr(_utils, "trace", SimpleNamespace(get_tracer_provider=lambda: SimpleNamespace())) + + _utils.configure(name="test", namespace="test") + + assert instrument_calls["exporter_kwargs"] == {"endpoint": "http://collector:4317", "timeout": 2.0} + + +def test_force_flush_flushes_meter_provider_when_present(monkeypatch): + # force_flush drains both the tracer and the meter provider when present, + # mirroring the batch-timer export safety net for the periodic reader. + calls = [] + tracer_provider = SimpleNamespace(force_flush=lambda timeout: calls.append(("traces", timeout))) + meter_provider = SimpleNamespace(force_flush=lambda timeout: calls.append(("metrics", timeout))) + monkeypatch.setattr(_utils.trace, "get_tracer_provider", lambda: tracer_provider) + monkeypatch.setattr(_utils.metrics, "get_meter_provider", lambda: meter_provider) + + _utils.force_flush() + + assert calls == [("traces", 3000), ("metrics", 3000)] + + +def test_force_flush_skips_meter_provider_without_support(monkeypatch): + # The default no-op meter provider exposes no force_flush; must not raise. + calls = [] + monkeypatch.setattr( + _utils.trace, "get_tracer_provider", lambda: SimpleNamespace(force_flush=lambda t: calls.append(t)) + ) + monkeypatch.setattr(_utils.metrics, "get_meter_provider", lambda: SimpleNamespace()) + + _utils.force_flush() + + assert calls == [3000]