From 7a174ce5f1cab788045d96bef895a1363ffbffc4 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 31 May 2026 17:17:49 +0000 Subject: [PATCH] Harden cross-backend run diagnostics --- docs/cli.md | 2 +- docs/internal/app.md | 2 + docs/operations.md | 4 + docs/troubleshooting.md | 4 +- internal/app/run.go | 40 ++++--- internal/app/run_test.go | 232 ++++++++++++++++++++++++++++++++++++++- 6 files changed, 262 insertions(+), 22 deletions(-) diff --git a/docs/cli.md b/docs/cli.md index 89dbf96..61749b3 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -82,7 +82,7 @@ go run ./cmd/distributor run --config examples/fan-out.yml --dry-run ## Output -`run` prints the number of configured pipelines, one line per pipeline, one line per planned destination action, and a final status line. Actions include: +`run` prints the number of configured pipelines, one line per pipeline, one line per planned destination action, and a final status line. Destination action lines include the bundle path, destination id, destination backend, action, outputs, and reason. Actions include: - `publish_new`: destination has no managed state and is empty. - `replace_older`: destination state is older than the source manifest. diff --git a/docs/internal/app.md b/docs/internal/app.md index 38a5e3f..ee6b043 100644 --- a/docs/internal/app.md +++ b/docs/internal/app.md @@ -39,6 +39,8 @@ Dry-run still loads config, opens backends, discovers bundles, inspects destinat `Run` returns immediately for config loading errors, context cancellation before work starts, source open errors, and source discovery errors. Per-destination backend, planning, execution, and notification errors are aggregated into one run error after remaining destinations have been attempted. +Run diagnostics include pipeline id, destination id, destination backend, and bundle path for destination-scoped failures. Source open and discovery failures include the source backend. + Stdout write errors are returned immediately because the caller's requested output stream can no longer be trusted. ## Boundaries diff --git a/docs/operations.md b/docs/operations.md index 5d7c7ab..15462fe 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -76,6 +76,8 @@ Do not edit `.distributor.json` by hand during normal operation. If it is missin Dry-run output is useful before publishing to confirm actions such as `publish_new`, `replace_older`, `skip_same`, and `skip_destination_newer`. +Destination action lines include the destination backend, so mixed local, SSH, and S3 fan-out runs can be audited before publication. + ## Retry and Replacement Behavior If a destination has matching `.distributor.json`, publication skips it as already published. @@ -90,6 +92,8 @@ If a destination path has files but no valid `.distributor.json`, publication fa If one destination fails in a fan-out run, independent later destinations are still planned and executed. The command exits non-zero after printing the final status if any destination failed. +Errors include the pipeline id, destination id, destination backend, and bundle path where applicable. + If a write fails during publication, `distributor` attempts to remove outputs written during that failed attempt so a retry does not see those partial outputs as unmanaged destination content. After a successful publish or replacement, the internal notifier hook runs. The current default notifier is a no-op. Skipped destinations do not invoke it. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 8e18aaf..164291d 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -286,7 +286,7 @@ Safe fix: adjust the source bundle contents or publish policy so source and gene ## A run failed after writing some files -Likely cause: a write failed partway through publication. Local execution attempts to clean up outputs written during the failed attempt. +Likely cause: a write failed partway through publication. Local, SSH, and S3 execution attempt to clean up outputs written during the failed attempt. Diagnostic: @@ -294,4 +294,4 @@ Diagnostic: find -maxdepth 2 -print ``` -Safe fix: inspect the destination before retrying. If only unrelated unmanaged files remain, move them aside or choose a clean destination. Re-run with `--dry-run` before publishing again. See [operations](operations.md). +Safe fix: use the pipeline id, destination id, backend, and bundle path printed in the run error to inspect the destination before retrying. If only unrelated unmanaged files remain, move them aside or choose a clean destination. Re-run with `--dry-run` before publishing again. See [operations](operations.md). diff --git a/internal/app/run.go b/internal/app/run.go index dce8cab..78408e6 100644 --- a/internal/app/run.go +++ b/internal/app/run.go @@ -38,6 +38,12 @@ func Run(ctx context.Context, options RunOptions) error { } func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error { + return runConfigWithBackendFactory(ctx, cfg, options, newBackendFactoryWithEnvironment) +} + +type backendFactoryProvider func(config.Environment) *backendFactory + +func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider) error { notifier := options.Notifier if notifier == nil { notifier = notify.Noop{} @@ -53,7 +59,7 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error return err } } - backends := newBackendFactoryWithEnvironment(secretLoad.Environment) + backends := provider(secretLoad.Environment) transforms := newTransformRegistry() if options.Stdout != nil { if _, err := fmt.Fprintf(options.Stdout, "Configured pipelines: %d\n", len(cfg.Pipelines)); err != nil { @@ -68,12 +74,12 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error } sourceBackend, err := backends.openSource(ctx, pipeline.Source) if err != nil { - return fmt.Errorf("pipeline %s: %w", pipeline.ID, err) + return fmt.Errorf("pipeline %s source backend %s: %w", pipeline.ID, pipeline.Source.Backend, err) } bundles, err := bundle.Discover(ctx, sourceBackend, "") if err != nil { closeBackend(sourceBackend) - return fmt.Errorf("pipeline %s discover source bundles: %w", pipeline.ID, err) + return fmt.Errorf("pipeline %s source backend %s discover source bundles: %w", pipeline.ID, pipeline.Source.Backend, err) } if options.Stdout != nil { if _, err := fmt.Fprintf(options.Stdout, "- pipeline=%s source=%s bundles=%d destinations=%s\n", pipeline.ID, pipeline.Source.Backend, len(bundles), destinationSummary(pipeline.Destinations)); err != nil { @@ -85,10 +91,10 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error for _, destination := range pipeline.Destinations { destinationBackend, err := backends.openDestination(ctx, destination) if err != nil { - failures.add(pipeline.ID, destination.ID, storage.DisplayPath(sourceBundle.RootRelativePath), err) + failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err) summary.recordFailure() if options.Stdout != nil { - writeErrorLine(options.Stdout, sourceBundle.RootRelativePath, destination.ID, err) + writeErrorLine(options.Stdout, sourceBundle.RootRelativePath, destination.ID, destination.Backend, err) } continue } @@ -117,11 +123,11 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error plan = publish.Plan{DestinationID: destination.ID, BundlePath: sourceBundle.RootRelativePath} } if options.Stdout != nil { - writePlanLine(options.Stdout, plan, err) + writePlanLine(options.Stdout, destination.Backend, plan, err) } if err != nil { deferCloseDestination() - failures.add(pipeline.ID, destination.ID, storage.DisplayPath(sourceBundle.RootRelativePath), err) + failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err) summary.recordFailure() continue } @@ -129,14 +135,14 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error if !options.DryRun { if err := publish.Execute(ctx, req, plan); err != nil { deferCloseDestination() - failures.add(pipeline.ID, destination.ID, storage.DisplayPath(sourceBundle.RootRelativePath), err) + failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err) summary.recordFailure() continue } if shouldNotify(plan.Action) { if err := notifier.Notify(ctx, notifyEvent(plan)); err != nil { deferCloseDestination() - failures.add(pipeline.ID, destination.ID, storage.DisplayPath(sourceBundle.RootRelativePath), err) + failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err) summary.recordFailure() continue } @@ -170,7 +176,7 @@ func closeBackend(backend storage.Backend) { _ = closeable.Close() } -func writePlanLine(w io.Writer, plan publish.Plan, planErr error) { +func writePlanLine(w io.Writer, backend string, plan publish.Plan, planErr error) { if w == nil { return } @@ -179,17 +185,17 @@ func writePlanLine(w io.Writer, plan publish.Plan, planErr error) { if destinationID == "" { destinationID = "unknown" } - fmt.Fprintf(w, " - bundle=%s destination=%s action=error reason=%q\n", storage.DisplayPath(plan.BundlePath), destinationID, planErr.Error()) + fmt.Fprintf(w, " - bundle=%s destination=%s backend=%s action=error reason=%q\n", storage.DisplayPath(plan.BundlePath), destinationID, backend, planErr.Error()) return } - fmt.Fprintf(w, " - bundle=%s destination=%s action=%s outputs=%s reason=%q\n", storage.DisplayPath(plan.BundlePath), plan.DestinationID, plan.Action, outputSummary(plan.Outputs), plan.Reason) + fmt.Fprintf(w, " - bundle=%s destination=%s backend=%s action=%s outputs=%s reason=%q\n", storage.DisplayPath(plan.BundlePath), plan.DestinationID, backend, plan.Action, outputSummary(plan.Outputs), plan.Reason) } -func writeErrorLine(w io.Writer, bundlePath, destinationID string, err error) { +func writeErrorLine(w io.Writer, bundlePath, destinationID, backend string, err error) { if w == nil { return } - fmt.Fprintf(w, " - bundle=%s destination=%s action=error reason=%q\n", storage.DisplayPath(bundlePath), destinationID, err.Error()) + fmt.Fprintf(w, " - bundle=%s destination=%s backend=%s action=error reason=%q\n", storage.DisplayPath(bundlePath), destinationID, backend, err.Error()) } func outputSummary(outputs []publish.Output) string { @@ -301,6 +307,7 @@ func (s runSummary) Line() string { type runFailure struct { pipelineID string destinationID string + backend string bundlePath string err error } @@ -309,10 +316,11 @@ type runFailures struct { items []runFailure } -func (f *runFailures) add(pipelineID, destinationID, bundlePath string, err error) { +func (f *runFailures) add(pipelineID, destinationID, backend, bundlePath string, err error) { f.items = append(f.items, runFailure{ pipelineID: pipelineID, destinationID: destinationID, + backend: backend, bundlePath: bundlePath, err: err, }) @@ -324,7 +332,7 @@ func (f runFailures) Error() string { } parts := make([]string, 0, len(f.items)) for _, item := range f.items { - parts = append(parts, fmt.Sprintf("pipeline %s destination %s bundle %s: %v", item.pipelineID, item.destinationID, item.bundlePath, item.err)) + parts = append(parts, fmt.Sprintf("pipeline %s destination %s backend %s bundle %s: %v", item.pipelineID, item.destinationID, item.backend, item.bundlePath, item.err)) } return "run failed: " + strings.Join(parts, "; ") } diff --git a/internal/app/run_test.go b/internal/app/run_test.go index aa4ce86..65bd187 100644 --- a/internal/app/run_test.go +++ b/internal/app/run_test.go @@ -10,11 +10,13 @@ import ( "testing" "time" + "gitea.maximumdirect.net/eric/distributor/internal/adapters/local" "gitea.maximumdirect.net/eric/distributor/internal/bundle" "gitea.maximumdirect.net/eric/distributor/internal/config" "gitea.maximumdirect.net/eric/distributor/internal/notify" "gitea.maximumdirect.net/eric/distributor/internal/state" "gitea.maximumdirect.net/eric/distributor/internal/storage" + "gitea.maximumdirect.net/eric/distributor/internal/storage/fake" "gitea.maximumdirect.net/eric/distributor/internal/testutil" ) @@ -38,7 +40,7 @@ func TestRunDryRunPrintsConfigSummary(t *testing.T) { for _, want := range []string{ "Configured pipelines: 1", "- pipeline=reports source=local bundles=1 destinations=archive", - "bundle=. destination=archive action=publish_new outputs=report.md,summary.txt", + "bundle=. destination=archive backend=local action=publish_new outputs=report.md,summary.txt", "Final status: ok planned=1 publish_new=1 replace_older=0 skipped=0 failed=0 dry_run=true", } { if !strings.Contains(output, want) { @@ -292,10 +294,13 @@ func TestRunContinuesAfterDestinationFailure(t *testing.T) { if err == nil || !strings.Contains(err.Error(), "fail_unmanaged") { t.Fatalf("Run() error = %v, want unmanaged failure", err) } + if !strings.Contains(err.Error(), "pipeline reports destination archive-one backend local bundle .") { + t.Fatalf("Run() error = %v, want backend context", err) + } output := stdout.String() for _, want := range []string{ - "destination=archive-one action=error", - "destination=archive-two action=publish_new", + "destination=archive-one backend=local action=error", + "destination=archive-two backend=local action=publish_new", "Final status: failed planned=1 publish_new=1 replace_older=0 skipped=0 failed=1 dry_run=false", } { if !strings.Contains(output, want) { @@ -506,6 +511,93 @@ func TestRunFansOutToLocalDestinations(t *testing.T) { assertFile(t, filepath.Join(secondDestination, "daily", "summary.txt"), "Summary\n") } +func TestRunFansOutWithDifferentPublishPolicies(t *testing.T) { + sourceRoot := t.TempDir() + archiveDestination := t.TempDir() + htmlDestination := t.TempDir() + writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + + err := Run(context.Background(), RunOptions{ConfigPath: writeMixedPolicyFanoutConfig(t, sourceRoot, archiveDestination, htmlDestination)}) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + assertFile(t, filepath.Join(archiveDestination, "report.md"), "# Report\nSunny.\n") + assertFile(t, filepath.Join(archiveDestination, "summary.txt"), "Summary\n") + if _, err := os.Stat(filepath.Join(archiveDestination, "report.html")); !os.IsNotExist(err) { + t.Fatalf("archive report.html stat error = %v, want not exist", err) + } + assertFileContains(t, filepath.Join(htmlDestination, "report.html"), "

Report

") + if _, err := os.Stat(filepath.Join(htmlDestination, "report.md")); !os.IsNotExist(err) { + t.Fatalf("html report.md stat error = %v, want not exist", err) + } +} + +func TestRunExercisesRemoteBackendShapesThroughCommonPath(t *testing.T) { + localSourceRoot := t.TempDir() + writeSourceBundle(t, localSourceRoot, "", testBundleOptions{}) + s3Source := fake.New() + testutil.WriteFakeSourceBundle(t, s3Source, "", testutil.BundleOptions{}) + sshSource := fake.New() + testutil.WriteFakeSourceBundle(t, sshSource, "", testutil.BundleOptions{}) + s3Destination := fake.New() + sshDestination := fake.New() + s3ToLocalDestination := t.TempDir() + sshToLocalDestination := t.TempDir() + + cfg := crossBackendConfig(localSourceRoot, s3ToLocalDestination, sshToLocalDestination) + if err := config.Validate(cfg); err != nil { + t.Fatalf("cross-backend config validation error = %v", err) + } + provider := fakeBackendFactoryProvider(t, map[string]storage.Backend{ + "s3:source-bucket": s3Source, + "s3:destination-bucket": s3Destination, + "ssh:/source": sshSource, + "ssh:/destination": sshDestination, + }) + + var dryRunOutput bytes.Buffer + if err := runConfigWithBackendFactory(context.Background(), cfg, RunOptions{DryRun: true, Stdout: &dryRunOutput}, provider); err != nil { + t.Fatalf("dry-run error = %v", err) + } + for _, want := range []string{ + "pipeline=local-to-s3 source=local", + "destination=object-archive backend=s3 action=publish_new", + "pipeline=s3-to-local source=s3", + "destination=local-archive backend=local action=publish_new", + "pipeline=local-to-ssh source=local", + "destination=ssh-archive backend=ssh action=publish_new", + "pipeline=ssh-to-local source=ssh", + "Final status: ok planned=4 publish_new=4 replace_older=0 skipped=0 failed=0 dry_run=true", + } { + if !strings.Contains(dryRunOutput.String(), want) { + t.Fatalf("dry-run output = %q, want substring %q", dryRunOutput.String(), want) + } + } + if hasAny, err := s3Destination.HasAny(context.Background(), ""); err != nil || hasAny { + t.Fatalf("s3 destination after dry-run hasAny=%t err=%v, want empty", hasAny, err) + } + if entries, err := os.ReadDir(s3ToLocalDestination); err != nil || len(entries) != 0 { + t.Fatalf("s3-to-local destination entries = %v err=%v, want empty", entries, err) + } + + var publishOutput bytes.Buffer + if err := runConfigWithBackendFactory(context.Background(), cfg, RunOptions{Stdout: &publishOutput}, provider); err != nil { + t.Fatalf("publish error = %v", err) + } + assertFakeFile(t, s3Destination, "report.md", "# Report\nSunny.\n") + assertFakeFile(t, sshDestination, "summary.txt", "Summary\n") + assertFile(t, filepath.Join(s3ToLocalDestination, "report.md"), "# Report\nSunny.\n") + assertFile(t, filepath.Join(sshToLocalDestination, "summary.txt"), "Summary\n") + + var repeatOutput bytes.Buffer + if err := runConfigWithBackendFactory(context.Background(), cfg, RunOptions{Stdout: &repeatOutput}, provider); err != nil { + t.Fatalf("repeat error = %v", err) + } + if got := strings.Count(repeatOutput.String(), "action=skip_same"); got != 4 { + t.Fatalf("repeat output = %q, skip_same count = %d, want 4", repeatOutput.String(), got) + } +} + func TestRunDryRunDoesNotWrite(t *testing.T) { sourceRoot := t.TempDir() destinationRoot := t.TempDir() @@ -583,6 +675,34 @@ func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestina return testutil.WriteFanoutLocalConfig(t, sourceRoot, firstDestination, secondDestination) } +func writeMixedPolicyFanoutConfig(t *testing.T, sourceRoot, archiveDestination, htmlDestination string) string { + t.Helper() + return writeConfigFile(t, ` +pipelines: + - id: reports + source: + backend: local + path: `+sourceRoot+` + destinations: + - id: archive + backend: local + path: `+archiveDestination+` + publish: + source: true + html: false + - id: html + backend: local + path: `+htmlDestination+` + publish: + source: false + html: true + transform: + markdown_to_html: + enabled: true + mode: sidecar +`) +} + func writeConfigFile(t *testing.T, body string) string { t.Helper() path := filepath.Join(t.TempDir(), "config.yml") @@ -624,6 +744,112 @@ func assertFileContains(t *testing.T, path, want string) { } } +func assertFakeFile(t *testing.T, backend *fake.Backend, path, want string) { + t.Helper() + data, err := backend.ReadFile(context.Background(), path) + if err != nil { + t.Fatalf("read fake file %s: %v", path, err) + } + if got := string(data); got != want { + t.Fatalf("%s = %q, want %q", path, got, want) + } +} + +func crossBackendConfig(localSourceRoot, s3ToLocalDestination, sshToLocalDestination string) config.Config { + cfg := config.Config{ + Pipelines: []config.Pipeline{ + { + ID: "local-to-s3", + Source: config.Backend{Backend: config.BackendLocal, Path: localSourceRoot}, + Destinations: []config.Destination{{ + ID: "object-archive", + Backend: config.BackendS3, + Endpoint: "http://s3.test", + Bucket: "destination-bucket", + }}, + }, + { + ID: "s3-to-local", + Source: config.Backend{ + Backend: config.BackendS3, + Endpoint: "http://s3.test", + Bucket: "source-bucket", + }, + Destinations: []config.Destination{{ + ID: "local-archive", + Backend: config.BackendLocal, + Path: s3ToLocalDestination, + }}, + }, + { + ID: "local-to-ssh", + Source: config.Backend{Backend: config.BackendLocal, Path: localSourceRoot}, + Destinations: []config.Destination{{ + ID: "ssh-archive", + Backend: config.BackendSSH, + Host: "ssh.test", + Path: "/destination", + }}, + }, + { + ID: "ssh-to-local", + Source: config.Backend{ + Backend: config.BackendSSH, + Host: "ssh.test", + Path: "/source", + }, + Destinations: []config.Destination{{ + ID: "local-archive", + Backend: config.BackendLocal, + Path: sshToLocalDestination, + }}, + }, + }, + } + config.ApplyDefaults(&cfg) + return cfg +} + +func fakeBackendFactoryProvider(t *testing.T, remoteBackends map[string]storage.Backend) backendFactoryProvider { + t.Helper() + return func(environment config.Environment) *backendFactory { + registry := storage.NewRegistry() + if err := registry.Register(config.BackendLocal, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + return local.New(cfg[storagePathKey]) + }); err != nil { + t.Fatalf("register local backend: %v", err) + } + if err := registry.Register(config.BackendS3, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + backend := remoteBackends["s3:"+cfg[s3BucketKey]] + if backend == nil { + return nil, fmt.Errorf("missing fake s3 backend for bucket %s", cfg[s3BucketKey]) + } + return backend, nil + }); err != nil { + t.Fatalf("register s3 backend: %v", err) + } + if err := registry.Register(config.BackendSSH, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + backend := remoteBackends["ssh:"+cfg[storagePathKey]] + if backend == nil { + return nil, fmt.Errorf("missing fake ssh backend for path %s", cfg[storagePathKey]) + } + return backend, nil + }); err != nil { + t.Fatalf("register ssh backend: %v", err) + } + return &backendFactory{registry: registry, environment: environment} + } +} + type recordingNotifier struct { events []notify.Event check func()