diff --git a/cloud-backups/cmd/oci_backup.go b/cloud-backups/cmd/oci_backup.go index be1ce8c..c6b9e0d 100644 --- a/cloud-backups/cmd/oci_backup.go +++ b/cloud-backups/cmd/oci_backup.go @@ -6,6 +6,7 @@ import ( "log/slog" "os" "os/signal" + "strings" "syscall" "time" @@ -15,6 +16,7 @@ import ( "github.com/relizaio/cloud-backup/internal/config" "github.com/relizaio/cloud-backup/internal/oras" "github.com/relizaio/cloud-backup/internal/orchestrator" + "github.com/relizaio/cloud-backup/internal/pipeline" "github.com/relizaio/cloud-backup/internal/registry" "github.com/relizaio/cloud-backup/internal/stats" "github.com/relizaio/cloud-backup/internal/storage" @@ -87,7 +89,7 @@ func runBackup() error { slog.Error("registry_login_failed", "error", err.Error()) return err } - defer authCtx.Cleanup() // guaranteed to run — no os.Exit below this point + defer authCtx.Cleanup() // guaranteed to run - no os.Exit below this point regClient := registry.New(cfg.RegistryHost, authCtx.ConfigDir, cfg.PlainHTTP) @@ -101,8 +103,8 @@ func runBackup() error { basePaths := cfg.CleanBasePaths() if len(basePaths) > 0 { slog.Info("running_preflight_auth_check", "target", basePaths[0]) - if err := regClient.PreflightCheck(ctx, basePaths[0]); err != nil { - slog.Error("preflight_check_failed", "error", err.Error()) + if err := pipeline.RunPreflightWithRetry(ctx, regClient, basePaths[0]); err != nil { + slog.Error("preflight_check_failed", "target", basePaths[0], "error", pipeline.TruncateCause(err.Error())) return err } slog.Info("preflight_check_passed") @@ -127,12 +129,42 @@ func runBackup() error { // 6. Report result stats.PrintSummary("backup_pipeline_completed", tracker, cfg.StorageType, time.Since(pipelineStart)) + + // PrintSummary only escalates when EVERY target is missing, so a partial skip + // otherwise stays at INFO and the run still exits 0 -- 49 of 50 repos could + // vanish silently. A skipped target produced no backup, so it is only + // tolerable when the absence is expected: under rolling months the + // PREVIOUS-month path may legitimately not exist. The CURRENT month is being + // actively written, and under explicit paths every target was named by an + // operator, so those absences are real gaps and must be visible. + // PrintSummary already covers the all-skipped case at ERROR; do not double-report. + if !allTargetsSkipped(tracker) { + var unexpected []string + for _, s := range tracker.GetSkipped() { + if !orchestrator.SkipIsExpected(s, cfg.AppendRollingMonths, time.Now().UTC()) { + unexpected = append(unexpected, s) + } + } + if len(unexpected) > 0 { + slog.Error("backup_targets_missing_from_registry", + "detail", "a skipped target produced no backup; absence is only expected for the previous-month rolling target", + "error", fmt.Sprintf("%d of %d target(s) produced no backup because the repository was reported absent: %s", + len(unexpected), tracker.GetTotal(), strings.Join(unexpected, ", "))) + } + } + if tracker.GetFailedCount() > 0 || (tracker.GetTotal() > 0 && tracker.GetTotal() == tracker.GetSkippedCount()) { return fmt.Errorf("backup pipeline completed with failures") } return nil } +// allTargetsSkipped mirrors the condition PrintSummary uses to raise +// pipeline_failed_all_repos_missing, so the two do not both alert on it. +func allTargetsSkipped(t *stats.Tracker) bool { + return t.GetTotal() > 0 && t.GetSkippedCount() == t.GetTotal() +} + func init() { ociCmd.AddCommand(backupCmd) backupCmd.Flags().StringSlice("registry-base-paths", []string{}, "Comma-separated list of target repositories (ENV: REGISTRY_BASE_PATHS)") diff --git a/cloud-backups/internal/orchestrator/backup.go b/cloud-backups/internal/orchestrator/backup.go index 6f36004..9d35b74 100644 --- a/cloud-backups/internal/orchestrator/backup.go +++ b/cloud-backups/internal/orchestrator/backup.go @@ -38,7 +38,7 @@ type BackupManager struct { EncPassword string // used only to build the modifier chain DumpPrefix string Timeout time.Duration - DeterministicName bool // when true, use last path segment as filename (no timestamp/random) — overwrites on re-run + DeterministicName bool // when true, use last path segment as filename (no timestamp/random) - overwrites on re-run } // RunBackups resolves the final target list and fans out concurrent backup workers. @@ -96,7 +96,7 @@ func (m *BackupManager) resolveTargets(basePaths []string, rollingMonths bool) [ slog.Info("rolling_months_strategy_enabled", "base_paths", basePaths) now := time.Now().UTC() currentMonth := now.Format("2006-01") - previousMonth := now.AddDate(0, 0, -now.Day()).Format("2006-01") + previousMonth := PreviousMonthSuffix(now) var targets []string for _, p := range basePaths { @@ -105,3 +105,21 @@ func (m *BackupManager) resolveTargets(basePaths []string, rollingMonths bool) [ } return targets } + +// PreviousMonthSuffix is the YYYY-MM appended to build the previous-month +// target. Exported because that target is the ONLY one under rolling months +// whose absence is legitimate (it may predate the deployment, or the month may +// have had no artifacts); callers deciding whether a skip is expected need to +// identify it, and must not re-derive the date arithmetic independently. +func PreviousMonthSuffix(now time.Time) string { + return now.AddDate(0, 0, -now.Day()).Format("2006-01") +} + +// SkipIsExpected reports whether a skipped (absent) target is a legitimate +// absence rather than a missing backup. Under explicit paths every target was +// named by an operator, so no absence is expected. Under rolling months only +// the previous-month target may legitimately not exist -- the CURRENT month is +// the one being actively written, and its absence means a real gap. +func SkipIsExpected(target string, rollingMonths bool, now time.Time) bool { + return rollingMonths && strings.HasSuffix(target, "-"+PreviousMonthSuffix(now)) +} diff --git a/cloud-backups/internal/orchestrator/backup_test.go b/cloud-backups/internal/orchestrator/backup_test.go index 64d9069..cddc896 100644 --- a/cloud-backups/internal/orchestrator/backup_test.go +++ b/cloud-backups/internal/orchestrator/backup_test.go @@ -72,7 +72,7 @@ func TestPGSuffixContract(t *testing.T) { t.Errorf("OCI suffix must not contain .dump: %q", ociSuffix) } - // PG suffix is .dump (or .dump.age) — no gzip layer at all. + // PG suffix is .dump (or .dump.age) - no gzip layer at all. pgSuffixPlain := ".dump" pgSuffixEnc := ".dump.age" for _, s := range []string{pgSuffixPlain, pgSuffixEnc} { @@ -137,7 +137,7 @@ func TestResolveTargets_RollingProducesCurrentAndPreviousMonth(t *testing.T) { t.Fatalf("got %d paths, want 2", len(got)) } // Both should be distinct (different months, unless we're on the 1st of the month - // and it wraps — but even then the function produces two entries) + // and it wraps - but even then the function produces two entries) if got[0] == got[1] { // This can happen on the 1st of the month when prev == current month boundary edge case // Just verify both are present and have month suffix @@ -236,3 +236,54 @@ func TestRunBackups_ContextCancelled(t *testing.T) { total := tracker.GetTotal() _ = total } + +// Under rolling months only the PREVIOUS-month target may legitimately be +// absent. The current month is being actively written, so treating its absence +// as expected would let a deleted or renamed repo pass as a healthy run -- the +// exact silent-gap this predicate exists to close. +func TestSkipIsExpected(t *testing.T) { + now := time.Date(2026, 8, 14, 3, 35, 0, 0, time.UTC) + + tests := []struct { + name string + target string + rollingMonths bool + want bool + }{ + {"previous month under rolling months", "rearm-artifacts/rebom-artifacts-2026-07", true, true}, + {"current month under rolling months", "rearm-artifacts/rebom-artifacts-2026-08", true, false}, + {"unrelated month under rolling months", "rearm-artifacts/rebom-artifacts-2026-05", true, false}, + {"previous month under explicit paths", "rearm-artifacts/rebom-artifacts-2026-07", false, false}, + {"explicit path", "rearm-artifacts/rebom-artifacts", false, false}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := SkipIsExpected(tc.target, tc.rollingMonths, now); got != tc.want { + t.Errorf("SkipIsExpected(%q, %v): got %v want %v", tc.target, tc.rollingMonths, got, tc.want) + } + }) + } +} + +// The suffix must match what resolveTargets actually builds, or the predicate +// would exempt nothing and re-open the noise it is meant to avoid. January is +// the case naive month arithmetic gets wrong. +func TestPreviousMonthSuffix_CrossesYearBoundary(t *testing.T) { + if got := PreviousMonthSuffix(time.Date(2026, 1, 5, 0, 0, 0, 0, time.UTC)); got != "2025-12" { + t.Errorf("January must roll back to the previous December, got %q", got) + } + + m := &BackupManager{} + now := time.Date(2026, 1, 5, 0, 0, 0, 0, time.UTC) + targets := m.resolveTargets([]string{"repo"}, true) + // resolveTargets uses time.Now(); assert the shape it produces agrees with + // the helper for the same instant rather than pinning a wall-clock date. + if len(targets) != 2 { + t.Fatalf("expected current+previous targets, got %v", targets) + } + if !strings.HasSuffix(targets[1], "-"+PreviousMonthSuffix(time.Now().UTC())) { + t.Errorf("second target %q must carry the previous-month suffix used by SkipIsExpected", targets[1]) + } + _ = now +} diff --git a/cloud-backups/internal/pipeline/stream.go b/cloud-backups/internal/pipeline/stream.go index 67adb0c..d000680 100644 --- a/cloud-backups/internal/pipeline/stream.go +++ b/cloud-backups/internal/pipeline/stream.go @@ -20,6 +20,10 @@ import ( const ( MaxBackupAttempts = 3 DefaultTimeout = 2 * time.Hour + // MaxCauseLength bounds a single attempt's cause inside an aggregated log + // field, so an alert payload stays ingestible. Generous enough to keep the + // failing URL and the tool's error line, which is what identifies the fault. + MaxCauseLength = 2000 ) var ( @@ -41,6 +45,8 @@ func RunWithRetry(ctx context.Context, src datasource.Source, storeProvider stor } }() + var attemptErrs []string + for attempt := 1; attempt <= MaxBackupAttempts; attempt++ { if ctx.Err() != nil { return @@ -48,6 +54,13 @@ func RunWithRetry(ctx context.Context, src datasource.Source, storeProvider stor slog.Info("backup_started", "target", target, "attempt", attempt) bytesUploaded, err := executeStream(ctx, src, storeProvider, target, backupName, nameSuffix, writerModifiers, timeout, deterministicName, totalHint) if err == nil { + // A target that needed a retry recovered on its own, so it is not + // operator-actionable and must not alert. It is still worth a line, + // because a repeatedly flaky registry shows up here first. + if len(attemptErrs) > 0 { + slog.Warn("backup_recovered_after_retry", "target", target, "attempts_used", attempt, + "earlier_failures", strings.Join(attemptErrs, " | ")) + } slog.Info("backup_successful", "target", target, "duration", time.Since(startTimer).Round(time.Second).String(), "size_human", stats.FormatBytes(bytesUploaded)) jobHandled = true tracker.RecordSuccess() @@ -55,36 +68,142 @@ func RunWithRetry(ctx context.Context, src datasource.Source, storeProvider stor return } // FAST-FAIL ON UNAUTHORIZED - if strings.Contains(err.Error(), "unauthorized") || strings.Contains(err.Error(), "authentication required") { - slog.Error("fatal_authentication_error", "target", target, "msg", "Credentials rejected. Halting retries.") + if isAuthRejection(err) { + // "detail", not "msg": slog's JSON handler already emits the event + // name as "msg", and a second "msg" attr is written verbatim, so + // every JSON parser in the alerting path keeps the LAST one and the + // event name never survives parsing. + slog.Error("fatal_authentication_error", "target", target, "detail", "Credentials rejected. Halting retries.", + "error", TruncateCause(err.Error())) jobHandled = true tracker.RecordFailure(target) return // Exit immediately, do not wait for backoff } if strings.Contains(err.Error(), "repository name not known to registry") { + // An absent repository is a legitimate skip, but only if it is the + // FIRST thing we saw. Reaching here after an attempt already failed + // for another reason means the classification is suspect (it is a + // substring match on a log tail), and swallowing it would retire the + // target with no ERROR at all. Surface those earlier causes. + if len(attemptErrs) > 0 { + slog.Error("repository_not_found_after_failed_attempts", "target", target, + "detail", "classified as absent only after earlier attempts failed; treat the absence as unconfirmed", + "error", strings.Join(attemptErrs, " | ")) + } slog.Warn("repository_not_found_skipping", "target", target) jobHandled = true tracker.RecordSkipped(target) return } - slog.Error("backup_attempt_failed", "target", target, "attempt", attempt, "error", err.Error()) + // WARN, not ERROR: this attempt may still be retried, and operator + // alerting fires on ERROR. A transient registry blip that the next + // attempt recovers from is not an incident, and paging on it trains + // operators to ignore the channel. The single ERROR for a target that + // really did fail is backup_exhausted below. + attemptErrs = append(attemptErrs, fmt.Sprintf("attempt %d: %s", attempt, TruncateCause(err.Error()))) + slog.Warn("backup_attempt_failed", "target", target, "attempt", attempt, "error", err.Error()) if attempt < MaxBackupAttempts { - backoff := RetryBackoffBase * time.Duration(1< MaxBackoffDuration { - backoff = MaxBackoffDuration - } - timer := time.NewTimer(backoff) - select { - case <-ctx.Done(): - timer.Stop() + if !waitBackoff(ctx, attempt) { + // Abandoned mid-retry. PrintSummary will report the target as + // failed, but only the causes we collected explain WHY, and they + // are otherwise WARN-only -- invisible where alerting is + // ERROR-only. Emit them before giving up. + slog.Error("backup_abandoned", "target", target, "attempts_used", len(attemptErrs), + "detail", "run cancelled before the retries were exhausted", + "error", strings.Join(attemptErrs, " | ")) return - case <-timer.C: } } } - slog.Error("backup_exhausted", "target", target) + // Carry every attempt's cause, because this is the only ERROR the operator + // sees for a failed target and an empty one is unactionable. + slog.Error("backup_exhausted", "target", target, "attempts_used", len(attemptErrs), + "error", strings.Join(attemptErrs, " | ")) +} + +// isAuthRejection reports whether the registry refused the credentials, in which +// case retrying cannot help. Matching on error text is this module's established +// idiom (see internal/registry/oras.go); the point of the helper is that the +// backup and preflight drivers share ONE definition of the predicate. +func isAuthRejection(err error) bool { + if err == nil { + return false + } + msg := err.Error() + return strings.Contains(msg, "unauthorized") || strings.Contains(msg, "authentication required") +} + +// TruncateCause bounds one attempt's cause before it goes into an aggregated +// log field. A cause carries the tail of the backup tool's own output (an 8KB +// tailBuffer), and backup_exhausted joins one per attempt -- unbounded, that is +// a single ~24KB field, which alerting backends truncate at an arbitrary point +// or reject outright. +// +// Keep the TAIL, not the head. The upstream buffer is already a tail buffer +// precisely because these tools print their diagnostic last, after progress +// chatter: `oras` ends with "Error: failed to ...". Head-truncating a tail +// buffer throws away the only line that names the fault and keeps the progress +// noise, so the surviving ERROR says nothing. +func TruncateCause(cause string) string { + if len(cause) <= MaxCauseLength { + return cause + } + // ToValidUTF8 because the cut can land mid-rune, and an invalid byte becomes + // U+FFFD once the record is JSON-encoded -- corruption that reads like a bug + // in the failure itself. + tail := strings.ToValidUTF8(cause[len(cause)-MaxCauseLength:], "") + return fmt.Sprintf("(truncated, %d bytes total, showing last %d) ...", len(cause), MaxCauseLength) + tail +} + +// waitBackoff sleeps the exponential backoff for the just-failed attempt. +// It reports false when ctx was cancelled while waiting, meaning the caller +// should give up rather than start another attempt. +func waitBackoff(ctx context.Context, attempt int) bool { + backoff := RetryBackoffBase * time.Duration(1< MaxBackoffDuration { + backoff = MaxBackoffDuration + } + timer := time.NewTimer(backoff) + defer timer.Stop() + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } +} + +// RunPreflightWithRetry probes the source with the same bounded retry/backoff a +// backup attempt gets. Preflight gates the WHOLE run, so without this a single +// transient network failure on one probe aborts every target with no retry at +// all -- a strictly worse outcome than the per-target failure it exists to +// prevent. Credential rejections still fail fast, since retrying cannot help. +func RunPreflightWithRetry(ctx context.Context, src datasource.Source, target string) error { + var lastErr error + for attempt := 1; attempt <= MaxBackupAttempts; attempt++ { + if ctx.Err() != nil { + return ctx.Err() + } + lastErr = src.PreflightCheck(ctx, target) + if lastErr == nil { + return nil + } + if isAuthRejection(lastErr) { + return lastErr + } + // Raw, not truncated: this is WARN and stays in the pod log, where the + // full detail is worth having. Truncation belongs on the ERROR that + // leaves the pod and lands in an alert payload. + slog.Warn("preflight_attempt_failed", "target", target, "attempt", attempt, "error", lastErr.Error()) + if attempt < MaxBackupAttempts { + if !waitBackoff(ctx, attempt) { + return lastErr + } + } + } + return lastErr } func executeStream(parentCtx context.Context, src datasource.Source, storeProvider storage.Provider, target, backupName, nameSuffix string, writerModifiers []WriterModifier, timeout time.Duration, deterministicName bool, totalHint int64) (int64, error) { @@ -179,7 +298,7 @@ func RunRestore(ctx context.Context, src datasource.Source, storeProvider storag pipeR, pipeW := io.Pipe() errChan := make(chan error, 1) - // 2. Goroutine: download → apply reader modifiers → write to pipeW + // 2. Goroutine: download -> apply reader modifiers -> write to pipeW go func() { var gErr error defer func() { diff --git a/cloud-backups/internal/pipeline/stream_test.go b/cloud-backups/internal/pipeline/stream_test.go index 9b7ff05..441a242 100644 --- a/cloud-backups/internal/pipeline/stream_test.go +++ b/cloud-backups/internal/pipeline/stream_test.go @@ -5,10 +5,13 @@ import ( "compress/gzip" "context" "errors" + "fmt" "io" + "log/slog" "os" "strings" + "sync" "sync/atomic" "testing" "time" @@ -26,8 +29,9 @@ func TestMain(m *testing.M) { // --- mocks --- type mockSource struct { - backupFn func(ctx context.Context, target string, out io.Writer) error - restoreFn func(ctx context.Context, target string, in io.Reader) error + backupFn func(ctx context.Context, target string, out io.Writer) error + restoreFn func(ctx context.Context, target string, in io.Reader) error + preflightFn func(ctx context.Context, target string) error } func (m *mockSource) Backup(ctx context.Context, target string, out io.Writer) error { @@ -36,7 +40,12 @@ func (m *mockSource) Backup(ctx context.Context, target string, out io.Writer) e func (m *mockSource) Restore(ctx context.Context, target string, in io.Reader) error { return m.restoreFn(ctx, target, in) } -func (m *mockSource) PreflightCheck(ctx context.Context, target string) error { return nil } +func (m *mockSource) PreflightCheck(ctx context.Context, target string) error { + if m.preflightFn == nil { + return nil + } + return m.preflightFn(ctx, target) +} type mockStorage struct { uploadFn func(ctx context.Context, path string, r io.Reader) error @@ -247,7 +256,7 @@ func TestOCI_NoEncryption_UploadIsGzipped(t *testing.T) { } // TestPG_NoEncryption_UploadIsNotGzipped verifies that the PG backup pipeline -// (no modifiers — pg_dump -Fc already compresses) does NOT add gzip on top. +// (no modifiers - pg_dump -Fc already compresses) does NOT add gzip on top. func TestPG_NoEncryption_UploadIsNotGzipped(t *testing.T) { var captured bytes.Buffer tracker := stats.New() @@ -414,3 +423,403 @@ func TestSuffixContract(t *testing.T) { t.Errorf("PG suffix must not contain 'gz' (no redundant compression): %q", pgSuffix) } } + +// --- alerting-level tests --- +// +// These pin the log LEVEL of each outcome, not just the text. Operator alerting +// fires on ERROR, so a level regression here silently either pages on a +// self-healing blip or, worse, hides a run that genuinely lost a backup. + +type capturedLog struct { + Level slog.Level + Msg string + Attrs map[string]string +} + +type captureHandler struct { + mu *sync.Mutex + recs *[]capturedLog +} + +func (h captureHandler) Enabled(context.Context, slog.Level) bool { return true } +func (h captureHandler) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h captureHandler) WithGroup(string) slog.Handler { return h } +func (h captureHandler) Handle(_ context.Context, r slog.Record) error { + attrs := map[string]string{} + r.Attrs(func(a slog.Attr) bool { + attrs[a.Key] = a.Value.String() + return true + }) + h.mu.Lock() + defer h.mu.Unlock() + *h.recs = append(*h.recs, capturedLog{Level: r.Level, Msg: r.Message, Attrs: attrs}) + return nil +} + +// captureLogs redirects the default slog logger for the duration of the test. +func captureLogs(t *testing.T) *[]capturedLog { + t.Helper() + recs := &[]capturedLog{} + prev := slog.Default() + slog.SetDefault(slog.New(captureHandler{mu: &sync.Mutex{}, recs: recs})) + t.Cleanup(func() { slog.SetDefault(prev) }) + return recs +} + +func findLog(recs *[]capturedLog, msg string) (capturedLog, bool) { + for _, r := range *recs { + if r.Msg == msg { + return r, true + } + } + return capturedLog{}, false +} + +func errorLevelMsgs(recs *[]capturedLog) []string { + var out []string + for _, r := range *recs { + if r.Level >= slog.LevelError { + out = append(out, r.Msg) + } + } + return out +} + +// A blip that the next attempt recovers from must not reach ERROR. This is the +// production false alarm: one refused TCP dial to the registry alerted even +// though the retry uploaded the backup seconds later. +func TestRunWithRetry_RecoveredAttemptDoesNotLogError(t *testing.T) { + recs := captureLogs(t) + tracker := stats.New() + var attempts atomic.Int32 + var captured bytes.Buffer + + src := &mockSource{backupFn: func(ctx context.Context, target string, out io.Writer) error { + if attempts.Add(1) == 1 { + return errors.New("dial tcp 10.0.5.5:443: connect: connection refused") + } + _, err := out.Write([]byte("ok")) + return err + }} + + RunWithRetry(context.Background(), src, captureStorage(&captured), "target", "prefix", ".dump", nil, tracker, 30*time.Second, false, 0) + + if got := tracker.GetFailedCount(); got != 0 { + t.Fatalf("recovered target must not count as failed, got %d", got) + } + if msgs := errorLevelMsgs(recs); len(msgs) != 0 { + t.Errorf("a recovered backup must emit no ERROR, got %v", msgs) + } + rec, ok := findLog(recs, "backup_attempt_failed") + if !ok { + t.Fatal("expected backup_attempt_failed to still be logged") + } + if rec.Level != slog.LevelWarn { + t.Errorf("backup_attempt_failed level: got %v want WARN", rec.Level) + } + if _, ok := findLog(recs, "backup_recovered_after_retry"); !ok { + t.Error("expected backup_recovered_after_retry so a flaky registry is still visible") + } +} + +// A target that really failed must emit exactly one ERROR, and that ERROR must +// carry the causes -- it is the only line the operator receives. +func TestRunWithRetry_ExhaustedLogsSingleErrorWithCauses(t *testing.T) { + recs := captureLogs(t) + tracker := stats.New() + var attempts atomic.Int32 + + src := &mockSource{backupFn: func(ctx context.Context, target string, out io.Writer) error { + return fmt.Errorf("boom %d", attempts.Add(1)) + }} + store := &mockStorage{uploadFn: func(ctx context.Context, path string, r io.Reader) error { + _, err := io.Copy(io.Discard, r) + return err + }} + + RunWithRetry(context.Background(), src, store, "target", "prefix", ".dump", nil, tracker, 30*time.Second, false, 0) + + if msgs := errorLevelMsgs(recs); len(msgs) != 1 || msgs[0] != "backup_exhausted" { + t.Fatalf("want exactly one ERROR (backup_exhausted), got %v", msgs) + } + rec, _ := findLog(recs, "backup_exhausted") + cause := rec.Attrs["error"] + if cause == "" { + t.Fatal("backup_exhausted must carry an error field; an empty alert is unactionable") + } + for i := 1; i <= MaxBackupAttempts; i++ { + if !strings.Contains(cause, fmt.Sprintf("boom %d", i)) { + t.Errorf("error field must include attempt %d cause, got %q", i, cause) + } + } +} + +// --- RunPreflightWithRetry tests --- + +// Preflight gates the whole run, so a transient failure here used to abort every +// target with no retry at all. +func TestRunPreflightWithRetry_RetriesTransientFailure(t *testing.T) { + captureLogs(t) + var calls atomic.Int32 + src := &mockSource{preflightFn: func(ctx context.Context, target string) error { + if calls.Add(1) < 3 { + return errors.New("dial tcp 10.0.5.5:443: connect: connection refused") + } + return nil + }} + + if err := RunPreflightWithRetry(context.Background(), src, "target"); err != nil { + t.Fatalf("preflight should have recovered, got %v", err) + } + if got := calls.Load(); got != 3 { + t.Errorf("expected 3 preflight attempts, got %d", got) + } +} + +func TestRunPreflightWithRetry_FastFailsOnUnauthorized(t *testing.T) { + captureLogs(t) + var calls atomic.Int32 + src := &mockSource{preflightFn: func(ctx context.Context, target string) error { + calls.Add(1) + return errors.New("unauthorized to access repo: check token scopes") + }} + + if err := RunPreflightWithRetry(context.Background(), src, "target"); err == nil { + t.Fatal("expected an error for rejected credentials") + } + if got := calls.Load(); got != 1 { + t.Errorf("credentials cannot fix themselves; want 1 attempt, got %d", got) + } +} + +func TestRunPreflightWithRetry_ExhaustsAndReturnsLastError(t *testing.T) { + captureLogs(t) + var calls atomic.Int32 + src := &mockSource{preflightFn: func(ctx context.Context, target string) error { + calls.Add(1) + return errors.New("connection refused") + }} + + if err := RunPreflightWithRetry(context.Background(), src, "target"); err == nil { + t.Fatal("expected the last error to be returned") + } + if got := calls.Load(); got != int32(MaxBackupAttempts) { + t.Errorf("want %d preflight attempts, got %d", MaxBackupAttempts, got) + } +} + +// An unbounded cause is not just untidy: backup_exhausted joins one per attempt, +// and the backup tool's own output tail can be kilobytes, so the single ERROR +// the operator receives can exceed what an alerting backend will ingest. +func TestTruncateCause_BoundsAggregatedField(t *testing.T) { + short := "dial tcp 10.0.5.5:443: connect: connection refused" + if got := TruncateCause(short); got != short { + t.Errorf("a short cause must pass through unchanged, got %q", got) + } + + long := strings.Repeat("x", MaxCauseLength*3) + got := TruncateCause(long) + if len(got) >= len(long) { + t.Fatalf("a long cause must shrink: got %d bytes, original %d", len(got), len(long)) + } + if !strings.Contains(got, "truncated") { + t.Error("a truncated cause must say so, or it reads as the whole error") + } + + // The property that matters. Upstream is a tailBuffer holding the LAST 8KB + // of the tool's output, and these tools print their diagnostic last, after + // progress chatter. Head-truncating a tail buffer keeps the noise and throws + // away the only line that names the fault, so the surviving ERROR says + // nothing -- which is exactly the bug this shape exists to prevent. + diagnostic := `Error: failed to find tags: dial tcp 10.0.5.5:443: connect: connection refused` + orasShaped := strings.Repeat("Uploading 1a2b3c4d sha256:deadbeef 100.00%\n", 400) + diagnostic + if len(orasShaped) <= MaxCauseLength { + t.Fatalf("fixture must exceed the cap to exercise truncation, got %d bytes", len(orasShaped)) + } + kept := TruncateCause(orasShaped) + if !strings.Contains(kept, diagnostic) { + t.Errorf("truncation dropped the diagnostic line, leaving only progress noise; got tail %q", + kept[max(0, len(kept)-120):]) + } +} + +// The aggregate that actually reaches the operator must stay bounded across all +// attempts, which is the property the truncation exists to protect. +func TestRunWithRetry_ExhaustedErrorFieldStaysBounded(t *testing.T) { + recs := captureLogs(t) + tracker := stats.New() + + src := &mockSource{backupFn: func(ctx context.Context, target string, out io.Writer) error { + return errors.New(strings.Repeat("y", 8192)) // an 8KB tool log tail + }} + store := &mockStorage{uploadFn: func(ctx context.Context, path string, r io.Reader) error { + _, err := io.Copy(io.Discard, r) + return err + }} + + RunWithRetry(context.Background(), src, store, "target", "prefix", ".dump", nil, tracker, 30*time.Second, false, 0) + + rec, ok := findLog(recs, "backup_exhausted") + if !ok { + t.Fatal("expected backup_exhausted") + } + limit := MaxCauseLength * MaxBackupAttempts * 2 + if got := len(rec.Attrs["error"]); got > limit { + t.Errorf("aggregated error field is %d bytes, want <= %d", got, limit) + } +} + +// Every terminal outcome, pinned by LEVEL. Alerting fires on ERROR, so a level +// regression on any row either pages on a self-healing blip or hides a run that +// lost a backup. Table-driven so a new outcome cannot be added without deciding +// which side of the alerting line it falls on. +func TestRunWithRetry_OutcomeLevels(t *testing.T) { + failWith := func(msg string) func(context.Context, string, io.Writer) error { + return func(context.Context, string, io.Writer) error { return errors.New(msg) } + } + recoverOnSecond := func() func(context.Context, string, io.Writer) error { + var n atomic.Int32 + return func(_ context.Context, _ string, out io.Writer) error { + if n.Add(1) == 1 { + return errors.New("connection refused") + } + _, err := out.Write([]byte("ok")) + return err + } + } + + tests := []struct { + name string + backupFn func(context.Context, string, io.Writer) error + wantErrors []string // ERROR-level events, i.e. what reaches the operator + }{ + {"success on first attempt", writePayload([]byte("ok")), nil}, + {"recovered after retry", recoverOnSecond(), nil}, + {"all attempts failed", failWith("boom"), []string{"backup_exhausted"}}, + {"credentials rejected", failWith("unauthorized: nope"), []string{"fatal_authentication_error"}}, + {"repository absent", failWith("repository name not known to registry"), nil}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + recs := captureLogs(t) + var captured bytes.Buffer + RunWithRetry(context.Background(), &mockSource{backupFn: tc.backupFn}, captureStorage(&captured), + "target", "prefix", ".dump", nil, stats.New(), 30*time.Second, false, 0) + + got := errorLevelMsgs(recs) + if len(got) != len(tc.wantErrors) { + t.Fatalf("ERROR-level events: got %v want %v", got, tc.wantErrors) + } + for i, want := range tc.wantErrors { + if got[i] != want { + t.Errorf("ERROR event %d: got %q want %q", i, got[i], want) + } + } + }) + } +} + +// slog's JSON handler emits the event name as "msg". An attr also named "msg" +// is written as a SECOND "msg" key, and every JSON parser keeps the last one -- +// so the event name silently disappears from the parsed record and alerting can +// no longer route on it. Nothing may reintroduce that. +func TestLogRecords_NeverUseMsgAsAttrKey(t *testing.T) { + cases := map[string]func(context.Context, string, io.Writer) error{ + "unauthorized": func(context.Context, string, io.Writer) error { + return errors.New("unauthorized: nope") + }, + "exhausted": func(context.Context, string, io.Writer) error { + return errors.New("boom") + }, + } + + for name, fn := range cases { + t.Run(name, func(t *testing.T) { + recs := captureLogs(t) + var captured bytes.Buffer + RunWithRetry(context.Background(), &mockSource{backupFn: fn}, captureStorage(&captured), + "target", "prefix", ".dump", nil, stats.New(), 30*time.Second, false, 0) + + for _, r := range *recs { + if _, clash := r.Attrs["msg"]; clash { + t.Errorf("event %q uses \"msg\" as an attr key, which shadows the event name after JSON parsing", r.Msg) + } + } + }) + } +} + +// A "repository absent" classification is a substring match on a log tail, so it +// can misfire on a transient failure (a content digest containing "404"). When +// it lands AFTER an attempt already failed for another reason, the absence is +// unconfirmed and swallowing it retires the target with no ERROR at all -- the +// path that lets a real repo go unbacked-up at exit 0. +func TestRunWithRetry_SkipAfterFailedAttemptsIsNotSilent(t *testing.T) { + recs := captureLogs(t) + var n atomic.Int32 + src := &mockSource{backupFn: func(context.Context, string, io.Writer) error { + if n.Add(1) == 1 { + return errors.New("dial tcp 10.0.5.5:443: connect: connection refused") + } + return errors.New("repository name not known to registry") + }} + var captured bytes.Buffer + + RunWithRetry(context.Background(), src, captureStorage(&captured), "target", "prefix", ".dump", nil, + stats.New(), 30*time.Second, false, 0) + + rec, ok := findLog(recs, "repository_not_found_after_failed_attempts") + if !ok { + t.Fatal("a skip preceded by a failed attempt must surface the earlier cause at ERROR") + } + if rec.Level != slog.LevelError { + t.Errorf("level: got %v want ERROR", rec.Level) + } + if !strings.Contains(rec.Attrs["error"], "connection refused") { + t.Errorf("the earlier cause must be carried, got %q", rec.Attrs["error"]) + } +} + +// ...but a clean first-attempt absence stays a quiet skip, or the previous-month +// rolling target would page every month. +func TestRunWithRetry_CleanSkipStaysQuiet(t *testing.T) { + recs := captureLogs(t) + src := &mockSource{backupFn: func(context.Context, string, io.Writer) error { + return errors.New("repository name not known to registry") + }} + var captured bytes.Buffer + + RunWithRetry(context.Background(), src, captureStorage(&captured), "target", "prefix", ".dump", nil, + stats.New(), 30*time.Second, false, 0) + + if msgs := errorLevelMsgs(recs); len(msgs) != 0 { + t.Errorf("a first-attempt absence must not alert, got %v", msgs) + } +} + +// Cancellation mid-retry leaves the target unfinished. PrintSummary will count it +// as failed, but only the collected causes explain why, and they are WARN-only. +func TestRunWithRetry_AbandonedMidRetryExplainsItself(t *testing.T) { + recs := captureLogs(t) + ctx, cancel := context.WithCancel(context.Background()) + src := &mockSource{backupFn: func(context.Context, string, io.Writer) error { + cancel() // cancel while the first backoff is pending + return errors.New("dial tcp 10.0.5.5:443: connect: connection refused") + }} + var captured bytes.Buffer + + RunWithRetry(ctx, src, captureStorage(&captured), "target", "prefix", ".dump", nil, + stats.New(), 30*time.Second, false, 0) + + rec, ok := findLog(recs, "backup_abandoned") + if !ok { + t.Fatal("abandoning mid-retry must emit an ERROR carrying the causes so far") + } + if rec.Level != slog.LevelError { + t.Errorf("level: got %v want ERROR", rec.Level) + } + if !strings.Contains(rec.Attrs["error"], "connection refused") { + t.Errorf("the cause must be carried, got %q", rec.Attrs["error"]) + } +} diff --git a/cloud-backups/internal/registry/oras.go b/cloud-backups/internal/registry/oras.go index 17f1969..0c507cf 100644 --- a/cloud-backups/internal/registry/oras.go +++ b/cloud-backups/internal/registry/oras.go @@ -82,7 +82,7 @@ func (c *OrasClient) Backup(ctx context.Context, registryPath string, out io.Wri if strings.Contains(logs, "unauthorized") || strings.Contains(logs, "authentication required") { return fmt.Errorf("unauthorized to access %s: check token scopes", fullPath) } - if strings.Contains(logs, "not found") || strings.Contains(logs, "404") { + if repositoryAbsent(logs) { slog.Warn("oras_backup_repository_not_found", "path", fullPath) return fmt.Errorf("repository name not known to registry: %s", fullPath) } @@ -143,6 +143,12 @@ func (c *OrasClient) PreflightCheck(ctx context.Context, registryPath string) er cmd := exec.CommandContext(ctx, "oras", preflightArgs...) cmd.Env = append(os.Environ(), fmt.Sprintf("DOCKER_CONFIG=%s", c.authDir)) + // Deliberately NOT a tailBuffer: `logs` is what the not-found and auth + // predicates below match on, so dropping the head can change the + // CLASSIFICATION, not just the message -- a probe whose "not found" token + // scrolls out of a bounded buffer stops being a tolerated absence and aborts + // the entire run. The size of the surfaced string is bounded at the call + // site instead, where it is only used for display. var stderrBuf strings.Builder cmd.Stderr = &stderrBuf @@ -151,7 +157,7 @@ func (c *OrasClient) PreflightCheck(ctx context.Context, registryPath string) er if strings.Contains(logs, "unauthorized") || strings.Contains(logs, "authentication required") { return fmt.Errorf("unauthorized to access %s: check token scopes", fullPath) } - if strings.Contains(logs, "not found") || strings.Contains(logs, "404") { + if repositoryAbsent(logs) { slog.Warn("oras_preflight_repository_not_found", "path", fullPath) return nil } @@ -160,6 +166,47 @@ func (c *OrasClient) PreflightCheck(ctx context.Context, registryPath string) er return nil } +// transportFailureMarkers are substrings that only a connectivity failure +// produces. A repository that is genuinely absent never emits them. +var transportFailureMarkers = []string{ + "connection refused", + "connection reset", + "dial tcp", + "i/o timeout", + "no such host", + "tls handshake", + "unexpected eof", + "context deadline exceeded", +} + +// repositoryAbsent reports whether the tool's output means "this repository +// does not exist" rather than "we could not reach the registry". +// +// The distinction decides whether the caller SKIPS the target (no backup, no +// alert, exit 0) or FAILS it, so a false positive silently loses a backup. Two +// traps, both observed: +// +// - Bare "404" is not evidence. Content digests are hex, and "404" appears in +// roughly 1.5% of them, so a refused dial on a repo with enough blobs in the +// log tail gets misread as an absence. Any transport marker therefore vetoes +// the classification outright. +// - Matching was case-sensitive and missed the canonical distribution error +// ("name unknown: repository name not known to registry") entirely, which +// made preflight abort whole runs against distribution-compatible +// registries. Match case-insensitively and include the canonical phrasing. +func repositoryAbsent(logs string) bool { + l := strings.ToLower(logs) + for _, marker := range transportFailureMarkers { + if strings.Contains(l, marker) { + return false + } + } + return strings.Contains(l, "name unknown") || + strings.Contains(l, "repository name not known") || + strings.Contains(l, "not found") || + strings.Contains(l, "404") +} + func makeRandBytes() ([]byte, error) { b := make([]byte, 8) if _, err := rand.Read(b); err != nil { diff --git a/cloud-backups/internal/registry/oras_test.go b/cloud-backups/internal/registry/oras_test.go index ea8ba59..6bcce22 100644 --- a/cloud-backups/internal/registry/oras_test.go +++ b/cloud-backups/internal/registry/oras_test.go @@ -44,7 +44,7 @@ func TestTailBuffer_MultipleWritesTruncation(t *testing.T) { if len(got) != 5 { t.Errorf("expected 5 bytes, got %d: %q", len(got), got) } - // After writing "12345" then "678", buf is "12345678" → truncated to last 5 = "45678" + // After writing "12345" then "678", buf is "12345678" -> truncated to last 5 = "45678" if got != "45678" { t.Errorf("got %q want %q", got, "45678") } @@ -120,3 +120,44 @@ func TestTailBuffer_MatchesBytesBuffer(t *testing.T) { t.Errorf("tailBuffer %q != bytes.Buffer %q", tb.String(), bb.String()) } } + +// This predicate decides SKIP (no backup, no alert, exit 0) versus FAIL, so a +// false positive silently loses a backup. Both directions are pinned. +func TestRepositoryAbsent(t *testing.T) { + tests := []struct { + name string + logs string + want bool + }{ + // Genuine absences must still skip, or the previous-month rolling + // target would fail every month. + {"canonical distribution error", `Error response from registry: name unknown: repository name not known to registry`, true}, + {"wrapped repository message", "repository name not known to registry: reg/repo", true}, + {"lowercase not found", "Error: repo not found", true}, + {"capitalised status text", "Error: unexpected status 404 Not Found", true}, + + // A transport failure is NOT an absence. A content digest containing + // "404" must never turn a refused dial into a silent skip -- this is the + // case that produced a successful-looking run with no backup. + { + name: "refused dial with 404 inside a digest", + logs: `Uploading sha256:a481e31691d2b86354c9ebbe3db446dd4041b514 100.00% +Error: failed to find tags: Get "https://reg/v2/org/repo/tags/list?last=x&n=100&orderby=": dial tcp 10.0.5.5:443: connect: connection refused`, + want: false, + }, + {"connection reset with 404 digest", "sha256:404aa1 ... read: connection reset by peer", false}, + {"i/o timeout with not found text", "Get \"https://reg/v2/\": i/o timeout, repo not found in cache", false}, + + // Neither absence nor transport failure. + {"generic server error", "Error: unexpected status code 500 Internal Server Error", false}, + {"empty", "", false}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := repositoryAbsent(tc.logs); got != tc.want { + t.Errorf("repositoryAbsent: got %v want %v\nlogs: %s", got, tc.want, tc.logs) + } + }) + } +} diff --git a/cloud-backups/internal/stats/tracker.go b/cloud-backups/internal/stats/tracker.go index cc82e5c..93654f5 100644 --- a/cloud-backups/internal/stats/tracker.go +++ b/cloud-backups/internal/stats/tracker.go @@ -4,6 +4,7 @@ import ( "fmt" "log/slog" "slices" + "strings" "sync" "time" ) @@ -33,6 +34,14 @@ func (t *Tracker) GetSuccess() int64 { t.mu.Lock(); defer t.mu.Unlock(); re func (t *Tracker) GetFailedCount() int64 { t.mu.Lock(); defer t.mu.Unlock(); return t.FailureCount } func (t *Tracker) GetSkippedCount() int64 { t.mu.Lock(); defer t.mu.Unlock(); return t.SkippedCount } +// GetSkipped returns the skipped paths. A skipped target produced no backup, so +// callers that know a skip is illegitimate need the names, not just the count. +func (t *Tracker) GetSkipped() []string { + t.mu.Lock() + defer t.mu.Unlock() + return slices.Clone(t.Skipped) +} + func (t *Tracker) RecordSkipped(path string) { t.mu.Lock() defer t.mu.Unlock() @@ -64,6 +73,31 @@ func FormatBytes(bytes int64) string { return fmt.Sprintf("%.2f %cB", float64(bytes)/float64(div), "KMGTPE"[exp]) } +// formatPaths renders a recorded path list for an alert payload. The list is +// capped at MaxPathsTracked while the count is not, and callers that pass a +// repeated target (pg audit-rotate records the database name once per archive) +// would otherwise render "rearm, rearm, rearm". Dedupe, and say so when the +// list is shorter than the count, so a truncated list cannot read as complete. +func formatPaths(paths []string, count int64) string { + if len(paths) == 0 { + return "(none recorded)" + } + unique := make([]string, 0, len(paths)) + seen := make(map[string]struct{}, len(paths)) + for _, p := range paths { + if _, dup := seen[p]; dup { + continue + } + seen[p] = struct{}{} + unique = append(unique, p) + } + rendered := strings.Join(unique, ", ") + if int64(len(paths)) < count { + return fmt.Sprintf("%s (showing %d of %d recorded)", rendered, len(paths), count) + } + return rendered +} + func PrintSummary(eventName string, t *Tracker, storageType string, duration time.Duration) { t.mu.Lock() defer t.mu.Unlock() @@ -88,10 +122,20 @@ func PrintSummary(eventName string, t *Tracker, storageType string, duration tim allSkipped := t.Total > 0 && t.SkippedCount == t.Total + // The failing paths are already inside summary, but alerting reads flat + // fields -- a nested map renders as an opaque blob, so the one ERROR that + // says the run lost data would arrive without saying which target. if allSkipped { - slog.Error("pipeline_failed_all_repos_missing", "summary", summary, "msg", "CRITICAL: No repositories found.") + // "detail", not "msg": slog's JSON handler emits the event name as + // "msg", so a "msg" attr becomes a duplicate key and every JSON parser + // keeps the last one -- the event name would never survive parsing. + slog.Error("pipeline_failed_all_repos_missing", "summary", summary, "detail", "CRITICAL: No repositories found.", + "error", fmt.Sprintf("no repositories found: all %d target(s) missing from the registry: %s", + t.Total, formatPaths(t.Skipped, t.SkippedCount))) } else if t.FailureCount > 0 { - slog.Error("pipeline_completed_with_failures", "summary", summary) + slog.Error("pipeline_completed_with_failures", "summary", summary, + "error", fmt.Sprintf("%d of %d target(s) failed: %s", + t.FailureCount, t.Total, formatPaths(t.Failed, t.FailureCount))) } else { slog.Info("pipeline_completed_successfully", "summary", summary) } diff --git a/cloud-backups/internal/stats/tracker_test.go b/cloud-backups/internal/stats/tracker_test.go index 51a589a..dfbedfb 100644 --- a/cloud-backups/internal/stats/tracker_test.go +++ b/cloud-backups/internal/stats/tracker_test.go @@ -152,3 +152,56 @@ func TestFormatBytes(t *testing.T) { }) } } + +// The joined path list is the only part of the alert that says WHICH target is +// broken, and it is built from a slice capped at MaxPathsTracked while the count +// is uncapped. A silently short list reads as complete. +func TestFormatPaths(t *testing.T) { + tests := []struct { + name string + paths []string + count int64 + want string + }{ + {"nothing recorded", nil, 0, "(none recorded)"}, + {"single", []string{"repo/a"}, 1, "repo/a"}, + { + // pg audit-rotate records the database name once per archive, so an + // undeduped join reads "rearm, rearm, rearm". + name: "repeated target is deduped", + paths: []string{"rearm", "rearm", "rearm"}, + count: 3, + want: "rearm", + }, + { + name: "truncated list says so", + paths: []string{"repo/a", "repo/b"}, + count: 150, + want: "repo/a, repo/b (showing 2 of 150 recorded)", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := formatPaths(tc.paths, tc.count); got != tc.want { + t.Errorf("formatPaths(%v, %d):\n got %q\nwant %q", tc.paths, tc.count, got, tc.want) + } + }) + } +} + +// GetSkipped must return a copy: callers iterate it while deciding whether to +// alert, and handing out the live slice would race the workers still running. +func TestGetSkipped_ReturnsCopy(t *testing.T) { + tr := New() + tr.RecordSkipped("repo/a") + + got := tr.GetSkipped() + if len(got) != 1 || got[0] != "repo/a" { + t.Fatalf("unexpected skipped list: %v", got) + } + got[0] = "mutated" + if again := tr.GetSkipped(); again[0] != "repo/a" { + t.Errorf("mutating the returned slice corrupted the tracker: %v", again) + } +}