From 07022cbeba775e886a738b4af4993fca12c01a1a Mon Sep 17 00:00:00 2001 From: Rodrigo Antunes Date: Tue, 27 Oct 2026 10:22:34 -0300 Subject: [PATCH 1/3] RHINENG-28312: fix evaluation span parenting Keep per-host evaluation spans under the active consumer when they share a trace. This makes upload and user-evaluation work appear inside the processing span that actually contains it. For mixed-trace batches, preserve each host's original trace and link its evaluation to the active consumer. Cover both evaluation labels, mixed batches, and missing or invalid parent contexts. Assisted-by: GPT-6 --- base/telemetry/kafka.go | 15 +++++-- base/telemetry/kafka_test.go | 85 +++++++++++++++++++++++++++++++++++- 2 files changed, 94 insertions(+), 6 deletions(-) diff --git a/base/telemetry/kafka.go b/base/telemetry/kafka.go index 417dc7720..dbf8ddfe6 100644 --- a/base/telemetry/kafka.go +++ b/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) { diff --git a/base/telemetry/kafka_test.go b/base/telemetry/kafka_test.go index eea15f3cf..6bfcc2fb1 100644 --- a/base/telemetry/kafka_test.go +++ b/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,7 +157,7 @@ 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()) @@ -165,6 +165,87 @@ func TestItemContextParentsOriginalTrace(t *testing.T) { 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") From 008632b428b13da9d689c9ebf88d0ed3d1331d77 Mon Sep 17 00:00:00 2001 From: Rodrigo Antunes Date: Tue, 27 Oct 2026 10:22:43 -0300 Subject: [PATCH 2/3] RHINENG-28312: measure listener dispatch latency Use each buffer's destination for its producer span and end the span immediately after evaluation publishing. Payload tracking must not inflate the evaluation publish duration. Record one buffer-wait span per traced host for time between listener processing and publication. Exclude processing overlap and retain zero-duration observations for synchronous flushes so span metrics count hosts without counting the same time twice. Skip empty flushes and release buffered span references after publishing. Add timing and failure regression coverage and document the metric boundaries. Assisted-by: GPT-6 --- base/telemetry/buffer.go | 40 +++++++ base/telemetry/buffer_test.go | 72 ++++++++++++ docs/md/architecture.md | 16 +++ listener/event_buffers.go | 32 +++++- listener/event_buffers_test.go | 203 +++++++++++++++++++++++++++++++++ listener/listener.go | 4 +- 6 files changed, 360 insertions(+), 7 deletions(-) create mode 100644 base/telemetry/buffer.go create mode 100644 base/telemetry/buffer_test.go create mode 100644 listener/event_buffers_test.go diff --git a/base/telemetry/buffer.go b/base/telemetry/buffer.go new file mode 100644 index 000000000..56f7ab168 --- /dev/null +++ b/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)) +} diff --git a/base/telemetry/buffer_test.go b/base/telemetry/buffer_test.go new file mode 100644 index 000000000..65c6bebbf --- /dev/null +++ b/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() +} diff --git a/docs/md/architecture.md b/docs/md/architecture.md index 6ed5af4b1..1550bca4a 100644 --- a/docs/md/architecture.md +++ b/docs/md/architecture.md @@ -77,6 +77,22 @@ and the [major migration runbook](major-migration-runbook.md). Using container C ### Components cooperation schema ![](graphics/schema.png) +### Listener buffer tracing + +The listener buffers host evaluation requests before publishing them. Created hosts use the configured +`CREATED_SYSTEMS_TOPIC` (normally `patchman.evaluator.user-evaluation`); updated hosts use `EVAL_TOPIC` +(normally `patchman.evaluator.upload`). The `send ` producer span covers the evaluation-message publish +and ends before payload-tracker messages are published. + +With messaging tracing enabled, each buffered host with a recorded listener processing span also gets an internal +`buffer wait ` span in its own trace. Its duration is the interval from the later of enqueue time and the end +of listener processing to the start of the evaluation publish. It excludes processing overlap and Kafka publishing. +If processing is still active at publication, as for the host that triggers a synchronous full-buffer flush, the +buffer-wait duration is zero. These zero observations keep span-metric averages representative of all traced hosts. +The spans retain the listener's organization and request attributes; their counts are per host, while producer +counts are per buffer flush. Buffer-wait durations can be added to listener processing durations without counting +the same interval twice. The synchronous producer remains inside the listener processing interval. + ### Template content update flow When Content Sources publishes a `template-updated` event: diff --git a/listener/event_buffers.go b/listener/event_buffers.go index 76502c5a2..36e3b08c7 100644 --- a/listener/event_buffers.go +++ b/listener/event_buffers.go @@ -10,20 +10,29 @@ 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) @@ -31,6 +40,7 @@ func (b *eventBuffer) initEventBuffer(evalWriter, ptWriter *mqueue.Writer) { }) 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] } diff --git a/listener/event_buffers_test.go b/listener/event_buffers_test.go new file mode 100644 index 000000000..47b7dc61f --- /dev/null +++ b/listener/event_buffers_test.go @@ -0,0 +1,203 @@ +package listener + +import ( + "app/base/mqueue" + "app/base/telemetry" + "app/base/utils" + "context" + "errors" + "testing" + "time" + + "github.com/bytedance/sonic" + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "go.opentelemetry.io/otel/trace" +) + +type bufferTestWriter func(context.Context, ...mqueue.KafkaMessage) error + +func (w bufferTestWriter) WriteMessages(ctx context.Context, messages ...mqueue.KafkaMessage) error { + return w(ctx, messages...) +} + +func setupBufferTest(t *testing.T, size int) *tracetest.SpanRecorder { + t.Helper() + previousProvider, previousPropagator := otel.GetTracerProvider(), otel.GetTextMapPropagator() + previousSize, previousTimeout := eventBufferSize, uploadEvalTimeout + t.Setenv("OTEL_ENABLED", "true") + t.Setenv("OTEL_MQ_ENABLED", "true") + require.NoError(t, telemetry.Init()) + // Use an in-memory provider; no spans go to the production OTLP exporter. + recorder := tracetest.NewSpanRecorder() + provider := sdktrace.NewTracerProvider( + sdktrace.WithSpanProcessor(recorder), + sdktrace.WithSpanProcessor(&telemetry.RHAttributeSpanProcessor{}), + ) + otel.SetTracerProvider(provider) + eventBufferSize, uploadEvalTimeout = size, time.Hour + t.Cleanup(func() { + require.NoError(t, provider.Shutdown(context.Background())) + require.NoError(t, telemetry.Shutdown(context.Background())) + otel.SetTracerProvider(previousProvider) + otel.SetTextMapPropagator(previousPropagator) + eventBufferSize, uploadEvalTimeout = previousSize, previousTimeout + }) + return recorder +} + +func newBufferTestEvent() *mqueue.PayloadTrackerEvent { + return &mqueue.PayloadTrackerEvent{ + InventoryID: uuid.New(), + OrgID: utils.PtrString("buffer-org"), + RequestID: utils.PtrString(uuid.NewString()), + } +} + +func findBufferTestSpans(recorder *tracetest.SpanRecorder, name string) []sdktrace.ReadOnlySpan { + var spans []sdktrace.ReadOnlySpan + for _, span := range recorder.Ended() { + if span.Name() == name { + spans = append(spans, span) + } + } + return spans +} + +func TestEventBufferProducerTopicAndLifetime(t *testing.T) { + for _, destination := range []string{"upload", "user-evaluation"} { + for _, failure := range []string{"none", "evaluation", "payload-tracker"} { + t.Run(destination+"/"+failure, func(t *testing.T) { + testBufferProducer(t, "patchman.evaluator."+destination, failure) + }) + } + } +} + +func testBufferProducer(t *testing.T, topic, failure string) { + t.Helper() + recorder := setupBufferTest(t, 1) + ctx, parent := telemetry.Tracer().Start(t.Context(), "processing") + defer parent.End() + event := newBufferTestEvent() + var producer trace.Span + var evaluationReturnedAt time.Time + var payloadCalls int + evaluationWriter := mqueue.Writer(bufferTestWriter(func(ctx context.Context, messages ...mqueue.KafkaMessage) error { + producer = trace.SpanFromContext(ctx) + assert.True(t, producer.IsRecording()) + require.Len(t, messages, 1) + var published mqueue.PlatformEvent + require.NoError(t, sonic.Unmarshal(messages[0].Value, &published)) + assert.Equal(t, []uuid.UUID{event.InventoryID}, published.SystemIDs) + assert.Equal(t, []string{telemetry.EncodeTraceparent(trace.ContextWithSpan(t.Context(), parent))}, + published.Traceparents) + evaluationReturnedAt = time.Now() + if failure == "evaluation" { + return errors.New("evaluation publish failed") + } + return nil + })) + payloadWriter := mqueue.Writer(bufferTestWriter(func(context.Context, ...mqueue.KafkaMessage) error { + payloadCalls++ + require.NotNil(t, producer) + assert.False(t, producer.IsRecording(), "producer must end before payload tracking starts") + if failure == "payload-tracker" { + return errors.New("payload tracking failed") + } + return nil + })) + var buffer eventBuffer + buffer.initEventBuffer(&evaluationWriter, &payloadWriter, topic) + t.Cleanup(func() { buffer.flushTimer.Stop() }) + buffer.bufferEvalEvents(event.InventoryID, 1, event, ctx) + assert.Equal(t, 1, payloadCalls) + spans := findBufferTestSpans(recorder, "send "+topic) + require.Len(t, spans, 1) + assert.Equal(t, trace.SpanKindProducer, spans[0].SpanKind()) + assert.Contains(t, spans[0].Attributes(), attribute.String("messaging.destination.name", topic)) + assert.False(t, spans[0].EndTime().Before(evaluationReturnedAt)) + wantStatus := codes.Unset + if failure == "evaluation" { + wantStatus = codes.Error + } + assert.Equal(t, wantStatus, spans[0].Status().Code) + buffer.flushEvalEvents() + assert.Len(t, findBufferTestSpans(recorder, "send "+topic), 1, "empty flush must not emit a producer") +} + +func TestEventBufferFullFlushWaitPerHost(t *testing.T) { + recorder := setupBufferTest(t, 2) + const topic = "full-buffer-evaluation" + writer := mqueue.Writer(&mqueue.MockKafkaWriter{}) + var buffer eventBuffer + buffer.initEventBuffer(&writer, &writer, topic) + t.Cleanup(func() { buffer.flushTimer.Stop() }) + ctx1, parent1 := telemetry.Tracer().Start(t.Context(), "first host") + event1 := newBufferTestEvent() + telemetry.SetRHAttributes(parent1, *event1.OrgID, *event1.RequestID) + buffer.bufferEvalEvents(event1.InventoryID, 1, event1, ctx1) + parent1.End() + ctx2, parent2 := telemetry.Tracer().Start(t.Context(), "second host") + event2 := newBufferTestEvent() + telemetry.SetRHAttributes(parent2, *event2.OrgID, *event2.RequestID) + buffer.bufferEvalEvents(event2.InventoryID, 1, event2, ctx2) + assert.True(t, parent2.IsRecording()) + parent2.End() + + waits := findBufferTestSpans(recorder, "buffer wait "+topic) + require.Len(t, waits, 2) + assert.Equal(t, parent1.SpanContext().TraceID(), waits[0].SpanContext().TraceID()) + assert.Equal(t, parent1.SpanContext().SpanID(), waits[0].Parent().SpanID()) + assert.Equal(t, parent1.(sdktrace.ReadOnlySpan).EndTime(), waits[0].StartTime()) + assert.True(t, waits[0].EndTime().After(waits[0].StartTime())) + assert.Equal(t, parent2.SpanContext().TraceID(), waits[1].SpanContext().TraceID()) + assert.Equal(t, parent2.SpanContext().SpanID(), waits[1].Parent().SpanID()) + assert.Equal(t, waits[1].StartTime(), waits[1].EndTime(), "synchronous host's wait is already in processing") + for i, event := range []*mqueue.PayloadTrackerEvent{event1, event2} { + assert.Contains(t, waits[i].Attributes(), attribute.String("rh.request_id", *event.RequestID)) + } + producer := findBufferTestSpans(recorder, "send "+topic) + require.Len(t, producer, 1) + assert.False(t, waits[0].EndTime().After(producer[0].StartTime()), "buffer wait excludes publishing") + buffer.flushEvalEvents() + assert.Len(t, findBufferTestSpans(recorder, "buffer wait "+topic), 2) +} + +func TestEventBufferTimerWait(t *testing.T) { + recorder := setupBufferTest(t, 2) + const topic = "timer-evaluation" + evaluationWriter := mqueue.Writer(&mqueue.MockKafkaWriter{}) + done := make(chan struct{}) + payloadWriter := mqueue.Writer(bufferTestWriter(func(context.Context, ...mqueue.KafkaMessage) error { + close(done) + return nil + })) + var buffer eventBuffer + buffer.initEventBuffer(&evaluationWriter, &payloadWriter, topic) + t.Cleanup(func() { buffer.flushTimer.Stop() }) + ctx, parent := telemetry.Tracer().Start(t.Context(), "timer host") + event := newBufferTestEvent() + buffer.bufferEvalEvents(event.InventoryID, 1, event, ctx) + parent.End() + buffer.flushTimer.Reset(time.Millisecond) + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("buffer timer did not publish") + } + buffer.lock.Lock() + defer buffer.lock.Unlock() + waits := findBufferTestSpans(recorder, "buffer wait "+topic) + require.Len(t, waits, 1) + assert.Equal(t, parent.(sdktrace.ReadOnlySpan).EndTime(), waits[0].StartTime()) + assert.True(t, waits[0].EndTime().After(waits[0].StartTime())) + assert.Empty(t, buffer.evalBuffer) + assert.Empty(t, buffer.traces) +} diff --git a/listener/listener.go b/listener/listener.go index 62ca812ac..2de900260 100644 --- a/listener/listener.go +++ b/listener/listener.go @@ -62,8 +62,8 @@ func configure() { ptWriter = mqueue.NewKafkaWriterFromEnv(ptTopic) createdSystemsWriter = mqueue.NewKafkaWriterFromEnv(createdTopic) - updatedEventsBuffer.initEventBuffer(&evalWriter, &ptWriter) - createdEventsBuffer.initEventBuffer(&createdSystemsWriter, &ptWriter) + updatedEventsBuffer.initEventBuffer(&evalWriter, &ptWriter, evalTopic) + createdEventsBuffer.initEventBuffer(&createdSystemsWriter, &ptWriter, createdTopic) configureListener() } From 424e2713dd07cab61dd20b9d4c9aab987ec37934 Mon Sep 17 00:00:00 2001 From: Rodrigo Antunes Date: Tue, 27 Oct 2026 10:29:39 -0300 Subject: [PATCH 3/3] RHINENG-28312: remove architecture doc additions Revert the listener buffer tracing section as requested, keeping the tracing fixes and regression tests in this change set. Assisted-by: GPT-6 --- docs/md/architecture.md | 16 ---------------- 1 file changed, 16 deletions(-) diff --git a/docs/md/architecture.md b/docs/md/architecture.md index 1550bca4a..6ed5af4b1 100644 --- a/docs/md/architecture.md +++ b/docs/md/architecture.md @@ -77,22 +77,6 @@ and the [major migration runbook](major-migration-runbook.md). Using container C ### Components cooperation schema ![](graphics/schema.png) -### Listener buffer tracing - -The listener buffers host evaluation requests before publishing them. Created hosts use the configured -`CREATED_SYSTEMS_TOPIC` (normally `patchman.evaluator.user-evaluation`); updated hosts use `EVAL_TOPIC` -(normally `patchman.evaluator.upload`). The `send ` producer span covers the evaluation-message publish -and ends before payload-tracker messages are published. - -With messaging tracing enabled, each buffered host with a recorded listener processing span also gets an internal -`buffer wait ` span in its own trace. Its duration is the interval from the later of enqueue time and the end -of listener processing to the start of the evaluation publish. It excludes processing overlap and Kafka publishing. -If processing is still active at publication, as for the host that triggers a synchronous full-buffer flush, the -buffer-wait duration is zero. These zero observations keep span-metric averages representative of all traced hosts. -The spans retain the listener's organization and request attributes; their counts are per host, while producer -counts are per buffer flush. Buffer-wait durations can be added to listener processing durations without counting -the same interval twice. The synchronous producer remains inside the listener processing interval. - ### Template content update flow When Content Sources publishes a `template-updated` event: