[*] rename sources.MonitoredDatabase to SourceConn (#671)

Also move some connection specific functions from reaper to
SourceConn methods, e.g. TryDiscoverExecutionEnv(), DoesFunctionExists()

Author: pashagolub

internal/cmdopts/cmdsource.go

@@ -54,7 +54,7 @@ func (cmd *SourcePingCommand) Execute(args []string) error {
 		case sources.SourcePostgresContinuous:
 			_, e = sources.ResolveDatabasesFromPostgres(s)
 		default:
-			mdb := &sources.MonitoredDatabase{Source: s}
+			mdb := &sources.SourceConn{Source: s}
 			e = mdb.Ping(context.Background())
 		}
 		if e != nil {

internal/metrics/logparse.go

@@ -84,7 +84,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.MonitoredDatabase) []MeasurementEnvelope {
+func eventCountsToMetricStoreMessages(eventCounts, eventCountsTotal map[string]int64, mdb *sources.SourceConn) []MeasurementEnvelope {
 	allSeverityCounts := make(Measurement)
 	for _, s := range PgSeverities {
 		parsedCount, ok := eventCounts[s]
@@ -110,7 +110,7 @@ func eventCountsToMetricStoreMessages(eventCounts, eventCountsTotal map[string]i
 	}}
 }
 
-func ParseLogs(ctx context.Context, mdb *sources.MonitoredDatabase, realDbname, metricName string, configMap map[string]float64, storeCh chan<- []MeasurementEnvelope) {
+func ParseLogs(ctx context.Context, mdb *sources.SourceConn, realDbname, metricName string, configMap map[string]float64, storeCh chan<- []MeasurementEnvelope) {
 
 	var latest, previous, serverMessagesLang string
 	var latestHandle *os.File

internal/reaper/cache.go

@@ -12,7 +12,7 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
-var monitoredDbCache map[string]*sources.MonitoredDatabase
+var monitoredDbCache map[string]*sources.SourceConn
 var monitoredDbCacheLock sync.RWMutex
 var MonitoredDatabasesSettings = make(map[string]MonitoredDatabaseSettings)
 var MonitoredDatabasesSettingsLock = sync.RWMutex{}
@@ -25,8 +25,8 @@ var lastDBSizeMB = make(map[string]int64)
 var lastDBSizeFetchTime = make(map[string]time.Time) // cached for DB_SIZE_CACHING_INTERVAL
 var lastDBSizeCheckLock sync.RWMutex
 
-var prevLoopMonitoredDBs sources.MonitoredDatabases // to be able to detect DBs removed from config
-var undersizedDBs = make(map[string]bool)           // DBs below the --min-db-size-mb limit, if set
+var prevLoopMonitoredDBs sources.SourceConns // to be able to detect DBs removed from config
+var undersizedDBs = make(map[string]bool)    // DBs below the --min-db-size-mb limit, if set
 var undersizedDBsLock = sync.RWMutex{}
 var recoveryIgnoredDBs = make(map[string]bool) // DBs in recovery state and OnlyIfMaster specified in config
 var recoveryIgnoredDBsLock = sync.RWMutex{}
@@ -35,16 +35,16 @@ var hostMetricIntervalMap = make(map[string]float64) // [db1_metric] = 30
 
 var lastSQLFetchError sync.Map
 
-func InitPGVersionInfoFetchingLockIfNil(md *sources.MonitoredDatabase) {
+func InitPGVersionInfoFetchingLockIfNil(md *sources.SourceConn) {
 	MonitoredDatabasesSettingsLock.Lock()
 	if _, ok := MonitoredDatabasesSettingsGetLock[md.Name]; !ok {
 		MonitoredDatabasesSettingsGetLock[md.Name] = &sync.RWMutex{}
 	}
 	MonitoredDatabasesSettingsLock.Unlock()
 }
 
-func UpdateMonitoredDBCache(data sources.MonitoredDatabases) {
-	monitoredDbCacheNew := make(map[string]*sources.MonitoredDatabase)
+func UpdateMonitoredDBCache(data sources.SourceConns) {
+	monitoredDbCacheNew := make(map[string]*sources.SourceConn)
 	for _, row := range data {
 		monitoredDbCacheNew[row.Name] = row
 	}
@@ -53,7 +53,7 @@ func UpdateMonitoredDBCache(data sources.MonitoredDatabases) {
 	monitoredDbCacheLock.Unlock()
 }
 
-func GetMonitoredDatabaseByUniqueName(name string) (*sources.MonitoredDatabase, error) {
+func GetMonitoredDatabaseByUniqueName(name string) (*sources.SourceConn, error) {
 	monitoredDbCacheLock.RLock()
 	defer monitoredDbCacheLock.RUnlock()
 	md, exists := monitoredDbCache[name]

internal/reaper/database.go

@@ -28,16 +28,9 @@ func DBExecRead(ctx context.Context, conn db.PgxIface, sql string, args ...any)
 	return nil, err
 }
 
-func GetConnByUniqueName(dbUnique string) db.PgxIface {
-	if md, err := GetMonitoredDatabaseByUniqueName(dbUnique); err == nil {
-		return md.Conn
-	}
-	return nil
-}
-
 func DBExecReadByDbUniqueName(ctx context.Context, dbUnique string, sql string, args ...any) (metrics.Measurements, error) {
 	var conn db.PgxIface
-	var md *sources.MonitoredDatabase
+	var md *sources.SourceConn
 	var err error
 	var tx pgx.Tx
 	if strings.TrimSpace(sql) == "" {
@@ -112,35 +105,6 @@ func DBGetSizeMB(ctx context.Context, dbUnique string) (int64, error) {
 	return lastDBSize, nil
 }
 
-func TryDiscoverExecutionEnv(ctx context.Context, dbUnique string) (execEnv string) {
-	sql := `select /* pgwatch_generated */
-	case
-	  when exists (select * from pg_settings where name = 'pg_qs.host_database' and setting = 'azure_sys') and version() ~* 'compiled by Visual C' then 'AZURE_SINGLE'
-	  when exists (select * from pg_settings where name = 'pg_qs.host_database' and setting = 'azure_sys') and version() ~* 'compiled by gcc' then 'AZURE_FLEXIBLE'
-	  when exists (select * from pg_settings where name = 'cloudsql.supported_extensions') then 'GOOGLE'
-	else
-	  'UNKNOWN'
-	end as exec_env`
-	_ = GetConnByUniqueName(dbUnique).QueryRow(ctx, sql).Scan(&execEnv)
-	return
-}
-
-func GetDBTotalApproxSize(ctx context.Context, dbUnique string) (int64, error) {
-	sqlApproxDBSize := `
-	select /* pgwatch_generated */
-		current_setting('block_size')::int8 * sum(relpages) as db_size_approx
-	from
-		pg_class c
-	where	/* works only for v9.1+*/
-		c.relpersistence != 't';
-	`
-	data, err := DBExecReadByDbUniqueName(ctx, dbUnique, sqlApproxDBSize)
-	if err != nil {
-		return 0, err
-	}
-	return data[0]["db_size_approx"].(int64), nil
-}
-
 // VersionToInt parses a given version and returns an integer  or
 // an error if unable to parse the version. Only parses valid semantic versions.
 // Performs checking that can find errors within the version.
@@ -158,7 +122,7 @@ func VersionToInt(version string) (v int) {
 
 var rBouncerAndPgpoolVerMatch = regexp.MustCompile(`\d+\.+\d+`) // extract $major.minor from "4.1.2 (karasukiboshi)" or "PgBouncer 1.12.0"
 
-func GetMonitoredDatabaseSettings(ctx context.Context, md *sources.MonitoredDatabase, noCache bool) (MonitoredDatabaseSettings, error) {
+func GetMonitoredDatabaseSettings(ctx context.Context, md *sources.SourceConn, noCache bool) (MonitoredDatabaseSettings, error) {
 	var dbSettings MonitoredDatabaseSettings
 	var dbNewSettings MonitoredDatabaseSettings
 	var ok bool
@@ -233,12 +197,12 @@ FROM
 			dbNewSettings.ExecEnv = dbSettings.ExecEnv // carry over as not likely to change ever
 		} else {
 			l.Debugf("determining the execution env...")
-			dbNewSettings.ExecEnv = TryDiscoverExecutionEnv(ctx, md.Name)
+			dbNewSettings.ExecEnv = md.DiscoverPlatform(ctx)
 		}
 
 		// to work around poor Azure Single Server FS functions performance for some metrics + the --min-db-size-mb filter
 		if dbNewSettings.ExecEnv == execEnvAzureSingle {
-			if approxSize, err := GetDBTotalApproxSize(ctx, md.Name); err == nil {
+			if approxSize, err := md.GetApproxSize(ctx); err == nil {
 				dbNewSettings.ApproxDBSizeB = approxSize
 			} else {
 				dbNewSettings.ApproxDBSizeB = dbSettings.ApproxDBSizeB
@@ -811,21 +775,6 @@ func FetchMetricsPgpool(ctx context.Context, msg MetricFetchConfig, vme Monitore
 	return retData, nil
 }
 
-func DoesFunctionExists(ctx context.Context, dbUnique, functionName string) bool {
-	log.GetLogger(ctx).Debug("Checking for function existence", dbUnique, functionName)
-	sql := fmt.Sprintf("select /* pgwatch_generated */ 1 from pg_proc join pg_namespace n on pronamespace = n.oid where proname = '%s' and n.nspname = 'public'", functionName)
-	data, err := DBExecReadByDbUniqueName(ctx, dbUnique, sql)
-	if err != nil {
-		log.GetLogger(ctx).Error("Failed to check for function existence", dbUnique, functionName, err)
-		return false
-	}
-	if len(data) > 0 {
-		log.GetLogger(ctx).Debugf("Function %s exists on %s", functionName, dbUnique)
-		return true
-	}
-	return false
-}
-
 // 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.
@@ -866,7 +815,7 @@ func TryCreateMissingExtensions(ctx context.Context, dbUnique string, extensionN
 }
 
 // Called once on daemon startup to try to create "metric fething helper" functions automatically
-func TryCreateMetricsFetchingHelpers(ctx context.Context, md *sources.MonitoredDatabase) (err error) {
+func TryCreateMetricsFetchingHelpers(ctx context.Context, md *sources.SourceConn) (err error) {
 	metricConfig := func() map[string]float64 {
 		if len(md.Metrics) > 0 {
 			return md.Metrics
@@ -919,7 +868,7 @@ func GetGoPsutilDiskPG(ctx context.Context, dbUnique string) (metrics.Measuremen
 	return psutil.GetGoPsutilDiskPG(data, dataTblsp)
 }
 
-func CloseResourcesForRemovedMonitoredDBs(metricsWriter sinks.Writer, currentDBs, prevLoopDBs sources.MonitoredDatabases, shutDownDueToRoleChange map[string]bool) {
+func CloseResourcesForRemovedMonitoredDBs(metricsWriter sinks.Writer, currentDBs, prevLoopDBs sources.SourceConns, shutDownDueToRoleChange map[string]bool) {
 	var curDBsMap = make(map[string]bool)
 
 	for _, curDB := range currentDBs {

internal/reaper/reaper.go

@@ -16,7 +16,7 @@ import (
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/sources"
 )
 
-var monitoredSources = make(sources.MonitoredDatabases, 0)
+var monitoredSources = make(sources.SourceConns, 0)
 var hostLastKnownStatusInRecovery = make(map[string]bool) // isInRecovery
 var metricConfig map[string]float64                       // set to host.Metrics or host.MetricsStandby (in case optional config defined and in recovery state
 var metricDefinitionMap *metrics.Metrics = &metrics.Metrics{}
@@ -68,7 +68,7 @@ func (r *Reaper) Reap(ctx context.Context) (err error) {
 
 		if DoesEmergencyTriggerfileExist(r.Metrics.EmergencyPauseTriggerfile) {
 			logger.Warningf("Emergency pause triggerfile detected at %s, ignoring currently configured DBs", r.Metrics.EmergencyPauseTriggerfile)
-			monitoredSources = make([]*sources.MonitoredDatabase, 0)
+			monitoredSources = make([]*sources.SourceConn, 0)
 		}
 
 		UpdateMonitoredDBCache(monitoredSources)
@@ -250,7 +250,7 @@ func (r *Reaper) Reap(ctx context.Context) (err error) {
 		logger.Debug("checking if any workers need to be shut down...")
 		for dbMetric, cancelFunc := range cancelFuncs {
 			var currentMetricConfig map[string]float64
-			var dbInfo *sources.MonitoredDatabase
+			var dbInfo *sources.SourceConn
 			var ok, dbRemovedFromConfig bool
 			singleMetricDisabled := false
 			splits := strings.Split(dbMetric, dbMetricJoinStr)
@@ -322,7 +322,7 @@ func (r *Reaper) Reap(ctx context.Context) (err error) {
 
 // metrics.ControlMessage notifies of shutdown + interval change
 func (r *Reaper) reapMetricMeasurements(ctx context.Context,
-	mdb *sources.MonitoredDatabase,
+	mdb *sources.SourceConn,
 	metricName string,
 	configMap map[string]float64) {
 
@@ -445,7 +445,7 @@ func StoreMetrics(metrics []metrics.MeasurementEnvelope, storageCh chan<- []metr
 	return 0, nil
 }
 
-func SyncMonitoredDBsToDatastore(ctx context.Context, monitoredDbs []*sources.MonitoredDatabase, persistenceChannel chan []metrics.MeasurementEnvelope) {
+func SyncMonitoredDBsToDatastore(ctx context.Context, monitoredDbs []*sources.SourceConn, persistenceChannel chan []metrics.MeasurementEnvelope) {
 	if len(monitoredDbs) > 0 {
 		msms := make([]metrics.MeasurementEnvelope, len(monitoredDbs))
 		now := time.Now()
@@ -547,7 +547,7 @@ func FetchMetrics(ctx context.Context,
 	var err error
 	var sql string
 	var data, cachedData metrics.Measurements
-	var md *sources.MonitoredDatabase
+	var md *sources.SourceConn
 	var fromCache, isCacheable bool
 
 	md, err = GetMonitoredDatabaseByUniqueName(msg.DBUniqueName)