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

Add S3-compatible storage backend

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-31 17:11:53 +00:00
parent 052aa8a64a
commit 14fa9c8000
27 changed files with 1334 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
internal/config/config.go
View File

@@ -28,7 +28,7 @@ type Destination struct {
Bucket string `yaml:"bucket"`
Prefix string `yaml:"prefix"`
Region string `yaml:"region"`
ForcePath bool `yaml:"force_path_style"`
ForcePath *bool `yaml:"force_path_style"`
Creds Credentials `yaml:"credentials"`
SSH SSH `yaml:",inline"`
Publish *PublishPolicy `yaml:"publish"`
@@ -47,7 +47,7 @@ type Backend struct {
Bucket string `yaml:"bucket"`
Prefix string `yaml:"prefix"`
Region string `yaml:"region"`
ForcePath bool `yaml:"force_path_style"`
ForcePath *bool `yaml:"force_path_style"`
Creds Credentials `yaml:"credentials"`
SSH SSH `yaml:",inline"`
}

19
internal/config/defaults.go
View File

@@ -22,6 +22,8 @@ const (
TransformModeSidecar = "sidecar"
)
const DefaultS3Region = "us-east-1"
func ApplyDefaults(cfg *Config) {
for pipelineIndex := range cfg.Pipelines {
pipeline := &cfg.Pipelines[pipelineIndex]
@@ -60,6 +62,9 @@ func applyBackendDefaults(backend *Backend) {
backend.SSH.HostKeyPolicy = HostKeyPolicyAcceptNew
}
}
if backend.Backend == BackendS3 {
applyS3Defaults(&backend.Region, &backend.Prefix, &backend.ForcePath)
}
}
func applyDestinationDefaults(destination *Destination) {
@@ -71,4 +76,18 @@ func applyDestinationDefaults(destination *Destination) {
destination.SSH.HostKeyPolicy = HostKeyPolicyAcceptNew
}
}
if destination.Backend == BackendS3 {
applyS3Defaults(&destination.Region, &destination.Prefix, &destination.ForcePath)
}
}
func applyS3Defaults(region, prefix *string, forcePath **bool) {
if *region == "" {
*region = DefaultS3Region
}
*prefix = NormalizeS3Prefix(*prefix)
if *forcePath == nil {
defaultForcePath := true
*forcePath = &defaultForcePath
}
}

69
internal/config/load_test.go
View File

@@ -155,6 +155,61 @@ pipelines:
}
}
func TestLoadFileDefaultsS3Config(t *testing.T) {
cfg := loadConfig(t, `
pipelines:
- id: s3-defaults
source:
backend: s3
endpoint: http://127.0.0.1:9000
bucket: source
prefix: /incoming/reports/
destinations:
- id: archive
backend: s3
endpoint: http://127.0.0.1:9000
bucket: destination
`)
source := cfg.Pipelines[0].Source
if got, want := source.Region, DefaultS3Region; got != want {
t.Fatalf("source region = %q, want %q", got, want)
}
if got, want := source.Prefix, "incoming/reports"; got != want {
t.Fatalf("source prefix = %q, want %q", got, want)
}
if !ForcePathStyle(source.ForcePath) {
t.Fatal("source force_path_style = false, want true")
}
destination := cfg.Pipelines[0].Destinations[0]
if got, want := destination.Region, DefaultS3Region; got != want {
t.Fatalf("destination region = %q, want %q", got, want)
}
if !ForcePathStyle(destination.ForcePath) {
t.Fatal("destination force_path_style = false, want true")
}
}
func TestLoadFilePreservesExplicitS3ForcePathStyleFalse(t *testing.T) {
cfg := loadConfig(t, `
pipelines:
- id: s3-force-path
source:
backend: s3
endpoint: https://s3.example.com
bucket: source
force_path_style: false
destinations:
- id: archive
backend: local
path: /archive
`)
if ForcePathStyle(cfg.Pipelines[0].Source.ForcePath) {
t.Fatal("force_path_style = true, want explicit false")
}
}
func TestLoadFileRejectsDuplicatePipelineIDs(t *testing.T) {
assertLoadError(t, `
pipelines:
@@ -214,6 +269,19 @@ func TestLoadFileRejectsMissingRequiredFields(t *testing.T) {
}
}
func TestLoadFileRejectsInvalidS3Config(t *testing.T) {
tests := map[string]string{
"prefix traversal": `pipelines: [{id: reports, source: {backend: s3, endpoint: "https://s3.example.com", bucket: source, prefix: "../reports"}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
"prefix backslash": `pipelines: [{id: reports, source: {backend: s3, endpoint: "https://s3.example.com", bucket: source, prefix: 'a\b'}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
"partial creds": `pipelines: [{id: reports, source: {backend: s3, endpoint: "https://s3.example.com", bucket: source, credentials: {access_key_id_env: ACCESS_KEY_ID}}, 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:
@@ -398,6 +466,7 @@ func TestExampleConfigsLoad(t *testing.T) {
"../../examples/local-html.yml",
"../../examples/fan-out.yml",
"../../examples/ssh-destination.yml",
"../../examples/s3-destination.yml",
} {
t.Run(path, func(t *testing.T) {
if _, err := LoadFile(path); err != nil {

19
internal/config/s3.go Normal file
View File

@@ -0,0 +1,19 @@
package config
import (
"strings"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
)
func NormalizeS3Prefix(prefix string) string {
return strings.Trim(prefix, "/")
}
func ValidateS3Prefix(prefix string) error {
return storage.ValidatePrefix(prefix)
}
func ForcePathStyle(value *bool) bool {
return value == nil || *value
}

3
internal/config/secrets.go
View File

@@ -80,6 +80,9 @@ func (e Environment) required(name string) (string, error) {
if !ok {
return "", fmt.Errorf("credential environment variable %s is not set", name)
}
if value == "" {
return "", fmt.Errorf("credential environment variable %s is empty", name)
}
return value, nil
}

11
internal/config/secrets_test.go
View File

@@ -207,6 +207,17 @@ func TestResolveCredentialsMissingReferenceFailsWithoutSecretValue(t *testing.T)
}
}
func TestResolveCredentialsRejectsEmptyReferencedValue(t *testing.T) {
env := NewEnvironment(map[string]string{"EMPTY": ""}, emptyLookup)
_, err := env.ResolveCredentials(Credentials{AccessKeyIDEnv: "EMPTY"})
if err == nil {
t.Fatal("ResolveCredentials() error = nil, want error")
}
if !strings.Contains(err.Error(), "EMPTY") || !strings.Contains(err.Error(), "empty") {
t.Fatalf("ResolveCredentials() error = %q, want empty variable context", err)
}
}
func writeSecret(t *testing.T, directory, name, value string) {
t.Helper()
if err := os.WriteFile(filepath.Join(directory, name), []byte(value), 0o600); err != nil {

12
internal/config/validate.go
View File

@@ -69,14 +69,14 @@ func Validate(cfg Config) error {
}
func validateSourceBackend(errs ValidationErrors, context string, backend Backend) ValidationErrors {
return validateBackend(errs, context, backend.Backend, backend.Host, backend.Port, backend.Path, backend.URI, backend.Endpoint, backend.Bucket, backend.SSH.HostKeyPolicy)
return validateBackend(errs, context, backend.Backend, backend.Host, backend.Port, backend.Path, backend.URI, backend.Endpoint, backend.Bucket, backend.Prefix, backend.SSH.HostKeyPolicy, backend.Creds)
}
func validateDestinationBackend(errs ValidationErrors, context string, destination Destination) ValidationErrors {
return validateBackend(errs, context, destination.Backend, destination.Host, destination.Port, destination.Path, destination.URI, destination.Endpoint, destination.Bucket, destination.SSH.HostKeyPolicy)
return validateBackend(errs, context, destination.Backend, destination.Host, destination.Port, destination.Path, destination.URI, destination.Endpoint, destination.Bucket, destination.Prefix, destination.SSH.HostKeyPolicy, destination.Creds)
}
func validateBackend(errs ValidationErrors, context, backend, host string, port int, path, uri, endpoint, bucket string, hostKeyPolicy HostKeyPolicy) ValidationErrors {
func validateBackend(errs ValidationErrors, context, backend, host string, port int, path, uri, endpoint, bucket, prefix string, hostKeyPolicy HostKeyPolicy, creds Credentials) ValidationErrors {
switch backend {
case "":
errs = append(errs, context+".backend is required")
@@ -112,6 +112,12 @@ func validateBackend(errs ValidationErrors, context, backend, host string, port
if bucket == "" {
errs = append(errs, context+".bucket is required for s3 backend")
}
if err := ValidateS3Prefix(prefix); err != nil {
errs = append(errs, context+".prefix must be a clean relative slash-separated path")
}
if (creds.AccessKeyIDEnv == "") != (creds.SecretAccessKeyEnv == "") {
errs = append(errs, context+".credentials.access_key_id_env and credentials.secret_access_key_env must be configured together")
}
default:
errs = append(errs, context+".backend "+backend+" is unsupported")
}