[*] fix hard coded epoch_ns measurement field name (#712)

Author: pashagolub

internal/metrics/logparse.go

@@ -85,7 +85,7 @@ func getFileWithNextModTimestamp(logsGlobPath, currentFile string) (string, erro
 
 // 1. add zero counts for severity levels that didn't have any occurrences in the log
 func eventCountsToMetricStoreMessages(eventCounts, eventCountsTotal map[string]int64, mdb *sources.SourceConn) []MeasurementEnvelope {
-	allSeverityCounts := make(Measurement)
+	allSeverityCounts := NewMeasurement(time.Now().UnixNano())
 	for _, s := range PgSeverities {
 		parsedCount, ok := eventCounts[s]
 		if ok {
@@ -100,7 +100,6 @@ func eventCountsToMetricStoreMessages(eventCounts, eventCountsTotal map[string]i
 			allSeverityCounts[strings.ToLower(s)+"_total"] = 0
 		}
 	}
-	allSeverityCounts["epoch_ns"] = time.Now().UnixNano()
 	return []MeasurementEnvelope{{
 		DBName:     mdb.Name,
 		SourceType: string(mdb.Kind),

internal/metrics/psutil/psutil.go

@@ -8,6 +8,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/cybertec-postgresql/pgwatch/v3/internal/metrics"
 	"github.com/shirou/gopsutil/v4/cpu"
 	"github.com/shirou/gopsutil/v4/disk"
 	"github.com/shirou/gopsutil/v4/load"
@@ -60,8 +61,7 @@ func GetGoPsutilCPU(interval time.Duration) ([]map[string]any, error) {
 		return nil, err
 	}
 
-	retMap := make(map[string]any)
-	retMap["epoch_ns"] = time.Now().UnixNano()
+	retMap := metrics.NewMeasurement(time.Now().UnixNano())
 	retMap["cpu_utilization"] = math.Round(100*goPsutilCalcCPUUtilization(prevTimeStat, curCallStats[0])) / 100
 	retMap["load_1m_norm"] = math.Round(100*la.Load1/float64(cpus)) / 100
 	retMap["load_1m"] = math.Round(100*la.Load1) / 100
@@ -83,8 +83,7 @@ func GetGoPsutilMem() ([]map[string]any, error) {
 		return nil, err
 	}
 
-	retMap := make(map[string]any)
-	retMap["epoch_ns"] = time.Now().UnixNano()
+	retMap := metrics.NewMeasurement(time.Now().UnixNano())
 	retMap["total"] = int64(vm.Total)
 	retMap["used"] = int64(vm.Used)
 	retMap["free"] = int64(vm.Free)
@@ -105,10 +104,9 @@ func GetGoPsutilDiskTotals() ([]map[string]any, error) {
 		return nil, err
 	}
 
-	retMap := make(map[string]any)
+	retMap := metrics.NewMeasurement(time.Now().UnixNano())
 	var readBytes, writeBytes, reads, writes float64
 
-	retMap["epoch_ns"] = time.Now().UnixNano()
 	for _, v := range d { // summarize all disk devices
 		readBytes += float64(v.ReadBytes) // datatype float is just an oversight in the original psutil helper
 		// but can't change it without causing problems on storage level (InfluxDB)
@@ -130,8 +128,7 @@ func GetLoadAvgLocal() ([]map[string]any, error) {
 		return nil, err
 	}
 
-	row := make(map[string]any)
-	row["epoch_ns"] = time.Now().UnixNano()
+	row := metrics.NewMeasurement(time.Now().UnixNano())
 	row["load_1min"] = la.Load1
 	row["load_5min"] = la.Load5
 	row["load_15min"] = la.Load15
@@ -157,8 +154,7 @@ func GetGoPsutilDiskPG(DataDirs, TblspaceDirs []map[string]any) ([]map[string]an
 
 	retRows := make([]map[string]any, 0)
 	epochNs := time.Now().UnixNano()
-	dd := make(map[string]any)
-	dd["epoch_ns"] = epochNs
+	dd := metrics.NewMeasurement(epochNs)
 	dd["tag_dir_or_tablespace"] = "data_directory"
 	dd["tag_path"] = dataDirPath
 	dd["total"] = float64(ddUsage.Total)
@@ -182,13 +178,12 @@ func GetGoPsutilDiskPG(DataDirs, TblspaceDirs []map[string]any) ([]map[string]an
 			return nil, err
 		}
 		if ldDevice != ddDevice { // no point to report same data in case of single folder configuration
-			ld := make(map[string]any)
+			ld := metrics.NewMeasurement(epochNs)
 			ldUsage, err := disk.Usage(logDirPath)
 			if err != nil {
 				return nil, err
 			}
 
-			ld["epoch_ns"] = epochNs
 			ld["tag_dir_or_tablespace"] = "log_directory"
 			ld["tag_path"] = logDirPath
 			ld["total"] = float64(ldUsage.Total)
@@ -218,8 +213,7 @@ func GetGoPsutilDiskPG(DataDirs, TblspaceDirs []map[string]any) ([]map[string]an
 				return nil, err
 			}
 
-			wd := make(map[string]any)
-			wd["epoch_ns"] = epochNs
+			wd := metrics.NewMeasurement(epochNs)
 			wd["tag_dir_or_tablespace"] = "pg_wal"
 			wd["tag_path"] = walDirPath
 			wd["total"] = float64(walUsage.Total)
@@ -248,8 +242,7 @@ func GetGoPsutilDiskPG(DataDirs, TblspaceDirs []map[string]any) ([]map[string]an
 			if err != nil {
 				return nil, err
 			}
-			ts := make(map[string]any)
-			ts["epoch_ns"] = epochNs
+			ts := metrics.NewMeasurement(epochNs)
 			ts["tag_dir_or_tablespace"] = tsName
 			ts["tag_path"] = tsPath
 			ts["total"] = float64(tsUsage.Total)

internal/metrics/psutil/psutil_test.go

@@ -4,6 +4,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cybertec-postgresql/pgwatch/v3/internal/metrics"
+
 	"github.com/stretchr/testify/assert"
 	"golang.org/x/exp/maps"
 )
@@ -17,7 +19,7 @@ func TestGetGoPsutilCPU(t *testing.T) {
 	a.NotEmpty(result)
 
 	// Check if the result contains the expected keys
-	expectedKeys := []string{"epoch_ns", "cpu_utilization", "load_1m_norm", "load_1m", "load_5m_norm", "load_5m", "user", "system", "idle", "iowait", "irqs", "other"}
+	expectedKeys := []string{metrics.EpochColumnName, "cpu_utilization", "load_1m_norm", "load_1m", "load_5m_norm", "load_5m", "user", "system", "idle", "iowait", "irqs", "other"}
 	resultKeys := maps.Keys(result[0])
 	a.ElementsMatch(resultKeys, expectedKeys)
 
@@ -36,7 +38,7 @@ func TestGetGoPsutilMem(t *testing.T) {
 	a.NotEmpty(result)
 
 	// Check if the result contains the expected keys
-	expectedKeys := []string{"epoch_ns", "total", "used", "free", "buff_cache", "available", "percent", "swap_total", "swap_used", "swap_free", "swap_percent"}
+	expectedKeys := []string{metrics.EpochColumnName, "total", "used", "free", "buff_cache", "available", "percent", "swap_total", "swap_used", "swap_free", "swap_percent"}
 	resultKeys := maps.Keys(result[0])
 	a.ElementsMatch(resultKeys, expectedKeys)
 }
@@ -52,7 +54,7 @@ func TestGetGoPsutilDiskTotals(t *testing.T) {
 	a.NotEmpty(result)
 
 	// Check if the result contains the expected keys
-	expectedKeys := []string{"epoch_ns", "read_bytes", "write_bytes", "read_count", "write_count"}
+	expectedKeys := []string{metrics.EpochColumnName, "read_bytes", "write_bytes", "read_count", "write_count"}
 	resultKeys := maps.Keys(result[0])
 	a.ElementsMatch(resultKeys, expectedKeys)
 }
@@ -66,7 +68,7 @@ func TestGetLoadAvgLocal(t *testing.T) {
 	a.NotEmpty(result)
 
 	// Check if the result contains the expected keys
-	expectedKeys := []string{"epoch_ns", "load_1min", "load_5min", "load_15min"}
+	expectedKeys := []string{metrics.EpochColumnName, "load_1min", "load_5min", "load_15min"}
 	resultKeys := maps.Keys(result[0])
 	a.ElementsMatch(resultKeys, expectedKeys)
 }

internal/metrics/types.go

@@ -1,5 +1,7 @@
 package metrics
 
+import "time"
+
 type (
 	ExtensionInfo struct {
 		ExtName       string `yaml:"ext_name"`
@@ -65,9 +67,37 @@ type Preset struct {
 	Metrics     map[string]float64
 }
 
+const (
+	EpochColumnName string = "epoch_ns" // this column (epoch in nanoseconds) is expected in every metric query
+	TagPrefix       string = "tag_"
+)
+
 type Measurement map[string]any
+
+func NewMeasurement(epoch int64) Measurement {
+	m := make(Measurement)
+	m[EpochColumnName] = epoch
+	return m
+}
+
+func (m Measurement) GetEpoch() int64 {
+	if v, ok := m[EpochColumnName]; ok {
+		if epoch, ok := v.(int64); ok {
+			return epoch
+		}
+	}
+	return time.Now().UnixNano()
+}
+
 type Measurements []map[string]any
 
+func (m Measurements) GetEpoch() int64 {
+	if len(m) == 0 {
+		return time.Now().UnixNano()
+	}
+	return Measurement(m[0]).GetEpoch()
+}
+
 type MeasurementEnvelope struct {
 	DBName           string
 	SourceType       string

internal/reaper/cache.go

@@ -1,14 +1,13 @@
 package reaper
 
 import (
-	"context"
 	"fmt"
+	"maps"
 	"sync"
 	"time"
 
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/metrics"
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/sources"
-	"github.com/sirupsen/logrus"
 )
 
 var monitoredDbCache map[string]*sources.SourceConn
@@ -53,53 +52,10 @@ func GetMonitoredDatabaseByUniqueName(name string) (*sources.SourceConn, error)
 	return md, nil
 }
 
-// LoadMetrics loads metric definitions from the reader
-func (r *Reaper) LoadMetrics() (err error) {
-	var newDefs *metrics.Metrics
-	if newDefs, err = r.MetricsReaderWriter.GetMetrics(); err != nil {
-		return
-	}
-	metricDefs.Assign(newDefs)
-	r.logger.
-		WithField("metrics", len(newDefs.MetricDefs)).
-		WithField("presets", len(newDefs.PresetDefs)).
-		Log(func() logrus.Level {
-			if len(newDefs.PresetDefs)*len(newDefs.MetricDefs) == 0 {
-				return logrus.WarnLevel
-			}
-			return logrus.InfoLevel
-		}(), "metrics and presets refreshed")
-	return
-}
-
-// LoadSources loads sources from the reader
-func (r *Reaper) LoadSources() (err error) {
-	if DoesEmergencyTriggerfileExist(r.Metrics.EmergencyPauseTriggerfile) {
-		r.logger.Warningf("Emergency pause triggerfile detected at %s, ignoring currently configured DBs", r.Metrics.EmergencyPauseTriggerfile)
-		monitoredSources = make([]*sources.SourceConn, 0)
-		return nil
-	}
-	if monitoredSources, err = monitoredSources.SyncFromReader(r.SourcesReaderWriter); err != nil {
-		return err
-	}
-	r.logger.WithField("sources", len(monitoredSources)).Info("sources refreshed")
-	return nil
-}
-
-// WriteMeasurements() writes the metrics to the sinks
-func (r *Reaper) WriteMeasurements(ctx context.Context) {
-	var err error
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case msg := <-r.measurementCh:
-			if err = r.SinksWriter.Write(msg); err != nil {
-				r.logger.Error(err)
-			}
-		}
-	}
-}
+var instanceMetricCache = make(map[string](metrics.Measurements)) // [dbUnique+metric]lastly_fetched_data
+var instanceMetricCacheLock = sync.RWMutex{}
+var instanceMetricCacheTimestamp = make(map[string]time.Time) // [dbUnique+metric]last_fetch_time
+var instanceMetricCacheTimestampLock = sync.RWMutex{}
 
 func GetFromInstanceCacheIfNotOlderThanSeconds(msg MetricFetchConfig, maxAgeSeconds int64) metrics.Measurements {
 	var clonedData metrics.Measurements
@@ -121,3 +77,34 @@ func GetFromInstanceCacheIfNotOlderThanSeconds(msg MetricFetchConfig, maxAgeSeco
 
 	return clonedData
 }
+
+func IsCacheableMetric(msg MetricFetchConfig, mvp metrics.Metric) bool {
+	switch msg.Source {
+	case sources.SourcePostgresContinuous, sources.SourcePatroniContinuous:
+		return false
+	default:
+		return mvp.IsInstanceLevel
+	}
+}
+
+func PutToInstanceCache(msg MetricFetchConfig, data metrics.Measurements) {
+	if len(data) == 0 {
+		return
+	}
+	dataCopy := deepCopyMetricData(data)
+	instanceMetricCacheLock.Lock()
+	instanceMetricCache[msg.DBUniqueNameOrig+msg.MetricName] = dataCopy
+	instanceMetricCacheLock.Unlock()
+
+	instanceMetricCacheTimestampLock.Lock()
+	instanceMetricCacheTimestamp[msg.DBUniqueNameOrig+msg.MetricName] = time.Now()
+	instanceMetricCacheTimestampLock.Unlock()
+}
+
+func deepCopyMetricData(data metrics.Measurements) metrics.Measurements {
+	newData := make(metrics.Measurements, len(data))
+	for i, dr := range data {
+		newData[i] = maps.Clone(dr)
+	}
+	return newData
+}

internal/reaper/database.go

@@ -289,15 +289,10 @@ func DetectSprocChanges(ctx context.Context, dbUnique string, vme MonitoredDatab
 			if !ok {
 				splits := strings.Split(sprocIdent, dbMetricJoinStr)
 				log.GetLogger(ctx).Info("detected delete of sproc:", splits[0], ", oid:", splits[1])
-				influxEntry := make(metrics.Measurement)
+				influxEntry := metrics.NewMeasurement(data.GetEpoch())
 				influxEntry["event"] = "drop"
 				influxEntry["tag_sproc"] = splits[0]
 				influxEntry["tag_oid"] = splits[1]
-				if len(data) > 0 {
-					influxEntry["epoch_ns"] = data[0]["epoch_ns"]
-				} else {
-					influxEntry["epoch_ns"] = time.Now().UnixNano()
-				}
 				detectedChanges = append(detectedChanges, influxEntry)
 				deletedSProcs = append(deletedSProcs, sprocIdent)
 				changeCounts.Dropped++
@@ -373,14 +368,9 @@ func DetectTableChanges(ctx context.Context, dbUnique string, vme MonitoredDatab
 			_, ok := currentTableMap[table]
 			if !ok {
 				log.GetLogger(ctx).Info("detected drop of table:", table)
-				influxEntry := make(metrics.Measurement)
+				influxEntry := metrics.NewMeasurement(data.GetEpoch())
 				influxEntry["event"] = "drop"
 				influxEntry["tag_table"] = table
-				if len(data) > 0 {
-					influxEntry["epoch_ns"] = data[0]["epoch_ns"]
-				} else {
-					influxEntry["epoch_ns"] = time.Now().UnixNano()
-				}
 				detectedChanges = append(detectedChanges, influxEntry)
 				deletedTables = append(deletedTables, table)
 				changeCounts.Dropped++
@@ -456,14 +446,9 @@ func DetectIndexChanges(ctx context.Context, dbUnique string, vme MonitoredDatab
 			_, ok := currentIndexMap[indexName]
 			if !ok {
 				log.GetLogger(ctx).Info("detected drop of index_name:", indexName)
-				influxEntry := make(metrics.Measurement)
+				influxEntry := metrics.NewMeasurement(data.GetEpoch())
 				influxEntry["event"] = "drop"
 				influxEntry["tag_index"] = indexName
-				if len(data) > 0 {
-					influxEntry["epoch_ns"] = data[0]["epoch_ns"]
-				} else {
-					influxEntry["epoch_ns"] = time.Now().UnixNano()
-				}
 				detectedChanges = append(detectedChanges, influxEntry)
 				deletedIndexes = append(deletedIndexes, indexName)
 				changeCounts.Dropped++
@@ -531,12 +516,7 @@ func DetectPrivilegeChanges(ctx context.Context, dbUnique string, vme MonitoredD
 				splits := strings.Split(objPrevRun, "#:#")
 				log.GetLogger(ctx).Infof("[%s][%s] detected removed object privileges: role=%s, object_type=%s, object=%s, privilege_type=%s",
 					dbUnique, specialMetricChangeEvents, splits[1], splits[0], splits[2], splits[3])
-				revokeEntry := make(metrics.Measurement)
-				if epochNs, ok := data[0]["epoch_ns"]; ok {
-					revokeEntry["epoch_ns"] = epochNs
-				} else {
-					revokeEntry["epoch_ns"] = time.Now().UnixNano()
-				}
+				revokeEntry := metrics.NewMeasurement(data.GetEpoch())
 				revokeEntry["object_type"] = splits[0]
 				revokeEntry["tag_role"] = splits[1]
 				revokeEntry["tag_object"] = splits[2]
@@ -658,9 +638,8 @@ func (r *Reaper) CheckForPGObjectChangesAndStore(ctx context.Context, dbUnique s
 		message = "Detected changes for \"" + dbUnique + "\" [Created/Altered/Dropped]:" + message
 		log.GetLogger(ctx).Info(message)
 		detectedChangesSummary := make(metrics.Measurements, 0)
-		influxEntry := make(metrics.Measurement)
+		influxEntry := metrics.NewMeasurement(time.Now().UnixNano())
 		influxEntry["details"] = message
-		influxEntry["epoch_ns"] = time.Now().UnixNano()
 		detectedChangesSummary = append(detectedChangesSummary, influxEntry)
 		md, _ := GetMonitoredDatabaseByUniqueName(dbUnique)
 		storageCh <- []metrics.MeasurementEnvelope{{DBName: dbUnique,
@@ -689,8 +668,7 @@ func FetchMetricsPgpool(ctx context.Context, msg MetricFetchConfig, vme Monitore
 			}
 
 			for _, row := range data {
-				retRow := make(metrics.Measurement)
-				retRow[epochColumnName] = epochNs
+				retRow := metrics.NewMeasurement(epochNs)
 				for k, v := range row {
 					vs := string(v.([]byte))
 					// need 1 tag so that Influx would not merge rows