RHINENG-28312: Fix ingestion span parenting and listener timing

base/telemetry/buffer.go

@@ -0,0 +1,40 @@
+package telemetry
+
+import (
+	"context"
+	"time"
+
+	"go.opentelemetry.io/otel/attribute"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"go.opentelemetry.io/otel/trace"
+)
+
+// RecordBufferWait records one host's additional wait before publication. Time
+// already covered by its local processing span is excluded. A still-running
+// processing span (including a synchronous flush) therefore gets a zero-duration
+// observation, keeping span-metric counts representative of all buffered hosts.
+func RecordBufferWait(parent trace.Span, topic string, enqueuedAt, publishAt time.Time) {
+	if !MQEnabled() {
+		return
+	}
+	processing, ok := parent.(sdktrace.ReadOnlySpan)
+	if !ok {
+		// Remote and unsampled spans cannot supply a local processing end time.
+		return
+	}
+	start := enqueuedAt
+	processedAt := processing.EndTime()
+	if processedAt.IsZero() || processedAt.After(publishAt) {
+		start = publishAt
+	} else if processedAt.After(start) {
+		start = processedAt
+	}
+
+	ctx := trace.ContextWithSpan(context.Background(), parent)
+	_, span := Tracer().Start(ctx, "buffer wait "+topic,
+		trace.WithTimestamp(start),
+		trace.WithSpanKind(trace.SpanKindInternal),
+		trace.WithAttributes(attribute.String("messaging.destination.name", topic)),
+	)
+	span.End(trace.WithTimestamp(publishAt))
+}

base/telemetry/buffer_test.go

@@ -0,0 +1,72 @@
+package telemetry
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/trace"
+)
+
+func TestRecordBufferWaitExcludesProcessingOverlap(t *testing.T) {
+	const topic = "evaluation"
+	enqueuedAt := time.Now()
+	publishAt := enqueuedAt.Add(time.Second)
+	cases := []struct {
+		name        string
+		processedAt time.Time
+		wantStart   time.Time
+	}{
+		{"already processed", enqueuedAt.Add(-time.Millisecond), enqueuedAt},
+		{"partial overlap", enqueuedAt.Add(100 * time.Millisecond), enqueuedAt.Add(100 * time.Millisecond)},
+		{"still processing", time.Time{}, publishAt},
+		{"ended during publish", publishAt.Add(time.Millisecond), publishAt},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			exp := setupKafkaTest(t)
+			_, parent := Tracer().Start(t.Context(), "processing", trace.WithTimestamp(enqueuedAt.Add(-time.Second)))
+			SetRHAttributes(parent, "buffer-org", "buffer-request")
+			if !tc.processedAt.IsZero() {
+				parent.End(trace.WithTimestamp(tc.processedAt))
+			}
+			RecordBufferWait(parent, topic, enqueuedAt, publishAt)
+			if tc.processedAt.IsZero() {
+				assert.True(t, parent.IsRecording())
+			} else {
+				assert.False(t, parent.IsRecording())
+			}
+			spans := exp.GetSpans()
+			require.NotEmpty(t, spans)
+			wait := spans[len(spans)-1]
+			assert.Equal(t, "buffer wait "+topic, wait.Name)
+			assert.Equal(t, tc.wantStart, wait.StartTime)
+			assert.Equal(t, publishAt, wait.EndTime)
+			assert.Equal(t, parent.SpanContext().SpanID(), wait.Parent.SpanID())
+			assert.Equal(t, parent.SpanContext().TraceID(), wait.SpanContext.TraceID())
+			assert.Equal(t, trace.SpanKindInternal, wait.SpanKind)
+			assert.Equal(t, topic, attr(wait, "messaging.destination.name"))
+			assert.Equal(t, "buffer-org", attr(wait, "rh.org_id"))
+			assert.Equal(t, "buffer-request", attr(wait, "rh.request_id"))
+			parent.End()
+		})
+	}
+}
+
+func TestRecordBufferWaitDisabledOrMissingLocalSpan(t *testing.T) {
+	exp := setupKafkaTest(t)
+	now := time.Now()
+	RecordBufferWait(trace.SpanFromContext(context.Background()), "evaluation", now, now)
+	remote := ContextFromTraceparent("00-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-bbbbbbbbbbbbbbbb-01")
+	RecordBufferWait(trace.SpanFromContext(remote), "evaluation", now, now)
+	assert.Empty(t, exp.GetSpans())
+
+	_, parent := Tracer().Start(t.Context(), "processing")
+	t.Setenv("OTEL_MQ_ENABLED", "false")
+	RecordBufferWait(parent, "evaluation", now, now)
+	assert.Empty(t, exp.GetSpans())
+	assert.True(t, parent.IsRecording())
+	parent.End()
+}

base/telemetry/kafka.go

@@ -137,18 +137,25 @@ func ProducerContext(ctx context.Context, topic string, links []trace.Link) (con
 	return Tracer().Start(ctx, "send "+topic, opts...)
 }
 
+// ItemContext nests work under the current span when it shares the item's trace.
+// Items from other traces retain their original parent and link to the current span.
 func ItemContext(parent context.Context, name, traceparent string) (context.Context, trace.Span) {
 	if !Enabled() {
 		return parent, trace.SpanFromContext(parent)
 	}
 	startCtx := parent
+	opts := []trace.SpanStartOption{trace.WithSpanKind(trace.SpanKindInternal)}
 	if traceparent != "" {
-		orig := ContextFromTraceparent(traceparent)
-		if trace.SpanContextFromContext(orig).IsValid() {
-			startCtx = trace.ContextWithRemoteSpanContext(parent, trace.SpanContextFromContext(orig))
+		origSC := trace.SpanContextFromContext(ContextFromTraceparent(traceparent))
+		parentSC := trace.SpanContextFromContext(parent)
+		if origSC.IsValid() && (!parentSC.IsValid() || origSC.TraceID() != parentSC.TraceID()) {
+			startCtx = trace.ContextWithRemoteSpanContext(parent, origSC)
+			if parentSC.IsValid() {
+				opts = append(opts, trace.WithLinks(trace.Link{SpanContext: parentSC}))
+			}
 		}
 	}
-	return Tracer().Start(startCtx, name, trace.WithSpanKind(trace.SpanKindInternal))
+	return Tracer().Start(startCtx, name, opts...)
 }
 
 func SetRHAttributes(span trace.Span, orgID, requestID string) {

base/telemetry/kafka_test.go

@@ -123,7 +123,7 @@ func TestDisabledProducerDoesNotEndParent(t *testing.T) {
 	assert.Empty(t, spans[0].Events, "a disabled producer must not record errors on the parent")
 }
 
-func TestItemContextParentsOriginalTrace(t *testing.T) {
+func TestItemContextParentsConsumerInSameTrace(t *testing.T) {
 	exp := setupKafkaTest(t)
 
 	origCtx, orig := Tracer().Start(context.Background(), "original-item")
@@ -157,14 +157,95 @@ func TestItemContextParentsOriginalTrace(t *testing.T) {
 
 	origSC := trace.SpanContextFromContext(ContextFromTraceparent(originalTraceparent))
 	assert.Equal(t, origSC.TraceID(), itemStub.SpanContext.TraceID())
-	assert.Equal(t, origSC.SpanID(), itemStub.Parent.SpanID())
+	assert.Equal(t, bulkStub.SpanContext.SpanID(), itemStub.Parent.SpanID())
 	assert.Empty(t, itemStub.Links)
 
 	assert.Equal(t, origSC.TraceID(), producerStub.SpanContext.TraceID())
 	assert.Equal(t, origSC.SpanID(), producerStub.Parent.SpanID())
 	assert.Equal(t, producerStub.SpanContext.SpanID(), bulkStub.Parent.SpanID())
 }
 
+func TestItemContextMixedTraceBatch(t *testing.T) {
+	for _, label := range []string{"upload", "user-evaluation"} {
+		t.Run(label, func(t *testing.T) {
+			exp := setupKafkaTest(t)
+			name := "evaluate " + label
+
+			ctx0, original0 := Tracer().Start(t.Context(), "original-0")
+			tp0 := EncodeTraceparent(ctx0)
+			original0.End()
+			ctx1, original1 := Tracer().Start(t.Context(), "original-1")
+			tp1 := EncodeTraceparent(ctx1)
+			original1.End()
+
+			producerCtx, producer := ProducerContext(ParentContextFromTraceparents([]string{tp0, tp1}),
+				"patchman.evaluator.upload", nil)
+			headers := Inject(producerCtx, nil)
+			producer.End()
+			bulkCtx, bulk := ConsumerContext(t.Context(), "patchman.evaluator.upload", headers)
+
+			_, item0 := ItemContext(bulkCtx, name, tp0)
+			item0.End()
+			_, item1 := ItemContext(bulkCtx, name, tp1)
+			item1.End()
+			bulk.End()
+
+			spans := flushSpans(t, exp)
+			var item0Stub, item1Stub tracetest.SpanStub
+			for _, s := range spans {
+				switch s.SpanContext.SpanID() {
+				case item0.SpanContext().SpanID():
+					item0Stub = s
+				case item1.SpanContext().SpanID():
+					item1Stub = s
+				}
+			}
+			require.NotEmpty(t, item0Stub.Name)
+			require.NotEmpty(t, item1Stub.Name)
+			assert.Equal(t, original0.SpanContext().TraceID(), item0Stub.SpanContext.TraceID())
+			assert.Equal(t, bulk.SpanContext().SpanID(), item0Stub.Parent.SpanID())
+			assert.Empty(t, item0Stub.Links)
+
+			assert.Equal(t, original1.SpanContext().TraceID(), item1Stub.SpanContext.TraceID())
+			assert.Equal(t, original1.SpanContext().SpanID(), item1Stub.Parent.SpanID())
+			require.Len(t, item1Stub.Links, 1)
+			assert.Equal(t, bulk.SpanContext(), item1Stub.Links[0].SpanContext)
+		})
+	}
+}
+
+func TestItemContextFallsBackToAvailableParent(t *testing.T) {
+	const name = "evaluate fallback"
+	for _, tt := range []struct {
+		name        string
+		traceparent string
+		withParent  bool
+	}{
+		{name: "missing traceparent", withParent: true},
+		{name: "invalid traceparent", traceparent: "invalid", withParent: true},
+		{name: "no consumer span"},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			exp := setupKafkaTest(t)
+			parentCtx, parent := Tracer().Start(t.Context(), "available-parent")
+			ctx, tp := parentCtx, tt.traceparent
+			if !tt.withParent {
+				ctx, tp = t.Context(), EncodeTraceparent(parentCtx)
+			}
+			_, item := ItemContext(ctx, name, tp)
+			item.End()
+			parent.End()
+
+			spans := flushSpans(t, exp)
+			require.Len(t, spans, 2)
+			assert.Equal(t, name, spans[0].Name)
+			assert.Equal(t, parent.SpanContext().TraceID(), spans[0].SpanContext.TraceID())
+			assert.Equal(t, parent.SpanContext().SpanID(), spans[0].Parent.SpanID())
+			assert.Empty(t, spans[0].Links)
+		})
+	}
+}
+
 func TestItemContextPreservesCallerContext(t *testing.T) {
 	setupKafkaTest(t)
 	originalCtx, original := Tracer().Start(t.Context(), "original")

listener/event_buffers.go

@@ -10,27 +10,37 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"go.opentelemetry.io/otel/trace"
 )
 
+type bufferedTrace struct {
+	parent     trace.Span
+	enqueuedAt time.Time
+}
+
 type eventBuffer struct {
 	evalBuffer mqueue.EvalDataSlice
 	ptBuffer   mqueue.PayloadTrackerEvents
+	traces     []bufferedTrace
 	lock       sync.Mutex
 	flushTimer *time.Timer
 	evalWriter *mqueue.Writer
 	ptWriter   *mqueue.Writer
+	evalTopic  string
 }
 
-func (b *eventBuffer) initEventBuffer(evalWriter, ptWriter *mqueue.Writer) {
+func (b *eventBuffer) initEventBuffer(evalWriter, ptWriter *mqueue.Writer, evalTopic string) {
 	b.evalBuffer = make(mqueue.EvalDataSlice, 0, eventBufferSize+1)
 	b.ptBuffer = make(mqueue.PayloadTrackerEvents, 0, eventBufferSize+1)
+	b.traces = nil
 	b.lock = sync.Mutex{}
 	b.flushTimer = time.AfterFunc(87600*time.Hour, func() {
 		utils.LogInfo(FlushedTimeoutBuffer)
 		b.flushEvalEvents()
 	})
 	b.evalWriter = evalWriter
 	b.ptWriter = ptWriter
+	b.evalTopic = evalTopic
 }
 
 // send events after full buffer or timeout
@@ -52,6 +62,9 @@ func (b *eventBuffer) bufferEvalEvents(
 	}
 	b.evalBuffer = append(b.evalBuffer, evalData)
 	b.ptBuffer = append(b.ptBuffer, *ptEvent)
+	if telemetry.MQEnabled() {
+		b.traces = append(b.traces, bufferedTrace{parent: trace.SpanFromContext(ctx), enqueuedAt: time.Now()})
+	}
 
 	b.flushTimer.Reset(uploadEvalTimeout)
 	shouldFlush := len(b.evalBuffer) >= eventBufferSize
@@ -67,23 +80,30 @@ func (b *eventBuffer) flushEvalEvents() {
 	tStart := time.Now()
 	b.lock.Lock()
 	defer b.lock.Unlock()
+	if len(b.evalBuffer) == 0 {
+		return
+	}
+	b.flushTimer.Stop()
 
 	tps := make([]string, 0, len(b.evalBuffer))
 	for _, e := range b.evalBuffer {
 		tps = append(tps, e.Traceparent)
 	}
+	publishAt := time.Now()
 	pctx, span := telemetry.ProducerContext(
 		telemetry.ParentContextFromTraceparents(tps),
-		utils.CoreCfg.EvalTopic,
+		b.evalTopic,
 		nil,
 	)
-	var err error
-	defer func() { telemetry.End(span, err) }()
-	err = mqueue.SendMessages(pctx, *b.evalWriter, b.evalBuffer)
+	err := mqueue.SendMessages(pctx, *b.evalWriter, b.evalBuffer)
+	telemetry.End(span, err)
 	if err != nil {
 		utils.LogError("err", err, ErrorKafkaSend)
 	}
 	utils.ObserveSecondsSince(tStart, messagePartDuration.WithLabelValues("buffer-sent-evaluator"))
+	for _, buffered := range b.traces {
+		telemetry.RecordBufferWait(buffered.parent, b.evalTopic, buffered.enqueuedAt, publishAt)
+	}
 	if ptErr := mqueue.SendMessages(base.Context, *b.ptWriter, b.ptBuffer); ptErr != nil {
 		utils.LogWarn("err", ptErr, WarnPayloadTracker)
 	}
@@ -93,4 +113,6 @@ func (b *eventBuffer) flushEvalEvents() {
 	// empty buffer
 	b.evalBuffer = b.evalBuffer[:0]
 	b.ptBuffer = b.ptBuffer[:0]
+	clear(b.traces)
+	b.traces = b.traces[:0]
 }