[core] add docs for metrics

Author: Michal Tichák

common/ecsmetrics/metric.go

@@ -30,7 +30,6 @@ import (
 	"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"
@@ -83,7 +82,7 @@ type KafkaWriter struct {
 }
 
 func (w *KafkaWriter) newMetric(name string) monitoring.Metric {
-	metric := ecsmetrics.NewMetric(name)
+	metric := monitoring.NewMetric(name)
 	metric.AddTag("topic", w.Topic)
 	return metric
 }
@@ -103,7 +102,7 @@ func NewWriterWithTopic(topic topic.Topic) *KafkaWriter {
 	}
 
 	writer.writeFunction = func(messages []kafka.Message, metric *monitoring.Metric) {
-		defer ecsmetrics.TimerNS(metric)()
+		defer monitoring.TimerNS(metric)()
 		if err := writer.WriteMessages(context.Background(), messages...); err != nil {
 			metric.SetFieldUInt64("messages_failed", uint64(len(messages)))
 			log.Errorf("failed to write %d messages to kafka with error: %v", len(messages), err)
@@ -243,7 +242,7 @@ func (w *KafkaWriter) WriteEventWithTimestamp(e interface{}, timestamp time.Time
 	metric := w.newMetric(KAFKAPREPARE)
 
 	func() {
-		defer ecsmetrics.TimerNS(&metric)()
+		defer monitoring.TimerNS(&metric)()
 		wrappedEvent, key, err := internalEventToKafkaEvent(e, timestamp)
 		if err != nil {
 			log.WithField("event", e).

common/event/writer.go

@@ -22,7 +22,7 @@
  * Intergovernmental Organization or submit itself to any jurisdiction.
  */
 
-package ecsmetrics
+package golangmetrics
 
 import (
 	internalmetrics "runtime/metrics"
@@ -57,7 +57,7 @@ func gather() monitoring.Metric {
 
 	internalmetrics.Read(samples)
 
-	metric := NewMetric("golangruntimemetrics")
+	metric := monitoring.NewMetric("golangruntimemetrics")
 
 	for _, sample := range samples {
 		switch sample.Value.Kind() {
@@ -76,7 +76,7 @@ func gather() monitoring.Metric {
 	return metric
 }
 
-func StartGolangMetrics(period time.Duration) {
+func Start(period time.Duration) {
 	log.WithField("level", infologger.IL_Devel).Info("Starting golang metrics reporting")
 	go func() {
 		log.Debug("Starting golang metrics goroutine")
@@ -96,7 +96,7 @@ func StartGolangMetrics(period time.Duration) {
 	}()
 }
 
-func StopGolangMetrics() {
+func Stop() {
 	endRequestChannel <- struct{}{}
 	<-endRequestChannel
 }

common/ecsmetrics/metrics.go common/golangmetrics/metrics.gocommon/ecsmetrics/metrics.go renamed to common/golangmetrics/metrics.go

@@ -22,29 +22,29 @@
  * Intergovernmental Organization or submit itself to any jurisdiction.
  */
 
-package ecsmetrics
+package monitoring
 
 import (
-	"testing"
+	"hash/maphash"
 	"time"
-
-	"github.com/AliceO2Group/Control/common/monitoring"
 )
 
-func measureFunc(metric *monitoring.Metric) {
-	defer TimerMS(metric)()
-	defer TimerNS(metric)()
-	time.Sleep(100 * time.Millisecond)
+type key struct {
+	nameTagsHash uint64
+	timestamp    time.Time
 }
 
-func TestSimpleStartStop(t *testing.T) {
-	metric := NewMetric("test")
-	measureFunc(&metric)
-	fields := metric.GetFields()
-	if fields["execution_time_ms"].(int64) < 100 {
-		t.Error("wrong milliseconds")
-	}
-	if fields["execution_time_ns"].(int64) < 100000000 {
-		t.Error("wrong nanoseconds")
+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/ecsmetrics/metrics_test.go common/monitoring/common.gocommon/ecsmetrics/metrics_test.go renamed to common/monitoring/common.go

@@ -49,17 +49,21 @@ type Metric struct {
 	timestamp time.Time
 }
 
-func NewMetric(name string, timestamp time.Time) Metric {
+// Return empty metric, it is used right now mostly in string
+// to report metrics correctly to influxdb use NewECSMetric
+func NewDefaultMetric(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
+// creates empty metric with tag subsystem=ECS, which
+// is used by Telegraf to send metrics from ECS to correct
+// bucket
+func NewMetric(name string) Metric {
+	metric := NewDefaultMetric(name, time.Now())
+	metric.AddTag("subsystem", "ECS")
+	return metric
 }
 
 func (metric *Metric) AddTag(tagName string, value string) {

common/monitoring/metric.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/metricsaggregate.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/metricsreservoirsampling.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()
 

common/monitoring/monitoring.go

@@ -69,13 +69,13 @@ func hasNumberOfMetrics(t *testing.T, timeout time.Duration, requiredMessages in
 }
 
 func TestSimpleStartStop(t *testing.T) {
-	go Run(1234, "/random", 100)
+	go Run(1234, "/random")
 	isRunningWithTimeout(t, time.Second)
 	Stop()
 }
 
 func TestStartMultipleStop(t *testing.T) {
-	go Run(1234, "/random", 100)
+	go Run(1234, "/random")
 	isRunningWithTimeout(t, time.Second)
 	Stop()
 	Stop()
@@ -86,7 +86,7 @@ func cleaningUpAfterTest() {
 }
 
 func initTest() {
-	go Run(12345, "notimportant", 100)
+	go Run(12345, "notimportant")
 }
 
 // decorator function that properly inits and cleans after higher level test of Monitoring package
@@ -131,7 +131,7 @@ func TestExportingMetrics(t *testing.T) {
 }
 
 func TestHttpRun(t *testing.T) {
-	go Run(9876, "/metrics", 10)
+	go Run(9876, "/metrics")
 	defer Stop()
 
 	isRunningWithTimeout(t, time.Second)
@@ -545,6 +545,24 @@ func TestMetricsHistogramObject(t *testing.T) {
 	}
 }
 
+func measureFunc(metric *Metric) {
+	defer TimerMS(metric)()
+	defer TimerNS(metric)()
+	time.Sleep(100 * time.Millisecond)
+}
+
+func TestTimers(t *testing.T) {
+	metric := NewMetric("test")
+	measureFunc(&metric)
+	fields := metric.fields
+	if fields["execution_time_ms"].(int64) < 100 {
+		t.Error("wrong milliseconds")
+	}
+	if fields["execution_time_ns"].(int64) < 100000000 {
+		t.Error("wrong nanoseconds")
+	}
+}
+
 func BenchmarkSimple(b *testing.B) {
 	cpuProfileFile, err := os.Create("cpu_profile.pprof")
 	if err != nil {
@@ -575,19 +593,8 @@ func BenchmarkSimple(b *testing.B) {
 	pprof.WriteHeapProfile(heapProfileFile)
 }
 
-// This benchmark cannot be run for too long as it will fill whole RAM even with
-// results:
-// goos: linux
-// goarch: amd64
-// pkg: github.com/AliceO2Group/Control/common/monitoring
-// cpu: 11th Gen Intel(R) Core(TM) i9-11900H @ 2.50GHz
-// BenchmarkSendingMetrics-16
-//
-// 123365481              192.6 ns/op
-// PASS
-// ok      github.com/AliceO2Group/Control/common/monitoring       44.686s
 func BenchmarkSendingMetrics(b *testing.B) {
-	go Run(12345, "/metrics", 100)
+	go Run(12345, "/metrics")
 
 	defer Stop()
 

common/monitoring/monitoring_test.go

@@ -0,0 +1,19 @@
+package monitoring
+
+import "time"
+
+// Timer* functions are meant to be used with defer statement to measure runtime of given function:
+// defer TimerNS(&metric)()
+func TimerMS(metric *Metric) func() {
+	start := time.Now()
+	return func() {
+		metric.SetFieldInt64("execution_time_ms", time.Since(start).Milliseconds())
+	}
+}
+
+func TimerNS(metric *Metric) func() {
+	start := time.Now()
+	return func() {
+		metric.SetFieldInt64("execution_time_ns", time.Since(start).Nanoseconds())
+	}
+}