RHINENG-23045: refactoring eventBuffer

Author: TenSt

listener/event_buffers.go

@@ -8,80 +8,73 @@ import (
 	"time"
 )
 
-// accumulate events and create group PlatformEvents to save some resources
-var eventBufferSize = 5 * mqueue.BatchSize
-
 type eventBuffer struct {
-	EvalBuffer mqueue.EvalDataSlice
-	PtBuffer   mqueue.PayloadTrackerEvents
-	Lock       sync.Mutex
+	evalBuffer mqueue.EvalDataSlice
+	ptBuffer   mqueue.PayloadTrackerEvents
+	lock       sync.Mutex
 	flushTimer *time.Timer
+	evalWriter *mqueue.Writer
+	ptWriter   *mqueue.Writer
 }
 
-var updatedEventsBuffer = eventBuffer{
-	EvalBuffer: make(mqueue.EvalDataSlice, 0, eventBufferSize+1),
-	PtBuffer:   make(mqueue.PayloadTrackerEvents, 0, eventBufferSize+1),
-	Lock:       sync.Mutex{},
-}
-
-var createdEventsBuffer = eventBuffer{
-	EvalBuffer: make(mqueue.EvalDataSlice, 0, eventBufferSize+1),
-	PtBuffer:   make(mqueue.PayloadTrackerEvents, 0, eventBufferSize+1),
-	Lock:       sync.Mutex{},
-}
-
-func (b *eventBuffer) initFlushTimer(w *mqueue.Writer) {
+func (b *eventBuffer) initEventBuffer(evalWriter, ptWriter *mqueue.Writer) {
+	b.evalBuffer = make(mqueue.EvalDataSlice, 0, eventBufferSize+1)
+	b.ptBuffer = make(mqueue.PayloadTrackerEvents, 0, eventBufferSize+1)
+	b.lock = sync.Mutex{}
 	b.flushTimer = time.AfterFunc(87600*time.Hour, func() {
 		utils.LogInfo(FlushedTimeoutBuffer)
-		b.flushEvalEvents(w)
+		b.flushEvalEvents()
 	})
+	b.evalWriter = evalWriter
+	b.ptWriter = ptWriter
 }
 
 // send events after full buffer or timeout
 func (b *eventBuffer) bufferEvalEvents(
 	inventoryID string,
 	rhAccountID int,
 	ptEvent *mqueue.PayloadTrackerEvent,
-	w *mqueue.Writer,
 ) {
 	tStart := time.Now()
 	defer utils.ObserveSecondsSince(tStart, messagePartDuration.WithLabelValues("buffer-eval-events"))
 
-	b.Lock.Lock()
+	b.lock.Lock()
 	evalData := mqueue.EvalData{
 		InventoryID: inventoryID,
 		RhAccountID: rhAccountID,
 		OrgID:       ptEvent.OrgID,
 		RequestID:   *ptEvent.RequestID,
 	}
-	b.EvalBuffer = append(b.EvalBuffer, evalData)
-	b.PtBuffer = append(b.PtBuffer, *ptEvent)
-	b.Lock.Unlock()
+	b.evalBuffer = append(b.evalBuffer, evalData)
+	b.ptBuffer = append(b.ptBuffer, *ptEvent)
 
 	b.flushTimer.Reset(uploadEvalTimeout)
-	if len(b.EvalBuffer) >= eventBufferSize {
+	shouldFlush := len(b.evalBuffer) >= eventBufferSize
+	b.lock.Unlock()
+
+	if shouldFlush {
 		utils.LogInfo(FlushedFullBuffer)
-		b.flushEvalEvents(w)
+		b.flushEvalEvents()
 	}
 }
 
-func (b *eventBuffer) flushEvalEvents(w *mqueue.Writer) {
+func (b *eventBuffer) flushEvalEvents() {
 	tStart := time.Now()
-	b.Lock.Lock()
-	defer b.Lock.Unlock()
-	err := mqueue.SendMessages(base.Context, *w, b.EvalBuffer)
+	b.lock.Lock()
+	defer b.lock.Unlock()
+	err := mqueue.SendMessages(base.Context, *b.evalWriter, b.evalBuffer)
 	if err != nil {
 		utils.LogError("err", err.Error(), ErrorKafkaSend)
 	}
 	utils.ObserveSecondsSince(tStart, messagePartDuration.WithLabelValues("buffer-sent-evaluator"))
-	err = mqueue.SendMessages(base.Context, ptWriter, b.PtBuffer)
+	err = mqueue.SendMessages(base.Context, *b.ptWriter, b.ptBuffer)
 	if err != nil {
 		utils.LogWarn("err", err.Error(), WarnPayloadTracker)
 	}
 	utils.ObserveSecondsSince(tStart, messagePartDuration.WithLabelValues("buffer-sent-payload-tracker"))
-	utils.LogDebug("evaluator_messages", len(b.EvalBuffer),
-		"payload_tracker_messages", len(b.PtBuffer), "flushed buffers")
+	utils.LogDebug("evaluator_messages", len(b.evalBuffer),
+		"payload_tracker_messages", len(b.ptBuffer), "flushed buffers")
 	// empty buffer
-	b.EvalBuffer = b.EvalBuffer[:0]
-	b.PtBuffer = b.PtBuffer[:0]
+	b.evalBuffer = b.evalBuffer[:0]
+	b.ptBuffer = b.ptBuffer[:0]
 }

listener/listener.go

@@ -29,6 +29,11 @@ var (
 	uploadEvalTimeout    time.Duration
 	deletionThreshold    time.Duration
 	useTraceLevel        bool
+
+	// Event buffers
+	eventBufferSize     = 5 * mqueue.BatchSize
+	updatedEventsBuffer eventBuffer
+	createdEventsBuffer eventBuffer
 )
 
 const (
@@ -46,6 +51,9 @@ func configure() {
 	ptWriter = mqueue.NewKafkaWriterFromEnv(ptTopic)
 	createdSystemsWriter = mqueue.NewKafkaWriterFromEnv(createdTopic)
 
+	updatedEventsBuffer.initEventBuffer(&evalWriter, &ptWriter)
+	createdEventsBuffer.initEventBuffer(&createdSystemsWriter, &ptWriter)
+
 	configureListener()
 }
 
@@ -74,9 +82,6 @@ func configureListener() {
 	useTraceLevel = log.IsLevelEnabled(log.TraceLevel)
 
 	validReporters = loadValidReporters()
-
-	updatedEventsBuffer.initFlushTimer(&evalWriter)
-	createdEventsBuffer.initFlushTimer(&createdSystemsWriter)
 }
 
 func loadValidReporters() map[string]int {

listener/upload.go

@@ -179,9 +179,9 @@ func HandleUpload(event HostEvent) error {
 
 	ptEvent.StatusMsg = ProcessingStatus
 	if event.Type == "created" {
-		createdEventsBuffer.bufferEvalEvents(sys.InventoryID, sys.RhAccountID, &ptEvent, &createdSystemsWriter)
+		createdEventsBuffer.bufferEvalEvents(sys.InventoryID, sys.RhAccountID, &ptEvent)
 	} else {
-		updatedEventsBuffer.bufferEvalEvents(sys.InventoryID, sys.RhAccountID, &ptEvent, &evalWriter)
+		updatedEventsBuffer.bufferEvalEvents(sys.InventoryID, sys.RhAccountID, &ptEvent)
 	}
 	logAndObserve(UploadSuccess, ReceivedSuccess, &event, &ptEvent, tStart, SuccessStatus, false)
 	return nil