[core] add docs for metrics

Author: Michal Tichák

common/monitoring/common.go

@@ -0,0 +1,50 @@
+/*
+ * === 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 (
+	"hash/maphash"
+	"time"
+)
+
+type key struct {
+	nameTagsHash uint64
+	timestamp    time.Time
+}
+
+func metricNameTagsToHash(hash *maphash.Hash, metric *Metric) {
+	hash.WriteString(metric.name)
+
+	for _, tag := range metric.tags {
+		hash.WriteString(tag.name)
+		hash.WriteString(tag.value)
+	}
+}
+
+func hashValueAndReset(hash *maphash.Hash) uint64 {
+	hashValue := hash.Sum64()
+	hash.Reset()
+	return hashValue
+}

common/monitoring/metricsaggregate.go

@@ -29,11 +29,6 @@ import (
 	"time"
 )
 
-type key struct {
-	nameTagsHash uint64
-	timestamp    time.Time
-}
-
 type bucketsType map[key]*Metric
 
 type MetricsAggregate struct {
@@ -59,21 +54,6 @@ func (this *MetricsAggregate) AddMetric(metric *Metric) {
 	}
 }
 
-func metricNameTagsToHash(hash *maphash.Hash, metric *Metric) {
-	hash.WriteString(metric.name)
-
-	for _, tag := range metric.tags {
-		hash.WriteString(tag.name)
-		hash.WriteString(tag.value)
-	}
-}
-
-func hashValueAndReset(hash *maphash.Hash) uint64 {
-	hashValue := hash.Sum64()
-	hash.Reset()
-	return hashValue
-}
-
 func (this *MetricsAggregate) Clear() {
 	this.hash.Reset()
 	clear(this.metricsBuckets)

common/monitoring/metricsreservoirsampling.go

@@ -46,6 +46,7 @@ type MetricsReservoirSampling struct {
 func NewMetricsReservoirSampling() *MetricsReservoirSampling {
 	metrics := &MetricsReservoirSampling{}
 	metrics.metricsBuckets = make(bucketsReservoirSampleType)
+	metrics.hash.SetSeed(maphash.MakeSeed())
 	return metrics
 }
 

common/monitoring/monitoring.go

@@ -36,10 +36,11 @@ import (
 )
 
 var (
-	server                   *http.Server
+	// scraping endpoint implementation
+	server *http.Server
+	// objects to store incoming metrics
 	metricsInternal          *MetricsAggregate
 	metricsHistogramInternal *MetricsReservoirSampling
-	// metrics      []Metric
 	// channel that is used to request end of metrics server, it sends notification when server ended.
 	// It needs to be read!!!
 	endChannel chan struct{}
@@ -59,7 +60,7 @@ var (
 	log = logger.New(logrus.StandardLogger(), "metrics").WithField("level", infologger.IL_Devel)
 )
 
-func initChannels(messageBufferSize int) {
+func initChannels() {
 	endChannel = make(chan struct{})
 	metricsRequestedChannel = make(chan struct{})
 	// 100 was chosen arbitrarily as a number that seemed sensible to be high enough to provide nice buffer if
@@ -79,11 +80,12 @@ func closeChannels() {
 }
 
 // this eventLoop is the main part that processes all metrics send to the package
-// 3 events can happen:
+// 4 events can happen:
 //  1. metricsChannel receives message from Send() method. We just add the new metric to metrics slice
-//  2. metricsRequestChannel receives request to dump and request existing metrics. We send shallow copy of existing
+//  2. metricsHistosChannel receives message from Send() method. We just add the new metric to metrics slice
+//  3. metricsRequestChannel receives request to dump and request existing metrics. We send shallow copy of existing
 //     metrics to requestor (via metricsExportedToRequest channel) while resetting current metrics slice
-//  3. receive request to stop monitoring via endChannel. We send confirmation through endChannel to notify caller
+//  4. receive request to stop monitoring via endChannel. We send confirmation through endChannel to notify caller
 //     that eventLoop stopped
 func eventLoop() {
 	for {
@@ -144,15 +146,14 @@ func handleFunc(endpointName string) {
 
 // \param port port where the scraping endpoint will be created
 // \param endpointName name of the endpoint, which must start with a slash eg. "/internalmetrics"
-// \param messageBufferSize size of buffer for messages where messages are kept between scraping request.
 //
 // If we attempt send more messages than the size of the buffer, these overflowing messages will be ignored and warning will be logged.
-func Run(port uint16, endpointName string, messageBufferSize int) error {
+func Run(port uint16, endpointName string) error {
 	if IsRunning() {
 		return nil
 	}
 
-	initChannels(messageBufferSize)
+	initChannels()
 
 	go eventLoop()
 

core/config.go

@@ -129,7 +129,6 @@ func setDefaults() error {
 	viper.SetDefault("enableKafka", true)
 	viper.SetDefault("logAllIL", false)
 	viper.SetDefault("metricsEndpoint", "8088/ecsmetrics")
-	viper.SetDefault("metricsBufferSize", 1000000)
 	return nil
 }
 
@@ -201,7 +200,6 @@ func setFlags() error {
 	pflag.Bool("enableKafka", viper.GetBool("enableKafka"), "Turn on the kafka messaging")
 	pflag.Bool("logAllIL", viper.GetBool("logAllIL"), "Send all the logs into IL, including Debug and Trace messages")
 	pflag.String("metricsEndpoint", viper.GetString("metricsEndpoint"), "Http endpoint from which metrics can be scraped: [port/endpoint]")
-	pflag.Int("metricsBufferSize", viper.GetInt("metricsBufferSize"), "Limit for how many metrics can be stored in buffer in between scraping requests")
 
 	pflag.Parse()
 	return viper.BindPFlags(pflag.CommandLine)

core/core.go

@@ -82,7 +82,7 @@ func runMetrics() {
 
 	go func() {
 		log.Infof("Starting to listen on endpoint %s:%d for metrics", endpoint, port)
-		if err := monitoring.Run(port, fmt.Sprintf("/%s", endpoint), viper.GetInt("metricsBufferSize")); err != nil && err != http.ErrServerClosed {
+		if err := monitoring.Run(port, fmt.Sprintf("/%s", endpoint)); err != nil && err != http.ErrServerClosed {
 			ecsmetrics.StopGolangMetrics()
 			log.Errorf("failed to run metrics on port %d and endpoint: %s")
 		}