[*] migrate to time-only partitioning in pg sink

cmd/pgwatch/doc.go

@@ -54,6 +54,9 @@
 //	                                     may be ignored by some sinks
 //	                                     (default: 950ms)
 //	                                     [$PW_BATCHING_DELAY]
+//  --partition-interval=                Time range for PostgreSQL sink time partitions. 
+// 										 Must be a valid PostgreSQL interval. 
+// 										 (default: 1 day) [$PW_PARTITION_INTERVAL]
 //	--retention=                         Delete metrics older than this. 
 //	                                     Must be a valid PostgreSQL interval. 
 //	                                     (default: 14 days) [$PW_RETENTION]

cmd/pgwatch/version.go

@@ -8,7 +8,7 @@ var (
 	version      = "unknown"
 	date         = "unknown"
 	configSchema = "00824"
-	sinkSchema   = "01180"
+	sinkSchema   = "01409"
 )
 
 func printVersion() {

docs/reference/cli_env.md

@@ -94,7 +94,7 @@ It reads the configuration from the specified sources and metrics, then begins c
 
 - `--partition-interval=`
 
-    Time range for PostgreSQL sink table partitions. Must be a valid PostgreSQL interval. (default: 1 week)  
+    Time range for PostgreSQL sink table partitions. Must be a valid PostgreSQL interval. (default: 1 day)  
     ENV: `$PW_PARTITION_INTERVAL`
 
     Example:

internal/sinks/cmdopts.go

@@ -6,7 +6,7 @@ import "time"
 type CmdOpts struct {
 	Sinks                 []string      `long:"sink" mapstructure:"sink" description:"URI where metrics will be stored, can be used multiple times" env:"PW_SINK"`
 	BatchingDelay         time.Duration `long:"batching-delay" mapstructure:"batching-delay" description:"Sink-specific batching flush delay; may be ignored by some sinks" default:"950ms" env:"PW_BATCHING_DELAY"`
-	PartitionInterval     string        `long:"partition-interval" mapstructure:"partition-interval" description:"Time range for PostgreSQL sink time partitions. Must be a valid PostgreSQL interval." default:"1 week" env:"PW_PARTITION_INTERVAL"`
+	PartitionInterval     string        `long:"partition-interval" mapstructure:"partition-interval" description:"Time range for PostgreSQL sink time partitions. Must be a valid PostgreSQL interval." default:"1 day" env:"PW_PARTITION_INTERVAL"`
 	RetentionInterval     string        `long:"retention" mapstructure:"retention" description:"Delete metrics older than this. Must be a valid PostgreSQL interval." default:"14 days" env:"PW_RETENTION"`
 	MaintenanceInterval   string        `long:"maintenance-interval" mapstructure:"maintenance-interval" description:"Run pgwatch maintenance tasks on sinks with this interval e.g., deleting old metrics; Set to zero to disable. Must be a valid PostgreSQL interval." default:"12 hours" env:"PW_MAINTENANCE_INTERVAL"`
 	RealDbnameField       string        `long:"real-dbname-field" mapstructure:"real-dbname-field" description:"Tag key for real database name" env:"PW_REAL_DBNAME_FIELD" default:"real_dbname"`

internal/sinks/postgres.go

@@ -70,8 +70,7 @@ type PostgresWriter struct {
 	input                    chan metrics.MeasurementEnvelope
 	lastError                chan error
 	forceRecreatePartitions  bool                                        // to signal override PG metrics storage cache
-	partitionMapMetric       map[string]ExistingPartitionInfo            // metric = min/max bounds
-	partitionMapMetricDbname map[string]map[string]ExistingPartitionInfo // metric[dbname = min/max bounds]
+	partitionMapMetric 		 map[string]ExistingPartitionInfo // metric = min/max bounds
 }
 
 // make sure *dbMetricReaderWriter implements the Migrator interface
@@ -97,8 +96,7 @@ func NewWriterFromPostgresConn(ctx context.Context, conn db.PgxPoolIface, opts *
 		lastError:                make(chan error),
 		sinkDb:                   conn,
 		forceRecreatePartitions:  false,
-		partitionMapMetric:       make(map[string]ExistingPartitionInfo),
-		partitionMapMetricDbname: make(map[string]map[string]ExistingPartitionInfo),
+		partitionMapMetric: 	  make(map[string]ExistingPartitionInfo),
 	}
 	l.Info("initialising measurements database...")
 	if err = pgw.init(); err != nil {
@@ -364,8 +362,7 @@ func (pgw *PostgresWriter) flush(msgs []metrics.MeasurementEnvelope) {
 		return
 	}
 	logger := log.GetLogger(pgw.ctx)
-	pgPartBounds := make(map[string]ExistingPartitionInfo)                  // metric=min/max
-	pgPartBoundsDbName := make(map[string]map[string]ExistingPartitionInfo) // metric=[dbname=min/max]
+	pgPartBounds := make(map[string]ExistingPartitionInfo) // metric=min/max
 	var err error
 
 	slices.SortFunc(msgs, func(a, b metrics.MeasurementEnvelope) int {
@@ -380,41 +377,21 @@ func (pgw *PostgresWriter) flush(msgs []metrics.MeasurementEnvelope) {
 	for _, msg := range msgs {
 		for _, dataRow := range msg.Data {
 			epochTime := time.Unix(0, metrics.Measurement(dataRow).GetEpoch())
-			switch pgw.metricSchema {
-			case DbStorageSchemaTimescale:
-				// set min/max timestamps to check/create partitions
-				bounds, ok := pgPartBounds[msg.MetricName]
-				if !ok || (ok && epochTime.Before(bounds.StartTime)) {
-					bounds.StartTime = epochTime
-					pgPartBounds[msg.MetricName] = bounds
-				}
-				if !ok || (ok && epochTime.After(bounds.EndTime)) {
-					bounds.EndTime = epochTime
-					pgPartBounds[msg.MetricName] = bounds
-				}
-			case DbStorageSchemaPostgres:
-				_, ok := pgPartBoundsDbName[msg.MetricName]
-				if !ok {
-					pgPartBoundsDbName[msg.MetricName] = make(map[string]ExistingPartitionInfo)
-				}
-				bounds, ok := pgPartBoundsDbName[msg.MetricName][msg.DBName]
-				if !ok || (ok && epochTime.Before(bounds.StartTime)) {
-					bounds.StartTime = epochTime
-					pgPartBoundsDbName[msg.MetricName][msg.DBName] = bounds
-				}
-				if !ok || (ok && epochTime.After(bounds.EndTime)) {
-					bounds.EndTime = epochTime
-					pgPartBoundsDbName[msg.MetricName][msg.DBName] = bounds
-				}
-			default:
-				logger.Fatal("unknown storage schema...")
+			bounds, ok := pgPartBounds[msg.MetricName]
+			if !ok || (ok && epochTime.Before(bounds.StartTime)) {
+				bounds.StartTime = epochTime
+				pgPartBounds[msg.MetricName] = bounds
+			}
+			if !ok || (ok && epochTime.After(bounds.EndTime)) {
+				bounds.EndTime = epochTime
+				pgPartBounds[msg.MetricName] = bounds
 			}
 		}
 	}
 
 	switch pgw.metricSchema {
 	case DbStorageSchemaPostgres:
-		err = pgw.EnsureMetricDbnameTime(pgPartBoundsDbName)
+		err = pgw.EnsureMetricTimePartsExist(pgPartBounds)
 	case DbStorageSchemaTimescale:
 		err = pgw.EnsureMetricTimescale(pgPartBounds)
 	default:
@@ -470,38 +447,23 @@ func (pgw *PostgresWriter) EnsureMetricTimescale(pgPartBounds map[string]Existin
 	return
 }
 
-func (pgw *PostgresWriter) EnsureMetricDbnameTime(metricDbnamePartBounds map[string]map[string]ExistingPartitionInfo) (err error) {
+func (pgw *PostgresWriter) EnsureMetricTimePartsExist(metricPartBounds map[string]ExistingPartitionInfo) (error) {
+	var err error
 	var rows pgx.Rows
-	sqlEnsure := `select * from admin.ensure_partition_metric_dbname_time($1, $2, $3, $4)`
-	for metric, dbnameTimestampMap := range metricDbnamePartBounds {
-		_, ok := pgw.partitionMapMetricDbname[metric]
-		if !ok {
-			pgw.partitionMapMetricDbname[metric] = make(map[string]ExistingPartitionInfo)
+	sqlEnsure := `select * from admin.ensure_partition_metric_time($1, $2, $3)`
+	for metric, pb := range metricPartBounds {
+		if pb.StartTime.IsZero() || pb.EndTime.IsZero() {
+			return fmt.Errorf("zero StartTime/EndTime in partitioning request: [%s:%v]", metric, pb)
 		}
-
-		for dbname, pb := range dbnameTimestampMap {
-			if pb.StartTime.IsZero() || pb.EndTime.IsZero() {
-				return fmt.Errorf("zero StartTime/EndTime in partitioning request: [%s:%v]", metric, pb)
-			}
-			partInfo, ok := pgw.partitionMapMetricDbname[metric][dbname]
-			if !ok || (ok && (pb.StartTime.Before(partInfo.StartTime))) || pgw.forceRecreatePartitions {
-				if rows, err = pgw.sinkDb.Query(pgw.ctx, sqlEnsure, metric, dbname, pb.StartTime, pgw.opts.PartitionInterval); err != nil {
-					return
-				}
-				if partInfo, err = pgx.CollectOneRow(rows, pgx.RowToStructByPos[ExistingPartitionInfo]); err != nil {
-					return err
-				}
-				pgw.partitionMapMetricDbname[metric][dbname] = partInfo
+		partInfo, ok := pgw.partitionMapMetric[metric]
+		if !ok || pb.EndTime.After(partInfo.EndTime) || pb.EndTime.Equal(partInfo.EndTime) || pgw.forceRecreatePartitions {
+			if rows, err = pgw.sinkDb.Query(pgw.ctx, sqlEnsure, metric, pb.EndTime, pgw.opts.PartitionInterval); err != nil {
+				return err
 			}
-			if pb.EndTime.After(partInfo.EndTime) || pb.EndTime.Equal(partInfo.EndTime) || pgw.forceRecreatePartitions {
-				if rows, err = pgw.sinkDb.Query(pgw.ctx, sqlEnsure, metric, dbname, pb.EndTime, pgw.opts.PartitionInterval); err != nil {
-					return
-				}
-				if partInfo, err = pgx.CollectOneRow(rows, pgx.RowToStructByPos[ExistingPartitionInfo]); err != nil {
-					return err
-				}
-				pgw.partitionMapMetricDbname[metric][dbname] = partInfo
+			if partInfo, err = pgx.CollectOneRow(rows, pgx.RowToStructByPos[ExistingPartitionInfo]); err != nil {
+				return err
 			}
+			pgw.partitionMapMetric[metric] = partInfo
 		}
 	}
 	return nil
@@ -631,6 +593,31 @@ var migrations func() migrator.Option = func() migrator.Option {
 			},
 		},
 
+		&migrator.Migration{
+			Name: "01409 Switch to time-only partitioning",
+			Func: func(ctx context.Context, tx pgx.Tx) error {
+				_, err := tx.Exec(ctx, `SELECT admin.drop_all_metric_tables()`)
+				if err != nil {
+					return err
+				}
+
+				_, err = tx.Exec(ctx, `
+					DROP FUNCTION IF EXISTS admin.ensure_partition_metric_dbname_time;
+				`)
+				if err != nil {
+					return err
+				}
+
+				_, err = tx.Exec(ctx, sqlMetricAdminFunctions)
+				if err != nil {
+					return err
+				}
+
+				_, err = tx.Exec(ctx, sqlMetricEnsurePartitionPostgres)
+				return err
+			},
+		},
+
 		// adding new migration here, update "admin"."migration" in "admin_schema.sql"!
 
 		// &migrator.Migration{

internal/sinks/postgres_schema_test.go

@@ -91,7 +91,7 @@ func TestPostgresWriterNeedsMigrationNoMigrationNeeded(t *testing.T) {
 	conn.ExpectQuery(`SELECT to_regclass`).
 		WithArgs("admin.migration").
 		WillReturnRows(pgxmock.NewRows([]string{"to_regclass"}).AddRow(true))
-	conn.ExpectQuery(`SELECT count`).WillReturnRows(pgxmock.NewRows([]string{"count"}).AddRow(2))
+	conn.ExpectQuery(`SELECT count`).WillReturnRows(pgxmock.NewRows([]string{"count"}).AddRow(3))
 
 	pgw := &PostgresWriter{ctx: ctx, sinkDb: conn}
 	needs, err := pgw.NeedsMigration()