[OCTRL-1001]: Levarage kafka-go writer's ability to send messages in batches

common/ecsmetrics/metric.go

@@ -12,3 +12,19 @@ func NewMetric(name string) monitoring.Metric {
 	metric.AddTag("subsystem", "ECS")
 	return metric
 }
+
+// Timer* functions are meant to be used with defer statement to measure runtime of given function:
+// defer TimerNS(&metric)()
+func TimerMS(metric *monitoring.Metric) func() {
+	start := time.Now()
+	return func() {
+		metric.AddValue("execution_time_ms", time.Since(start).Milliseconds())
+	}
+}
+
+func TimerNS(metric *monitoring.Metric) func() {
+	start := time.Now()
+	return func() {
+		metric.AddValue("execution_time_ns", time.Since(start).Nanoseconds())
+	}
+}

common/ecsmetrics/metrics_test.go

@@ -0,0 +1,28 @@
+package ecsmetrics
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/AliceO2Group/Control/common/monitoring"
+)
+
+func measureFunc(metric *monitoring.Metric) {
+	defer TimerMS(metric)()
+	defer TimerNS(metric)()
+	time.Sleep(100 * time.Millisecond)
+}
+
+func TestSimpleStartStop(t *testing.T) {
+	metric := NewMetric("test")
+	measureFunc(&metric)
+	fmt.Println(metric.Values["execution_time_ms"])
+	fmt.Println(metric.Values["execution_time_ns"])
+	if metric.Values["execution_time_ms"].(int64) < 100 {
+		t.Error("wrong milliseconds")
+	}
+	if metric.Values["execution_time_ns"].(int64) < 100000000 {
+		t.Error("wrong nanoseconds")
+	}
+}

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()
+		})
+	})
+})