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/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") 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() }