[core] kafka writer does not discard messages

See OCTRL-1001 for more details

Author: Michal Tichák

common/event/event_suite_test.go

@@ -0,0 +1,13 @@
+package event_test
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestEvent(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Event Suite")
+}

common/event/fifobuffer.go

@@ -0,0 +1,87 @@
+/*
+ * === This file is part of ALICE O² ===
+ *
+ * Copyright 2025 CERN and copyright holders of ALICE O².
+ * Author: Teo Mrnjavac <teo.mrnjavac@cern.ch>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * In applying this license CERN does not waive the privileges and
+ * immunities granted to it by virtue of its status as an
+ * Intergovernmental Organization or submit itself to any jurisdiction.
+ */
+
+package event
+
+import (
+	"math"
+	"sync"
+)
+
+// This structure is meant to be used as a threadsafe FIFO with builtin waiting for new data
+// in its Pop and PopMultiple functions. It is meant to be used with multiple goroutines, it is a
+// waste of synchronization mechanisms if used synchronously.
+type FifoBuffer[T any] struct {
+	lock sync.Mutex
+	cond sync.Cond
+
+	buffer []T
+}
+
+func NewFifoBuffer[T any]() (result FifoBuffer[T]) {
+	result = FifoBuffer[T]{
+		lock: sync.Mutex{},
+	}
+	result.cond = *sync.NewCond(&result.lock)
+	return
+}
+
+func (this *FifoBuffer[T]) Push(value T) {
+	this.cond.L.Lock()
+	this.buffer = append(this.buffer, value)
+	this.cond.Signal()
+	this.cond.L.Unlock()
+}
+
+// Blocks until it has some value in internal buffer
+func (this *FifoBuffer[T]) PopMultiple(numberToPop uint) (result []T) {
+	this.cond.L.Lock()
+	defer this.cond.L.Unlock()
+
+	for len(this.buffer) == 0 {
+		this.cond.Wait()
+		// this check is used when ReleaseGoroutines is called on waiting goroutine
+		if len(this.buffer) == 0 {
+			return
+		}
+	}
+
+	result = make([]T, int(math.Min(float64(numberToPop), float64(len(this.buffer)))))
+	copy(result, this.buffer[0:len(result)])
+	this.buffer = this.buffer[len(result):]
+
+	return
+}
+
+func (this *FifoBuffer[T]) Length() int {
+	this.cond.L.Lock()
+	defer this.cond.L.Unlock()
+	return len(this.buffer)
+}
+
+func (this *FifoBuffer[T]) ReleaseGoroutines() {
+	this.cond.L.Lock()
+	this.cond.Broadcast()
+	this.cond.L.Unlock()
+}

common/event/fifobuffer_test.go

@@ -0,0 +1,129 @@
+/*
+ * === This file is part of ALICE O² ===
+ *
+ * Copyright 2025 CERN and copyright holders of ALICE O².
+ * Author: Teo Mrnjavac <teo.mrnjavac@cern.ch>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * In applying this license CERN does not waive the privileges and
+ * immunities granted to it by virtue of its status as an
+ * Intergovernmental Organization or submit itself to any jurisdiction.
+ */
+
+package event
+
+import (
+	"sync"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("FifoBuffer", func() {
+	When("Poping lower amount of items than inside of a buffer", func() {
+		It("returns requested items", func() {
+			buffer := NewFifoBuffer[int]()
+			buffer.Push(1)
+			buffer.Push(2)
+			buffer.Push(3)
+
+			Expect(buffer.Length()).To(Equal(3))
+
+			results := buffer.PopMultiple(2)
+			Expect(results).To(Equal([]int{1, 2}))
+		})
+	})
+
+	When("Poping higher amount of items than inside of a buffer", func() {
+		It("returns only available items", func() {
+			buffer := NewFifoBuffer[int]()
+			buffer.Push(1)
+
+			results := buffer.PopMultiple(2)
+			Expect(results).To(Equal([]int{1}))
+		})
+	})
+
+	When("We use buffer with multiple goroutines pushing first (PopMultiple)", func() {
+		It("is synchronised properly", func() {
+			buffer := NewFifoBuffer[int]()
+			channel := make(chan struct{})
+
+			wg := sync.WaitGroup{}
+			wg.Add(2)
+
+			go func() {
+				buffer.Push(1)
+				channel <- struct{}{}
+				wg.Done()
+			}()
+
+			go func() {
+				<-channel
+				result := buffer.PopMultiple(42)
+				Expect(result, 1)
+				wg.Done()
+			}()
+
+			wg.Wait()
+		})
+	})
+
+	When("We use buffer with multiple goroutines popping first", func() {
+		It("is synchronised properly", func() {
+			buffer := NewFifoBuffer[int]()
+			channel := make(chan struct{})
+
+			wg := sync.WaitGroup{}
+			wg.Add(2)
+
+			go func() {
+				// Pop is blocking is we have empty buffer, so we notify before
+				channel <- struct{}{}
+				result := buffer.PopMultiple(42)
+				Expect(result, 1)
+				wg.Done()
+			}()
+
+			go func() {
+				<-channel
+				buffer.Push(1)
+				wg.Done()
+			}()
+
+			wg.Wait()
+		})
+	})
+
+	When("We block FifoBuffer without data and call Release", func() {
+		It("releases goroutines properly", func() {
+			buffer := NewFifoBuffer[int]()
+			everythingDone := sync.WaitGroup{}
+			channel := make(chan struct{})
+
+			everythingDone.Add(1)
+			go func() {
+				channel <- struct{}{}
+				buffer.PopMultiple(42)
+				everythingDone.Done()
+			}()
+			<-channel
+			time.Sleep(100 * time.Millisecond)
+			buffer.ReleaseGoroutines()
+			everythingDone.Wait()
+		})
+	})
+})

common/event/writer.go

@@ -27,11 +27,14 @@ package event
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
+	"github.com/AliceO2Group/Control/common/ecsmetrics"
 	"github.com/AliceO2Group/Control/common/event/topic"
 	"github.com/AliceO2Group/Control/common/logger"
 	"github.com/AliceO2Group/Control/common/logger/infologger"
+	"github.com/AliceO2Group/Control/common/monitoring"
 	pb "github.com/AliceO2Group/Control/common/protos"
 	"github.com/segmentio/kafka-go"
 	"github.com/sirupsen/logrus"
@@ -53,9 +56,26 @@ func (*DummyWriter) WriteEvent(interface{})                         {}
 func (*DummyWriter) WriteEventWithTimestamp(interface{}, time.Time) {}
 func (*DummyWriter) Close()                                         {}
 
+// Kafka writer is used to convert events from events.proto into kafka messages and to write them.
+// it is built with 2 workers:
+//
+//	#1 is gathering kafka.Message from any goroutine which sends message into buffered channel and puts them into FifoBuffer.
+//	#2 is poping any messages from FifoBuffer and sends them to Kafka
+//
+//	The reason for this setup over setting Async: true in kafka.Writer is the ability to have some error handling
+//	of failed messages. Moreover if we used only one worker that gathers messages from channel and then sends them directly to Kafka,
+//	we would block whole core if we receive lot of messages at once. So we split functionality into two workers: one is
+//	putting all messages into the buffer, so if we have a lot of messages buffer just grows without blocking whole core and the
+//	second does all the sending. This setup allows us to gather messages from any amount of goroutines without blocking/losing messages.
+//	Another benefit is batching messages instead of writing them one by one.
 type KafkaWriter struct {
 	*kafka.Writer
-	toWriteChan chan kafka.Message
+	toBatchMessagesChan chan kafka.Message
+	messageBuffer       FifoBuffer[kafka.Message]
+	// NOTE: existence of this is to be able to test the writer without actually setting up kafka
+	writeFunction  func([]kafka.Message)
+	runningWorkers sync.WaitGroup
+	batchingDoneCh chan struct{}
 }
 
 func NewWriterWithTopic(topic topic.Topic) *KafkaWriter {
@@ -66,16 +86,27 @@ func NewWriterWithTopic(topic topic.Topic) *KafkaWriter {
 			Balancer:               &kafka.Hash{},
 			AllowAutoTopicCreation: true,
 		},
-		toWriteChan: make(chan kafka.Message, 1000),
+		toBatchMessagesChan: make(chan kafka.Message, 100),
+		messageBuffer:       NewFifoBuffer[kafka.Message](),
+		runningWorkers:      sync.WaitGroup{},
+		batchingDoneCh:      make(chan struct{}, 1),
+	}
+
+	writer.writeFunction = func(messages []kafka.Message) {
+		writer.WriteMessages(context.Background(), messages...)
 	}
 
 	go writer.writingLoop()
+	go writer.batchingLoop()
+
 	return writer
 }
 
 func (w *KafkaWriter) Close() {
 	if w != nil {
-		close(w.toWriteChan)
+		w.runningWorkers.Add(2)
+		close(w.toBatchMessagesChan)
+		w.runningWorkers.Wait()
 		w.Writer.Close()
 	}
 }
@@ -86,17 +117,36 @@ func (w *KafkaWriter) WriteEvent(e interface{}) {
 	}
 }
 
-// TODO: we can optimise this to write multiple message at once
 func (w *KafkaWriter) writingLoop() {
-	for message := range w.toWriteChan {
-		err := w.WriteMessages(context.Background(), message)
-		if err != nil {
-			log.WithField("level", infologger.IL_Support).
-				Errorf("failed to write async kafka message: %w", err)
+	for {
+		select {
+		case <-w.batchingDoneCh:
+			w.runningWorkers.Done()
+			return
+		default:
+			messagesToSend := w.messageBuffer.PopMultiple(100)
+			if len(messagesToSend) == 0 {
+				continue
+			}
+			w.writeFunction(messagesToSend)
+
+			metric := ecsmetrics.NewMetric("kafka")
+			metric.AddTag("topic", w.Topic)
+			metric.AddValue("sentmessages", len(messagesToSend))
+			monitoring.Send(metric)
 		}
 	}
 }
 
+func (w *KafkaWriter) batchingLoop() {
+	for message := range w.toBatchMessagesChan {
+		w.messageBuffer.Push(message)
+	}
+	w.batchingDoneCh <- struct{}{}
+	w.messageBuffer.ReleaseGoroutines()
+	w.runningWorkers.Done()
+}
+
 type HasEnvID interface {
 	GetEnvironmentId() string
 }
@@ -109,6 +159,7 @@ func extractAndConvertEnvID[T HasEnvID](object T) []byte {
 	return nil
 }
 
+// TODO: there should be written test to convert all of these messages
 func internalEventToKafkaEvent(internalEvent interface{}, timestamp time.Time) (kafkaEvent *pb.Event, key []byte, err error) {
 	kafkaEvent = &pb.Event{
 		Timestamp:     timestamp.UnixMilli(),
@@ -188,9 +239,5 @@ func (w *KafkaWriter) WriteEventWithTimestamp(e interface{}, timestamp time.Time
 		return
 	}
 
-	select {
-	case w.toWriteChan <- message:
-	default:
-		log.Warnf("Writer of kafka topic [%s] cannot write because channel is full, discarding a message", w.Writer.Topic)
-	}
+	w.toBatchMessagesChan <- message
 }