diff --git a/cmd/pgwatch/doc.go b/cmd/pgwatch/doc.go index 9eb3cb6d47..8e91f6cbe7 100644 --- a/cmd/pgwatch/doc.go +++ b/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] diff --git a/cmd/pgwatch/version.go b/cmd/pgwatch/version.go index 6ffdf3d124..d65b9f0ec8 100644 --- a/cmd/pgwatch/version.go +++ b/cmd/pgwatch/version.go @@ -8,7 +8,7 @@ var ( version = "unknown" date = "unknown" configSchema = "00824" - sinkSchema = "01180" + sinkSchema = "01409" ) func printVersion() { diff --git a/docs/reference/cli_env.md b/docs/reference/cli_env.md index 46e9d55cf9..8a77c7f9b0 100644 --- a/docs/reference/cli_env.md +++ b/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: diff --git a/internal/sinks/cmdopts.go b/internal/sinks/cmdopts.go index 6c3de93bd4..07c118f182 100644 --- a/internal/sinks/cmdopts.go +++ b/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"` diff --git a/internal/sinks/postgres.go b/internal/sinks/postgres.go index cd1d0848ba..97fc45c49b 100644 --- a/internal/sinks/postgres.go +++ b/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{ diff --git a/internal/sinks/postgres_schema_test.go b/internal/sinks/postgres_schema_test.go index 69774ca48f..5b9573aa90 100644 --- a/internal/sinks/postgres_schema_test.go +++ b/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() diff --git a/internal/sinks/postgres_test.go b/internal/sinks/postgres_test.go index f35b3aa72e..aa8f6ba4b5 100644 --- a/internal/sinks/postgres_test.go +++ b/internal/sinks/postgres_test.go @@ -645,24 +645,21 @@ func TestPartitionInterval(t *testing.T) { conn, err := pgx.Connect(ctx, connStr) r.NoError(err) - m := map[string]map[string]ExistingPartitionInfo{ + m := map[string]ExistingPartitionInfo{ "test_metric": { - "test_db": { - time.Now(), time.Now().Add(time.Hour), - }, + time.Now(), time.Now().Add(time.Hour), }, } - err = pgw.EnsureMetricDbnameTime(m) + err = pgw.EnsureMetricTimePartsExist(m) r.NoError(err) var partitionsNum int err = conn.QueryRow(ctx, "SELECT COUNT(*) FROM pg_partition_tree('test_metric');").Scan(&partitionsNum) a.NoError(err) - // 1 the metric table itself + 1 dbname partition - // + 4 time partitions (1 we asked for + 3 precreated) - a.Equal(6, partitionsNum) + // 1 the metric table itself + 4 time partitions (1 we asked for + 3 precreated) + a.Equal(5, partitionsNum) - part := pgw.partitionMapMetricDbname["test_metric"]["test_db"] + part := pgw.partitionMapMetric["test_metric"] // partition bounds should have a difference of 3 weeks a.Equal(part.StartTime.Add(3*7*24*time.Hour), part.EndTime) } @@ -744,29 +741,26 @@ func Test_Maintain(t *testing.T) { err = pgw.EnsureMetricDummy("test_metric_b") r.NoError(err) - // Create partitions for each dbname + // Create time-based partitions for each metric _, err = conn.Exec(ctx, ` - CREATE TABLE subpartitions.test_metric_a_db1 PARTITION OF public.test_metric_a FOR VALUES IN ('db1'); - CREATE TABLE subpartitions.test_metric_a_db2 PARTITION OF public.test_metric_a FOR VALUES IN ('db2'); - CREATE TABLE subpartitions.test_metric_a_db3 PARTITION OF public.test_metric_a FOR VALUES IN ('db3'); - CREATE TABLE subpartitions.test_metric_b_db1 PARTITION OF public.test_metric_b FOR VALUES IN ('db1'); - CREATE TABLE subpartitions.test_metric_b_db2 PARTITION OF public.test_metric_b FOR VALUES IN ('db2'); + CREATE TABLE subpartitions.test_metric_a_2024w01 PARTITION OF public.test_metric_a FOR VALUES FROM ('2024-01-01') TO ('2024-01-08'); + CREATE TABLE subpartitions.test_metric_b_2024w01 PARTITION OF public.test_metric_b FOR VALUES FROM ('2024-01-01') TO ('2024-01-08'); `) r.NoError(err) // Directly insert test data with different dbnames _, err = conn.Exec(ctx, ` - INSERT INTO test_metric_a (time, dbname, data) VALUES - (now(), 'db1', '{}'::jsonb), - (now(), 'db2', '{}'::jsonb), - (now(), 'db3', '{}'::jsonb) + INSERT INTO test_metric_a (time, dbname, data) VALUES + ('2024-01-03', 'db1', '{}'::jsonb), + ('2024-01-03', 'db2', '{}'::jsonb), + ('2024-01-03', 'db3', '{}'::jsonb) `) r.NoError(err) _, err = conn.Exec(ctx, ` - INSERT INTO test_metric_b (time, dbname, data) VALUES - (now(), 'db1', '{}'::jsonb), - (now(), 'db2', '{}'::jsonb) + INSERT INTO test_metric_b (time, dbname, data) VALUES + ('2024-01-03', 'db1', '{}'::jsonb), + ('2024-01-03', 'db2', '{}'::jsonb) `) r.NoError(err) @@ -794,16 +788,16 @@ func Test_Maintain(t *testing.T) { err = pgw.EnsureMetricDummy("test_metric_c") r.NoError(err) - // Create partition for db_active + // Create time partition for the metric _, err = conn.Exec(ctx, ` - CREATE TABLE subpartitions.test_metric_c_db_active PARTITION OF public.test_metric_c FOR VALUES IN ('db_active'); + CREATE TABLE subpartitions.test_metric_c_2024w01 PARTITION OF public.test_metric_c FOR VALUES FROM ('2024-01-01') TO ('2024-01-08'); `) r.NoError(err) // Directly insert test data with one active dbname _, err = conn.Exec(ctx, ` - INSERT INTO test_metric_c (time, dbname, data) VALUES - (now(), 'db_active', '{}'::jsonb) + INSERT INTO test_metric_c (time, dbname, data) VALUES + ('2024-01-03', 'db_active', '{}'::jsonb) `) r.NoError(err) @@ -845,14 +839,14 @@ func Test_Maintain(t *testing.T) { r.NoError(err) _, err = conn.Exec(ctx, ` - CREATE TABLE subpartitions.test_metric_d_db1 PARTITION OF public.test_metric_d FOR VALUES IN ('db1'); + CREATE TABLE subpartitions.test_metric_d_2024w01 PARTITION OF public.test_metric_d FOR VALUES FROM ('2024-01-01') TO ('2024-01-08'); `) r.NoError(err) // Directly insert test data for only db1 _, err = conn.Exec(ctx, ` - INSERT INTO test_metric_d (time, dbname, data) VALUES - (now(), 'db1', '{}'::jsonb) + INSERT INTO test_metric_d (time, dbname, data) VALUES + ('2024-01-03', 'db1', '{}'::jsonb) `) r.NoError(err) @@ -908,43 +902,39 @@ func Test_Maintain(t *testing.T) { err = pgw.SyncMetric("test", "test_metric_2", AddOp) r.NoError(err) - // create the 2nd level dbname partition - _, err = conn.Exec(ctx, "CREATE TABLE subpartitions.test_metric_2_dbname PARTITION OF public.test_metric_2 FOR VALUES IN ('test') PARTITION BY RANGE (time)") - a.NoError(err) - boundStart := time.Now().Add(-1 * 2 * 24 * time.Hour).Format("2006-01-02") boundEnd := time.Now().Add(-1 * 24 * time.Hour).Format("2006-01-02") - // create the 3rd level time partition with end bound yesterday + // create the time partition with end bound yesterday _, err = conn.Exec(ctx, fmt.Sprintf( - `CREATE TABLE subpartitions.test_metric_2_dbname_time - PARTITION OF subpartitions.test_metric_2_dbname + `CREATE TABLE subpartitions.test_metric_2_yesterday + PARTITION OF public.test_metric_2 FOR VALUES FROM ('%s') TO ('%s')`, boundStart, boundEnd), ) a.NoError(err) - _, err = conn.Exec(ctx, "COMMENT ON TABLE subpartitions.test_metric_2_dbname_time IS $$pgwatch-generated-metric-dbname-time-lvl$$") + _, err = conn.Exec(ctx, "COMMENT ON TABLE subpartitions.test_metric_2_yesterday IS $$pgwatch-generated-metric-time-lvl$$") a.NoError(err) var partitionsNum int err = conn.QueryRow(ctx, "SELECT COUNT(*) FROM pg_partition_tree('test_metric_2');").Scan(&partitionsNum) a.NoError(err) - a.Equal(3, partitionsNum) + a.Equal(2, partitionsNum) pgw.opts.RetentionInterval = "2 days" pgw.DeleteOldPartitions() // 1 day < 2 days, shouldn't delete anything err = conn.QueryRow(ctx, "SELECT COUNT(*) FROM pg_partition_tree('test_metric_2');").Scan(&partitionsNum) a.NoError(err) - a.Equal(3, partitionsNum) + a.Equal(2, partitionsNum) pgw.opts.RetentionInterval = "1 hour" pgw.DeleteOldPartitions() // 1 day > 1 hour, should delete the partition err = conn.QueryRow(ctx, "SELECT COUNT(*) FROM pg_partition_tree('test_metric_2');").Scan(&partitionsNum) a.NoError(err) - a.Equal(2, partitionsNum) + a.Equal(1, partitionsNum) }) t.Run("Epoch to Duration Conversion", func(_ *testing.T) { @@ -976,9 +966,9 @@ func Test_Maintain(t *testing.T) { }) } -// TestEnsureMetricDbnameTime_SpecialSourceNames verifies that source names with special characters +// TestEnsureMetricTimePartsExist_SpecialMetricNames verifies that metric names with special characters // (dots, uppercase, hyphens, underscores) are accepted by the partition functions. -func TestEnsureMetricDbnameTime_SpecialSourceNames(t *testing.T) { +func TestEnsureMetricTimePartsExist_SpecialMetricNames(t *testing.T) { r := require.New(t) a := assert.New(t) @@ -999,41 +989,41 @@ func TestEnsureMetricDbnameTime_SpecialSourceNames(t *testing.T) { r.NoError(err) specialNames := []string{ - "source.new", - "Source.With.Dots", - "UPPERCASE_SOURCE", - "MixedCase_Source-123", - "source-with-hyphens", - "source_with_underscores", - "Source123.Test_Name-456", + "metric.new", + "Metric.With.Dots", + "UPPERCASE_METRIC", + "MixedCase_Metric-123", + "metric-with-hyphens", + "metric_with_underscores", + "Metric123.Test_Name-456", } - m := make(map[string]map[string]ExistingPartitionInfo) - m["test_metric"] = make(map[string]ExistingPartitionInfo) + m := make(map[string]ExistingPartitionInfo) for _, name := range specialNames { - m["test_metric"][name] = ExistingPartitionInfo{ + m[name] = ExistingPartitionInfo{ StartTime: time.Now(), EndTime: time.Now().Add(time.Hour), } } - err = pgw.EnsureMetricDbnameTime(m) - r.NoError(err, "EnsureMetricDbnameTime should handle special source names") + err = pgw.EnsureMetricTimePartsExist(m) + r.NoError(err, "EnsureMetricTimePartsExist should handle special metric names") conn, err := pgx.Connect(ctx, connStr) r.NoError(err) defer conn.Close(ctx) var partitionCount int - err = conn.QueryRow(ctx, "SELECT COUNT(*) FROM pg_partition_tree('test_metric') WHERE level = 1").Scan(&partitionCount) + err = conn.QueryRow(ctx, `SELECT COUNT(*) FROM pg_partition_tree('"metric.new"') WHERE level = 1`).Scan(&partitionCount) r.NoError(err) - a.Equal(len(specialNames), partitionCount, "expected one dbname-level partition per special source name") + // 4 time partitions (1 requested + 3 precreated) per metric + a.Equal(4, partitionCount) } -// TestEnsureMetricDbnameTime_IdempotentAcrossRestarts verifies that repeated calls to -// EnsureMetricDbnameTime with fresh writer instances (simulating process restarts) +// TestEnsureMetricTimePartsExist_IdempotentAcrossRestarts verifies that repeated calls to +// EnsureMetricTimePartsExist with fresh writer instances (simulating process restarts) // do not create duplicate partitions. -func TestEnsureMetricDbnameTime_IdempotentAcrossRestarts(t *testing.T) { +func TestEnsureMetricTimePartsExist_IdempotentAcrossRestarts(t *testing.T) { r := require.New(t) a := assert.New(t) @@ -1051,20 +1041,17 @@ func TestEnsureMetricDbnameTime_IdempotentAcrossRestarts(t *testing.T) { BatchingDelay: time.Second, } - m := map[string]map[string]ExistingPartitionInfo{ + m := map[string]ExistingPartitionInfo{ "restart_test_metric": { - "test_source": { - StartTime: time.Now(), - EndTime: time.Now().Add(time.Hour), - }, + StartTime: time.Now(), + EndTime: time.Now().Add(time.Hour), }, } - var partitionCountAfterFirst int for i := range 5 { pgw, err := NewPostgresWriter(ctx, connStr, opts) r.NoError(err) - r.NoError(pgw.EnsureMetricDbnameTime(m)) + r.NoError(pgw.EnsureMetricTimePartsExist(m)) conn, err := pgx.Connect(ctx, connStr) r.NoError(err) @@ -1072,12 +1059,6 @@ func TestEnsureMetricDbnameTime_IdempotentAcrossRestarts(t *testing.T) { err = conn.QueryRow(ctx, "SELECT COUNT(*) FROM pg_partition_tree('restart_test_metric') WHERE isleaf").Scan(&count) conn.Close(ctx) r.NoError(err) - - if i == 0 { - partitionCountAfterFirst = count - } else { - a.Equal(partitionCountAfterFirst, count, - "partition count should not grow on restart %d", i+1) - } + a.Equal(4, count, "partition count should not grow on restart %d", i+1) } } diff --git a/internal/sinks/sql/README.md b/internal/sinks/sql/README.md index e198b62e6f..787dcc463b 100644 --- a/internal/sinks/sql/README.md +++ b/internal/sinks/sql/README.md @@ -1,7 +1,7 @@ ## Rollout sequence pgwatch automatically handles database schema for metric measurements. If `timescale` extension is available then `timescale` -schema will be used by default. Otherwise `metric-dbname-time` schema is applied. +schema will be used by default. Otherwise `metric-time` schema is applied. If one wants to init the schema without running the monitoring, they should use `--init` command-line parameter, e.g. @@ -11,9 +11,9 @@ pgwatch --config=postgresql://pgwatch:pgwatchadmin@localhost/pgwatch --sink=post ## Schema types -### metric-dbname-time +### metric-time -A single top level table for each distinct metric in the "public" schema + 2 levels of subpartitions ("dbname" + weekly time based) in the "subpartitions" schema. +A single top level table for each distinct metric in the "public" schema with time-based partitioning in the "subpartitions" schema. Provides the fastest query runtimes when having long retention intervals / lots of metrics data or slow disks and accessing mostly only a single DB's metrics at a time. @@ -27,18 +27,13 @@ Something like below will be done by the gatherer AUTOMATICALLY: ```sql create table public."mymetric" (LIKE admin.metrics_template) - PARTITION BY LIST (dbname); + PARTITION BY RANGE (time); COMMENT ON TABLE public."mymetric" IS 'pgwatch-generated-metric-lvl'; -create table subpartitions."mymetric_mydbname" +create table subpartitions."mymetric_y2019w01" -- week calculated dynamically of course PARTITION OF public."mymetric" - FOR VALUES IN ('my-dbname') PARTITION BY RANGE (time); -COMMENT ON TABLE subpartitions."mymetric_mydbname" IS 'pgwatch-generated-metric-dbname-lvl'; - -create table subpartitions."mymetric_mydbname_y2019w01" -- month calculated dynamically of course - PARTITION OF subpartitions."mymetric_mydbname" FOR VALUES FROM ('2019-01-01') TO ('2019-01-07'); -COMMENT ON TABLE subpartitions."mymetric_mydbname_y2019w01" IS 'pgwatch-generated-metric-dbname-time-lvl'; +COMMENT ON TABLE subpartitions."mymetric_y2019w01" IS 'pgwatch-generated-metric-time-lvl'; ``` ### timescale diff --git a/internal/sinks/sql/admin_functions.sql b/internal/sinks/sql/admin_functions.sql index 390d198d75..cb5bca9453 100644 --- a/internal/sinks/sql/admin_functions.sql +++ b/internal/sinks/sql/admin_functions.sql @@ -14,7 +14,7 @@ BEGIN END IF; IF l_schema_type = 'postgres' THEN - EXECUTE format('CREATE TABLE public.%I (LIKE admin.metrics_template INCLUDING INDEXES) PARTITION BY LIST (dbname)', metric); + EXECUTE format('CREATE TABLE public.%I (LIKE admin.metrics_template INCLUDING INDEXES) PARTITION BY RANGE (time)', metric); ELSIF l_schema_type = 'timescale' THEN PERFORM admin.ensure_partition_timescale(metric); END IF; @@ -110,7 +110,7 @@ BEGIN c.relkind IN ('r', 'p') AND c.relnamespace = 'subpartitions'::regnamespace AND (regexp_match(pg_catalog.pg_get_expr(c.relpartbound, c.oid), E'TO \\((''.*?'')'))[1]::timestamp < (now() - older_than) - AND pg_catalog.obj_description(c.oid, 'pg_class') IN ('pgwatch-generated-metric-time-lvl', 'pgwatch-generated-metric-dbname-time-lvl') + AND pg_catalog.obj_description(c.oid, 'pg_class') = 'pgwatch-generated-metric-time-lvl' ORDER BY 1; WHEN 'timescale' THEN RETURN QUERY @@ -167,53 +167,6 @@ BEGIN END; $SQL$ LANGUAGE plpgsql; -/* -drop_source_partitions removes all metric data for a decommissioned source - p_source_name - the dbname (source) whose partitions should be dropped -Returns the number of patritions dropped -*/ -CREATE OR REPLACE FUNCTION admin.drop_source_partitions(p_source_name text) -RETURNS int AS -$SQL$ -DECLARE - r record; - v_dropped_count int := 0; -BEGIN - FOR r IN - SELECT - c.oid::regclass::text AS partition_name, - parent.oid::regclass::text AS metric_table, - parent.relname AS metric_name - FROM pg_catalog.pg_class c - JOIN pg_catalog.pg_inherits i ON c.oid = i.inhrelid - JOIN pg_catalog.pg_class parent ON parent.oid = i.inhparent - WHERE c.relnamespace = 'subpartitions'::regnamespace - AND pg_catalog.obj_description(parent.oid, 'pg_class') = 'pgwatch-generated-metric-lvl' - AND (regexp_match( - pg_catalog.pg_get_expr(c.relpartbound, c.oid), - E'FOR VALUES IN \\(''(.+?)''\\)' - ))[1] = p_source_name - LOOP - -- advisory lock matching ensure_partition_metric_dbname_time - PERFORM pg_advisory_xact_lock( - regexp_replace(md5(r.metric_name), E'\\D', '', 'g')::varchar(10)::int8 - ); - - RAISE NOTICE 'detaching partition: % from %', r.partition_name, r.metric_table; - EXECUTE 'ALTER TABLE ' || r.metric_table || ' DETACH PARTITION ' || r.partition_name; - - RAISE NOTICE 'dropping partition: %', r.partition_name; - EXECUTE 'DROP TABLE IF EXISTS ' || r.partition_name; - - v_dropped_count := v_dropped_count + 1; - END LOOP; - - DELETE FROM admin.all_distinct_dbname_metrics WHERE dbname = p_source_name; - - RETURN v_dropped_count; -END; -$SQL$ LANGUAGE plpgsql; - /* maintain_unique_sources() maintains a mapping of unique sources in each metric table in admin.all_distinct_dbname_metrics. This is used to avoid listing the same source diff --git a/internal/sinks/sql/admin_schema.sql b/internal/sinks/sql/admin_schema.sql index d5abe9b57c..bdb5fcbe80 100644 --- a/internal/sinks/sql/admin_schema.sql +++ b/internal/sinks/sql/admin_schema.sql @@ -95,5 +95,6 @@ INSERT INTO admin.migration (id, version) VALUES (0, '01110 Apply postgres sink schema migrations'), - (1, '01180 Apply admin functions migrations for v5'); + (1, '01180 Apply admin functions migrations for v5'), + (2, '01409 Switch to time-only partitioning'); -- apply new migration value to `sinkSchema` in `cmd/pgwatch/version.go` file as well diff --git a/internal/sinks/sql/ensure_partition_postgres.sql b/internal/sinks/sql/ensure_partition_postgres.sql index 923ff99494..3db97ebe98 100644 --- a/internal/sinks/sql/ensure_partition_postgres.sql +++ b/internal/sinks/sql/ensure_partition_postgres.sql @@ -1,16 +1,14 @@ /* - ensure_partition_metric_dbname_time creates partitioned metric tables if not already existing + ensure_partition_metric_time creates partitioned metric tables if not already existing metric - name of the metric (top level table) - dbname - name of the database (2nd level partition) metric_timestamp - timestamp of the metric (used to determine the time partition) partition_period - interval for partitioning (e.g., '1 week', '1 day', '1 month') partitions_to_precreate - how many future time partitions to create (default 3) part_available_from - output parameter, start time of the time partition where the given metric_timestamp fits in part_available_to - output parameter, end time of the time partition where the given metric_timestamp fits in */ -CREATE OR REPLACE FUNCTION admin.ensure_partition_metric_dbname_time( +CREATE OR REPLACE FUNCTION admin.ensure_partition_metric_time( metric text, - dbname text, metric_timestamp timestamptz, partition_period interval default '1 week'::interval, partitions_to_precreate int default 3, @@ -18,18 +16,16 @@ CREATE OR REPLACE FUNCTION admin.ensure_partition_metric_dbname_time( OUT part_available_to timestamptz) RETURNS record AS /* - creates a top level metric table, a dbname partition and a time partition if not already existing. + creates a top level metric table and time partitions if not already existing. returns time partition start/end date */ $SQL$ DECLARE - l_part_name_2nd text; - l_part_name_3rd text; + l_part_name text; l_part_start timestamptz; l_part_end timestamptz; ideal_length int; l_template_table text := 'admin.metrics_template'; - MAX_IDENT_LEN CONSTANT integer := current_setting('max_identifier_length')::int; l_partition_format text; l_time_suffix text; l_existing_upper_bound timestamptz; @@ -50,31 +46,15 @@ BEGIN PERFORM pg_advisory_xact_lock(regexp_replace( md5(metric) , E'\\D', '', 'g')::varchar(10)::int8); - -- 1. level IF to_regclass('public.' || quote_ident(metric)) IS NULL THEN - EXECUTE format('CREATE TABLE public.%I (LIKE admin.metrics_template INCLUDING INDEXES) PARTITION BY LIST (dbname)', metric); + EXECUTE format('CREATE TABLE public.%I (LIKE admin.metrics_template INCLUDING INDEXES) PARTITION BY RANGE (time)', metric); EXECUTE format('COMMENT ON TABLE public.%I IS $$pgwatch-generated-metric-lvl$$', metric); END IF; -- 2. level - l_part_name_2nd := metric || '_' || dbname; - IF char_length(l_part_name_2nd) > MAX_IDENT_LEN -- use "dbname" hash instead of name for overly long ones - THEN - ideal_length = MAX_IDENT_LEN - char_length(format('%s_', metric)); - l_part_name_2nd := metric || '_' || substring(md5(dbname) from 1 for ideal_length); - END IF; - IF to_regclass('subpartitions.' || quote_ident(l_part_name_2nd)) IS NULL - THEN - EXECUTE format('CREATE TABLE subpartitions.%I PARTITION OF public.%I FOR VALUES IN (%L) PARTITION BY RANGE (time)', - l_part_name_2nd, metric, dbname); - EXECUTE format('COMMENT ON TABLE subpartitions.%I IS $$pgwatch-generated-metric-dbname-lvl$$', l_part_name_2nd); - END IF; - - -- 3. level - -- Get existing partition upper bound SELECT max(substring(pg_catalog.pg_get_expr(c.relpartbound, c.oid, true) from 'TO \(''([^'']+)''')::timestamptz) INTO l_existing_upper_bound @@ -83,7 +63,7 @@ BEGIN JOIN pg_catalog.pg_class parent ON parent.oid = i.inhparent WHERE c.relispartition AND c.relnamespace = 'subpartitions'::regnamespace - AND parent.relname = l_part_name_2nd; + AND parent.relname = metric; -- Determine starting point for new partitions IF l_existing_upper_bound IS NOT NULL THEN @@ -100,7 +80,7 @@ BEGIN -- For hourly periods (>= 1 hour, < 1 day) l_part_start := date_trunc('hour', metric_timestamp); END CASE; - + -- For the first partition, set the available range part_available_from := l_part_start; part_available_to := l_part_start + partition_period; @@ -109,7 +89,7 @@ BEGIN -- Create partitions FOR i IN 0..partitions_to_precreate LOOP l_part_end := l_part_start + partition_period; - + -- Update the available range for the first partition only if we started from metric_timestamp IF i = 0 AND l_existing_upper_bound IS NULL THEN part_available_from := l_part_start; @@ -125,19 +105,13 @@ BEGIN END IF; l_time_suffix := to_char(l_part_start, l_partition_format); - l_part_name_3rd := format('%s_%s_%s', metric, dbname, l_time_suffix); - - IF char_length(l_part_name_3rd) > MAX_IDENT_LEN -- use "dbname" hash instead of name for overly long ones - THEN - ideal_length = MAX_IDENT_LEN - char_length(format('%s__%s', metric, l_time_suffix)); - l_part_name_3rd := format('%s_%s_%s', metric, substring(md5(dbname) from 1 for ideal_length), l_time_suffix); - END IF; + l_part_name := format('%s_%s', metric, l_time_suffix); - IF to_regclass('subpartitions.' || quote_ident(l_part_name_3rd)) IS NULL + IF to_regclass('subpartitions.' || quote_ident(l_part_name)) IS NULL THEN - EXECUTE format('CREATE TABLE subpartitions.%I PARTITION OF subpartitions.%I FOR VALUES FROM ($$%s$$) TO ($$%s$$)', - l_part_name_3rd, l_part_name_2nd, l_part_start, l_part_end); - EXECUTE format('COMMENT ON TABLE subpartitions.%I IS $$pgwatch-generated-metric-dbname-time-lvl$$', l_part_name_3rd); + EXECUTE format('CREATE TABLE subpartitions.%I PARTITION OF public.%I FOR VALUES FROM ($$%s$$) TO ($$%s$$)', + l_part_name, metric, l_part_start, l_part_end); + EXECUTE format('COMMENT ON TABLE subpartitions.%I IS $$pgwatch-generated-metric-time-lvl$$', l_part_name); END IF; l_part_start := l_part_end; @@ -156,14 +130,12 @@ BEGIN JOIN pg_catalog.pg_class parent ON parent.oid = i.inhparent WHERE c.relispartition AND n.nspname = 'subpartitions' - AND parent.relname = l_part_name_2nd + AND parent.relname = metric ) AS partitions - WHERE metric_timestamp >= lower_text::timestamptz + WHERE metric_timestamp >= lower_text::timestamptz AND metric_timestamp < upper_text::timestamptz LIMIT 1; END IF; - + END; $SQL$ LANGUAGE plpgsql; - --- GRANT EXECUTE ON FUNCTION admin.ensure_partition_metric_dbname_time(text,text,timestamp with time zone,interval,integer) TO pgwatch;