▽

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  					l.WithField("source", sr.md.Name).WithField("version", version).Warning("no SQL found for metric version")
   209  					sr.lastFetch[name] = time.Now()
   210  					break
   211  				}
   212  				if sr.isDegraded(name) {
   213  					if err = sr.fetchMetric(ctx, batchEntry{metricName: name, metric: metric, sql: sql}); err != nil {
   214  						l.WithError(err).WithField("metric", name).Error("degraded metric fetch failed")
   215  					} else {
   216  						l.WithField("metric", name).Info("degraded metric recovered, returning to batch execution")
   217  						sr.clearDegraded(name)
   218  					}
   219  					sr.lastFetch[name] = time.Now()
   220  					break
   221  				}
   222  				batch = append(batch, batchEntry{metricName: name, metric: metric, sql: sql})
   223  				continue
   224  			}
   225  			if err != nil {
   226  				l.WithError(err).WithField("metric", name).Error("failed to fetch metric")
   227  			}
   228  		}
   229  
   230  		if len(batch) > 0 {
   231  			if sr.md.IsPostgresSource() {
   232  				err = sr.executeBatch(ctx, batch)
   233  			} else {
   234  				for _, e := range batch {
   235  					err = errors.Join(err, sr.fetchMetric(ctx, e))
   236  				}
   237  			}
   238  
   239  			if errs, ok := err.(interface{ Unwrap() []error }); ok {
   240  				for _, err = range errs.Unwrap() {
   241  					l.WithError(err).Error("failed to fetch metric")
   242  				}
   243  			} else if err != nil {
   244  				l.WithError(err).Error("failed to fetch metric")
   245  			}
   246  
   247  			now := time.Now()
   248  			for _, e := range batch {
   249  				sr.lastFetch[e.metricName] = now
   250  			}
   251  		}
   252  		select {
   253  		case <-ctx.Done():
   254  			return
   255  		case <-time.After(sr.calcTickInterval()):
   256  		}
   257  	}
   258  }
   259  
   260  // executeBatch sends all SQLs in a single pgx.Batch round-trip, dispatching
   261  // each result immediately as it arrives. If any query fails, PostgreSQL's
   262  // extended protocol aborts all subsequent queries in the same sync boundary
   263  // (cascade failure). Any entry that returns an error from the batch is retried
   264  // individually via fetchMetric to isolate real failures from cascade failures.
   265  // Entries that fail even after the individual retry are marked as degraded
   266  // so that subsequent runs use fetchMetric for them until they recover.
   267  func (sr *DbConnReaper) executeBatch(ctx context.Context, entries []batchEntry) error {
   268  	batch := &pgx.Batch{}
   269  	for _, e := range entries {
   270  		batch.Queue(e.sql)
   271  	}
   272  
   273  	// Bound the batch round-trip by the tick interval (with the floor from
   274  	// db.MinFetchTimeout) so a wedged connection cannot stall the source
   275  	// worker past the next tick. The derived ctx embeds an op-named cause
   276  	// so the error chain is greppable per call site.
   277  	bctx, cancel := db.WithFetchTimeout(ctx, "batch", sr.calcTickInterval())
   278  	defer cancel()
   279  
   280  	br := sr.md.Conn.SendBatch(bctx, batch)
   281  	defer func() { _ = br.Close() }()
   282  
   283  	var (
   284  		errs    error
   285  		retries []batchEntry
   286  	)
   287  	for _, e := range entries {
   288  		rows, err := br.Query()
   289  		if err != nil {
   290  			// May be a real error or a cascade from an earlier failure; retry individually.
   291  			retries = append(retries, e)
   292  			continue
   293  		}
   294  		errs = errors.Join(errs, sr.CollectAndDispatch(ctx, rows, e.metricName, e.metric))
   295  	}
   296  
   297  	for _, e := range retries {
   298  		if err := sr.fetchMetric(ctx, e); err != nil {
   299  			errs = errors.Join(errs, fmt.Errorf("failed to fetch metric %s: %v", e.metricName, err))
   300  			log.GetLogger(ctx).WithField("metric", e.metricName).Warning("metric degraded after repeated failures, switching to individual fetch")
   301  			sr.markDegraded(e.metricName)
   302  		}
   303  	}
   304  	return errs
   305  }
   306  
   307  // fetchMetric executes a single SQL query and returns the resulting measurements.
   308  func (sr *DbConnReaper) fetchMetric(ctx context.Context, entry batchEntry) error {
   309  	// Bound the per-metric round-trip by max(GetMetricInterval, MinFetchTimeout).
   310  	// GetMetricInterval already returns a time.Duration in seconds — do NOT
   311  	// multiply by time.Second again.
   312  	interval := sr.md.GetMetricInterval(entry.metricName)
   313  	op := "fetch " + entry.metricName
   314  	fctx, cancel := db.WithFetchTimeout(ctx, op, interval)
   315  	defer cancel()
   316  	if err := sr.fetchMetricOnce(fctx, entry); err != nil {
   317  		return fmt.Errorf("%s: %w", op, err)
   318  	}
   319  	return nil
   320  }
   321  
   322  // fetchMetricOnce is the unwrapped round-trip — deadline-bound by the caller.
   323  func (sr *DbConnReaper) fetchMetricOnce(ctx context.Context, entry batchEntry) error {
   324  	rows, err := sr.md.Conn.Query(ctx, entry.sql, pgx.QueryExecModeSimpleProtocol)
   325  	if err != nil {
   326  		return err
   327  	}
   328  	return sr.CollectAndDispatch(ctx, rows, entry.metricName, entry.metric)
   329  }
   330  
   331  // CollectAndDispatch is a helper that collects rows from a pgx.Rows and dispatches them.
   332  func (sr *DbConnReaper) CollectAndDispatch(ctx context.Context, rows pgx.Rows, name string, metric metrics.Metric) error {
   333  	data, err := pgx.CollectRows(rows, metrics.RowToMeasurement)
   334  	if err != nil {
   335  		return err
   336  	}
   337  	if len(data) > 0 {
   338  		sr.dispatchMetricData(ctx, name, metric, data)
   339  	}
   340  	return nil
   341  }
   342  
   343  // fetchOSMetric handles gopsutil-based OS metrics.
   344  func (sr *DbConnReaper) fetchOSMetric(ctx context.Context, name string) error {
   345  	msg, err := sr.reaper.FetchStatsDirectlyFromOS(ctx, sr.md, name)
   346  	if err != nil {
   347  		return fmt.Errorf("could not read metric from OS: %v", err)
   348  	}
   349  	if msg != nil && len(msg.Data) > 0 {
   350  		log.GetLogger(ctx).WithField("metric", name).WithField("rows", len(msg.Data)).Info("measurements fetched")
   351  		sr.reaper.measurementCh <- *msg
   352  	}
   353  	return nil
   354  }
   355  
   356  // fetchSpecialMetric handles change_events and instance_up metrics.
   357  func (sr *DbConnReaper) fetchSpecialMetric(ctx context.Context, name, storageName string) error {
   358  	var (
   359  		data metrics.Measurements
   360  		err  error
   361  	)
   362  	switch name {
   363  	case specialMetricChangeEvents:
   364  		data, err = sr.reaper.GetObjectChangesMeasurement(ctx, sr.md)
   365  	case specialMetricInstanceUp:
   366  		data, err = sr.reaper.GetInstanceUpMeasurement(ctx, sr.md)
   367  	}
   368  	if err != nil {
   369  		return fmt.Errorf("failed to fetch special metric: %v", err)
   370  	}
   371  	if len(data) > 0 {
   372  		sr.sendEnvelope(ctx, name, storageName, data)
   373  	}
   374  	return err
   375  }
   376  
   377  // runLogParser launches the server log event counts parser.
   378  func (sr *DbConnReaper) runLogParser(ctx context.Context) error {
   379  	lp, err := NewLogParser(ctx, sr.md, sr.reaper.measurementCh)
   380  	if err != nil {
   381  		return fmt.Errorf("failed to initialize log parser: %v", err)
   382  	}
   383  	if err := lp.ParseLogs(); err != nil {
   384  		return fmt.Errorf("log parser error: %v", err)
   385  	}
   386  	return nil
   387  }
   388  
   389  // detectServerRestart checks for PostgreSQL server restarts via postmaster_uptime_s
   390  // in db_stats metric data and emits an object_changes measurement if detected.
   391  func (sr *DbConnReaper) detectServerRestart(ctx context.Context, data metrics.Measurements) {
   392  	if len(data) == 0 {
   393  		return
   394  	}
   395  	uptimeS, ok := data[0]["postmaster_uptime_s"].(int64)
   396  	if !ok {
   397  		return
   398  	}
   399  	prev := sr.lastUptimeS
   400  	sr.lastUptimeS = uptimeS
   401  	if prev > 0 && uptimeS < prev {
   402  		l := log.GetLogger(ctx)
   403  		l.Warning("Detected server restart (or failover)")
   404  		entry := metrics.NewMeasurement(data.GetEpoch())
   405  		entry["details"] = "Detected server restart (or failover)"
   406  		sr.reaper.measurementCh <- metrics.MeasurementEnvelope{
   407  			DBName:     sr.md.Name,
   408  			MetricName: "object_changes",
   409  			Data:       metrics.Measurements{entry},
   410  			CustomTags: sr.md.CustomTags,
   411  		}
   412  	}
   413  }
   414  
   415  // QueryMeasurements runs the given SQL on the source connection and returns
   416  // the collected rows. The round-trip is bounded by db.ChangeDetectionTimeout
   417  // so a hung source cannot block the change-detection sweep indefinitely.
   418  func QueryMeasurements(ctx context.Context, md *sources.DbConn, sql string, args ...any) (metrics.Measurements, error) {
   419  	if strings.TrimSpace(sql) == "" {
   420  		return nil, errors.New("empty SQL")
   421  	}
   422  	qctx, cancel := db.WithOpTimeout(ctx, "change_detection_query", db.ChangeDetectionTimeout)
   423  	defer cancel()
   424  	// For non-postgres connections (e.g. pgbouncer, pgpool), use simple protocol
   425  	if !md.IsPostgresSource() {
   426  		args = append([]any{pgx.QueryExecModeSimpleProtocol}, args...)
   427  	}
   428  	// lock_timeout is set at connection level via RuntimeParams, no need for transaction wrapper
   429  	rows, err := md.Conn.Query(qctx, sql, args...)
   430  	if err == nil {
   431  		return pgx.CollectRows(rows, metrics.RowToMeasurement)
   432  	}
   433  	if errors.Is(err, context.DeadlineExceeded) {
   434  		return nil, fmt.Errorf("change_detection_query: %w", context.Cause(qctx))
   435  	}
   436  	return nil, err
   437  }
   438  
   439  func (r *reaper) DetectSprocChanges(ctx context.Context, md *sources.DbConn) (changeCounts ChangeDetectionResults) {
   440  	detectedChanges := make(metrics.Measurements, 0)
   441  	var firstRun bool
   442  	l := log.GetLogger(ctx)
   443  	changeCounts.Target = "functions"
   444  	l.Debug("checking for sproc changes...")
   445  	if _, ok := md.ChangeState["sproc_hashes"]; !ok {
   446  		firstRun = true
   447  		md.ChangeState["sproc_hashes"] = make(map[string]string)
   448  	}
   449  	mvp, ok := metricDefs.GetMetricDef("sproc_hashes")
   450  	if !ok {
   451  		l.Error("could not get sproc_hashes sql")
   452  		return
   453  	}
   454  	data, err := QueryMeasurements(ctx, md, mvp.GetSQL(int(md.Version)))
   455  	if err != nil {
   456  		l.Error(err)
   457  		return
   458  	}
   459  	for _, dr := range data {
   460  		objIdent := dr["tag_sproc"].(string) + dbMetricJoinStr + dr["tag_oid"].(string)
   461  		prevHash, ok := md.ChangeState["sproc_hashes"][objIdent]
   462  		ll := l.WithField("sproc", dr["tag_sproc"]).WithField("oid", dr["tag_oid"])
   463  		if ok { // we have existing state
   464  			if prevHash != dr["md5"].(string) {
   465  				ll.Debug("change detected")
   466  				dr["event"] = "alter"
   467  				detectedChanges = append(detectedChanges, dr)
   468  				md.ChangeState["sproc_hashes"][objIdent] = dr["md5"].(string)
   469  				changeCounts.Altered++
   470  			}
   471  		} else { // check for new / delete
   472  			if !firstRun {
   473  				ll.Debug("new sproc detected")
   474  				dr["event"] = "create"
   475  				detectedChanges = append(detectedChanges, dr)
   476  				changeCounts.Created++
   477  			}
   478  			md.ChangeState["sproc_hashes"][objIdent] = dr["md5"].(string)
   479  		}
   480  	}
   481  	// detect deletes
   482  	if !firstRun && len(md.ChangeState["sproc_hashes"]) != len(data) {
   483  		currentOidMap := make(map[string]bool)
   484  		for _, dr := range data {
   485  			currentOidMap[dr["tag_sproc"].(string)+dbMetricJoinStr+dr["tag_oid"].(string)] = true
   486  		}
   487  		for sprocIdent := range md.ChangeState["sproc_hashes"] {
   488  			_, ok := currentOidMap[sprocIdent]
   489  			if !ok {
   490  				splits := strings.Split(sprocIdent, dbMetricJoinStr)
   491  				l.WithField("sproc", splits[0]).WithField("oid", splits[1]).Debug("deleted sproc detected")
   492  				m := metrics.NewMeasurement(data.GetEpoch())
   493  				m["event"] = "drop"
   494  				m["tag_sproc"] = splits[0]
   495  				m["tag_oid"] = splits[1]
   496  				detectedChanges = append(detectedChanges, m)
   497  				delete(md.ChangeState["sproc_hashes"], sprocIdent)
   498  				changeCounts.Dropped++
   499  			}
   500  		}
   501  	}
   502  	l.Debugf("sproc changes detected: %d", len(detectedChanges))
   503  	if len(detectedChanges) > 0 {
   504  		r.measurementCh <- metrics.MeasurementEnvelope{
   505  			DBName:     md.Name,
   506  			MetricName: "sproc_changes",
   507  			Data:       detectedChanges,
   508  			CustomTags: md.CustomTags,
   509  		}
   510  	}
   511  	return changeCounts
   512  }
   513  
   514  func (r *reaper) DetectTableChanges(ctx context.Context, md *sources.DbConn) ChangeDetectionResults {
   515  	detectedChanges := make(metrics.Measurements, 0)
   516  	var firstRun bool
   517  	var changeCounts ChangeDetectionResults
   518  	l := log.GetLogger(ctx)
   519  	changeCounts.Target = "tables"
   520  	l.Debug("checking for table changes...")
   521  	if _, ok := md.ChangeState["table_hashes"]; !ok {
   522  		firstRun = true
   523  		md.ChangeState["table_hashes"] = make(map[string]string)
   524  	}
   525  	mvp, ok := metricDefs.GetMetricDef("table_hashes")
   526  	if !ok {
   527  		l.Error("could not get table_hashes sql")
   528  		return changeCounts
   529  	}
   530  	data, err := QueryMeasurements(ctx, md, mvp.GetSQL(int(md.Version)))
   531  	if err != nil {
   532  		l.Error(err)
   533  		return changeCounts
   534  	}
   535  	for _, dr := range data {
   536  		objIdent := dr["tag_table"].(string)
   537  		prevHash, ok := md.ChangeState["table_hashes"][objIdent]
   538  		ll := l.WithField("table", dr["tag_table"])
   539  		if ok { // we have existing state
   540  			if prevHash != dr["md5"].(string) {
   541  				ll.Debug("change detected")
   542  				dr["event"] = "alter"
   543  				detectedChanges = append(detectedChanges, dr)
   544  				md.ChangeState["table_hashes"][objIdent] = dr["md5"].(string)
   545  				changeCounts.Altered++
   546  			}
   547  		} else { // check for new / delete
   548  			if !firstRun {
   549  				ll.Debug("new table detected")
   550  				dr["event"] = "create"
   551  				detectedChanges = append(detectedChanges, dr)
   552  				changeCounts.Created++
   553  			}
   554  			md.ChangeState["table_hashes"][objIdent] = dr["md5"].(string)
   555  		}
   556  	}
   557  	// detect deletes
   558  	if !firstRun && len(md.ChangeState["table_hashes"]) != len(data) {
   559  		deletedTables := make([]string, 0)
   560  		// turn resultset to map => [table]=true for faster checks
   561  		currentTableMap := make(map[string]bool)
   562  		for _, dr := range data {
   563  			currentTableMap[dr["tag_table"].(string)] = true
   564  		}
   565  		for table := range md.ChangeState["table_hashes"] {
   566  			_, ok := currentTableMap[table]
   567  			if !ok {
   568  				l.WithField("table", table).Debug("deleted table detected")
   569  				influxEntry := metrics.NewMeasurement(data.GetEpoch())
   570  				influxEntry["event"] = "drop"
   571  				influxEntry["tag_table"] = table
   572  				detectedChanges = append(detectedChanges, influxEntry)
   573  				deletedTables = append(deletedTables, table)
   574  				changeCounts.Dropped++
   575  			}
   576  		}
   577  		for _, deletedTable := range deletedTables {
   578  			delete(md.ChangeState["table_hashes"], deletedTable)
   579  		}
   580  	}
   581  	l.Debugf("table changes detected: %d", len(detectedChanges))
   582  	if len(detectedChanges) > 0 {
   583  		r.measurementCh <- metrics.MeasurementEnvelope{
   584  			DBName:     md.Name,
   585  			MetricName: "table_changes",
   586  			Data:       detectedChanges,
   587  			CustomTags: md.CustomTags,
   588  		}
   589  	}
   590  	return changeCounts
   591  }
   592  
   593  func (r *reaper) DetectIndexChanges(ctx context.Context, md *sources.DbConn) ChangeDetectionResults {
   594  	detectedChanges := make(metrics.Measurements, 0)
   595  	var firstRun bool
   596  	var changeCounts ChangeDetectionResults
   597  	l := log.GetLogger(ctx)
   598  	changeCounts.Target = "indexes"
   599  	l.Debug("checking for index changes...")
   600  	if _, ok := md.ChangeState["index_hashes"]; !ok {
   601  		firstRun = true
   602  		md.ChangeState["index_hashes"] = make(map[string]string)
   603  	}
   604  	mvp, ok := metricDefs.GetMetricDef("index_hashes")
   605  	if !ok {
   606  		l.Error("could not get index_hashes sql")
   607  		return changeCounts
   608  	}
   609  	data, err := QueryMeasurements(ctx, md, mvp.GetSQL(int(md.Version)))
   610  	if err != nil {
   611  		l.Error(err)
   612  		return changeCounts
   613  	}
   614  	for _, dr := range data {
   615  		objIdent := dr["tag_index"].(string)
   616  		prevHash, ok := md.ChangeState["index_hashes"][objIdent]
   617  		ll := l.WithField("index", dr["tag_index"]).WithField("table", dr["table"])
   618  		if ok { // we have existing state
   619  			if prevHash != (dr["md5"].(string) + dr["is_valid"].(string)) {
   620  				ll.Debug("change detected")
   621  				dr["event"] = "alter"
   622  				detectedChanges = append(detectedChanges, dr)
   623  				md.ChangeState["index_hashes"][objIdent] = dr["md5"].(string) + dr["is_valid"].(string)
   624  				changeCounts.Altered++
   625  			}
   626  		} else { // check for new / delete
   627  			if !firstRun {
   628  				ll.Debug("new index detected")
   629  				dr["event"] = "create"
   630  				detectedChanges = append(detectedChanges, dr)
   631  				changeCounts.Created++
   632  			}
   633  			md.ChangeState["index_hashes"][objIdent] = dr["md5"].(string) + dr["is_valid"].(string)
   634  		}
   635  	}
   636  	// detect deletes
   637  	if !firstRun && len(md.ChangeState["index_hashes"]) != len(data) {
   638  		deletedIndexes := make([]string, 0)
   639  		// turn resultset to map => [table]=true for faster checks
   640  		currentIndexMap := make(map[string]bool)
   641  		for _, dr := range data {
   642  			currentIndexMap[dr["tag_index"].(string)] = true
   643  		}
   644  		for indexName := range md.ChangeState["index_hashes"] {
   645  			_, ok := currentIndexMap[indexName]
   646  			if !ok {
   647  				l.WithField("index", indexName).Debug("deleted index detected")
   648  				influxEntry := metrics.NewMeasurement(data.GetEpoch())
   649  				influxEntry["event"] = "drop"
   650  				influxEntry["tag_index"] = indexName
   651  				detectedChanges = append(detectedChanges, influxEntry)
   652  				deletedIndexes = append(deletedIndexes, indexName)
   653  				changeCounts.Dropped++
   654  			}
   655  		}
   656  		for _, deletedIndex := range deletedIndexes {
   657  			delete(md.ChangeState["index_hashes"], deletedIndex)
   658  		}
   659  	}
   660  	l.Debugf("index changes detected: %d", len(detectedChanges))
   661  	if len(detectedChanges) > 0 {
   662  		r.measurementCh <- metrics.MeasurementEnvelope{
   663  			DBName:     md.Name,
   664  			MetricName: "index_changes",
   665  			Data:       detectedChanges,
   666  			CustomTags: md.CustomTags,
   667  		}
   668  	}
   669  	return changeCounts
   670  }
   671  
   672  func (r *reaper) DetectPrivilegeChanges(ctx context.Context, md *sources.DbConn) ChangeDetectionResults {
   673  	detectedChanges := make(metrics.Measurements, 0)
   674  	var firstRun bool
   675  	var changeCounts ChangeDetectionResults
   676  	l := log.GetLogger(ctx)
   677  	changeCounts.Target = "privileges"
   678  	l.Debug("checking object privilege changes...")
   679  	if _, ok := md.ChangeState["object_privileges"]; !ok {
   680  		firstRun = true
   681  		md.ChangeState["object_privileges"] = make(map[string]string)
   682  	}
   683  	mvp, ok := metricDefs.GetMetricDef("privilege_changes")
   684  	if !ok || mvp.GetSQL(int(md.Version)) == "" {
   685  		l.Warning("could not get SQL for 'privilege_changes'. cannot detect privilege changes")
   686  		return changeCounts
   687  	}
   688  	// returns rows of: object_type, tag_role, tag_object, privilege_type
   689  	data, err := QueryMeasurements(ctx, md, mvp.GetSQL(int(md.Version)))
   690  	if err != nil {
   691  		l.Error(err)
   692  		return changeCounts
   693  	}
   694  	currentState := make(map[string]bool)
   695  	for _, dr := range data {
   696  		objIdent := fmt.Sprintf("%s#:#%s#:#%s#:#%s", dr["object_type"], dr["tag_role"], dr["tag_object"], dr["privilege_type"])
   697  		ll := l.WithField("role", dr["tag_role"]).
   698  			WithField("object_type", dr["object_type"]).
   699  			WithField("object", dr["tag_object"]).
   700  			WithField("privilege_type", dr["privilege_type"])
   701  		if firstRun {
   702  			md.ChangeState["object_privileges"][objIdent] = ""
   703  		} else {
   704  			_, ok := md.ChangeState["object_privileges"][objIdent]
   705  			if !ok {
   706  				ll.Debug("new object privileges detected")
   707  				dr["event"] = "GRANT"
   708  				detectedChanges = append(detectedChanges, dr)
   709  				changeCounts.Created++
   710  				md.ChangeState["object_privileges"][objIdent] = ""
   711  			}
   712  			currentState[objIdent] = true
   713  		}
   714  	}
   715  	// check revokes - exists in old state only
   716  	if !firstRun && len(currentState) > 0 {
   717  		for objPrevRun := range md.ChangeState["object_privileges"] {
   718  			if _, ok := currentState[objPrevRun]; !ok {
   719  				splits := strings.Split(objPrevRun, "#:#")
   720  				l.WithField("role", splits[1]).
   721  					WithField("object_type", splits[0]).
   722  					WithField("object", splits[2]).
   723  					WithField("privilege_type", splits[3]).
   724  					Debug("removed object privileges detected")
   725  				revokeEntry := metrics.NewMeasurement(data.GetEpoch())
   726  				revokeEntry["object_type"] = splits[0]
   727  				revokeEntry["tag_role"] = splits[1]
   728  				revokeEntry["tag_object"] = splits[2]
   729  				revokeEntry["privilege_type"] = splits[3]
   730  				revokeEntry["event"] = "REVOKE"
   731  				detectedChanges = append(detectedChanges, revokeEntry)
   732  				changeCounts.Dropped++
   733  				delete(md.ChangeState["object_privileges"], objPrevRun)
   734  			}
   735  		}
   736  	}
   737  	l.Debugf("object privilege changes detected: %d", len(detectedChanges))
   738  	if len(detectedChanges) > 0 {
   739  		r.measurementCh <- metrics.MeasurementEnvelope{
   740  			DBName:     md.Name,
   741  			MetricName: "privilege_changes",
   742  			Data:       detectedChanges,
   743  			CustomTags: md.CustomTags,
   744  		}
   745  	}
   746  	return changeCounts
   747  }
   748  
   749  func (r *reaper) DetectConfigurationChanges(ctx context.Context, md *sources.DbConn) ChangeDetectionResults {
   750  	detectedChanges := make(metrics.Measurements, 0)
   751  	var firstRun bool
   752  	var changeCounts ChangeDetectionResults
   753  	l := log.GetLogger(ctx)
   754  	changeCounts.Target = "settings"
   755  	l.Debug("checking for configuration changes...")
   756  	if _, ok := md.ChangeState["configuration_hashes"]; !ok {
   757  		firstRun = true
   758  		md.ChangeState["configuration_hashes"] = make(map[string]string)
   759  	}
   760  	mvp, ok := metricDefs.GetMetricDef("configuration_hashes")
   761  	if !ok {
   762  		l.Error("could not get configuration_hashes sql")
   763  		return changeCounts
   764  	}
   765  	rows, err := md.Conn.Query(ctx, mvp.GetSQL(md.Version))
   766  	if err != nil {
   767  		l.Error(err)
   768  		return changeCounts
   769  	}
   770  	defer rows.Close()
   771  	var (
   772  		objIdent, objValue string
   773  		epoch              int64
   774  	)
   775  	for rows.Next() {
   776  		if rows.Scan(&epoch, &objIdent, &objValue) != nil {
   777  			return changeCounts
   778  		}
   779  		prevРash, ok := md.ChangeState["configuration_hashes"][objIdent]
   780  		ll := l.WithField("setting", objIdent)
   781  		if ok { // we have existing state
   782  			if prevРash != objValue {
   783  				ll.Warningf("settings change detected: %s = %s (prev: %s)", objIdent, objValue, prevРash)
   784  				detectedChanges = append(detectedChanges, metrics.Measurement{
   785  					metrics.EpochColumnName: epoch,
   786  					"tag_setting":           objIdent,
   787  					"value":                 objValue,
   788  					"event":                 "alter"})
   789  				md.ChangeState["configuration_hashes"][objIdent] = objValue
   790  				changeCounts.Altered++
   791  			}
   792  		} else { // check for new, delete not relevant here (pg_upgrade)
   793  			md.ChangeState["configuration_hashes"][objIdent] = objValue
   794  			if firstRun {
   795  				continue
   796  			}
   797  			ll.Debug("new setting detected")
   798  			detectedChanges = append(detectedChanges, metrics.Measurement{
   799  				metrics.EpochColumnName: epoch,
   800  				"tag_setting":           objIdent,
   801  				"value":                 objValue,
   802  				"event":                 "create"})
   803  			changeCounts.Created++
   804  		}
   805  	}
   806  	l.Debugf("configuration changes detected: %d", len(detectedChanges))
   807  	if len(detectedChanges) > 0 {
   808  		r.measurementCh <- metrics.MeasurementEnvelope{
   809  			DBName:     md.Name,
   810  			MetricName: "configuration_changes",
   811  			Data:       detectedChanges,
   812  			CustomTags: md.CustomTags,
   813  		}
   814  	}
   815  	return changeCounts
   816  }
   817  
   818  // GetInstanceUpMeasurement returns a single measurement with "instance_up" metric
   819  // used to detect if the instance is up or down
   820  func (r *reaper) GetInstanceUpMeasurement(ctx context.Context, md *sources.DbConn) (metrics.Measurements, error) {
   821  	return metrics.Measurements{
   822  		metrics.Measurement{
   823  			metrics.EpochColumnName: time.Now().UnixNano(),
   824  			"instance_up": func() int {
   825  				if md.Conn.Ping(ctx) == nil {
   826  					return 1
   827  				}
   828  				return 0
   829  			}(), // true if connection is up
   830  		},
   831  	}, nil // always return nil error for the status metric
   832  }
   833  
   834  func (r *reaper) GetObjectChangesMeasurement(ctx context.Context, md *sources.DbConn) (metrics.Measurements, error) {
   835  	md.Lock()
   836  	defer md.Unlock()
   837  	spN := r.DetectSprocChanges(ctx, md)
   838  	tblN := r.DetectTableChanges(ctx, md)
   839  	idxN := r.DetectIndexChanges(ctx, md)
   840  	cnfN := r.DetectConfigurationChanges(ctx, md)
   841  	privN := r.DetectPrivilegeChanges(ctx, md)
   842  	if spN.Total()+tblN.Total()+idxN.Total()+cnfN.Total()+privN.Total() == 0 {
   843  		return nil, nil
   844  	}
   845  	m := metrics.NewMeasurement(time.Now().UnixNano())
   846  	m["details"] = strings.Join([]string{spN.String(), tblN.String(), idxN.String(), cnfN.String(), privN.String()}, " ")
   847  	return metrics.Measurements{m}, nil
   848  }
   849