Add HTTP upload configuration support

internal/config/load_test.go

@@ -208,6 +208,123 @@ pipelines:
 	}
 }
 
+func TestLoadFileDefaultsHTTPServerConfig(t *testing.T) {
+	cfg := loadConfig(t, `
+pipelines:
+  - id: reports
+    source:
+      backend: local
+      path: /source
+    destinations:
+      - id: archive
+        backend: local
+        path: /archive
+`)
+
+	server := cfg.Server.HTTP
+	if got, want := server.Bind, DefaultHTTPBind; got != want {
+		t.Fatalf("server.http.bind = %q, want %q", got, want)
+	}
+	if got, want := server.StagingRoot, DefaultHTTPStagingRoot; got != want {
+		t.Fatalf("server.http.staging_root = %q, want %q", got, want)
+	}
+	if got, want := *server.MaxUploadSize, DefaultHTTPMaxUploadSize; got != want {
+		t.Fatalf("server.http.max_upload_size = %s, want %s", got, want)
+	}
+	if got, want := server.QueueSize, DefaultHTTPQueueSize; got != want {
+		t.Fatalf("server.http.queue_size = %d, want %d", got, want)
+	}
+	if got, want := server.MaxConcurrency, DefaultHTTPMaxConcurrency; got != want {
+		t.Fatalf("server.http.max_concurrency = %d, want %d", got, want)
+	}
+	if got, want := *server.Retention, DefaultHTTPRetention; got != want {
+		t.Fatalf("server.http.retention = %s, want %s", got, want)
+	}
+}
+
+func TestLoadFileAcceptsHTTPUploadSourceConfig(t *testing.T) {
+	cfg := loadConfig(t, `
+server:
+  http:
+    bind: 127.0.0.1:9090
+    staging_root: /srv/distributor/staging
+    max_upload_size: 64MB
+    queue_size: 32
+    max_concurrency: 2
+    retention: 48h
+pipelines:
+  - id: weather-daily
+    source:
+      backend: http_upload
+      token_env: WEATHER_DAILY_UPLOAD_TOKEN
+      staging_path: /srv/distributor/staging/weather-daily
+      max_upload_size: 32MB
+    destinations:
+      - id: archive
+        backend: local
+        path: /archive
+`)
+
+	server := cfg.Server.HTTP
+	if got, want := server.Bind, "127.0.0.1:9090"; got != want {
+		t.Fatalf("server.http.bind = %q, want %q", got, want)
+	}
+	if got, want := server.StagingRoot, "/srv/distributor/staging"; got != want {
+		t.Fatalf("server.http.staging_root = %q, want %q", got, want)
+	}
+	if got, want := *server.MaxUploadSize, ByteSize(64*1024*1024); got != want {
+		t.Fatalf("server.http.max_upload_size = %s, want %s", got, want)
+	}
+	if got, want := server.QueueSize, 32; got != want {
+		t.Fatalf("server.http.queue_size = %d, want %d", got, want)
+	}
+	if got, want := server.MaxConcurrency, 2; got != want {
+		t.Fatalf("server.http.max_concurrency = %d, want %d", got, want)
+	}
+	if got, want := server.Retention.String(), "48h0m0s"; got != want {
+		t.Fatalf("server.http.retention = %s, want %s", got, want)
+	}
+
+	source := cfg.Pipelines[0].Source
+	if got, want := source.Backend, BackendHTTPUpload; got != want {
+		t.Fatalf("source.backend = %q, want %q", got, want)
+	}
+	if got, want := source.Upload.TokenEnv, "WEATHER_DAILY_UPLOAD_TOKEN"; got != want {
+		t.Fatalf("source.token_env = %q, want %q", got, want)
+	}
+	if got, want := source.Upload.StagingPath, "/srv/distributor/staging/weather-daily"; got != want {
+		t.Fatalf("source.staging_path = %q, want %q", got, want)
+	}
+	if got, want := *source.Upload.MaxUploadSize, ByteSize(32*1024*1024); got != want {
+		t.Fatalf("source.max_upload_size = %s, want %s", got, want)
+	}
+}
+
+func TestLoadFileDefaultsHTTPUploadSourceConfig(t *testing.T) {
+	cfg := loadConfig(t, `
+server:
+  http:
+    max_upload_size: 12MB
+pipelines:
+  - id: weather-daily
+    source:
+      backend: http_upload
+      token_env: WEATHER_DAILY_UPLOAD_TOKEN
+    destinations:
+      - id: archive
+        backend: local
+        path: /archive
+`)
+
+	source := cfg.Pipelines[0].Source
+	if got, want := source.Upload.StagingPath, "/var/spool/distributor/weather-daily"; got != want {
+		t.Fatalf("source.staging_path = %q, want %q", got, want)
+	}
+	if got, want := *source.Upload.MaxUploadSize, ByteSize(12*1024*1024); got != want {
+		t.Fatalf("source.max_upload_size = %s, want %s", got, want)
+	}
+}
+
 func TestLoadFileValidBackendConfigs(t *testing.T) {
 	tests := map[string]string{
 		"local": `
@@ -396,6 +513,26 @@ func TestLoadFileRejectsInvalidS3Config(t *testing.T) {
 	}
 }
 
+func TestLoadFileRejectsInvalidHTTPUploadConfig(t *testing.T) {
+	tests := map[string]string{
+		"server size":             `server: {http: {max_upload_size: 20XB}}`,
+		"source size":             `pipelines: [{id: reports, source: {backend: http_upload, token_env: UPLOAD_TOKEN, max_upload_size: 20XB}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+		"zero source size":        `pipelines: [{id: reports, source: {backend: http_upload, token_env: UPLOAD_TOKEN, max_upload_size: 0B}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+		"server duration":         `server: {http: {retention: forever}}`,
+		"zero server duration":    `server: {http: {retention: 0s}}`,
+		"missing token env":       `pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+		"destination http upload": `pipelines: [{id: reports, source: {backend: local, path: /source}, destinations: [{id: ingest, backend: http_upload}]}]`,
+		"literal token":           `pipelines: [{id: reports, source: {backend: http_upload, token: secret, token_env: UPLOAD_TOKEN}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+		"unknown server field":    `server: {http: {surprise: true}}`,
+		"unknown source field":    `pipelines: [{id: reports, source: {backend: http_upload, token_env: UPLOAD_TOKEN, surprise: true}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+	}
+	for name, body := range tests {
+		t.Run(name, func(t *testing.T) {
+			assertLoadError(t, body, "")
+		})
+	}
+}
+
 func TestLoadFileDefaultsSSHConfig(t *testing.T) {
 	cfg := loadConfig(t, `
 pipelines: