[!] add Measurements group and use URIs for --sink options (#368)

* split Metrics into Metrics & Measurements
* move options to appropriate groups
* deprecate `UseConnPooling` option
* improve Postgres writer log entries

Author: pashagolub

src/config/cmdoptions.go

@@ -33,15 +33,14 @@ func InitSQLConnPoolForMonitoredDBIfNil(md sources.MonitoredDatabase) error {
 	}
 
 	conn, err := db.New(mainContext, md.ConnStr, func(conf *pgxpool.Config) error {
-		conf.MaxConns = int32(opts.MaxParallelConnectionsPerDb)
+		conf.MaxConns = int32(opts.Sources.MaxParallelConnectionsPerDb)
 		return nil
 	})
 	if err != nil {
 		return err
 	}
 
 	monitoredDbConnCache[md.DBUniqueName] = conn
-	logger.Debugf("[%s] Connection pool initialized with max %d parallel connections. Conn pooling: %v", md.DBUniqueName, opts.MaxParallelConnectionsPerDb, opts.UseConnPooling)
 
 	return nil
 }
@@ -57,13 +56,11 @@ func CloseOrLimitSQLConnPoolForMonitoredDBIfAny(dbUnique string) {
 
 	if IsDBUndersized(dbUnique) || IsDBIgnoredBasedOnRecoveryState(dbUnique) {
 
-		if opts.UseConnPooling {
-			s := conn.Stat()
-			if s.TotalConns() > 1 {
-				logger.Debugf("[%s] Limiting SQL connection pool to max 1 connection due to dormant state ...", dbUnique)
-				// conn.SetMaxIdleConns(1)
-				// conn.SetMaxOpenConns(1)
-			}
+		s := conn.Stat()
+		if s.TotalConns() > 1 {
+			logger.Debugf("[%s] Limiting SQL connection pool to max 1 connection due to dormant state ...", dbUnique)
+			// conn.SetMaxIdleConns(1)
+			// conn.SetMaxOpenConns(1)
 		}
 
 	} else { // removed from config
@@ -281,7 +278,7 @@ func DBGetPGVersion(ctx context.Context, dbUnique string, srcType sources.Kind,
 		verNew.IsInRecovery = data[0]["pg_is_in_recovery"].(bool)
 		verNew.RealDbname = data[0]["current_database"].(string)
 
-		if verNew.Version > VersionToInt("10.0") && opts.Metric.SystemIdentifierField > "" {
+		if verNew.Version > VersionToInt("10.0") && opts.Measurements.SystemIdentifierField > "" {
 			logger.Debugf("[%s] determining system identifier version (pg ver: %v)", dbUnique, verNew.VersionStr)
 			data, err := DBExecReadByDbUniqueName(ctx, dbUnique, sqlSysid)
 			if err == nil && len(data) > 0 {
@@ -940,12 +937,12 @@ func TryCreateMetricsFetchingHelpers(dbUnique string) error {
 	}
 
 	if fileBasedMetrics {
-		helpers, _, err := metrics.ReadMetricsFromFolder(mainContext, path.Join(opts.Metric.MetricsFolder, metrics.FileBasedMetricHelpersDir))
+		helpers, _, err := metrics.ReadMetricsFromFolder(mainContext, path.Join(opts.Metrics.MetricsFolder, metrics.FileBasedMetricHelpersDir))
 		if err != nil {
-			logger.Errorf("Failed to fetch helpers from \"%s\": %s", path.Join(opts.Metric.MetricsFolder, metrics.FileBasedMetricHelpersDir), err)
+			logger.Errorf("Failed to fetch helpers from \"%s\": %s", path.Join(opts.Metrics.MetricsFolder, metrics.FileBasedMetricHelpersDir), err)
 			return err
 		}
-		logger.Debug("%d helper definitions found from \"%s\"...", len(helpers), path.Join(opts.Metric.MetricsFolder, metrics.FileBasedMetricHelpersDir))
+		logger.Debug("%d helper definitions found from \"%s\"...", len(helpers), path.Join(opts.Metrics.MetricsFolder, metrics.FileBasedMetricHelpersDir))
 
 		for helperName := range helpers {
 			if strings.Contains(helperName, "windows") {

src/database.go

@@ -29,7 +29,8 @@ type PgxPoolIface interface {
 	PgxIface
 	Acquire(ctx context.Context) (*pgxpool.Conn, error)
 	BeginTx(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error)
-	Stat() *pgxpool.Stat
 	Close()
+	Config() *pgxpool.Config
 	Ping(ctx context.Context) error
+	Stat() *pgxpool.Stat
 }

src/db/conn.go

@@ -14,7 +14,7 @@ import (
 func newFormatter(noColors bool) *Formatter {
 	return &Formatter{
 		HideKeys:        false,
-		FieldsOrder:     []string{"source", "metric", "sql", "params"},
+		FieldsOrder:     []string{"source", "sink", "metric", "rows", "sql", "params"},
 		TimestampFormat: "2006-01-02 15:04:05.000",
 		ShowFullLevel:   true,
 		NoColors:        noColors,

src/log/formatter.go

@@ -188,9 +188,6 @@ func VersionToInt(v string) uint {
 }
 
 func RestoreSQLConnPoolLimitsForPreviouslyDormantDB(dbUnique string) {
-	if !opts.UseConnPooling {
-		return
-	}
 	monitoredDbConnCacheLock.Lock()
 	defer monitoredDbConnCacheLock.Unlock()
 
@@ -460,8 +457,8 @@ func FetchMetrics(ctx context.Context, msg MetricFetchMessage, hostState map[str
 	}
 
 	isCacheable = IsCacheableMetric(msg, mvp)
-	if isCacheable && opts.InstanceLevelCacheMaxSeconds > 0 && msg.Interval.Seconds() > float64(opts.InstanceLevelCacheMaxSeconds) {
-		cachedData = GetFromInstanceCacheIfNotOlderThanSeconds(msg, opts.InstanceLevelCacheMaxSeconds)
+	if isCacheable && opts.Metrics.InstanceLevelCacheMaxSeconds > 0 && msg.Interval.Seconds() > float64(opts.Metrics.InstanceLevelCacheMaxSeconds) {
+		cachedData = GetFromInstanceCacheIfNotOlderThanSeconds(msg, opts.Metrics.InstanceLevelCacheMaxSeconds)
 		if len(cachedData) > 0 {
 			fromCache = true
 			goto send_to_storageChannel
@@ -472,7 +469,7 @@ retry_with_superuser_sql: // if 1st fetch with normal SQL fails, try with SU SQL
 
 	sql = mvp.SQL
 
-	if opts.Metric.NoHelperFunctions && mvp.CallsHelperFunctions && mvp.SQLSU != "" {
+	if opts.Metrics.NoHelperFunctions && mvp.CallsHelperFunctions && mvp.SQLSU != "" {
 		logger.Debugf("[%s:%s] Using SU SQL instead of normal one due to --no-helper-functions input", msg.DBUniqueName, msg.MetricName)
 		sql = mvp.SQLSU
 		retryWithSuperuserSQL = false
@@ -557,13 +554,13 @@ retry_with_superuser_sql: // if 1st fetch with normal SQL fails, try with SU SQL
 
 	}
 
-	if isCacheable && opts.InstanceLevelCacheMaxSeconds > 0 && msg.Interval.Seconds() > float64(opts.InstanceLevelCacheMaxSeconds) {
+	if isCacheable && opts.Metrics.InstanceLevelCacheMaxSeconds > 0 && msg.Interval.Seconds() > float64(opts.Metrics.InstanceLevelCacheMaxSeconds) {
 		PutToInstanceCache(msg, data)
 	}
 
 send_to_storageChannel:
 
-	if (opts.Metric.RealDbnameField > "" || opts.Metric.SystemIdentifierField > "") && msg.Source == sources.SourcePostgres {
+	if (opts.Measurements.RealDbnameField > "" || opts.Measurements.SystemIdentifierField > "") && msg.Source == sources.SourcePostgres {
 		dbPgVersionMapLock.RLock()
 		ver := dbPgVersionMap[msg.DBUniqueName]
 		dbPgVersionMapLock.RUnlock()
@@ -658,16 +655,16 @@ func AddDbnameSysinfoIfNotExistsToQueryResultData(msg MetricFetchMessage, data m
 
 	logger.Debugf("Enriching all rows of [%s:%s] with sysinfo (%s) / real dbname (%s) if set. ", msg.DBUniqueName, msg.MetricName, ver.SystemIdentifier, ver.RealDbname)
 	for _, dr := range data {
-		if opts.Metric.RealDbnameField > "" && ver.RealDbname > "" {
-			old, ok := dr[opts.Metric.RealDbnameField]
+		if opts.Measurements.RealDbnameField > "" && ver.RealDbname > "" {
+			old, ok := dr[opts.Measurements.RealDbnameField]
 			if !ok || old == "" {
-				dr[opts.Metric.RealDbnameField] = ver.RealDbname
+				dr[opts.Measurements.RealDbnameField] = ver.RealDbname
 			}
 		}
-		if opts.Metric.SystemIdentifierField > "" && ver.SystemIdentifier > "" {
-			old, ok := dr[opts.Metric.SystemIdentifierField]
+		if opts.Measurements.SystemIdentifierField > "" && ver.SystemIdentifier > "" {
+			old, ok := dr[opts.Measurements.SystemIdentifierField]
 			if !ok || old == "" {
-				dr[opts.Metric.SystemIdentifierField] = ver.SystemIdentifier
+				dr[opts.Measurements.SystemIdentifierField] = ver.SystemIdentifier
 			}
 		}
 		enrichedData = append(enrichedData, dr)
@@ -779,7 +776,7 @@ func MetricGathererLoop(ctx context.Context,
 		}
 
 		// 1st try local overrides for some metrics if operating in push mode
-		if opts.DirectOSStats && IsDirectlyFetchableMetric(metricName) {
+		if opts.Metrics.DirectOSStats && IsDirectlyFetchableMetric(metricName) {
 			metricStoreMessages, err = FetchStatsDirectlyFromOS(mfm, vme, mvp)
 			if err != nil {
 				l.WithError(err).Errorf("Could not reader metric directly from OS")
@@ -1316,10 +1313,10 @@ func DoesEmergencyTriggerfileExist() bool {
 	// In highly automated K8s / IaC environments such a temporary change might involve pull requests, peer reviews, CI/CD etc
 	// which can all take too long vs "exec -it pgwatch3-pod -- touch /tmp/pgwatch3-emergency-pause".
 	// After creating the file it can still take up to --servers-refresh-loop-seconds (2min def.) for change to take effect!
-	if opts.EmergencyPauseTriggerfile == "" {
+	if opts.Metrics.EmergencyPauseTriggerfile == "" {
 		return false
 	}
-	_, err := os.Stat(opts.EmergencyPauseTriggerfile)
+	_, err := os.Stat(opts.Metrics.EmergencyPauseTriggerfile)
 	return err == nil
 }
 
@@ -1374,7 +1371,7 @@ func LoadMetricDefs(ctx context.Context) (err error) {
 	var metricDefs metrics.MetricVersionDefs
 	var renamingDefs map[string]string
 	if fileBasedMetrics {
-		metricDefs, renamingDefs, err = metrics.ReadMetricsFromFolder(ctx, opts.Metric.MetricsFolder)
+		metricDefs, renamingDefs, err = metrics.ReadMetricsFromFolder(ctx, opts.Metrics.MetricsFolder)
 	} else {
 		metricDefs, renamingDefs, err = metrics.ReadMetricsFromPostgres(ctx, configDb)
 	}
@@ -1457,7 +1454,7 @@ func main() {
 	if err != nil {
 		logger.Fatal(err)
 	}
-	if opts.Source.Init {
+	if opts.Sources.Init {
 		return
 	}
 
@@ -1500,13 +1497,13 @@ func main() {
 				logger.Fatal("could not fetch active hosts - check config!", err)
 			} else {
 				logger.Error("could not fetch active hosts, using last valid config data. err:", err)
-				time.Sleep(time.Second * time.Duration(opts.Source.Refresh))
+				time.Sleep(time.Second * time.Duration(opts.Sources.Refresh))
 				continue
 			}
 		}
 
 		if DoesEmergencyTriggerfileExist() {
-			logger.Warningf("Emergency pause triggerfile detected at %s, ignoring currently configured DBs", opts.EmergencyPauseTriggerfile)
+			logger.Warningf("Emergency pause triggerfile detected at %s, ignoring currently configured DBs", opts.Metrics.EmergencyPauseTriggerfile)
 			monitoredDbs = make([]sources.MonitoredDatabase, 0)
 		}
 
@@ -1543,7 +1540,7 @@ func main() {
 			metricConfig = host.Metrics
 			wasInstancePreviouslyDormant := IsDBDormant(dbUnique)
 
-			if host.Encryption == "aes-gcm-256" && len(opts.Source.AesGcmKeyphrase) == 0 && len(opts.Source.AesGcmKeyphraseFile) == 0 {
+			if host.Encryption == "aes-gcm-256" && len(opts.Sources.AesGcmKeyphrase) == 0 && len(opts.Sources.AesGcmKeyphraseFile) == 0 {
 				// Warn if any encrypted hosts found but no keyphrase given
 				logger.Warningf("Encrypted password type found for host \"%s\", but no decryption keyphrase specified. Use --aes-gcm-keyphrase or --aes-gcm-keyphrase-file params", dbUnique)
 			}
@@ -1589,7 +1586,7 @@ func main() {
 				}
 
 				if !opts.Ping && host.IsSuperuser && host.IsPostgresSource() && !ver.IsInRecovery {
-					if opts.Metric.NoHelperFunctions {
+					if opts.Metrics.NoHelperFunctions {
 						logger.Infof("[%s] Skipping rollout out helper functions due to the --no-helper-functions flag ...", dbUnique)
 					} else {
 						logger.Infof("Trying to create helper functions if missing for \"%s\"...", dbUnique)
@@ -1602,10 +1599,10 @@ func main() {
 			if host.IsPostgresSource() {
 				var DBSizeMB int64
 
-				if opts.Source.MinDbSizeMB >= 8 { // an empty DB is a bit less than 8MB
+				if opts.Sources.MinDbSizeMB >= 8 { // an empty DB is a bit less than 8MB
 					DBSizeMB, _ = DBGetSizeMB(dbUnique) // ignore errors, i.e. only remove from monitoring when we're certain it's under the threshold
 					if DBSizeMB != 0 {
-						if DBSizeMB < opts.Source.MinDbSizeMB {
+						if DBSizeMB < opts.Sources.MinDbSizeMB {
 							logger.Infof("[%s] DB will be ignored due to the --min-db-size-mb filter. Current (up to %v cached) DB size = %d MB", dbUnique, dbSizeCachingInterval, DBSizeMB)
 							hostsToShutDownDueToRoleChange[dbUnique] = true // for the case when DB size was previosly above the threshold
 							SetUndersizedDBState(dbUnique, true)
@@ -1640,8 +1637,8 @@ func main() {
 					RestoreSQLConnPoolLimitsForPreviouslyDormantDB(dbUnique)
 				}
 
-				if mainLoopCount == 0 && opts.TryCreateListedExtsIfMissing != "" && !ver.IsInRecovery {
-					extsToCreate := strings.Split(opts.TryCreateListedExtsIfMissing, ",")
+				if mainLoopCount == 0 && opts.Sources.TryCreateListedExtsIfMissing != "" && !ver.IsInRecovery {
+					extsToCreate := strings.Split(opts.Sources.TryCreateListedExtsIfMissing, ",")
 					extsCreated := TryCreateMissingExtensions(dbUnique, extsToCreate, ver.Extensions)
 					logger.Infof("[%s] %d/%d extensions created based on --try-create-listed-exts-if-missing input %v", dbUnique, len(extsCreated), len(extsToCreate), extsCreated)
 				}
@@ -1804,9 +1801,9 @@ func main() {
 		mainLoopCount++
 		prevLoopMonitoredDBs = monitoredDbs
 
-		logger.Debugf("main sleeping %ds...", opts.Source.Refresh)
+		logger.Debugf("main sleeping %ds...", opts.Sources.Refresh)
 		select {
-		case <-time.After(time.Second * time.Duration(opts.Source.Refresh)):
+		case <-time.After(time.Second * time.Duration(opts.Sources.Refresh)):
 			// pass
 		case <-mainContext.Done():
 			return

src/main.go

@@ -3,7 +3,6 @@ package sinks
 import (
 	"context"
 	"encoding/json"
-	"io"
 	"sync/atomic"
 
 	"github.com/cybertec-postgresql/pgwatch3/metrics"
@@ -16,24 +15,22 @@ var (
 )
 
 type JSONWriter struct {
-	ctx      context.Context
-	filename string
-	w        io.Writer
+	ctx context.Context
+	lw  *lumberjack.Logger
 }
 
 func NewJSONWriter(ctx context.Context, fname string) (*JSONWriter, error) {
 	return &JSONWriter{
-		ctx:      ctx,
-		filename: fname,
-		w:        &lumberjack.Logger{Filename: fname, Compress: true},
+		ctx: ctx,
+		lw:  &lumberjack.Logger{Filename: fname, Compress: true},
 	}, nil
 }
 
 func (jw *JSONWriter) Write(msgs []metrics.MeasurementMessage) error {
 	if len(msgs) == 0 {
 		return nil
 	}
-	enc := json.NewEncoder(jw.w)
+	enc := json.NewEncoder(jw.lw)
 	for _, msg := range msgs {
 		dataRow := map[string]any{
 			"metric":      msg.MetricName,

src/sinks/json_file.go

@@ -3,6 +3,8 @@ package sinks
 import (
 	"context"
 	"errors"
+	"fmt"
+	"strings"
 	"sync"
 
 	"github.com/cybertec-postgresql/pgwatch3/config"
@@ -23,37 +25,34 @@ type MultiWriter struct {
 }
 
 // NewMultiWriter creates and returns new instance of MultiWriter struct.
-func NewMultiWriter(ctx context.Context, opts *config.Options, metricDefs metrics.MetricVersionDefs) (*MultiWriter, error) {
+func NewMultiWriter(ctx context.Context, opts *config.Options, metricDefs metrics.MetricVersionDefs) (mw *MultiWriter, err error) {
+	var w Writer
 	logger := log.GetLogger(ctx)
-	mw := &MultiWriter{}
-	for _, f := range opts.Metric.JSONStorageFile {
-		jw, err := NewJSONWriter(ctx, f)
-		if err != nil {
-			return nil, err
+	mw = &MultiWriter{}
+	for _, s := range opts.Measurements.Sinks {
+		scheme, path, found := strings.Cut(s, "://")
+		if !found || scheme == "" || path == "" {
+			return nil, fmt.Errorf("malformed sink URI %s", s)
 		}
-		mw.AddWriter(jw)
-		logger.WithField("file", f).Info(`JSON output enabled`)
-	}
-
-	for _, connstr := range opts.Metric.PGMetricStoreConnStr {
-		pgw, err := NewPostgresWriter(ctx, connstr, opts, metricDefs)
-		if err != nil {
-			return nil, err
+		switch scheme {
+		case "jsonfile":
+			w, err = NewJSONWriter(ctx, path)
+		case "postgres", "postgresql":
+			w, err = NewPostgresWriter(ctx, s, opts, metricDefs)
+		case "prometheus":
+			w, err = NewPrometheusWriter(ctx, path)
+		default:
+			return nil, fmt.Errorf("unknown schema %s in sink URI %s", scheme, s)
 		}
-		mw.AddWriter(pgw)
-		logger.WithField("connstr", connstr).Info(`PostgreSQL output enabled`)
-	}
-
-	if opts.Metric.PrometheusListenAddr > "" {
-		promw, err := NewPrometheusWriter(ctx, opts)
 		if err != nil {
 			return nil, err
 		}
-		mw.AddWriter(promw)
-		logger.WithField("listen", opts.Metric.PrometheusListenAddr).Info(`Prometheus output enabled`)
+		mw.AddWriter(w)
+		logger.WithField("sink", s).Info(`measurements sink added`)
 	}
+
 	if len(mw.writers) == 0 {
-		return nil, errors.New("no storages specified for metrics")
+		return nil, errors.New("no sinks specified for measurements")
 	}
 	return mw, nil
 }