Merge branch 'main' into unified-storage-sql-garbage-collection

pkg/storage/unified/sql/backend.go

@@ -242,6 +242,23 @@ func (b *backend) initPruner(ctx context.Context) error {
 	return nil
 }
 
+func (b *backend) initGarbageCollection(ctx context.Context) error {
+	// TODO
+
+	/*
+		# Garbage collect every minute?
+
+		gr = getGroupResources()
+		for each gr in grs:
+		  rows = garbageCollect(gr)
+		  log rows deleted
+		  sleep 1s
+
+	*/
+
+	return nil
+}
+
 func (b *backend) IsHealthy(ctx context.Context, _ *resourcepb.HealthCheckRequest) (*resourcepb.HealthCheckResponse, error) {
 	// ctxLogger := s.log.FromContext(log.WithContextualAttributes(ctx, []any{"method", "isHealthy"}))
 

pkg/storage/unified/sql/data/resource_history_gc_batch.sql

@@ -0,0 +1,13 @@
+DELETE FROM {{ .Ident "resource_history" }} h
+WHERE h.{{ .Ident "group" }} = {{ .Arg .Group }}
+  AND h.{{ .Ident "resource" }} = {{ .Arg .Resource }}
+  AND h.{{ .Ident "action" }} = 3
+  AND h.{{ .Ident "resource_version" }} < {{ .Arg .CutoffTimestamp }}
+  AND NOT EXISTS (
+    SELECT 1 FROM {{ .Ident "resource" }} r
+    WHERE r.{{ .Ident "namespace" }} = h.{{ .Ident "namespace" }}
+      AND r.{{ .Ident "group" }} = h.{{ .Ident "group" }}
+      AND r.{{ .Ident "resource" }} = h.{{ .Ident "resource" }}
+      AND r.{{ .Ident "name" }} = h.{{ .Ident "name" }}
+  )
+LIMIT {{ .Arg .BatchSize }};

pkg/storage/unified/sql/db/migrations/resource_mig.go

@@ -220,5 +220,11 @@ func initResourceTables(mg *migrator.Migrator) string {
 	mg.AddMigration("Change key_path collation of resource_history in postgres", migrator.NewRawSQLMigration("").Postgres(`ALTER TABLE resource_history ALTER COLUMN key_path TYPE VARCHAR(2048) COLLATE "C";`))
 	mg.AddMigration("Change key_path collation of resource_events in postgres", migrator.NewRawSQLMigration("").Postgres(`ALTER TABLE resource_events ALTER COLUMN key_path TYPE VARCHAR(2048) COLLATE "C";`))
 
+	mg.AddMigration("Add index to resource_history for garbage collection", migrator.NewAddIndexMigration(resource_history_table, &migrator.Index{
+		Cols: []string{"group", "resource", "action", "resource_version", "name"},
+		Type: migrator.IndexType,
+		Name: "IDX_resource_history_resource_action_version_name",
+	}))
+
 	return marker
 }