From 0f1ef9e622c79b5f2b73405436ecd22923f10a74 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Wed, 3 Jun 2026 15:26:49 +0000 Subject: [PATCH] Add HTTP upload end-to-end coverage --- docs/operations.md | 82 +++- internal/app/upload_http_integration_test.go | 419 +++++++++++++++++++ 2 files changed, 500 insertions(+), 1 deletion(-) create mode 100644 internal/app/upload_http_integration_test.go diff --git a/docs/operations.md b/docs/operations.md index a0c8d6c..c4d5dec 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -68,6 +68,83 @@ Validate one configured source without opening destinations: go run ./cmd/distributor validate --config examples/local-publish.yml --pipeline example-source-bundle ``` +## HTTP Upload Workflow + +`distributor serve` runs the HTTP upload API for pipelines whose source backend +is `http_upload`. Each upload token maps to one configured pipeline, and each +accepted archive is staged, validated, and published through the same +destination fan-out path used by local source runs. + +Minimal local HTTP upload configuration: + +```yaml +server: + http: + bind: 127.0.0.1:8080 + staging_root: /var/spool/distributor + max_upload_size: 20MB + queue_size: 16 + max_concurrency: 1 + retention: 24h +secrets: + directory: /run/secrets/distributor +pipelines: + - id: reports + source: + backend: http_upload + token_env: REPORTS_UPLOAD_TOKEN + destinations: + - id: archive + backend: local + path: /srv/reports/archive +``` + +Create `/run/secrets/distributor/REPORTS_UPLOAD_TOKEN` or set the real process +environment variable before starting the server. Distributor does not read +literal upload tokens from YAML. + +Start the server: + +```sh +go run ./cmd/distributor serve --config +``` + +Submit a tar or tar.gz source bundle: + +```sh +curl -X POST http://127.0.0.1:8080/upload \ + -H "Authorization: Bearer $REPORTS_UPLOAD_TOKEN" \ + -H "Content-Type: application/gzip" \ + --data-binary @bundle.tar.gz +``` + +Successful admission returns a run id: + +```json +{"run_id":"reports.20260603T120000Z.abcdef12","status":"accepted"} +``` + +Poll status until it reaches `succeeded` or `failed`: + +```sh +curl http://127.0.0.1:8080/runs/ +``` + +The status record includes the completed run report on successful publication +or error details on failure. Status is memory-only and expires after +`server.http.retention`; completed staged bundle directories are removed on +expiry. Restarting the process clears upload status and queue state. + +Use `GET /healthz` for readiness after config and tokens load: + +```sh +curl http://127.0.0.1:8080/healthz +``` + +The default bind address is private loopback. Put TLS, public routing, +rate-limiting, and external access policy in a reverse proxy or deployment +layer. + ## Filesystem Layout Source bundles are discovered beneath the configured source root. Each bundle is a directory containing `manifest.json`. @@ -322,7 +399,10 @@ secrets: directory: /run/secrets/distributor ``` -The directory is loaded during `run` and configured-source `validate` or `inspect` before any backend is opened. If the directory is missing, unreadable, or contains an invalid secret filename, the command fails before storage work starts. +The directory is loaded during `run`, `serve`, and configured-source `validate` +or `inspect` before credential-consuming work starts. If the directory is +missing, unreadable, or contains an invalid secret filename, the command fails +before storage work starts. Real process environment values take precedence over files with the same name. If the values differ and stdout is enabled, `run` and configured-source diagnostics print a warning naming the ignored secret file variable without printing either value. The process environment is not changed. diff --git a/internal/app/upload_http_integration_test.go b/internal/app/upload_http_integration_test.go new file mode 100644 index 0000000..ca98e1d --- /dev/null +++ b/internal/app/upload_http_integration_test.go @@ -0,0 +1,419 @@ +package app + +import ( + "archive/tar" + "bytes" + "compress/gzip" + "context" + "encoding/json" + "fmt" + "io" + "io/fs" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + "time" + + "gitea.maximumdirect.net/eric/distributor/internal/config" + "gitea.maximumdirect.net/eric/distributor/internal/ingest" + "gitea.maximumdirect.net/eric/distributor/internal/storage" + "gitea.maximumdirect.net/eric/distributor/internal/testutil" +) + +func TestHTTPUploadPublishesTarAndGzipFanout(t *testing.T) { + tests := []struct { + name string + compressed bool + contentType string + }{ + {name: "tar", contentType: ingest.ContentTypeTar}, + {name: "gzip", compressed: true, contentType: ingest.ContentTypeGzip}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + firstDestination := t.TempDir() + secondDestination := t.TempDir() + cfg := httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{ + id: "reports", + tokenEnv: "REPORTS_TOKEN", + stagingPath: filepath.Join(t.TempDir(), "reports"), + destinations: []string{firstDestination, secondDestination}, + }}, 4, 1) + handler, err := newUploadHTTPHandler(context.Background(), cfg, uploadHTTPTestEnvironment(map[string]string{ + "REPORTS_TOKEN": "reports-secret", + })) + if err != nil { + t.Fatalf("newUploadHTTPHandler() error = %v", err) + } + server := httptest.NewServer(handler) + defer server.Close() + + runID := submitHTTPUpload(t, server, "reports-secret", tt.contentType, bundleArchive(t, tt.compressed, testutil.BundleOptions{})) + record := waitForHTTPUploadStatus(t, server, runID, UploadStatusSucceeded) + + if record.Report == nil { + t.Fatal("completed status report = nil, want run report") + } + if record.Report.Summary.Status != "ok" { + t.Fatalf("summary status = %q, want ok", record.Report.Summary.Status) + } + if got, want := len(record.Report.Actions), 2; got != want { + t.Fatalf("action count = %d, want %d", got, want) + } + assertPublishedBundle(t, firstDestination) + assertPublishedBundle(t, secondDestination) + }) + } +} + +func TestHTTPUploadInvalidArchiveFailsWithoutPublishing(t *testing.T) { + destination := t.TempDir() + cfg := httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{ + id: "reports", + tokenEnv: "REPORTS_TOKEN", + stagingPath: filepath.Join(t.TempDir(), "reports"), + destinations: []string{destination}, + }}, 4, 1) + handler, err := newUploadHTTPHandler(context.Background(), cfg, uploadHTTPTestEnvironment(map[string]string{ + "REPORTS_TOKEN": "reports-secret", + })) + if err != nil { + t.Fatalf("newUploadHTTPHandler() error = %v", err) + } + server := httptest.NewServer(handler) + defer server.Close() + + runID := submitHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("not a tar archive")) + record := waitForHTTPUploadStatus(t, server, runID, UploadStatusFailed) + + if record.Error == "" { + t.Fatal("failed status error is empty") + } + if record.Report != nil { + t.Fatalf("failed staging report = %#v, want nil", record.Report) + } + assertDirectoryEmpty(t, destination) +} + +func TestHTTPUploadSamePipelineRequestsSerialize(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + release := make(chan struct{}) + started := make(chan struct{}, 1) + coordinator := newUploadCoordinator(ctx, httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{ + id: "reports", + tokenEnv: "REPORTS_TOKEN", + stagingPath: filepath.Join(t.TempDir(), "reports"), + destinations: []string{t.TempDir()}, + }}, 4, 2), uploadCoordinatorHooks{ + randomSuffix: uploadTestSuffixes("00000001", "00000002"), + stage: successfulUploadStage, + run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) { + select { + case started <- struct{}{}: + default: + } + <-release + return RunReport{}, nil + }, + }) + handler := uploadHTTPHandler{ + coordinator: coordinator, + tokens: map[string]string{"reports-secret": "reports"}, + limits: map[string]int64{"reports": 1024}, + } + server := httptest.NewServer(handler) + defer server.Close() + + firstRunID := submitHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("first")) + waitForRunStart(t, started) + first := waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusRunning) + secondRunID := submitHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("second")) + second := waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusQueued) + + if first.PipelineID != "reports" || second.PipelineID != "reports" { + t.Fatalf("statuses = %#v %#v, want same pipeline", first, second) + } + if got := coordinator.RunningCount(); got != 1 { + t.Fatalf("running count = %d, want 1", got) + } + close(release) + waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusSucceeded) + waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusSucceeded) +} + +func TestHTTPUploadDifferentPipelinesRunConcurrently(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + release := make(chan struct{}) + started := make(chan string, 2) + coordinator := newUploadCoordinator(ctx, httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{ + { + id: "reports-one", + tokenEnv: "REPORTS_ONE_TOKEN", + stagingPath: filepath.Join(t.TempDir(), "reports-one"), + destinations: []string{t.TempDir()}, + }, + { + id: "reports-two", + tokenEnv: "REPORTS_TWO_TOKEN", + stagingPath: filepath.Join(t.TempDir(), "reports-two"), + destinations: []string{t.TempDir()}, + }, + }, 4, 2), uploadCoordinatorHooks{ + randomSuffix: uploadTestSuffixes("00000001", "00000002"), + stage: successfulUploadStage, + run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) { + started <- options.PipelineID + <-release + return RunReport{}, nil + }, + }) + handler := uploadHTTPHandler{ + coordinator: coordinator, + tokens: map[string]string{ + "one-secret": "reports-one", + "two-secret": "reports-two", + }, + limits: map[string]int64{ + "reports-one": 1024, + "reports-two": 1024, + }, + } + server := httptest.NewServer(handler) + defer server.Close() + + firstRunID := submitHTTPUpload(t, server, "one-secret", ingest.ContentTypeTar, []byte("first")) + secondRunID := submitHTTPUpload(t, server, "two-secret", ingest.ContentTypeTar, []byte("second")) + waitForStartedPipelines(t, started, "reports-one", "reports-two") + waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusRunning) + waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusRunning) + + if got := coordinator.RunningCount(); got != 2 { + t.Fatalf("running count = %d, want 2", got) + } + close(release) + waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusSucceeded) + waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusSucceeded) +} + +type httpUploadPipelineSpec struct { + id string + tokenEnv string + stagingPath string + destinations []string +} + +func httpUploadIntegrationConfig(t *testing.T, pipelines []httpUploadPipelineSpec, queueSize, maxConcurrency int) config.Config { + t.Helper() + size := config.ByteSize(1024 * 1024) + retention := config.Duration(time.Minute) + cfg := config.Config{ + Server: config.Server{HTTP: config.HTTPServer{ + Bind: config.DefaultHTTPBind, + StagingRoot: t.TempDir(), + MaxUploadSize: &size, + QueueSize: queueSize, + MaxConcurrency: maxConcurrency, + Retention: &retention, + }}, + } + for _, spec := range pipelines { + pipeline := config.Pipeline{ + ID: spec.id, + Source: config.Backend{ + Backend: config.BackendHTTPUpload, + Upload: config.HTTPUpload{ + TokenEnv: spec.tokenEnv, + StagingPath: spec.stagingPath, + MaxUploadSize: &size, + }, + }, + } + for index, destination := range spec.destinations { + pipeline.Destinations = append(pipeline.Destinations, config.Destination{ + ID: fmt.Sprintf("archive-%d", index+1), + Backend: config.BackendLocal, + Path: destination, + Publish: &config.PublishPolicy{Source: true}, + }) + } + cfg.Pipelines = append(cfg.Pipelines, pipeline) + } + config.ApplyDefaults(&cfg) + return cfg +} + +func submitHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) UploadRunID { + t.Helper() + request, err := http.NewRequest(http.MethodPost, server.URL+"/upload", bytes.NewReader(body)) + if err != nil { + t.Fatalf("NewRequest() error = %v", err) + } + request.Header.Set("Authorization", "Bearer "+token) + request.Header.Set("Content-Type", contentType) + response, err := server.Client().Do(request) + if err != nil { + t.Fatalf("POST /upload error = %v", err) + } + defer response.Body.Close() + if response.StatusCode != http.StatusAccepted { + t.Fatalf("POST /upload status = %d, want %d", response.StatusCode, http.StatusAccepted) + } + var accepted uploadAcceptedResponse + if err := json.NewDecoder(response.Body).Decode(&accepted); err != nil { + t.Fatalf("decode accepted response: %v", err) + } + if accepted.RunID == "" || accepted.Status != UploadStatusAccepted { + t.Fatalf("accepted response = %#v, want run id and accepted status", accepted) + } + return accepted.RunID +} + +func waitForHTTPUploadStatus(t *testing.T, server *httptest.Server, runID UploadRunID, status UploadStatus) UploadRunRecord { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + var latest UploadRunRecord + var latestStatus int + for time.Now().Before(deadline) { + latest, latestStatus = getHTTPUploadStatus(t, server, runID) + if latestStatus == http.StatusOK && latest.Status == status { + return latest + } + time.Sleep(time.Millisecond) + } + t.Fatalf("timed out waiting for status %s; latest HTTP status=%d record=%#v", status, latestStatus, latest) + return UploadRunRecord{} +} + +func getHTTPUploadStatus(t *testing.T, server *httptest.Server, runID UploadRunID) (UploadRunRecord, int) { + t.Helper() + response, err := server.Client().Get(server.URL + "/runs/" + string(runID)) + if err != nil { + t.Fatalf("GET /runs error = %v", err) + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + return UploadRunRecord{}, response.StatusCode + } + var record UploadRunRecord + if err := json.NewDecoder(response.Body).Decode(&record); err != nil { + t.Fatalf("decode run status: %v", err) + } + return record, response.StatusCode +} + +func bundleArchive(t *testing.T, compressed bool, opts testutil.BundleOptions) []byte { + t.Helper() + root := t.TempDir() + testutil.WriteSourceBundle(t, root, "", opts) + return tarDirectory(t, root, compressed) +} + +func tarDirectory(t *testing.T, root string, compressed bool) []byte { + t.Helper() + var output bytes.Buffer + var writer io.WriteCloser = nopWriteCloser{writer: &output} + if compressed { + gzipWriter := gzip.NewWriter(&output) + writer = gzipWriter + } + tarWriter := tar.NewWriter(writer) + if err := filepath.WalkDir(root, func(filePath string, entry fs.DirEntry, err error) error { + if err != nil { + return err + } + if entry.IsDir() { + return nil + } + relative, err := filepath.Rel(root, filePath) + if err != nil { + return err + } + data, err := os.ReadFile(filePath) + if err != nil { + return err + } + header := &tar.Header{ + Name: filepath.ToSlash(relative), + Mode: 0o600, + Size: int64(len(data)), + } + if err := tarWriter.WriteHeader(header); err != nil { + return err + } + if _, err := tarWriter.Write(data); err != nil { + return err + } + return nil + }); err != nil { + t.Fatalf("walk bundle: %v", err) + } + if err := tarWriter.Close(); err != nil { + t.Fatalf("close tar: %v", err) + } + if err := writer.Close(); err != nil { + t.Fatalf("close archive: %v", err) + } + return output.Bytes() +} + +type nopWriteCloser struct { + writer io.Writer +} + +func (writer nopWriteCloser) Write(data []byte) (int, error) { + return writer.writer.Write(data) +} + +func (writer nopWriteCloser) Close() error { + return nil +} + +func assertPublishedBundle(t *testing.T, destinationRoot string) { + t.Helper() + testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n") + testutil.AssertFile(t, filepath.Join(destinationRoot, "summary.txt"), "Summary\n") + if _, err := os.Stat(filepath.Join(destinationRoot, storage.StateFileName)); err != nil { + t.Fatalf("destination state stat: %v", err) + } +} + +func assertDirectoryEmpty(t *testing.T, root string) { + t.Helper() + entries, err := os.ReadDir(root) + if err != nil { + t.Fatalf("ReadDir() error = %v", err) + } + if len(entries) != 0 { + t.Fatalf("directory %s has %d entries, want empty", root, len(entries)) + } +} + +func waitForRunStart(t *testing.T, started <-chan struct{}) { + t.Helper() + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("timed out waiting for run start") + } +} + +func waitForStartedPipelines(t *testing.T, started <-chan string, want ...string) { + t.Helper() + remaining := map[string]bool{} + for _, pipelineID := range want { + remaining[pipelineID] = true + } + deadline := time.After(time.Second) + for len(remaining) > 0 { + select { + case pipelineID := <-started: + delete(remaining, pipelineID) + case <-deadline: + t.Fatalf("timed out waiting for pipelines to start; remaining=%v", remaining) + } + } +}