Use prometheus metrics implementations

Author: UnAfraid

config/metrics_mapping.go

@@ -6,18 +6,21 @@ import (
 
 type MetricsMapping struct {
 	// Counter, Gauge
-	Name        string   `yaml:"name"`
-	Description string   `yaml:"description"`
-	Namespace   string   `yaml:"namespace"`
-	Subsystem   string   `yaml:"subsystem"`
-	Labels      []string `yaml:"labels"`
+	Name        string            `yaml:"name"`
+	Description string            `yaml:"description"`
+	Namespace   string            `yaml:"namespace"`
+	Subsystem   string            `yaml:"subsystem"`
+	Labels      []string          `yaml:"labels"`
+	ConstLabels map[string]string `yaml:"constLabels"`
 	// Histogram
 	Buckets                         []float64     `yaml:"buckets"`
 	NativeHistogramBucketFactor     float64       `yaml:"nativeHistogramBucketFactor"`
 	NativeHistogramZeroThreshold    float64       `yaml:"nativeHistogramZeroThreshold"`
 	NativeHistogramMaxBucketNumber  uint32        `yaml:"nativeHistogramMaxBucketNumber"`
 	NativeHistogramMinResetDuration time.Duration `yaml:"nativeHistogramMinResetDuration"`
 	NativeHistogramMaxZeroThreshold float64       `yaml:"nativeHistogramMaxZeroThreshold"`
+	NativeHistogramMaxExemplars     int           `yaml:"nativeHistogramMaxExemplars"`
+	NativeHistogramExemplarTTL      time.Duration `yaml:"nativeHistogramExemplarTTL"`
 	// Summary
 	Objectives map[float64]float64 `yaml:"objectives"`
 	MaxAge     time.Duration       `yaml:"maxAge"`

input/transform/impl/bletomqtt/ble-to-mqtt.go

@@ -36,7 +36,10 @@ func (btm *bleToMqtt) Transform(_ context.Context, data *input.Data) (*input.Dat
 		return nil, fmt.Errorf("ble-to-mqtt transform: failed to json decode input event data: %w", err)
 	}
 
-	logrus.WithField("payload", inputEventData).Debug("input transformed")
+	logrus.
+		WithField("payload", inputEventData).
+		Debug("input transformed")
+
 	if serviceData, exists := inputEventData.ServiceData[btm.sdk]; exists {
 		payload, err := hex.DecodeString(serviceData)
 		if err != nil {

metrics/collector.go

@@ -1,85 +1,216 @@
 package metrics
 
 import (
+	"context"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/sirupsen/logrus"
 
 	"github.com/nikiforov-soft/yasp/config"
 	"github.com/nikiforov-soft/yasp/internal/syncx"
 )
 
 type collector struct {
-	metricsMapping *config.MetricsMapping
-	metrics        *syncx.Map[string, *metric]
+	closeCtx          context.Context
+	closeCtxCancel    context.CancelFunc
+	desc              *prometheus.Desc
+	metricVec         any
+	metrics           *syncx.Map[string, metricHistory]
+	metricsMapping    *config.MetricsMapping
+	stalenessInterval time.Duration
+}
+
+func newCollector(metricsMapping *config.MetricsMapping, stalenessInterval time.Duration) (*collector, error) {
+	metricVec, err := newMetricVec(metricsMapping)
+	if err != nil {
+		return nil, err
+	}
+
+	closeCtx, closeCtxCancel := context.WithCancel(context.Background())
+	c := &collector{
+		closeCtx:          closeCtx,
+		closeCtxCancel:    closeCtxCancel,
+		desc:              newDesc(metricsMapping),
+		metricVec:         metricVec,
+		metrics:           &syncx.Map[string, metricHistory]{},
+		metricsMapping:    metricsMapping,
+		stalenessInterval: stalenessInterval,
+	}
+
+	if err := prometheus.Register(c); err != nil {
+		return nil, fmt.Errorf("failed to register metrics collector: %w", err)
+	}
+
+	if stalenessInterval > 0 {
+		go c.runMetricsCleanupTask()
+	}
+
+	return c, nil
 }
 
 func (c *collector) Describe(ch chan<- *prometheus.Desc) {
-	ch <- c.buildDesc()
+	ch <- c.desc
 }
 
 func (c *collector) Collect(ch chan<- prometheus.Metric) {
-	c.metrics.Range(func(_ string, m *metric) bool {
-		metricValue, err := c.collectMetrics(m)
+	c.metrics.Range(func(_ string, m metricHistory) bool {
+		metricValue, err := c.getMetric(m.labels)
 		if err != nil {
-			ch <- prometheus.NewInvalidMetric(c.buildDesc(), err)
+			ch <- prometheus.NewInvalidMetric(c.desc, err)
 		} else {
-			ch <- prometheus.NewMetricWithTimestamp(m.timestamp, metricValue)
+			ch <- prometheus.NewMetricWithTimestamp(m.lastUpdatedAt, metricValue)
 		}
 		return true
 	})
 }
 
-func (c *collector) buildDesc() *prometheus.Desc {
-	return prometheus.NewDesc(
-		prometheus.BuildFQName(c.metricsMapping.Namespace, c.metricsMapping.Subsystem, c.metricsMapping.Name),
-		c.metricsMapping.Description,
-		c.metricsMapping.Labels,
-		nil,
-	)
+func (c *collector) Close() {
+	c.closeCtxCancel()
+	prometheus.Unregister(c)
+}
+
+func (c *collector) Observe(value float64, labels prometheus.Labels) error {
+	switch m := c.metricVec.(type) {
+	case *prometheus.CounterVec:
+		m.With(labels).Add(value)
+	case *prometheus.GaugeVec:
+		m.With(labels).Set(value)
+	case *prometheus.HistogramVec:
+		m.With(labels).Observe(value)
+	case *prometheus.SummaryVec:
+		m.With(labels).Observe(value)
+	default:
+		return fmt.Errorf("unknown metricVec type: %T", c.metricVec)
+	}
+
+	c.metrics.Store(computeHash(c.metricsMapping, labels), metricHistory{
+		labels:        labels,
+		lastUpdatedAt: time.Now(),
+	})
+
+	return nil
+}
+
+func (c *collector) getMetric(labels prometheus.Labels) (prometheus.Metric, error) {
+	switch m := c.metricVec.(type) {
+	case *prometheus.CounterVec:
+		return m.GetMetricWith(labels)
+	case *prometheus.GaugeVec:
+		return m.GetMetricWith(labels)
+	case *prometheus.HistogramVec:
+		return m.MetricVec.GetMetricWith(labels)
+	case *prometheus.SummaryVec:
+		return m.MetricVec.GetMetricWith(labels)
+	default:
+		return nil, fmt.Errorf("unsupported metricVec type: %T", m)
+	}
+}
+
+func (c *collector) runMetricsCleanupTask() {
+	ticker := time.NewTicker(time.Second)
+	for {
+		select {
+		case <-c.closeCtx.Done():
+			return
+		case <-ticker.C:
+			c.pruneStaleMetrics()
+		}
+	}
 }
 
-func (c *collector) collectMetrics(m *metric) (prometheus.Metric, error) {
-	desc := c.buildDesc()
-	labels := flattenLabels(c.metricsMapping.Labels, m.labels)
+func (c *collector) pruneStaleMetrics() {
+	var keysToDelete []string
+	c.metrics.Range(func(key string, value metricHistory) bool {
+		if time.Since(value.lastUpdatedAt) > c.stalenessInterval {
+			keysToDelete = append(keysToDelete, key)
+		}
+		return true
+	})
+
+	for _, key := range keysToDelete {
+		mh, ok := c.metrics.LoadAndDelete(key)
+		if !ok {
+			continue
+		}
+
+		logrus.
+			WithField("name", c.metricsMapping.Name).
+			WithField("labels", mh.labels).
+			WithField("updatedAt", mh.lastUpdatedAt).
+			Debug("stale metric removed")
 
-	switch strings.ToLower(c.metricsMapping.Type) {
+		switch m := c.metricVec.(type) {
+		case *prometheus.CounterVec:
+			m.Delete(mh.labels)
+		case *prometheus.GaugeVec:
+			m.Delete(mh.labels)
+		case *prometheus.HistogramVec:
+			m.Delete(mh.labels)
+		case *prometheus.SummaryVec:
+			m.Delete(mh.labels)
+		}
+	}
+}
+
+func newMetricVec(mapping *config.MetricsMapping) (any, error) {
+	switch strings.ToLower(mapping.Type) {
 	case "counter":
-		return prometheus.NewConstMetric(
-			desc,
-			prometheus.CounterValue,
-			m.value,
-			labels...,
-		)
+		return prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace:   mapping.Namespace,
+			Subsystem:   mapping.Subsystem,
+			Name:        mapping.Name,
+			ConstLabels: mapping.ConstLabels,
+			Help:        mapping.Description,
+		}, mapping.Labels), nil
 	case "gauge":
-		return prometheus.NewConstMetric(
-			desc,
-			prometheus.GaugeValue,
-			m.value,
-			labels...,
-		)
-	case "histogram":
-		buckets := make(map[float64]uint64)
-		for bucket, count := range m.histogramBuckets {
-			buckets[bucket] = uint64(count)
-		}
-		return prometheus.NewConstHistogram(
-			desc,
-			m.histogramCount,
-			m.histogramSum,
-			buckets,
-			labels...,
-		)
+		return prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace:   mapping.Namespace,
+			Subsystem:   mapping.Subsystem,
+			Name:        mapping.Name,
+			ConstLabels: mapping.ConstLabels,
+			Help:        mapping.Description,
+		}, mapping.Labels), nil
 	case "summary":
-		return prometheus.NewConstSummary(
-			desc,
-			m.summaryCount,
-			m.summarySum,
-			m.summaryQuantiles,
-			labels...,
-		)
+		return prometheus.NewSummaryVec(prometheus.SummaryOpts{
+			Namespace:   mapping.Namespace,
+			Subsystem:   mapping.Subsystem,
+			Name:        mapping.Name,
+			Help:        mapping.Description,
+			ConstLabels: mapping.ConstLabels,
+			Objectives:  mapping.Objectives,
+			MaxAge:      mapping.MaxAge,
+			AgeBuckets:  mapping.AgeBuckets,
+			BufCap:      mapping.BufCap,
+		}, mapping.Labels), nil
+	case "histogram":
+		return prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Namespace:                       mapping.Namespace,
+			Subsystem:                       mapping.Subsystem,
+			Name:                            mapping.Name,
+			Help:                            mapping.Description,
+			ConstLabels:                     mapping.ConstLabels,
+			Buckets:                         mapping.Buckets,
+			NativeHistogramBucketFactor:     mapping.NativeHistogramBucketFactor,
+			NativeHistogramZeroThreshold:    mapping.NativeHistogramZeroThreshold,
+			NativeHistogramMaxBucketNumber:  mapping.NativeHistogramMaxBucketNumber,
+			NativeHistogramMinResetDuration: mapping.NativeHistogramMinResetDuration,
+			NativeHistogramMaxZeroThreshold: mapping.NativeHistogramMaxZeroThreshold,
+			NativeHistogramMaxExemplars:     mapping.NativeHistogramMaxExemplars,
+			NativeHistogramExemplarTTL:      mapping.NativeHistogramExemplarTTL,
+		}, mapping.Labels), nil
 	default:
-		return nil, fmt.Errorf("unsupported metric type: %s", c.metricsMapping.Type)
+		return nil, fmt.Errorf("unsupported type: %s", mapping.Type)
 	}
 }
+
+func newDesc(mapping *config.MetricsMapping) *prometheus.Desc {
+	return prometheus.NewDesc(
+		prometheus.BuildFQName(mapping.Namespace, mapping.Subsystem, mapping.Name),
+		mapping.Description,
+		mapping.Labels,
+		mapping.ConstLabels,
+	)
+}

metrics/helper.go

@@ -3,6 +3,8 @@ package metrics
 import (
 	"crypto/sha256"
 	"encoding/hex"
+	"fmt"
+	"slices"
 
 	"github.com/prometheus/client_golang/prometheus"
 
@@ -17,14 +19,40 @@ func flattenLabels(keys []string, labels map[string]string) []string {
 	return result
 }
 
-func computeHash(key Key, mapping *config.MetricsMapping, labels prometheus.Labels) string {
+func computeHash(mapping *config.MetricsMapping, labels prometheus.Labels) string {
 	hash := sha256.New()
-	hash.Write([]byte(key.String()))
-	for i, v := range flattenLabels(mapping.Labels, labels) {
-		if i != 0 {
-			hash.Write([]byte("_"))
+	hash.Write([]byte(mapping.Namespace))
+	hash.Write([]byte(","))
+	hash.Write([]byte(mapping.Subsystem))
+	hash.Write([]byte(","))
+	hash.Write([]byte(mapping.Name))
+	if len(mapping.Labels) != 0 {
+		hash.Write([]byte(","))
+		for i, v := range flattenLabels(mapping.Labels, labels) {
+			if i != 0 {
+				hash.Write([]byte("_"))
+			}
+			hash.Write([]byte(v))
 		}
-		hash.Write([]byte(v))
 	}
 	return hex.EncodeToString(hash.Sum(nil))
 }
+
+func validateLabels(key Key, mapping *config.MetricsMapping, labels prometheus.Labels) error {
+	if len(labels) != len(mapping.Labels) {
+		return fmt.Errorf("mismatched label name/value count for %s expected %d got %d", key, len(mapping.Labels), len(labels))
+	}
+
+	for _, k := range mapping.Labels {
+		if _, exists := labels[k]; !exists {
+			return fmt.Errorf("missing required label %s for: %s", k, key)
+		}
+	}
+
+	for k := range labels {
+		if !slices.Contains(mapping.Labels, k) {
+			return fmt.Errorf("provided unknown label %s for: %s", k, key)
+		}
+	}
+	return nil
+}