diff --git a/internal/k8s/client.go b/internal/k8s/client.go index 60e5709..f034e44 100644 --- a/internal/k8s/client.go +++ b/internal/k8s/client.go @@ -4,6 +4,7 @@ import ( "fmt" "os" + "k8s.io/client-go/dynamic" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" "k8s.io/client-go/tools/clientcmd" @@ -12,6 +13,7 @@ import ( // Client wraps the Kubernetes interface used by Soteria. type Client struct { Clientset kubernetes.Interface + Dynamic dynamic.Interface } var ( @@ -20,6 +22,9 @@ var ( newForConfigFn = func(cfg *rest.Config) (kubernetes.Interface, error) { return kubernetes.NewForConfig(cfg) } + newDynamicForConfigFn = func(cfg *rest.Config) (dynamic.Interface, error) { + return dynamic.NewForConfig(cfg) + } ) // New returns a Kubernetes client, preferring in-cluster config and falling back to KUBECONFIG. @@ -40,6 +45,10 @@ func New() (*Client, error) { if err != nil { return nil, fmt.Errorf("build clientset: %w", err) } + dynamicClient, err := newDynamicForConfigFn(cfg) + if err != nil { + return nil, fmt.Errorf("build dynamic client: %w", err) + } - return &Client{Clientset: clientset}, nil + return &Client{Clientset: clientset, Dynamic: dynamicClient}, nil } diff --git a/internal/k8s/client_test.go b/internal/k8s/client_test.go index b646498..b315e4c 100644 --- a/internal/k8s/client_test.go +++ b/internal/k8s/client_test.go @@ -7,6 +7,7 @@ import ( "strings" "testing" + "k8s.io/client-go/dynamic" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" ) @@ -84,6 +85,9 @@ users: if client == nil || client.Clientset == nil { t.Fatalf("expected populated client, got %#v", client) } + if client.Dynamic == nil { + t.Fatalf("expected populated dynamic client, got %#v", client) + } } func TestNewWrapsClientsetConstructionFailures(t *testing.T) { @@ -106,14 +110,36 @@ func TestNewWrapsClientsetConstructionFailures(t *testing.T) { } } +func TestNewWrapsDynamicClientConstructionFailures(t *testing.T) { + restore := swapClientTestHooks() + defer restore() + + inClusterConfigFn = func() (*rest.Config, error) { + return &rest.Config{Host: "https://127.0.0.1:6443"}, nil + } + newDynamicForConfigFn = func(*rest.Config) (dynamic.Interface, error) { + return nil, errors.New("dynamic exploded") + } + + client, err := New() + if err == nil || client != nil { + t.Fatalf("expected dynamic client construction error, got client=%#v err=%v", client, err) + } + if !strings.Contains(err.Error(), "build dynamic client: dynamic exploded") { + t.Fatalf("expected wrapped dynamic client error, got %v", err) + } +} + func swapClientTestHooks() func() { originalInCluster := inClusterConfigFn originalBuildConfig := buildConfigFromFlagsFn originalNewForConfig := newForConfigFn + originalNewDynamicForConfig := newDynamicForConfigFn return func() { inClusterConfigFn = originalInCluster buildConfigFromFlagsFn = originalBuildConfig newForConfigFn = originalNewForConfig + newDynamicForConfigFn = originalNewDynamicForConfig } } diff --git a/internal/k8s/longhorn_backups.go b/internal/k8s/longhorn_backups.go new file mode 100644 index 0000000..bb62590 --- /dev/null +++ b/internal/k8s/longhorn_backups.go @@ -0,0 +1,89 @@ +package k8s + +import ( + "context" + "fmt" + "sort" + "strings" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" +) + +const longhornSystemNamespace = "longhorn-system" + +var longhornBackupGVR = schema.GroupVersionResource{ + Group: "longhorn.io", + Version: "v1beta2", + Resource: "backups", +} + +// LonghornBackupSummary is the subset of the Longhorn Backup CRD that Soteria +// needs while waiting for an async backup to reach a terminal state. +type LonghornBackupSummary struct { + Name string + SnapshotName string + State string + Error string + Progress int64 + CreatedAt string +} + +// 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) + if snapshotName == "" { + return LonghornBackupSummary{}, false, fmt.Errorf("snapshot name is required") + } + if c.Dynamic == nil { + return LonghornBackupSummary{}, false, fmt.Errorf("dynamic Kubernetes client is unavailable") + } + + list, err := c.Dynamic.Resource(longhornBackupGVR).Namespace(longhornSystemNamespace).List(ctx, metav1.ListOptions{}) + if err != nil { + return LonghornBackupSummary{}, false, fmt.Errorf("list Longhorn backups: %w", err) + } + + matches := make([]LonghornBackupSummary, 0, len(list.Items)) + for _, item := range list.Items { + summary := summarizeLonghornBackup(item) + if summary.SnapshotName == snapshotName { + matches = append(matches, summary) + } + } + if len(matches) == 0 { + 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 + }) + return matches[0], true, nil +} + +func summarizeLonghornBackup(item unstructured.Unstructured) LonghornBackupSummary { + snapshotName, _, _ := unstructured.NestedString(item.Object, "spec", "snapshotName") + state, _, _ := unstructured.NestedString(item.Object, "status", "state") + errorMessage, _, _ := unstructured.NestedString(item.Object, "status", "error") + progress, _, _ := unstructured.NestedInt64(item.Object, "status", "progress") + createdAt, _, _ := unstructured.NestedString(item.Object, "status", "backupCreatedAt") + if strings.TrimSpace(createdAt) == "" { + createdAt, _, _ = unstructured.NestedString(item.Object, "status", "snapshotCreatedAt") + } + if strings.TrimSpace(createdAt) == "" { + createdAt = item.GetCreationTimestamp().UTC().Format("2006-01-02T15:04:05Z") + } + + return LonghornBackupSummary{ + Name: item.GetName(), + SnapshotName: strings.TrimSpace(snapshotName), + State: strings.TrimSpace(state), + Error: strings.TrimSpace(errorMessage), + Progress: progress, + CreatedAt: strings.TrimSpace(createdAt), + } +} diff --git a/internal/k8s/longhorn_backups_test.go b/internal/k8s/longhorn_backups_test.go new file mode 100644 index 0000000..5acb9cb --- /dev/null +++ b/internal/k8s/longhorn_backups_test.go @@ -0,0 +1,99 @@ +package k8s + +import ( + "context" + "errors" + "strings" + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + dynamicfake "k8s.io/client-go/dynamic/fake" + k8stesting "k8s.io/client-go/testing" +) + +func TestGetLonghornBackupBySnapshotCoversLookupPaths(t *testing.T) { + matchingOlder := longhornBackupObject("backup-old", "snap-a", "Completed", "", 100, "2026-04-20T10:00:00Z") + matchingNewer := longhornBackupObject("backup-new", "snap-a", "Error", "storage cap exceeded", 0, "2026-04-20T11:00:00Z") + other := longhornBackupObject("backup-other", "snap-b", "Completed", "", 100, "2026-04-20T12:00:00Z") + client := &Client{Dynamic: newLonghornBackupDynamic(matchingOlder, matchingNewer, other)} + + backup, found, err := client.GetLonghornBackupBySnapshot(context.Background(), "snap-a") + if err != nil || !found { + t.Fatalf("expected matching backup, found=%v err=%v", found, err) + } + if backup.Name != "backup-new" || backup.State != "Error" || backup.Error != "storage cap exceeded" || backup.Progress != 0 { + t.Fatalf("expected newest matching backup with status fields, got %#v", backup) + } + + if _, found, err := client.GetLonghornBackupBySnapshot(context.Background(), "missing"); err != nil || found { + t.Fatalf("expected missing snapshot to be cleanly absent, found=%v err=%v", found, err) + } +} + +func TestGetLonghornBackupBySnapshotValidatesClientAndInput(t *testing.T) { + client := &Client{Dynamic: newLonghornBackupDynamic()} + if _, _, err := client.GetLonghornBackupBySnapshot(context.Background(), " "); err == nil || !strings.Contains(err.Error(), "snapshot name is required") { + t.Fatalf("expected snapshot validation error, got %v", err) + } + + client.Dynamic = nil + if _, _, err := client.GetLonghornBackupBySnapshot(context.Background(), "snap-a"); err == nil || !strings.Contains(err.Error(), "dynamic Kubernetes client is unavailable") { + t.Fatalf("expected missing dynamic client error, got %v", err) + } +} + +func TestGetLonghornBackupBySnapshotWrapsListErrorAndFallbackTimestamp(t *testing.T) { + client := &Client{Dynamic: newLonghornBackupDynamic(longhornBackupObject("backup-a", "snap-a", "Completed", "", 100, ""))} + + backup, found, err := client.GetLonghornBackupBySnapshot(context.Background(), "snap-a") + if err != nil || !found { + t.Fatalf("expected fallback timestamp backup, found=%v err=%v", found, err) + } + if backup.CreatedAt == "" || backup.SnapshotName != "snap-a" { + t.Fatalf("expected fallback timestamp and snapshot name, got %#v", backup) + } + + 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.GetLonghornBackupBySnapshot(context.Background(), "snap-a"); 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(), + map[schema.GroupVersionResource]string{longhornBackupGVR: "BackupList"}, + objects..., + ) +} + +func longhornBackupObject(name, snapshotName, state, errorMessage string, progress int64, createdAt string) *unstructured.Unstructured { + item := &unstructured.Unstructured{Object: map[string]any{ + "apiVersion": "longhorn.io/v1beta2", + "kind": "Backup", + "metadata": map[string]any{ + "name": name, + "namespace": longhornSystemNamespace, + }, + "spec": map[string]any{ + "snapshotName": snapshotName, + }, + "status": map[string]any{ + "state": state, + "error": errorMessage, + "progress": progress, + }, + }} + if createdAt != "" { + _ = unstructured.SetNestedField(item.Object, createdAt, "status", "backupCreatedAt") + } + item.SetCreationTimestamp(metav1.Now()) + return item +} diff --git a/internal/server/backup_handlers.go b/internal/server/backup_handlers.go index fb34895..fa3b82e 100644 --- a/internal/server/backup_handlers.go +++ b/internal/server/backup_handlers.go @@ -247,6 +247,16 @@ func (s *Server) executeBackup(ctx context.Context, req api.BackupRequest, reque if _, err := s.longhorn.SnapshotBackup(ctx, volumeName, backupID, labels, s.cfg.LonghornBackupMode); err != nil { return api.BackupResponse{}, "backend_error", err } + backup, result, err := s.waitForLonghornBackup(ctx, backupID) + if err != nil { + if backup.Name != "" { + response.Backup = backup.Name + } + return response, result, err + } + if backup.Name != "" { + response.Backup = backup.Name + } if err := s.pruneLonghornBackups(ctx, volumeName, req.Namespace, req.PVC, resolvedKeepLast); err != nil { return api.BackupResponse{}, "backend_error", err } diff --git a/internal/server/backup_handlers_test.go b/internal/server/backup_handlers_test.go index f4c9e93..f1e9a34 100644 --- a/internal/server/backup_handlers_test.go +++ b/internal/server/backup_handlers_test.go @@ -484,8 +484,71 @@ func TestExecuteBackupAndStatusHelpers(t *testing.T) { } }) + t.Run("longhorn waits for completed backup crd", func(t *testing.T) { + kube := &fakeKubeClient{ + longhornBackups: []k8s.LonghornBackupSummary{{ + Name: "backup-finished", + SnapshotName: "soteria-backup-apps-data-20260420-101112", + State: "Completed", + Progress: 100, + }}, + } + srv := newBackupTestServer( + &config.Config{AuthRequired: false, BackupDriver: "longhorn"}, + &backupTestKubeClient{restoreTestKubeClient: &restoreTestKubeClient{fakeKubeClient: kube}}, + &backupTestLonghornClient{ + restoreTestLonghornClient: &restoreTestLonghornClient{fakeLonghornClient: &fakeLonghornClient{}}, + }, + ) + backupNameFn = func(prefix, subject string) string { return "soteria-backup-apps-data-20260420-101112" } + defer func() { backupNameFn = defaultBackupName }() + + response, result, err := srv.executeBackup(context.Background(), api.BackupRequest{Namespace: "apps", PVC: "data"}, "brad") + if result != "success" || err != nil { + t.Fatalf("expected completed Longhorn backup success, got result=%q err=%v", result, err) + } + if response.Backup != "backup-finished" { + t.Fatalf("expected response to report completed backup name, got %#v", response) + } + }) + + t.Run("longhorn reports async backup failure", func(t *testing.T) { + kube := &fakeKubeClient{ + longhornBackups: []k8s.LonghornBackupSummary{{ + Name: "backup-failed", + SnapshotName: "soteria-backup-apps-data-20260420-101112", + State: "Error", + Error: "storage cap exceeded", + }}, + } + srv := newBackupTestServer( + &config.Config{AuthRequired: false, BackupDriver: "longhorn"}, + &backupTestKubeClient{restoreTestKubeClient: &restoreTestKubeClient{fakeKubeClient: kube}}, + &backupTestLonghornClient{ + restoreTestLonghornClient: &restoreTestLonghornClient{fakeLonghornClient: &fakeLonghornClient{}}, + }, + ) + backupNameFn = func(prefix, subject string) string { return "soteria-backup-apps-data-20260420-101112" } + defer func() { backupNameFn = defaultBackupName }() + + response, result, err := srv.executeBackup(context.Background(), api.BackupRequest{Namespace: "apps", PVC: "data"}, "brad") + if result != "backend_error" || err == nil || !strings.Contains(err.Error(), "storage cap exceeded") { + t.Fatalf("expected async backup backend error, got response=%#v result=%q err=%v", response, result, err) + } + if response.Backup != "backup-failed" { + t.Fatalf("expected response to include failed backup name, got %#v", response) + } + }) + t.Run("longhorn prunes soteria backups by keep last", func(t *testing.T) { keepLast := 2 + kube := &fakeKubeClient{ + longhornBackups: []k8s.LonghornBackupSummary{{ + Name: "backup-created", + SnapshotName: "soteria-backup-apps-data-20260420-101112", + State: "Completed", + }}, + } fakeLonghorn := &fakeLonghornClient{backups: []longhorn.Backup{ { Name: "backup-keep-new", @@ -520,9 +583,11 @@ func TestExecuteBackupAndStatusHelpers(t *testing.T) { }} srv := newBackupTestServer( &config.Config{AuthRequired: false, BackupDriver: "longhorn"}, - &backupTestKubeClient{restoreTestKubeClient: &restoreTestKubeClient{fakeKubeClient: &fakeKubeClient{}}}, + &backupTestKubeClient{restoreTestKubeClient: &restoreTestKubeClient{fakeKubeClient: kube}}, &backupTestLonghornClient{restoreTestLonghornClient: &restoreTestLonghornClient{fakeLonghornClient: fakeLonghorn}}, ) + backupNameFn = func(prefix, subject string) string { return "soteria-backup-apps-data-20260420-101112" } + defer func() { backupNameFn = defaultBackupName }() _, result, err := srv.executeBackup(context.Background(), api.BackupRequest{Namespace: "apps", PVC: "data", KeepLast: &keepLast}, "brad") if result != "success" || err != nil { t.Fatalf("expected longhorn backup success with pruning, got result=%q err=%v", result, err) diff --git a/internal/server/longhorn_backup_wait.go b/internal/server/longhorn_backup_wait.go new file mode 100644 index 0000000..5d8e9e2 --- /dev/null +++ b/internal/server/longhorn_backup_wait.go @@ -0,0 +1,52 @@ +package server + +import ( + "context" + "errors" + "fmt" + "strings" + "time" + + "scm.bstein.dev/bstein/soteria/internal/k8s" +) + +var ( + longhornBackupWaitTimeout = 2 * time.Minute + longhornBackupPollInterval = 5 * time.Second +) + +func (s *Server) waitForLonghornBackup(ctx context.Context, snapshotName string) (k8s.LonghornBackupSummary, string, error) { + waitCtx, cancel := context.WithTimeout(ctx, longhornBackupWaitTimeout) + defer cancel() + + ticker := time.NewTicker(longhornBackupPollInterval) + defer ticker.Stop() + + for { + backup, found, err := s.client.GetLonghornBackupBySnapshot(waitCtx, snapshotName) + if err != nil { + return k8s.LonghornBackupSummary{}, "backend_error", err + } + if found { + switch strings.ToLower(strings.TrimSpace(backup.State)) { + case "completed": + return backup, "success", nil + case "error", "failed": + message := strings.TrimSpace(backup.Error) + if message == "" { + message = fmt.Sprintf("Longhorn backup %s failed", backup.Name) + } + return backup, "backend_error", errors.New(message) + } + } + + select { + case <-waitCtx.Done(): + if found { + return backup, "in_progress", fmt.Errorf("Longhorn backup %s for snapshot %s is still %s at %d%%", backup.Name, snapshotName, backup.State, backup.Progress) + } + return k8s.LonghornBackupSummary{}, "in_progress", fmt.Errorf("Longhorn backup for snapshot %s did not appear before timeout", snapshotName) + case <-ticker.C: + } + } +} diff --git a/internal/server/longhorn_backup_wait_test.go b/internal/server/longhorn_backup_wait_test.go new file mode 100644 index 0000000..4afa92c --- /dev/null +++ b/internal/server/longhorn_backup_wait_test.go @@ -0,0 +1,53 @@ +package server + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "scm.bstein.dev/bstein/soteria/internal/config" + "scm.bstein.dev/bstein/soteria/internal/k8s" +) + +func TestWaitForLonghornBackupCoversLookupErrorAndTimeout(t *testing.T) { + originalTimeout := longhornBackupWaitTimeout + originalPoll := longhornBackupPollInterval + longhornBackupWaitTimeout = 20 * time.Millisecond + longhornBackupPollInterval = 5 * time.Millisecond + defer func() { + longhornBackupWaitTimeout = originalTimeout + longhornBackupPollInterval = originalPoll + }() + + srv := newBackupTestServer( + &config.Config{AuthRequired: false, BackupDriver: "longhorn"}, + &backupTestKubeClient{restoreTestKubeClient: &restoreTestKubeClient{fakeKubeClient: &fakeKubeClient{ + longhornBackupLookupErr: errors.New("lookup exploded"), + }}}, + &backupTestLonghornClient{restoreTestLonghornClient: &restoreTestLonghornClient{fakeLonghornClient: &fakeLonghornClient{}}}, + ) + if _, result, err := srv.waitForLonghornBackup(context.Background(), "snap-a"); result != "backend_error" || err == nil || !strings.Contains(err.Error(), "lookup exploded") { + t.Fatalf("expected lookup backend error, result=%q err=%v", result, err) + } + + srv.client = &backupTestKubeClient{restoreTestKubeClient: &restoreTestKubeClient{fakeKubeClient: &fakeKubeClient{ + longhornBackups: []k8s.LonghornBackupSummary{{ + Name: "backup-running", + SnapshotName: "snap-a", + State: "InProgress", + Progress: 42, + }}, + }}} + if _, result, err := srv.waitForLonghornBackup(context.Background(), "snap-a"); result != "in_progress" || err == nil || !strings.Contains(err.Error(), "backup-running") { + t.Fatalf("expected in-progress timeout with backup name, result=%q err=%v", result, err) + } + + srv.client = &backupTestKubeClient{restoreTestKubeClient: &restoreTestKubeClient{fakeKubeClient: &fakeKubeClient{ + longhornBackups: []k8s.LonghornBackupSummary{}, + }}} + if _, result, err := srv.waitForLonghornBackup(context.Background(), "snap-missing"); result != "in_progress" || err == nil || !strings.Contains(err.Error(), "did not appear") { + t.Fatalf("expected missing backup timeout, result=%q err=%v", result, err) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 7f590cb..99fd51c 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -25,6 +25,7 @@ type kubeClient interface { ReadBackupJobLog(ctx context.Context, namespace, jobName string) (string, error) 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) 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 7f7ca5f..b262778 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -30,6 +30,8 @@ type fakeKubeClient struct { lastRestoreReq api.RestoreTestRequest targetExists bool secretData map[string][]byte + longhornBackups []k8s.LonghornBackupSummary + longhornBackupLookupErr error } type secretErrorKubeClient struct { @@ -142,6 +144,26 @@ func (f *fakeKubeClient) ReadBackupJobLog(_ context.Context, namespace, jobName return f.jobLogs[key], nil } +func (f *fakeKubeClient) GetLonghornBackupBySnapshot(_ context.Context, snapshotName string) (k8s.LonghornBackupSummary, bool, error) { + if f.longhornBackupLookupErr != nil { + return k8s.LonghornBackupSummary{}, false, f.longhornBackupLookupErr + } + if f.longhornBackups == nil { + return k8s.LonghornBackupSummary{ + Name: "backup-completed", + SnapshotName: snapshotName, + State: "Completed", + Progress: 100, + }, true, nil + } + for _, backup := range f.longhornBackups { + if backup.SnapshotName == snapshotName { + return backup, true, nil + } + } + return k8s.LonghornBackupSummary{}, false, nil +} + func (f *fakeKubeClient) PersistentVolumeClaimExists(_ context.Context, _, _ string) (bool, error) { return f.targetExists, nil } diff --git a/internal/server/server_utilities.go b/internal/server/server_utilities.go index d5dcbb4..eb472c2 100644 --- a/internal/server/server_utilities.go +++ b/internal/server/server_utilities.go @@ -215,7 +215,13 @@ func writeError(w http.ResponseWriter, status int, message string) { writeJSON(w, status, map[string]string{"error": message}) } +var backupNameFn = defaultBackupName + func backupName(prefix, value string) string { + return backupNameFn(prefix, value) +} + +func defaultBackupName(prefix, value string) string { base := sanitizeName(fmt.Sprintf("soteria-%s-%s", prefix, value)) timestamp := time.Now().UTC().Format("20060102-150405") name := fmt.Sprintf("%s-%s", base, timestamp)