fixup! [core] OCTRL-1003

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: 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 ecsmetrics
 
 import (
@@ -7,7 +31,7 @@ import (
 )
 
 func NewMetric(name string) monitoring.Metric {
-	metric := monitoring.Metric{Name: name, Timestamp: time.Now()}
+	metric := monitoring.NewMetric(name, time.Now())
 	metric.AddTag("subsystem", "ECS")
 	return metric
 }
@@ -17,13 +41,13 @@ func NewMetric(name string) monitoring.Metric {
 func TimerMS(metric *monitoring.Metric) func() {
 	start := time.Now()
 	return func() {
-		metric.AddValueInt64("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.AddValueInt64("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: 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 ecsmetrics
 
 import (
@@ -38,9 +62,9 @@ func gather() monitoring.Metric {
 	for _, sample := range samples {
 		switch sample.Value.Kind() {
 		case internalmetrics.KindUint64:
-			metric.AddValueUInt64(sample.Name, sample.Value.Uint64())
+			metric.SetFieldUInt64(sample.Name, sample.Value.Uint64())
 		case internalmetrics.KindFloat64:
-			metric.AddValueFloat64(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

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: 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 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.AddValueUInt64("messages_failed", uint64(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,8 +145,8 @@ func (w *KafkaWriter) writingLoop() {
 			}
 
 			metric := w.newMetric(KAFKAWRITER)
-			metric.AddValueUInt64("messages_sent", uint64(len(messagesToSend)))
-			metric.AddValueUInt64("messages_failed", 0)
+			metric.SetFieldUInt64("messages_sent", uint64(len(messagesToSend)))
+			metric.SetFieldUInt64("messages_failed", 0)
 
 			w.writeFunction(messagesToSend, &metric)
 

common/monitoring/metric.go

@@ -1,3 +1,27 @@
+/*
+ * === 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 monitoring
 
 import (
@@ -10,58 +34,70 @@ import (
 
 type (
 	Tag struct {
-		Name  string
-		Value string
+		name  string
+		value string
 	}
 
 	TagsType   []Tag
-	ValuesType map[string]interface{}
+	FieldsType map[string]any
 )
 
 type Metric struct {
-	Name      string
-	Values    ValuesType
-	Tags      TagsType
-	Timestamp time.Time
+	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})
+	metric.tags = append(metric.tags, Tag{name: tagName, value: value})
 }
 
-func (metric *Metric) addValue(valueName string, value interface{}) {
-	if metric.Values == nil {
-		metric.Values = make(ValuesType)
+func (metric *Metric) setField(fieldName string, field any) {
+	if metric.fields == nil {
+		metric.fields = make(FieldsType)
 	}
-	metric.Values[valueName] = value
+	metric.fields[fieldName] = field
 }
 
-func (metric *Metric) AddValueInt64(valueName string, value int64) {
-	metric.addValue(valueName, value)
+func (metric *Metric) SetFieldInt64(fieldName string, field int64) {
+	metric.setField(fieldName, field)
 }
 
-func (metric *Metric) AddValueUInt64(valueName string, value uint64) {
-	metric.addValue(valueName, value)
+func (metric *Metric) SetFieldUInt64(fieldName string, field uint64) {
+	metric.setField(fieldName, field)
 }
 
-func (metric *Metric) AddValueFloat64(valueName string, value float64) {
-	metric.addValue(valueName, value)
+func (metric *Metric) SetFieldFloat64(fieldName string, field float64) {
+	metric.setField(fieldName, field)
 }
 
-func (metric *Metric) MergeValues(other *Metric) {
-	for valueName, value := range other.Values {
-		if storedValue, ok := metric.Values[valueName]; ok {
-			switch v := value.(type) {
+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.Values[valueName] = v + storedValue.(int64)
+				metric.fields[fieldName] = v + storedField.(int64)
 			case uint64:
-				metric.Values[valueName] = v + storedValue.(uint64)
+				metric.fields[fieldName] = v + storedField.(uint64)
 			case float64:
-				metric.Values[valueName] = v + storedValue.(float64)
-			case string:
+				metric.fields[fieldName] = v + storedField.(float64)
 			}
 		} else {
-			metric.Values[valueName] = value
+			metric.fields[fieldName] = field
 		}
 	}
 }
@@ -70,16 +106,16 @@ 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)
+		enc.StartLine(metric.name)
+		for _, tag := range metric.tags {
+			enc.AddTag(tag.name, tag.value)
 		}
 
-		for valueName, value := range metric.Values {
-			// we cannot panic as we provide accessors only for allowed type with AddValue*
-			enc.AddField(valueName, lp.MustNewValue(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)
+		enc.EndLine(metric.timestamp)
 	}
 
 	if err := enc.Err(); err != nil {

common/monitoring/metricsaggregate.go

@@ -1,3 +1,27 @@
+/*
+ * === 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 monitoring
 
 import (
@@ -27,20 +51,20 @@ func (this *MetricsAggregate) AddMetric(metric *Metric) {
 	metricNameTagsToHash(&this.hash, metric)
 	hashKey := hashValueAndReset(&this.hash)
 
-	k := key{nameTagsHash: hashKey, timestamp: time.Unix(metric.Timestamp.Unix(), 0)}
+	k := key{nameTagsHash: hashKey, timestamp: time.Unix(metric.timestamp.Unix(), 0)}
 	if storedMetric, ok := this.metricsBuckets[k]; ok {
-		storedMetric.MergeValues(metric)
+		storedMetric.MergeFields(metric)
 	} else {
 		this.metricsBuckets[k] = metric
 	}
 }
 
 func metricNameTagsToHash(hash *maphash.Hash, metric *Metric) {
-	hash.WriteString(metric.Name)
+	hash.WriteString(metric.name)
 
-	for _, tag := range metric.Tags {
-		hash.WriteString(tag.Name)
-		hash.WriteString(tag.Value)
+	for _, tag := range metric.tags {
+		hash.WriteString(tag.name)
+		hash.WriteString(tag.value)
 	}
 }
 
@@ -58,7 +82,7 @@ func (this *MetricsAggregate) Clear() {
 func (this *MetricsAggregate) GetMetrics() []Metric {
 	var result []Metric
 	for key, metric := range this.metricsBuckets {
-		metric.Timestamp = key.timestamp
+		metric.timestamp = key.timestamp
 		result = append(result, *metric)
 	}
 	return result