diff --git a/internal/k8s/longhorn_backups.go b/internal/k8s/longhorn_backups.go index bb62590..ad23038 100644 --- a/internal/k8s/longhorn_backups.go +++ b/internal/k8s/longhorn_backups.go @@ -24,12 +24,36 @@ var longhornBackupGVR = schema.GroupVersionResource{ type LonghornBackupSummary struct { Name string SnapshotName string + Namespace string + PVC string State string Error string Progress int64 CreatedAt string } +// ListLonghornBackups returns Soteria-owned Longhorn Backup CRDs. +func (c *Client) ListLonghornBackups(ctx context.Context) ([]LonghornBackupSummary, error) { + if c.Dynamic == nil { + return nil, fmt.Errorf("dynamic Kubernetes client is unavailable") + } + + list, err := c.Dynamic.Resource(longhornBackupGVR).Namespace(longhornSystemNamespace).List(ctx, metav1.ListOptions{}) + if err != nil { + return nil, fmt.Errorf("list Longhorn backups: %w", err) + } + + items := make([]LonghornBackupSummary, 0, len(list.Items)) + for _, item := range list.Items { + summary := summarizeLonghornBackup(item) + if summary.Namespace != "" && summary.PVC != "" { + items = append(items, summary) + } + } + sortLonghornBackupsNewestFirst(items) + return items, nil +} + // GetLonghornBackupBySnapshot returns the newest Longhorn Backup CRD for a snapshot. func (c *Client) GetLonghornBackupBySnapshot(ctx context.Context, snapshotName string) (LonghornBackupSummary, bool, error) { snapshotName = strings.TrimSpace(snapshotName) @@ -56,12 +80,7 @@ func (c *Client) GetLonghornBackupBySnapshot(ctx context.Context, snapshotName s return LonghornBackupSummary{}, false, nil } - sort.Slice(matches, func(i, j int) bool { - if matches[i].CreatedAt == matches[j].CreatedAt { - return matches[i].Name > matches[j].Name - } - return matches[i].CreatedAt > matches[j].CreatedAt - }) + sortLonghornBackupsNewestFirst(matches) return matches[0], true, nil } @@ -70,6 +89,7 @@ func summarizeLonghornBackup(item unstructured.Unstructured) LonghornBackupSumma state, _, _ := unstructured.NestedString(item.Object, "status", "state") errorMessage, _, _ := unstructured.NestedString(item.Object, "status", "error") progress, _, _ := unstructured.NestedInt64(item.Object, "status", "progress") + specLabels, _, _ := unstructured.NestedStringMap(item.Object, "spec", "labels") createdAt, _, _ := unstructured.NestedString(item.Object, "status", "backupCreatedAt") if strings.TrimSpace(createdAt) == "" { createdAt, _, _ = unstructured.NestedString(item.Object, "status", "snapshotCreatedAt") @@ -81,9 +101,20 @@ func summarizeLonghornBackup(item unstructured.Unstructured) LonghornBackupSumma return LonghornBackupSummary{ Name: item.GetName(), SnapshotName: strings.TrimSpace(snapshotName), + Namespace: strings.TrimSpace(specLabels["soteria.bstein.dev/namespace"]), + PVC: strings.TrimSpace(specLabels["soteria.bstein.dev/pvc"]), State: strings.TrimSpace(state), Error: strings.TrimSpace(errorMessage), Progress: progress, CreatedAt: strings.TrimSpace(createdAt), } } + +func sortLonghornBackupsNewestFirst(items []LonghornBackupSummary) { + sort.Slice(items, func(i, j int) bool { + if items[i].CreatedAt == items[j].CreatedAt { + return items[i].Name > items[j].Name + } + return items[i].CreatedAt > items[j].CreatedAt + }) +} diff --git a/internal/k8s/longhorn_backups_test.go b/internal/k8s/longhorn_backups_test.go index 5acb9cb..15738f6 100644 --- a/internal/k8s/longhorn_backups_test.go +++ b/internal/k8s/longhorn_backups_test.go @@ -66,6 +66,45 @@ func TestGetLonghornBackupBySnapshotWrapsListErrorAndFallbackTimestamp(t *testin } } +func TestListLonghornBackupsFiltersSoteriaLabelsAndSorts(t *testing.T) { + older := longhornBackupObject("backup-older", "snap-a", "Completed", "", 100, "2026-04-20T10:00:00Z") + newer := longhornBackupObject("backup-newer", "snap-b", "Error", "storage cap exceeded", 0, "2026-04-20T11:00:00Z") + setLonghornBackupLabels(newer, "ops", "cache") + unlabeled := longhornBackupObject("backup-unlabeled", "snap-c", "Completed", "", 100, "2026-04-20T12:00:00Z") + _ = unstructured.SetNestedStringMap(unlabeled.Object, map[string]string{}, "spec", "labels") + + client := &Client{Dynamic: newLonghornBackupDynamic(older, newer, unlabeled)} + backups, err := client.ListLonghornBackups(context.Background()) + if err != nil { + t.Fatalf("list Longhorn backups: %v", err) + } + if len(backups) != 2 { + t.Fatalf("expected two Soteria-labeled backups, got %#v", backups) + } + if backups[0].Name != "backup-newer" || backups[0].Namespace != "ops" || backups[0].PVC != "cache" { + t.Fatalf("expected newest ops/cache backup first, got %#v", backups[0]) + } + if backups[1].Name != "backup-older" || backups[1].Namespace != "apps" || backups[1].PVC != "data" { + t.Fatalf("expected older apps/data backup second, got %#v", backups[1]) + } +} + +func TestListLonghornBackupsValidatesClientAndWrapsListError(t *testing.T) { + client := &Client{} + if _, err := client.ListLonghornBackups(context.Background()); err == nil || !strings.Contains(err.Error(), "dynamic Kubernetes client is unavailable") { + t.Fatalf("expected missing dynamic client error, got %v", err) + } + + failingDynamic := newLonghornBackupDynamic() + failingDynamic.PrependReactor("list", "backups", func(k8stesting.Action) (bool, runtime.Object, error) { + return true, nil, errors.New("list exploded") + }) + client.Dynamic = failingDynamic + if _, err := client.ListLonghornBackups(context.Background()); err == nil || !strings.Contains(err.Error(), "list exploded") { + t.Fatalf("expected wrapped list error, got %v", err) + } +} + func newLonghornBackupDynamic(objects ...runtime.Object) *dynamicfake.FakeDynamicClient { return dynamicfake.NewSimpleDynamicClientWithCustomListKinds( runtime.NewScheme(), @@ -74,6 +113,13 @@ func newLonghornBackupDynamic(objects ...runtime.Object) *dynamicfake.FakeDynami ) } +func setLonghornBackupLabels(item *unstructured.Unstructured, namespace, pvc string) { + _ = unstructured.SetNestedStringMap(item.Object, map[string]string{ + "soteria.bstein.dev/namespace": namespace, + "soteria.bstein.dev/pvc": pvc, + }, "spec", "labels") +} + func longhornBackupObject(name, snapshotName, state, errorMessage string, progress int64, createdAt string) *unstructured.Unstructured { item := &unstructured.Unstructured{Object: map[string]any{ "apiVersion": "longhorn.io/v1beta2", @@ -84,6 +130,10 @@ func longhornBackupObject(name, snapshotName, state, errorMessage string, progre }, "spec": map[string]any{ "snapshotName": snapshotName, + "labels": map[string]any{ + "soteria.bstein.dev/namespace": "apps", + "soteria.bstein.dev/pvc": "data", + }, }, "status": map[string]any{ "state": state, diff --git a/internal/server/inventory_builder.go b/internal/server/inventory_builder.go index 6426aa8..ba6b652 100644 --- a/internal/server/inventory_builder.go +++ b/internal/server/inventory_builder.go @@ -18,6 +18,7 @@ func (s *Server) buildInventory(ctx context.Context) (api.InventoryResponse, err } resticJobsByPVC, resticLookupErrors := s.prefetchResticBackupJobs(ctx, pvcs) + longhornBackupsByPVC, longhornLookupErr := s.prefetchLonghornBackupCRDs(ctx) groups := make(map[string][]api.PVCInventory) for _, summary := range pvcs { @@ -38,7 +39,7 @@ func (s *Server) buildInventory(ctx context.Context) (api.InventoryResponse, err groups[summary.Namespace] = append(groups[summary.Namespace], entry) continue } - s.enrichPVCInventory(ctx, &entry, resticJobsByPVC, resticLookupErrors) + s.enrichPVCInventory(ctx, &entry, resticJobsByPVC, resticLookupErrors, longhornBackupsByPVC, longhornLookupErr) groups[summary.Namespace] = append(groups[summary.Namespace], entry) } @@ -98,22 +99,59 @@ func (s *Server) prefetchResticBackupJobs(ctx context.Context, pvcs []k8s.PVCSum return jobsByPVC, lookupErrors } +func (s *Server) prefetchLonghornBackupCRDs(ctx context.Context) (map[string][]k8s.LonghornBackupSummary, error) { + if s.cfg.BackupDriver != "longhorn" { + return nil, nil + } + + backups, err := s.client.ListLonghornBackups(ctx) + if err != nil { + return nil, err + } + + backupsByPVC := map[string][]k8s.LonghornBackupSummary{} + for _, backup := range backups { + if backup.Namespace == "" || backup.PVC == "" { + continue + } + key := backup.Namespace + "/" + backup.PVC + backupsByPVC[key] = append(backupsByPVC[key], backup) + } + for key := range backupsByPVC { + sortLonghornBackupsNewestFirst(backupsByPVC[key]) + } + return backupsByPVC, nil +} + func (s *Server) enrichPVCInventory( ctx context.Context, entry *api.PVCInventory, resticJobsByPVC map[string][]k8s.BackupJobSummary, resticLookupErrors map[string]error, + longhornBackupsByPVC map[string][]k8s.LonghornBackupSummary, + longhornLookupErr error, ) { switch s.cfg.BackupDriver { case "longhorn": + backupCRDs := []k8s.LonghornBackupSummary{} + if longhornLookupErr == nil && longhornBackupsByPVC != nil { + backupCRDs = longhornBackupsByPVC[entry.Namespace+"/"+entry.PVC] + applyLonghornBackupCRDStatus(entry, backupCRDs) + } backups, err := s.longhorn.ListBackups(ctx, entry.Volume) if err != nil { entry.Healthy = false entry.HealthReason = "lookup_failed" entry.Error = err.Error() + if longhornLookupErr != nil { + entry.Error = entry.Error + "; " + longhornLookupErr.Error() + } return } entry.BackupCount = len(backups) + if len(backupCRDs) > entry.BackupCount { + entry.BackupCount = len(backupCRDs) + } totalBackupSize := int64(0) completedBackups := 0 for _, backup := range backups { @@ -132,6 +170,9 @@ func (s *Server) enrichPVCInventory( } else { entry.HealthReason = "no_completed" } + if longhornLookupErr != nil { + entry.Error = longhornLookupErr.Error() + } return } entry.LastBackupAt = latest.Created @@ -242,6 +283,52 @@ func (s *Server) enrichPVCInventory( } } +func sortLonghornBackupsNewestFirst(items []k8s.LonghornBackupSummary) { + sort.Slice(items, func(i, j int) bool { + if items[i].CreatedAt == items[j].CreatedAt { + return items[i].Name > items[j].Name + } + return items[i].CreatedAt > items[j].CreatedAt + }) +} + +func applyLonghornBackupCRDStatus(entry *api.PVCInventory, backups []k8s.LonghornBackupSummary) { + if len(backups) == 0 { + return + } + entry.LastJobName = backups[0].Name + entry.LastJobState = backups[0].State + entry.LastJobStartedAt = backups[0].CreatedAt + entry.LastJobProgressPct = longhornBackupProgressPct(backups[0]) + + active := 0 + for _, backup := range backups { + switch strings.ToLower(strings.TrimSpace(backup.State)) { + case "completed", "error", "failed": + continue + default: + active++ + } + } + entry.ActiveBackups = active +} + +func longhornBackupProgressPct(backup k8s.LonghornBackupSummary) int { + switch strings.ToLower(strings.TrimSpace(backup.State)) { + case "completed": + return 100 + case "error", "failed": + return 100 + } + if backup.Progress > 0 { + if backup.Progress > 100 { + return 100 + } + return int(backup.Progress) + } + return 20 +} + func sortBackupJobsNewestFirst(items []k8s.BackupJobSummary) { sort.Slice(items, func(i, j int) bool { left := items[i].CompletionTime diff --git a/internal/server/inventory_builder_test.go b/internal/server/inventory_builder_test.go index 8780cad..2fae6d8 100644 --- a/internal/server/inventory_builder_test.go +++ b/internal/server/inventory_builder_test.go @@ -117,6 +117,86 @@ func TestBuildInventoryLonghornSortsNamespacesAndCalculatesHealth(t *testing.T) } } +func TestBuildInventoryLonghornTracksFailedBackupCRDAttempts(t *testing.T) { + recent := time.Now().UTC().Add(-30 * time.Minute).Format(time.RFC3339) + client := &inventoryTestKubeClient{ + fakeKubeClient: &fakeKubeClient{ + pvcs: []k8s.PVCSummary{ + {Namespace: "apps", Name: "data", VolumeName: "vol-data", Phase: "Bound", StorageClass: "fast"}, + }, + longhornBackups: []k8s.LonghornBackupSummary{ + { + Name: "backup-failed", + Namespace: "apps", + PVC: "data", + State: "Error", + Error: "storage cap exceeded", + CreatedAt: recent, + }, + }, + }, + } + longhornClient := &inventoryTestLonghornClient{ + fakeLonghornClient: &fakeLonghornClient{}, + listBackupsByVolume: map[string][]longhorn.Backup{ + "vol-data": {}, + }, + } + srv := newInventoryTestServer(&config.Config{ + BackupDriver: "longhorn", + BackupMaxAge: 24 * time.Hour, + }, client, longhornClient) + + inventory, err := srv.buildInventory(context.Background()) + if err != nil { + t.Fatalf("build longhorn inventory: %v", err) + } + entry := inventory.Namespaces[0].PVCs[0] + if entry.BackupCount != 1 || entry.CompletedBackups != 0 || entry.ActiveBackups != 0 { + t.Fatalf("expected failed CRD attempt to count without active/completed backups, got %#v", entry) + } + if entry.LastJobName != "backup-failed" || entry.LastJobState != "Error" || entry.LastJobStartedAt != recent { + t.Fatalf("expected latest failed CRD metadata, got %#v", entry) + } + if entry.LastJobProgressPct != 100 { + t.Fatalf("expected failed CRD to report terminal progress, got %#v", entry.LastJobProgressPct) + } + if entry.Healthy || entry.HealthReason != "missing" { + t.Fatalf("expected no completed backup health, got %#v", entry) + } +} + +func TestPrefetchLonghornBackupCRDsCoversDriverFilterAndErrors(t *testing.T) { + resticSrv := newInventoryTestServer(&config.Config{BackupDriver: "restic"}, &inventoryTestKubeClient{fakeKubeClient: &fakeKubeClient{}}, &inventoryTestLonghornClient{fakeLonghornClient: &fakeLonghornClient{}}) + if backups, err := resticSrv.prefetchLonghornBackupCRDs(context.Background()); err != nil || backups != nil { + t.Fatalf("expected non-longhorn driver to skip CRD prefetch, backups=%#v err=%v", backups, err) + } + + errSrv := newInventoryTestServer(&config.Config{BackupDriver: "longhorn"}, &inventoryTestKubeClient{fakeKubeClient: &fakeKubeClient{longhornBackupLookupErr: errors.New("list CRDs exploded")}}, &inventoryTestLonghornClient{fakeLonghornClient: &fakeLonghornClient{}}) + if _, err := errSrv.prefetchLonghornBackupCRDs(context.Background()); err == nil || !strings.Contains(err.Error(), "list CRDs exploded") { + t.Fatalf("expected CRD lookup error, got %v", err) + } + + client := &inventoryTestKubeClient{ + fakeKubeClient: &fakeKubeClient{ + longhornBackups: []k8s.LonghornBackupSummary{ + {Name: "older", Namespace: "apps", PVC: "data", CreatedAt: "2026-04-20T10:00:00Z"}, + {Name: "newer", Namespace: "apps", PVC: "data", CreatedAt: "2026-04-20T11:00:00Z"}, + {Name: "ignored", Namespace: "", PVC: "data", CreatedAt: "2026-04-20T12:00:00Z"}, + }, + }, + } + srv := newInventoryTestServer(&config.Config{BackupDriver: "longhorn"}, client, &inventoryTestLonghornClient{fakeLonghornClient: &fakeLonghornClient{}}) + backups, err := srv.prefetchLonghornBackupCRDs(context.Background()) + if err != nil { + t.Fatalf("prefetch Longhorn backup CRDs: %v", err) + } + got := backups["apps/data"] + if len(got) != 2 || got[0].Name != "newer" || got[1].Name != "older" { + t.Fatalf("expected grouped backups sorted newest first, got %#v", got) + } +} + func TestBuildInventoryMarksExcludedPVCsHealthy(t *testing.T) { client := &inventoryTestKubeClient{ fakeKubeClient: &fakeKubeClient{ @@ -182,7 +262,7 @@ func TestEnrichPVCInventoryCoversLonghornAndResticBranches(t *testing.T) { for _, tc := range testCases { entry := tc.entry.toAPI() - srv.enrichPVCInventory(context.Background(), &entry, nil, nil) + srv.enrichPVCInventory(context.Background(), &entry, nil, nil, nil, nil) if entry.HealthReason != tc.want || entry.Healthy != tc.ok { t.Fatalf("%s/%s: expected %q healthy=%v, got %#v", entry.Namespace, entry.PVC, tc.want, tc.ok, entry) } @@ -225,7 +305,7 @@ func TestEnrichPVCInventoryCoversLonghornAndResticBranches(t *testing.T) { } entry := apiPVCInventory{Namespace: "apps", PVC: "data", Volume: "vol-data"}.toAPI() - srv.enrichPVCInventory(context.Background(), &entry, resticJobsByPVC, nil) + srv.enrichPVCInventory(context.Background(), &entry, resticJobsByPVC, nil, nil, nil) if !entry.Healthy || entry.HealthReason != "fresh" { t.Fatalf("expected fresh restic pvc, got %#v", entry) } @@ -237,25 +317,25 @@ func TestEnrichPVCInventoryCoversLonghornAndResticBranches(t *testing.T) { } running := apiPVCInventory{Namespace: "apps", PVC: "running", Volume: "vol-running"}.toAPI() - srv.enrichPVCInventory(context.Background(), &running, resticJobsByPVC, nil) + srv.enrichPVCInventory(context.Background(), &running, resticJobsByPVC, nil, nil, nil) if running.Healthy || running.HealthReason != "in_progress" || running.ActiveBackups != 1 { t.Fatalf("expected in-progress restic pvc, got %#v", running) } missing := apiPVCInventory{Namespace: "apps", PVC: "missing", Volume: "vol-missing"}.toAPI() - srv.enrichPVCInventory(context.Background(), &missing, resticJobsByPVC, nil) + srv.enrichPVCInventory(context.Background(), &missing, resticJobsByPVC, nil, nil, nil) if missing.Healthy || missing.HealthReason != "missing" { t.Fatalf("expected missing restic pvc, got %#v", missing) } badTime := apiPVCInventory{Namespace: "apps", PVC: "bad-time", Volume: "vol-bad-time"}.toAPI() - srv.enrichPVCInventory(context.Background(), &badTime, resticJobsByPVC, nil) + srv.enrichPVCInventory(context.Background(), &badTime, resticJobsByPVC, nil, nil, nil) if badTime.Healthy || badTime.HealthReason != "unknown_timestamp" { t.Fatalf("expected unknown timestamp restic pvc, got %#v", badTime) } lookupFailed := apiPVCInventory{Namespace: "ops", PVC: "data", Volume: "vol-ops"}.toAPI() - srv.enrichPVCInventory(context.Background(), &lookupFailed, resticJobsByPVC, map[string]error{"ops": errors.New("list jobs exploded")}) + srv.enrichPVCInventory(context.Background(), &lookupFailed, resticJobsByPVC, map[string]error{"ops": errors.New("list jobs exploded")}, nil, nil) if lookupFailed.Healthy || lookupFailed.HealthReason != "lookup_failed" || lookupFailed.Error != "list jobs exploded" { t.Fatalf("expected lookup failure restic pvc, got %#v", lookupFailed) } @@ -344,6 +424,38 @@ func TestSortBackupJobsNewestFirstUsesCompletionCreatedAndNameTiebreakers(t *tes } } +func TestLonghornBackupInventoryHelpers(t *testing.T) { + backups := []k8s.LonghornBackupSummary{ + {Name: "backup-a", CreatedAt: "2026-04-20T10:00:00Z"}, + {Name: "backup-c", CreatedAt: "2026-04-20T11:00:00Z"}, + {Name: "backup-b", CreatedAt: "2026-04-20T10:00:00Z"}, + } + sortLonghornBackupsNewestFirst(backups) + got := []string{backups[0].Name, backups[1].Name, backups[2].Name} + want := []string{"backup-c", "backup-b", "backup-a"} + for index := range want { + if got[index] != want[index] { + t.Fatalf("expected sorted Longhorn backups %v, got %v", want, got) + } + } + + testCases := []struct { + backup k8s.LonghornBackupSummary + want int + }{ + {backup: k8s.LonghornBackupSummary{State: "Completed"}, want: 100}, + {backup: k8s.LonghornBackupSummary{State: "Failed"}, want: 100}, + {backup: k8s.LonghornBackupSummary{State: "Running", Progress: 45}, want: 45}, + {backup: k8s.LonghornBackupSummary{State: "Running", Progress: 150}, want: 100}, + {backup: k8s.LonghornBackupSummary{State: "Running"}, want: 20}, + } + for _, tc := range testCases { + if got := longhornBackupProgressPct(tc.backup); got != tc.want { + t.Fatalf("expected progress %d for %#v, got %d", tc.want, tc.backup, got) + } + } +} + type apiPVCInventory struct { Namespace string PVC string diff --git a/internal/server/policy_cycle_test.go b/internal/server/policy_cycle_test.go index f51f878..8373878 100644 --- a/internal/server/policy_cycle_test.go +++ b/internal/server/policy_cycle_test.go @@ -325,3 +325,50 @@ func TestRunPolicyCycleSkipsDetachedLonghornVolumesWithoutConsumingLimit(t *test t.Fatalf("expected one successful policy backup, got %f", got) } } + +func TestRunPolicyCycleThrottlesRecentLonghornAttemptAfterStaleBackup(t *testing.T) { + now := time.Now().UTC() + stale := now.Add(-48 * time.Hour).Format(time.RFC3339) + recentAttempt := now.Add(-30 * time.Minute).Format(time.RFC3339) + client := &policyCycleTestKubeClient{ + inventoryTestKubeClient: &inventoryTestKubeClient{ + fakeKubeClient: &fakeKubeClient{ + pvcs: []k8s.PVCSummary{ + {Namespace: "apps", Name: "data", VolumeName: "vol-data", Phase: "Bound"}, + }, + longhornBackups: []k8s.LonghornBackupSummary{ + {Name: "backup-failed", Namespace: "apps", PVC: "data", State: "Error", CreatedAt: recentAttempt}, + }, + }, + }, + } + longhornClient := &inventoryTestLonghornClient{ + fakeLonghornClient: &fakeLonghornClient{}, + listBackupsByVolume: map[string][]longhorn.Backup{ + "vol-data": { + {Name: "backup-stale", Created: stale, State: "Completed", Size: "10"}, + }, + }, + } + srv := &Server{ + cfg: &config.Config{ + BackupDriver: "longhorn", + BackupMaxAge: 24 * time.Hour, + }, + client: client, + longhorn: longhornClient, + metrics: newTelemetry(), + policies: map[string]api.BackupPolicy{ + "apps__all": {ID: "apps__all", Namespace: "apps", IntervalHours: 6, Enabled: true, Dedupe: true}, + }, + } + + srv.runPolicyCycle(context.Background()) + + if longhornClient.createSnapshotName != "" { + t.Fatalf("expected recent failed attempt to throttle Longhorn policy backup, got snapshot %q", longhornClient.createSnapshotName) + } + if got := metricCount(srv.metrics.policyBackups, map[string]string{"result": "not_due"}); got != 1 { + t.Fatalf("expected not_due metric for recent failed attempt, got %f", got) + } +} diff --git a/internal/server/policy_runtime.go b/internal/server/policy_runtime.go index a7ba619..b243454 100644 --- a/internal/server/policy_runtime.go +++ b/internal/server/policy_runtime.go @@ -120,14 +120,7 @@ func (s *Server) runPolicyCycle(ctx context.Context) { continue } - lastRunRef := strings.TrimSpace(pvc.LastBackupAt) - if lastRunRef == "" { - // If no successful backup exists yet, fall back to the most recent job start - // so failed attempts are still throttled by interval_hours. - lastRunRef = strings.TrimSpace(pvc.LastJobStartedAt) - } - - if !backupDue(lastRunRef, effective.IntervalHours) { + if !backupDue(policyRunReference(pvc), effective.IntervalHours) { s.metrics.RecordPolicyBackup("not_due") continue } @@ -439,3 +432,21 @@ func backupDue(lastBackupAt string, intervalHours float64) bool { interval := time.Duration(intervalHours * float64(time.Hour)) return time.Since(timestamp) >= interval } + +func policyRunReference(pvc api.PVCInventory) string { + lastBackup := strings.TrimSpace(pvc.LastBackupAt) + lastJob := strings.TrimSpace(pvc.LastJobStartedAt) + backupTime, backupOK := parseBackupTime(lastBackup) + jobTime, jobOK := parseBackupTime(lastJob) + switch { + case backupOK && jobOK: + if jobTime.After(backupTime) { + return lastJob + } + return lastBackup + case jobOK: + return lastJob + default: + return lastBackup + } +} diff --git a/internal/server/policy_runtime_test.go b/internal/server/policy_runtime_test.go index 56b3168..d50d4a6 100644 --- a/internal/server/policy_runtime_test.go +++ b/internal/server/policy_runtime_test.go @@ -163,6 +163,18 @@ func TestPolicySliceFromMapAndBackupDueHelpers(t *testing.T) { if backupDue(time.Now().UTC().Add(-30*time.Minute).Format(time.RFC3339), 0) { t.Fatalf("expected default interval fallback to keep recent backup not due") } + + oldBackup := time.Now().UTC().Add(-48 * time.Hour).Format(time.RFC3339) + recentJob := time.Now().UTC().Add(-30 * time.Minute).Format(time.RFC3339) + if got := policyRunReference(api.PVCInventory{LastBackupAt: oldBackup, LastJobStartedAt: recentJob}); got != recentJob { + t.Fatalf("expected recent job attempt to win over stale backup, got %q", got) + } + if got := policyRunReference(api.PVCInventory{LastBackupAt: "not-a-time", LastJobStartedAt: recentJob}); got != recentJob { + t.Fatalf("expected valid job attempt to win over invalid backup timestamp, got %q", got) + } + if got := policyRunReference(api.PVCInventory{LastBackupAt: "not-a-time"}); got != "not-a-time" { + t.Fatalf("expected invalid backup timestamp to be preserved for due fallback, got %q", got) + } } func TestLoadPoliciesRejectsInvalidDocuments(t *testing.T) { diff --git a/internal/server/server.go b/internal/server/server.go index 99fd51c..698427f 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -26,6 +26,7 @@ type kubeClient interface { ListBoundPVCs(ctx context.Context) ([]k8s.PVCSummary, error) ListPVCMounts(ctx context.Context, namespace, pvcName string) ([]k8s.PVCMount, error) GetLonghornBackupBySnapshot(ctx context.Context, snapshotName string) (k8s.LonghornBackupSummary, bool, error) + ListLonghornBackups(ctx context.Context) ([]k8s.LonghornBackupSummary, error) PersistentVolumeClaimExists(ctx context.Context, namespace, pvcName string) (bool, error) LoadSecretData(ctx context.Context, namespace, secretName, key string) ([]byte, error) SaveSecretData(ctx context.Context, namespace, secretName, key string, value []byte, labels map[string]string) error diff --git a/internal/server/server_test.go b/internal/server/server_test.go index b262778..0f0b653 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -164,6 +164,14 @@ func (f *fakeKubeClient) GetLonghornBackupBySnapshot(_ context.Context, snapshot return k8s.LonghornBackupSummary{}, false, nil } +func (f *fakeKubeClient) ListLonghornBackups(_ context.Context) ([]k8s.LonghornBackupSummary, error) { + if f.longhornBackupLookupErr != nil { + return nil, f.longhornBackupLookupErr + } + items := append([]k8s.LonghornBackupSummary{}, f.longhornBackups...) + return items, nil +} + func (f *fakeKubeClient) PersistentVolumeClaimExists(_ context.Context, _, _ string) (bool, error) { return f.targetExists, nil }