diff --git a/go/adk/cmd/main.go b/go/adk/cmd/main.go index 6927b4d38e..db043cf2f4 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" "github.com/kagent-dev/kagent/go/core/pkg/env" "github.com/kagent-dev/kagent/go/pkg/logging" ) @@ -190,12 +191,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. @@ -234,6 +238,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 *slog.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 b5f3dfeb02..f20801cab6 100644 --- a/go/adk/pkg/a2a/executor.go +++ b/go/adk/pkg/a2a/executor.go @@ -34,6 +34,11 @@ type KAgentExecutorConfig struct { Stream bool AppName string Logger *slog.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 @@ -57,6 +62,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, @@ -66,6 +73,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. @@ -89,6 +97,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. 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..072c7d7a86 --- /dev/null +++ b/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") +} diff --git a/go/adk/pkg/config/config_usage.go b/go/adk/pkg/config/config_usage.go index 69c86264c6..e2a0cde5f7 100644 --- a/go/adk/pkg/config/config_usage.go +++ b/go/adk/pkg/config/config_usage.go @@ -111,3 +111,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 +} diff --git a/go/adk/pkg/telemetry/metrics.go b/go/adk/pkg/telemetry/metrics.go new file mode 100644 index 0000000000..e46529b0af --- /dev/null +++ b/go/adk/pkg/telemetry/metrics.go @@ -0,0 +1,143 @@ +package telemetry + +import ( + "context" + "fmt" + "sync/atomic" + + "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 atomic.Pointer[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) { + histogram := tokenUsageHistogram.Load() + if histogram == 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...) + (*histogram).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") + + var exporter sdkmetric.Exporter + var err error + switch protocol { + case "http/protobuf": + exporter, err = otlpmetrichttp.New(ctx) + case "grpc": + exporter, err = otlpmetricgrpc.New(ctx) + default: + return nil, fmt.Errorf("unsupported OTLP metrics protocol %q", protocol) + } + if err != nil { + return nil, fmt.Errorf("create metric exporter: %w", 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) + histogram, 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.Store(nil) + return + } + tokenUsageHistogram.Store(&histogram) +} diff --git a/go/adk/pkg/telemetry/metrics_init_test.go b/go/adk/pkg/telemetry/metrics_init_test.go new file mode 100644 index 0000000000..8deff95c42 --- /dev/null +++ b/go/adk/pkg/telemetry/metrics_init_test.go @@ -0,0 +1,142 @@ +package telemetry + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "sync" + "testing" + + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" + logglobal "go.opentelemetry.io/otel/log/global" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/resource" +) + +func TestMetricEndpointPaths(t *testing.T) { + for _, test := range []struct{ name, endpointPath, signalPath, wantPath string }{ + {name: "generic endpoint", wantPath: "/v1/metrics"}, + {name: "generic prefix", endpointPath: "/collector", wantPath: "/collector/v1/metrics"}, + {name: "signal override", signalPath: "/custom-metrics", wantPath: "/custom-metrics"}, + } { + t.Run(test.name, func(t *testing.T) { + paths := make(chan string, 4) + collector := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + paths <- request.URL.Path + writer.Header().Set("Content-Type", "application/x-protobuf") + })) + t.Cleanup(collector.Close) + t.Setenv("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL", "http/protobuf") + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", collector.URL+test.endpointPath) + t.Setenv("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", "") + if test.signalPath != "" { + t.Setenv("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", collector.URL+test.signalPath) + } + provider, err := newMeterProvider(t.Context(), resource.Empty()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, provider.Shutdown(context.Background())) }) + histogram, err := provider.Meter("test").Int64Histogram("tokens") + require.NoError(t, err) + histogram.Record(t.Context(), 10) + require.NoError(t, provider.ForceFlush(t.Context())) + select { + case path := <-paths: + require.Equal(t, test.wantPath, path) + default: + t.Fatal("metrics were not exported") + } + }) + } +} + +func TestInitSignalGates(t *testing.T) { + collector := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + writer.Header().Set("Content-Type", "application/x-protobuf") + })) + t.Cleanup(collector.Close) + for _, test := range []struct { + name, traces, logs, metrics, protocol string + wantErr bool + }{ + {name: "default off", protocol: "invalid"}, + {name: "metrics only", metrics: "true"}, + {name: "traces only", traces: "true"}, + {name: "logs only", logs: "true"}, + {name: "all signals", traces: "true", logs: "true", metrics: "true"}, + {name: "metric failure preserves globals", traces: "true", logs: "true", metrics: "true", protocol: "invalid", wantErr: true}, + } { + t.Run(test.name, func(t *testing.T) { + t.Setenv("OTEL_TRACING_ENABLED", test.traces) + t.Setenv("OTEL_LOGGING_ENABLED", test.logs) + t.Setenv("OTEL_METRICS_ENABLED", test.metrics) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", collector.URL) + t.Setenv("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL", test.protocol) + t.Setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "http/protobuf") + previousTrace := otel.GetTracerProvider() + previousLog := logglobal.GetLoggerProvider() + previousMeter := otel.GetMeterProvider() + previousPropagator := otel.GetTextMapPropagator() + previousHistogram := tokenUsageHistogram.Load() + t.Cleanup(func() { + otel.SetTracerProvider(previousTrace) + logglobal.SetLoggerProvider(previousLog) + otel.SetMeterProvider(previousMeter) + otel.SetTextMapPropagator(previousPropagator) + tokenUsageHistogram.Store(previousHistogram) + }) + shutdown, enabled, err := Init(t.Context(), "test", "test") + if test.wantErr { + require.Error(t, err) + require.Nil(t, shutdown) + } else { + require.NoError(t, err) + require.Equal(t, test.traces == "true" || test.logs == "true" || test.metrics == "true", enabled) + require.NoError(t, shutdown(t.Context())) + } + if test.traces != "true" || test.wantErr { + require.Same(t, previousTrace, otel.GetTracerProvider()) + } + if test.logs != "true" || test.wantErr { + require.Same(t, previousLog, logglobal.GetLoggerProvider()) + } + if test.metrics != "true" || test.wantErr { + require.Same(t, previousMeter, otel.GetMeterProvider()) + } + }) + } +} + +func TestTokenRecorderConcurrentInitialization(t *testing.T) { + previous := tokenUsageHistogram.Load() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(sdkmetric.NewManualReader())) + t.Cleanup(func() { + tokenUsageHistogram.Store(previous) + require.NoError(t, provider.Shutdown(context.Background())) + }) + var workers sync.WaitGroup + for range 4 { + workers.Go(func() { + for range 100 { + initTokenUsageRecorder(provider) + RecordTokenUsage(t.Context(), TokenUsage{InputTokens: 1}) + } + }) + } + workers.Wait() +} + +func TestShutdownContinuesAfterTraceError(t *testing.T) { + traceErr := errors.New("trace shutdown failed") + metricErr := errors.New("metric shutdown failed") + var stopped []string + err := shutdownProviders(t.Context(), []func(context.Context) error{ + func(context.Context) error { stopped = append(stopped, "traces"); return traceErr }, + func(context.Context) error { stopped = append(stopped, "logs"); return nil }, + func(context.Context) error { stopped = append(stopped, "metrics"); return metricErr }, + }) + require.ErrorIs(t, err, traceErr) + require.ErrorIs(t, err, metricErr) + require.Equal(t, []string{"traces", "logs", "metrics"}, stopped) +} diff --git a/go/adk/pkg/telemetry/metrics_test.go b/go/adk/pkg/telemetry/metrics_test.go new file mode 100644 index 0000000000..cf1781724a --- /dev/null +++ b/go/adk/pkg/telemetry/metrics_test.go @@ -0,0 +1,176 @@ +package telemetry + +import ( + "context" + "testing" + + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" +) + +// withTokenRecorder installs a manual meter reader so tests can inspect the +// gen_ai.client.token.usage histogram after RecordTokenUsage calls. +func withTokenRecorder(t *testing.T) *sdkmetric.ManualReader { + t.Helper() + prev := tokenUsageHistogram.Load() + reader := sdkmetric.NewManualReader() + mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + initTokenUsageRecorder(mp) + t.Cleanup(func() { + tokenUsageHistogram.Store(prev) + _ = mp.Shutdown(context.Background()) + }) + return reader +} + +// tokenUsagePoint returns the unique data point on gen_ai.client.token.usage +// whose attributes carry the given token type, failing the test otherwise. +func tokenUsagePoint(t *testing.T, reader *sdkmetric.ManualReader, tokenType string) metricdata.HistogramDataPoint[int64] { + t.Helper() + var rm metricdata.ResourceMetrics + if err := reader.Collect(context.Background(), &rm); err != nil { + t.Fatalf("collect: %v", err) + } + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name != metricGenAIClientTokenUsage { + continue + } + hist, ok := m.Data.(metricdata.Histogram[int64]) + if !ok { + t.Fatalf("expected histogram data, got %T", m.Data) + } + for _, dp := range hist.DataPoints { + if v, _ := dp.Attributes.Value(attribute.Key(attrGenAITokenType)); v.AsString() == tokenType { + return dp + } + } + } + } + t.Fatalf("no data point with gen_ai.token.type=%q on %s", tokenType, metricGenAIClientTokenUsage) + return metricdata.HistogramDataPoint[int64]{} +} + +func attrString(t *testing.T, dp metricdata.HistogramDataPoint[int64], key string) string { + t.Helper() + v, ok := dp.Attributes.Value(attribute.Key(key)) + if !ok { + t.Fatalf("attribute %q not set on data point", key) + } + s := v.AsString() + if s == "" { + t.Fatalf("attribute %q empty on data point", key) + } + return s +} + +func TestRecordTokenUsage_RecordsInputAndOutput(t *testing.T) { + reader := withTokenRecorder(t) + + RecordTokenUsage(context.Background(), TokenUsage{ + RequestModel: "gpt-4o", + ResponseModel: "gpt-4o-2024-05-13", + ProviderName: "openai", + AgentName: "my-agent", + InputTokens: 10, + OutputTokens: 5, + }) + + input := tokenUsagePoint(t, reader, tokenTypeInput) + if got := input.Count; got != 1 { + t.Fatalf("input count = %d, want 1", got) + } + if got := input.Sum; got != 10 { + t.Fatalf("input sum = %d, want 10", got) + } + if got := attrString(t, input, attrGenAIRequestModel); got != "gpt-4o" { + t.Errorf("input request model = %q", got) + } + if got := attrString(t, input, attrGenAIResponseModel); got != "gpt-4o-2024-05-13" { + t.Errorf("input response model = %q", got) + } + if got := attrString(t, input, attrGenAIProviderName); got != "openai" { + t.Errorf("input provider = %q", got) + } + if got := attrString(t, input, attrGenAIAgentName); got != "my-agent" { + t.Errorf("input agent = %q", got) + } + + output := tokenUsagePoint(t, reader, tokenTypeOutput) + if got := output.Sum; got != 5 { + t.Fatalf("output sum = %d, want 5", got) + } + if got := output.Count; got != 1 { + t.Fatalf("output count = %d, want 1", got) + } +} + +func TestRecordTokenUsage_ResponseModelFallsBackToRequest(t *testing.T) { + reader := withTokenRecorder(t) + + RecordTokenUsage(context.Background(), TokenUsage{ + RequestModel: "gemini-2.5-flash", + ProviderName: "gcp.gemini", + AgentName: "a", + InputTokens: 3, + }) + + pt := tokenUsagePoint(t, reader, tokenTypeInput) + if got := attrString(t, pt, attrGenAIResponseModel); got != "gemini-2.5-flash" { + t.Errorf("response model = %q, want fallback to request model", got) + } +} + +func TestRecordTokenUsage_SkipsZeroAndNegative(t *testing.T) { + reader := withTokenRecorder(t) + + // All-zero, and negative/zero mixes, must not create data points. + RecordTokenUsage(context.Background(), TokenUsage{ + RequestModel: "gpt-4o", ProviderName: "openai", AgentName: "my-agent", + }) + RecordTokenUsage(context.Background(), TokenUsage{ + RequestModel: "gpt-4o", ProviderName: "openai", InputTokens: -1, + }) + + var rm metricdata.ResourceMetrics + if err := reader.Collect(context.Background(), &rm); err != nil { + t.Fatalf("collect: %v", err) + } + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name == metricGenAIClientTokenUsage { + t.Fatalf("expected no data on %s for zero/negative counts", metricGenAIClientTokenUsage) + } + } + } +} + +func TestRecordTokenUsage_NoopWhenNotInitialized(t *testing.T) { + prev := tokenUsageHistogram.Load() + tokenUsageHistogram.Store(nil) + t.Cleanup(func() { tokenUsageHistogram.Store(prev) }) + + RecordTokenUsage(context.Background(), TokenUsage{ + RequestModel: "gpt-4o", ProviderName: "openai", InputTokens: 10, OutputTokens: 5, + }) // must not panic when metrics are disabled +} + +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.gemini", + "gemini_anthropic": "gcp.gemini", + "bedrock": "aws.bedrock", + "ollama": "ollama", + "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) + } + } +} diff --git a/go/adk/pkg/telemetry/tracing.go b/go/adk/pkg/telemetry/tracing.go index 3a1d22a936..78fdd544ce 100644 --- a/go/adk/pkg/telemetry/tracing.go +++ b/go/adk/pkg/telemetry/tracing.go @@ -2,6 +2,7 @@ package telemetry import ( "context" + "errors" "fmt" "net/url" "os" @@ -14,13 +15,13 @@ import ( "go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp" "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc" "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp" + logglobal "go.opentelemetry.io/otel/log/global" "go.opentelemetry.io/otel/propagation" sdklog "go.opentelemetry.io/otel/sdk/log" "go.opentelemetry.io/otel/sdk/resource" sdktrace "go.opentelemetry.io/otel/sdk/trace" semconv "go.opentelemetry.io/otel/semconv/v1.36.0" "go.opentelemetry.io/otel/trace" - adktelemetry "google.golang.org/adk/v2/telemetry" ) // SetKAgentSpanAttributes sets kagent span attributes in the OpenTelemetry context @@ -49,14 +50,19 @@ func StartInvocationSpan(ctx context.Context) (context.Context, trace.Span) { // 3s and is configurable via KAGENT_TRACE_FLUSH_TIMEOUT_MS. func ForceFlush(ctx context.Context) { type flusher interface{ ForceFlush(context.Context) error } - fp, ok := otel.GetTracerProvider().(flusher) - if !ok { - return - } flushCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), flushTimeout()) defer cancel() - if err := fp.ForceFlush(flushCtx); err != nil { - otel.Handle(err) + if fp, ok := otel.GetTracerProvider().(flusher); ok { + if err := fp.ForceFlush(flushCtx); err != nil { + otel.Handle(err) + } + } + // A periodic metric reader may never fire before the actor is suspended, + // so drain it alongside spans (see newMeterProvider). + if mp, ok := otel.GetMeterProvider().(flusher); ok { + if err := mp.ForceFlush(flushCtx); err != nil { + otel.Handle(err) + } } } @@ -88,6 +94,17 @@ func Init(ctx context.Context, serviceName string, serviceNamespace string) (shu if !isTelemetryEnabled() { return func(context.Context) error { return nil }, false, nil } + var shutdowns []func(context.Context) error + shutdownAll := func(shutdownCtx context.Context) error { + return shutdownProviders(shutdownCtx, shutdowns) + } + defer func() { + if err != nil { + cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + defer cancel() + err = errors.Join(err, shutdownAll(cleanupCtx)) + } + }() telemetryResource, err := newTelemetryResource(ctx, serviceName, serviceNamespace) if err != nil { @@ -96,39 +113,60 @@ func Init(ctx context.Context, serviceName string, serviceNamespace string) (shu tracingEnabled := strings.EqualFold(strings.TrimSpace(os.Getenv("OTEL_TRACING_ENABLED")), "true") loggingEnabled := strings.EqualFold(strings.TrimSpace(os.Getenv("OTEL_LOGGING_ENABLED")), "true") - otelOpts := []adktelemetry.Option{adktelemetry.WithResource(telemetryResource)} + metricsEnabled := strings.EqualFold(strings.TrimSpace(os.Getenv("OTEL_METRICS_ENABLED")), "true") + var tracerProvider *sdktrace.TracerProvider + var loggerProvider *sdklog.LoggerProvider if tracingEnabled { - tracerProvider, tpErr := newTracerProvider(ctx, telemetryResource) - if tpErr != nil { - return nil, true, tpErr + tracerProvider, err = newTracerProvider(ctx, telemetryResource) + if err != nil { + return nil, true, err } - otelOpts = append(otelOpts, adktelemetry.WithTracerProvider(tracerProvider)) + shutdowns = append(shutdowns, tracerProvider.Shutdown) } if loggingEnabled { - loggerProvider, lpErr := newLoggerProvider(ctx, telemetryResource) - if lpErr != nil { - return nil, true, lpErr + loggerProvider, err = newLoggerProvider(ctx, telemetryResource) + if err != nil { + return nil, true, err } - otelOpts = append(otelOpts, adktelemetry.WithLoggerProvider(loggerProvider)) + shutdowns = append(shutdowns, loggerProvider.Shutdown) } - telemetryProviders, telErr := adktelemetry.New(ctx, otelOpts...) - if telErr != nil { - return nil, true, telErr + if metricsEnabled { + meterProvider, mpErr := newMeterProvider(ctx, telemetryResource) + if mpErr != nil { + return nil, true, mpErr + } + otel.SetMeterProvider(meterProvider) + initTokenUsageRecorder(meterProvider) + shutdowns = append(shutdowns, meterProvider.Shutdown) + } + if tracerProvider != nil { + otel.SetTracerProvider(tracerProvider) + } + if loggerProvider != nil { + logglobal.SetLoggerProvider(loggerProvider) } - telemetryProviders.SetGlobalOtelProviders() otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator( propagation.TraceContext{}, propagation.Baggage{}, )) - return telemetryProviders.Shutdown, true, nil + return shutdownAll, true, nil +} + +func shutdownProviders(ctx context.Context, shutdowns []func(context.Context) error) error { + var shutdownErrors []error + for _, shutdown := range shutdowns { + shutdownErrors = append(shutdownErrors, shutdown(ctx)) + } + return errors.Join(shutdownErrors...) } func isTelemetryEnabled() bool { return strings.EqualFold(strings.TrimSpace(os.Getenv("OTEL_TRACING_ENABLED")), "true") || - strings.EqualFold(strings.TrimSpace(os.Getenv("OTEL_LOGGING_ENABLED")), "true") + strings.EqualFold(strings.TrimSpace(os.Getenv("OTEL_LOGGING_ENABLED")), "true") || + strings.EqualFold(strings.TrimSpace(os.Getenv("OTEL_METRICS_ENABLED")), "true") } // resolveOTLPProtocol returns the OTLP protocol for the given signal, diff --git a/go/core/internal/controller/mcpserver/reconciler.go b/go/core/internal/controller/mcpserver/reconciler.go index 78666b6dd4..3be7062800 100644 --- a/go/core/internal/controller/mcpserver/reconciler.go +++ b/go/core/internal/controller/mcpserver/reconciler.go @@ -30,10 +30,12 @@ import ( toolservice "github.com/kagent-dev/kagent/go/core/internal/service/tool" "github.com/kagent-dev/kagent/go/pkg/logging" kmcp "github.com/kagent-dev/kmcp/api/v1alpha1" + corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" apiMeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/tools/events" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller" @@ -41,6 +43,11 @@ import ( ) const ( + eventReconcileFailed = "ReconcileFailed" + eventValidationFailed = "ValidationFailed" + eventToolsDiscovered = "ToolsDiscovered" + actionReconcile = "Reconcile" + actionDiscover = "Discover" mcpServerGroupKind = "MCPServer.kagent.dev" refreshInterval = 5 * time.Minute readinessPollInterval = 10 * time.Second @@ -65,10 +72,11 @@ type Reconciler struct { client client.Client discoverer ToolDiscoverer catalog CatalogStore + recorder events.EventRecorder } -func New(client client.Client, discoverer ToolDiscoverer, catalog CatalogStore) *Reconciler { - return &Reconciler{client: client, discoverer: discoverer, catalog: catalog} +func New(client client.Client, discoverer ToolDiscoverer, catalog CatalogStore, recorder events.EventRecorder) *Reconciler { + return &Reconciler{client: client, discoverer: discoverer, catalog: catalog, recorder: recorder} } func (r *Reconciler) SetupWithManager(manager ctrl.Manager) error { @@ -119,6 +127,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, request reconcile.Request) ( Ref: request.NamespacedName, GroupKind: mcpServerGroupKind, }) if err != nil { + r.recordEvent(server, corev1.EventTypeWarning, eventReconcileFailed, actionReconcile, + "failed to discover MCPServer tools: %v", err) catalogErr := r.updateCatalog(ctx, server, nil, false) return reconcile.Result{}, errors.Join( fmt.Errorf("discover MCPServer tools: %w", err), @@ -128,6 +138,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, request reconcile.Request) ( discovered, err := toolcatalog.NormalizeTools(tools) if err != nil { + r.recordEvent(server, corev1.EventTypeWarning, eventValidationFailed, actionReconcile, + "invalid MCPServer tool discovery: %v", err) catalogErr := r.updateCatalog(ctx, server, nil, false) return reconcile.Result{}, errors.Join( err, @@ -135,11 +147,23 @@ func (r *Reconciler) Reconcile(ctx context.Context, request reconcile.Request) ( ) } if err := r.updateCatalog(ctx, server, discovered, true); err != nil { + r.recordEvent(server, corev1.EventTypeWarning, eventReconcileFailed, actionReconcile, + "failed to update MCPServer tool catalog: %v", err) return reconcile.Result{}, fmt.Errorf("update MCPServer tool catalog: %w", err) } + r.recordEvent(server, corev1.EventTypeNormal, eventToolsDiscovered, actionDiscover, "Discovered %d MCP tools", len(discovered)) return reconcile.Result{RequeueAfter: refreshInterval}, nil } +// recordEvent emits a Kubernetes Event against the reconciled object. It is a +// no-op when no recorder is wired on this reconciler. +func (r *Reconciler) recordEvent(object client.Object, eventType, reason, action, messageFmt string, args ...any) { + if r.recorder == nil { + return + } + r.recorder.Eventf(object, nil, eventType, reason, action, messageFmt, args...) +} + func isReady(server *kmcp.MCPServer) bool { condition := apiMeta.FindStatusCondition(server.Status.Conditions, string(kmcp.MCPServerConditionReady)) return condition != nil && condition.Status == metav1.ConditionTrue && condition.ObservedGeneration == server.Generation diff --git a/go/core/internal/controller/mcpserver/reconciler_test.go b/go/core/internal/controller/mcpserver/reconciler_test.go index 2bbaad0eb6..a210370b64 100644 --- a/go/core/internal/controller/mcpserver/reconciler_test.go +++ b/go/core/internal/controller/mcpserver/reconciler_test.go @@ -19,6 +19,7 @@ package mcpserver import ( "context" "errors" + "strings" "testing" "time" @@ -31,6 +32,7 @@ import ( "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/tools/events" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" @@ -74,7 +76,7 @@ func TestReconcileDiscoversReadyMCPServer(t *testing.T) { }} catalog := &fakeCatalog{} - result, err := New(testClient(t, server), discoverer, catalog).Reconcile(t.Context(), ctrl.Request{ + result, err := New(testClient(t, server), discoverer, catalog, nil).Reconcile(t.Context(), ctrl.Request{ NamespacedName: client.ObjectKeyFromObject(server), }) if err != nil { @@ -112,7 +114,7 @@ func TestReconcileWaitsForCurrentReadyCondition(t *testing.T) { discoverer := &fakeDiscoverer{} catalog := &fakeCatalog{tools: []*v1alpha3.MCPTool{{Name: "stale"}}} - result, err := New(testClient(t, server), discoverer, catalog).Reconcile(t.Context(), ctrl.Request{ + result, err := New(testClient(t, server), discoverer, catalog, nil).Reconcile(t.Context(), ctrl.Request{ NamespacedName: client.ObjectKeyFromObject(server), }) if err != nil { @@ -133,7 +135,7 @@ func TestReconcileClearsCatalogAfterDiscoveryFailure(t *testing.T) { discoverer := &fakeDiscoverer{err: errors.New("unavailable")} catalog := &fakeCatalog{tools: []*v1alpha3.MCPTool{{Name: "stale"}}} - _, err := New(testClient(t, server), discoverer, catalog).Reconcile(t.Context(), ctrl.Request{ + _, err := New(testClient(t, server), discoverer, catalog, nil).Reconcile(t.Context(), ctrl.Request{ NamespacedName: client.ObjectKeyFromObject(server), }) if err == nil { @@ -148,7 +150,7 @@ func TestReconcileDeletesCatalogProjection(t *testing.T) { catalog := &fakeCatalog{} request := ctrl.Request{NamespacedName: types.NamespacedName{Namespace: "test", Name: "gone"}} - if _, err := New(testClient(t), &fakeDiscoverer{}, catalog).Reconcile(t.Context(), request); err != nil { + if _, err := New(testClient(t), &fakeDiscoverer{}, catalog, nil).Reconcile(t.Context(), request); err != nil { t.Fatalf("Reconcile() error = %v", err) } want := "test/gone|" + mcpServerGroupKind @@ -199,3 +201,86 @@ func testClient(t *testing.T, objects ...client.Object) client.Client { func apiMetaTestMapper() *apiMeta.DefaultRESTMapper { return apiMeta.NewDefaultRESTMapper([]schema.GroupVersion{kmcp.GroupVersion}) } + +// TestReconcileEmitsToolsDiscovered verifies a Normal ToolsDiscovered event is +// emitted when a ready MCPServer is successfully discovered. +func TestReconcileEmitsToolsDiscovered(t *testing.T) { + server := readyServer() + discoverer := &fakeDiscoverer{tools: []toolservice.MCPAppTool{{Name: "zeta"}}} + recorder := events.NewFakeRecorder(1) + + result, err := New(testClient(t, server), discoverer, &fakeCatalog{}, recorder).Reconcile(t.Context(), ctrl.Request{ + NamespacedName: client.ObjectKeyFromObject(server), + }) + if err != nil { + t.Fatalf("Reconcile() error = %v", err) + } + if result.RequeueAfter != 5*time.Minute { + t.Fatalf("Reconcile() requeue = %s, want 5m", result.RequeueAfter) + } + + select { + case ev := <-recorder.Events: + if !strings.Contains(ev, "Normal ToolsDiscovered ") { + t.Fatalf("unexpected event: %q", ev) + } + default: + t.Fatal("expected a ToolsDiscovered event, got none") + } +} + +// TestReconcileEmitsValidationFailed verifies a malformed discovery triggers a +// Warning ValidationFailed event. +func TestReconcileEmitsValidationFailed(t *testing.T) { + server := readyServer() + discoverer := &fakeDiscoverer{tools: []toolservice.MCPAppTool{{Name: " "}}} + recorder := events.NewFakeRecorder(1) + + if _, err := New(testClient(t, server), discoverer, &fakeCatalog{}, recorder).Reconcile(t.Context(), ctrl.Request{ + NamespacedName: client.ObjectKeyFromObject(server), + }); err == nil { + t.Fatal("expected validation error") + } + + select { + case event := <-recorder.Events: + if !strings.Contains(event, "Warning ValidationFailed ") { + t.Fatalf("unexpected event: %q", event) + } + default: + t.Fatal("expected a ValidationFailed event, got none") + } +} + +// TestReconcileNoEventsWithoutRecorder verifies event emission is skipped when +// no recorder is wired, keeping zero behavioral change by default. +func TestReconcileNoEventsWithoutRecorder(t *testing.T) { + server := readyServer() + discoverer := &fakeDiscoverer{tools: []toolservice.MCPAppTool{{Name: "zeta"}}} + + if _, err := New(testClient(t, server), discoverer, &fakeCatalog{}, nil). + Reconcile(t.Context(), ctrl.Request{ + NamespacedName: client.ObjectKeyFromObject(server), + }); err != nil { + t.Fatalf("Reconcile() error = %v", err) + } +} + +func TestReconcileEmitsEmptyDiscovery(t *testing.T) { + server := readyServer() + recorder := events.NewFakeRecorder(1) + _, err := New(testClient(t, server), &fakeDiscoverer{}, &fakeCatalog{}, recorder).Reconcile(t.Context(), ctrl.Request{ + NamespacedName: client.ObjectKeyFromObject(server), + }) + if err != nil { + t.Fatal(err) + } + select { + case event := <-recorder.Events: + if !strings.Contains(event, "Normal ToolsDiscovered Discovered 0 MCP tools") { + t.Fatalf("unexpected event: %q", event) + } + default: + t.Fatal("expected an event for successful empty discovery") + } +} diff --git a/go/core/internal/controller/remotemcpserver/reconciler.go b/go/core/internal/controller/remotemcpserver/reconciler.go index 69c4ea38c3..5d3444895c 100644 --- a/go/core/internal/controller/remotemcpserver/reconciler.go +++ b/go/core/internal/controller/remotemcpserver/reconciler.go @@ -34,6 +34,7 @@ import ( apiMeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/tools/events" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" @@ -43,9 +44,14 @@ import ( ) const ( - conditionAccepted = "Accepted" - remoteGroupKind = "RemoteMCPServer.kagent.dev" - refreshInterval = 5 * time.Minute + eventReconcileFailed = "ReconcileFailed" + eventValidationFailed = "ValidationFailed" + eventToolsDiscovered = "ToolsDiscovered" + actionReconcile = "Reconcile" + actionDiscover = "Discover" + conditionAccepted = "Accepted" + remoteGroupKind = "RemoteMCPServer.kagent.dev" + refreshInterval = 5 * time.Minute ) // ToolDiscoverer returns the tools currently advertised by one MCP server. @@ -66,10 +72,11 @@ type Reconciler struct { client client.Client discoverer ToolDiscoverer catalog CatalogStore + recorder events.EventRecorder } -func New(client client.Client, discoverer ToolDiscoverer, catalog CatalogStore) *Reconciler { - return &Reconciler{client: client, discoverer: discoverer, catalog: catalog} +func New(client client.Client, discoverer ToolDiscoverer, catalog CatalogStore, recorder events.EventRecorder) *Reconciler { + return &Reconciler{client: client, discoverer: discoverer, catalog: catalog, recorder: recorder} } func (r *Reconciler) SetupWithManager(manager ctrl.Manager) error { @@ -93,6 +100,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, request reconcile.Request) ( Ref: request.NamespacedName, GroupKind: remoteGroupKind, }) if err != nil { + r.recordEvent(server, corev1.EventTypeWarning, eventReconcileFailed, actionReconcile, + "failed to discover RemoteMCPServer tools: %v", err) statusErr := r.updateStatus(ctx, server, nil, metav1.ConditionFalse, "DiscoveryFailed", err.Error()) catalogErr := r.updateCatalog(ctx, server, nil, false) return reconcile.Result{}, errors.Join( @@ -104,6 +113,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, request reconcile.Request) ( discovered, err := toolcatalog.NormalizeTools(tools) if err != nil { + r.recordEvent(server, corev1.EventTypeWarning, eventValidationFailed, actionReconcile, + "invalid RemoteMCPServer tool discovery: %v", err) statusErr := r.updateStatus(ctx, server, nil, metav1.ConditionFalse, "InvalidDiscovery", err.Error()) catalogErr := r.updateCatalog(ctx, server, nil, false) return reconcile.Result{}, errors.Join( @@ -112,16 +123,36 @@ func (r *Reconciler) Reconcile(ctx context.Context, request reconcile.Request) ( wrapError("clear invalid RemoteMCPServer tool catalog", catalogErr), ) } + // Only surface a Normal event on the transition to a successful discovery, + // so the periodic refresh timer does not re-emit it forever. + accepted := apiMeta.FindStatusCondition(server.Status.Conditions, conditionAccepted) + firstDiscovery := accepted == nil || accepted.Status != metav1.ConditionTrue || accepted.ObservedGeneration != server.Generation message := fmt.Sprintf("Discovered %d MCP tools", len(discovered)) + if err := r.updateCatalog(ctx, server, discovered, true); err != nil { + r.recordEvent(server, corev1.EventTypeWarning, eventReconcileFailed, actionReconcile, + "failed to update RemoteMCPServer tool catalog: %v", err) + return reconcile.Result{}, fmt.Errorf("update RemoteMCPServer tool catalog: %w", err) + } if err := r.updateStatus(ctx, server, discovered, metav1.ConditionTrue, "DiscoverySucceeded", message); err != nil { + r.recordEvent(server, corev1.EventTypeWarning, eventReconcileFailed, actionReconcile, + "failed to update RemoteMCPServer discovery status: %v", err) return reconcile.Result{}, fmt.Errorf("update RemoteMCPServer discovery status: %w", err) } - if err := r.updateCatalog(ctx, server, discovered, true); err != nil { - return reconcile.Result{}, fmt.Errorf("update RemoteMCPServer tool catalog: %w", err) + if firstDiscovery { + r.recordEvent(server, corev1.EventTypeNormal, eventToolsDiscovered, actionDiscover, "Discovered %d MCP tools", len(discovered)) } return reconcile.Result{RequeueAfter: refreshInterval}, nil } +// recordEvent emits a Kubernetes Event against the reconciled object. It is a +// no-op when no recorder is wired on this reconciler. +func (r *Reconciler) recordEvent(object client.Object, eventType, reason, action, messageFmt string, args ...any) { + if r.recorder == nil { + return + } + r.recorder.Eventf(object, nil, eventType, reason, action, messageFmt, args...) +} + func (r *Reconciler) updateCatalog(ctx context.Context, server *v1alpha3.RemoteMCPServer, tools []*v1alpha3.MCPTool, connected bool) error { name := client.ObjectKeyFromObject(server).String() var lastConnected *time.Time diff --git a/go/core/internal/controller/remotemcpserver/reconciler_test.go b/go/core/internal/controller/remotemcpserver/reconciler_test.go index 4dde94624b..24ff2d002d 100644 --- a/go/core/internal/controller/remotemcpserver/reconciler_test.go +++ b/go/core/internal/controller/remotemcpserver/reconciler_test.go @@ -19,6 +19,7 @@ package remotemcpserver import ( "context" "errors" + "github.com/stretchr/testify/require" "testing" "time" @@ -30,6 +31,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/tools/events" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" @@ -42,15 +44,17 @@ type fakeDiscoverer struct { } type fakeCatalog struct { - server *database.ToolServer - tools []*v1alpha3.MCPTool - deleted string + server *database.ToolServer + tools []*v1alpha3.MCPTool + err error + deletedTools string + deleted string } func (f *fakeCatalog) RefreshToolServer(_ context.Context, server *database.ToolServer, tools ...*v1alpha3.MCPTool) error { f.server = server f.tools = tools - return nil + return f.err } func (f *fakeCatalog) DeleteToolServer(_ context.Context, name, groupKind string) error { @@ -71,7 +75,7 @@ func TestReconcilePublishesSortedDiscovery(t *testing.T) { {Name: "alpha", Description: "first"}, }} catalog := &fakeCatalog{} - reconciler := New(kube, discoverer, catalog) + reconciler := New(kube, discoverer, catalog, nil) result, err := reconciler.Reconcile(t.Context(), ctrl.Request{NamespacedName: client.ObjectKeyFromObject(server)}) if err != nil { @@ -110,7 +114,7 @@ func TestReconcilePublishesFailureAndClearsStaleTools(t *testing.T) { discoverer := &fakeDiscoverer{err: errors.New("upstream unavailable")} catalog := &fakeCatalog{} - _, err := New(kube, discoverer, catalog).Reconcile(t.Context(), ctrl.Request{NamespacedName: client.ObjectKeyFromObject(server)}) + _, err := New(kube, discoverer, catalog, nil).Reconcile(t.Context(), ctrl.Request{NamespacedName: client.ObjectKeyFromObject(server)}) if err == nil { t.Fatal("Reconcile() error = nil, want discovery failure") } @@ -131,7 +135,7 @@ func TestReconcileDeletesCatalogProjection(t *testing.T) { catalog := &fakeCatalog{} request := ctrl.Request{NamespacedName: types.NamespacedName{Namespace: "test", Name: "gone"}} - if _, err := New(testClient(t), &fakeDiscoverer{}, catalog).Reconcile(t.Context(), request); err != nil { + if _, err := New(testClient(t), &fakeDiscoverer{}, catalog, nil).Reconcile(t.Context(), request); err != nil { t.Fatalf("Reconcile() error = %v", err) } want := "test/gone|" + remoteGroupKind @@ -177,6 +181,71 @@ func testServer() *v1alpha3.RemoteMCPServer { } } +func TestReconcileEvents(t *testing.T) { + for _, test := range []struct { + name string + tools []toolservice.MCPAppTool + condition *metav1.Condition + discoveryErr error + catalogErr error + wantEvent string + wantErr bool + }{ + {name: "initial empty discovery", wantEvent: "Normal ToolsDiscovered Discovered 0 MCP tools"}, + {name: "initial discovery", tools: []toolservice.MCPAppTool{{Name: "tool"}}, wantEvent: "Normal ToolsDiscovered Discovered 1 MCP tools"}, + {name: "unchanged successful discovery", condition: &metav1.Condition{Status: metav1.ConditionTrue, ObservedGeneration: 3}}, + {name: "new generation", condition: &metav1.Condition{Status: metav1.ConditionTrue, ObservedGeneration: 2}, wantEvent: "Normal ToolsDiscovered Discovered 0 MCP tools"}, + {name: "recovery to empty discovery", condition: &metav1.Condition{Status: metav1.ConditionFalse, ObservedGeneration: 3}, wantEvent: "Normal ToolsDiscovered Discovered 0 MCP tools"}, + {name: "discovery failure", discoveryErr: errors.New("unavailable"), wantErr: true, wantEvent: "Warning ReconcileFailed failed to discover RemoteMCPServer tools: unavailable"}, + {name: "validation failure", tools: []toolservice.MCPAppTool{{Name: " "}}, wantErr: true, wantEvent: "Warning ValidationFailed invalid RemoteMCPServer tool discovery:"}, + {name: "catalog failure", catalogErr: errors.New("unavailable"), wantErr: true, wantEvent: "Warning ReconcileFailed failed to update RemoteMCPServer tool catalog:"}, + } { + t.Run(test.name, func(t *testing.T) { + server := testServer() + if test.condition != nil { + condition := *test.condition + condition.Type = conditionAccepted + server.Status.Conditions = []metav1.Condition{condition} + } + recorder := events.NewFakeRecorder(4) + catalog := &fakeCatalog{err: test.catalogErr} + reconciler := New(testClient(t, server), &fakeDiscoverer{tools: test.tools, err: test.discoveryErr}, catalog, recorder) + request := ctrl.Request{NamespacedName: client.ObjectKeyFromObject(server)} + _, err := reconciler.Reconcile(t.Context(), request) + if test.wantErr { + require.Error(t, err) + } else { + require.NoError(t, err) + } + if test.wantEvent != "" { + select { + case event := <-recorder.Events: + require.Contains(t, event, test.wantEvent) + default: + t.Fatal("expected reconcile event") + } + } + require.Empty(t, recorder.Events) + if !test.wantErr { + _, err = reconciler.Reconcile(t.Context(), request) + require.NoError(t, err) + require.Empty(t, recorder.Events) + } + if test.catalogErr != nil { + catalog.err = nil + _, err = reconciler.Reconcile(t.Context(), request) + require.NoError(t, err) + select { + case event := <-recorder.Events: + require.Contains(t, event, "Normal ToolsDiscovered Discovered 0 MCP tools") + default: + t.Fatal("catalog retry lost the success event") + } + } + }) + } +} + func testClient(t *testing.T, objects ...client.Object) client.Client { t.Helper() scheme := runtime.NewScheme() diff --git a/go/core/internal/telemetry/logging.go b/go/core/internal/telemetry/logging.go new file mode 100644 index 0000000000..186099b6b7 --- /dev/null +++ b/go/core/internal/telemetry/logging.go @@ -0,0 +1,89 @@ +package telemetry + +import ( + "context" + "fmt" + "log/slog" + + "github.com/go-logr/logr" + "github.com/go-logr/zapr" + otelzap "go.opentelemetry.io/contrib/bridges/otelzap" + "go.opentelemetry.io/contrib/exporters/autoexport" + logglobal "go.opentelemetry.io/otel/log/global" + sdklog "go.opentelemetry.io/otel/sdk/log" + "go.uber.org/zap" + + "github.com/kagent-dev/kagent/go/core/pkg/env" +) + +// loggerBridgeName is the instrumentation scope under which the controller's +// own logs are emitted by the otelzap bridge, matching the module that owns +// the controller. +const loggerBridgeName = "github.com/kagent-dev/kagent/go/core" + +// InitLoggerProvider configures an OTLP LoggerProvider and registers it as the +// global OTel logger provider. The exporter type and endpoint are read from the +// standard OTEL environment variables via autoexport, mirroring +// InitTracerProvider, so log and trace pipelines share the same OTLP config. +// The returned shutdown function must be called on process exit to flush +// in-flight log records. When OTEL_LOGGING_ENABLED is unset (the default) the +// pipeline is not created and a no-op shutdown is returned. +func InitLoggerProvider(ctx context.Context, serviceVersion string) (func(context.Context) error, error) { + if !env.OtelLoggingEnabled.Get() { + return func(context.Context) error { return nil }, nil + } + + res, err := newTelemetryResource(ctx, serviceVersion) + if err != nil { + return nil, fmt.Errorf("create logging resource: %w", err) + } + + exporter, err := autoexport.NewLogExporter(ctx) + if err != nil { + return nil, fmt.Errorf("create log exporter: %w", err) + } + + lp := sdklog.NewLoggerProvider( + sdklog.WithProcessor(sdklog.NewBatchProcessor(exporter)), + sdklog.WithResource(res), + ) + + logglobal.SetLoggerProvider(lp) + + return lp.Shutdown, nil +} + +// ControllerLogHandler adds an otelzap bridge while preserving the original +// handler's output and level filtering. The global logger provider delegates +// to the SDK once InitLoggerProvider runs, including for early startup loggers. +func ControllerLogHandler(handler slog.Handler) slog.Handler { + if !env.OtelLoggingEnabled.Get() { + return handler + } + bridgeCore := otelzap.NewCore(loggerBridgeName, + otelzap.WithLoggerProvider(logglobal.GetLoggerProvider()), + ) + return &controllerLogHandler{ + Handler: handler, + output: slog.NewMultiHandler(handler, logr.ToSlogHandler(zapr.NewLogger(zap.New(bridgeCore)))), + } +} + +type controllerLogHandler struct { + slog.Handler + output slog.Handler +} + +var _ slog.Handler = (*controllerLogHandler)(nil) + +func (handler *controllerLogHandler) Handle(ctx context.Context, record slog.Record) error { + return handler.output.Handle(ctx, record) +} + +func (handler *controllerLogHandler) WithAttrs(attrs []slog.Attr) slog.Handler { + return &controllerLogHandler{Handler: handler.Handler.WithAttrs(attrs), output: handler.output.WithAttrs(attrs)} +} + +func (handler *controllerLogHandler) WithGroup(name string) slog.Handler { + return &controllerLogHandler{Handler: handler.Handler.WithGroup(name), output: handler.output.WithGroup(name)} +} diff --git a/go/core/internal/telemetry/logging_test.go b/go/core/internal/telemetry/logging_test.go new file mode 100644 index 0000000000..8ee2ab3a97 --- /dev/null +++ b/go/core/internal/telemetry/logging_test.go @@ -0,0 +1,113 @@ +package telemetry + +import ( + "bytes" + "context" + "encoding/json" + "log/slog" + "testing" + + "github.com/kagent-dev/kagent/go/pkg/logging" + "github.com/stretchr/testify/require" + otellog "go.opentelemetry.io/otel/log" + logglobal "go.opentelemetry.io/otel/log/global" + sdklog "go.opentelemetry.io/otel/sdk/log" +) + +type logExporter struct { + records []sdklog.Record +} + +var _ sdklog.Exporter = (*logExporter)(nil) + +func (exporter *logExporter) Export(_ context.Context, records []sdklog.Record) error { + for _, record := range records { + exporter.records = append(exporter.records, record.Clone()) + } + return nil +} + +func (*logExporter) Shutdown(context.Context) error { return nil } +func (*logExporter) ForceFlush(context.Context) error { return nil } + +func TestControllerLogHandler(t *testing.T) { + for _, enabled := range []string{"", "false", "true"} { + t.Run("enabled="+enabled, func(t *testing.T) { + t.Setenv("OTEL_LOGGING_ENABLED", enabled) + exporter := &logExporter{} + provider := sdklog.NewLoggerProvider(sdklog.WithProcessor(sdklog.NewSimpleProcessor(exporter))) + previous := logglobal.GetLoggerProvider() + logglobal.SetLoggerProvider(provider) + t.Cleanup(func() { + require.NoError(t, provider.Shutdown(context.Background())) + logglobal.SetLoggerProvider(previous) + }) + var output bytes.Buffer + base, err := logging.New(&output, "warn") + require.NoError(t, err) + handler := ControllerLogHandler(base.Handler()) + if enabled != "true" { + require.Same(t, base.Handler(), handler) + } + logger := slog.New(handler).With("component", "controller").WithGroup("reconcile").With("name", "tools") + logger.InfoContext(t.Context(), "filtered") + logger.WarnContext(t.Context(), "discovery failed", "attempt", 2) + var record struct { + Level string `json:"level"` + Message string `json:"msg"` + Component string `json:"component"` + Reconcile struct { + Name string `json:"name"` + Attempt int `json:"attempt"` + } `json:"reconcile"` + } + require.NoError(t, json.Unmarshal(output.Bytes(), &record)) + require.Equal(t, "WARN", record.Level) + require.Equal(t, "discovery failed", record.Message) + require.Equal(t, "controller", record.Component) + require.Equal(t, "tools", record.Reconcile.Name) + require.Equal(t, 2, record.Reconcile.Attempt) + if enabled == "true" { + require.Len(t, exporter.records, 1) + require.Equal(t, "discovery failed", exporter.records[0].Body().AsString()) + require.Equal(t, otellog.SeverityWarn, exporter.records[0].Severity()) + } else { + require.Empty(t, exporter.records) + } + }) + } +} + +func TestInitLoggerProvider(t *testing.T) { + for _, test := range []struct { + name, gate, exporter, resource string + wantErr bool + }{ + {name: "default off ignores broken config", exporter: "invalid", resource: "invalid"}, + {name: "explicit off ignores broken config", gate: "false", exporter: "invalid"}, + {name: "enabled", gate: "true", exporter: "none"}, + {name: "exporter failure", gate: "true", exporter: "invalid", wantErr: true}, + {name: "resource failure", gate: "true", exporter: "none", resource: "invalid", wantErr: true}, + } { + t.Run(test.name, func(t *testing.T) { + t.Setenv("OTEL_LOGGING_ENABLED", test.gate) + t.Setenv("OTEL_LOGS_EXPORTER", test.exporter) + t.Setenv("OTEL_RESOURCE_ATTRIBUTES", test.resource) + previous := logglobal.GetLoggerProvider() + t.Cleanup(func() { logglobal.SetLoggerProvider(previous) }) + shutdown, err := InitLoggerProvider(t.Context(), "test") + if test.wantErr { + require.Error(t, err) + require.Nil(t, shutdown) + } else { + require.NoError(t, err) + require.NoError(t, shutdown(t.Context())) + } + if test.gate != "true" || test.wantErr { + require.Same(t, previous, logglobal.GetLoggerProvider()) + } else { + require.IsType(t, &sdklog.LoggerProvider{}, logglobal.GetLoggerProvider()) + } + }) + } +} diff --git a/go/core/internal/telemetry/tracing.go b/go/core/internal/telemetry/tracing.go index 71528adf6e..2218e57a26 100644 --- a/go/core/internal/telemetry/tracing.go +++ b/go/core/internal/telemetry/tracing.go @@ -33,11 +33,34 @@ func InitTracerProvider(ctx context.Context, serviceVersion string) (func(contex return func(context.Context) error { return nil }, nil } + res, err := newTelemetryResource(ctx, serviceVersion) + if err != nil { + return nil, fmt.Errorf("create tracing resource: %w", err) + } + exporter, err := autoexport.NewSpanExporter(ctx) if err != nil { return nil, fmt.Errorf("create span exporter: %w", err) } + tp := sdktrace.NewTracerProvider( + sdktrace.WithBatcher(exporter), + sdktrace.WithResource(res), + ) + + otel.SetTracerProvider(tp) + otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator( + propagation.TraceContext{}, + propagation.Baggage{}, + )) + + return tp.Shutdown, nil +} + +// newTelemetryResource builds the OTel resource shared by the controller's +// signal pipelines (traces and logs), so every signal carries the same +// resource attributes. +func newTelemetryResource(ctx context.Context, serviceVersion string) (*resource.Resource, error) { instanceID, err := os.Hostname() if err != nil || instanceID == "" { instanceID = uuid.New().String() @@ -59,25 +82,9 @@ func InitTracerProvider(ctx context.Context, serviceVersion string) (func(contex attrs = append(attrs, semconv.K8SNodeName(node)) } - res, err := resource.New(ctx, + return resource.New(ctx, resource.WithTelemetrySDK(), resource.WithAttributes(attrs...), resource.WithFromEnv(), ) - if err != nil { - return nil, fmt.Errorf("create OTEL resource: %w", err) - } - - tp := sdktrace.NewTracerProvider( - sdktrace.WithBatcher(exporter), - sdktrace.WithResource(res), - ) - - otel.SetTracerProvider(tp) - otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator( - propagation.TraceContext{}, - propagation.Baggage{}, - )) - - return tp.Shutdown, nil } diff --git a/go/core/internal/telemetry/tracing_test.go b/go/core/internal/telemetry/tracing_test.go index 627ff2a2a1..a6b5c0fc70 100644 --- a/go/core/internal/telemetry/tracing_test.go +++ b/go/core/internal/telemetry/tracing_test.go @@ -26,6 +26,8 @@ func restoreGlobals(t *testing.T) { func TestInitTracerProviderDisabled(t *testing.T) { restoreGlobals(t) t.Setenv("OTEL_TRACING_ENABLED", "false") + t.Setenv("OTEL_TRACES_EXPORTER", "invalid") + t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "invalid") before := otel.GetTracerProvider() shutdown, err := telemetry.InitTracerProvider(context.Background(), "test") diff --git a/go/core/pkg/app/app.go b/go/core/pkg/app/app.go index 98ebdbb54e..eb4ca20631 100644 --- a/go/core/pkg/app/app.go +++ b/go/core/pkg/app/app.go @@ -137,6 +137,7 @@ func SetupLogger() error { if err != nil { return fmt.Errorf("parse LOG_LEVEL: %w", err) } + logger = slog.New(telemetry.ControllerLogHandler(logger.Handler())) slog.SetDefault(logger) ctrl.SetLogger(logging.AsLogr(logger)) return nil @@ -165,6 +166,18 @@ func Run(ctx context.Context, opts Options) error { } }() + shutdownLogging, err := telemetry.InitLoggerProvider(ctx, version.Version) + if err != nil { + return fmt.Errorf("initialize logging: %w", err) + } + defer func() { + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := shutdownLogging(shutdownCtx); err != nil { + logger.ErrorContext(shutdownCtx, "failed to shut down logging", "error", err) + } + }() + dbURL, err := database.ResolveURL(env("POSTGRES_DATABASE_URL", "postgres://postgres:kagent@kagent-postgresql.kagent.svc.cluster.local:5432/postgres"), os.Getenv("POSTGRES_DATABASE_URL_FILE")) if err != nil { return err @@ -253,11 +266,11 @@ func Run(ctx context.Context, opts Options) error { } } mcpClient := toolservice.NewRuntimeMCPClient(manager.GetClient()) - remoteMCPDiscovery := remotemcpcontroller.New(manager.GetClient(), mcpClient, store) + remoteMCPDiscovery := remotemcpcontroller.New(manager.GetClient(), mcpClient, store, manager.GetEventRecorder("remotemcpserver")) if err := remoteMCPDiscovery.SetupWithManager(manager); err != nil { return fmt.Errorf("set up RemoteMCPServer discovery: %w", err) } - mcpServerDiscovery := mcpservercontroller.New(manager.GetClient(), mcpClient, store) + mcpServerDiscovery := mcpservercontroller.New(manager.GetClient(), mcpClient, store, manager.GetEventRecorder("mcpserver-catalog")) if err := mcpServerDiscovery.SetupWithManager(manager); err != nil { return fmt.Errorf("set up MCPServer discovery: %w", err) } diff --git a/go/core/pkg/app/logging_test.go b/go/core/pkg/app/logging_test.go new file mode 100644 index 0000000000..7e10a86077 --- /dev/null +++ b/go/core/pkg/app/logging_test.go @@ -0,0 +1,112 @@ +package app + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "testing" + "time" + + "github.com/kagent-dev/kagent/go/core/internal/telemetry" + "github.com/kagent-dev/kagent/go/pkg/logging" + "github.com/stretchr/testify/require" + collectorlogsv1 "go.opentelemetry.io/proto/otlp/collector/logs/v1" + "google.golang.org/protobuf/proto" + ctrl "sigs.k8s.io/controller-runtime" +) + +func TestSetupLoggerStartup(t *testing.T) { + if os.Getenv("KAGENT_TEST_LOGGER_CHILD") == "true" { + testLoggerStartup(t) + return + } + for _, gate := range []string{"", "true"} { + t.Run("gate="+gate, func(t *testing.T) { + t.Setenv("OTEL_LOGGING_ENABLED", gate) + t.Setenv("LOG_LEVEL", "info") + ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second) + defer cancel() + command := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestSetupLoggerStartup$") + command.Env = append(os.Environ(), "KAGENT_TEST_LOGGER_CHILD=true") + var stderr, stdout bytes.Buffer + command.Stderr, command.Stdout = &stderr, &stdout + require.NoError(t, command.Run(), "stdout: %s; stderr: %s", stdout.String(), stderr.String()) + var messages []string + scanner := bufio.NewScanner(&stderr) + for scanner.Scan() { + var record struct { + Level string `json:"level"` + Message string `json:"msg"` + Component string `json:"component"` + } + require.NoError(t, json.Unmarshal(scanner.Bytes(), &record)) + require.Equal(t, "INFO", record.Level) + require.Equal(t, "controller", record.Component) + messages = append(messages, record.Message) + } + require.NoError(t, scanner.Err()) + require.ElementsMatch(t, []string{"startup logger", "runtime logger", "context logger"}, messages) + }) + } +} + +func TestRunValidatesLogLevelBeforeTelemetry(t *testing.T) { + t.Setenv("LOG_LEVEL", "invalid") + t.Setenv("OTEL_TRACING_ENABLED", "true") + t.Setenv("OTEL_TRACES_EXPORTER", "invalid") + require.ErrorContains(t, Run(t.Context(), Options{}), "parse LOG_LEVEL") +} + +func testLoggerStartup(t *testing.T) { + t.Helper() + requests := make(chan []byte, 4) + collector := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + body, err := io.ReadAll(request.Body) + if err != nil || request.URL.Path != "/v1/logs" { + writer.WriteHeader(http.StatusBadRequest) + return + } + requests <- body + writer.Header().Set("Content-Type", "application/x-protobuf") + })) + t.Cleanup(collector.Close) + t.Setenv("OTEL_LOGS_EXPORTER", "otlp") + t.Setenv("OTEL_EXPORTER_OTLP_LOGS_PROTOCOL", "http/protobuf") + t.Setenv("OTEL_EXPORTER_OTLP_LOGS_ENDPOINT", collector.URL+"/v1/logs") + t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "") + require.NoError(t, SetupLogger()) + startupLogger := slog.Default() + require.NoError(t, SetupLogger()) + shutdown, err := telemetry.InitLoggerProvider(t.Context(), "test") + require.NoError(t, err) + startupLogger.DebugContext(t.Context(), "filtered") + startupLogger.InfoContext(t.Context(), "startup logger", "component", "controller") + ctrl.Log.Info("runtime logger", "component", "controller") + ctx := logging.IntoContext(t.Context(), slog.Default()) + logging.FromContext(ctx).InfoContext(ctx, "context logger", "component", "controller") + require.NoError(t, shutdown(t.Context())) + var messages []string + for len(requests) > 0 { + var request collectorlogsv1.ExportLogsServiceRequest + require.NoError(t, proto.Unmarshal(<-requests, &request)) + for _, resource := range request.ResourceLogs { + for _, scope := range resource.ScopeLogs { + for _, record := range scope.LogRecords { + messages = append(messages, record.Body.GetStringValue()) + } + } + } + } + if os.Getenv("OTEL_LOGGING_ENABLED") == "true" { + require.ElementsMatch(t, []string{"startup logger", "runtime logger", "context logger"}, messages) + } else { + require.Empty(t, messages) + } +} diff --git a/go/go.mod b/go/go.mod index 06cd5fbf40..ccd615b6d4 100644 --- a/go/go.mod +++ b/go/go.mod @@ -21,6 +21,7 @@ require ( github.com/charmbracelet/x/ansi v0.11.8 github.com/glebarez/sqlite v1.11.0 github.com/go-logr/logr v1.4.4 + github.com/go-logr/zapr v1.3.0 github.com/golang-jwt/jwt/v5 v5.3.1 github.com/google/go-containerregistry v0.21.9 github.com/google/jsonschema-go v0.4.3 @@ -50,17 +51,24 @@ require ( github.com/stretchr/testify v1.12.1 github.com/testcontainers/testcontainers-go v0.44.0 github.com/testcontainers/testcontainers-go/modules/postgres v0.44.0 + go.opentelemetry.io/contrib/bridges/otelzap v0.19.0 go.opentelemetry.io/contrib/exporters/autoexport v0.69.0 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.69.0 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 go.opentelemetry.io/otel v1.44.0 go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.20.0 go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.20.0 + go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.44.0 + go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.44.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0 + go.opentelemetry.io/otel/log v0.20.0 + go.opentelemetry.io/otel/metric v1.44.0 go.opentelemetry.io/otel/sdk v1.44.0 go.opentelemetry.io/otel/sdk/log v0.20.0 + go.opentelemetry.io/otel/sdk/metric v1.44.0 go.opentelemetry.io/otel/trace v1.44.0 + go.uber.org/zap v1.28.0 golang.org/x/sync v0.22.0 golang.org/x/term v0.45.0 golang.org/x/text v0.41.0 @@ -423,20 +431,14 @@ require ( go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/bridges/prometheus v0.69.0 // indirect go.opentelemetry.io/contrib/detectors/gcp v1.44.0 // indirect - go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.44.0 // indirect - go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.44.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 // indirect go.opentelemetry.io/otel/exporters/prometheus v0.66.0 // indirect go.opentelemetry.io/otel/exporters/stdout/stdoutlog v0.20.0 // indirect go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.44.0 // indirect go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.44.0 // indirect - go.opentelemetry.io/otel/log v0.20.0 // indirect - go.opentelemetry.io/otel/metric v1.44.0 // indirect - go.opentelemetry.io/otel/sdk/metric v1.44.0 // indirect go.opentelemetry.io/proto/otlp v1.11.0 // indirect go.uber.org/atomic v1.11.0 // indirect go.uber.org/multierr v1.11.0 // indirect - go.uber.org/zap v1.28.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.5 // indirect go.yaml.in/yaml/v4 v4.0.0-rc.6 // indirect diff --git a/go/go.sum b/go/go.sum index 1e494f8c2d..ca7ad881ea 100644 --- a/go/go.sum +++ b/go/go.sum @@ -1214,6 +1214,8 @@ go.opencensus.io v0.20.2/go.mod h1:6WKK9ahsWS3RSO+PY9ZHZUfv2irvY6gN279GOPZjmmk= go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/contrib/bridges/otelzap v0.19.0 h1:48Eq3xxFx2KlL/tF7lnl42kKJBDlhNTLRzv0h154JnM= +go.opentelemetry.io/contrib/bridges/otelzap v0.19.0/go.mod h1:cQbV77F0u6HmtZPiQD9oxp2esaOEb4uLqIta6OFIKOk= go.opentelemetry.io/contrib/bridges/prometheus v0.69.0 h1:saQoWg5845Q8TojpqeVStS7zGwVZ6bc5W2PJavTPiBM= go.opentelemetry.io/contrib/bridges/prometheus v0.69.0/go.mod h1:AAaS6xs5AyqMdR3Ir0nSWK+QudL2XM8Vbw5INzUxNc8= go.opentelemetry.io/contrib/detectors/gcp v1.44.0 h1:NmLfL734pJhM0JKaYd2Y28+nY9dPRWYAAbxhRCrKXPw= @@ -1250,6 +1252,8 @@ go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.44.0 h1:bl2S7Ubua0Nms+D go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.44.0/go.mod h1:L0hRV50XdVIODHUfWEqGRCXQvj2rV82STVo12FMFBU0= go.opentelemetry.io/otel/log v0.20.0 h1:/5i0vuHxCLWUfChWG41K9wkM0jafruPw9NU1/RCJirs= go.opentelemetry.io/otel/log v0.20.0/go.mod h1:wOcMcjsZpG8x7Bak7IhSi/lg8wscV2C1VdrKCLPlt0E= +go.opentelemetry.io/otel/log/logtest v0.20.0 h1:+tsZVE15N+RWyN9lUzsRyw7hMZXNMepGu105Eim82/k= +go.opentelemetry.io/otel/log/logtest v0.20.0/go.mod h1:zS9Ryx9RrEAG2tgapMBSvacwhVSSOGSaSiWWgW3NPlQ= go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= go.opentelemetry.io/otel/metric/x v0.66.0 h1:YkCrx1zLOChi9ZcZ6euupOcsgzbVlec7D/xoEU1+cTA= diff --git a/helm/kagent/templates/rbac/writer-role.yaml b/helm/kagent/templates/rbac/writer-role.yaml index 5866de2725..17625b8d07 100644 --- a/helm/kagent/templates/rbac/writer-role.yaml +++ b/helm/kagent/templates/rbac/writer-role.yaml @@ -1,4 +1,11 @@ {{- define "kagent.writer.rules" -}} +- apiGroups: + - events.k8s.io + resources: + - events + verbs: + - create + - patch - apiGroups: - kagent.dev resources: diff --git a/helm/kagent/tests/rbac_test.yaml b/helm/kagent/tests/rbac_test.yaml index 719eb6c102..7671cf6d8f 100644 --- a/helm/kagent/tests/rbac_test.yaml +++ b/helm/kagent/tests/rbac_test.yaml @@ -7,6 +7,28 @@ templates: - rbac/getter-rolebinding.yaml - rbac/writer-rolebinding.yaml tests: + - it: should allow reconcile events with cluster-wide RBAC + template: rbac/writer-role.yaml + asserts: + - contains: + path: rules + content: + apiGroups: ["events.k8s.io"] + resources: ["events"] + verbs: ["create", "patch"] + + - it: should allow reconcile events with namespace-scoped RBAC + template: rbac/writer-role.yaml + set: + rbac.namespaces: [NAMESPACE] + asserts: + - contains: + path: rules + content: + apiGroups: ["events.k8s.io"] + resources: ["events"] + verbs: ["create", "patch"] + - it: should render controller serviceaccount template: controller-serviceaccount.yaml asserts: @@ -251,4 +273,4 @@ tests: - equal: path: metadata.namespace value: other-ns - documentIndex: 1 \ No newline at end of file + documentIndex: 1