bstein/soteria
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

backup: wait for longhorn backup completion

Browse Source
This commit is contained in:
codex 2026-07-16 01:54:42 -03:00
parent 18bbd814f8
commit a81ada7358
11 changed files with 434 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
internal/k8s/client.go
View File

@ -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
}

26
internal/k8s/client_test.go
View File

@ -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
}
}

89
internal/k8s/longhorn_backups.go Normal file
View File

@ -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),
}
}

99
internal/k8s/longhorn_backups_test.go Normal file
View File

@ -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
}

10
internal/server/backup_handlers.go
View File

@ -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
}

67
internal/server/backup_handlers_test.go
View File

@ -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)

52
internal/server/longhorn_backup_wait.go Normal file
View File

@ -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:
}
}
}

53
internal/server/longhorn_backup_wait_test.go Normal file
View File

@ -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)
}
}

1
internal/server/server.go
View File

@ -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

22
internal/server/server_test.go
View File

@ -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
}

6
internal/server/server_utilities.go
View File

@ -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)