diff --git a/docs/genai-token-metrics.md b/docs/genai-token-metrics.md new file mode 100644 index 0000000000..c0a5f9a33a --- /dev/null +++ b/docs/genai-token-metrics.md @@ -0,0 +1,102 @@ +# GenAI token-usage metrics + +kagent's **Go ADK** agent runtime records the OpenTelemetry GenAI-semconv metric +[`gen_ai.client.token.usage`](https://opentelemetry.io/docs/specs/semconv/gen-ai/gen-ai-metrics/#metric-gen_aiclienttokenusage) +using the native Prometheus client library and exposes it for scraping. It lets you graph and alert +on token spend per model / provider without parsing traces. + +## What is emitted + +A Prometheus histogram, served at **`/metrics`** on the agent's HTTP port: + +| Prometheus name | OTel semconv | Notes | +| --- | --- | --- | +| `gen_ai_client_token_usage` | `gen_ai.client.token.usage` | histogram, semconv-recommended buckets | + +Labels (semconv attributes, dots → underscores), aligned with what the upstream Google ADK Python +runtime emits for the same instrument so a single dashboard works across both runtimes: + +| Label | Values | +| --- | --- | +| `gen_ai_token_type` | `input`, `output` (output = candidate + reasoning tokens) | +| `gen_ai_operation_name` | `chat` | +| `gen_ai_provider_name` | well-known value, e.g. `openai`, `anthropic`, `gcp.vertex_ai`, `aws.bedrock`, `azure.ai.openai` | +| `gen_ai_request_model` | configured model, e.g. `gpt-4o` | +| `gen_ai_response_model` | model the provider served (falls back to request model) | +| `gen_ai_agent_name` | agent that produced the tokens (the kagent app name) | +| `error_type` | set on failed requests; empty otherwise | + +One observation is recorded per LLM call (streaming partial chunks are not double-counted). + +## Configuration + +- **Runtime**: available on Declarative agents with `runtime: go`. (The Python runtime + records the same metric from upstream Google ADK.) +- **Gate**: recording and the `/metrics` endpoint are **default-OFF**, matching kagent's other + observability gates. Set `OTEL_METRICS_ENABLED=true` to turn them on: + + ```yaml + env: + - name: OTEL_METRICS_ENABLED + value: "true" + ``` + +- To have Prometheus scrape the endpoint, annotate the Go-runtime agent pods: + + ```yaml + metadata: + annotations: + prometheus.io/scrape: "true" + prometheus.io/port: "" + prometheus.io/path: "/metrics" + ``` + +### Scraping it + +Any Prometheus-compatible scraper that honors pod annotations will pick agents up. With an +OpenTelemetry Collector, add a `prometheus` receiver job with pod discovery: + +```yaml +receivers: + prometheus: + config: + scrape_configs: + - job_name: kagent-agents + kubernetes_sd_configs: [{ role: pod }] + relabel_configs: + - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_scrape] + regex: "true" + action: keep + - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_path] + target_label: __metrics_path__ + - source_labels: [__address__, __meta_kubernetes_pod_annotation_prometheus_io_port] + regex: ([^:]+)(?::\d+)?;(\d+) + replacement: $$1:$$2 + target_label: __address__ +``` + +## Verifying + +```bash +# exec into a Go-runtime agent pod and curl its metrics endpoint +kubectl exec -- wget -qO- localhost:/metrics | grep gen_ai_client_token_usage +``` + +Typical output after two chat requests (note `_count` equals the number of LLM calls, not stream +chunks, and the semconv labels including `gen_ai.agent.name`, response model, and an empty +`error.type`): + +```text +# HELP gen_ai_client_token_usage Measures the number of input and output tokens used by GenAI requests. +# TYPE gen_ai_client_token_usage histogram +gen_ai_client_token_usage_sum{error_type="",gen_ai_agent_name="my_agent",gen_ai_operation_name="chat",gen_ai_provider_name="gcp.vertex_ai",gen_ai_request_model="gemini-2.5-flash",gen_ai_response_model="gemini-2.5-flash",gen_ai_token_type="input"} 137 +gen_ai_client_token_usage_count{error_type="",gen_ai_agent_name="my_agent",gen_ai_operation_name="chat",gen_ai_provider_name="gcp.vertex_ai",gen_ai_request_model="gemini-2.5-flash",gen_ai_response_model="gemini-2.5-flash",gen_ai_token_type="input"} 2 +gen_ai_client_token_usage_sum{error_type="",gen_ai_agent_name="my_agent",gen_ai_operation_name="chat",gen_ai_provider_name="gcp.vertex_ai",gen_ai_request_model="gemini-2.5-flash",gen_ai_response_model="gemini-2.5-flash",gen_ai_token_type="output"} 91 +gen_ai_client_token_usage_count{error_type="",gen_ai_agent_name="my_agent",gen_ai_operation_name="chat",gen_ai_provider_name="gcp.vertex_ai",gen_ai_request_model="gemini-2.5-flash",gen_ai_response_model="gemini-2.5-flash",gen_ai_token_type="output"} 2 +``` + +## Follow-ups + +- Optional OTLP **push** (in addition to the scrape endpoint) via the OpenTelemetry + [Prometheus→OTLP bridge](https://pkg.go.dev/go.opentelemetry.io/contrib/bridges/prometheus), for + environments that push to an OTLP collector rather than scrape. \ No newline at end of file diff --git a/go/adk/cmd/main.go b/go/adk/cmd/main.go index c165bf67f8..b423a24c72 100644 --- a/go/adk/cmd/main.go +++ b/go/adk/cmd/main.go @@ -20,6 +20,7 @@ import ( runnerpkg "github.com/kagent-dev/kagent/go/adk/pkg/runner" "github.com/kagent-dev/kagent/go/adk/pkg/session" "github.com/kagent-dev/kagent/go/adk/pkg/telemetry" + "github.com/kagent-dev/kagent/go/api/adk" "go.uber.org/zap" "go.uber.org/zap/zapcore" ) @@ -218,12 +219,15 @@ func main() { } stream := agentConfig.GetStream() + modelName, providerName := resolveModelLabels(agentConfig) executor := a2a.NewKAgentExecutor(a2a.KAgentExecutorConfig{ RunnerConfig: runnerConfig, SessionService: sessionService, Stream: stream, AppName: appName, Logger: logger, + ModelName: modelName, + ProviderName: providerName, }) // Build the agent card. @@ -262,6 +266,16 @@ func main() { } } +// resolveModelLabels derives the gen_ai.request.model / gen_ai.provider.name +// labels for token-usage metrics from the agent config. Returns empty strings +// when no model is configured; the metric simply omits those attributes. +func resolveModelLabels(agentConfig *adk.AgentConfig) (model, provider string) { + if agentConfig == nil || agentConfig.Model == nil { + return "", "" + } + return config.ModelName(agentConfig.Model), telemetry.SemconvProviderName(agentConfig.Model.GetType()) +} + func deriveAppName(kagentName, kagentNamespace string, agentCard *a2atype.AgentCard, logger logr.Logger) string { if kagentNamespace != "" && kagentName != "" { namespace := strings.ReplaceAll(kagentNamespace, "-", "_") diff --git a/go/adk/pkg/a2a/executor.go b/go/adk/pkg/a2a/executor.go index de46a9b181..3d4ab314c6 100644 --- a/go/adk/pkg/a2a/executor.go +++ b/go/adk/pkg/a2a/executor.go @@ -33,6 +33,11 @@ type KAgentExecutorConfig struct { Stream bool AppName string Logger logr.Logger + // ModelName and ProviderName label GenAI token-usage metrics + // (gen_ai.request.model / gen_ai.provider.name). Both may be empty, in + // which case the corresponding metric attributes are omitted. + ModelName string + ProviderName string } // KAgentExecutor keeps kagent's request/session glue around the upstream ADK @@ -75,6 +80,7 @@ func NewKAgentExecutor(cfg KAgentExecutorConfig) *KAgentExecutor { } processed.Artifact.SetMeta(apia2a.TimelinePositionMetadataKey, position.UTC().Format(time.RFC3339Nano)) } + recordTokenUsage(cfg.ModelName, cfg.ProviderName, cfg.AppName, event) return nil }, OutputMode: adka2a.OutputArtifactPerEvent, @@ -88,6 +94,30 @@ func NewKAgentExecutor(cfg KAgentExecutorConfig) *KAgentExecutor { } } +// recordTokenUsage records GenAI token usage for a single ADK event on the +// gen_ai.client.token.usage histogram. Partial (streaming) events are skipped: +// a streamed LLM call emits many Partial chunks but usage is reported once on +// the aggregated non-partial event, so this counts one observation per LLM +// call, not per stream chunk. Output combines candidate + reasoning tokens. +func recordTokenUsage(modelName, providerName, agentName string, adkEvent *adksession.Event) { + if adkEvent == nil { + return + } + um := adkEvent.UsageMetadata + if um == nil || adkEvent.Partial { + return + } + telemetry.RecordTokenUsage(telemetry.TokenUsage{ + RequestModel: modelName, + ResponseModel: adkEvent.ModelVersion, + Provider: providerName, + AgentName: agentName, + ErrorType: adkEvent.ErrorCode, + InputTokens: int64(um.PromptTokenCount), + OutputTokens: int64(um.CandidatesTokenCount) + int64(um.ThoughtsTokenCount), + }) +} + // UserIDCallInterceptor returns an a2asrv.CallInterceptor that extracts the // x-user-id HTTP header from the incoming request metadata and sets it as the // authenticated user on the CallContext. diff --git a/go/adk/pkg/a2a/executor_metrics_test.go b/go/adk/pkg/a2a/executor_metrics_test.go new file mode 100644 index 0000000000..c5bba9ed78 --- /dev/null +++ b/go/adk/pkg/a2a/executor_metrics_test.go @@ -0,0 +1,57 @@ +package a2a + +import ( + "testing" + + "github.com/prometheus/client_golang/prometheus" + adkmodel "google.golang.org/adk/v2/model" + adksession "google.golang.org/adk/v2/session" + "google.golang.org/genai" +) + +// tokenUsageSeriesCount returns how many label series currently exist on the +// gen_ai_client_token_usage histogram across the default registry. +func tokenUsageSeriesCount(t *testing.T) int { + t.Helper() + mfs, err := prometheus.DefaultGatherer.Gather() + if err != nil { + t.Fatalf("gather: %v", err) + } + for _, mf := range mfs { + if mf.GetName() != "gen_ai_client_token_usage" { + continue + } + return len(mf.GetMetric()) + } + return 0 +} + +func TestRecordTokenUsage_RecordsPerLLMCall(t *testing.T) { + t.Setenv("OTEL_METRICS_ENABLED", "true") + series := 0 + + // Partial (streaming) events must be skipped: a streamed call emits many + // partial chunks but usage is reported once on the final non-partial event. + recordTokenUsage("gpt-4o", "openai", "my-agent", &adksession.Event{ + LLMResponse: adkmodel.LLMResponse{ + Partial: true, + ModelVersion: "gpt-4o-2024-11-20", + UsageMetadata: &genai.GenerateContentResponseUsageMetadata{PromptTokenCount: 100, CandidatesTokenCount: 42}, + }, + }) + if got := tokenUsageSeriesCount(t); got != series { + t.Fatalf("partial event must not record tokens, got %d series", got) + } + + // The aggregated non-partial event records one input and one output series. + recordTokenUsage("gpt-4o", "openai", "my-agent", &adksession.Event{ + LLMResponse: adkmodel.LLMResponse{ + ModelVersion: "gpt-4o-2023-11-20", + UsageMetadata: &genai.GenerateContentResponseUsageMetadata{PromptTokenCount: 100, CandidatesTokenCount: 40, ThoughtsTokenCount: 2}, + }, + }) + got := tokenUsageSeriesCount(t) + if got != 2 { + t.Fatalf("expected input+output series after one LLM call, got %d", got) + } +} diff --git a/go/adk/pkg/a2a/server/server.go b/go/adk/pkg/a2a/server/server.go index 88ceca9b76..ff512f441c 100644 --- a/go/adk/pkg/a2a/server/server.go +++ b/go/adk/pkg/a2a/server/server.go @@ -59,6 +59,11 @@ func NewA2AServer(agentCard a2atype.AgentCard, executor a2asrv.AgentExecutor, lo mux := http.NewServeMux() RegisterHealthEndpoints(mux) mux.Handle(a2asrv.WellKnownAgentCardPath, a2asrv.NewStaticAgentCardHandler(&agentCard)) + // Serve Prometheus metrics for scraping when the metrics gate is on. This + // endpoint is excluded from request tracing and span flushing below. + if telemetry.MetricsEnabled() { + mux.Handle("/metrics", telemetry.MetricsHandler()) + } mux.Handle("/", jsonrpcHandler) grpcServer := grpc.NewServer() @@ -81,6 +86,8 @@ func NewA2AServer(agentCard a2atype.AgentCard, executor a2asrv.AgentExecutor, lo return false case r.URL.Path == "/health", r.URL.Path == "/healthz", r.URL.Path == a2asrv.WellKnownAgentCardPath: return false + case r.URL.Path == "/metrics": + return false default: return true } diff --git a/go/adk/pkg/config/config_usage.go b/go/adk/pkg/config/config_usage.go index 01cea9ae1c..ee4cf018b9 100644 --- a/go/adk/pkg/config/config_usage.go +++ b/go/adk/pkg/config/config_usage.go @@ -139,3 +139,10 @@ func getModelName(m adk.Model) string { return "unknown" } } + +// ModelName returns the configured model identifier (e.g. "gpt-4o", +// "claude-3-5-sonnet"), or "unknown" for an unrecognized model type. It is the +// exported form of getModelName, used to label GenAI telemetry. +func ModelName(m adk.Model) string { + return getModelName(m) +} diff --git a/go/adk/pkg/telemetry/metrics.go b/go/adk/pkg/telemetry/metrics.go new file mode 100644 index 0000000000..421e7aacaf --- /dev/null +++ b/go/adk/pkg/telemetry/metrics.go @@ -0,0 +1,149 @@ +package telemetry + +import ( + "net/http" + "os" + "strings" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +// GenAI token-usage instrumentation using the native Prometheus client library. +// +// Metric and attribute names follow the OpenTelemetry GenAI semantic +// conventions (semconv 1.40.0), mapped to Prometheus naming (dots -> underscores): +// - metric gen_ai.client.token.usage -> gen_ai_client_token_usage +// - attrs gen_ai.token.type, gen_ai.request.model, gen_ai.response.model, +// gen_ai.provider.name, gen_ai.agent.name, gen_ai.operation.name, error.type +// +// The attribute set mirrors what the upstream Google ADK Python runtime emits +// for the same instrument, so a single dashboard works across the Go and Python +// runtimes. +// +// https://opentelemetry.io/docs/specs/semconv/gen-ai/gen-ai-metrics/ +const ( + metricGenAIClientTokenUsage = "gen_ai_client_token_usage" + + labelGenAITokenType = "gen_ai_token_type" + labelGenAIOperationName = "gen_ai_operation_name" + labelGenAIProviderName = "gen_ai_provider_name" + labelGenAIRequestModel = "gen_ai_request_model" + labelGenAIResponseModel = "gen_ai_response_model" + labelGenAIAgentName = "gen_ai_agent_name" + labelErrorType = "error_type" + + tokenTypeInput = "input" + tokenTypeOutput = "output" + + // operationChat is the gen_ai.operation.name for the chat-completion calls + // that produce the token usage recorded here. + operationChat = "chat" + + // metricsEnabledEnvVar gates metrics recording and the /metrics endpoint. + metricsEnabledEnvVar = "OTEL_METRICS_ENABLED" +) + +// tokenUsageBuckets is the explicit bucket layout recommended by the GenAI +// metrics semantic conventions for gen_ai.client.token.usage. +// https://opentelemetry.io/docs/specs/semconv/gen-ai/gen-ai-metrics/#metric-gen_aiclienttokenusage +var tokenUsageBuckets = []float64{1, 4, 16, 64, 256, 1024, 4096, 16384, 65536, 262144, 1048576, 4194304, 16777216, 67108864} + +// tokenUsage is the gen_ai.client.token.usage histogram, registered on the +// default Prometheus registry so it is served by MetricsHandler alongside the +// standard Go/process collectors. It is always registered; MetricsEnabled +// gates whether observations are recorded, keeping the scrape endpoint stable. +var tokenUsage = promauto.NewHistogramVec( + prometheus.HistogramOpts{ + Name: metricGenAIClientTokenUsage, + Help: "Measures the number of input and output tokens used by GenAI requests.", + Buckets: tokenUsageBuckets, + }, + []string{ + labelGenAITokenType, + labelGenAIOperationName, + labelGenAIProviderName, + labelGenAIRequestModel, + labelGenAIResponseModel, + labelGenAIAgentName, + labelErrorType, + }, +) + +// MetricsEnabled reports whether the GenAI token metrics pipeline is on, i.e. +// the OTEL_METRICS_ENABLED gate used consistently across kagent's runtimes. +func MetricsEnabled() bool { + return strings.EqualFold(strings.TrimSpace(os.Getenv(metricsEnabledEnvVar)), "true") +} + +// MetricsHandler returns an http.Handler that serves the agent's Prometheus +// metrics, intended to be mounted at /metrics for scraping. +func MetricsHandler() http.Handler { + return promhttp.Handler() +} + +// SemconvProviderName maps a kagent model type (adk.Model.GetType()) to the +// OpenTelemetry GenAI well-known gen_ai.provider.name value. Types without a +// well-known mapping (e.g. ollama, sap_ai_core, custom) pass through unchanged. +// https://opentelemetry.io/docs/specs/semconv/registry/attributes/gen-ai/#gen-ai-provider-name +func SemconvProviderName(modelType string) string { + switch modelType { + case "openai": + return "openai" + case "azure_openai": + return "azure.ai.openai" + case "anthropic": + return "anthropic" + case "gemini": + return "gcp.gemini" + case "gemini_vertex_ai", "gemini_anthropic": + return "gcp.vertex_ai" + case "bedrock": + return "aws.bedrock" + default: + return modelType + } +} + +// TokenUsage carries the per-request labels and counts for one recording on the +// gen_ai.client.token.usage histogram. +type TokenUsage struct { + // RequestModel is gen_ai.request.model (the configured model). + RequestModel string + // ResponseModel is gen_ai.response.model (the model the provider actually + // served). Falls back to RequestModel when empty. + ResponseModel string + // Provider is gen_ai.provider.name (a semconv well-known value). + Provider string + // AgentName is gen_ai.agent.name (the agent that produced the tokens). + AgentName string + // ErrorType is error.type; empty for successful requests. + ErrorType string + // InputTokens / OutputTokens are the token counts (output = candidate + + // reasoning tokens). Non-positive counts are skipped. + InputTokens int64 + OutputTokens int64 +} + +// RecordTokenUsage records input/output token counts on the +// gen_ai.client.token.usage histogram. If the metric pipeline is disabled +// (OTEL_METRICS_ENABLED unset or not "true"), it is a no-op. Zero/negative +// counts are skipped. +func RecordTokenUsage(u TokenUsage) { + if !MetricsEnabled() { + return + } + responseModel := u.ResponseModel + if responseModel == "" { + responseModel = u.RequestModel + } + if u.InputTokens > 0 { + tokenUsage.WithLabelValues(tokenTypeInput, operationChat, u.Provider, u.RequestModel, responseModel, u.AgentName, u.ErrorType). + Observe(float64(u.InputTokens)) + } + if u.OutputTokens > 0 { + tokenUsage.WithLabelValues(tokenTypeOutput, operationChat, u.Provider, u.RequestModel, responseModel, u.AgentName, u.ErrorType). + Observe(float64(u.OutputTokens)) + } +} diff --git a/go/adk/pkg/telemetry/metrics_test.go b/go/adk/pkg/telemetry/metrics_test.go new file mode 100644 index 0000000000..884209d71a --- /dev/null +++ b/go/adk/pkg/telemetry/metrics_test.go @@ -0,0 +1,120 @@ +package telemetry + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/prometheus/client_golang/prometheus/testutil" +) + +// TestRecordTokenUsage_RecordsHistogram verifies input/output token counts are +// recorded as two separate series on the gen_ai_client_token_usage histogram. +func TestRecordTokenUsage_RecordsHistogram(t *testing.T) { + t.Setenv(metricsEnabledEnvVar, "true") + tokenUsage.Reset() + + RecordTokenUsage(TokenUsage{ + RequestModel: "gpt-4o", Provider: "openai", InputTokens: 100, OutputTokens: 42, + }) + + // One series per token type (input, output). + if got := testutil.CollectAndCount(tokenUsage); got != 2 { + t.Fatalf("expected 2 histogram series (input+output), got %d", got) + } +} + +func TestRecordTokenUsage_SkipsZero(t *testing.T) { + t.Setenv(metricsEnabledEnvVar, "true") + tokenUsage.Reset() + + RecordTokenUsage(TokenUsage{RequestModel: "gpt-4o", Provider: "openai"}) + + if got := testutil.CollectAndCount(tokenUsage); got != 0 { + t.Fatalf("expected no series for zero token counts, got %d", got) + } +} + +func TestRecordTokenUsage_Disabled(t *testing.T) { + tokenUsage.Reset() + + RecordTokenUsage(TokenUsage{RequestModel: "gpt-4o", Provider: "openai", InputTokens: 100}) + + if got := testutil.CollectAndCount(tokenUsage); got != 0 { + t.Fatalf("expected no series when metrics are disabled, got %d", got) + } +} + +func TestRecordTokenUsage_ResponseModelFallback(t *testing.T) { + t.Setenv(metricsEnabledEnvVar, "true") + tokenUsage.Reset() + RecordTokenUsage(TokenUsage{RequestModel: "gemini-2.5-flash", Provider: "gcp.vertex_ai", InputTokens: 3}) + RecordTokenUsage(TokenUsage{RequestModel: "gemini-2.5-flash", ResponseModel: "gemini-2.5-flash-002", Provider: "gcp.vertex_ai", InputTokens: 3}) + + body := serveMetrics(t) + if !strings.Contains(body, "gen_ai_response_model=\"gemini-2.5-flash\"") { + t.Errorf("expected response model to fall back to request model") + } + if !strings.Contains(body, "gen_ai_response_model=\"gemini-2.5-flash-002\"") { + t.Errorf("expected explicit response model to be used") + } +} + +func TestMetricsHandler_ServesLabels(t *testing.T) { + t.Setenv(metricsEnabledEnvVar, "true") + tokenUsage.Reset() + RecordTokenUsage(TokenUsage{ + RequestModel: "claude-3-5-sonnet", ResponseModel: "claude-3-5-sonnet-20241022", + Provider: "anthropic", AgentName: "my-agent", ErrorType: "overloaded_error", + InputTokens: 10, OutputTokens: 5, + }) + + body := serveMetrics(t) + for _, want := range []string{ + "gen_ai_client_token_usage_count", + "gen_ai_token_type=\"input\"", + "gen_ai_token_type=\"output\"", + "gen_ai_operation_name=\"chat\"", + "gen_ai_provider_name=\"anthropic\"", + "gen_ai_request_model=\"claude-3-5-sonnet\"", + "gen_ai_response_model=\"claude-3-5-sonnet-20241022\"", + "gen_ai_agent_name=\"my-agent\"", + "error_type=\"overloaded_error\"", + } { + if !strings.Contains(body, want) { + t.Errorf("metrics output missing %q", want) + } + } +} + +func TestSemconvProviderName(t *testing.T) { + cases := map[string]string{ + "openai": "openai", + "azure_openai": "azure.ai.openai", + "anthropic": "anthropic", + "gemini": "gcp.gemini", + "gemini_vertex_ai": "gcp.vertex_ai", + "gemini_anthropic": "gcp.vertex_ai", + "bedrock": "aws.bedrock", + "ollama": "ollama", + "sap_ai_core": "sap_ai_core", + "some-custom": "some-custom", + } + for in, want := range cases { + if got := SemconvProviderName(in); got != want { + t.Errorf("SemconvProviderName(%q) = %q, want %q", in, got, want) + } + } +} + +func serveMetrics(t *testing.T) string { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/metrics", nil) + rec := httptest.NewRecorder() + MetricsHandler().ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + return rec.Body.String() +}