prometheus.go

Documentation: github.com/cybertec-postgresql/pgwatch/v6/internal/sinks

     1  package sinks
     2  
     3  import (
     4  	"context"
     5  	"fmt"
     6  	"maps"
     7  	"net"
     8  	"net/http"
     9  	"slices"
    10  	"strings"
    11  	"sync"
    12  	"time"
    13  
    14  	"github.com/cybertec-postgresql/pgwatch/v6/internal/log"
    15  	"github.com/cybertec-postgresql/pgwatch/v6/internal/metrics"
    16  	"github.com/prometheus/client_golang/prometheus"
    17  	"github.com/prometheus/client_golang/prometheus/promhttp"
    18  	"github.com/prometheus/common/model"
    19  )
    20  
    21  type PromMetricCache = map[string]map[string]metrics.MeasurementEnvelope // [dbUnique][metric]lastly_fetched_data
    22  
    23  // PrometheusWriter is a Prometheus exporter that implements the prometheus.Collector
    24  // interface using the "unchecked collector" pattern (empty Describe method).
    25  //
    26  // Design decisions based on Prometheus exporter guidelines
    27  // (https://prometheus.io/docs/instrumenting/writing_exporters/#collectors):
    28  //
    29  //   - Metrics are collected periodically by reaper and cached in-memory.
    30  //     On scrape, the collector reads a snapshot of the cache
    31  //     and emits fresh NewConstMetric values. The cache is NOT consumed on
    32  //     scrape, so parallel or back-to-back scrapes see the same data until the
    33  //     next Write() updates arrive.
    34  //
    35  //   - This is an "unchecked collector": Describe() sends no descriptors, which
    36  //     tells the Prometheus registry to skip consistency checks. This is necessary
    37  //     because the set of metrics is dynamic (driven by monitored databases and
    38  //     their query results). Safety is ensured by deduplicating metric identities
    39  //     within each Collect() call.
    40  //
    41  //   - Label keys are always sorted lexicographically before building descriptors
    42  //     and label value slices. This guarantees deterministic descriptor identity
    43  //     regardless of Go map iteration order.
    44  type PrometheusWriter struct {
    45  	sync.RWMutex
    46  	logger    log.Logger
    47  	ctx       context.Context
    48  	gauges    map[string]([]string) // map of metric names to their gauge column names
    49  	Namespace string
    50  	Cache     PromMetricCache // [dbUnique][metric]lastly_fetched_data
    51  
    52  	// Self-instrumentation metrics
    53  	lastScrapeErrors    prometheus.Gauge
    54  	totalScrapes        prometheus.Counter
    55  	totalScrapeFailures prometheus.Counter
    56  }
    57  
    58  const promInstanceUpStateMetric = "instance_up"
    59  
    60  // timestamps older than that will be ignored on the Prom scraper side anyway, so better don't emit at all and just log a notice
    61  const promCacheTTL = time.Minute * time.Duration(10)
    62  
    63  func NewPrometheusWriter(ctx context.Context, connstr string) (promw *PrometheusWriter, err error) {
    64  	addr, namespace, found := strings.Cut(connstr, "/")
    65  	if !found || namespace == "" {
    66  		namespace = "pgwatch"
    67  	}
    68  	l := log.GetLogger(ctx).WithField("sink", "prometheus").WithField("address", addr)
    69  	ctx = log.WithLogger(ctx, l)
    70  
    71  	promw = &PrometheusWriter{
    72  		ctx:       ctx,
    73  		logger:    l,
    74  		Namespace: namespace,
    75  		Cache:     make(PromMetricCache),
    76  		lastScrapeErrors: prometheus.NewGauge(prometheus.GaugeOpts{
    77  			Namespace: namespace,
    78  			Name:      "exporter_last_scrape_errors",
    79  			Help:      "Last scrape error count for all monitored hosts / metrics",
    80  		}),
    81  		totalScrapes: prometheus.NewCounter(prometheus.CounterOpts{
    82  			Namespace: namespace,
    83  			Name:      "exporter_total_scrapes",
    84  			Help:      "Total scrape attempts.",
    85  		}),
    86  		totalScrapeFailures: prometheus.NewCounter(prometheus.CounterOpts{
    87  			Namespace: namespace,
    88  			Name:      "exporter_total_scrape_failures",
    89  			Help:      "Number of errors while executing metric queries",
    90  		}),
    91  	}
    92  
    93  	if err = prometheus.Register(promw); err != nil {
    94  		return
    95  	}
    96  
    97  	promServer := &http.Server{
    98  		Addr: addr,
    99  		Handler: promhttp.HandlerFor(
   100  			prometheus.DefaultGatherer,
   101  			promhttp.HandlerOpts{
   102  				ErrorLog:      promw,
   103  				ErrorHandling: promhttp.ContinueOnError,
   104  			},
   105  		),
   106  	}
   107  
   108  	ln, err := net.Listen("tcp", promServer.Addr)
   109  	if err != nil {
   110  		return nil, err
   111  	}
   112  
   113  	go func() { log.GetLogger(ctx).Error(promServer.Serve(ln)) }()
   114  
   115  	l.Info(`measurements sink is activated`)
   116  	return
   117  }
   118  
   119  // Println implements promhttp.Logger
   120  func (promw *PrometheusWriter) Println(v ...any) {
   121  	promw.logger.Errorln(v...)
   122  }
   123  
   124  // DefineMetrics is called by reaper on startup and whenever metric definitions change
   125  func (promw *PrometheusWriter) DefineMetrics(metrics *metrics.Metrics) (err error) {
   126  	promw.Lock()
   127  	defer promw.Unlock()
   128  	promw.gauges = make(map[string]([]string))
   129  	for name, m := range metrics.MetricDefs {
   130  		promw.gauges[name] = m.Gauges
   131  	}
   132  	return nil
   133  }
   134  
   135  // Write is called by reaper whenever new measurement data arrives
   136  func (promw *PrometheusWriter) Write(msg metrics.MeasurementEnvelope) error {
   137  	if len(msg.Data) == 0 {
   138  		return nil
   139  	}
   140  	promw.AddCacheEntry(msg.DBName, msg.MetricName, msg)
   141  	return nil
   142  }
   143  
   144  // SyncMetric is called by reaper when a metric or monitored source is removed or added,
   145  // allowing the writer to purge or initialize cache entries as needed
   146  func (promw *PrometheusWriter) SyncMetric(sourceName, metricName string, op SyncOp) error {
   147  	switch op {
   148  	case DeleteOp:
   149  		promw.PurgeCacheEntry(sourceName, metricName)
   150  	case AddOp:
   151  		promw.InitCacheEntry(sourceName)
   152  	}
   153  	return nil
   154  }
   155  
   156  var notSupportedMetrics = map[string]struct{}{
   157  	"change_events":     {}, // fully consist of text columns
   158  	"pgbouncer_stats":   {}, // this and below metrics column names cannot be renamed with tag_ prefix
   159  	"pgbouncer_clients": {},
   160  	"pgpool_processes":  {},
   161  	"pgpool_stats":      {},
   162  }
   163  
   164  func (promw *PrometheusWriter) AddCacheEntry(dbUnique, metric string, msgArr metrics.MeasurementEnvelope) { // cache structure: [dbUnique][metric]lastly_fetched_data
   165  	if _, ok := notSupportedMetrics[metric]; ok && msgArr.SourceKind != "prometheus" {
   166  		return // not supported for DB-sourced metrics
   167  	}
   168  	promw.Lock()
   169  	defer promw.Unlock()
   170  	if _, ok := promw.Cache[dbUnique]; !ok {
   171  		promw.Cache[dbUnique] = make(map[string]metrics.MeasurementEnvelope)
   172  	}
   173  	promw.Cache[dbUnique][metric] = msgArr
   174  }
   175  
   176  func (promw *PrometheusWriter) InitCacheEntry(dbUnique string) {
   177  	promw.Lock()
   178  	defer promw.Unlock()
   179  	if _, ok := promw.Cache[dbUnique]; !ok {
   180  		promw.Cache[dbUnique] = make(map[string]metrics.MeasurementEnvelope)
   181  	}
   182  }
   183  
   184  func (promw *PrometheusWriter) PurgeCacheEntry(dbUnique, metric string) {
   185  	promw.Lock()
   186  	defer promw.Unlock()
   187  	if metric == "" {
   188  		delete(promw.Cache, dbUnique) // whole host removed from config
   189  		return
   190  	}
   191  	delete(promw.Cache[dbUnique], metric)
   192  }
   193  
   194  // Describe is intentionally empty to make PrometheusWriter an "unchecked
   195  // collector" per the prometheus.Collector contract
   196  func (promw *PrometheusWriter) Describe(_ chan<- *prometheus.Desc) {
   197  }
   198  
   199  // Collect implements prometheus.Collector. It reads a snapshot of the metric
   200  // cache and emits const metrics. Parallel scrapes see the same data until
   201  // background Write() calls update it
   202  func (promw *PrometheusWriter) Collect(ch chan<- prometheus.Metric) {
   203  	promw.totalScrapes.Add(1)
   204  	ch <- promw.totalScrapes
   205  
   206  	promw.RLock()
   207  	if len(promw.Cache) == 0 {
   208  		promw.RUnlock()
   209  		promw.logger.Warning("No dbs configured for monitoring. Check config")
   210  		ch <- promw.totalScrapeFailures
   211  		promw.lastScrapeErrors.Set(0)
   212  		ch <- promw.lastScrapeErrors
   213  		return
   214  	}
   215  	snapshot := promw.snapshotCache()
   216  	promw.RUnlock()
   217  
   218  	var rows int
   219  	var lastScrapeErrors float64
   220  
   221  	t1 := time.Now()
   222  	for _, metricsMessages := range snapshot {
   223  		for _, envelope := range metricsMessages {
   224  			written, errors := promw.WritePromMetrics(envelope, ch)
   225  			lastScrapeErrors += float64(errors)
   226  			rows += written
   227  		}
   228  	}
   229  	promw.logger.WithField("count", rows).WithField("elapsed", time.Since(t1)).Info("measurements written")
   230  	ch <- promw.totalScrapeFailures
   231  	promw.lastScrapeErrors.Set(lastScrapeErrors)
   232  	ch <- promw.lastScrapeErrors
   233  }
   234  
   235  // snapshotCache creates a shallow copy of the cache map hierarchy.
   236  // Must be called under at least promw.RLock().
   237  // The MeasurementEnvelope values are not deep-copied because writers always
   238  // replace entire envelopes (never mutate them in place).
   239  func (promw *PrometheusWriter) snapshotCache() PromMetricCache {
   240  	snapshot := make(PromMetricCache, len(promw.Cache))
   241  	for db, metricMap := range promw.Cache {
   242  		snapshot[db] = maps.Clone(metricMap)
   243  	}
   244  	return snapshot
   245  }
   246  
   247  // WritePromMetrics converts a MeasurementEnvelope into Prometheus const metrics
   248  // and sends them directly to ch. Returns the count of metrics written and errors encountered.
   249  //
   250  // For prom-sourced envelopes (SourceKind == "prometheus") the original family
   251  // name is used as the fully-qualified metric name with no pgwatch namespace
   252  // prefix, and each measurement's own epoch_ns is used as the timestamp.
   253  // For all other sources the existing namespace+family+field naming and a
   254  // single envelope-level timestamp are applied.
   255  func (promw *PrometheusWriter) WritePromMetrics(msg metrics.MeasurementEnvelope, ch chan<- prometheus.Metric) (written int, errorCount int) {
   256  	if len(msg.Data) == 0 {
   257  		return
   258  	}
   259  
   260  	isPromSource := msg.SourceKind == "prometheus"
   261  
   262  	promw.RLock()
   263  	gauges := promw.gauges[msg.MetricName]
   264  	promw.RUnlock()
   265  
   266  	// baseEpochTime is used for staleness check and as the per-row timestamp
   267  	// for non-prom sources (all rows in one envelope share the same instant).
   268  	baseEpochTime := time.Unix(0, msg.Data.GetEpoch())
   269  	if baseEpochTime.Before(time.Now().Add(-promCacheTTL)) {
   270  		promw.logger.Debugf("Dropping metric %s:%s cache set due to staleness (>%v)...", msg.DBName, msg.MetricName, promCacheTTL)
   271  		promw.PurgeCacheEntry(msg.DBName, msg.MetricName)
   272  		return
   273  	}
   274  
   275  	seen := make(map[string]any)
   276  
   277  	// fqName and the value type depend only on the field name, and a Desc
   278  	// only on (fqName, help, label keys); both are therefore memoized per
   279  	// envelope instead of being rebuilt for every row.
   280  	type fieldMeta struct {
   281  		fqName    string
   282  		valueType prometheus.ValueType
   283  	}
   284  	metaByField := make(map[string]fieldMeta)
   285  	type descKey struct {
   286  		fqName    string
   287  		labelKeys string
   288  	}
   289  	descs := make(map[descKey]*prometheus.Desc)
   290  	fqNamePrefix := promw.Namespace + "_" + msg.MetricName + "_"
   291  
   292  	for _, measurement := range msg.Data {
   293  		var labels map[string]string
   294  		if msg.CustomTags != nil {
   295  			labels = maps.Clone(msg.CustomTags)
   296  		} else {
   297  			labels = make(map[string]string)
   298  		}
   299  		labels["dbname"] = msg.DBName
   300  		fields := make(map[string]float64)
   301  
   302  		// Prom-sourced measurements each carry their own timestamp.
   303  		rowEpochTime := baseEpochTime
   304  		for k, v := range measurement {
   305  			if k == metrics.EpochColumnName {
   306  				if isPromSource {
   307  					if ns, ok := v.(int64); ok && ns != 0 {
   308  						rowEpochTime = time.Unix(0, ns)
   309  					}
   310  				}
   311  				continue
   312  			}
   313  			if v == nil || v == "" {
   314  				continue
   315  			}
   316  
   317  			if tag, found := strings.CutPrefix(k, metrics.TagPrefix); found {
   318  				labels[tag] = fmt.Sprintf("%v", v)
   319  				continue
   320  			}
   321  			switch t := v.(type) {
   322  			case int:
   323  				fields[k] = float64(t)
   324  			case int32:
   325  				fields[k] = float64(t)
   326  			case int64:
   327  				fields[k] = float64(t)
   328  			case float32:
   329  				fields[k] = float64(t)
   330  			case float64:
   331  				fields[k] = t
   332  			case bool:
   333  				if t {
   334  					fields[k] = 1
   335  				} else {
   336  					fields[k] = 0
   337  				}
   338  			default:
   339  				promw.logger.Debugf("skipping scraping column %s of [%s:%s], unsupported datatype: %v", k, msg.DBName, msg.MetricName, t)
   340  			}
   341  		}
   342  
   343  		// Sort label keys for deterministic descriptor identity.
   344  		labelKeys := slices.Sorted(maps.Keys(labels))
   345  		labelValues := make([]string, len(labelKeys))
   346  		for i, k := range labelKeys {
   347  			labelValues[i] = labels[k]
   348  		}
   349  		joinedLabelValues := strings.Join(labelValues, "_")
   350  		// model.SeparatorByte cannot occur in label names.
   351  		joinedLabelKeys := strings.Join(labelKeys, string(model.SeparatorByte))
   352  
   353  		for field, value := range fields {
   354  			meta, ok := metaByField[field]
   355  			if !ok {
   356  				if isPromSource {
   357  					// Prom→prom proxy path: ScrapeAll stores exactly one value
   358  					// column per family, named after the family itself. Skip any
   359  					// other numeric column that may appear unexpectedly.
   360  					if field != msg.MetricName {
   361  						continue
   362  					}
   363  					meta = fieldMeta{fqName: field, valueType: prometheus.UntypedValue}
   364  				} else {
   365  					meta.valueType = prometheus.CounterValue
   366  					if msg.MetricName == promInstanceUpStateMetric ||
   367  						len(gauges) > 0 && (gauges[0] == "*" || slices.Contains(gauges, field)) {
   368  						meta.valueType = prometheus.GaugeValue
   369  					}
   370  					if msg.MetricName == promInstanceUpStateMetric {
   371  						meta.fqName = promw.Namespace + "_" + msg.MetricName
   372  					} else {
   373  						meta.fqName = fqNamePrefix + field
   374  					}
   375  				}
   376  				metaByField[field] = meta
   377  			}
   378  
   379  			// skip if this exact identity was already emitted in this scrape
   380  			identity := meta.fqName + "_" + joinedLabelValues
   381  			if _, dup := seen[identity]; dup {
   382  				promw.logger.
   383  					WithField("metric", msg.MetricName).
   384  					Warning("duplicate metric identity dropped, prefix differentiating string columns with tag_")
   385  				errorCount++
   386  				continue
   387  			}
   388  			seen[identity] = struct{}{}
   389  
   390  			dk := descKey{meta.fqName, joinedLabelKeys}
   391  			desc, ok := descs[dk]
   392  			if !ok {
   393  				desc = prometheus.NewDesc(meta.fqName, msg.MetricName, labelKeys, nil)
   394  				descs[dk] = desc
   395  			}
   396  			m, err := prometheus.NewConstMetric(desc, meta.valueType, value, labelValues...)
   397  			if err != nil {
   398  				promw.logger.Warningf("skipping metric %s of [%s:%s]: %v", meta.fqName, msg.DBName, msg.MetricName, err)
   399  				errorCount++
   400  				continue
   401  			}
   402  			ch <- prometheus.NewMetricWithTimestamp(rowEpochTime, m)
   403  			written++
   404  		}
   405  	}
   406  	return
   407  }
   408