[*] fix maintenance task to delete entries of dropped tables (#1020)

* introduce `--maintenance-interval` flag

* modify `--retention` to accept PostgreSQL interval string.

* Use new flags to control deletion and maintenance tasks.

* Test `MaintainUniqueSources()` and `DeleteOldPartitions()`

* update docs reference page to mention new flag.

* run `DeleteOldPartitions()` with `retentionInterval` not `maintenanceInterval`

* Rename test

* Capitalize `set` in flag description.

* Fix typo in comment

* Update maintenance task to remove entries of dropped metric tables from listing table.

* Test removing entries of dropped tables from the listing table.

* Add empty line at end of file

* Ensure there is no error dropping table

* Use `!=` operator instead of `NOT ... = ...`

* Preserve `public.` prefix in `sqlDistinct` query

* substitue `break` with `continue` to avoid breaking
the loop without trimming all elements in `allDistinctMetricTables`

* this is more informative in the logs also.

Author: 0xgouda

internal/sinks/postgres.go

@@ -520,6 +520,7 @@ func (pgw *PostgresWriter) MaintainUniqueSources() {
 	SELECT dbname FROM t WHERE dbname NOTNULL ORDER BY 1`
 	sqlDelete := `DELETE FROM admin.all_distinct_dbname_metrics WHERE NOT dbname = ANY($1) and metric = $2`
 	sqlDeleteAll := `DELETE FROM admin.all_distinct_dbname_metrics WHERE metric = $1`
+	sqlDroppedTables := `DELETE FROM admin.all_distinct_dbname_metrics WHERE metric != ALL($1)`
 	sqlAdd := `
 		INSERT INTO admin.all_distinct_dbname_metrics 
 		SELECT u, $2 FROM (select unnest($1::text[]) as u) x
@@ -545,17 +546,20 @@ func (pgw *PostgresWriter) MaintainUniqueSources() {
 		return
 	}
 
-	for _, tableName := range allDistinctMetricTables {
+	for i, tableName := range allDistinctMetricTables {
 		foundDbnamesMap := make(map[string]bool)
 		foundDbnamesArr := make([]string, 0)
+
 		metricName := strings.Replace(tableName, "public.", "", 1)
+		// later usage in sqlDroppedTables requires no "public." prefix
+		allDistinctMetricTables[i] = metricName 
 
 		logger.Debugf("Refreshing all_distinct_dbname_metrics listing for metric: %s", metricName)
 		rows, _ := pgw.sinkDb.Query(pgw.ctx, fmt.Sprintf(sqlDistinct, tableName, tableName))
 		ret, err := pgx.CollectRows(rows, pgx.RowTo[string])
 		if err != nil {
-			logger.Errorf("Could not refresh Postgres all_distinct_dbname_metrics listing table for '%s': %s", metricName, err)
-			break
+			logger.Errorf("Could not refresh Postgres all_distinct_dbname_metrics listing table for metric '%s': %s", metricName, err)
+			continue
 		}
 		for _, drDbname := range ret {
 			foundDbnamesMap[drDbname] = true // "set" behaviour, don't want duplicates
@@ -589,6 +593,12 @@ func (pgw *PostgresWriter) MaintainUniqueSources() {
 		time.Sleep(time.Minute)
 	}
 
+	cmdTag, err := pgw.sinkDb.Exec(pgw.ctx, sqlDroppedTables, allDistinctMetricTables)
+	if err != nil {
+		logger.Errorf("Could not refresh Postgres all_distinct_dbname_metrics listing table for dropped metric tables: %s", err)
+	} else if cmdTag.RowsAffected() > 0 {
+		logger.Infof("Removed %d stale entries for dropped tables from all_distinct_dbname_metrics listing table", cmdTag.RowsAffected())
+	}
 }
 
 func (pgw *PostgresWriter) AddDBUniqueMetricToListingTable(dbUnique, metric string) error {

internal/sinks/postgres_test.go

@@ -704,6 +704,15 @@ func Test_MaintainUniqueSources_DeleteOldPartitions(t *testing.T) {
 		err = conn.QueryRow(ctx, "SELECT count(*) FROM admin.all_distinct_dbname_metrics;").Scan(&numOfEntries)
 		a.NoError(err)
 		a.Equal(1, numOfEntries)
+
+		_, err = conn.Exec(ctx, "DROP TABLE test_metric_1;")
+		r.NoError(err)
+
+		// the corresponding entry should be deleted
+		pgw.MaintainUniqueSources()
+		err = conn.QueryRow(ctx, "SELECT count(*) FROM admin.all_distinct_dbname_metrics;").Scan(&numOfEntries)
+		a.NoError(err)
+		a.Equal(0, numOfEntries)
 	})
 
 	t.Run("DeleteOldPartitions", func(_ *testing.T) {