feat(cloud-backups): support rotating a non-audit table (metrics_audit)

cloud-backups/cmd/pg_audit_rotate.go

@@ -164,22 +164,34 @@ func runPGAuditRotate() error {
 	// live-read data on a deployment that does have them; --drop-instance-rows is the
 	// conscious override. Only meaningful pre-rotation (the fresh table has none), so
 	// on the steady-state cron this is a cheap count returning 0.
+	//
+	// This guard is SPECIFIC to the generic audit table's frozen-instances semantics:
+	// it only applies to a table that HAS an entity_name column. Another rotated table
+	// (e.g. metrics_audit) has no such column and no such rows, so the guard is skipped
+	// there -- and running the count unconditionally would error on the missing column.
 	if !cfg.DropInstanceRows {
-		rows, err := pgClient.QueryRows(ctx, countInstancesSQL(cfg.PGSchema, cfg.AuditTable))
+		hasEntityName, err := pgClient.QueryRows(ctx, columnExistsSQL(cfg.PGSchema, cfg.AuditTable, "entity_name"))
 		if err != nil {
-			slog.Error("instances_precount_failed", "error", err.Error())
+			slog.Error("entity_name_column_check_failed", "error", err.Error())
 			return err
 		}
-		// Proceed only on a definitive count of exactly 0; refuse on any rows OR an
-		// unexpected result shape (never assume zero from an ambiguous answer).
-		if len(rows) != 1 || rows[0] != "0" {
-			got := "an unexpected count result"
-			if len(rows) == 1 {
-				got = rows[0] + " entity_name='instances' row(s)"
+		if len(hasEntityName) == 1 && hasEntityName[0] == "t" {
+			rows, err := pgClient.QueryRows(ctx, countInstancesSQL(cfg.PGSchema, cfg.AuditTable))
+			if err != nil {
+				slog.Error("instances_precount_failed", "error", err.Error())
+				return err
+			}
+			// Proceed only on a definitive count of exactly 0; refuse on any rows OR an
+			// unexpected result shape (never assume zero from an ambiguous answer).
+			if len(rows) != 1 || rows[0] != "0" {
+				got := "an unexpected count result"
+				if len(rows) == 1 {
+					got = rows[0] + " entity_name='instances' row(s)"
+				}
+				err := fmt.Errorf("%s.%s returned %s still read by the app (InstanceService); audit-rotate does not carry them forward, so the app's instance-revision reads would return empty once they age out of the DB (the rows are still backed up to the permanent bucket). Set --drop-instance-rows to proceed as a conscious cutover", cfg.PGSchema, cfg.AuditTable, got)
+				slog.Error("instances_rows_present_refusing", "error", err.Error())
+				return err
 			}
-			err := fmt.Errorf("%s.%s returned %s still read by the app (InstanceService); audit-rotate does not carry them forward, so the app's instance-revision reads would return empty once they age out of the DB (the rows are still backed up to the permanent bucket). Set --drop-instance-rows to proceed as a conscious cutover", cfg.PGSchema, cfg.AuditTable, got)
-			slog.Error("instances_rows_present_refusing", "error", err.Error())
-			return err
 		}
 	}
 
@@ -697,6 +709,14 @@ func countInstancesSQL(schema, audit string) string {
 	return fmt.Sprintf("SELECT count(*) FROM %s.%s WHERE entity_name = 'instances';", schema, audit)
 }
 
+// columnExistsSQL returns 't'/'f' for whether the table has the named column. Used to
+// scope the entity_name-specific instances preflight to tables that actually have that
+// column (the generic audit table); a table without it -- e.g. metrics_audit -- skips the
+// guard rather than erroring on the missing column.
+func columnExistsSQL(schema, table, column string) string {
+	return fmt.Sprintf("SELECT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema = '%s' AND table_name = '%s' AND column_name = '%s');", schema, table, column)
+}
+
 // nonOwnerGrantsSQL lists the roles (or PUBLIC) that hold a GRANT on the audit table
 // other than its owner. CREATE TABLE ... LIKE ... INCLUDING ALL copies neither table
 // ownership nor ACLs, so any such grant silently vanishes on the fresh table after a

cloud-backups/cmd/pg_audit_rotate_test.go

@@ -234,6 +234,18 @@ func TestCountInstancesSQL(t *testing.T) {
 	}
 }
 
+// columnExistsSQL scopes the entity_name instances guard to tables that have the column
+// (the generic audit table), so a table without it (e.g. metrics_audit) skips the guard
+// instead of erroring on the missing column.
+func TestColumnExistsSQL(t *testing.T) {
+	got := columnExistsSQL("rearm", "metrics_audit", "entity_name")
+	for _, want := range []string{"SELECT EXISTS", "information_schema.columns", "table_schema = 'rearm'", "table_name = 'metrics_audit'", "column_name = 'entity_name'"} {
+		if !strings.Contains(got, want) {
+			t.Errorf("columnExistsSQL missing %q in:\n%s", want, got)
+		}
+	}
+}
+
 func TestNonOwnerGrantsSQL(t *testing.T) {
 	got := nonOwnerGrantsSQL("rearm", "audit")
 	for _, want := range []string{"aclexplode(c.relacl)", "'rearm.audit'::regclass", "acl.grantee <> c.relowner", "'PUBLIC'"} {