Address PR review: AppendAction, OTel semconv, trace injection tests

- extproc_out.go: Set AppendAction=OVERWRITE_IF_EXISTS_OR_ADD on injected trace headers
- extproc.go: Replace ate.target_addr with stable OTel semconv server.address + server.port
- extproc_test.go: Add inMemoryExporter + tests for traceparent matching parent span trace ID and AppendAction default
- resumer_test.go: Add resumerSpanExporter + test for ResumeActor child span sharing parent trace ID and span ID

Author: krsnaSuraj

cmd/atenet/internal/router/extproc.go

@@ -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"
@@ -181,7 +182,8 @@ func (s *ExtProcServer) handleRequestHeaders(
 	slog.InfoContext(ctx, "Route ok", slog.String("actor", actorName), slog.String("targetAddr", targetAddr))
 
 	span.SetAttributes(
-		attribute.String("ate.target_addr", targetAddr),
+		semconv.ServerAddress(workerIP),
+		semconv.ServerPort(80),
 		attribute.String("ate.actor.template.namespace", tmplNs),
 		attribute.String("ate.actor.template.name", tmplName),
 	)

cmd/atenet/internal/router/extproc_out.go

@@ -59,6 +59,7 @@ func injectTraceContext(ctx context.Context, mutation *extproc.HeaderMutation) {
 				Key:   k,
 				Value: v,
 			},
+			AppendAction: corev3.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD,
 		})
 	}
 }

cmd/atenet/internal/router/extproc_test.go

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

cmd/atenet/internal/router/resumer_test.go

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