Source file src/github.com/cybertec-postgresql/pgwatch/v6/internal/reaper/database.go

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

     1  package reaper
     2  
     3  import (
     4  	"cmp"
     5  	"context"
     6  	"errors"
     7  	"fmt"
     8  	"strings"
     9  	"sync"
    10  	"time"
    11  
    12  	"github.com/cybertec-postgresql/pgwatch/v6/internal/db"
    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  	"github.com/jackc/pgx/v5"
    17  )
    18  
    19  const minTickInterval = 1 // seconds - floor for GCD to help handle zero/negative intervals
    20  
    21  var _ Reaper = (*DbConnReaper)(nil)
    22  
    23  // DbConnReaper manages metric collection for a single monitored database source.
    24  // Instead of one goroutine per metric it runs a single GCD-based tick loop
    25  // and batches SQL queries via pgx.Batch when the source is a real Postgres
    26  // connection (non-pgbouncer, non-pgpool).
    27  type DbConnReaper struct {
    28  	reaper      *reaper
    29  	md          *sources.DbConn
    30  	lastFetch   map[string]time.Time
    31  	lastUptimeS int64 // last seen postmaster_uptime_s for restart detection
    32  
    33  	logParserStarted bool // server_log_event_counts streaming parser was started; runs until ctx cancel
    34  
    35  	degradedMu      sync.RWMutex
    36  	degradedMetrics map[string]struct{} // metrics that failed individual retry; executed via fetchMetric until they recover
    37  }
    38  
    39  // NewDbConnReaper creates a SourceReaper for the given source connection.
    40  func NewDbConnReaper(r *reaper, md *sources.DbConn) *DbConnReaper {
    41  	return &DbConnReaper{
    42  		reaper:          r,
    43  		md:              md,
    44  		lastFetch:       make(map[string]time.Time),
    45  		degradedMetrics: make(map[string]struct{}),
    46  	}
    47  }
    48  
    49  // isDegraded reports whether the named metric is in the degraded set.
    50  func (sr *DbConnReaper) isDegraded(name string) bool {
    51  	sr.degradedMu.RLock()
    52  	_, ok := sr.degradedMetrics[name]
    53  	sr.degradedMu.RUnlock()
    54  	return ok
    55  }
    56  
    57  // markDegraded adds name to the degraded set.
    58  func (sr *DbConnReaper) markDegraded(name string) {
    59  	sr.degradedMu.Lock()
    60  	sr.degradedMetrics[name] = struct{}{}
    61  	sr.degradedMu.Unlock()
    62  }
    63  
    64  // clearDegraded removes name from the degraded set.
    65  func (sr *DbConnReaper) clearDegraded(name string) {
    66  	sr.degradedMu.Lock()
    67  	delete(sr.degradedMetrics, name)
    68  	sr.degradedMu.Unlock()
    69  }
    70  
    71  // activeMetrics returns a snapshot copy of the currently active metric intervals
    72  // as durations, delegating recovery-state selection to md.ActiveMetrics().
    73  func (sr *DbConnReaper) activeMetrics() map[string]time.Duration {
    74  	c := make(map[string]time.Duration)
    75  	for k, v := range sr.md.ActiveMetrics() {
    76  		c[k] = time.Duration(v) * time.Second
    77  	}
    78  	return c
    79  }
    80  
    81  // GCDSlice computes GCD across a slice. Returns 0 for empty input.
    82  func GCDSlice(vals []int) int {
    83  	if len(vals) == 0 {
    84  		return 0
    85  	}
    86  	g := vals[0]
    87  	for _, v := range vals[1:] {
    88  		for v != 0 {
    89  			g, v = v, g%v
    90  		}
    91  	}
    92  	return g
    93  }
    94  
    95  // calcTickInterval computes GCD of all metric intervals with a minimum floor.
    96  func (sr *DbConnReaper) calcTickInterval() time.Duration {
    97  	am := sr.activeMetrics()
    98  	intervals := make([]int, 0, len(am))
    99  	for _, d := range am {
   100  		intervals = append(intervals, max(int(d.Seconds()), minTickInterval))
   101  	}
   102  	return time.Duration(max(GCDSlice(intervals), minTickInterval)) * time.Second
   103  }
   104  
   105  // cacheKey returns the instance-level cache key for the given metric.
   106  func (sr *DbConnReaper) cacheKey(m metrics.Metric, name string) string {
   107  	age := sr.reaper.Metrics.CacheAge()
   108  	if m.IsInstanceLevel && age > 0 && sr.md.GetMetricInterval(name) < age {
   109  		return fmt.Sprintf("%s:%s", sr.md.GetClusterIdentifier(), name)
   110  	}
   111  	return ""
   112  }
   113  
   114  // isRoleExcluded returns true if the metric should be skipped based on the
   115  // source's recovery state (e.g. primary-only metric on a standby).
   116  func (sr *DbConnReaper) isRoleExcluded(m metrics.Metric) bool {
   117  	sr.md.RLock()
   118  	defer sr.md.RUnlock()
   119  	return (m.PrimaryOnly() && sr.md.IsInRecovery) || (m.StandbyOnly() && !sr.md.IsInRecovery)
   120  }
   121  
   122  // sendEnvelope adds sysinfo and dispatches a MeasurementEnvelope to the
   123  // measurement channel.
   124  func (sr *DbConnReaper) sendEnvelope(ctx context.Context, name, storageName string, data metrics.Measurements) {
   125  	log.GetLogger(ctx).WithField("metric", name).WithField("rows", len(data)).Info("measurements fetched")
   126  	sr.reaper.AddSysinfoToMeasurements(data, sr.md)
   127  	sr.reaper.measurementCh <- metrics.MeasurementEnvelope{
   128  		DBName:     sr.md.Name,
   129  		MetricName: cmp.Or(storageName, name),
   130  		Data:       data,
   131  		CustomTags: sr.md.CustomTags,
   132  	}
   133  }
   134  
   135  // dispatchMetricData handles the post-fetch workflow for a collected metric:
   136  // caching, sysinfo enrichment, sending, and restart detection.
   137  func (sr *DbConnReaper) dispatchMetricData(ctx context.Context, name string, metric metrics.Metric, data metrics.Measurements) {
   138  	if key := sr.cacheKey(metric, name); key != "" {
   139  		sr.reaper.measurementCache.Put(key, data)
   140  	}
   141  	sr.sendEnvelope(ctx, name, metric.StorageName, data)
   142  	if name == "db_stats" {
   143  		sr.detectServerRestart(ctx, data)
   144  	}
   145  }
   146  
   147  // batchEntry holds the minimum info needed to execute and dispatch a metric query.
   148  type batchEntry struct {
   149  	metricName string
   150  	metric     metrics.Metric
   151  	sql        string
   152  }
   153  
   154  // Run is the main loop for a single source. It replaces N per-metric goroutines
   155  // with one goroutine that batches SQL queries at GCD-aligned ticks.
   156  func (sr *DbConnReaper) Reap(ctx context.Context) {
   157  	l := log.GetLogger(ctx).WithField("source", sr.md.Name)
   158  	ctx = log.WithLogger(ctx, l)
   159  	var err error
   160  	for {
   161  		if err = sr.md.FetchRuntimeInfo(ctx, false); err != nil {
   162  			l.WithError(err).Warning("could not refresh runtime info")
   163  		}
   164  
   165  		now := time.Now()
   166  		var batch []batchEntry
   167  
   168  		for name, interval := range sr.activeMetrics() {
   169  			if interval <= 0 {
   170  				continue
   171  			}
   172  			if lf := sr.lastFetch[name]; !lf.IsZero() && now.Sub(lf) < interval {
   173  				continue
   174  			}
   175  
   176  			metric, ok := metricDefs.GetMetricDef(name)
   177  			if !ok || sr.isRoleExcluded(metric) {
   178  				continue
   179  			}
   180  			switch {
   181  			case name == specialMetricServerLogEventCounts:
   182  				if !sr.logParserStarted {
   183  					sr.logParserStarted = true // streaming parser starts once per worker lifetime and runs until ctx cancel
   184  					go func() {
   185  						if e := sr.runLogParser(ctx); e != nil {
   186  							l.WithError(e).Error("log parser error")
   187  						}
   188  					}()
   189  				}
   190  			case IsDirectlyFetchableMetric(sr.md, name):
   191  				err = sr.fetchOSMetric(ctx, name)
   192  				sr.lastFetch[name] = time.Now()
   193  			case name == specialMetricChangeEvents || name == specialMetricInstanceUp:
   194  				err = sr.fetchSpecialMetric(ctx, name, metric.StorageName)
   195  				sr.lastFetch[name] = time.Now()
   196  			default:
   197  				if cached := sr.reaper.GetMeasurementCache(sr.cacheKey(metric, name)); len(cached) > 0 {
   198  					l.WithField("metric", name).Info("instance level cache hit")
   199  					sr.sendEnvelope(ctx, name, metric.StorageName, cached)
   200  					sr.lastFetch[name] = time.Now()
   201  					break
   202  				}
   203  				sr.md.RLock()
   204  				version := sr.md.Version
   205  				sr.md.RUnlock()
   206  				sql := metric.GetSQL(version)
   207  				if sql == "" {
   208  					ll := l.WithField("metric", name).WithField("version", version)
   209  					if len(metric.SQLs) > 0 {
   210  						// SQL defined only for newer server versions, expected, e.g. PG 19-only metrics on PG 18
   211  						ll.Debug("metric not supported by source version")
   212  					} else {
   213  						ll.Warning("no SQL found for metric version")
   214  					}
   215  					sr.lastFetch[name] = time.Now()
   216  					break
   217  				}
   218  				if sr.isDegraded(name) {
   219  					if err = sr.fetchMetric(ctx, batchEntry{metricName: name, metric: metric, sql: sql}); err != nil {
   220  						l.WithError(err).WithField("metric", name).Error("degraded metric fetch failed")
   221  					} else {
   222  						l.WithField("metric", name).Info("degraded metric recovered, returning to batch execution")
   223  						sr.clearDegraded(name)
   224  					}
   225  					sr.lastFetch[name] = time.Now()
   226  					break
   227  				}
   228  				batch = append(batch, batchEntry{metricName: name, metric: metric, sql: sql})
   229  				continue
   230  			}
   231  			if err != nil {
   232  				l.WithError(err).WithField("metric", name).Error("failed to fetch metric")
   233  			}
   234  		}
   235  
   236  		if len(batch) > 0 {
   237  			if sr.md.IsPostgresSource() {
   238  				err = sr.executeBatch(ctx, batch)
   239  			} else {
   240  				for _, e := range batch {
   241  					err = errors.Join(err, sr.fetchMetric(ctx, e))
   242  				}
   243  			}
   244  
   245  			if errs, ok := err.(interface{ Unwrap() []error }); ok {
   246  				for _, err = range errs.Unwrap() {
   247  					l.WithError(err).Error("failed to fetch metric")
   248  				}
   249  			} else if err != nil {
   250  				l.WithError(err).Error("failed to fetch metric")
   251  			}
   252  
   253  			now := time.Now()
   254  			for _, e := range batch {
   255  				sr.lastFetch[e.metricName] = now
   256  			}
   257  		}
   258  		select {
   259  		case <-ctx.Done():
   260  			return
   261  		case <-time.After(sr.calcTickInterval()):
   262  		}
   263  	}
   264  }
   265  
   266  // executeBatch sends all SQLs in a single pgx.Batch round-trip, dispatching
   267  // each result immediately as it arrives. If any query fails, PostgreSQL's
   268  // extended protocol aborts all subsequent queries in the same sync boundary
   269  // (cascade failure). Any entry that returns an error from the batch is retried
   270  // individually via fetchMetric to isolate real failures from cascade failures.
   271  // Entries that fail even after the individual retry are marked as degraded
   272  // so that subsequent runs use fetchMetric for them until they recover.
   273  func (sr *DbConnReaper) executeBatch(ctx context.Context, entries []batchEntry) error {
   274  	batch := &pgx.Batch{}
   275  	for _, e := range entries {
   276  		batch.Queue(e.sql)
   277  	}
   278  
   279  	// Bound the batch round-trip by the tick interval (with the floor from
   280  	// db.MinFetchTimeout) so a wedged connection cannot stall the source
   281  	// worker past the next tick. The derived ctx embeds an op-named cause
   282  	// so the error chain is greppable per call site.
   283  	bctx, cancel := db.WithFetchTimeout(ctx, "batch", sr.calcTickInterval())
   284  	defer cancel()
   285  
   286  	br := sr.md.Conn.SendBatch(bctx, batch)
   287  	defer func() { _ = br.Close() }()
   288  
   289  	var (
   290  		errs    error
   291  		retries []batchEntry
   292  	)
   293  	for _, e := range entries {
   294  		rows, err := br.Query()
   295  		if err != nil {
   296  			// May be a real error or a cascade from an earlier failure; retry individually.
   297  			retries = append(retries, e)
   298  			continue
   299  		}
   300  		errs = errors.Join(errs, sr.CollectAndDispatch(ctx, rows, e.metricName, e.metric))
   301  	}
   302  
   303  	for _, e := range retries {
   304  		if err := sr.fetchMetric(ctx, e); err != nil {
   305  			errs = errors.Join(errs, fmt.Errorf("failed to fetch metric %s: %v", e.metricName, err))
   306  			log.GetLogger(ctx).WithField("metric", e.metricName).Warning("metric degraded after repeated failures, switching to individual fetch")
   307  			sr.markDegraded(e.metricName)
   308  		}
   309  	}
   310  	return errs
   311  }
   312  
   313  // fetchMetric executes a single SQL query and returns the resulting measurements.
   314  func (sr *DbConnReaper) fetchMetric(ctx context.Context, entry batchEntry) error {
   315  	// Bound the per-metric round-trip by max(GetMetricInterval, MinFetchTimeout).
   316  	// GetMetricInterval already returns a time.Duration in seconds — do NOT
   317  	// multiply by time.Second again.
   318  	interval := sr.md.GetMetricInterval(entry.metricName)
   319  	op := "fetch " + entry.metricName
   320  	fctx, cancel := db.WithFetchTimeout(ctx, op, interval)
   321  	defer cancel()
   322  	if err := sr.fetchMetricOnce(fctx, entry); err != nil {
   323  		return fmt.Errorf("%s: %w", op, err)
   324  	}
   325  	return nil
   326  }
   327  
   328  // fetchMetricOnce is the unwrapped round-trip — deadline-bound by the caller.
   329  func (sr *DbConnReaper) fetchMetricOnce(ctx context.Context, entry batchEntry) error {
   330  	rows, err := sr.md.Conn.Query(ctx, entry.sql, pgx.QueryExecModeSimpleProtocol)
   331  	if err != nil {
   332  		return err
   333  	}
   334  	return sr.CollectAndDispatch(ctx, rows, entry.metricName, entry.metric)
   335  }
   336  
   337  // CollectAndDispatch is a helper that collects rows from a pgx.Rows and dispatches them.
   338  func (sr *DbConnReaper) CollectAndDispatch(ctx context.Context, rows pgx.Rows, name string, metric metrics.Metric) error {
   339  	data, err := pgx.CollectRows(rows, metrics.RowToMeasurement)
   340  	if err != nil {
   341  		return err
   342  	}
   343  	if len(data) > 0 {
   344  		sr.dispatchMetricData(ctx, name, metric, data)
   345  	}
   346  	return nil
   347  }
   348  
   349  // fetchOSMetric handles gopsutil-based OS metrics.
   350  func (sr *DbConnReaper) fetchOSMetric(ctx context.Context, name string) error {
   351  	msg, err := sr.reaper.FetchStatsDirectlyFromOS(ctx, sr.md, name)
   352  	if err != nil {
   353  		return fmt.Errorf("could not read metric from OS: %v", err)
   354  	}
   355  	if msg != nil && len(msg.Data) > 0 {
   356  		log.GetLogger(ctx).WithField("metric", name).WithField("rows", len(msg.Data)).Info("measurements fetched")
   357  		sr.reaper.measurementCh <- *msg
   358  	}
   359  	return nil
   360  }
   361  
   362  // fetchSpecialMetric handles change_events and instance_up metrics.
   363  func (sr *DbConnReaper) fetchSpecialMetric(ctx context.Context, name, storageName string) error {
   364  	var (
   365  		data metrics.Measurements
   366  		err  error
   367  	)
   368  	switch name {
   369  	case specialMetricChangeEvents:
   370  		data, err = sr.reaper.GetObjectChangesMeasurement(ctx, sr.md)
   371  	case specialMetricInstanceUp:
   372  		data, err = sr.reaper.GetInstanceUpMeasurement(ctx, sr.md)
   373  	}
   374  	if err != nil {
   375  		return fmt.Errorf("failed to fetch special metric: %v", err)
   376  	}
   377  	if len(data) > 0 {
   378  		sr.sendEnvelope(ctx, name, storageName, data)
   379  	}
   380  	return err
   381  }
   382  
   383  // runLogParser launches the server log event counts parser.
   384  func (sr *DbConnReaper) runLogParser(ctx context.Context) error {
   385  	lp, err := NewLogParser(ctx, sr.md, sr.reaper.measurementCh)
   386  	if err != nil {
   387  		return fmt.Errorf("failed to initialize log parser: %v", err)
   388  	}
   389  	if err := lp.ParseLogs(); err != nil {
   390  		return fmt.Errorf("log parser error: %v", err)
   391  	}
   392  	return nil
   393  }
   394  
   395  // detectServerRestart checks for PostgreSQL server restarts via postmaster_uptime_s
   396  // in db_stats metric data and emits an object_changes measurement if detected.
   397  func (sr *DbConnReaper) detectServerRestart(ctx context.Context, data metrics.Measurements) {
   398  	if len(data) == 0 {
   399  		return
   400  	}
   401  	uptimeS, ok := data[0]["postmaster_uptime_s"].(int64)
   402  	if !ok {
   403  		return
   404  	}
   405  	prev := sr.lastUptimeS
   406  	sr.lastUptimeS = uptimeS
   407  	if prev > 0 && uptimeS < prev {
   408  		l := log.GetLogger(ctx)
   409  		l.Warning("Detected server restart (or failover)")
   410  		entry := metrics.NewMeasurement(data.GetEpoch())
   411  		entry["details"] = "Detected server restart (or failover)"
   412  		sr.reaper.measurementCh <- metrics.MeasurementEnvelope{
   413  			DBName:     sr.md.Name,
   414  			MetricName: "object_changes",
   415  			Data:       metrics.Measurements{entry},
   416  			CustomTags: sr.md.CustomTags,
   417  		}
   418  	}
   419  }
   420  
   421  // QueryMeasurements runs the given SQL on the source connection and returns
   422  // the collected rows. The round-trip is bounded by db.ChangeDetectionTimeout
   423  // so a hung source cannot block the change-detection sweep indefinitely.
   424  func QueryMeasurements(ctx context.Context, md *sources.DbConn, sql string, args ...any) (metrics.Measurements, error) {
   425  	if strings.TrimSpace(sql) == "" {
   426  		return nil, errors.New("empty SQL")
   427  	}
   428  	qctx, cancel := db.WithOpTimeout(ctx, "change_detection_query", db.ChangeDetectionTimeout)
   429  	defer cancel()
   430  	// For non-postgres connections (e.g. pgbouncer, pgpool), use simple protocol
   431  	if !md.IsPostgresSource() {
   432  		args = append([]any{pgx.QueryExecModeSimpleProtocol}, args...)
   433  	}
   434  	// lock_timeout is set at connection level via RuntimeParams, no need for transaction wrapper
   435  	rows, err := md.Conn.Query(qctx, sql, args...)
   436  	if err == nil {
   437  		return pgx.CollectRows(rows, metrics.RowToMeasurement)
   438  	}
   439  	if errors.Is(err, context.DeadlineExceeded) {
   440  		return nil, fmt.Errorf("change_detection_query: %w", context.Cause(qctx))
   441  	}
   442  	return nil, err
   443  }
   444  
   445  func (r *reaper) DetectSprocChanges(ctx context.Context, md *sources.DbConn) (changeCounts ChangeDetectionResults) {
   446  	detectedChanges := make(metrics.Measurements, 0)
   447  	var firstRun bool
   448  	l := log.GetLogger(ctx)
   449  	changeCounts.Target = "functions"
   450  	l.Debug("checking for sproc changes...")
   451  	if _, ok := md.ChangeState["sproc_hashes"]; !ok {
   452  		firstRun = true
   453  		md.ChangeState["sproc_hashes"] = make(map[string]string)
   454  	}
   455  	mvp, ok := metricDefs.GetMetricDef("sproc_hashes")
   456  	if !ok {
   457  		l.Error("could not get sproc_hashes sql")
   458  		return
   459  	}
   460  	data, err := QueryMeasurements(ctx, md, mvp.GetSQL(int(md.Version)))
   461  	if err != nil {
   462  		l.Error(err)
   463  		return
   464  	}
   465  	for _, dr := range data {
   466  		objIdent := dr["tag_sproc"].(string) + dbMetricJoinStr + dr["tag_oid"].(string)
   467  		prevHash, ok := md.ChangeState["sproc_hashes"][objIdent]
   468  		ll := l.WithField("sproc", dr["tag_sproc"]).WithField("oid", dr["tag_oid"])
   469  		if ok { // we have existing state
   470  			if prevHash != dr["md5"].(string) {
   471  				ll.Debug("change detected")
   472  				dr["event"] = "alter"
   473  				detectedChanges = append(detectedChanges, dr)
   474  				md.ChangeState["sproc_hashes"][objIdent] = dr["md5"].(string)
   475  				changeCounts.Altered++
   476  			}
   477  		} else { // check for new / delete
   478  			if !firstRun {
   479  				ll.Debug("new sproc detected")
   480  				dr["event"] = "create"
   481  				detectedChanges = append(detectedChanges, dr)
   482  				changeCounts.Created++
   483  			}
   484  			md.ChangeState["sproc_hashes"][objIdent] = dr["md5"].(string)
   485  		}
   486  	}
   487  	// detect deletes
   488  	if !firstRun && len(md.ChangeState["sproc_hashes"]) != len(data) {
   489  		currentOidMap := make(map[string]bool)
   490  		for _, dr := range data {
   491  			currentOidMap[dr["tag_sproc"].(string)+dbMetricJoinStr+dr["tag_oid"].(string)] = true
   492  		}
   493  		for sprocIdent := range md.ChangeState["sproc_hashes"] {
   494  			_, ok := currentOidMap[sprocIdent]
   495  			if !ok {
   496  				splits := strings.Split(sprocIdent, dbMetricJoinStr)
   497  				l.WithField("sproc", splits[0]).WithField("oid", splits[1]).Debug("deleted sproc detected")
   498  				m := metrics.NewMeasurement(data.GetEpoch())
   499  				m["event"] = "drop"
   500  				m["tag_sproc"] = splits[0]
   501  				m["tag_oid"] = splits[1]
   502  				detectedChanges = append(detectedChanges, m)
   503  				delete(md.ChangeState["sproc_hashes"], sprocIdent)
   504  				changeCounts.Dropped++
   505  			}
   506  		}
   507  	}
   508  	l.Debugf("sproc changes detected: %d", len(detectedChanges))
   509  	if len(detectedChanges) > 0 {
   510  		r.measurementCh <- metrics.MeasurementEnvelope{
   511  			DBName:     md.Name,
   512  			MetricName: "sproc_changes",
   513  			Data:       detectedChanges,
   514  			CustomTags: md.CustomTags,
   515  		}
   516  	}
   517  	return changeCounts
   518  }
   519  
   520  func (r *reaper) DetectTableChanges(ctx context.Context, md *sources.DbConn) ChangeDetectionResults {
   521  	detectedChanges := make(metrics.Measurements, 0)
   522  	var firstRun bool
   523  	var changeCounts ChangeDetectionResults
   524  	l := log.GetLogger(ctx)
   525  	changeCounts.Target = "tables"
   526  	l.Debug("checking for table changes...")
   527  	if _, ok := md.ChangeState["table_hashes"]; !ok {
   528  		firstRun = true
   529  		md.ChangeState["table_hashes"] = make(map[string]string)
   530  	}
   531  	mvp, ok := metricDefs.GetMetricDef("table_hashes")
   532  	if !ok {
   533  		l.Error("could not get table_hashes sql")
   534  		return changeCounts
   535  	}
   536  	data, err := QueryMeasurements(ctx, md, mvp.GetSQL(int(md.Version)))
   537  	if err != nil {
   538  		l.Error(err)
   539  		return changeCounts
   540  	}
   541  	for _, dr := range data {
   542  		objIdent := dr["tag_table"].(string)
   543  		prevHash, ok := md.ChangeState["table_hashes"][objIdent]
   544  		ll := l.WithField("table", dr["tag_table"])
   545  		if ok { // we have existing state
   546  			if prevHash != dr["md5"].(string) {
   547  				ll.Debug("change detected")
   548  				dr["event"] = "alter"
   549  				detectedChanges = append(detectedChanges, dr)
   550  				md.ChangeState["table_hashes"][objIdent] = dr["md5"].(string)
   551  				changeCounts.Altered++
   552  			}
   553  		} else { // check for new / delete
   554  			if !firstRun {
   555  				ll.Debug("new table detected")
   556  				dr["event"] = "create"
   557  				detectedChanges = append(detectedChanges, dr)
   558  				changeCounts.Created++
   559  			}
   560  			md.ChangeState["table_hashes"][objIdent] = dr["md5"].(string)
   561  		}
   562  	}
   563  	// detect deletes
   564  	if !firstRun && len(md.ChangeState["table_hashes"]) != len(data) {
   565  		deletedTables := make([]string, 0)
   566  		// turn resultset to map => [table]=true for faster checks
   567  		currentTableMap := make(map[string]bool)
   568  		for _, dr := range data {
   569  			currentTableMap[dr["tag_table"].(string)] = true
   570  		}
   571  		for table := range md.ChangeState["table_hashes"] {
   572  			_, ok := currentTableMap[table]
   573  			if !ok {
   574  				l.WithField("table", table).Debug("deleted table detected")
   575  				influxEntry := metrics.NewMeasurement(data.GetEpoch())
   576  				influxEntry["event"] = "drop"
   577  				influxEntry["tag_table"] = table
   578  				detectedChanges = append(detectedChanges, influxEntry)
   579  				deletedTables = append(deletedTables, table)
   580  				changeCounts.Dropped++
   581  			}
   582  		}
   583  		for _, deletedTable := range deletedTables {
   584  			delete(md.ChangeState["table_hashes"], deletedTable)
   585  		}
   586  	}
   587  	l.Debugf("table changes detected: %d", len(detectedChanges))
   588  	if len(detectedChanges) > 0 {
   589  		r.measurementCh <- metrics.MeasurementEnvelope{
   590  			DBName:     md.Name,
   591  			MetricName: "table_changes",
   592  			Data:       detectedChanges,
   593  			CustomTags: md.CustomTags,
   594  		}
   595  	}
   596  	return changeCounts
   597  }
   598  
   599  func (r *reaper) DetectIndexChanges(ctx context.Context, md *sources.DbConn) ChangeDetectionResults {
   600  	detectedChanges := make(metrics.Measurements, 0)
   601  	var firstRun bool
   602  	var changeCounts ChangeDetectionResults
   603  	l := log.GetLogger(ctx)
   604  	changeCounts.Target = "indexes"
   605  	l.Debug("checking for index changes...")
   606  	if _, ok := md.ChangeState["index_hashes"]; !ok {
   607  		firstRun = true
   608  		md.ChangeState["index_hashes"] = make(map[string]string)
   609  	}
   610  	mvp, ok := metricDefs.GetMetricDef("index_hashes")
   611  	if !ok {
   612  		l.Error("could not get index_hashes sql")
   613  		return changeCounts
   614  	}
   615  	data, err := QueryMeasurements(ctx, md, mvp.GetSQL(int(md.Version)))
   616  	if err != nil {
   617  		l.Error(err)
   618  		return changeCounts
   619  	}
   620  	for _, dr := range data {
   621  		objIdent := dr["tag_index"].(string)
   622  		prevHash, ok := md.ChangeState["index_hashes"][objIdent]
   623  		ll := l.WithField("index", dr["tag_index"]).WithField("table", dr["table"])
   624  		if ok { // we have existing state
   625  			if prevHash != (dr["md5"].(string) + dr["is_valid"].(string)) {
   626  				ll.Debug("change detected")
   627  				dr["event"] = "alter"
   628  				detectedChanges = append(detectedChanges, dr)
   629  				md.ChangeState["index_hashes"][objIdent] = dr["md5"].(string) + dr["is_valid"].(string)
   630  				changeCounts.Altered++
   631  			}
   632  		} else { // check for new / delete
   633  			if !firstRun {
   634  				ll.Debug("new index detected")
   635  				dr["event"] = "create"
   636  				detectedChanges = append(detectedChanges, dr)
   637  				changeCounts.Created++
   638  			}
   639  			md.ChangeState["index_hashes"][objIdent] = dr["md5"].(string) + dr["is_valid"].(string)
   640  		}
   641  	}
   642  	// detect deletes
   643  	if !firstRun && len(md.ChangeState["index_hashes"]) != len(data) {
   644  		deletedIndexes := make([]string, 0)
   645  		// turn resultset to map => [table]=true for faster checks
   646  		currentIndexMap := make(map[string]bool)
   647  		for _, dr := range data {
   648  			currentIndexMap[dr["tag_index"].(string)] = true
   649  		}
   650  		for indexName := range md.ChangeState["index_hashes"] {
   651  			_, ok := currentIndexMap[indexName]
   652  			if !ok {
   653  				l.WithField("index", indexName).Debug("deleted index detected")
   654  				influxEntry := metrics.NewMeasurement(data.GetEpoch())
   655  				influxEntry["event"] = "drop"
   656  				influxEntry["tag_index"] = indexName
   657  				detectedChanges = append(detectedChanges, influxEntry)
   658  				deletedIndexes = append(deletedIndexes, indexName)
   659  				changeCounts.Dropped++
   660  			}
   661  		}
   662  		for _, deletedIndex := range deletedIndexes {
   663  			delete(md.ChangeState["index_hashes"], deletedIndex)
   664  		}
   665  	}
   666  	l.Debugf("index changes detected: %d", len(detectedChanges))
   667  	if len(detectedChanges) > 0 {
   668  		r.measurementCh <- metrics.MeasurementEnvelope{
   669  			DBName:     md.Name,
   670  			MetricName: "index_changes",
   671  			Data:       detectedChanges,
   672  			CustomTags: md.CustomTags,
   673  		}
   674  	}
   675  	return changeCounts
   676  }
   677  
   678  func (r *reaper) DetectPrivilegeChanges(ctx context.Context, md *sources.DbConn) ChangeDetectionResults {
   679  	detectedChanges := make(metrics.Measurements, 0)
   680  	var firstRun bool
   681  	var changeCounts ChangeDetectionResults
   682  	l := log.GetLogger(ctx)
   683  	changeCounts.Target = "privileges"
   684  	l.Debug("checking object privilege changes...")
   685  	if _, ok := md.ChangeState["object_privileges"]; !ok {
   686  		firstRun = true
   687  		md.ChangeState["object_privileges"] = make(map[string]string)
   688  	}
   689  	mvp, ok := metricDefs.GetMetricDef("privilege_changes")
   690  	if !ok || mvp.GetSQL(int(md.Version)) == "" {
   691  		l.Warning("could not get SQL for 'privilege_changes'. cannot detect privilege changes")
   692  		return changeCounts
   693  	}
   694  	// returns rows of: object_type, tag_role, tag_object, privilege_type
   695  	data, err := QueryMeasurements(ctx, md, mvp.GetSQL(int(md.Version)))
   696  	if err != nil {
   697  		l.Error(err)
   698  		return changeCounts
   699  	}
   700  	currentState := make(map[string]bool)
   701  	for _, dr := range data {
   702  		objIdent := fmt.Sprintf("%s#:#%s#:#%s#:#%s", dr["object_type"], dr["tag_role"], dr["tag_object"], dr["privilege_type"])
   703  		ll := l.WithField("role", dr["tag_role"]).
   704  			WithField("object_type", dr["object_type"]).
   705  			WithField("object", dr["tag_object"]).
   706  			WithField("privilege_type", dr["privilege_type"])
   707  		if firstRun {
   708  			md.ChangeState["object_privileges"][objIdent] = ""
   709  		} else {
   710  			_, ok := md.ChangeState["object_privileges"][objIdent]
   711  			if !ok {
   712  				ll.Debug("new object privileges detected")
   713  				dr["event"] = "GRANT"
   714  				detectedChanges = append(detectedChanges, dr)
   715  				changeCounts.Created++
   716  				md.ChangeState["object_privileges"][objIdent] = ""
   717  			}
   718  			currentState[objIdent] = true
   719  		}
   720  	}
   721  	// check revokes - exists in old state only
   722  	if !firstRun && len(currentState) > 0 {
   723  		for objPrevRun := range md.ChangeState["object_privileges"] {
   724  			if _, ok := currentState[objPrevRun]; !ok {
   725  				splits := strings.Split(objPrevRun, "#:#")
   726  				l.WithField("role", splits[1]).
   727  					WithField("object_type", splits[0]).
   728  					WithField("object", splits[2]).
   729  					WithField("privilege_type", splits[3]).
   730  					Debug("removed object privileges detected")
   731  				revokeEntry := metrics.NewMeasurement(data.GetEpoch())
   732  				revokeEntry["object_type"] = splits[0]
   733  				revokeEntry["tag_role"] = splits[1]
   734  				revokeEntry["tag_object"] = splits[2]
   735  				revokeEntry["privilege_type"] = splits[3]
   736  				revokeEntry["event"] = "REVOKE"
   737  				detectedChanges = append(detectedChanges, revokeEntry)
   738  				changeCounts.Dropped++
   739  				delete(md.ChangeState["object_privileges"], objPrevRun)
   740  			}
   741  		}
   742  	}
   743  	l.Debugf("object privilege changes detected: %d", len(detectedChanges))
   744  	if len(detectedChanges) > 0 {
   745  		r.measurementCh <- metrics.MeasurementEnvelope{
   746  			DBName:     md.Name,
   747  			MetricName: "privilege_changes",
   748  			Data:       detectedChanges,
   749  			CustomTags: md.CustomTags,
   750  		}
   751  	}
   752  	return changeCounts
   753  }
   754  
   755  func (r *reaper) DetectConfigurationChanges(ctx context.Context, md *sources.DbConn) ChangeDetectionResults {
   756  	detectedChanges := make(metrics.Measurements, 0)
   757  	var firstRun bool
   758  	var changeCounts ChangeDetectionResults
   759  	l := log.GetLogger(ctx)
   760  	changeCounts.Target = "settings"
   761  	l.Debug("checking for configuration changes...")
   762  	if _, ok := md.ChangeState["configuration_hashes"]; !ok {
   763  		firstRun = true
   764  		md.ChangeState["configuration_hashes"] = make(map[string]string)
   765  	}
   766  	mvp, ok := metricDefs.GetMetricDef("configuration_hashes")
   767  	if !ok {
   768  		l.Error("could not get configuration_hashes sql")
   769  		return changeCounts
   770  	}
   771  	rows, err := md.Conn.Query(ctx, mvp.GetSQL(md.Version))
   772  	if err != nil {
   773  		l.Error(err)
   774  		return changeCounts
   775  	}
   776  	defer rows.Close()
   777  	var (
   778  		objIdent, objValue string
   779  		epoch              int64
   780  	)
   781  	for rows.Next() {
   782  		if rows.Scan(&epoch, &objIdent, &objValue) != nil {
   783  			return changeCounts
   784  		}
   785  		prevРash, ok := md.ChangeState["configuration_hashes"][objIdent]
   786  		ll := l.WithField("setting", objIdent)
   787  		if ok { // we have existing state
   788  			if prevРash != objValue {
   789  				ll.Warningf("settings change detected: %s = %s (prev: %s)", objIdent, objValue, prevРash)
   790  				detectedChanges = append(detectedChanges, metrics.Measurement{
   791  					metrics.EpochColumnName: epoch,
   792  					"tag_setting":           objIdent,
   793  					"value":                 objValue,
   794  					"event":                 "alter"})
   795  				md.ChangeState["configuration_hashes"][objIdent] = objValue
   796  				changeCounts.Altered++
   797  			}
   798  		} else { // check for new, delete not relevant here (pg_upgrade)
   799  			md.ChangeState["configuration_hashes"][objIdent] = objValue
   800  			if firstRun {
   801  				continue
   802  			}
   803  			ll.Debug("new setting detected")
   804  			detectedChanges = append(detectedChanges, metrics.Measurement{
   805  				metrics.EpochColumnName: epoch,
   806  				"tag_setting":           objIdent,
   807  				"value":                 objValue,
   808  				"event":                 "create"})
   809  			changeCounts.Created++
   810  		}
   811  	}
   812  	l.Debugf("configuration changes detected: %d", len(detectedChanges))
   813  	if len(detectedChanges) > 0 {
   814  		r.measurementCh <- metrics.MeasurementEnvelope{
   815  			DBName:     md.Name,
   816  			MetricName: "configuration_changes",
   817  			Data:       detectedChanges,
   818  			CustomTags: md.CustomTags,
   819  		}
   820  	}
   821  	return changeCounts
   822  }
   823  
   824  // GetInstanceUpMeasurement returns a single measurement with "instance_up" metric
   825  // used to detect if the instance is up or down
   826  func (r *reaper) GetInstanceUpMeasurement(ctx context.Context, md *sources.DbConn) (metrics.Measurements, error) {
   827  	return metrics.Measurements{
   828  		metrics.Measurement{
   829  			metrics.EpochColumnName: time.Now().UnixNano(),
   830  			"instance_up": func() int {
   831  				if md.Conn.Ping(ctx) == nil {
   832  					return 1
   833  				}
   834  				return 0
   835  			}(), // true if connection is up
   836  		},
   837  	}, nil // always return nil error for the status metric
   838  }
   839  
   840  func (r *reaper) GetObjectChangesMeasurement(ctx context.Context, md *sources.DbConn) (metrics.Measurements, error) {
   841  	md.Lock()
   842  	defer md.Unlock()
   843  	spN := r.DetectSprocChanges(ctx, md)
   844  	tblN := r.DetectTableChanges(ctx, md)
   845  	idxN := r.DetectIndexChanges(ctx, md)
   846  	cnfN := r.DetectConfigurationChanges(ctx, md)
   847  	privN := r.DetectPrivilegeChanges(ctx, md)
   848  	if spN.Total()+tblN.Total()+idxN.Total()+cnfN.Total()+privN.Total() == 0 {
   849  		return nil, nil
   850  	}
   851  	m := metrics.NewMeasurement(time.Now().UnixNano())
   852  	m["details"] = strings.Join([]string{spN.String(), tblN.String(), idxN.String(), cnfN.String(), privN.String()}, " ")
   853  	return metrics.Measurements{m}, nil
   854  }
   855