const (
metricCPULoad = "cpu_load"
metricPsutilCPU = "psutil_cpu"
metricPsutilDisk = "psutil_disk"
metricPsutilDiskIoTotal = "psutil_disk_io_total"
metricPsutilMem = "psutil_mem"
)
const (
sqlPgDirs = `select current_setting('data_directory') as dd, current_setting('log_directory') as ld`
sqlTsDirs = `select spcname::text as name, pg_catalog.pg_tablespace_location(oid) as location from pg_catalog.pg_tablespace where not spcname like any(array[E'pg\\_%'])`
)
const (
monitoredDbsDatastoreSyncIntervalSeconds = 600 // write actively monitored DBs listing to metrics store after so many seconds
monitoredDbsDatastoreSyncMetricName = "configured_dbs" // FYI - for Postgres datastore there's also the admin.all_unique_dbnames table with all recent DB unique names with some metric data
dbSizeCachingInterval = 30 * time.Minute
dbMetricJoinStr = "¤¤¤" // just some unlikely string for a DB name to avoid using maps of maps for DB+metric data
)
const (
recoPrefix = "reco_" // special handling for metrics with such prefix, data stored in RECO_METRIC_NAME
recoMetricName = "recommendations"
specialMetricChangeEvents = "change_events"
specialMetricServerLogEventCounts = "server_log_event_counts"
specialMetricInstanceUp = "instance_up"
)
var directlyFetchableOSMetrics = []string{metricPsutilCPU, metricPsutilDisk, metricPsutilDiskIoTotal, metricPsutilMem, metricCPULoad}
var hostLastKnownStatusInRecovery = make(map[string]bool) // isInRecovery
var lastSQLFetchError sync.Map
var metricDefs = NewConcurrentMetricDefs()
var metricsConfig map[string]float64 // set to host.Metrics or host.MetricsStandby (in case optional config defined and in recovery state
var monitoredDbCache map[string]*sources.SourceConn
var monitoredDbCacheLock sync.RWMutex
var prevCPULoadTimeStats cpu.TimesStat
"cache" of last CPU utilization stats for GetGoPsutilCPU to get more exact results and not having to sleep
var prevCPULoadTimeStatsLock sync.RWMutex
var prevCPULoadTimestamp time.Time
var specialMetrics = map[string]bool{recoMetricName: true, specialMetricChangeEvents: true, specialMetricServerLogEventCounts: true}
func CheckFolderExistsAndReadable(path string) bool
func DoesEmergencyTriggerfileExist(fname string) bool
func GetAllRecoMetricsForVersion() (metrics.MetricDefs, error)
func GetGoPsutilCPU(interval float64) ([]map[string]any, error)
Simulates "psutil" metric output. Assumes the result from last call as input, otherwise uses a 1s measurement
func GetGoPsutilDiskPG(DataDirs, TblspaceDirs []map[string]any) ([]map[string]any, error)
func GetGoPsutilDiskTotals() ([]map[string]any, error)
func GetGoPsutilMem() ([]map[string]any, error)
func GetLoadAvgLocal() ([]map[string]any, error)
func GetMonitoredDatabaseByUniqueName(ctx context.Context, name string) (*sources.SourceConn, error)
func GetPathUnderlyingDeviceID(path string) (uint64, error)
func GetRecommendations(ctx context.Context, dbUnique string, md *sources.SourceConn) (metrics.Measurements, error)
func IsDirectlyFetchableMetric(metric string) bool
func QueryMeasurements(ctx context.Context, dbUnique string, sql string, args ...any) (metrics.Measurements, error)
func TryCreateMetricsFetchingHelpers(ctx context.Context, md *sources.SourceConn) (err error)
Called once on daemon startup to try to create "metric fething helper" functions automatically
func TryCreateMissingExtensions(ctx context.Context, dbUnique string, extensionNames []string, existingExtensions map[string]int) []string
Called once on daemon startup if some commonly wanted extension (most notably pg_stat_statements) is missing. With newer Postgres version can even succeed if the user is not a real superuser due to some cloud-specific whitelisting or "trusted extensions" (a feature from v13). Ignores errors.
func UpdateMonitoredDBCache(data sources.SourceConns)
func goPsutilCalcCPUUtilization(probe0, probe1 cpu.TimesStat) float64
type ChangeDetectionResults struct {
Created int
Altered int
Dropped int
}
func DetectConfigurationChanges(ctx context.Context, md *sources.SourceConn, storageCh chan<- []metrics.MeasurementEnvelope, hostState map[string]map[string]string) ChangeDetectionResults
func DetectIndexChanges(ctx context.Context, md *sources.SourceConn, storageCh chan<- []metrics.MeasurementEnvelope, hostState map[string]map[string]string) ChangeDetectionResults
func DetectPrivilegeChanges(ctx context.Context, md *sources.SourceConn, storageCh chan<- []metrics.MeasurementEnvelope, hostState map[string]map[string]string) ChangeDetectionResults
func DetectSprocChanges(ctx context.Context, md *sources.SourceConn, storageCh chan<- []metrics.MeasurementEnvelope, hostState map[string]map[string]string) ChangeDetectionResults
func DetectTableChanges(ctx context.Context, md *sources.SourceConn, storageCh chan<- []metrics.MeasurementEnvelope, hostState map[string]map[string]string) ChangeDetectionResults
type ConcurrentMetricDefs struct {
*metrics.Metrics
sync.RWMutex
}
func NewConcurrentMetricDefs() *ConcurrentMetricDefs
func (cmd *ConcurrentMetricDefs) Assign(newDefs *metrics.Metrics)
func (cmd *ConcurrentMetricDefs) GetMetricDef(name string) (m metrics.Metric, ok bool)
func (cmd *ConcurrentMetricDefs) GetPresetDef(name string) (m metrics.Preset, ok bool)
func (cmd *ConcurrentMetricDefs) GetPresetMetrics(name string) (m map[string]float64)
type ExistingPartitionInfo struct {
StartTime time.Time
EndTime time.Time
}
type InstanceMetricCache struct {
cache map[string](metrics.Measurements) // [dbUnique+metric]lastly_fetched_data
sync.RWMutex
}
func NewInstanceMetricCache() *InstanceMetricCache
func (imc *InstanceMetricCache) Get(key string, age time.Duration) metrics.Measurements
func (imc *InstanceMetricCache) Put(key string, data metrics.Measurements)
Reaper is the struct that responsible for fetching metrics measurements from the sources and storing them to the sinks
type Reaper struct {
*cmdopts.Options
ready atomic.Bool
measurementCh chan []metrics.MeasurementEnvelope
measurementCache *InstanceMetricCache
logger log.Logger
monitoredSources sources.SourceConns
prevLoopMonitoredDBs sources.SourceConns
cancelFuncs map[string]context.CancelFunc
}
func NewReaper(ctx context.Context, opts *cmdopts.Options) (r *Reaper)
NewReaper creates a new Reaper instance
func (r *Reaper) AddSysinfoToMeasurements(data metrics.Measurements, md *sources.SourceConn)
func (r *Reaper) CheckForPGObjectChangesAndStore(ctx context.Context, dbUnique string, md *sources.SourceConn, hostState map[string]map[string]string)
func (r *Reaper) CloseResourcesForRemovedMonitoredDBs(shutDownDueToRoleChange map[string]bool)
func (r *Reaper) CreateSourceHelpers(ctx context.Context, srcL log.Logger, monitoredSource *sources.SourceConn)
CreateSourceHelpers creates the extensions and metric helpers for the monitored source
func (r *Reaper) FetchMetric(ctx context.Context, md *sources.SourceConn, metricName string, hostState map[string]map[string]string) (_ *metrics.MeasurementEnvelope, err error)
func (r *Reaper) FetchStatsDirectlyFromOS(ctx context.Context, md *sources.SourceConn, metricName string) (*metrics.MeasurementEnvelope, error)
func (r *Reaper) LoadMetrics() (err error)
LoadMetrics loads metric definitions from the reader
func (r *Reaper) LoadSources() (err error)
LoadSources loads sources from the reader
func (r *Reaper) Ready() bool
Ready() returns true if the service is healthy and operating correctly
func (r *Reaper) Reap(ctx context.Context)
Reap() starts the main monitoring loop. It is responsible for fetching metrics measurements from the sources and storing them to the sinks. It also manages the lifecycle of the metric gatherers. In case of a source or metric definition change, it will start or stop the gatherers accordingly.
func (r *Reaper) ShutdownOldWorkers(ctx context.Context, hostsToShutDownDueToRoleChange map[string]bool)
func (r *Reaper) WriteMeasurements(ctx context.Context)
WriteMeasurements() writes the metrics to the sinks
func (r *Reaper) WriteMonitoredSources(ctx context.Context)
WriteMonitoredSources writes actively monitored DBs listing to sinks every monitoredDbsDatastoreSyncIntervalSeconds (default 10min)
func (r *Reaper) reapMetricMeasurements(ctx context.Context, md *sources.SourceConn, metricName string)
metrics.ControlMessage notifies of shutdown + interval change