[*] exclude MetricDef from MeasurementEnvelope (#787)

Sink doesn't need to know all information about metric so passing
the whole definition is an overkill. This will decrease a memory
consumption. Instead implement `MetricsDefiner` interface for sinks
that needs metric specific information, e.g. gauges for Prometheus

Author: pashagolub

internal/metrics/types.go

@@ -152,7 +152,6 @@ type MeasurementEnvelope struct {
 	MetricName       string
 	CustomTags       map[string]string
 	Data             Measurements
-	MetricDef        Metric
 	RealDbname       string
 	SystemIdentifier string
 }

internal/reaper/file.go

@@ -66,14 +66,12 @@ func (r *Reaper) FetchStatsDirectlyFromOS(ctx context.Context, md *sources.Sourc
 	if err != nil {
 		return nil, err
 	}
-	m, _ := metricDefs.GetMetricDef(metricName)
 	return &metrics.MeasurementEnvelope{
 		DBName:           md.Name,
 		SourceType:       string(md.Kind),
 		MetricName:       metricName,
 		CustomTags:       md.CustomTags,
 		Data:             data,
-		MetricDef:        m,
 		RealDbname:       md.RealDbname,
 		SystemIdentifier: md.SystemIdentifier,
 	}, nil

internal/reaper/metric.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/metrics"
+	"github.com/cybertec-postgresql/pgwatch/v3/internal/sinks"
 	"github.com/sirupsen/logrus"
 )
 
@@ -77,6 +78,9 @@ func (r *Reaper) LoadMetrics() (err error) {
 		return
 	}
 	metricDefs.Assign(newDefs)
+	if definer, ok := r.SinksWriter.(sinks.MetricsDefiner); ok {
+		err = definer.DefineMetrics(newDefs)
+	}
 	r.logger.
 		WithField("metrics", len(newDefs.MetricDefs)).
 		WithField("presets", len(newDefs.PresetDefs)).

internal/reaper/reaper.go

@@ -500,7 +500,7 @@ func (r *Reaper) FetchMetric(ctx context.Context, md *sources.SourceConn, metric
 				return nil, err
 			}
 			if metricName == specialMetricInstanceUp {
-				l.WithError(err).Debugf("[%s:%s] failed to fetch metrics. marking instance as not up", md.Name, metricName)
+				l.WithError(err).Debugf("failed to fetch metrics. marking instance as not up")
 				data = make(metrics.Measurements, 1)
 				data[0] = metrics.NewMeasurement(time.Now().UnixNano())
 				data[0]["is_up"] = 0 // should be updated if the "instance_up" metric definition is changed
@@ -521,7 +521,6 @@ send_to_storageChannel:
 		MetricName:       cmp.Or(metric.StorageName, metricName),
 		Data:             data,
 		CustomTags:       md.CustomTags,
-		MetricDef:        metric,
 		RealDbname:       md.RealDbname,
 		SystemIdentifier: md.SystemIdentifier}, nil
 }

internal/sinks/multiwriter.go

@@ -16,6 +16,11 @@ type Writer interface {
 	Write(msgs metrics.MeasurementEnvelope) error
 }
 
+// MetricDefiner is an interface for passing metric definitions to a sink.
+type MetricsDefiner interface {
+	DefineMetrics(metric *metrics.Metrics) error
+}
+
 // MultiWriter ensures the simultaneous storage of data in several storages.
 type MultiWriter struct {
 	writers []Writer
@@ -62,6 +67,15 @@ func (mw *MultiWriter) AddWriter(w Writer) {
 	mw.Unlock()
 }
 
+func (mw *MultiWriter) DefineMetrics(metrics *metrics.Metrics) (err error) {
+	for _, w := range mw.writers {
+		if definer, ok := w.(MetricsDefiner); ok {
+			err = errors.Join(err, definer.DefineMetrics(metrics))
+		}
+	}
+	return nil
+}
+
 func (mw *MultiWriter) SyncMetric(dbUnique, metricName string, op SyncOp) (err error) {
 	for _, w := range mw.writers {
 		err = errors.Join(err, w.SyncMetric(dbUnique, metricName, op))

internal/sinks/prometheus.go

@@ -24,6 +24,7 @@ type PrometheusWriter struct {
 	lastScrapeErrors                  prometheus.Gauge
 	totalScrapes, totalScrapeFailures prometheus.Counter
 	PrometheusNamespace               string
+	gauges                            map[string]([]string) // map of metric names to their gauge names, used for Prometheus gauge metrics
 }
 
 const promInstanceUpStateMetric = "instance_up"
@@ -77,6 +78,14 @@ func NewPrometheusWriter(ctx context.Context, connstr string) (promw *Prometheus
 	return
 }
 
+func (promw *PrometheusWriter) DefineMetrics(metrics *metrics.Metrics) (err error) {
+	promw.gauges = make(map[string]([]string))
+	for name, m := range metrics.MetricDefs {
+		promw.gauges[name] = m.Gauges
+	}
+	return nil
+}
+
 func (promw *PrometheusWriter) Write(msg metrics.MeasurementEnvelope) error {
 	if len(msg.Data) == 0 { // no batching in async prom mode, so using 0 indexing ok
 		return nil
@@ -194,6 +203,8 @@ func (promw *PrometheusWriter) MetricStoreMessageToPromMetrics(msg metrics.Measu
 		return promMetrics
 	}
 
+	gauges := promw.gauges[msg.MetricName]
+
 	epochTime = time.Unix(0, msg.Data.GetEpoch())
 
 	if epochTime.Before(time.Now().Add(-promScrapingStalenessHardDropLimit)) {
@@ -247,8 +258,7 @@ func (promw *PrometheusWriter) MetricStoreMessageToPromMetrics(msg metrics.Measu
 		for field, value := range fields {
 			fieldPromDataType := prometheus.CounterValue
 			if msg.MetricName == promInstanceUpStateMetric ||
-				len(msg.MetricDef.Gauges) > 0 &&
-					(msg.MetricDef.Gauges[0] == "*" || slices.Contains(msg.MetricDef.Gauges, field)) {
+				len(gauges) > 0 && (gauges[0] == "*" || slices.Contains(gauges, field)) {
 				fieldPromDataType = prometheus.GaugeValue
 			}
 			var desc *prometheus.Desc