LOGC-6: Implement S3 uploader

.github/workflows/tests.yaml

@@ -4,6 +4,10 @@ name: tests
 on:
   push:
 
+permissions:
+  contents: read
+  packages: read
+
 jobs:
   build-test:
     name: Build and test
@@ -21,6 +25,19 @@ jobs:
         with:
           go-version: '1.24'
 
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Login to Registry
+        uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.repository_owner }}
+          password: ${{ github.token }}
+
+      - name: Start Workbench
+        uses: scality/[email protected]
+
       - name: Start ClickHouse
         run: docker compose up -d
 
@@ -52,6 +69,12 @@ jobs:
           name: log-courier
           path: ./bin/log-courier
 
+      - name: Stop Workbench
+        if: always()
+        run: |
+          workbench logs
+          workbench down
+
       - name: Stop ClickHouse
         if: always()
         run: |

docker-compose.yml

@@ -5,7 +5,7 @@ services:
     image: docker.io/clickhouse/clickhouse-server:24.3.2.23-alpine
     container_name: log-courier-clickhouse
     ports:
-      - "9000:9000"  # Native protocol
+      - "9002:9000"  # Native protocol
       - "8123:8123"  # HTTP interface
     environment:
       CLICKHOUSE_DB: logs

env/default/.gitignore

@@ -0,0 +1,4 @@
+# Generated by workbench
+/*
+!values.yaml
+!.gitignore

env/default/values.yaml

@@ -0,0 +1,17 @@
+global:
+  log_level: info
+
+features:
+  scuba:
+    enabled: false
+    enable_service_user: false
+
+
+cloudserver:
+  image: ghcr.io/scality/cloudserver:7.70.77
+
+vault:
+  image: ghcr.io/scality/vault:7.81.0
+
+s3_metadata:
+  image: ghcr.io/scality/metadata:8.19.0-standalone

go.mod

@@ -14,6 +14,24 @@ require (
 	github.com/ClickHouse/clickhouse-go/v2 v2.40.3 // indirect
 	github.com/Masterminds/semver/v3 v3.4.0 // indirect
 	github.com/andybalholm/brotli v1.2.0 // indirect
+	github.com/aws/aws-sdk-go-v2 v1.39.4 // indirect
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.2 // indirect
+	github.com/aws/aws-sdk-go-v2/config v1.31.15 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.18.19 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.11 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.11 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.11 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.11 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.2 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.2 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.11 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.11 // indirect
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.89.0 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.29.8 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.3 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.38.9 // indirect
+	github.com/aws/smithy-go v1.23.1 // indirect
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/go-faster/city v1.0.1 // indirect
 	github.com/go-faster/errors v0.7.1 // indirect

pkg/logcourier/config.go

@@ -29,5 +29,15 @@
 		return fmt.Errorf("consumer.time-threshold-seconds must be positive, got %d", timeThreshold)
 	}
 
+	maxRetryAttempts := ConfigSpec.GetInt("s3.max-retry-attempts")
+	if maxRetryAttempts <= 0 {
+		return fmt.Errorf("s3.max-retry-attempts must be positive, got %d", maxRetryAttempts)
+	}
+
+	maxBackoffDelay := ConfigSpec.GetInt("s3.max-backoff-delay-seconds")
+	if maxBackoffDelay <= 0 {
+		return fmt.Errorf("s3.max-backoff-delay-seconds must be positive, got %d", maxBackoffDelay)
+	}
+
 	return nil
 }

pkg/logcourier/configspec.go

@@ -9,7 +9,7 @@ var ConfigSpec = util.ConfigSpec{
 	// ClickHouse connection
 	"clickhouse.url": util.ConfigVarSpec{
 		Help:         "ClickHouse connection URL",
-		DefaultValue: "localhost:9000",
+		DefaultValue: "localhost:9002",
 		EnvVar:       "LOG_COURIER_CLICKHOUSE_URL",
 	},
 	"clickhouse.username": util.ConfigVarSpec{
@@ -40,6 +40,33 @@ var ConfigSpec = util.ConfigSpec{
 		EnvVar:       "LOG_COURIER_CONSUMER_TIME_THRESHOLD_SECONDS",
 	},
 
+	// S3 configuration
+	"s3.endpoint": util.ConfigVarSpec{
+		Help:         "S3 endpoint URL",
+		DefaultValue: "127.0.0.1:8000",
+		EnvVar:       "S3_ENDPOINT",
+	},
+	"s3.access-key-id": util.ConfigVarSpec{
+		Help:         "S3 access key ID",
+		DefaultValue: "",
+		EnvVar:       "S3_ACCESS_KEY_ID",
+	},
+	"s3.secret-access-key": util.ConfigVarSpec{
+		Help:         "S3 secret access key",
+		DefaultValue: "",
+		EnvVar:       "S3_SECRET_ACCESS_KEY",
+	},
+	"s3.max-retry-attempts": util.ConfigVarSpec{
+		Help:         "Maximum number of retry attempts for S3 operations (including initial request)",
+		DefaultValue: 3,
+		EnvVar:       "S3_MAX_RETRY_ATTEMPTS",
+	},
+	"s3.max-backoff-delay-seconds": util.ConfigVarSpec{
+		Help:         "Maximum backoff delay in seconds between S3 retry attempts",
+		DefaultValue: 20,
+		EnvVar:       "S3_MAX_BACKOFF_DELAY_SECONDS",
+	},
+
 	// General
 	"log-level": util.ConfigVarSpec{
 		Help:         "Log level (error|warn|info|debug)",

pkg/s3/client.go

@@ -0,0 +1,87 @@
+package s3
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/aws/retry"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+)
+
+const (
+	defaultRegion = "us-east-1"
+)
+
+// Client wraps S3 client
+type Client struct {
+	s3Client *s3.Client
+}
+
+// Config holds S3 client configuration
+type Config struct {
+	Endpoint        string
+	AccessKeyID     string
+	SecretAccessKey string
+	MaxRetryAttempts int
+	MaxBackoffDelay time.Duration
+}
+
+// NewClient creates a new S3 client
+func NewClient(ctx context.Context, cfg Config) (*Client, error) {
+	if cfg.AccessKeyID == "" || cfg.SecretAccessKey == "" {
+		return nil, fmt.Errorf("access key ID and secret access key are required")
+	}
+
+	var optFns []func(*config.LoadOptions) error
+
+	// Set region and credentials
+	optFns = append(optFns,
+		config.WithRegion(defaultRegion),
+		config.WithCredentialsProvider(
+			credentials.NewStaticCredentialsProvider(
+				cfg.AccessKeyID,
+				cfg.SecretAccessKey,
+				"",
+			),
+		),
+	)
+
+	// Set retry configuration if non-zero values provided
+	if cfg.MaxRetryAttempts > 0 || cfg.MaxBackoffDelay > 0 {
+		optFns = append(optFns, config.WithRetryer(func() aws.Retryer {
+			retryer := retry.NewStandard()
+			var result aws.Retryer = retryer
+			if cfg.MaxRetryAttempts > 0 {
+				result = retry.AddWithMaxAttempts(result, cfg.MaxRetryAttempts)
+			}
+			if cfg.MaxBackoffDelay > 0 {
+				result = retry.AddWithMaxBackoffDelay(result, cfg.MaxBackoffDelay)
+			}
+			return result
+		}))
+	}
+
+	awsCfg, err := config.LoadDefaultConfig(ctx, optFns...)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load AWS config: %w", err)
+	}
+
+	// Create S3 client
+	s3ClientOpts := []func(*s3.Options){}
+
+	// Set endpoint
+	if cfg.Endpoint != "" {
+		s3ClientOpts = append(s3ClientOpts, func(o *s3.Options) {
+			o.BaseEndpoint = aws.String(cfg.Endpoint)
+			o.UsePathStyle = true // Required for non-AWS S3-compatible services
+		})
+	}
+
+	s3Client := s3.NewFromConfig(awsCfg, s3ClientOpts...)
+
+	return &Client{s3Client: s3Client}, nil
+}

pkg/s3/uploader.go

@@ -0,0 +1,37 @@
+package s3
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+)
+
+// Uploader uploads log objects to S3
+type Uploader struct {
+	client *Client
+}
+
+// NewUploader creates a new uploader
+func NewUploader(client *Client) *Uploader {
+	return &Uploader{client: client}
+}
+
+// Upload uploads a log object to the specified bucket.
+// Retries are handled automatically by the SDK client based on its retry configuration.
+func (u *Uploader) Upload(ctx context.Context, bucket, key string, content []byte) error {
+	_, err := u.client.s3Client.PutObject(ctx, &s3.PutObjectInput{
+		Bucket:      aws.String(bucket),
+		Key:         aws.String(key),
+		Body:        bytes.NewReader(content),
+		ContentType: aws.String("text/plain"),
+	})
+
+	if err != nil {
+		return fmt.Errorf("failed to upload to S3: bucket=%s, key=%s: %w", bucket, key, err)
+	}
+
+	return nil
+}