From c8873bbd0ecb3c7e85d833a84f7f5ad3fd2476ae Mon Sep 17 00:00:00 2001 From: krsnaSuraj Date: Mon, 13 Jul 2026 00:09:43 +0530 Subject: [PATCH] fix(atenet): propagate trace context through router and namespace span/metric attrs Fixes trace detachment in the router where two issues broke the trace chain: 1. Primary: resumer.go's singleflight used context.Background() inside DoChan, so otelgrpc saw no parent span and ateapi handler spans appeared in a separate trace. Now propagates the caller's span context into the background context. 2. Secondary: the ext_proc HeadersResponse only rewrote :authority and never injected traceparent/tracestate into the header mutation. Now injects the trace context so the upstream worker receives the same trace even when Envoy tracing is not configured. 3. Span and metric attributes used bare keys (atespace, actor, etc.) instead of the ate.* convention established in #412. Renamed to ate.atespace, ate.actor_name, ate.target_addr, ate.actor_template_namespace, ate.actor_template_name, ate.outcome. Fixes #427 --- cmd/atenet/internal/router/extproc.go | 20 ++++++++++++++++---- cmd/atenet/internal/router/extproc_out.go | 18 ++++++++++++++++++ cmd/atenet/internal/router/extproc_test.go | 22 ++++++++++++---------- cmd/atenet/internal/router/resumer.go | 7 +++++-- 4 files changed, 51 insertions(+), 16 deletions(-) diff --git a/cmd/atenet/internal/router/extproc.go b/cmd/atenet/internal/router/extproc.go index 203c668d1..64c857978 100644 --- a/cmd/atenet/internal/router/extproc.go +++ b/cmd/atenet/internal/router/extproc.go @@ -147,6 +147,10 @@ func (s *ExtProcServer) handleRequestHeaders( // Host is invalid, respond with 404. return nil, metadata, "", "", "", invalidHostErr(metadata.host, err) } + span.SetAttributes( + attribute.String("ate.atespace", atespace), + attribute.String("ate.actor_name", actorName), + ) slog.InfoContext(ctx, "ResumeActor", slog.String("atespace", atespace), slog.String("actor", actorName)) actor, err := s.resumer.ResumeActor(ctx, atespace, actorName) @@ -176,9 +180,17 @@ func (s *ExtProcServer) handleRequestHeaders( slog.InfoContext(ctx, "Route ok", slog.String("actor", actorName), slog.String("targetAddr", targetAddr)) - // Route by rewriting the :authority header. + span.SetAttributes( + attribute.String("ate.target_addr", targetAddr), + attribute.String("ate.actor_template_namespace", tmplNs), + attribute.String("ate.actor_template_name", tmplName), + ) + + // Route by rewriting the :authority header and injecting trace context + // so the upstream worker receives the same trace as the ingress path. mutation := &extprocv3.HeaderMutation{} addAuthorityMutation(targetAddr, mutation) + injectTraceContext(ctx, mutation) return &extprocv3.HeadersResponse{ Response: &extprocv3.CommonResponse{ @@ -192,9 +204,9 @@ func (s *ExtProcServer) recordRouteDuration(ctx context.Context, d time.Duration return } s.routeDuration.Record(ctx, d.Seconds(), metric.WithAttributes( - attribute.String("actor_template_namespace", tmplNs), - attribute.String("actor_template_name", tmplName), - attribute.String("outcome", outcome), + attribute.String("ate.actor_template_namespace", tmplNs), + attribute.String("ate.actor_template_name", tmplName), + attribute.String("ate.outcome", outcome), )) } diff --git a/cmd/atenet/internal/router/extproc_out.go b/cmd/atenet/internal/router/extproc_out.go index 98d21d93e..8c15c6dda 100644 --- a/cmd/atenet/internal/router/extproc_out.go +++ b/cmd/atenet/internal/router/extproc_out.go @@ -18,6 +18,8 @@ import ( 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. @@ -43,6 +45,22 @@ func addAuthorityMutation(auth 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) { + headers := make(map[string]string) + otel.GetTextMapPropagator().Inject(ctx, propagation.MapCarrier(headers)) + for k, v := range headers { + mutation.SetHeaders = append(mutation.SetHeaders, &corev3.HeaderValueOption{ + Header: &corev3.HeaderValue{ + Key: k, + Value: v, + }, + }) + } +} + 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 81f97a944..9cd102707 100644 --- a/cmd/atenet/internal/router/extproc_test.go +++ b/cmd/atenet/internal/router/extproc_test.go @@ -230,17 +230,19 @@ func TestExtProcHeadersEvaluation(t *testing.T) { } mutation := res.Response.GetHeaderMutation() - if len(mutation.GetSetHeaders()) != 1 { - t.Fatalf("expected exactly one Header option set, found: %v", mutation.GetSetHeaders()) - } - - headerOption := mutation.GetSetHeaders()[0] - if strings.ToLower(headerOption.Header.Key) != ":authority" { - t.Errorf("invalid resulting dynamic parameter key: %s", headerOption.Header.Key) + headers := mutation.GetSetHeaders() + var found bool + for _, h := range headers { + if strings.ToLower(h.Header.Key) == ":authority" { + found = true + if string(h.Header.RawValue) != tc.expectedTarget { + t.Errorf("invalid destination mapping found: %s, expected: %s", h.Header.RawValue, tc.expectedTarget) + } + break + } } - - if string(headerOption.Header.RawValue) != tc.expectedTarget { - t.Errorf("invalid destination mapping found: %s, expected: %s", headerOption.Header.RawValue, tc.expectedTarget) + if !found { + t.Errorf(":authority header not found in mutation, got: %v", headers) } // Confirm that query logs recorded metric trace details diff --git a/cmd/atenet/internal/router/resumer.go b/cmd/atenet/internal/router/resumer.go index db2231c06..1fa8034c3 100644 --- a/cmd/atenet/internal/router/resumer.go +++ b/cmd/atenet/internal/router/resumer.go @@ -46,8 +46,8 @@ func NewActorResumer(apiClient ateapipb.ControlClient) *ActorResumer { func (r *ActorResumer) ResumeActor(ctx context.Context, atespace, actorName string) (*ateapipb.Actor, error) { ctx, span := otel.Tracer(routerServiceName).Start(ctx, "ResumeActor", trace.WithAttributes( - attribute.String("atespace", atespace), - attribute.String("actor", actorName), + attribute.String("ate.atespace", atespace), + attribute.String("ate.actor_name", actorName), )) defer span.End() @@ -57,6 +57,9 @@ func (r *ActorResumer) ResumeActor(ctx context.Context, atespace, actorName stri // resume operation continues running for Caller 2 and Caller 3 without failing. bgCtx, bgCancel := context.WithTimeout(context.Background(), 15*time.Second) defer bgCancel() + // Propagate the caller's span context so the gRPC spans are children + // of ResumeActor rather than appearing as a separate trace. + bgCtx = trace.ContextWithSpanContext(bgCtx, trace.SpanContextFromContext(ctx)) backoff := wait.Backoff{ Steps: 7,