Merge pull request #569 from flanksource/feat/job-history-cleanup

feat: clean old job histories

db/job_history.go

@@ -1,6 +1,7 @@
 package db
 
 import (
+	"github.com/flanksource/commons/logger"
 	"github.com/flanksource/duty/models"
 	"github.com/flanksource/incident-commander/api"
 	"github.com/google/uuid"
@@ -15,16 +16,27 @@ func PersistJobHistory(ctx api.Context, h *models.JobHistory) error {
 	return ctx.DB().Table("job_history").Save(h).Error
 }
 
-func DeleteOldJobHistoryRows(ctx api.Context, keepLatest int) error {
-	return ctx.DB().Exec(`
+func DeleteOldJobHistoryRows(ctx api.Context, keepLatestSuccess, keepLatestFailed int) error {
+	query := `
     WITH ordered_history AS (
-        SELECT
-            id, resource_type, resource_id, name, created_at,
-            ROW_NUMBER() OVER (PARTITION by resource_id, resource_type, name ORDER BY created_at DESC)
-        FROM job_history
+      SELECT
+        id, 
+        status,
+        ROW_NUMBER() OVER (PARTITION by resource_id, name, status ORDER BY created_at DESC)
+      FROM job_history
     )
     DELETE FROM job_history WHERE id IN (
-        SELECT id FROM ordered_history WHERE row_number > ?
-    )
-    `, keepLatest).Error
+      SELECT id FROM ordered_history WHERE 
+        (row_number > ? AND status IN (?, ?))
+        OR (row_number > ? AND status IN (?, ?))
+    )`
+
+	res := ctx.DB().Exec(query,
+		keepLatestSuccess, models.StatusSuccess, models.StatusFinished,
+		keepLatestFailed, models.StatusFailed, models.StatusWarning)
+	if res.RowsAffected > 0 {
+		logger.Infof("deleted %d job history rows", res.RowsAffected)
+	}
+
+	return res.Error
 }

jobs/job_history_cleanup.go

@@ -7,7 +7,7 @@ import (
 )
 
 func CleanupJobHistoryTable() {
-	if err := db.DeleteOldJobHistoryRows(api.DefaultContext, 5); err != nil {
+	if err := db.DeleteOldJobHistoryRows(api.DefaultContext, 3, 10); err != nil {
 		logger.Errorf("Error deleting old job history rows: %v", err)
 	}
 }

upstream/reconcile.go

@@ -23,7 +23,7 @@ func SyncWithUpstream(ctx api.Context) error {
 	for _, table := range api.TablesToReconcile {
 		if err := reconciler.Sync(ctx, table); err != nil {
 			jobHistory.AddError(err.Error())
-			logger.Errorf("failed to sync table %s: %w", table, err)
+			logger.Errorf("failed to sync table %s: %v", table, err)
 		} else {
 			jobHistory.IncrSuccess()
 		}