Fix backup check notifications and workload recovery; release 2026.09.21.r001
This commit is contained in:
1 parent
8f75a04db6
commit
5009dc202b
14 files changed
+295
-38
No files matched your search
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"encoding/xml"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
@@ -138,16 +139,21 @@ func (w *WorkloadManager) Prepare(ctx context.Context, job model.Job) (Prepared,
|
||||
}
|
||||
discovered, err := w.captureMetadata(ctx, job.Type, source.WorkloadID, metadataDir)
|
||||
if err != nil {
|
||||
w.Cleanup(context.Background(), prepared)
|
||||
return Prepared{}, err
|
||||
return Prepared{}, errors.Join(err, w.Cleanup(context.Background(), prepared))
|
||||
}
|
||||
prepared.Sources = append(prepared.Sources, discovered...)
|
||||
if job.Consistency.Mode == "stop" {
|
||||
if err := w.stop(ctx, job.Type, source.WorkloadID, time.Duration(job.ShutdownSecs)*time.Second); err != nil {
|
||||
w.Cleanup(context.Background(), prepared)
|
||||
return Prepared{}, err
|
||||
running, err := w.running(ctx, job.Type, source.WorkloadID)
|
||||
if err != nil {
|
||||
return Prepared{}, errors.Join(err, w.Cleanup(context.Background(), prepared))
|
||||
}
|
||||
if running {
|
||||
// Record before stopping: cancellation can happen after the daemon stopped it.
|
||||
prepared.Stopped = append(prepared.Stopped, source)
|
||||
if err := w.stop(ctx, job.Type, source.WorkloadID, time.Duration(job.ShutdownSecs)*time.Second); err != nil {
|
||||
return Prepared{}, errors.Join(err, w.Cleanup(context.Background(), prepared))
|
||||
}
|
||||
}
|
||||
prepared.Stopped = append(prepared.Stopped, source)
|
||||
}
|
||||
}
|
||||
prepared.Sources = uniqueStrings(prepared.Sources)
|
||||
@@ -203,22 +209,56 @@ func (w *WorkloadManager) FlashDevice(ctx context.Context) (string, error) {
|
||||
}
|
||||
|
||||
func (w *WorkloadManager) Cleanup(ctx context.Context, prepared Prepared) error {
|
||||
if _, hasDeadline := ctx.Deadline(); !hasDeadline {
|
||||
var cancel context.CancelFunc
|
||||
ctx, cancel = context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel()
|
||||
}
|
||||
var first error
|
||||
var failures []error
|
||||
// Recovery must continue even when the backup was cancelled. Each workload
|
||||
// receives its own deadline so one slow start cannot starve the others.
|
||||
for i := len(prepared.Stopped) - 1; i >= 0; i-- {
|
||||
source := prepared.Stopped[i]
|
||||
if err := w.start(ctx, prepared.Kind, source.WorkloadID); err != nil && first == nil {
|
||||
first = err
|
||||
startCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 120*time.Second)
|
||||
if err := w.start(startCtx, prepared.Kind, source.WorkloadID); err != nil {
|
||||
failures = append(failures, fmt.Errorf("restart %s: %w", source.WorkloadID, err))
|
||||
}
|
||||
cancel()
|
||||
}
|
||||
if prepared.MetadataDir != "" {
|
||||
_ = os.RemoveAll(prepared.MetadataDir)
|
||||
if err := os.RemoveAll(prepared.MetadataDir); err != nil {
|
||||
failures = append(failures, err)
|
||||
}
|
||||
}
|
||||
return errors.Join(failures...)
|
||||
}
|
||||
|
||||
func (w *WorkloadManager) running(ctx context.Context, kind model.JobType, id string) (bool, error) {
|
||||
if kind == model.JobDocker {
|
||||
output, err := exec.CommandContext(ctx, "docker", "inspect", "--format", "{{json .State}}", id).Output()
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("source: inspect state of %s: %w", id, err)
|
||||
}
|
||||
var state struct {
|
||||
Running bool
|
||||
Paused bool
|
||||
Restarting bool
|
||||
}
|
||||
if err := json.Unmarshal(output, &state); err != nil {
|
||||
return false, err
|
||||
}
|
||||
if state.Paused || state.Restarting {
|
||||
return false, fmt.Errorf("source: %s is paused or restarting; refusing to change its state", id)
|
||||
}
|
||||
return state.Running, nil
|
||||
}
|
||||
output, err := exec.CommandContext(ctx, "virsh", "domstate", id).Output()
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("source: inspect VM state %s: %w", id, err)
|
||||
}
|
||||
switch strings.TrimSpace(string(output)) {
|
||||
case "running":
|
||||
return true, nil
|
||||
case "shut off":
|
||||
return false, nil
|
||||
default:
|
||||
return false, fmt.Errorf("source: VM %s is in an unsupported state", id)
|
||||
}
|
||||
return first
|
||||
}
|
||||
|
||||
func (w *WorkloadManager) captureMetadata(ctx context.Context, kind model.JobType, id, dir string) ([]string, error) {
|
||||
@@ -249,7 +289,10 @@ func (w *WorkloadManager) captureMetadata(ctx context.Context, kind model.JobTyp
|
||||
var paths []string
|
||||
if len(inspected) > 0 {
|
||||
for _, mount := range inspected[0].Mounts {
|
||||
if mount.Type == "bind" && filepath.IsAbs(mount.Source) {
|
||||
if mount.Type == "bind" || mount.Type == "volume" {
|
||||
if !filepath.IsAbs(mount.Source) {
|
||||
return nil, fmt.Errorf("source: container %s has a %s mount without an absolute backup path", id, mount.Type)
|
||||
}
|
||||
paths = append(paths, mount.Source)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user