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

Route upload client by pipeline

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-08 04:37:33 +00:00
parent 1c5d7198e3
commit ce43a6044a
4 changed files with 142 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
pkg/upload/client.go
View File

@@ -15,6 +15,7 @@ import (
"os"
"path"
"path/filepath"
"regexp"
"strings"
"time"
@@ -22,7 +23,6 @@ import (
)
const (
uploadPath = "upload"
runsPath = "runs"
idempotencyKeyHeader = "Idempotency-Key"
defaultHTTPTimeout = 30 * time.Second
@@ -34,6 +34,8 @@ const (
redactedSecret = "[redacted]"
)
var pipelineIDPattern = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9._-]*$`)
func NewClient(opts ClientOptions) (*Client, error) {
endpoint, err := cleanEndpoint(opts.Endpoint)
if err != nil {
@@ -65,6 +67,9 @@ func (c *Client) UploadBundle(ctx context.Context, opts UploadBundleOptions) (Re
if opts.Validate && opts.DisableValidation {
return Result{}, fmt.Errorf("validate and disable validation cannot both be set")
}
if err := validatePipelineID(opts.PipelineID); err != nil {
return Result{}, err
}
if opts.Root == "" {
return Result{}, fmt.Errorf("root is required")
}
@@ -85,7 +90,7 @@ func (c *Client) UploadBundle(ctx context.Context, opts UploadBundleOptions) (Re
if err != nil {
return Result{}, err
}
return c.uploadArchive(ctx, archive, key)
return c.uploadArchive(ctx, opts.PipelineID, archive, key)
}
func (c *Client) UploadFiles(ctx context.Context, opts UploadFilesOptions) (Result, error) {
@@ -95,6 +100,9 @@ func (c *Client) UploadFiles(ctx context.Context, opts UploadFilesOptions) (Resu
if opts.Validate && opts.DisableValidation {
return Result{}, fmt.Errorf("validate and disable validation cannot both be set")
}
if err := validatePipelineID(opts.PipelineID); err != nil {
return Result{}, err
}
if opts.ID == "" {
return Result{}, fmt.Errorf("id is required")
}
@@ -131,7 +139,7 @@ func (c *Client) UploadFiles(ctx context.Context, opts UploadFilesOptions) (Resu
if err != nil {
return Result{}, err
}
return c.uploadArchive(ctx, archive, key)
return c.uploadArchive(ctx, opts.PipelineID, archive, key)
}
func (c *Client) Status(ctx context.Context, runID string) (RunStatus, error) {
@@ -167,7 +175,7 @@ func (c *Client) Status(ctx context.Context, runID string) (RunStatus, error) {
return status, nil
}
func (c *Client) uploadArchive(ctx context.Context, archive []byte, idempotencyKey string) (Result, error) {
func (c *Client) uploadArchive(ctx context.Context, pipelineID string, archive []byte, idempotencyKey string) (Result, error) {
if ctx == nil {
ctx = context.Background()
}
@@ -176,7 +184,7 @@ func (c *Client) uploadArchive(ctx context.Context, archive []byte, idempotencyK
if err := ctx.Err(); err != nil {
return Result{}, err
}
result, retry, err := c.uploadAttempt(ctx, archive, idempotencyKey)
result, retry, err := c.uploadAttempt(ctx, pipelineID, archive, idempotencyKey)
if err == nil {
return result, nil
}
@@ -191,8 +199,8 @@ func (c *Client) uploadArchive(ctx context.Context, archive []byte, idempotencyK
return Result{}, lastErr
}
func (c *Client) uploadAttempt(ctx context.Context, archive []byte, idempotencyKey string) (Result, bool, error) {
request, err := http.NewRequestWithContext(ctx, http.MethodPost, c.uploadURL(), bytes.NewReader(archive))
func (c *Client) uploadAttempt(ctx context.Context, pipelineID string, archive []byte, idempotencyKey string) (Result, bool, error) {
request, err := http.NewRequestWithContext(ctx, http.MethodPost, c.uploadURL(pipelineID), bytes.NewReader(archive))
if err != nil {
return Result{}, false, c.redactError(err)
}
@@ -227,8 +235,8 @@ func (c *Client) authorize(request *http.Request) {
request.Header.Set("Authorization", authorizationPrefix+c.token)
}
func (c *Client) uploadURL() string {
return joinEndpointPath(c.endpoint, uploadPath)
func (c *Client) uploadURL(pipelineID string) string {
return joinEndpointPath(c.endpoint, "v1", "pipelines", pipelineID, "upload")
}
func (c *Client) statusURL(runID string) string {
@@ -343,6 +351,16 @@ func uploadIdempotencyKey(value string) (string, error) {
return value, nil
}
func validatePipelineID(value string) error {
if value == "" {
return fmt.Errorf("pipeline id is required")
}
if !pipelineIDPattern.MatchString(value) {
return fmt.Errorf("pipeline id must be a slug-like identifier")
}
return nil
}
func validateIdempotencyKey(value string) error {
if value == "" {
return fmt.Errorf("idempotency key is required")

102
pkg/upload/client_test.go
View File

@@ -47,7 +47,7 @@ func TestNewClientValidatesOptions(t *testing.T) {
if err != nil {
t.Fatalf("NewClient() error = %v", err)
}
if got, want := client.uploadURL(), "http://127.0.0.1:8080/base/upload"; got != want {
if got, want := client.uploadURL("reports.daily"), "http://127.0.0.1:8080/base/v1/pipelines/reports.daily/upload"; got != want {
t.Fatalf("upload URL = %q, want %q", got, want)
}
if client.httpClient == nil || client.httpClient.Timeout == 0 {
@@ -65,7 +65,7 @@ func TestUploadBundleSendsCallerKeyAndManifestArchive(t *testing.T) {
}
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if got, want := r.URL.Path, "/upload"; got != want {
if got, want := r.URL.Path, "/v1/pipelines/reports.daily/upload"; got != want {
t.Fatalf("path = %q, want %q", got, want)
}
if got, want := r.Header.Get("Authorization"), "Bearer secret-token"; got != want {
@@ -93,6 +93,7 @@ func TestUploadBundleSendsCallerKeyAndManifestArchive(t *testing.T) {
t.Fatalf("NewClient() error = %v", err)
}
result, err := client.UploadBundle(context.Background(), UploadBundleOptions{
PipelineID: "reports.daily",
Root: root,
IdempotencyKey: "producer.retry:one",
})
@@ -112,6 +113,9 @@ func TestUploadFilesBuildsTemporaryBundleWithoutTouchingSources(t *testing.T) {
}
tempDir := t.TempDir()
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if got, want := r.URL.Path, "/v1/pipelines/reports.files/upload"; got != want {
t.Fatalf("path = %q, want %q", got, want)
}
entries := readArchiveEntries(t, r.Body)
if got := string(entries["manifest.json"]); !strings.Contains(got, `"id": "reports.from.files"`) {
t.Fatalf("manifest = %s, want uploaded id", got)
@@ -131,6 +135,7 @@ func TestUploadFilesBuildsTemporaryBundleWithoutTouchingSources(t *testing.T) {
t.Fatalf("NewClient() error = %v", err)
}
_, err = client.UploadFiles(context.Background(), UploadFilesOptions{
PipelineID: "reports.files",
ID: "reports.from.files",
Files: []sourcebundle.BundleFile{{
SourcePath: sourcePath,
@@ -153,6 +158,77 @@ func TestUploadFilesBuildsTemporaryBundleWithoutTouchingSources(t *testing.T) {
}
}
func TestUploadMethodsRequirePipelineIDBeforeLocalWork(t *testing.T) {
var requests atomic.Int64
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
requests.Add(1)
t.Fatal("server should not receive request")
}))
defer server.Close()
client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret", HTTPClient: server.Client()})
if err != nil {
t.Fatalf("NewClient() error = %v", err)
}
missingRoot := filepath.Join(t.TempDir(), "missing")
if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: missingRoot}); err == nil || !strings.Contains(err.Error(), "pipeline id is required") {
t.Fatalf("UploadBundle() error = %v, want missing pipeline id", err)
}
tempDir := t.TempDir()
sourcePath := filepath.Join(t.TempDir(), "report.md")
if err := os.WriteFile(sourcePath, []byte("data"), 0o600); err != nil {
t.Fatalf("write source: %v", err)
}
if _, err := client.UploadFiles(context.Background(), UploadFilesOptions{
ID: "reports.from.files",
Files: []sourcebundle.BundleFile{{
SourcePath: sourcePath,
Path: "report.md",
}},
TempDir: tempDir,
}); err == nil || !strings.Contains(err.Error(), "pipeline id is required") {
t.Fatalf("UploadFiles() error = %v, want missing pipeline id", err)
}
entries, err := os.ReadDir(tempDir)
if err != nil {
t.Fatalf("read temp dir: %v", err)
}
if len(entries) != 0 {
t.Fatalf("temp dir entries = %d, want no local bundle work", len(entries))
}
if got := requests.Load(); got != 0 {
t.Fatalf("requests = %d, want 0", got)
}
}
func TestUploadMethodsRejectInvalidPipelineIDBeforeHTTPRequest(t *testing.T) {
root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "data"}})
var requests atomic.Int64
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
requests.Add(1)
t.Fatal("server should not receive request")
}))
defer server.Close()
client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret", HTTPClient: server.Client()})
if err != nil {
t.Fatalf("NewClient() error = %v", err)
}
for _, pipelineID := range []string{".reports", "reports/daily", "reports daily"} {
t.Run(pipelineID, func(t *testing.T) {
_, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: pipelineID, Root: root})
if err == nil || !strings.Contains(err.Error(), "pipeline id must be a slug-like identifier") {
t.Fatalf("UploadBundle() error = %v, want invalid pipeline id", err)
}
})
}
if got := requests.Load(); got != 0 {
t.Fatalf("requests = %d, want 0", got)
}
}
func TestUploadBundleValidationFailurePreventsHTTPRequest(t *testing.T) {
root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "original"}})
if err := os.WriteFile(filepath.Join(root, "report.md"), []byte("changed"), 0o600); err != nil {
@@ -169,7 +245,7 @@ func TestUploadBundleValidationFailurePreventsHTTPRequest(t *testing.T) {
if err != nil {
t.Fatalf("NewClient() error = %v", err)
}
if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root}); err == nil {
if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root}); err == nil {
t.Fatal("UploadBundle() error = nil, want validation error")
}
if got := requests.Load(); got != 0 {
@@ -193,7 +269,7 @@ func TestUploadBundleCanDisableLocalValidation(t *testing.T) {
if err != nil {
t.Fatalf("NewClient() error = %v", err)
}
if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, DisableValidation: true}); err != nil {
if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, DisableValidation: true}); err != nil {
t.Fatalf("UploadBundle() error = %v", err)
}
if got := requests.Load(); got != 1 {
@@ -206,6 +282,9 @@ func TestGeneratedIdempotencyKeyIsReusedAcrossRetry(t *testing.T) {
var attempts atomic.Int64
var keys []string
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if got, want := r.URL.Path, "/v1/pipelines/reports/upload"; got != want {
t.Fatalf("path = %q, want %q", got, want)
}
keys = append(keys, r.Header.Get(idempotencyKeyHeader))
if attempts.Add(1) == 1 {
writeJSONError(w, http.StatusServiceUnavailable, "busy", false)
@@ -224,7 +303,7 @@ func TestGeneratedIdempotencyKeyIsReusedAcrossRetry(t *testing.T) {
if err != nil {
t.Fatalf("NewClient() error = %v", err)
}
if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root}); err != nil {
if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root}); err != nil {
t.Fatalf("UploadBundle() error = %v", err)
}
if got, want := attempts.Load(), int64(2); got != want {
@@ -276,7 +355,7 @@ func TestUploadResponseParsingAndNoRetryStatuses(t *testing.T) {
if err != nil {
t.Fatalf("NewClient() error = %v", err)
}
_, err = client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "key"})
_, err = client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "key"})
if err == nil {
t.Fatal("UploadBundle() error = nil, want error")
}
@@ -309,7 +388,7 @@ func TestTokenRedactedFromHTTPError(t *testing.T) {
if err != nil {
t.Fatalf("NewClient() error = %v", err)
}
_, err = client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "key"})
_, err = client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "key"})
if err == nil {
t.Fatal("UploadBundle() error = nil, want error")
}
@@ -330,6 +409,9 @@ func TestNetworkRetryUsesSameIdempotencyKey(t *testing.T) {
Token: "secret",
HTTPClient: &http.Client{Transport: roundTripFunc(func(request *http.Request) (*http.Response, error) {
keys = append(keys, request.Header.Get(idempotencyKeyHeader))
if got, want := request.URL.Path, "/v1/pipelines/reports/upload"; got != want {
t.Fatalf("path = %q, want %q", got, want)
}
if attempts.Add(1) == 1 {
return nil, temporaryNetworkError{}
}
@@ -346,7 +428,7 @@ func TestNetworkRetryUsesSameIdempotencyKey(t *testing.T) {
if err != nil {
t.Fatalf("NewClient() error = %v", err)
}
result, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "network-retry"})
result, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "network-retry"})
if err != nil {
t.Fatalf("UploadBundle() error = %v", err)
}
@@ -381,7 +463,7 @@ func TestContextCancellationDuringRetryBackoff(t *testing.T) {
if err != nil {
t.Fatalf("NewClient() error = %v", err)
}
_, err = client.UploadBundle(ctx, UploadBundleOptions{Root: root, IdempotencyKey: "cancel"})
_, err = client.UploadBundle(ctx, UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "cancel"})
if !errors.Is(err, context.Canceled) {
t.Fatalf("UploadBundle() error = %v, want context.Canceled", err)
}
@@ -446,7 +528,7 @@ func TestInvalidCallerIdempotencyKeyPreventsHTTPRequest(t *testing.T) {
if err != nil {
t.Fatalf("NewClient() error = %v", err)
}
if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "bad key"}); err == nil {
if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "bad key"}); err == nil {
t.Fatal("UploadBundle() error = nil, want invalid key error")
}
if got := requests.Load(); got != 0 {

29
pkg/upload/doc.go
View File

@@ -12,9 +12,10 @@
//
// NewClient creates a Client from ClientOptions. Endpoint is required and must
// be an http or https distributor server base URL without userinfo, query, or
// fragment. The client derives /upload for submissions and /runs/<run-id> for
// status checks. Token is required and is sent as Authorization: Bearer <token>.
// Token values are redacted from errors produced by the client.
// fragment. The client derives /v1/pipelines/<pipeline-id>/upload for
// submissions and /runs/<run-id> for status checks. Token is required and is
// sent as Authorization: Bearer <token>. Token values are redacted from errors
// produced by the client.
//
// HTTPClient is optional. When omitted, the package uses a client with a
// conservative timeout. Retry is optional; zero values select safe defaults.
@@ -23,20 +24,28 @@
//
// # Upload Workflows
//
// UploadBundle uploads an existing local source bundle root. The root must
// contain manifest.json. By default, UploadBundle loads the manifest and
// validates the complete local bundle with pkg/bundle before making any HTTP
// request. The generated gzip-compressed tar archive contains manifest.json and
// exactly the manifest-listed files; unlisted files are not uploaded.
// UploadBundle uploads an existing local source bundle root to the configured
// PipelineID. PipelineID is required and must match the server's slug-like
// pipeline id syntax. The root must contain manifest.json. By default,
// UploadBundle loads the manifest and validates the complete local bundle with
// pkg/bundle before making any HTTP request. The generated gzip-compressed tar
// archive contains manifest.json and exactly the manifest-listed files;
// unlisted files are not uploaded.
//
// UploadFiles is the convenience workflow for producer applications that have
// generated files but have not yet assembled a bundle directory. It uses
// generated files but have not yet assembled a bundle directory. PipelineID is
// required and selects the configured distributor workflow. UploadFiles uses
// pkg/bundle to create a temporary complete bundle from explicit
// bundle.BundleFile values, validates it by default, archives it, uploads it,
// and removes temporary files when the call returns. UploadFiles does not write
// into producer source directories. A zero Created timestamp follows
// pkg/bundle defaulting behavior.
//
// The producer contract has four separate identifiers: the bearer token
// authenticates the client, PipelineID selects the distributor workflow, the
// source manifest ID identifies the logical artifact within that workflow, and
// IdempotencyKey identifies one producer run and retry group.
//
// Validation is enabled by default. Set DisableValidation when the application
// has already performed equivalent local validation and wants to skip the
// package's validation step. Validate and DisableValidation must not both be
@@ -91,6 +100,7 @@
// }
//
// result, err := client.UploadFiles(ctx, upload.UploadFilesOptions{
// PipelineID: "reports.daily",
// ID: "reports.daily.2026-06-06",
// IdempotencyKey: "reports.daily.2026-06-06",
// Files: []bundle.BundleFile{
@@ -115,6 +125,7 @@
// Example: upload an existing bundle root.
//
// result, err := client.UploadBundle(ctx, upload.UploadBundleOptions{
// PipelineID: "reports.daily",
// Root: "/var/lib/reports/daily-2026-06-06",
// IdempotencyKey: "reports.daily.2026-06-06",
// })

2
pkg/upload/types.go
View File

@@ -30,6 +30,7 @@ type RetryOptions struct {
}
type UploadBundleOptions struct {
PipelineID string
Root string
Validate bool
DisableValidation bool
@@ -37,6 +38,7 @@ type UploadBundleOptions struct {
}
type UploadFilesOptions struct {
PipelineID string
ID string
Created time.Time
Files []bundle.BundleFile