[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

@@ -7,8 +7,7 @@ import (
 )
 
 func NewMetric(name string) monitoring.Metric {
-	timestamp := time.Now()
-	metric := monitoring.Metric{Name: name, Timestamp: timestamp.UnixMilli()}
+	metric := monitoring.Metric{Name: name, Timestamp: time.Now()}
 	metric.AddTag("subsystem", "ECS")
 	return metric
 }
@@ -18,13 +17,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.AddValueInt64("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.AddValueInt64("execution_time_ns", time.Since(start).Nanoseconds())
 	}
 }

common/ecsmetrics/metrics.go

@@ -38,9 +38,9 @@ func gather() monitoring.Metric {
 	for _, sample := range samples {
 		switch sample.Value.Kind() {
 		case internalmetrics.KindUint64:
-			metric.AddValue(sample.Name, sample.Value.Uint64())
+			metric.AddValueUInt64(sample.Name, sample.Value.Uint64())
 		case internalmetrics.KindFloat64:
-			metric.AddValue(sample.Name, sample.Value.Float64())
+			metric.AddValueFloat64(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 +64,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/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.AddValueUInt64("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.AddValueUInt64("messages_sent", uint64(len(messagesToSend)))
+			metric.AddValueUInt64("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,91 @@
 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
+	ValuesType map[string]interface{}
 )
 
 type Metric struct {
-	Name      string     `json:"name"`
-	Values    ValuesType `json:"values"`
-	Tags      TagsType   `json:"tags,omitempty"`
-	Timestamp int64      `json:"timestamp"`
+	Name      string
+	Values    ValuesType
+	Tags      TagsType
+	Timestamp time.Time
 }
 
-func (metric *Metric) AddTag(tagName string, value any) {
-	if metric.Tags == nil {
-		metric.Tags = make(TagsType)
-	}
-	metric.Tags[tagName] = value
+func (metric *Metric) AddTag(tagName string, value string) {
+	metric.Tags = append(metric.Tags, Tag{Name: tagName, Value: value})
 }
 
-func (metric *Metric) AddValue(valueName string, value any) {
+func (metric *Metric) addValue(valueName string, value interface{}) {
 	if metric.Values == nil {
 		metric.Values = make(ValuesType)
 	}
 	metric.Values[valueName] = value
 }
+
+func (metric *Metric) AddValueInt64(valueName string, value int64) {
+	metric.addValue(valueName, value)
+}
+
+func (metric *Metric) AddValueUInt64(valueName string, value uint64) {
+	metric.addValue(valueName, value)
+}
+
+func (metric *Metric) AddValueFloat64(valueName string, value float64) {
+	metric.addValue(valueName, value)
+}
+
+func (metric *Metric) MergeValues(other *Metric) {
+	for valueName, value := range other.Values {
+		if storedValue, ok := metric.Values[valueName]; ok {
+			switch v := value.(type) {
+			case int64:
+				metric.Values[valueName] = v + storedValue.(int64)
+			case uint64:
+				metric.Values[valueName] = v + storedValue.(uint64)
+			case float64:
+				metric.Values[valueName] = v + storedValue.(float64)
+			case string:
+			}
+		} else {
+			metric.Values[valueName] = value
+		}
+	}
+}
+
+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 valueName, value := range metric.Values {
+			// we cannot panic as we provide accessors only for allowed type with AddValue*
+			enc.AddField(valueName, lp.MustNewValue(value))
+		}
+		enc.EndLine(metric.Timestamp)
+	}
+
+	if err := enc.Err(); err != nil {
+		return err
+	}
+
+	_, err := fmt.Fprintf(writer, "%s", enc.Bytes())
+	return err
+}

common/monitoring/metricsaggregate.go

@@ -0,0 +1,65 @@
+package monitoring
+
+import (
+	"hash/maphash"
+	"time"
+)
+
+type key struct {
+	nameTagsHash uint64
+	timestamp    time.Time
+}
+
+type bucketsType map[key]*Metric
+
+type MetricsAggregate struct {
+	hash           maphash.Hash
+	metricsBuckets bucketsType
+}
+
+func NewMetricsAggregate() *MetricsAggregate {
+	metrics := &MetricsAggregate{}
+	metrics.metricsBuckets = make(bucketsType)
+	return metrics
+}
+
+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)}
+	if storedMetric, ok := this.metricsBuckets[k]; ok {
+		storedMetric.MergeValues(metric)
+	} else {
+		this.metricsBuckets[k] = 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)
+}
+
+func (this *MetricsAggregate) GetMetrics() []Metric {
+	var result []Metric
+	for key, metric := range this.metricsBuckets {
+		metric.Timestamp = key.timestamp
+		result = append(result, *metric)
+	}
+	return result
+}

common/monitoring/metricsreservoirsampling.go

@@ -0,0 +1,144 @@
+package monitoring
+
+import (
+	"hash/maphash"
+	"math/rand"
+	"sort"
+	"time"
+)
+
+type bucketsReservoirSampleType map[key]*metricReservoirSample
+
+type metricReservoirSample struct {
+	metric    Metric
+	reservoir reservoirSampling
+}
+
+type MetricsReservoirSampling struct {
+	hash           maphash.Hash
+	metricsBuckets bucketsReservoirSampleType
+}
+
+func NewMetricsReservoirSampling() *MetricsReservoirSampling {
+	metrics := &MetricsReservoirSampling{}
+	metrics.metricsBuckets = make(bucketsReservoirSampleType)
+	return metrics
+}
+
+func metricValueToFloat64(value interface{}) float64 {
+	var asserted float64
+	switch v := value.(type) {
+	case int64:
+		asserted = float64(v)
+	case uint64:
+		asserted = float64(v)
+	case float64:
+		asserted = v
+	}
+	return asserted
+}
+
+func (this *MetricsReservoirSampling) AddMetric(metric *Metric) {
+	for valueKey, value := range metric.Values {
+		metricNameTagsToHash(&this.hash, metric)
+		this.hash.WriteString(valueKey)
+		k := key{nameTagsHash: hashValueAndReset(&this.hash), timestamp: time.Unix(metric.Timestamp.Unix(), 0)}
+		if storedMetric, ok := this.metricsBuckets[k]; !ok {
+			newReservoir := newReservoirSampling(valueKey, 1000)
+			newReservoir.AddPoint(metricValueToFloat64(value))
+			this.metricsBuckets[k] = &metricReservoirSample{metric: *metric, reservoir: newReservoir}
+		} else {
+			storedMetric.reservoir.AddPoint(metricValueToFloat64(value))
+		}
+	}
+}
+
+func (this *MetricsReservoirSampling) Clear() {
+	this.hash.Reset()
+	clear(this.metricsBuckets)
+}
+
+func (this *MetricsReservoirSampling) GetMetrics() []Metric {
+	var result []Metric
+	for key, reservoirMetric := range this.metricsBuckets {
+		m := Metric{Name: reservoirMetric.metric.Name, Tags: reservoirMetric.metric.Tags, Timestamp: key.timestamp}
+
+		mean, median, min, p10, p30, p70, p90, max, count, poolSize := reservoirMetric.reservoir.GetStats()
+
+		m.AddValueFloat64(reservoirMetric.reservoir.name+"_mean", mean)
+		m.AddValueFloat64(reservoirMetric.reservoir.name+"_median", median)
+		m.AddValueFloat64(reservoirMetric.reservoir.name+"_min", min)
+		m.AddValueFloat64(reservoirMetric.reservoir.name+"_p10", p10)
+		m.AddValueFloat64(reservoirMetric.reservoir.name+"_p30", p30)
+		m.AddValueFloat64(reservoirMetric.reservoir.name+"_p70", p70)
+		m.AddValueFloat64(reservoirMetric.reservoir.name+"_p90", p90)
+		m.AddValueFloat64(reservoirMetric.reservoir.name+"_max", max)
+		m.AddValueUInt64(reservoirMetric.reservoir.name+"_count", count)
+		m.AddValueUInt64(reservoirMetric.reservoir.name+"_poolsize", poolSize)
+
+		result = append(result, m)
+	}
+	return result
+}
+
+type reservoirSampling struct {
+	samples         []float64
+	samplesLimit    uint64
+	name            string
+	countSinceReset uint64
+}
+
+func newReservoirSampling(name string, limit uint64) reservoirSampling {
+	return reservoirSampling{
+		samples:         make([]float64, 0, limit),
+		samplesLimit:    limit,
+		name:            name,
+		countSinceReset: 0,
+	}
+}
+
+// https://en.wikipedia.org/wiki/Reservoir_sampling
+func (this *reservoirSampling) AddPoint(val float64) {
+	this.countSinceReset += 1
+	if len(this.samples) < int(this.samplesLimit) {
+		this.samples = append(this.samples, val)
+	} else {
+		if j := rand.Int63n(int64(this.countSinceReset)); j < int64(len(this.samples)) {
+			this.samples[j] = val
+		}
+	}
+}
+
+func (this *reservoirSampling) indexForPercentile(percentile int) int {
+	return int(float64(len(this.samples)) * 0.01 * float64(percentile))
+}
+
+func (this *reservoirSampling) Reset() {
+	this.samples = this.samples[:0]
+	this.countSinceReset = 0
+}
+
+func (this *reservoirSampling) GetStats() (mean float64, median float64, min float64, percentile10 float64, percentile30 float64, percentile70 float64, percentile90 float64, max float64, count uint64, poolSize uint64) {
+	sort.Slice(this.samples, func(i, j int) bool { return this.samples[i] < this.samples[j] })
+
+	samplesCount := len(this.samples)
+	if samplesCount == 0 {
+		return 0, 0, 0, 0, 0, 0, 0, 0, 0, 0
+	}
+
+	var sum float64
+	for _, val := range this.samples {
+		sum += float64(val)
+	}
+
+	return sum / float64(samplesCount),
+		this.samples[this.indexForPercentile(50)],
+		this.samples[0],
+		this.samples[this.indexForPercentile(10)],
+		this.samples[this.indexForPercentile(30)],
+		this.samples[this.indexForPercentile(70)],
+		this.samples[this.indexForPercentile(90)],
+		this.samples[len(this.samples)-1],
+		this.countSinceReset,
+		uint64(len(this.samples))
+}