From 3f5d9ebccbb9e02b068e97bc8248bf40b22c7213 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 12 Aug 2026 06:57:01 +0000 Subject: [PATCH 1/2] feat(jobs): set environment.id on Consumer entry spans Annotate go-queue otel Consumer spans for EnvironmentScrape and SitespeedScrape with low-cardinality environment.id (and outcome) so Datadog can facet process spans without opening child Internal spans. Co-authored-by: Soner --- api/internal/jobs/consumer_span.go | 41 ++++++++++++++ api/internal/jobs/consumer_span_test.go | 73 +++++++++++++++++++++++++ api/internal/jobs/handlers.go | 11 +++- 3 files changed, 123 insertions(+), 2 deletions(-) create mode 100644 api/internal/jobs/consumer_span.go create mode 100644 api/internal/jobs/consumer_span_test.go diff --git a/api/internal/jobs/consumer_span.go b/api/internal/jobs/consumer_span.go new file mode 100644 index 00000000..85ce25c6 --- /dev/null +++ b/api/internal/jobs/consumer_span.go @@ -0,0 +1,41 @@ +package jobs + +import ( + "context" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" +) + +// annotateConsumerEnvironment copies the environment id onto the active +// go-queue Consumer span (EnvironmentScrape process / SitespeedScrape process). +// Child Internal spans already carry richer environment attributes; Datadog +// facets on the entry span need this low-cardinality id without opening +// children. Do not add high-cardinality fields (URL, emails, extension lists). +func annotateConsumerEnvironment(ctx context.Context, environmentID int32) { + trace.SpanFromContext(ctx).SetAttributes( + attribute.Int("environment.id", int(environmentID)), + ) +} + +// annotateConsumerOutcome stamps a tiny ok/error enum on the active Consumer +// span after the handler returns. Mirrors go-queue otel status (error vs ok) +// as a facetable attribute; business-specific outcomes stay on metrics/child spans. +func annotateConsumerOutcome(ctx context.Context, err error) { + outcome := "ok" + if err != nil { + outcome = "error" + } + trace.SpanFromContext(ctx).SetAttributes( + attribute.String("outcome", outcome), + ) +} + +// runEnvironmentJob annotates the Consumer span with environment.id up front +// and outcome when the job finishes, then returns the job error unchanged. +func runEnvironmentJob(ctx context.Context, environmentID int32, run func(context.Context) error) error { + annotateConsumerEnvironment(ctx, environmentID) + err := run(ctx) + annotateConsumerOutcome(ctx, err) + return err +} diff --git a/api/internal/jobs/consumer_span_test.go b/api/internal/jobs/consumer_span_test.go new file mode 100644 index 00000000..1830ee3e --- /dev/null +++ b/api/internal/jobs/consumer_span_test.go @@ -0,0 +1,73 @@ +package jobs + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + sdktrace "go.opentelemetry.io/otel/sdk/trace" +) + +func TestRunEnvironmentJobSetsConsumerSpanAttributes(t *testing.T) { + tp := sdktrace.NewTracerProvider() + tracer := tp.Tracer("test") + ctx, span := tracer.Start(context.Background(), "EnvironmentScrape process") + + err := runEnvironmentJob(ctx, 42, func(context.Context) error { + return nil + }) + require.NoError(t, err) + span.End() + + attrs := spanAttributes(span.(sdktrace.ReadOnlySpan)) + assert.Equal(t, int64(42), attrs["environment.id"].AsInt64()) + assert.Equal(t, "ok", attrs["outcome"].AsString()) +} + +func TestRunEnvironmentJobSetsErrorOutcome(t *testing.T) { + tp := sdktrace.NewTracerProvider() + tracer := tp.Tracer("test") + ctx, span := tracer.Start(context.Background(), "SitespeedScrape process") + + jobErr := errors.New("sitespeed failed") + err := runEnvironmentJob(ctx, 7, func(context.Context) error { + return jobErr + }) + assert.ErrorIs(t, err, jobErr) + span.End() + + attrs := spanAttributes(span.(sdktrace.ReadOnlySpan)) + assert.Equal(t, int64(7), attrs["environment.id"].AsInt64()) + assert.Equal(t, "error", attrs["outcome"].AsString()) +} + +func TestRunEnvironmentJobAnnotatesParentNotChild(t *testing.T) { + // Handlers receive the Consumer span in ctx; job body may start Internal + // children. Attributes must stay on the Consumer entry span. + tp := sdktrace.NewTracerProvider() + tracer := tp.Tracer("test") + ctx, consumer := tracer.Start(context.Background(), "EnvironmentScrape process") + + err := runEnvironmentJob(ctx, 99, func(ctx context.Context) error { + _, child := tracer.Start(ctx, "environment.scrape") + child.End() + return nil + }) + require.NoError(t, err) + consumer.End() + + attrs := spanAttributes(consumer.(sdktrace.ReadOnlySpan)) + assert.Equal(t, int64(99), attrs["environment.id"].AsInt64()) + assert.Equal(t, "ok", attrs["outcome"].AsString()) +} + +func spanAttributes(span sdktrace.ReadOnlySpan) map[string]attribute.Value { + out := make(map[string]attribute.Value, len(span.Attributes())) + for _, attr := range span.Attributes() { + out[string(attr.Key)] = attr.Value + } + return out +} diff --git a/api/internal/jobs/handlers.go b/api/internal/jobs/handlers.go index 0036079a..7f6d1ff7 100644 --- a/api/internal/jobs/handlers.go +++ b/api/internal/jobs/handlers.go @@ -60,13 +60,20 @@ func RegisterHandlers(bus *goqueue.Bus, handlers Handlers) error { } goqueue.HandleFunc(bus, TransportName, func(ctx context.Context, message EnvironmentScrape) error { - return handlers.EnvironmentScraper.Scrape(ctx, message.EnvironmentID) + // Annotate the go-queue otel Consumer span before Scrape starts a child + // Internal span, so Datadog can facet EnvironmentScrape process by environment.id. + return runEnvironmentJob(ctx, message.EnvironmentID, func(ctx context.Context) error { + return handlers.EnvironmentScraper.Scrape(ctx, message.EnvironmentID) + }) }) goqueue.HandleFunc(bus, TransportName, func(ctx context.Context, message StoreExtensionSync) error { + // Names are high-cardinality; keep them off the Consumer entry span. return handlers.StoreExtensionSynchronizer.Sync(ctx, message.Names, message.ShopwareVersion) }) goqueue.HandleFunc(bus, TransportName, func(ctx context.Context, message SitespeedScrape) error { - return handlers.SitespeedScraper.Scrape(ctx, message.EnvironmentID) + return runEnvironmentJob(ctx, message.EnvironmentID, func(ctx context.Context) error { + return handlers.SitespeedScraper.Scrape(ctx, message.EnvironmentID) + }) }) goqueue.HandleFunc(bus, TransportName, func(ctx context.Context, _ LockCleanup) error { return handlers.Cleanup.CleanupLocks(ctx) From e333f6bd6d324f48695c84c3efbce357c140aad0 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 14 Aug 2026 14:23:15 +0000 Subject: [PATCH 2/2] docs(jobs): note Consumer span errors are owned by queueotel Clarify that runEnvironmentJob must not RecordError/SetStatus; go-queue otel middleware already does that after the handler returns. Co-authored-by: Soner --- api/internal/jobs/consumer_span.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/api/internal/jobs/consumer_span.go b/api/internal/jobs/consumer_span.go index 85ce25c6..8c53cd1c 100644 --- a/api/internal/jobs/consumer_span.go +++ b/api/internal/jobs/consumer_span.go @@ -33,6 +33,9 @@ func annotateConsumerOutcome(ctx context.Context, err error) { // runEnvironmentJob annotates the Consumer span with environment.id up front // and outcome when the job finishes, then returns the job error unchanged. +// Do not RecordError / SetStatus here: queueotel.Middleware already does that +// on the Consumer span after the handler returns; duplicating it would emit +// two exception events for the same failure. func runEnvironmentJob(ctx context.Context, environmentID int32, run func(context.Context) error) error { annotateConsumerEnvironment(ctx, environmentID) err := run(ctx)