eric/distributor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

Harden cross-backend run diagnostics

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-31 17:17:49 +00:00
parent 14fa9c8000
commit 7a174ce5f1
6 changed files with 262 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
internal/app/run.go
View File

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

232
internal/app/run_test.go
View File

@@ -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"), "<h1>Report</h1>")
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()