diff --git a/cmd/atenet/internal/router/extproc.go b/cmd/atenet/internal/router/extproc.go index 795afd583..f7cb05566 100644 --- a/cmd/atenet/internal/router/extproc.go +++ b/cmd/atenet/internal/router/extproc.go @@ -30,6 +30,7 @@ import ( "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/metric" "go.opentelemetry.io/otel/propagation" + semconv "go.opentelemetry.io/otel/semconv/v1.40.0" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -153,6 +154,7 @@ func (s *ExtProcServer) handleRequestHeaders( // Host is invalid, respond with 404. return nil, metadata, "", "", "", ResumeOutcomeNone, invalidHostErr(metadata.host, err) } + span.SetAttributes(ateattr.ActorRefAttributes(actorRef)...) // Admit the request to the parking lot before resuming. While resume is // in-flight the request occupies a slot; if the actor's worker pool is @@ -186,6 +188,12 @@ func (s *ExtProcServer) handleRequestHeaders( return nil, metadata, "", tmplNs, tmplName, resumeOutcome, newReqError(envoy_type.StatusCode_InternalServerError, "actor %s routing failed", actorRef) } + // Record the resolved worker endpoint with the stable OTel semconv + // attributes (replaces the earlier custom ate.target_addr key). + span.SetAttributes( + semconv.ServerAddress(workerIP), + semconv.ServerPort(443), + ) // The actor is reached through the in-worker atunnel ingress server, which // listens on :443 (mTLS) and forwards to the actor's :80. The worker no @@ -200,9 +208,11 @@ func (s *ExtProcServer) handleRequestHeaders( // Route by telling the ORIGINAL_DST cluster which worker atunnel address to // dial, without touching :authority — atunnel authorizes the actor by the - // original Host (actor DNS name). + // original Host (actor DNS name). Inject the trace context so the upstream + // worker receives the same trace as the ingress path. mutation := &extprocv3.HeaderMutation{} addOriginalDstMutation(targetAddr, mutation) + injectTraceContext(ctx, mutation) return &extprocv3.HeadersResponse{ Response: &extprocv3.CommonResponse{ diff --git a/cmd/atenet/internal/router/extproc_out.go b/cmd/atenet/internal/router/extproc_out.go index 964a1a4f7..ddf9fdf2e 100644 --- a/cmd/atenet/internal/router/extproc_out.go +++ b/cmd/atenet/internal/router/extproc_out.go @@ -15,9 +15,13 @@ package router import ( + "context" + corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" extproc "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3" envoy_type "github.com/envoyproxy/go-control-plane/envoy/type/v3" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/propagation" ) // reqError carries an HTTP-mappable status code and a client-safe message. @@ -53,6 +57,30 @@ func addOriginalDstMutation(dst string, mut *extproc.HeaderMutation) { ) } +// injectTraceContext injects the trace context from ctx into the header mutation, +// so that Envoy forwards traceparent/tracestate to the upstream worker and spans +// are connected across the full request path. +func injectTraceContext(ctx context.Context, mutation *extproc.HeaderMutation) { + injectTraceContextWithPropagator(ctx, mutation, otel.GetTextMapPropagator()) +} + +// injectTraceContextWithPropagator is the injectable core of injectTraceContext. +// Tests pass an explicit propagator so they can assert the injected headers +// without swapping the process-global text map propagator. +func injectTraceContextWithPropagator(ctx context.Context, mutation *extproc.HeaderMutation, propagator propagation.TextMapPropagator) { + headers := make(map[string]string) + propagator.Inject(ctx, propagation.MapCarrier(headers)) + for k, v := range headers { + mutation.SetHeaders = append(mutation.SetHeaders, &corev3.HeaderValueOption{ + Header: &corev3.HeaderValue{ + Key: k, + Value: v, + }, + AppendAction: corev3.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD, + }) + } +} + func immediateResponse(statusCode envoy_type.StatusCode, message string) *extproc.ProcessingResponse { return &extproc.ProcessingResponse{ Response: &extproc.ProcessingResponse_ImmediateResponse{ diff --git a/cmd/atenet/internal/router/extproc_test.go b/cmd/atenet/internal/router/extproc_test.go index ab08d0c59..a800f5a92 100644 --- a/cmd/atenet/internal/router/extproc_test.go +++ b/cmd/atenet/internal/router/extproc_test.go @@ -29,8 +29,11 @@ import ( extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3" envoy_type "github.com/envoyproxy/go-control-plane/envoy/type/v3" "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/propagation" sdkmetric "go.opentelemetry.io/otel/sdk/metric" "go.opentelemetry.io/otel/sdk/metric/metricdata" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -418,3 +421,64 @@ func TestRecordRouteDuration_Attributes(t *testing.T) { } } } + +func TestInjectTraceContext_AddsTraceparentMatchingParentSpan(t *testing.T) { + exporter := tracetest.NewInMemoryExporter() + tp := sdktrace.NewTracerProvider( + sdktrace.WithSampler(sdktrace.AlwaysSample()), + sdktrace.WithSpanProcessor(sdktrace.NewSimpleSpanProcessor(exporter)), + ) + + ctx, span := tp.Tracer("test").Start(context.Background(), "parent") + parentTraceID := span.SpanContext().TraceID() + + mutation := &extprocv3.HeaderMutation{} + injectTraceContextWithPropagator(ctx, mutation, propagation.TraceContext{}) + span.End() + + var traceparent string + for _, h := range mutation.GetSetHeaders() { + if strings.ToLower(h.Header.Key) == "traceparent" { + traceparent = h.Header.Value + if h.GetAppendAction() != corev3.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD { + t.Errorf("AppendAction = %v, want OVERWRITE_IF_EXISTS_OR_ADD", h.GetAppendAction()) + } + break + } + } + if traceparent == "" { + t.Fatal("traceparent not found in header mutation") + } + + parts := strings.Split(traceparent, "-") + if len(parts) < 2 { + t.Fatalf("invalid traceparent format: %s", traceparent) + } + if parts[1] != parentTraceID.String() { + t.Errorf("trace ID in traceparent = %s, want parent trace ID = %s", parts[1], parentTraceID.String()) + } +} + +func TestInjectTraceContext_AppendActionDefaultsToOverwrite(t *testing.T) { + exporter := tracetest.NewInMemoryExporter() + tp := sdktrace.NewTracerProvider( + sdktrace.WithSampler(sdktrace.AlwaysSample()), + sdktrace.WithSpanProcessor(sdktrace.NewSimpleSpanProcessor(exporter)), + ) + + ctx, span := tp.Tracer("test").Start(context.Background(), "parent") + defer span.End() + + mutation := &extprocv3.HeaderMutation{} + injectTraceContextWithPropagator(ctx, mutation, propagation.TraceContext{}) + + for _, h := range mutation.GetSetHeaders() { + if strings.ToLower(h.Header.Key) == "traceparent" { + if h.GetAppendAction() != corev3.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD { + t.Errorf("AppendAction = %v, want OVERWRITE_IF_EXISTS_OR_ADD", h.GetAppendAction()) + } + return + } + } + t.Error("traceparent not found in mutation") +} diff --git a/cmd/atenet/internal/router/resumer.go b/cmd/atenet/internal/router/resumer.go index e6b01f705..3c5c7719b 100644 --- a/cmd/atenet/internal/router/resumer.go +++ b/cmd/atenet/internal/router/resumer.go @@ -91,6 +91,11 @@ type ActorResumer struct { apiClient ateapipb.ControlClient flight singleflight.Group + // tracer records the ResumeActor span. Defaults to the package-level + // router tracer; injectable for tests so they can assert span lineage + // without swapping the global TracerProvider. + tracer trace.Tracer + // parkEnabled makes transient worker-pool saturation (FailedPrecondition) // retryable, so a request is parked and retried until budget rather than // failing immediately. @@ -124,9 +129,19 @@ func withParking(cfg ParkedRequestConfig) resumerOption { } } +// withTracer overrides the tracer used to record the ResumeActor span. Tests +// use this to capture spans with a local TracerProvider instead of swapping +// the process-global one. +func withTracer(tracer trace.Tracer) resumerOption { + return func(r *ActorResumer) { + r.tracer = tracer + } +} + func NewActorResumer(apiClient ateapipb.ControlClient, opts ...resumerOption) *ActorResumer { r := &ActorResumer{ apiClient: apiClient, + tracer: otel.Tracer(routerServiceName), budget: failFastResumeBudget, backoff: resumeBackoff(defaultParkedRequestRetryInterval, defaultParkedRequestRetryFactor, defaultParkedRequestRetryJitter), @@ -161,7 +176,7 @@ func (r *ActorResumer) retryable(err error) bool { // requests within the process and, when parking is enabled, holds the request // while retrying transient failures until the budget elapses. func (r *ActorResumer) ResumeActor(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.Actor, ResumeOutcome, error) { - ctx, span := otel.Tracer(routerServiceName).Start(ctx, "ResumeActor", + ctx, span := r.tracer.Start(ctx, "ResumeActor", trace.WithAttributes(ateattr.ActorRefAttributes(actorRef)...)) defer span.End() @@ -179,6 +194,11 @@ func (r *ActorResumer) ResumeActor(ctx context.Context, actorRef resources.Actor // one control-plane RPC per hot actor (see docs/request-parking.md). bgCtx, bgCancel := context.WithTimeout(context.Background(), r.budget) defer bgCancel() + // Propagate the caller's span context into the background context so the + // gRPC spans created by the flight are children of ResumeActor rather + // than starting a fresh root trace. Without this, singleflight's + // context.Background() detaches the trace chain at the resumer. + bgCtx = trace.ContextWithSpanContext(bgCtx, trace.SpanContextFromContext(ctx)) backoff := r.backoff diff --git a/cmd/atenet/internal/router/resumer_test.go b/cmd/atenet/internal/router/resumer_test.go index 9e358b9c5..ac4afebde 100644 --- a/cmd/atenet/internal/router/resumer_test.go +++ b/cmd/atenet/internal/router/resumer_test.go @@ -23,6 +23,8 @@ import ( "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -511,3 +513,57 @@ func TestResumeBackoffHasNoCap(t *testing.T) { t.Errorf("resume backoff Steps must be high so the budget bounds the wait; got %d", b.Steps) } } + +func TestActorResumer_ResumeChildSharesParentTraceID(t *testing.T) { + const testActorName = "actor-a" + const testAtespace = "team-a" + const expectedIP = "10.0.0.52" + + exporter := tracetest.NewInMemoryExporter() + tp := sdktrace.NewTracerProvider( + sdktrace.WithSampler(sdktrace.AlwaysSample()), + sdktrace.WithSpanProcessor(sdktrace.NewSimpleSpanProcessor(exporter)), + ) + + ctx, parent := tp.Tracer("test").Start(context.Background(), "parent") + parentTraceID := parent.SpanContext().TraceID() + parentSpanID := parent.SpanContext().SpanID() + + mock := &resumerMockClient{ + resumeFn: func(ctx context.Context, in *ateapipb.ResumeActorRequest, opts ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) { + return &ateapipb.ResumeActorResponse{ + Actor: &ateapipb.Actor{ + Status: ateapipb.Actor_STATUS_RUNNING, + AteomPodIp: expectedIP, + }, + }, nil + }, + } + + // withTracer injects the test tracer so the ResumeActor span lands in the + // in-memory exporter without swapping the process-global TracerProvider. + resumer := NewActorResumer(mock, withTracer(tp.Tracer("test"))) + _, _, err := resumer.ResumeActor(ctx, resources.ActorRef{Atespace: testAtespace, Name: testActorName}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + parent.End() + + var resumeSpan tracetest.SpanStub + for _, s := range exporter.GetSpans() { + if s.Name == "ResumeActor" { + resumeSpan = s + break + } + } + if resumeSpan.Name == "" { + t.Fatal("ResumeActor span not found in exporter") + } + + if got, want := resumeSpan.SpanContext.TraceID(), parentTraceID; got != want { + t.Errorf("ResumeActor trace ID = %s, want parent trace ID = %s", got, want) + } + if got, want := resumeSpan.Parent.SpanID(), parentSpanID; got != want { + t.Errorf("ResumeActor parent span ID = %s, want %s", got, want) + } +}