Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 18 additions & 4 deletions cmd/atenet/internal/router/extproc.go
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import (
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/metric"
"go.opentelemetry.io/otel/propagation"
"go.opentelemetry.io/otel/semconv/v1.40.0"
"google.golang.org/grpc"

"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
Expand Down Expand Up @@ -147,6 +148,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)
Expand Down Expand Up @@ -176,9 +181,18 @@ 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(
semconv.ServerAddress(workerIP),
semconv.ServerPort(80),
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{
Expand All @@ -192,9 +206,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),
Comment on lines +209 to +211

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Renaming these metric labels changes the time series identity, I left these out of #412 for this reason. Totally fine to do it, but make sure to call it out explicitly in the release note as it's a "breaking change".

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point - completely agreed. I''ve updated the release note in the PR description to explicitly call out the route_duration label rename as BREAKING, noting that dashboards and alert rules need updating.

))
}

Expand Down
21 changes: 21 additions & 0 deletions cmd/atenet/internal/router/extproc_out.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -43,6 +47,23 @@ 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{

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You're not setting AppendAction, so this defaults to APPEND_IF_EXISTS_OR_ADD. The request already has a traceparent, so Envoy appends a second value and not replace it. The w3c propagator receives a multi value traceparent, which is invalid per spec, so the worker starts a fresh root span.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Great catch! I''ve set AppendAction to OVERWRITE_IF_EXISTS_OR_ADD on all injected trace headers so Envoy replaces the existing traceparent instead of appending a second (invalid) value.

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{
Expand Down
123 changes: 113 additions & 10 deletions cmd/atenet/internal/router/extproc_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,13 +21,17 @@ import (
"errors"
"log/slog"
"strings"
"sync"
"testing"
"time"

"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
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"
"go.opentelemetry.io/otel/propagation"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
Expand Down Expand Up @@ -230,17 +234,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()

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you add a small test with a local TracerProvider + in-memory exporter, start a parent span, and assert the mutation has a traceparent matching it? Also same idea for the resumer reparenting.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done! Added TestInjectTraceContext_AddsTraceparentMatchingParentSpan - sets up a local TracerProvider with in-memory exporter, starts a parent span, calls injectTraceContext, and asserts the injected traceparent matches the parent span''s trace ID. Also added TestInjectTraceContext_AppendActionDefaultsToOverwrite as a focused AppendAction check.

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
Expand All @@ -252,3 +258,100 @@ func TestExtProcHeadersEvaluation(t *testing.T) {
})
}
}

// inMemoryExporter stores ended spans for test inspection.
type inMemoryExporter struct {
mu sync.Mutex
spans []sdktrace.ReadOnlySpan
}

func (e *inMemoryExporter) ExportSpans(_ context.Context, spans []sdktrace.ReadOnlySpan) error {
e.mu.Lock()
defer e.mu.Unlock()
e.spans = append(e.spans, spans...)
return nil
}

func (e *inMemoryExporter) Shutdown(context.Context) error {
e.mu.Lock()
defer e.mu.Unlock()
e.spans = nil
return nil
}

func (e *inMemoryExporter) Ended() []sdktrace.ReadOnlySpan {
e.mu.Lock()
defer e.mu.Unlock()
ret := make([]sdktrace.ReadOnlySpan, len(e.spans))
copy(ret, e.spans)
return ret
}

func TestInjectTraceContext_AddsTraceparentMatchingParentSpan(t *testing.T) {
prevProp := otel.GetTextMapPropagator()
otel.SetTextMapPropagator(propagation.TraceContext{})
t.Cleanup(func() { otel.SetTextMapPropagator(prevProp) })

exporter := &inMemoryExporter{}
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{}
injectTraceContext(ctx, mutation)
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) {
prevProp := otel.GetTextMapPropagator()
otel.SetTextMapPropagator(propagation.TraceContext{})
t.Cleanup(func() { otel.SetTextMapPropagator(prevProp) })

exporter := &inMemoryExporter{}
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{}
injectTraceContext(ctx, mutation)

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")
}
7 changes: 5 additions & 2 deletions cmd/atenet/internal/router/resumer.go
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Comment on lines +49 to +50

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

FYI these are what I am setting in #412 Can we plan the rebase to call the ateattr helpers instead of redoing it once #412 is merged?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good call! Once #412 merges, I''ll rebase this branch and replace the inline ate.* attribute strings with the ateattr helpers. Will track that as a follow-up so it doesn''t block the trace context fix here.

))
defer span.End()

Expand All @@ -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))

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This looks good, but can you please add test asserting the resume child span shares the parent traceid?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done! Added TestActorResumer_ResumeChildSharesParentTraceID - sets up a local TracerProvider + in-memory exporter, starts a parent span, calls ResumeActor, and asserts the ResumeActor span shares the parent''s trace ID and parent span ID.


backoff := wait.Backoff{
Steps: 7,
Expand Down
85 changes: 85 additions & 0 deletions cmd/atenet/internal/router/resumer_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@ import (
"time"

"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
"go.opentelemetry.io/otel"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
Expand Down Expand Up @@ -168,3 +170,86 @@ func TestActorResumer_ResumeActor(t *testing.T) {
}
})
}

type resumerSpanExporter struct {
mu sync.Mutex
spans []sdktrace.ReadOnlySpan
}

func (e *resumerSpanExporter) ExportSpans(_ context.Context, spans []sdktrace.ReadOnlySpan) error {
e.mu.Lock()
defer e.mu.Unlock()
e.spans = append(e.spans, spans...)
return nil
}

func (e *resumerSpanExporter) Shutdown(context.Context) error {
e.mu.Lock()
defer e.mu.Unlock()
e.spans = nil
return nil
}

func (e *resumerSpanExporter) Ended() []sdktrace.ReadOnlySpan {
e.mu.Lock()
defer e.mu.Unlock()
ret := make([]sdktrace.ReadOnlySpan, len(e.spans))
copy(ret, e.spans)
return ret
}

func TestActorResumer_ResumeChildSharesParentTraceID(t *testing.T) {
const testActorName = "actor-a"
const testAtespace = "team-a"
const expectedIP = "10.0.0.52"

exporter := &resumerSpanExporter{}
tp := sdktrace.NewTracerProvider(
sdktrace.WithSampler(sdktrace.AlwaysSample()),
sdktrace.WithSpanProcessor(sdktrace.NewSimpleSpanProcessor(exporter)),
)
prevTP := otel.GetTracerProvider()
otel.SetTracerProvider(tp)
t.Cleanup(func() { otel.SetTracerProvider(prevTP) })

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{
Metadata: &ateapipb.ResourceMetadata{Name: testActorName},
Status: ateapipb.Actor_STATUS_RUNNING,
AteomPodIp: expectedIP,
},
}, nil
},
}

resumer := NewActorResumer(mock)
_, err := resumer.ResumeActor(ctx, testAtespace, testActorName)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
parent.End()

var resumeSpan sdktrace.ReadOnlySpan
for _, s := range exporter.Ended() {
if s.Name() == "ResumeActor" {
resumeSpan = s
break
}
}
if resumeSpan == nil {
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)
}
}