diff --git a/cmd/atenet/internal/router/extproc.go b/cmd/atenet/internal/router/extproc.go index 203c668d..64c85797 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 98d21d93..8c15c6dd 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 81f97a94..9cd10270 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 db2231c0..1fa8034c 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,