diff --git a/cmd/snowglobe/helpers.go b/cmd/snowglobe/helpers.go index 4aa7c2c..03b3d43 100644 --- a/cmd/snowglobe/helpers.go +++ b/cmd/snowglobe/helpers.go @@ -13,6 +13,7 @@ import ( "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" logapi "go.opentelemetry.io/otel/log" + sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/trace" ) @@ -222,6 +223,54 @@ func urlHost(raw string) string { return raw } +// httpServerAttrs returns the attributes an HTTP SERVER span must carry beyond what the caller sets: +// url.scheme and url.path (Required by the OpenTelemetry HTTP server span table) and, when the span is a +// child of an HTTP CLIENT span, that client's server.address and server.port. The convention says these +// "are intended, whenever possible, to be the same on the client and server sides", and asks a server to +// do the best effort from the request it received, so a server reached through a named host reports that +// host (simulated here: the front door is a Host header the generator does not really have). +// +// url.path is the path of the request the server received. When the parent client says what it requested +// (url.full), that is the path; the fallback argument is used only for a server with no HTTP client parent +// (a webhook arriving from outside). A third-party server span is never emitted from here. +func httpServerAttrs(parent context.Context, fallbackPath string) []attribute.KeyValue { + path := fallbackPath + var extra []attribute.KeyValue + if p, ok := trace.SpanFromContext(parent).(sdktrace.ReadOnlySpan); ok && p.SpanKind() == trace.SpanKindClient { + var isHTTP bool + var addr, port *attribute.KeyValue + var full string + for _, a := range p.Attributes() { + a := a + switch string(a.Key) { + case "http.request.method": + isHTTP = true + case "server.address": + addr = &a + case "server.port": + port = &a + case "url.full": + full = a.Value.AsString() + } + } + if isHTTP { + if u, err := url.Parse(full); err == nil && u.Path != "" { + path = u.EscapedPath() + } + if addr != nil { + extra = append(extra, *addr) + if port != nil { + extra = append(extra, *port) + } + } + } + } + return append([]attribute.KeyValue{ + attribute.String("url.scheme", "https"), + attribute.String("url.path", path), + }, extra...) +} + type modelInfo struct { name string // model identifier system string // gen_ai.provider.name value diff --git a/cmd/snowglobe/scenarios.go b/cmd/snowglobe/scenarios.go index e288fe1..a7748b9 100644 --- a/cmd/snowglobe/scenarios.go +++ b/cmd/snowglobe/scenarios.go @@ -37,6 +37,7 @@ func createOrderFlow(ctx context.Context) { attribute.String("user_agent.original", "Mozilla/5.0"), attribute.String("network.peer.address", randomIP()), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/orders")...), ) defer gateway.End() emitLog(ctx, "api-gateway", logapi.SeverityInfo, "Incoming request POST /api/v2/orders", @@ -255,6 +256,7 @@ func searchAndBrowseFlow(ctx context.Context) { attribute.Int("http.response.status_code", 200), attribute.String("search.query", randomSearchTerm()), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/search")...), ) defer gateway.End() sleep(5, 15) @@ -333,6 +335,7 @@ func userLoginFlow(ctx context.Context) { attribute.String("http.request.method", "POST"), attribute.String("http.route", "/api/v2/auth/login"), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/auth/login")...), ) defer func() { if isFailure { @@ -414,6 +417,7 @@ func failedPaymentFlow(ctx context.Context) { attribute.Int("http.response.status_code", 402), attribute.String("network.peer.address", randomIP()), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/orders")...), ) exc := randomException(paymentExceptions) recordException(ctx, "api-gateway", gateway, exc.excType, exc.message, exc.stacktrace) @@ -510,6 +514,7 @@ func bulkNotificationFlow(ctx context.Context) { attribute.String("http.route", "/api/v2/admin/digest"), attribute.Int("http.response.status_code", 202), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/admin/digest")...), ) defer gateway.End() sleep(3, 10) @@ -605,6 +610,7 @@ func healthCheckFlow(ctx context.Context) { attribute.String("http.route", "/healthz"), attribute.Int("http.response.status_code", 200), ), + trace.WithAttributes(httpServerAttrs(ctx, "/healthz")...), ) defer gateway.End() sleep(1, 5) @@ -668,6 +674,7 @@ func inventorySyncFlow(ctx context.Context) { attribute.String("http.route", "/api/v2/admin/inventory/sync"), attribute.Int("http.response.status_code", 202), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/admin/inventory/sync")...), ) defer gateway.End() sleep(3, 10) @@ -870,6 +877,7 @@ func stripeWebhookFlow(ctx context.Context) { attribute.String("webhook.id", fmt.Sprintf("evt_%016d", rand.Int63())), attribute.String("service.peer.name", "stripe-api"), ), + trace.WithAttributes(httpServerAttrs(ctx, "/webhooks/stripe")...), ) defer webhook.End() sleep(5, 15) @@ -992,6 +1000,7 @@ func recommendationFlow(ctx context.Context) { attribute.String("http.route", "/api/v2/recommendations"), attribute.Int("http.response.status_code", 200), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/recommendations")...), ) defer gateway.End() sleep(3, 10) @@ -1118,6 +1127,7 @@ func addToCartFlow(ctx context.Context) { attribute.String("http.route", "/api/v2/cart/items"), attribute.Int("http.response.status_code", 200), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/cart/items")...), ) defer gateway.End() sleep(3, 10) @@ -1263,6 +1273,7 @@ func fullCheckoutFlow(ctx context.Context) { attribute.Int("http.response.status_code", 200), attribute.String("network.peer.address", randomIP()), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/checkout")...), ) defer gateway.End() emitLog(ctx, "api-gateway", logapi.SeverityInfo, "Checkout initiated", @@ -1593,6 +1604,7 @@ func shippingUpdateFlow(ctx context.Context) { attribute.String("shipping.tracking_number", fmt.Sprintf("1Z%09d", rand.Intn(999999999))), attribute.String("shipping.status", shippingStatus), ), + trace.WithAttributes(httpServerAttrs(ctx, "/webhooks/carrier")...), ) defer webhook.End() sleep(5, 15) @@ -1739,6 +1751,7 @@ func sagaCompensationFlow(ctx context.Context) { attribute.String("http.route", "/api/v2/checkout"), attribute.Int("http.response.status_code", 402), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/checkout")...), ) defer gateway.End() sleep(5, 15) @@ -2014,6 +2027,7 @@ func timeoutCascadeFlow(ctx context.Context) { attribute.Int("http.response.status_code", 504), attribute.Int("retry.attempt", 1), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/search")...), ) sleep(5, 10) @@ -2051,6 +2065,7 @@ func timeoutCascadeFlow(ctx context.Context) { attribute.Bool("circuit_breaker.open", true), attribute.String("circuit_breaker.fallback", "stale_cache"), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/search")...), ) sleep(3, 8) diff --git a/cmd/snowglobe/scenarios_ai.go b/cmd/snowglobe/scenarios_ai.go index e7d3d80..b64e037 100644 --- a/cmd/snowglobe/scenarios_ai.go +++ b/cmd/snowglobe/scenarios_ai.go @@ -40,6 +40,7 @@ func ragSearchFlow(ctx context.Context) { attribute.Int("http.response.status_code", 200), attribute.String("search.query", query), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/search/ai")...), ) defer gateway.End() sleep(3, 10) @@ -209,6 +210,7 @@ func aiChatbotFlow(ctx context.Context) { attribute.String("http.route", "/api/v2/chat"), attribute.Int("http.response.status_code", 200), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/chat")...), ) defer gateway.End() sleep(3, 8) @@ -370,6 +372,7 @@ func contentModerationFlow(ctx context.Context) { attribute.String("http.request.method", "POST"), attribute.String("http.route", "/api/v2/reviews"), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/reviews")...), ) defer gateway.End() sleep(3, 8) @@ -542,6 +545,7 @@ func multiStepAgentFlow(ctx context.Context) { attribute.String("http.route", "/api/v2/agent/task"), attribute.Int("http.response.status_code", 200), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/agent/task")...), ) defer gateway.End() sleep(3, 8) @@ -767,6 +771,7 @@ func returnRefundFlow(ctx context.Context) { attribute.String("http.route", "/api/v2/returns"), attribute.Int("http.response.status_code", 200), ), + trace.WithAttributes(httpServerAttrs(ctx, "/api/v2/returns")...), ) defer gateway.End() sleep(5, 15) diff --git a/cmd/snowglobe/semconv_e1_server_test.go b/cmd/snowglobe/semconv_e1_server_test.go new file mode 100644 index 0000000..4eaadf4 --- /dev/null +++ b/cmd/snowglobe/semconv_e1_server_test.go @@ -0,0 +1,118 @@ +package main + +import ( + "net/url" + "testing" + + "go.opentelemetry.io/otel/sdk/trace" + apitrace "go.opentelemetry.io/otel/trace" +) + +// SP-120 E1. The HTTP semantic conventions (http-spans, semconv 1.44.0) make these Required on a SERVER +// span: http.request.method, url.path, url.scheme. server.address is Recommended, server.port is +// Conditionally Required when server.address is set, and both "are intended, whenever possible, to be the +// same on the client and server sides". A span is an HTTP span here when it carries http.request.method. + +func TestHTTPServerSpansCarryRequiredAttributes(t *testing.T) { + spans := recordedSpans(t) + + var httpServer int + violations := map[string]int{} + for _, s := range spans { + if s.SpanKind() != apitrace.SpanKindServer { + continue + } + if _, ok := attrOf(s, "http.request.method"); !ok { + continue + } + httpServer++ + for _, m := range missing(s, "url.path", "url.scheme") { + violations["SERVER "+s.Name()+" missing "+m]++ + } + if _, ok := attrOf(s, "server.address"); ok { + for _, m := range missing(s, "server.port") { + violations["SERVER "+s.Name()+" missing "+m+" (server.address set)"]++ + } + } + } + if httpServer == 0 { + t.Fatal("recorded no HTTP server spans: the harness is not driving the flows") + } + reportViolations(t, violations, "Required-attribute") +} + +// For EVERY HTTP client -> HTTP server pair the server reports what the client asked for: the same host, +// the same port, and the path of the client's url.full. Presence alone would let a wrong value through +// (the healthz server once said /healthz while its client requested /api/v2/healthz). +func TestHTTPServerSpanMirrorsCallingClientRequest(t *testing.T) { + spans := recordedSpans(t) + + byID := map[apitrace.SpanID]trace.ReadOnlySpan{} + for _, s := range spans { + byID[s.SpanContext().SpanID()] = s + } + + pairs := 0 + sawHealthz := false + for _, s := range spans { + if s.SpanKind() != apitrace.SpanKindServer { + continue + } + if _, ok := attrOf(s, "http.request.method"); !ok { + continue + } + parent, ok := byID[s.Parent().SpanID()] + if !ok || parent.SpanKind() != apitrace.SpanKindClient { + continue + } + if _, isHTTP := attrOf(parent, "http.request.method"); !isHTTP { + continue + } + pairs++ + if s.Name() == "GET /healthz" { + sawHealthz = true + } + + full, _ := attrOf(parent, "url.full") + u, err := url.Parse(full.AsString()) + if err != nil || u.Path == "" { + t.Errorf("client %q has no parseable url.full %q", parent.Name(), full.AsString()) + continue + } + if got, ok := attrOf(s, "url.path"); !ok || got.AsString() != u.EscapedPath() { + t.Errorf("server %q url.path=%q ok=%v, want the client's requested path %q", s.Name(), got.AsString(), ok, u.EscapedPath()) + } + if got, ok := attrOf(s, "server.address"); !ok || got.AsString() != u.Hostname() { + t.Errorf("server %q server.address=%q ok=%v, want the client's host %q", s.Name(), got.AsString(), ok, u.Hostname()) + } + pAddr, _ := attrOf(parent, "server.address") + if got, _ := attrOf(s, "server.address"); got.AsString() != pAddr.AsString() { + t.Errorf("server %q server.address=%q differs from client attribute %q", s.Name(), got.AsString(), pAddr.AsString()) + } + pPort, _ := attrOf(parent, "server.port") + if got, ok := attrOf(s, "server.port"); !ok || got.AsInt64() != pPort.AsInt64() { + t.Errorf("server %q server.port=%d ok=%v, want the client's %d", s.Name(), got.AsInt64(), ok, pPort.AsInt64()) + } + if got, _ := attrOf(s, "url.scheme"); got.AsString() != u.Scheme { + t.Errorf("server %q url.scheme=%q, client url.full scheme %q", s.Name(), got.AsString(), u.Scheme) + } + } + if pairs == 0 { + t.Fatal("no HTTP client -> HTTP server pair observed") + } + if !sawHealthz { + t.Fatal("the health-check pair (the one that once disagreed) was not observed") + } +} + +// Nothing is invented: only the shop front door is ever a server host. +func TestNoInventedThirdPartyServers(t *testing.T) { + for _, s := range recordedSpans(t) { + if s.SpanKind() != apitrace.SpanKindServer { + continue + } + if v, ok := attrOf(s, "server.address"); ok && v.AsString() != "shop.example.com" { + t.Errorf("SERVER span %q carries server.address=%q: only the shop front door is served here", s.Name(), v.AsString()) + } + } +} diff --git a/cmd/snowglobe/semconv_testutil_test.go b/cmd/snowglobe/semconv_testutil_test.go new file mode 100644 index 0000000..f72c4c4 --- /dev/null +++ b/cmd/snowglobe/semconv_testutil_test.go @@ -0,0 +1,161 @@ +package main + +import ( + "context" + "os" + "reflect" + "regexp" + "runtime" + "sort" + "strings" + "sync" + "testing" + + "go.opentelemetry.io/otel/attribute" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "go.opentelemetry.io/otel/trace" +) + +// Shared machinery for the semantic-convention conformance tests (SP-120 e-002/e-004). +// +// The tests drive every scenario flow through the real tracer() indirection (tracerPool) into a +// SpanRecorder, so they assert what the generator actually emits, not what the source text says. + +var ( + flowRe = regexp.MustCompile(`\{"[^"]+", (\w+Flow),`) + tracerRe = regexp.MustCompile(`tracer\("([\w-]+)"\)`) +) + +func allFlows() []func(context.Context) { + return []func(context.Context){ + createOrderFlow, searchAndBrowseFlow, userLoginFlow, addToCartFlow, fullCheckoutFlow, + healthCheckFlow, failedPaymentFlow, bulkNotificationFlow, inventorySyncFlow, scheduledReportFlow, + stripeWebhookFlow, recommendationFlow, shippingUpdateFlow, returnRefundFlow, sagaCompensationFlow, + timeoutCascadeFlow, ragSearchFlow, aiChatbotFlow, contentModerationFlow, multiStepAgentFlow, + } +} + +func flowName(f func(context.Context)) string { + n := runtime.FuncForPC(reflect.ValueOf(f).Pointer()).Name() + return n[strings.LastIndex(n, ".")+1:] +} + +func readSource(t *testing.T, names ...string) string { + t.Helper() + var all string + for _, n := range names { + b, err := os.ReadFile(n) + if err != nil { + t.Fatalf("read %s: %v", n, err) + } + all += string(b) + } + return all +} + +var ( + recordOnce sync.Once + recordSpans []sdktrace.ReadOnlySpan +) + +// recordedSpans runs every flow 25 times against a recorder, once for the whole test binary (the flows +// sleep, so several tests each re-running 500 flows cost 22 s apiece), and returns the ended spans. +func recordedSpans(t *testing.T) []sdktrace.ReadOnlySpan { + t.Helper() + recordOnce.Do(func() { recordSpans = recordAllFlows(t, 25) }) + if recordSpans == nil { + t.Fatal("recording failed in an earlier test") + } + return recordSpans +} + +func recordAllFlows(t *testing.T, perFlow int) []sdktrace.ReadOnlySpan { + t.Helper() + + // Every flow registered in main.go must be driven here, by NAME, so a new or swapped scenario cannot + // escape this test. + registered := map[string]bool{} + for _, m := range flowRe.FindAllStringSubmatch(readSource(t, "main.go"), -1) { + registered[m[1]] = true + } + driven := map[string]bool{} + for _, f := range allFlows() { + driven[flowName(f)] = true + } + if !reflect.DeepEqual(registered, driven) { + var r, d []string + for k := range registered { + r = append(r, k) + } + for k := range driven { + d = append(d, k) + } + sort.Strings(r) + sort.Strings(d) + t.Fatalf("main.go registers flows %v, test drives %v: update allFlows()", r, d) + } + + rec := tracetest.NewSpanRecorder() + tp := sdktrace.NewTracerProvider(sdktrace.WithSampler(sdktrace.AlwaysSample()), sdktrace.WithSpanProcessor(rec)) + defer func() { _ = tp.Shutdown(context.Background()) }() + + saved := tracerPool + tracerPool = map[string][]trace.Tracer{} + defer func() { tracerPool = saved }() + for _, m := range tracerRe.FindAllStringSubmatch(readSource(t, "scenarios.go", "scenarios_ai.go", "helpers.go"), -1) { + tracerPool[m[1]] = []trace.Tracer{tp.Tracer(m[1])} + } + + savedErr, savedCons, savedLogs := errorMultiplier, consumersEnabled, logsDisabled + errorMultiplier, consumersEnabled, logsDisabled = 1.0, true, true + defer func() { errorMultiplier, consumersEnabled, logsDisabled = savedErr, savedCons, savedLogs }() + + var wg sync.WaitGroup + for _, f := range allFlows() { + for i := 0; i < perFlow; i++ { + wg.Add(1) + go func(f func(context.Context)) { + defer wg.Done() + f(context.Background()) + }(f) + } + } + wg.Wait() + return rec.Ended() +} + +func attrOf(s sdktrace.ReadOnlySpan, key string) (attribute.Value, bool) { + for _, a := range s.Attributes() { + if string(a.Key) == key { + return a.Value, true + } + } + return attribute.Value{}, false +} + +func missing(s sdktrace.ReadOnlySpan, keys ...string) []string { + var out []string + for _, k := range keys { + if _, ok := attrOf(s, k); !ok { + out = append(out, k) + } + } + return out +} + +func reportViolations(t *testing.T, violations map[string]int, what string) { + t.Helper() + if len(violations) == 0 { + return + } + var keys []string + for k := range violations { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + t.Errorf("%s (x%d)", k, violations[k]) + } + t.Errorf("%d distinct %s violations", len(violations), what) +}