diff --git a/go/core/cmd/controller-v2/main.go b/go/core/cmd/controller-v2/main.go index aeca8a3c62..f2833c4e13 100644 --- a/go/core/cmd/controller-v2/main.go +++ b/go/core/cmd/controller-v2/main.go @@ -42,6 +42,8 @@ import ( prompttemplateservice "github.com/kagent-dev/kagent/go/core/internal/service/prompttemplate" systemservice "github.com/kagent-dev/kagent/go/core/internal/service/system" toolservice "github.com/kagent-dev/kagent/go/core/internal/service/tool" + "github.com/kagent-dev/kagent/go/core/internal/telemetry" + "github.com/kagent-dev/kagent/go/core/internal/version" "github.com/kagent-dev/kagent/go/core/pkg/auth" "github.com/kagent-dev/kagent/go/core/pkg/migrations" "github.com/kagent-dev/kagent/go/core/v2/a2agateway" @@ -68,7 +70,6 @@ import ( func main() { ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() - logLevel := zapcore.InfoLevel if value := os.Getenv("ZAP_LOG_LEVEL"); value != "" { if err := logLevel.Set(value); err != nil { @@ -76,6 +77,19 @@ func main() { } } ctrl.SetLogger(zap.New(zap.Level(logLevel))) + // otelgrpc snapshots the global TracerProvider and propagator when its handler + // is constructed, so tracing has to be registered before any server is built. + shutdownTracing, err := telemetry.InitTracerProvider(ctx, version.Version) + if err != nil { + log.Fatalf("initialize tracing: %v", err) + } + defer func() { + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := shutdownTracing(shutdownCtx); err != nil { + log.Printf("shutdown tracing: %v", 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 { diff --git a/go/core/internal/telemetry/tracing_test.go b/go/core/internal/telemetry/tracing_test.go new file mode 100644 index 0000000000..627ff2a2a1 --- /dev/null +++ b/go/core/internal/telemetry/tracing_test.go @@ -0,0 +1,66 @@ +package telemetry_test + +import ( + "context" + "net/http" + "testing" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/propagation" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + + "github.com/kagent-dev/kagent/go/core/internal/telemetry" +) + +// restoreGlobals puts the process-wide OTEL registrations back after a test. +func restoreGlobals(t *testing.T) { + t.Helper() + tracerProvider := otel.GetTracerProvider() + propagator := otel.GetTextMapPropagator() + t.Cleanup(func() { + otel.SetTracerProvider(tracerProvider) + otel.SetTextMapPropagator(propagator) + }) +} + +func TestInitTracerProviderDisabled(t *testing.T) { + restoreGlobals(t) + t.Setenv("OTEL_TRACING_ENABLED", "false") + + before := otel.GetTracerProvider() + shutdown, err := telemetry.InitTracerProvider(context.Background(), "test") + if err != nil { + t.Fatal(err) + } + if err := shutdown(context.Background()); err != nil { + t.Fatalf("shutdown: %v", err) + } + if otel.GetTracerProvider() != before { + t.Fatal("disabled tracing replaced the global TracerProvider") + } +} + +func TestInitTracerProviderRegistersGlobals(t *testing.T) { + restoreGlobals(t) + t.Setenv("OTEL_TRACING_ENABLED", "true") + // "none" selects a noop exporter, so the test dials no collector. + t.Setenv("OTEL_TRACES_EXPORTER", "none") + + shutdown, err := telemetry.InitTracerProvider(context.Background(), "test") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = shutdown(context.Background()) }) + + if _, ok := otel.GetTracerProvider().(*sdktrace.TracerProvider); !ok { + t.Fatalf("global TracerProvider = %T, want *sdktrace.TracerProvider", otel.GetTracerProvider()) + } + + ctx, span := otel.Tracer("test").Start(context.Background(), "span") + defer span.End() + header := http.Header{} + otel.GetTextMapPropagator().Inject(ctx, propagation.HeaderCarrier(header)) + if header.Get("traceparent") == "" { + t.Fatal("registered propagator did not inject traceparent") + } +} diff --git a/go/core/v2/translator/compiler_test.go b/go/core/v2/translator/compiler_test.go index f7217bc9f4..b62a86da26 100644 --- a/go/core/v2/translator/compiler_test.go +++ b/go/core/v2/translator/compiler_test.go @@ -292,6 +292,42 @@ func TestCompileAgentTemplateResolvesCredentialsForSubstrate(t *testing.T) { } } +func TestCompileAgentTemplateForwardsOtelEnvironment(t *testing.T) { + t.Setenv("OTEL_TRACING_ENABLED", "true") + harness := &v1alpha3.Harness{ + ObjectMeta: metav1.ObjectMeta{Name: "kagent", Namespace: "test"}, + Spec: v1alpha3.HarnessSpec{ + Kagent: &v1alpha3.KagentHarness{}, + AllowedAgentTemplates: &v1alpha3.HarnessAgentTemplateAdmission{Selector: metav1.LabelSelector{}}, + Workload: v1alpha3.HarnessWorkload{Image: "example.com/kagent@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}, + Substrate: v1alpha3.HarnessSubstratePolicy{ + WorkerPoolRef: corev1.LocalObjectReference{Name: "default"}, + SnapshotPolicy: v1alpha3.HarnessSnapshotPolicy{Location: "snapshots"}, + }, + }, + } + template := &v1alpha3.AgentTemplate{ + ObjectMeta: metav1.ObjectMeta{Name: "helper", Namespace: "test"}, + Spec: v1alpha3.AgentTemplateSpec{ + ModelConfig: &corev1.LocalObjectReference{Name: "default-model"}, + SystemPrompt: "help", + }, + } + spec, err := compiler(t, modelConfig()).CompileAgentTemplate(context.Background(), harness, template) + if err != nil { + t.Fatal(err) + } + for _, variable := range spec.Environment { + if variable.Name == "OTEL_TRACING_ENABLED" { + if variable.Value != "true" { + t.Fatalf("OTEL_TRACING_ENABLED = %q, want %q", variable.Value, "true") + } + return + } + } + t.Fatalf("OTEL_TRACING_ENABLED missing from runtime revision environment: %+v", spec.Environment) +} + func TestCompileAgentTemplateSharedAgent(t *testing.T) { selector := &v1alpha3.HarnessAgentTemplateAdmission{Selector: metav1.LabelSelector{MatchLabels: map[string]string{"runtime": "kagent"}}} harness := &v1alpha3.Harness{ diff --git a/go/core/v2/translator/kagent/compiler.go b/go/core/v2/translator/kagent/compiler.go index ab5e8d80d3..3d563c6ba9 100644 --- a/go/core/v2/translator/kagent/compiler.go +++ b/go/core/v2/translator/kagent/compiler.go @@ -73,6 +73,7 @@ func (c *Compiler) Compile(ctx context.Context, input *v2translator.HarnessInput corev1.EnvVar{Name: "KAGENT_A2A_GRPC_ADDRESS", Value: "[::]:80"}, corev1.EnvVar{Name: "KAGENT_PRE_RESPONSE_TRACE_FLUSH", Value: "true"}, ) + environment = append(environment, v2translator.OtelEnvFromProcess()...) environment = adkconfig.DedupeEnv(environment) provenance, err := c.config.BuildProvenance(ctx, harness, compiled.Templates, compiled.Models, environment) if err != nil { diff --git a/go/core/v2/translator/otel.go b/go/core/v2/translator/otel.go new file mode 100644 index 0000000000..60dcbdc6bb --- /dev/null +++ b/go/core/v2/translator/otel.go @@ -0,0 +1,31 @@ +package translator + +import ( + "os" + + corev1 "k8s.io/api/core/v1" +) + +// These are the tracing settings read by the runtime; trace-specific values +// take precedence over their generic OTLP counterparts. +// Keep this list explicit: headers may contain credentials and resource +// attributes belong to the controller rather than its agent runtimes. +var otelEnvNames = []string{ + "OTEL_TRACING_ENABLED", + "OTEL_EXPORTER_OTLP_ENDPOINT", + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", + "OTEL_EXPORTER_OTLP_PROTOCOL", + "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL", +} + +// OtelEnvFromProcess returns the controller's supported tracing configuration +// for the agent runtime. +func OtelEnvFromProcess() []corev1.EnvVar { + envVars := make([]corev1.EnvVar, 0, len(otelEnvNames)) + for _, name := range otelEnvNames { + if value, found := os.LookupEnv(name); found { + envVars = append(envVars, corev1.EnvVar{Name: name, Value: value}) + } + } + return envVars +} diff --git a/go/core/v2/translator/otel_test.go b/go/core/v2/translator/otel_test.go new file mode 100644 index 0000000000..500d7699d4 --- /dev/null +++ b/go/core/v2/translator/otel_test.go @@ -0,0 +1,32 @@ +package translator_test + +import ( + "reflect" + "testing" + + "github.com/kagent-dev/kagent/go/core/v2/translator" + corev1 "k8s.io/api/core/v1" +) + +func TestOtelEnvFromProcess(t *testing.T) { + t.Setenv("OTEL_TRACING_ENABLED", "true") + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "collector:4317") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "http://collector:4317") + t.Setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "http/protobuf") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL", "grpc") + t.Setenv("OTEL_EXPORTER_OTLP_HEADERS", "authorization=secret") + t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "service.name=controller") + t.Setenv("OTEL_SERVICE_NAME", "controller") + + got := translator.OtelEnvFromProcess() + want := []corev1.EnvVar{ + {Name: "OTEL_TRACING_ENABLED", Value: "true"}, + {Name: "OTEL_EXPORTER_OTLP_ENDPOINT", Value: "collector:4317"}, + {Name: "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", Value: "http://collector:4317"}, + {Name: "OTEL_EXPORTER_OTLP_PROTOCOL", Value: "http/protobuf"}, + {Name: "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL", Value: "grpc"}, + } + if !reflect.DeepEqual(got, want) { + t.Errorf("OtelEnvFromProcess() = %#v, want %#v", got, want) + } +}