fix(atenet): propagate trace context through router and namespace span/metric attrs

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{

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{

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")
+}

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
 

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)
+	}
+}