[core] OCTRL-1003

-aggregating metrics so we don't flod influxdb
-changed json to influxdb line format to properly handle tags

Author: Michal Tichák

common/ecsmetrics/metric.go

@@ -1,3 +1,27 @@
+/*
+ * === This file is part of ALICE O² ===
+ *
+ * Copyright 2025 CERN and copyright holders of ALICE O².
+ * Author: Michal Tichak <michal.tichak@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 ecsmetrics
 
 import (
@@ -7,8 +31,7 @@ import (
 )
 
 func NewMetric(name string) monitoring.Metric {
-	timestamp := time.Now()
-	metric := monitoring.Metric{Name: name, Timestamp: timestamp.UnixMilli()}
+	metric := monitoring.NewMetric(name, time.Now())
 	metric.AddTag("subsystem", "ECS")
 	return metric
 }
@@ -18,13 +41,13 @@ func NewMetric(name string) monitoring.Metric {
 func TimerMS(metric *monitoring.Metric) func() {
 	start := time.Now()
 	return func() {
-		metric.AddValue("execution_time_ms", time.Since(start).Milliseconds())
+		metric.SetFieldInt64("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())
+		metric.SetFieldInt64("execution_time_ns", time.Since(start).Nanoseconds())
 	}
 }

common/ecsmetrics/metrics.go

@@ -1,3 +1,27 @@
+/*
+ * === This file is part of ALICE O² ===
+ *
+ * Copyright 2025 CERN and copyright holders of ALICE O².
+ * Author: Michal Tichak <michal.tichak@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 ecsmetrics
 
 import (
@@ -38,9 +62,9 @@ func gather() monitoring.Metric {
 	for _, sample := range samples {
 		switch sample.Value.Kind() {
 		case internalmetrics.KindUint64:
-			metric.AddValue(sample.Name, sample.Value.Uint64())
+			metric.SetFieldUInt64(sample.Name, sample.Value.Uint64())
 		case internalmetrics.KindFloat64:
-			metric.AddValue(sample.Name, sample.Value.Float64())
+			metric.SetFieldFloat64(sample.Name, sample.Value.Float64())
 		case internalmetrics.KindFloat64Histogram:
 			log.WithField("level", infologger.IL_Devel).Warningf("Error: Histogram is not supported yet for metric [%s]", sample.Name)
 			continue
@@ -64,7 +88,8 @@ func StartGolangMetrics(period time.Duration) {
 				return
 			default:
 				log.Debug("sending golang metrics")
-				monitoring.Send(gather())
+				metric := gather()
+				monitoring.Send(&metric)
 				time.Sleep(period)
 			}
 		}

common/ecsmetrics/metrics_test.go

@@ -1,7 +1,30 @@
+/*
+ * === This file is part of ALICE O² ===
+ *
+ * Copyright 2025 CERN and copyright holders of ALICE O².
+ * Author: Michal Tichak <michal.tichak@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 ecsmetrics
 
 import (
-	"fmt"
 	"testing"
 	"time"
 
@@ -17,12 +40,11 @@ func measureFunc(metric *monitoring.Metric) {
 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 {
+	fields := metric.GetFields()
+	if fields["execution_time_ms"].(int64) < 100 {
 		t.Error("wrong milliseconds")
 	}
-	if metric.Values["execution_time_ns"].(int64) < 100000000 {
+	if fields["execution_time_ns"].(int64) < 100000000 {
 		t.Error("wrong nanoseconds")
 	}
 }

common/event/writer.go

@@ -105,7 +105,7 @@ func NewWriterWithTopic(topic topic.Topic) *KafkaWriter {
 	writer.writeFunction = func(messages []kafka.Message, metric *monitoring.Metric) {
 		defer ecsmetrics.TimerNS(metric)()
 		if err := writer.WriteMessages(context.Background(), messages...); err != nil {
-			metric.AddValue("messages_failed", len(messages))
+			metric.SetFieldUInt64("messages_failed", uint64(len(messages)))
 			log.Errorf("failed to write %d messages to kafka with error: %v", len(messages), err)
 		}
 	}
@@ -145,12 +145,12 @@ func (w *KafkaWriter) writingLoop() {
 			}
 
 			metric := w.newMetric(KAFKAWRITER)
-			metric.AddValue("messages_sent", len(messagesToSend))
-			metric.AddValue("messages_failed", 0)
+			metric.SetFieldUInt64("messages_sent", uint64(len(messagesToSend)))
+			metric.SetFieldUInt64("messages_failed", 0)
 
 			w.writeFunction(messagesToSend, &metric)
 
-			monitoring.Send(metric)
+			monitoring.Send(&metric)
 		}
 	}
 }
@@ -262,5 +262,5 @@ func (w *KafkaWriter) WriteEventWithTimestamp(e interface{}, timestamp time.Time
 		w.toBatchMessagesChan <- message
 	}()
 
-	monitoring.Send(metric)
+	monitoring.Send(&metric)
 }

common/monitoring/metric.go

@@ -1,27 +1,127 @@
+/*
+ * === This file is part of ALICE O² ===
+ *
+ * Copyright 2025 CERN and copyright holders of ALICE O².
+ * Author: Michal Tichak <michal.tichak@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 monitoring
 
+import (
+	"fmt"
+	"io"
+	"time"
+
+	lp "github.com/influxdata/line-protocol/v2/lineprotocol"
+)
+
 type (
-	TagsType   map[string]any
-	ValuesType map[string]any
+	Tag struct {
+		name  string
+		value string
+	}
+
+	TagsType   []Tag
+	FieldsType map[string]any
 )
 
 type Metric struct {
-	Name      string     `json:"name"`
-	Values    ValuesType `json:"values"`
-	Tags      TagsType   `json:"tags,omitempty"`
-	Timestamp int64      `json:"timestamp"`
+	name      string
+	fields    FieldsType
+	tags      TagsType
+	timestamp time.Time
+}
+
+func NewMetric(name string, timestamp time.Time) Metric {
+	return Metric{
+		name: name, timestamp: timestamp,
+	}
+}
+
+func (metric *Metric) GetFields() (fields FieldsType) {
+	for fieldName, field := range metric.fields {
+		fields[fieldName] = field
+	}
+	return
+}
+
+func (metric *Metric) AddTag(tagName string, value string) {
+	metric.tags = append(metric.tags, Tag{name: tagName, value: value})
 }
 
-func (metric *Metric) AddTag(tagName string, value any) {
-	if metric.Tags == nil {
-		metric.Tags = make(TagsType)
+func (metric *Metric) setField(fieldName string, field any) {
+	if metric.fields == nil {
+		metric.fields = make(FieldsType)
 	}
-	metric.Tags[tagName] = value
+	metric.fields[fieldName] = field
 }
 
-func (metric *Metric) AddValue(valueName string, value any) {
-	if metric.Values == nil {
-		metric.Values = make(ValuesType)
+func (metric *Metric) SetFieldInt64(fieldName string, field int64) {
+	metric.setField(fieldName, field)
+}
+
+func (metric *Metric) SetFieldUInt64(fieldName string, field uint64) {
+	metric.setField(fieldName, field)
+}
+
+func (metric *Metric) SetFieldFloat64(fieldName string, field float64) {
+	metric.setField(fieldName, field)
+}
+
+func (metric *Metric) MergeFields(other *Metric) {
+	for fieldName, field := range other.fields {
+		if storedField, ok := metric.fields[fieldName]; ok {
+			switch v := field.(type) {
+			case int64:
+				metric.fields[fieldName] = v + storedField.(int64)
+			case uint64:
+				metric.fields[fieldName] = v + storedField.(uint64)
+			case float64:
+				metric.fields[fieldName] = v + storedField.(float64)
+			}
+		} else {
+			metric.fields[fieldName] = field
+		}
+	}
+}
+
+func Format(writer io.Writer, metrics []Metric) error {
+	var enc lp.Encoder
+
+	for _, metric := range metrics {
+		enc.StartLine(metric.name)
+		for _, tag := range metric.tags {
+			enc.AddTag(tag.name, tag.value)
+		}
+
+		for fieldName, field := range metric.fields {
+			// we cannot panic as we provide accessors only for allowed type with AddField*
+			enc.AddField(fieldName, lp.MustNewValue(field))
+		}
+		enc.EndLine(metric.timestamp)
 	}
-	metric.Values[valueName] = value
+
+	if err := enc.Err(); err != nil {
+		return err
+	}
+
+	_, err := fmt.Fprintf(writer, "%s", enc.Bytes())
+	return err
 }