feat(jobs): facet Consumer spans by environment.id in Datadog

api/internal/jobs/consumer_span.go

@@ -0,0 +1,44 @@
+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.
+// 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)
+	annotateConsumerOutcome(ctx, err)
+	return err
+}

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
+}

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)