feat(kagent-core): wire a MeterProvider so google-adk GenAI metrics export (gated OTEL_METRICS_ENABLED)

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

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 "<default>")
+
+        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")

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()