From 2714bb33603a3bcf2d45142491b4279bbfc4c18f Mon Sep 17 00:00:00 2001 From: Rhythm Garg Date: Fri, 14 Aug 2026 15:26:38 +0000 Subject: [PATCH 1/2] fix(cloud-backups): stop paging on self-healing backup retries The oci-artifacts-backup CronJob alerted on a customer prod instance on 2026-08-08. Investigation showed the backup never actually failed: attempt 1 died on a refused TCP dial to the ACR private endpoint during tag pagination, and attempt 2 completed the upload. The alert fired only because backup_attempt_failed was logged at ERROR, and the operator's rule pages on any level=ERROR. Only ERROR-level logs leave the pod, so the successful outcome was invisible and every blip read as an outage. Underneath, `oras backup` aborts a whole repo on one ECONNREFUSED: oras-go's retry predicate retries dial TIMEOUTS only, so a refused connection is never retried (reproduced against a fake ACR paginator; a reset, a 429 and a 500 injected at the same point all recover). The tool's own 3x retry already absorbs that, so the registry behaviour is left alone and the reporting is fixed instead. Alerting levels: - backup_attempt_failed drops to WARN while a retry remains. The single ERROR for a target that really failed is backup_exhausted, which now carries attempts_used and every attempt's cause -- it previously had no error field at all, so a genuine give-up alerted with a blank cause. - New WARN backup_recovered_after_retry, so a repeatedly flaky registry is still visible without paging. - pipeline_completed_with_failures and pipeline_failed_all_repos_missing gained a flat error field naming the targets; the failing paths were only inside a nested summary map, which alerting renders opaquely. - Both sites used "msg" as an attr key. slog's JSON handler already emits the event name as "msg", so the record carried a duplicate key and every JSON parser kept the last one -- the event name never survived parsing. Renamed to "detail". - Causes are truncated per attempt (MaxCauseLength) before being joined, bounding what was a ~24KB single field, and PreflightCheck's stderr now uses a tailBuffer like its Backup/Restore siblings instead of an unbounded builder. Behaviour: - PreflightCheck ran outside the retry loop, so the same transient failure landing on the probe aborted the entire job for every target with no retry at all. It now shares the backup retry/backoff via RunPreflightWithRetry, with the auth fast-fail preserved. - Under explicit registry paths, a skipped (absent) repository is now an ERROR. PrintSummary only escalates when EVERY target is missing, so 49 of 50 could vanish at INFO with exit 0. Rolling months stays exempt, since it deliberately fabricates a previous-month path that legitimately may not exist yet. Verified end to end against a local rig reproducing the exact production error (fake ACR Link-header pagination + oras 1.3.3 + minio): the Aug-8 scenario now emits zero ERROR lines, a preflight refusal survives, and a sustained refusal emits exactly two ERRORs that name the target and the causes. New tests pin the level of all five terminal outcomes and are mutation-checked -- reverting each guarantee fails its test. Co-Authored-By: Claude ReARM-Agent: 1420896f-adf5-4843-896f-d863cfcc6528 ReARM-Agentic-Session: 94afdb9d-05d3-44ad-abcd-92e753ddf2df --- cloud-backups/cmd/oci_backup.go | 24 +- cloud-backups/internal/pipeline/stream.go | 125 ++++++- .../internal/pipeline/stream_test.go | 330 +++++++++++++++++- cloud-backups/internal/registry/oras.go | 7 +- cloud-backups/internal/stats/tracker.go | 23 +- 5 files changed, 484 insertions(+), 25 deletions(-) diff --git a/cloud-backups/cmd/oci_backup.go b/cloud-backups/cmd/oci_backup.go index be1ce8c..ea31817 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,6 +129,22 @@ 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 stays at INFO and the run still exits 0 -- 49 of 50 repos could + // vanish silently. Under explicit paths each target is one an operator named, + // so any skip is a missing backup and must be visible. Rolling months is + // exempt: it deliberately fabricates a previous-month path that legitimately + // may not exist yet, and alerting on that would recreate the noise this + // tool's retry levels were just fixed to avoid. + if !cfg.AppendRollingMonths { + if skipped := tracker.GetSkipped(); len(skipped) > 0 { + slog.Error("backup_targets_missing_from_registry", + "error", fmt.Sprintf("%d of %d explicitly configured target(s) produced no backup because the repository was not found: %s", + len(skipped), tracker.GetTotal(), strings.Join(skipped, ", "))) + } + } + if tracker.GetFailedCount() > 0 || (tracker.GetTotal() > 0 && tracker.GetTotal() == tracker.GetSkippedCount()) { return fmt.Errorf("backup pipeline completed with failures") } diff --git a/cloud-backups/internal/pipeline/stream.go b/cloud-backups/internal/pipeline/stream.go index 67adb0c..7b6aa48 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,8 +68,13 @@ 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 @@ -69,22 +87,101 @@ func RunWithRetry(ctx context.Context, src datasource.Source, storeProvider stor 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) { 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 (up to +// MaxCauseLength * several, e.g. the 8KB oras tail), 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. Keeping the head +// preserves the part that names the failure. +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. + head := strings.ToValidUTF8(cause[:MaxCauseLength], "") + return head + fmt.Sprintf("... (truncated, %d bytes total)", len(cause)) +} + +// 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 +276,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..a040e56 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,316 @@ 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.HasPrefix(got, strings.Repeat("x", MaxCauseLength)) { + t.Error("truncation must keep the head, which names the failure") + } + if !strings.Contains(got, "truncated") { + t.Error("a truncated cause must say so, or it reads as the whole error") + } +} + +// 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) + } + } + }) + } +} diff --git a/cloud-backups/internal/registry/oras.go b/cloud-backups/internal/registry/oras.go index 17f1969..293202f 100644 --- a/cloud-backups/internal/registry/oras.go +++ b/cloud-backups/internal/registry/oras.go @@ -143,8 +143,11 @@ 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)) - var stderrBuf strings.Builder - cmd.Stderr = &stderrBuf + // tailBuffer, matching Backup and Restore above. An unbounded builder lets a + // chatty failure produce an arbitrarily large error string, and unlike its + // siblings this one is surfaced at ERROR straight into an alert payload. + stderrBuf := &tailBuffer{max: 8192} + cmd.Stderr = stderrBuf if err := cmd.Run(); err != nil { logs := stderrBuf.String() diff --git a/cloud-backups/internal/stats/tracker.go b/cloud-backups/internal/stats/tracker.go index cc82e5c..f88685a 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() @@ -88,10 +97,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, strings.Join(t.Skipped, ", "))) } 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, strings.Join(t.Failed, ", "))) } else { slog.Info("pipeline_completed_successfully", "summary", summary) } From 8c4f64468a1ff04916753544e8c93a6212e7ea7a Mon Sep 17 00:00:00 2001 From: Rhythm Garg Date: Fri, 14 Aug 2026 16:02:46 +0000 Subject: [PATCH 2/2] fix(cloud-backups): close the silent-skip paths found by adversarial review Four independent review lenses were run against the previous commit, each blind to the others. One REFUTED its central safety claim with a reproducible counterexample, and the others found defects the first commit introduced. All of the following are proven by execution, not inspection. The counterexample: a repository can be silently skipped, producing no backup, no ERROR and exit 0. `repositoryAbsent` classified any log tail containing bare "404" as "repository does not exist". Content digests are hex and contain "404" in ~1.5% of cases, so a refused dial on a repo with enough blobs in the 8KB tail was misread as an absence -- the same ECONNREFUSED that started this investigation. The skip then set jobHandled, bypassed the remaining retries, and under the shipped appendRollingMonths=true default produced pipeline_completed_successfully. - repositoryAbsent now vetoes the classification on any transport marker (connection refused/reset, dial tcp, i/o timeout, no such host, TLS handshake, unexpected EOF, deadline exceeded), matches case-insensitively, and recognises the canonical distribution error ("name unknown"). The case-sensitivity gap was also making preflight abort entire runs against distribution-compatible registries, verified live -- the previous commit's preflight retry merely made that failure 60s slower. - A skip that lands AFTER an attempt already failed for another reason is now an ERROR carrying the earlier causes, since the absence is unconfirmed. - The rolling-months exemption for missing targets was too broad: it exempted the whole run, but only the PREVIOUS-month target may legitimately be absent. The current month is being actively written. orchestrator.SkipIsExpected now decides this, sharing the date arithmetic with resolveTargets rather than re-deriving it. Regressions the previous commit introduced, now fixed: - TruncateCause kept the HEAD of what is already a TAIL buffer. These tools print their diagnostic last, so the surviving ERROR contained upload progress noise and the line naming the fault was dropped - exactly inverting the intent. It now keeps the tail. - PreflightCheck was switched to a tailBuffer, but its output is what the absence/auth predicates match on, so bounding it changed CLASSIFICATION, not just message size. Reverted; the surfaced string is bounded at the call site instead, where it is display-only. - The missing-targets ERROR counted len(skipped), a slice capped at MaxPathsTracked=100, so 150 skips of 200 reported "100 of 200". - PrintSummary joined those same capped lists with no marker, and pg audit-rotate records the database name once per archive, so the field read "3 of 3 target(s) failed: rearm, rearm, rearm". formatPaths now dedupes and discloses truncation. - Abandoning mid-retry (context cancelled during backoff) returned silently; the causes were WARN-only and therefore invisible. Now emits backup_abandoned at ERROR. - pipeline_failed_all_repos_missing and backup_targets_missing_from_registry both fired for the all-skipped case. Deduplicated. Everything added outside internal/pipeline previously had no tests, which is why the cap bug got through. Added coverage for repositoryAbsent (both directions), SkipIsExpected, PreviousMonthSuffix across a year boundary, formatPaths, GetSkipped aliasing, and the three new ERROR paths. Verified end to end on the rig: the Aug-8 transient refusal still emits zero ERRORs; an absent CURRENT-month repo now emits one ERROR naming it; an absent previous-month repo stays silent; a sustained refusal emits two ERRORs carrying the causes. go test -race clean. Reviewed but deliberately NOT fixed here, filed separately: encryption silently disabled when the optional secret key is absent (writes a plaintext backup, exit 0); no dead man's switch; a panic in the upload goroutine escapes the recover(); ~36h worst-case time-to-first-ERROR. Co-Authored-By: Claude ReARM-Agent: 1420896f-adf5-4843-896f-d863cfcc6528 ReARM-Agentic-Session: 94afdb9d-05d3-44ad-abcd-92e753ddf2df --- cloud-backups/cmd/oci_backup.go | 36 ++++--- cloud-backups/internal/orchestrator/backup.go | 22 ++++- .../internal/orchestrator/backup_test.go | 55 ++++++++++- cloud-backups/internal/pipeline/stream.go | 36 +++++-- .../internal/pipeline/stream_test.go | 93 ++++++++++++++++++- cloud-backups/internal/registry/oras.go | 58 ++++++++++-- cloud-backups/internal/registry/oras_test.go | 43 ++++++++- cloud-backups/internal/stats/tracker.go | 29 +++++- cloud-backups/internal/stats/tracker_test.go | 53 +++++++++++ 9 files changed, 390 insertions(+), 35 deletions(-) diff --git a/cloud-backups/cmd/oci_backup.go b/cloud-backups/cmd/oci_backup.go index ea31817..c6b9e0d 100644 --- a/cloud-backups/cmd/oci_backup.go +++ b/cloud-backups/cmd/oci_backup.go @@ -130,18 +130,26 @@ 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 stays at INFO and the run still exits 0 -- 49 of 50 repos could - // vanish silently. Under explicit paths each target is one an operator named, - // so any skip is a missing backup and must be visible. Rolling months is - // exempt: it deliberately fabricates a previous-month path that legitimately - // may not exist yet, and alerting on that would recreate the noise this - // tool's retry levels were just fixed to avoid. - if !cfg.AppendRollingMonths { - if skipped := tracker.GetSkipped(); len(skipped) > 0 { + // 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", - "error", fmt.Sprintf("%d of %d explicitly configured target(s) produced no backup because the repository was not found: %s", - len(skipped), tracker.GetTotal(), strings.Join(skipped, ", "))) + "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, ", "))) } } @@ -151,6 +159,12 @@ func runBackup() error { 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 7b6aa48..d000680 100644 --- a/cloud-backups/internal/pipeline/stream.go +++ b/cloud-backups/internal/pipeline/stream.go @@ -81,6 +81,16 @@ func RunWithRetry(ctx context.Context, src datasource.Source, storeProvider stor } 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) @@ -96,6 +106,13 @@ func RunWithRetry(ctx context.Context, src datasource.Source, storeProvider stor slog.Warn("backup_attempt_failed", "target", target, "attempt", attempt, "error", err.Error()) if attempt < MaxBackupAttempts { 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 } } @@ -119,11 +136,16 @@ func isAuthRejection(err error) bool { } // 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 (up to -// MaxCauseLength * several, e.g. the 8KB oras tail), 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. Keeping the head -// preserves the part that names the failure. +// 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 @@ -131,8 +153,8 @@ func TruncateCause(cause string) string { // 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. - head := strings.ToValidUTF8(cause[:MaxCauseLength], "") - return head + fmt.Sprintf("... (truncated, %d bytes total)", len(cause)) + 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. diff --git a/cloud-backups/internal/pipeline/stream_test.go b/cloud-backups/internal/pipeline/stream_test.go index a040e56..441a242 100644 --- a/cloud-backups/internal/pipeline/stream_test.go +++ b/cloud-backups/internal/pipeline/stream_test.go @@ -622,12 +622,25 @@ func TestTruncateCause_BoundsAggregatedField(t *testing.T) { if len(got) >= len(long) { t.Fatalf("a long cause must shrink: got %d bytes, original %d", len(got), len(long)) } - if !strings.HasPrefix(got, strings.Repeat("x", MaxCauseLength)) { - t.Error("truncation must keep the head, which names the failure") - } 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 @@ -736,3 +749,77 @@ func TestLogRecords_NeverUseMsgAsAttrKey(t *testing.T) { }) } } + +// 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 293202f..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,18 +143,21 @@ 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)) - // tailBuffer, matching Backup and Restore above. An unbounded builder lets a - // chatty failure produce an arbitrarily large error string, and unlike its - // siblings this one is surfaced at ERROR straight into an alert payload. - stderrBuf := &tailBuffer{max: 8192} - cmd.Stderr = stderrBuf + // 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 if err := cmd.Run(); err != nil { logs := stderrBuf.String() 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 } @@ -163,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 f88685a..93654f5 100644 --- a/cloud-backups/internal/stats/tracker.go +++ b/cloud-backups/internal/stats/tracker.go @@ -73,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() @@ -106,11 +131,11 @@ func PrintSummary(eventName string, t *Tracker, storageType string, duration tim // 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, strings.Join(t.Skipped, ", "))) + t.Total, formatPaths(t.Skipped, t.SkippedCount))) } else if t.FailureCount > 0 { slog.Error("pipeline_completed_with_failures", "summary", summary, "error", fmt.Sprintf("%d of %d target(s) failed: %s", - t.FailureCount, t.Total, strings.Join(t.Failed, ", "))) + 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) + } +}