source_reaper.go

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

     1  package reaper
     2  
     3  import (
     4  	"cmp"
     5  	"context"
     6  	"fmt"
     7  	"time"
     8  
     9  	"github.com/cybertec-postgresql/pgwatch/v5/internal/log"
    10  	"github.com/cybertec-postgresql/pgwatch/v5/internal/metrics"
    11  	"github.com/cybertec-postgresql/pgwatch/v5/internal/sources"
    12  	"github.com/jackc/pgx/v5"
    13  )
    14  
    15  const minTickInterval = 1 // seconds - floor for GCD to help handle zero/negative intervals
    16  
    17  // SourceReaper manages metric collection for a single monitored source.
    18  // Instead of one goroutine per metric it runs a single GCD-based tick loop
    19  // and batches SQL queries via pgx.Batch when the source is a real Postgres
    20  // connection (non-pgbouncer, non-pgpool).
    21  type SourceReaper struct {
    22  	reaper          *Reaper
    23  	md              *sources.SourceConn
    24  	lastFetch       map[string]time.Time
    25  	lastUptimeS     int64               // last seen postmaster_uptime_s for restart detection
    26  	degradedMetrics map[string]struct{} // metrics that failed individual retry; executed via fetchMetric until they recover
    27  }
    28  
    29  // NewSourceReaper creates a SourceReaper for the given source connection.
    30  func NewSourceReaper(r *Reaper, md *sources.SourceConn) *SourceReaper {
    31  	sr := &SourceReaper{
    32  		reaper:          r,
    33  		md:              md,
    34  		lastFetch:       make(map[string]time.Time),
    35  		degradedMetrics: make(map[string]struct{}),
    36  	}
    37  	return sr
    38  }
    39  
    40  // activeMetrics returns a snapshot copy of the currently active metric intervals
    41  // based on the source's recovery state. Copying under the lock prevents data
    42  // races when the caller iterates after the lock is released.
    43  func (sr *SourceReaper) activeMetrics() map[string]time.Duration {
    44  	sr.md.RLock()
    45  	defer sr.md.RUnlock()
    46  	am := sr.md.Metrics
    47  	if sr.md.IsInRecovery && len(sr.md.MetricsStandby) > 0 {
    48  		am = sr.md.MetricsStandby
    49  	}
    50  	c := make(map[string]time.Duration, len(am))
    51  	for k, v := range am {
    52  		c[k] = time.Duration(v) * time.Second
    53  	}
    54  	return c
    55  }
    56  
    57  // GCDSlice computes GCD across a slice. Returns 0 for empty input.
    58  func GCDSlice(vals []int) int {
    59  	if len(vals) == 0 {
    60  		return 0
    61  	}
    62  	g := vals[0]
    63  	for _, v := range vals[1:] {
    64  		for v != 0 {
    65  			g, v = v, g%v
    66  		}
    67  	}
    68  	return g
    69  }
    70  
    71  // calcTickInterval computes GCD of all metric intervals with a minimum floor.
    72  func (sr *SourceReaper) calcTickInterval() time.Duration {
    73  	am := sr.activeMetrics()
    74  	intervals := make([]int, 0, len(am))
    75  	for _, d := range am {
    76  		intervals = append(intervals, max(int(d.Seconds()), minTickInterval))
    77  	}
    78  	return time.Duration(max(GCDSlice(intervals), minTickInterval)) * time.Second
    79  }
    80  
    81  // cacheKey returns the instance-level cache key for the given metric.
    82  func (sr *SourceReaper) cacheKey(m metrics.Metric, name string) string {
    83  	age := sr.reaper.Metrics.CacheAge()
    84  	if m.IsInstanceLevel && age > 0 && sr.md.GetMetricInterval(name) < age {
    85  		return fmt.Sprintf("%s:%s", sr.md.GetClusterIdentifier(), name)
    86  	}
    87  	return ""
    88  }
    89  
    90  // isRoleExcluded returns true if the metric should be skipped based on the
    91  // source's recovery state (e.g. primary-only metric on a standby).
    92  func (sr *SourceReaper) isRoleExcluded(m metrics.Metric) bool {
    93  	sr.md.RLock()
    94  	defer sr.md.RUnlock()
    95  	return (m.PrimaryOnly() && sr.md.IsInRecovery) || (m.StandbyOnly() && !sr.md.IsInRecovery)
    96  }
    97  
    98  // sendEnvelope adds sysinfo and dispatches a MeasurementEnvelope to the
    99  // measurement channel.
   100  func (sr *SourceReaper) sendEnvelope(ctx context.Context, name, storageName string, data metrics.Measurements) {
   101  	log.GetLogger(ctx).WithField("metric", name).WithField("rows", len(data)).Info("measurements fetched")
   102  	sr.reaper.AddSysinfoToMeasurements(data, sr.md)
   103  	sr.reaper.measurementCh <- metrics.MeasurementEnvelope{
   104  		DBName:     sr.md.Name,
   105  		MetricName: cmp.Or(storageName, name),
   106  		Data:       data,
   107  		CustomTags: sr.md.CustomTags,
   108  	}
   109  }
   110  
   111  // dispatchMetricData handles the post-fetch workflow for a collected metric:
   112  // caching, sysinfo enrichment, sending, and restart detection.
   113  func (sr *SourceReaper) dispatchMetricData(ctx context.Context, name string, metric metrics.Metric, data metrics.Measurements) {
   114  	if key := sr.cacheKey(metric, name); key != "" {
   115  		sr.reaper.measurementCache.Put(key, data)
   116  	}
   117  	sr.sendEnvelope(ctx, name, metric.StorageName, data)
   118  	if name == "db_stats" {
   119  		sr.detectServerRestart(ctx, data)
   120  	}
   121  }
   122  
   123  // batchEntry holds the minimum info needed to execute and dispatch a metric query.
   124  type batchEntry struct {
   125  	metricName string
   126  	metric     metrics.Metric
   127  	sql        string
   128  }
   129  
   130  // Run is the main loop for a single source. It replaces N per-metric goroutines
   131  // with one goroutine that batches SQL queries at GCD-aligned ticks.
   132  func (sr *SourceReaper) Run(ctx context.Context) {
   133  	l := log.GetLogger(ctx).WithField("source", sr.md.Name)
   134  	ctx = log.WithLogger(ctx, l)
   135  	var err error
   136  	for {
   137  		if err = sr.md.FetchRuntimeInfo(ctx, false); err != nil {
   138  			l.WithError(err).Warning("could not refresh runtime info")
   139  		}
   140  
   141  		now := time.Now()
   142  		var batch []batchEntry
   143  
   144  		for metricName, metricInterval := range sr.activeMetrics() {
   145  			if metricInterval <= 0 {
   146  				continue
   147  			}
   148  			if lf := sr.lastFetch[metricName]; !lf.IsZero() && now.Sub(lf) < metricInterval {
   149  				continue
   150  			}
   151  
   152  			metric, ok := metricDefs.GetMetricDef(metricName)
   153  			if !ok || sr.isRoleExcluded(metric) {
   154  				continue
   155  			}
   156  
   157  			switch {
   158  			case metricName == specialMetricServerLogEventCounts:
   159  				if sr.lastFetch[metricName].IsZero() {
   160  					go func() {
   161  						if e := sr.runLogParser(ctx); e != nil {
   162  							l.WithError(e).Error("log parser error")
   163  						}
   164  					}()
   165  				}
   166  			case IsDirectlyFetchableMetric(sr.md, metricName):
   167  				err = sr.fetchOSMetric(ctx, metricName)
   168  				sr.lastFetch[metricName] = time.Now()
   169  			case metricName == specialMetricChangeEvents || metricName == specialMetricInstanceUp:
   170  				err = sr.fetchSpecialMetric(ctx, metricName, metric.StorageName)
   171  				sr.lastFetch[metricName] = time.Now()
   172  			default:
   173  				if cached := sr.reaper.GetMeasurementCache(sr.cacheKey(metric, metricName)); len(cached) > 0 {
   174  					l.WithField("metric", metricName).Info("instance level cache hit")
   175  					sr.sendEnvelope(ctx, metricName, metric.StorageName, cached)
   176  					sr.lastFetch[metricName] = time.Now()
   177  					break
   178  				}
   179  				sr.md.RLock()
   180  				version := sr.md.Version
   181  				sr.md.RUnlock()
   182  				sql := metric.GetSQL(version)
   183  				if sql == "" {
   184  					l.WithField("metric", metricName).WithField("version", version).Warning("no SQL found for metric version")
   185  					sr.lastFetch[metricName] = time.Now()
   186  					break
   187  				}
   188  				if _, degraded := sr.degradedMetrics[metricName]; degraded {
   189  					if err = sr.fetchMetric(ctx, batchEntry{metricName: metricName, metric: metric, sql: sql}); err != nil {
   190  						l.WithError(err).WithField("metric", metricName).Error("degraded metric fetch failed")
   191  					} else {
   192  						l.WithField("metric", metricName).Info("degraded metric recovered, returning to batch execution")
   193  						delete(sr.degradedMetrics, metricName)
   194  					}
   195  					sr.lastFetch[metricName] = time.Now()
   196  					break
   197  				}
   198  				batch = append(batch, batchEntry{metricName: metricName, metric: metric, sql: sql})
   199  			}
   200  			if err != nil {
   201  				l.WithError(err).WithField("metric", metricName).Error("failed to fetch metric")
   202  			}
   203  		}
   204  
   205  		if len(batch) > 0 {
   206  			// executeBatch/fetchMetrics log every failure individually with the
   207  			// metric name attached, so there is nothing to re-log here.
   208  			if sr.md.IsPostgresSource() {
   209  				sr.executeBatch(ctx, batch)
   210  			} else {
   211  				sr.fetchMetrics(ctx, batch)
   212  			}
   213  
   214  			now := time.Now()
   215  			for _, e := range batch {
   216  				sr.lastFetch[e.metricName] = now
   217  			}
   218  		}
   219  		select {
   220  		case <-ctx.Done():
   221  			return
   222  		case <-time.After(sr.calcTickInterval()):
   223  		}
   224  	}
   225  }
   226  
   227  // executeBatch sends all SQLs in a single pgx.Batch round-trip, dispatching
   228  // each result immediately as it arrives. If any query fails, PostgreSQL's
   229  // extended protocol aborts all subsequent queries in the same sync boundary
   230  // (cascade failure). Any entry that returns an error from the batch is retried
   231  // individually via fetchMetric to isolate real failures from cascade failures.
   232  // Entries that fail even after the individual retry are marked as degraded
   233  // so that subsequent runs use fetchMetric for them until they recover.
   234  //
   235  // Failures are logged per-metric as they happen.
   236  func (sr *SourceReaper) executeBatch(ctx context.Context, entries []batchEntry) {
   237  	batch := &pgx.Batch{}
   238  	for _, e := range entries {
   239  		batch.Queue(e.sql)
   240  	}
   241  
   242  	// A failing query aborts the rest of the batch (cascade), and the per-query
   243  	// retry below may also fail. Those failures are expected and handled, so run
   244  	// them under a context that downgrades pgx tracer errors to debug and avoids
   245  	// flooding the log with BatchClose/Query dumps. We log the real outcome once.
   246  	pgxCtx := log.WithSuppressedPgxErrors(ctx)
   247  	br := sr.md.Conn.SendBatch(pgxCtx, batch)
   248  	l := log.GetLogger(ctx)
   249  
   250  	var retries []batchEntry
   251  	for _, e := range entries {
   252  		rows, err := br.Query()
   253  		if err != nil {
   254  			// May be a real error or a cascade from an earlier failure; retry individually.
   255  			// Don't log here: a cascade failure is expected noise and the retry will
   256  			// log a single accurate error if the metric truly fails.
   257  			retries = append(retries, e)
   258  			continue
   259  		}
   260  		if err = sr.CollectAndDispatch(ctx, rows, e.metricName, e.metric); err != nil {
   261  			retries = append(retries, e)
   262  		}
   263  	}
   264  
   265  	// Close the batch explicitly before retrying to release the connection back to the
   266  	// pool. A deferred close would hold the connection through the retry loop, causing a
   267  	// potential deadlock
   268  	if err := br.Close(); err != nil {
   269  		l.WithError(err).Debug("failed to close batch")
   270  	}
   271  
   272  	sr.fetchMetrics(ctx, retries)
   273  }
   274  
   275  // fetchMetrics fetches a batch of entries individually, marking any that fail
   276  // as degraded so subsequent runs will continue to use individual fetching for them.
   277  // Each failure is logged on its own line with the metric name attached.
   278  func (sr *SourceReaper) fetchMetrics(ctx context.Context, entries []batchEntry) {
   279  	for _, e := range entries {
   280  		if err := sr.fetchMetric(ctx, e); err != nil {
   281  			log.GetLogger(ctx).
   282  				WithField("metric", e.metricName).
   283  				WithError(err).
   284  				Error("failed to fetch metric")
   285  			sr.degradedMetrics[e.metricName] = struct{}{}
   286  		}
   287  	}
   288  }
   289  
   290  // fetchMetric executes a single SQL query and returns the resulting measurements.
   291  // pgx tracer errors are suppressed because the caller logs a single, accurate
   292  // message (with the metric name) on failure; the raw SQL/args dump is only
   293  // emitted at debug level.
   294  func (sr *SourceReaper) fetchMetric(ctx context.Context, entry batchEntry) error {
   295  	rows, err := sr.md.Conn.Query(log.WithSuppressedPgxErrors(ctx), entry.sql, pgx.QueryExecModeSimpleProtocol)
   296  	if err != nil {
   297  		return err
   298  	}
   299  	return sr.CollectAndDispatch(ctx, rows, entry.metricName, entry.metric)
   300  }
   301  
   302  // CollectAndDispatch is a helper that collects rows from a pgx.Rows and dispatches them.
   303  func (sr *SourceReaper) CollectAndDispatch(ctx context.Context, rows pgx.Rows, name string, metric metrics.Metric) error {
   304  	data, err := pgx.CollectRows(rows, metrics.RowToMeasurement)
   305  	if err != nil {
   306  		return err
   307  	}
   308  	if len(data) > 0 {
   309  		sr.dispatchMetricData(ctx, name, metric, data)
   310  	}
   311  	return nil
   312  }
   313  
   314  // fetchOSMetric handles gopsutil-based OS metrics.
   315  func (sr *SourceReaper) fetchOSMetric(ctx context.Context, name string) error {
   316  	msg, err := sr.reaper.FetchStatsDirectlyFromOS(ctx, sr.md, name)
   317  	if err != nil {
   318  		return fmt.Errorf("could not read metric from OS: %v", err)
   319  	}
   320  	if msg != nil && len(msg.Data) > 0 {
   321  		log.GetLogger(ctx).WithField("metric", name).WithField("rows", len(msg.Data)).Info("measurements fetched")
   322  		sr.reaper.measurementCh <- *msg
   323  	}
   324  	return nil
   325  }
   326  
   327  // fetchSpecialMetric handles change_events and instance_up metrics.
   328  func (sr *SourceReaper) fetchSpecialMetric(ctx context.Context, name, storageName string) error {
   329  	var (
   330  		data metrics.Measurements
   331  		err  error
   332  	)
   333  	switch name {
   334  	case specialMetricChangeEvents:
   335  		data, err = sr.reaper.GetObjectChangesMeasurement(ctx, sr.md)
   336  	case specialMetricInstanceUp:
   337  		data, err = sr.reaper.GetInstanceUpMeasurement(ctx, sr.md)
   338  	}
   339  	if err != nil {
   340  		return fmt.Errorf("failed to fetch special metric: %v", err)
   341  	}
   342  	if len(data) > 0 {
   343  		sr.sendEnvelope(ctx, name, storageName, data)
   344  	}
   345  	return err
   346  }
   347  
   348  // runLogParser launches the server log event counts parser.
   349  func (sr *SourceReaper) runLogParser(ctx context.Context) error {
   350  	lp, err := NewLogParser(ctx, sr.md, sr.reaper.measurementCh)
   351  	if err != nil {
   352  		return fmt.Errorf("failed to initialize log parser: %v", err)
   353  	}
   354  	if err := lp.ParseLogs(); err != nil {
   355  		return fmt.Errorf("log parser error: %v", err)
   356  	}
   357  	return nil
   358  }
   359  
   360  // detectServerRestart checks for PostgreSQL server restarts via postmaster_uptime_s
   361  // in db_stats metric data and emits an object_changes measurement if detected.
   362  func (sr *SourceReaper) detectServerRestart(ctx context.Context, data metrics.Measurements) {
   363  	if len(data) == 0 {
   364  		return
   365  	}
   366  	uptimeS, ok := data[0]["postmaster_uptime_s"].(int64)
   367  	if !ok {
   368  		return
   369  	}
   370  	prev := sr.lastUptimeS
   371  	sr.lastUptimeS = uptimeS
   372  	if prev > 0 && uptimeS < prev {
   373  		l := log.GetLogger(ctx)
   374  		l.Warning("Detected server restart (or failover)")
   375  		entry := metrics.NewMeasurement(data.GetEpoch())
   376  		entry["details"] = "Detected server restart (or failover)"
   377  		sr.reaper.measurementCh <- metrics.MeasurementEnvelope{
   378  			DBName:     sr.md.Name,
   379  			MetricName: "object_changes",
   380  			Data:       metrics.Measurements{entry},
   381  			CustomTags: sr.md.CustomTags,
   382  		}
   383  	}
   384  }
   385