From f9436a74234ce21fb2739a3ed7add9b098699685 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Wed, 3 Jun 2026 15:10:07 +0000 Subject: [PATCH] Add local source pipeline execution --- docs/internal/app.md | 20 ++++++ internal/app/run.go | 98 ++++++++++++++++++++++++--- internal/app/run_test.go | 140 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 250 insertions(+), 8 deletions(-) diff --git a/docs/internal/app.md b/docs/internal/app.md index 16cef77..63c0ee8 100644 --- a/docs/internal/app.md +++ b/docs/internal/app.md @@ -23,6 +23,13 @@ config path, pipeline ID, dry-run flag, force flag, and optional notifier. It loads the same config as `Run`, narrows execution to exactly one configured pipeline, and returns a `RunReport` without writing command output. +`RunPipelineWithLocalSource` is the app-layer single-pipeline entrypoint for an +already prepared local source bundle root. It accepts the same pipeline +selection and execution options as `RunPipeline` plus a local source root path. +It loads config, selects one configured pipeline, opens the supplied source +root as a local backend, validates exactly that root bundle, and then uses the +same destination fan-out path as normal runs. + `Validate` and `Inspect` accept either a local path or one configured pipeline source. They share source backend construction with run workflows and never open destination backends. @@ -64,6 +71,14 @@ pipeline. It uses the same backend factory, secret loading, transform registry, warning generation, destination planning, publish execution, notification behavior, and failure aggregation as `Run`. +`RunPipelineWithLocalSource` follows the same flow after pipeline selection +except for source opening and source discovery. It opens the supplied local +source root directly, validates the root bundle before opening any destinations, +and passes the resulting local source backend and bundle into the same +destination planning and execution loop. Destination code receives the normal +storage backend and bundle values and does not depend on how the source root was +prepared. + ## Coordination `PipelineRunCoordinator` wraps `RunPipeline` with in-memory admission control. @@ -89,6 +104,11 @@ work starts, source open errors, and source discovery errors. `RunPipeline` returns `PipelineNotFoundError` when the requested pipeline ID is not configured. Callers can detect that condition with `IsPipelineNotFound`. +`RunPipelineWithLocalSource` also returns `PipelineNotFoundError` for an unknown +pipeline ID. It returns before destination opening when the supplied local +source root is missing, cannot be opened, or does not validate as one complete +source bundle. + Per-destination backend, planning, execution, and notification errors are aggregated into one run error after remaining destinations have been attempted. Destination diagnostics include pipeline ID, destination ID, backend, and diff --git a/internal/app/run.go b/internal/app/run.go index 63389ba..9d3bc58 100644 --- a/internal/app/run.go +++ b/internal/app/run.go @@ -29,6 +29,15 @@ type RunPipelineOptions struct { Notifier notify.Notifier } +type RunPipelineWithLocalSourceOptions struct { + ConfigPath string + PipelineID string + SourceRoot string + DryRun bool + Force bool + Notifier notify.Notifier +} + func Run(ctx context.Context, options RunOptions) error { if err := ValidateOutputFormat(options.OutputFormat); err != nil { return err @@ -64,6 +73,25 @@ func RunPipeline(ctx context.Context, options RunPipelineOptions) (RunReport, er return runPipelineConfig(ctx, cfg, options) } +func RunPipelineWithLocalSource(ctx context.Context, options RunPipelineWithLocalSourceOptions) (RunReport, error) { + if err := ctx.Err(); err != nil { + return RunReport{}, err + } + if options.SourceRoot == "" { + return RunReport{}, fmt.Errorf("source root is required") + } + + configPath := options.ConfigPath + if configPath == "" { + configPath = config.DefaultConfigPath + } + cfg, err := config.LoadFile(configPath) + if err != nil { + return RunReport{}, err + } + return runPipelineConfigWithLocalSource(ctx, cfg, options) +} + func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error { return runConfigWithBackendFactory(ctx, cfg, options, newBackendFactoryWithEnvironment) } @@ -74,12 +102,17 @@ func runPipelineConfig(ctx context.Context, cfg config.Config, options RunPipeli return runPipelineConfigWithBackendFactory(ctx, cfg, options, newBackendFactoryWithEnvironment) } +func runPipelineConfigWithLocalSource(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) { + return runPipelineConfigWithLocalSourceAndBackendFactory(ctx, cfg, options, newBackendFactoryWithEnvironment) +} + func runPipelineConfigWithBackendFactory(ctx context.Context, cfg config.Config, options RunPipelineOptions, provider backendFactoryProvider) (RunReport, error) { pipeline, ok := findPipeline(cfg, options.PipelineID) if !ok { return RunReport{}, PipelineNotFoundError{ID: options.PipelineID} } return buildRunReportWithBackendFactory(ctx, config.Config{ + Server: cfg.Server, Secrets: cfg.Secrets, Pipelines: []config.Pipeline{pipeline}, }, RunOptions{ @@ -89,6 +122,25 @@ func runPipelineConfigWithBackendFactory(ctx context.Context, cfg config.Config, }, provider) } +func runPipelineConfigWithLocalSourceAndBackendFactory(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions, provider backendFactoryProvider) (RunReport, error) { + pipeline, ok := findPipeline(cfg, options.PipelineID) + if !ok { + return RunReport{}, PipelineNotFoundError{ID: options.PipelineID} + } + return buildRunReport(ctx, config.Config{ + Server: cfg.Server, + Secrets: cfg.Secrets, + Pipelines: []config.Pipeline{pipeline}, + }, RunOptions{ + DryRun: options.DryRun, + Force: options.Force, + Notifier: options.Notifier, + }, provider, &localSourceRoot{ + pipelineID: options.PipelineID, + root: options.SourceRoot, + }) +} + func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider) error { report, err := buildRunReportWithBackendFactory(ctx, cfg, options, provider) if err != nil && !IsPartialResultError(err) { @@ -101,6 +153,15 @@ func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options } func buildRunReportWithBackendFactory(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider) (RunReport, error) { + return buildRunReport(ctx, cfg, options, provider, nil) +} + +type localSourceRoot struct { + pipelineID string + root string +} + +func buildRunReport(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider, sourceRoot *localSourceRoot) (RunReport, error) { notifier := options.Notifier if notifier == nil { notifier = notify.Noop{} @@ -125,18 +186,13 @@ func buildRunReportWithBackendFactory(ctx context.Context, cfg config.Config, op for _, pipeline := range cfg.Pipelines { pipelineWarnings := sshWarnings(pipeline) report.addWarnings(pipelineWarnings) - sourceBackend, err := backends.openSource(ctx, pipeline.Source) + sourceBackend, bundles, sourceBackendName, err := openPipelineSource(ctx, backends, pipeline, sourceRoot) if err != nil { - return report, 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 report, fmt.Errorf("pipeline %s source backend %s discover source bundles: %w", pipeline.ID, pipeline.Source.Backend, err) + return report, err } report.Pipelines = append(report.Pipelines, RunPipelineSummary{ ID: pipeline.ID, - SourceBackend: pipeline.Source.Backend, + SourceBackend: sourceBackendName, BundleCount: len(bundles), Destinations: destinationIDs(pipeline.Destinations), Warnings: pipelineWarnings, @@ -251,6 +307,32 @@ func buildRunReportWithBackendFactory(ctx context.Context, cfg config.Config, op return report, nil } +func openPipelineSource(ctx context.Context, backends *backendFactory, pipeline config.Pipeline, sourceRoot *localSourceRoot) (storage.Backend, []bundle.Bundle, string, error) { + if sourceRoot != nil && sourceRoot.pipelineID == pipeline.ID { + sourceBackend, err := backends.openLocalPath(ctx, sourceRoot.root) + if err != nil { + return nil, nil, config.BackendLocal, fmt.Errorf("pipeline %s source backend %s: %w", pipeline.ID, config.BackendLocal, err) + } + sourceBundle, err := bundle.Validate(ctx, sourceBackend, "") + if err != nil { + closeBackend(sourceBackend) + return nil, nil, config.BackendLocal, fmt.Errorf("pipeline %s source backend %s validate source bundle: %w", pipeline.ID, config.BackendLocal, err) + } + return sourceBackend, []bundle.Bundle{sourceBundle}, config.BackendLocal, nil + } + + sourceBackend, err := backends.openSource(ctx, pipeline.Source) + if err != nil { + return nil, nil, pipeline.Source.Backend, 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 nil, nil, pipeline.Source.Backend, fmt.Errorf("pipeline %s source backend %s discover source bundles: %w", pipeline.ID, pipeline.Source.Backend, err) + } + return sourceBackend, bundles, pipeline.Source.Backend, nil +} + type closeableBackend interface { Close() error } diff --git a/internal/app/run_test.go b/internal/app/run_test.go index a87c108..1cb5b57 100644 --- a/internal/app/run_test.go +++ b/internal/app/run_test.go @@ -3,6 +3,7 @@ package app import ( "bytes" "context" + "encoding/json" "fmt" "os" "path/filepath" @@ -220,6 +221,116 @@ func TestRunPublishesNewLocalBundle(t *testing.T) { } } +func TestRunPipelineWithLocalSourcePublishesConfiguredDestination(t *testing.T) { + sourceRoot := t.TempDir() + destinationRoot := t.TempDir() + writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + + report, err := RunPipelineWithLocalSource(context.Background(), RunPipelineWithLocalSourceOptions{ + ConfigPath: writeUploadPipelineConfig(t, destinationRoot), + PipelineID: "reports", + SourceRoot: sourceRoot, + }) + if err != nil { + t.Fatalf("RunPipelineWithLocalSource() error = %v", err) + } + + if got, want := report.Summary.Status, "ok"; got != want { + t.Fatalf("report status = %q, want %q", got, want) + } + if got, want := len(report.Pipelines), 1; got != want { + t.Fatalf("pipeline count = %d, want %d", got, want) + } + if got, want := report.Pipelines[0].SourceBackend, config.BackendLocal; got != want { + t.Fatalf("source backend = %q, want %q", got, want) + } + if got, want := report.Pipelines[0].BundleCount, 1; got != want { + t.Fatalf("bundle count = %d, want %d", got, want) + } + if got, want := len(report.Actions), 1; got != want { + t.Fatalf("action count = %d, want %d", got, want) + } + if report.Actions[0].PipelineID != "reports" || report.Actions[0].DestinationID != "archive" { + t.Fatalf("action = %#v, want reports/archive action", report.Actions[0]) + } + testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n") + testutil.AssertFile(t, filepath.Join(destinationRoot, "summary.txt"), "Summary\n") +} + +func TestRunPipelineWithLocalSourceValidatesBeforeDestinationWrites(t *testing.T) { + sourceRoot := t.TempDir() + destinationRoot := t.TempDir() + writeJSONManifest(t, sourceRoot, testutil.ValidManifest(testutil.BundleOptions{})) + + _, err := RunPipelineWithLocalSource(context.Background(), RunPipelineWithLocalSourceOptions{ + ConfigPath: writeUploadPipelineConfig(t, destinationRoot), + PipelineID: "reports", + SourceRoot: sourceRoot, + }) + if err == nil { + t.Fatal("RunPipelineWithLocalSource() error = nil, want validation error") + } + if !strings.Contains(err.Error(), "validate source bundle") { + t.Fatalf("RunPipelineWithLocalSource() error = %v, want source validation context", err) + } + entries, readErr := os.ReadDir(destinationRoot) + if readErr != nil { + t.Fatalf("ReadDir() error = %v", readErr) + } + if len(entries) != 0 { + t.Fatalf("destination entries = %d, want no writes", len(entries)) + } +} + +func TestRunPipelineWithLocalSourcePublishesToRegisteredDestinationBackends(t *testing.T) { + sourceRoot := t.TempDir() + writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + s3Destination := fake.New() + sshDestination := fake.New() + cfg := config.Config{ + Pipelines: []config.Pipeline{{ + ID: "reports", + Source: config.Backend{ + Backend: config.BackendHTTPUpload, + Upload: config.HTTPUpload{TokenEnv: "UPLOAD_TOKEN"}, + }, + Destinations: []config.Destination{ + { + ID: "object-archive", + Backend: config.BackendS3, + Endpoint: "http://s3.test", + Bucket: "destination-bucket", + }, + { + ID: "ssh-archive", + Backend: config.BackendSSH, + Host: "ssh.test", + Path: "/destination", + }, + }, + }}, + } + config.ApplyDefaults(&cfg) + provider := fakeBackendFactoryProvider(t, map[string]storage.Backend{ + "s3:destination-bucket": s3Destination, + "ssh:/destination": sshDestination, + }) + + report, err := runPipelineConfigWithLocalSourceAndBackendFactory(context.Background(), cfg, RunPipelineWithLocalSourceOptions{ + PipelineID: "reports", + SourceRoot: sourceRoot, + }, provider) + if err != nil { + t.Fatalf("runPipelineConfigWithLocalSourceAndBackendFactory() error = %v", err) + } + + if got, want := len(report.Actions), 2; got != want { + t.Fatalf("action count = %d, want %d", got, want) + } + testutil.AssertFakeFile(t, s3Destination, "report.md", "# Report\nSunny.\n") + testutil.AssertFakeFile(t, sshDestination, "summary.txt", "Summary\n") +} + func TestRunExplicitPreserveRelativePathMappingMatchesDefault(t *testing.T) { sourceRoot := t.TempDir() destinationRoot := t.TempDir() @@ -1516,6 +1627,21 @@ func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestina return testutil.WriteFanoutLocalConfig(t, sourceRoot, firstDestination, secondDestination) } +func writeUploadPipelineConfig(t *testing.T, destinationRoot string) string { + t.Helper() + return writeConfigFile(t, ` +pipelines: + - id: reports + source: + backend: http_upload + token_env: UPLOAD_TOKEN + destinations: + - id: archive + backend: local + path: `+destinationRoot+` +`) +} + func writeTwoPipelineConfig(t *testing.T, firstSource, firstDestination, secondSource, secondDestination string) string { t.Helper() return writeConfigFile(t, ` @@ -1553,6 +1679,20 @@ func writeDestinationState(t *testing.T, root, relative string, manifest bundle. testutil.WriteDestinationState(t, root, relative, manifest, testutil.DestinationStateOptions{}) } +func writeJSONManifest(t *testing.T, root string, manifest bundle.Manifest) { + t.Helper() + if err := os.MkdirAll(root, 0o755); err != nil { + t.Fatalf("mkdir manifest root: %v", err) + } + data, err := json.MarshalIndent(manifest, "", " ") + if err != nil { + t.Fatalf("marshal manifest: %v", err) + } + if err := os.WriteFile(filepath.Join(root, bundle.ManifestName), append(data, '\n'), 0o600); err != nil { + t.Fatalf("write manifest: %v", err) + } +} + func readStateFile(t *testing.T, path string) state.DistributorState { t.Helper() return testutil.ReadDestinationState(t, path)