feat(core): wire a MeterProvider into the Python runtime (gated OTEL_METRICS_ENABLED)

helm/kagent/templates/controller-configmap.yaml

@@ -19,15 +19,19 @@ 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 }}
   OTEL_EXPORTER_OTLP_TRACES_INSECURE: {{ .Values.otel.tracing.exporter.otlp.insecure | quote }}
   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 }}

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

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,33 @@ 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.
     """
     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)
+    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 "<default>")
+        # 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")

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]