mqtt2prometheus/pkg/metrics/collector.go

package metrics

import (
	"fmt"
	"github.com/hikhvar/mqtt2prometheus/pkg/config"
	gocache "github.com/patrickmn/go-cache"
	"github.com/prometheus/client_golang/prometheus"
	"go.uber.org/zap"
	"time"
)

type Collector interface {
	prometheus.Collector
	Observe(deviceID string, collection MetricCollection)
}

type MemoryCachedCollector struct {
	cache        *gocache.Cache
	descriptions []*prometheus.Desc
	logger       *zap.Logger
}

type Metric struct {
	Description *prometheus.Desc
	Value       float64
	ValueType   prometheus.ValueType
	IngestTime  time.Time
	Topic       string
}

type CacheItem struct {
	DeviceID string
	Metric   Metric
}

type MetricCollection []Metric

func NewCollector(defaultTimeout time.Duration, possibleMetrics []config.MetricConfig, logger *zap.Logger) Collector {
	var descs []*prometheus.Desc
	for _, m := range possibleMetrics {
		descs = append(descs, m.PrometheusDescription())
	}
	return &MemoryCachedCollector{
		cache:        gocache.New(defaultTimeout, defaultTimeout*10),
		descriptions: descs,
		logger:       logger,
	}
}

func (c *MemoryCachedCollector) Observe(deviceID string, collection MetricCollection) {
	for _, m := range collection {
		item := CacheItem{
			DeviceID: deviceID,
			Metric:   m,
		}
		c.cache.Set(fmt.Sprintf("%s-%s", deviceID, m.Description.String()), item, gocache.DefaultExpiration)
	}
}

func (c *MemoryCachedCollector) Describe(ch chan<- *prometheus.Desc) {
	for i := range c.descriptions {
		ch <- c.descriptions[i]
	}
}

func (c *MemoryCachedCollector) Collect(mc chan<- prometheus.Metric) {
	for _, metricsRaw := range c.cache.Items() {
		item := metricsRaw.Object.(CacheItem)
		device, metric := item.DeviceID, item.Metric
		if metric.Description == nil {
			c.logger.Warn("empty description", zap.String("topic", metric.Topic), zap.Float64("value", metric.Value))
		}
		m := prometheus.MustNewConstMetric(
			metric.Description,
			metric.ValueType,
			metric.Value,
			device,
			metric.Topic,
		)
		mc <- prometheus.NewMetricWithTimestamp(metric.IngestTime, m)

	}
}