feat(go): add gated OTLP metrics, reconcile events, and log bridge (#2148)

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,17 @@ func main() {
 	}
 }
 
+// resolveModelLabels derives the gen_ai.request.model / gen_ai.provider.name
+// attributes for token-usage metrics from the agent config. The provider name
+// is mapped to its OpenTelemetry GenAI semconv value; both are empty strings
+// when no model is configured, so 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, "-", "_")

go/adk/pkg/a2a/executor.go

@@ -33,6 +33,11 @@ type KAgentExecutorConfig struct {
 	Stream         bool
 	AppName        string
 	Logger         logr.Logger
+	// ModelName and ProviderName label the GenAI token-usage metric
+	// (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
@@ -56,6 +61,8 @@ func NewKAgentExecutor(cfg KAgentExecutorConfig) *KAgentExecutor {
 	if cfg.SessionService != nil {
 		runnerConfig.SessionService = cfg.SessionService
 	}
+	modelName := cfg.ModelName
+	providerName := cfg.ProviderName
 	builtin := adka2a.NewExecutor(adka2a.ExecutorConfig{
 		RunnerConfig:       runnerConfig,
 		RunConfig:          runConfig,
@@ -65,6 +72,7 @@ func NewKAgentExecutor(cfg KAgentExecutorConfig) *KAgentExecutor {
 			if event.InvocationID != "" {
 				trace.SpanFromContext(ctx).SetAttributes(attribute.String("gcp.vertex.agent.invocation_id", event.InvocationID))
 			}
+			recordTokenUsage(ctx, event, modelName, providerName, cfg.AppName)
 			// Preserve the artifact's protocol type while giving current A2A clients a
 			// common ordering key. A2A #2129 will replace this with native artifact
 			// start/end generations and a task timeline.
@@ -88,6 +96,39 @@ func NewKAgentExecutor(cfg KAgentExecutorConfig) *KAgentExecutor {
 	}
 }
 
+// recordTokenUsage records GenAI token usage for a single agent 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 records one input + one output
+// observation per LLM call, not per stream chunk. Output combines candidate +
+// reasoning tokens, matching the Python runtime's accounting.
+func recordTokenUsage(ctx context.Context, adkEvent *adksession.Event, modelName, providerName, agentName string) {
+	if usage, ok := tokenUsageFromEvent(adkEvent, modelName, providerName, agentName); ok {
+		telemetry.RecordTokenUsage(ctx, usage)
+	}
+}
+
+// tokenUsageFromEvent derives GenAI token usage from a single ADK session event.
+// 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
+// yields one input + one output observation per LLM call, not per stream chunk.
+// Output combines candidate + reasoning tokens, matching the Python runtime's
+// accounting. ok=false when there is nothing to record.
+func tokenUsageFromEvent(adkEvent *adksession.Event, modelName, providerName, agentName string) (telemetry.TokenUsage, bool) {
+	usage := adkEvent.UsageMetadata
+	if usage == nil || adkEvent.Partial {
+		return telemetry.TokenUsage{}, false
+	}
+	return telemetry.TokenUsage{
+		RequestModel:  modelName,
+		ResponseModel: adkEvent.ModelVersion,
+		ProviderName:  providerName,
+		AgentName:     agentName,
+		InputTokens:   int64(usage.PromptTokenCount),
+		OutputTokens:  int64(usage.CandidatesTokenCount) + int64(usage.ThoughtsTokenCount),
+	}, true
+}
+
 // 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.

go/adk/pkg/a2a/executor_metrics_test.go

@@ -0,0 +1,67 @@
+package a2a
+
+import (
+	"testing"
+
+	adkmodel "google.golang.org/adk/v2/model"
+	adksession "google.golang.org/adk/v2/session"
+	"google.golang.org/genai"
+)
+
+// TestTokenUsageFromEvent_Values verifies the input/output token accounting:
+// output combines candidate + reasoning tokens, and the model/provider/agent
+// labels flow through.
+func TestTokenUsageFromEvent_Values(t *testing.T) {
+	usage := &genai.GenerateContentResponseUsageMetadata{
+		PromptTokenCount:     10,
+		CandidatesTokenCount: 5,
+		ThoughtsTokenCount:   3,
+	}
+	event := &adksession.Event{
+		LLMResponse: adkmodel.LLMResponse{
+			ModelVersion:  "gemini-2.5-flash",
+			UsageMetadata: usage,
+		},
+	}
+
+	got, ok := tokenUsageFromEvent(event, "gemini-2.5-flash", "gcp.gemini", "my-agent")
+	if !ok {
+		t.Fatal("expected usage to be recorded")
+	}
+	if got.InputTokens != 10 {
+		t.Errorf("input = %d, want 10", got.InputTokens)
+	}
+	if got.OutputTokens != 8 {
+		t.Errorf("output = %d, want 8 (candidates 5 + thoughts 3)", got.OutputTokens)
+	}
+	if got.RequestModel != "gemini-2.5-flash" || got.ProviderName != "gcp.gemini" || got.AgentName != "my-agent" {
+		t.Errorf("labels not propagated: %+v", got)
+	}
+}
+
+// TestTokenUsageFromEvent_SkipsPartialAndNil verifies streamed chunks that carry
+// usage metadata are skipped (one observation per LLM call), and events without
+// usage produce nothing.
+func TestTokenUsageFromEvent_SkipsPartialAndNil(t *testing.T) {
+	usage := &genai.GenerateContentResponseUsageMetadata{PromptTokenCount: 10, CandidatesTokenCount: 5}
+
+	if _, ok := tokenUsageFromEvent(
+		&adksession.Event{LLMResponse: adkmodel.LLMResponse{Partial: true, UsageMetadata: usage}},
+		"m", "p", "a",
+	); ok {
+		t.Error("partial event must not be recorded")
+	}
+
+	if _, ok := tokenUsageFromEvent(&adksession.Event{}, "m", "p", "a"); ok {
+		t.Error("event without usage metadata must not be recorded")
+	}
+}
+
+// TestRecordTokenUsage_NoopWithUninitializedRecorder verifies the executor
+// recording path is a no-op before telemetry metrics are initialized (metrics
+// disabled), so it cannot panic.
+func TestRecordTokenUsage_NoopWithUninitializedRecorder(t *testing.T) {
+	usage := &genai.GenerateContentResponseUsageMetadata{PromptTokenCount: 10, CandidatesTokenCount: 5}
+	event := &adksession.Event{LLMResponse: adkmodel.LLMResponse{UsageMetadata: usage}}
+	recordTokenUsage(t.Context(), event, "gemini-2.5-flash", "gcp.gemini", "my-agent")
+}

go/adk/pkg/config/config_usage.go

@@ -139,3 +139,14 @@ func getModelName(m adk.Model) string {
 		return "unknown"
 	}
 }
+
+// ModelName returns the configured model's identifier (e.g. "gpt-4o"), or ""
+// when no model is configured. This labels the gen_ai.request.model token-usage
+// metric attribute.
+func ModelName(m adk.Model) string {
+	name := getModelName(m)
+	if name == "unknown" {
+		return ""
+	}
+	return name
+}

go/adk/pkg/telemetry/metrics.go

@@ -0,0 +1,152 @@
+package telemetry
+
+import (
+	"context"
+	"net/url"
+
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
+	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp"
+	"go.opentelemetry.io/otel/metric"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/resource"
+)
+
+// GenAI token-usage instrumentation. The metric and attribute names follow
+// the OpenTelemetry GenAI semantic conventions, and the attribute set matches
+// the one upstream google-adk records for gen_ai.client.token.usage, so a
+// single dashboard works across the Go and Python runtimes.
+const (
+	metricGenAIClientTokenUsage = "gen_ai.client.token.usage"
+	genAIMeterScope             = "gcp.vertex.agent"
+
+	attrGenAITokenType     = "gen_ai.token.type"
+	attrGenAIRequestModel  = "gen_ai.request.model"
+	attrGenAIResponseModel = "gen_ai.response.model"
+	attrGenAIProviderName  = "gen_ai.provider.name"
+	attrGenAIAgentName     = "gen_ai.agent.name"
+
+	tokenTypeInput  = "input"
+	tokenTypeOutput = "output"
+)
+
+// tokenUsageHistogram records gen_ai.client.token.usage per LLM call. It is
+// set when the meter provider is initialized (metrics enabled); otherwise it
+// stays nil and recording is a cheap no-op, keeping the gate default-OFF.
+var tokenUsageHistogram metric.Int64Histogram
+
+// TokenUsage carries the per-LLM-call labels and token counts for one
+// recording on the gen_ai.client.token.usage histogram. RequestModel and
+// ProviderName are resolved at startup from the agent config; ResponseModel
+// falls back to RequestModel when a specific response model is unavailable.
+type TokenUsage struct {
+	RequestModel  string
+	ResponseModel string
+	ProviderName  string
+	AgentName     string
+	InputTokens   int64
+	OutputTokens  int64
+}
+
+// RecordTokenUsage records input + output token counts on the
+// gen_ai.client.token.usage histogram, one observation per token type.
+// Zero/negative counts are skipped, and nothing is recorded when metrics are
+// disabled or initialization failed (the instrument is nil).
+func RecordTokenUsage(ctx context.Context, usage TokenUsage) {
+	h := tokenUsageHistogram
+	if h == nil {
+		return
+	}
+	responseModel := usage.ResponseModel
+	if responseModel == "" {
+		responseModel = usage.RequestModel
+	}
+	base := []attribute.KeyValue{
+		attribute.String(attrGenAIRequestModel, usage.RequestModel),
+		attribute.String(attrGenAIResponseModel, responseModel),
+		attribute.String(attrGenAIProviderName, usage.ProviderName),
+		attribute.String(attrGenAIAgentName, usage.AgentName),
+	}
+	recordToken := func(tokenType string, count int64) {
+		if count <= 0 {
+			return
+		}
+		opts := append([]attribute.KeyValue{attribute.String(attrGenAITokenType, tokenType)}, base...)
+		h.Record(ctx, count, metric.WithAttributes(opts...))
+	}
+	recordToken(tokenTypeInput, usage.InputTokens)
+	recordToken(tokenTypeOutput, usage.OutputTokens)
+}
+
+// SemconvProviderName maps a kagent model type to its OpenTelemetry GenAI
+// gen_ai.provider.name value. Unknown types pass through unchanged so custom
+// providers keep their configured identity.
+func SemconvProviderName(modelType string) string {
+	switch modelType {
+	case "openai":
+		return "openai"
+	case "azure_openai":
+		return "azure.ai.openai"
+	case "anthropic":
+		return "anthropic"
+	case "gemini", "gemini_vertex_ai", "gemini_anthropic":
+		return "gcp.gemini"
+	case "bedrock":
+		return "aws.bedrock"
+	default:
+		return modelType
+	}
+}
+
+// newMeterProvider builds a MeterProvider with a periodic OTLP metric exporter,
+// sharing the endpoint/protocol resolution used by traces and logs.
+func newMeterProvider(ctx context.Context, res *resource.Resource) (*sdkmetric.MeterProvider, error) {
+	protocol := resolveOTLPProtocol("METRICS")
+	endpoint := resolveEndpoint("METRICS")
+
+	var exporter sdkmetric.Exporter
+	var err error
+	switch protocol {
+	case "http/protobuf":
+		var opts []otlpmetrichttp.Option
+		if endpoint != "" {
+			opts = append(opts, otlpmetrichttp.WithEndpointURL(endpoint))
+		}
+		exporter, err = otlpmetrichttp.New(ctx, opts...)
+	default:
+		var opts []otlpmetricgrpc.Option
+		if endpoint != "" {
+			if u, parseErr := url.Parse(endpoint); parseErr == nil && u.Scheme != "" && u.Host != "" {
+				opts = append(opts, otlpmetricgrpc.WithEndpointURL(u.String()))
+			} else {
+				opts = append(opts, otlpmetricgrpc.WithEndpoint(endpoint))
+			}
+		}
+		exporter, err = otlpmetricgrpc.New(ctx, opts...)
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	return sdkmetric.NewMeterProvider(
+		sdkmetric.WithReader(sdkmetric.NewPeriodicReader(exporter)),
+		sdkmetric.WithResource(res),
+	), nil
+}
+
+// initTokenUsageRecorder binds the gen_ai.client.token.usage histogram to the
+// given meter scope. It is called after setting the global meter provider.
+func initTokenUsageRecorder(mp *sdkmetric.MeterProvider) {
+	meter := mp.Meter(genAIMeterScope)
+	var err error
+	tokenUsageHistogram, err = meter.Int64Histogram(
+		metricGenAIClientTokenUsage,
+		metric.WithUnit("{token}"),
+		metric.WithDescription("Number of input and output tokens used by GenAI requests."),
+	)
+	if err != nil {
+		otel.Handle(err)
+		tokenUsageHistogram = nil
+	}
+}