[+] allow to specify sink retention period as interval value (#988)

Retention is now treated as Postgres interval. This allow people to use
custom retention intervals, e.g. "1 month", "14 days", "42 hours"

Author: pashagolub

cmd/pgwatch/doc.go

@@ -60,7 +60,7 @@
 //	                                     (default: 950ms)
 //	                                     [$PW_BATCHING_DELAY]
 //	--retention=                         If set, metrics older than that will
-//	                                     be deleted (default: 14)
+//	                                     be deleted (default: "14 days")
 //	                                     [$PW_RETENTION]
 //	--real-dbname-field=                 Tag key for real database name
 //	                                     (default: real_dbname)

docs/reference/cli_env.md

@@ -99,7 +99,7 @@ It reads the configuration from the specified sources and metrics, then begins c
 
 - `--retention=`
 
-    If set, metrics older than that will be deleted (default: 14).  
+    If set, metrics older than that will be deleted (default: "14 days").  
     ENV: `$PW_RETENTION`
 
 - `--real-dbname-field=`

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"`
-	Retention             int           `long:"retention" mapstructure:"retention" description:"If set, metrics older than that will be deleted" default:"14" env:"PW_RETENTION"`
+	Retention             string        `long:"retention" mapstructure:"retention" description:"If set, metrics older than that will be deleted" default:"14 days" env:"PW_RETENTION"`
 	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"`
 	SystemIdentifierField string        `long:"system-identifier-field" mapstructure:"system-identifier-field" description:"Tag key for system identifier value" env:"PW_SYSTEM_IDENTIFIER_FIELD" default:"sys_id"`
 }

internal/sinks/postgres.go

@@ -43,7 +43,11 @@ func NewWriterFromPostgresConn(ctx context.Context, conn db.PgxPoolIface, opts *
 		lastError: make(chan error),
 		sinkDb:    conn,
 	}
+	var runDeleteOldPartitions bool
 	if err = db.Init(ctx, pgw.sinkDb, func(ctx context.Context, conn db.PgxIface) error {
+		if err = conn.QueryRow(ctx, "SELECT $1::interval > '0'::interval", opts.Retention).Scan(&runDeleteOldPartitions); err != nil {
+			return err
+		}
 		l.Info("initialising measurements database...")
 		exists, err := db.DoesSchemaExist(ctx, conn, "admin")
 		if err != nil || exists {
@@ -64,7 +68,9 @@ func NewWriterFromPostgresConn(ctx context.Context, conn db.PgxPoolIface, opts *
 	if err = pgw.EnsureBuiltinMetricDummies(); err != nil {
 		return
 	}
-	go pgw.deleteOldPartitions()
+	if runDeleteOldPartitions {
+		go pgw.deleteOldPartitions()
+	}
 	go pgw.maintainUniqueSources()
 	go pgw.poll()
 	l.Info(`measurements sink is activated`)
@@ -452,14 +458,11 @@ func (pgw *PostgresWriter) EnsureMetricDbnameTime(metricDbnamePartBounds map[str
 
 // deleteOldPartitions is a background task that deletes old partitions from the measurements DB
 func (pgw *PostgresWriter) deleteOldPartitions() {
-	if pgw.opts.Retention <= 0 {
-		return
-	}
 	l := log.GetLogger(pgw.ctx)
 	for {
 		var partsDropped int
-		err := pgw.sinkDb.QueryRow(pgw.ctx, `SELECT admin.drop_old_time_partitions(older_than => $1)`,
-			24*time.Hour*time.Duration(pgw.opts.Retention)).Scan(&partsDropped)
+		err := pgw.sinkDb.QueryRow(pgw.ctx, `SELECT admin.drop_old_time_partitions(older_than => $1::interval)`,
+			pgw.opts.Retention).Scan(&partsDropped)
 		if err != nil {
 			l.Error("Could not drop old time partitions:", err)
 		} else if partsDropped > 0 {

internal/sinks/postgres_test.go

@@ -43,13 +43,14 @@ func TestNewWriterFromPostgresConn(t *testing.T) {
 	assert.NoError(t, err)
 
 	conn.ExpectPing()
+	conn.ExpectQuery("SELECT \\$1::interval").WithArgs("1 year").WillReturnRows(pgxmock.NewRows([]string{"col"}).AddRow(true))
 	conn.ExpectQuery("SELECT EXISTS").WithArgs("admin").WillReturnRows(pgxmock.NewRows([]string{"schema_type"}).AddRow(true))
 	conn.ExpectQuery("SELECT schema_type").WillReturnRows(pgxmock.NewRows([]string{"schema_type"}).AddRow(true))
 	for _, m := range metrics.GetDefaultBuiltInMetrics() {
 		conn.ExpectExec("SELECT admin.ensure_dummy_metrics_table").WithArgs(m).WillReturnResult(pgxmock.NewResult("EXECUTE", 1))
 	}
 
-	opts := &CmdOpts{BatchingDelay: time.Hour, Retention: 356}
+	opts := &CmdOpts{BatchingDelay: time.Hour, Retention: "1 year"}
 	pgw, err := NewWriterFromPostgresConn(ctx, conn, opts)
 	assert.NoError(t, err)
 	assert.NotNil(t, pgw)