prometheus.go

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

     1  package reaper
     2  
     3  import (
     4  	"context"
     5  	"errors"
     6  	"fmt"
     7  	"io"
     8  	"time"
     9  
    10  	dto "github.com/prometheus/client_model/go"
    11  	"github.com/prometheus/common/expfmt"
    12  
    13  	"github.com/cybertec-postgresql/pgwatch/v6/internal/log"
    14  	"github.com/cybertec-postgresql/pgwatch/v6/internal/metrics"
    15  	"github.com/cybertec-postgresql/pgwatch/v6/internal/sources"
    16  )
    17  
    18  const defaultScrapeInterval = 60 * time.Second
    19  
    20  var _ Reaper = (*PromReaper)(nil)
    21  
    22  // PromReaper drives metric scraping for a single Prometheus source.
    23  // It runs a GCD-based tick loop and applies per-family emit-interval gating.
    24  type PromReaper struct {
    25  	reaper      *reaper
    26  	md          *sources.PromConn
    27  	lastEmitted map[string]time.Time
    28  }
    29  
    30  // NewPromSourceReaper creates a PromReaper for the given Prometheus source.
    31  func NewPromSourceReaper(r *reaper, md *sources.PromConn) *PromReaper {
    32  	return &PromReaper{
    33  		reaper:      r,
    34  		md:          md,
    35  		lastEmitted: make(map[string]time.Time),
    36  	}
    37  }
    38  
    39  // calcScrapeInterval returns the GCD of all configured metric intervals.
    40  // Defaults to defaultScrapeInterval when no metrics are configured (scrape-all mode).
    41  // Individual intervals below minTickInterval are floored to minTickInterval.
    42  func (pr *PromReaper) calcScrapeInterval() time.Duration {
    43  	pr.md.RLock()
    44  	m := pr.md.Metrics
    45  	pr.md.RUnlock()
    46  
    47  	if len(m) == 0 {
    48  		return defaultScrapeInterval
    49  	}
    50  	intervals := make([]int, 0, len(m))
    51  	for _, v := range m {
    52  		intervals = append(intervals, max(v, minTickInterval))
    53  	}
    54  	return time.Duration(max(GCDSlice(intervals), minTickInterval)) * time.Second
    55  }
    56  
    57  // Reap is the main loop for a Prometheus source. It scrapes all metric families
    58  // on every GCD tick and emits envelopes that have passed their per-family interval.
    59  func (pr *PromReaper) Reap(ctx context.Context) {
    60  	l := log.GetLogger(ctx).WithField("source", pr.md.Name)
    61  	ctx = log.WithLogger(ctx, l)
    62  
    63  	pr.md.RLock()
    64  	scrapeAll := len(pr.md.Metrics) == 0
    65  	pr.md.RUnlock()
    66  
    67  	if scrapeAll {
    68  		l.Warning("no metrics configured for prometheus source, using scrape-all mode with 60 s interval")
    69  	}
    70  
    71  	for {
    72  		envelopes, err := pr.ScrapeAll(ctx)
    73  		if err != nil {
    74  			l.WithError(err).Warning("prometheus scrape failed")
    75  		} else {
    76  			now := time.Now()
    77  			for _, env := range envelopes {
    78  				if !scrapeAll {
    79  					emitInterval := pr.md.GetMetricInterval(env.MetricName)
    80  					if emitInterval > 0 {
    81  						if last := pr.lastEmitted[env.MetricName]; !last.IsZero() && now.Sub(last) < emitInterval {
    82  							continue
    83  						}
    84  					}
    85  				}
    86  				env.DBName = pr.md.Name
    87  				env.CustomTags = pr.md.CustomTags
    88  				env.SourceKind = string(sources.SourcePrometheus)
    89  				pr.reaper.measurementCh <- env
    90  				pr.lastEmitted[env.MetricName] = now
    91  			}
    92  		}
    93  
    94  		select {
    95  		case <-ctx.Done():
    96  			return
    97  		case <-time.After(pr.calcScrapeInterval()):
    98  		}
    99  	}
   100  }
   101  
   102  // ScrapeAll fetches Prometheus exposition metrics from pr.md and returns one
   103  // MeasurementEnvelope per metric family. Each sample becomes one Measurement
   104  // with tag_<label> columns (skipping __name__), a value column named after the
   105  // family, and epoch_ns set from the sample timestamp (ms→ns) or time.Now().
   106  func (pr *PromReaper) ScrapeAll(ctx context.Context) ([]metrics.MeasurementEnvelope, error) {
   107  	resp, err := pr.md.Scrape(ctx)
   108  	if err != nil {
   109  		return nil, err
   110  	}
   111  	defer resp.Body.Close()
   112  
   113  	if resp.StatusCode >= 300 {
   114  		return nil, fmt.Errorf("scrapeall: unexpected status %s", resp.Status)
   115  	}
   116  
   117  	contentType := expfmt.ResponseFormat(resp.Header)
   118  	decoder := expfmt.NewDecoder(resp.Body, contentType)
   119  
   120  	var result []metrics.MeasurementEnvelope
   121  	for {
   122  		var mf dto.MetricFamily
   123  		if err := decoder.Decode(&mf); err != nil {
   124  			if errors.Is(err, io.EOF) {
   125  				break
   126  			}
   127  			return nil, fmt.Errorf("scrapeall: decoding: %w", err)
   128  		}
   129  
   130  		familyName := mf.GetName()
   131  		samples := make(metrics.Measurements, 0, len(mf.GetMetric()))
   132  		for _, m := range mf.GetMetric() {
   133  			measurement := make(metrics.Measurement)
   134  
   135  			for _, lp := range m.GetLabel() {
   136  				if lp.GetName() != "__name__" {
   137  					measurement[metrics.TagPrefix+lp.GetName()] = lp.GetValue()
   138  				}
   139  			}
   140  
   141  			measurement[familyName] = promMetricValue(m, mf.GetType())
   142  
   143  			if ts := m.GetTimestampMs(); ts != 0 {
   144  				measurement[metrics.EpochColumnName] = ts * 1_000_000
   145  			} else {
   146  				measurement[metrics.EpochColumnName] = time.Now().UnixNano()
   147  			}
   148  
   149  			samples = append(samples, measurement)
   150  		}
   151  
   152  		result = append(result, metrics.MeasurementEnvelope{
   153  			MetricName: familyName,
   154  			SourceKind: string(pr.md.Kind),
   155  			Data:       samples,
   156  		})
   157  	}
   158  	return result, nil
   159  }
   160  
   161  // promMetricValue extracts the primary float64 sample value from a metric
   162  // based on its family type.
   163  func promMetricValue(m *dto.Metric, mtype dto.MetricType) float64 {
   164  	switch mtype {
   165  	case dto.MetricType_GAUGE:
   166  		return m.GetGauge().GetValue()
   167  	case dto.MetricType_COUNTER:
   168  		return m.GetCounter().GetValue()
   169  	case dto.MetricType_HISTOGRAM:
   170  		return float64(m.GetHistogram().GetSampleCount())
   171  	case dto.MetricType_SUMMARY:
   172  		return m.GetSummary().GetSampleSum()
   173  	default:
   174  		return m.GetUntyped().GetValue()
   175  	}
   176  }
   177