chore: added s3 flag

constants/constants.go

@@ -22,6 +22,13 @@ const (
 	StringifiedData        = "data"
 	DefaultReadPreference  = "secondaryPreferred"
 	EncryptionKey          = "OLAKE_ENCRYPTION_KEY"
+	EnvStorageMode         = "OLAKE_STORAGE_MODE"
+	StorageModeS3          = "s3"
+	EnvS3Region            = "OLAKE_S3_REGION"
+	EnvS3AccessKeyID       = "OLAKE_S3_ACCESS_KEY_ID"
+	EnvS3SecretAccessKey   = "OLAKE_S3_SECRET_ACCESS_KEY" // #nosec G101 -- env var name, not a credential
+	EnvS3SessionToken      = "OLAKE_S3_SESSION_TOKEN"     // #nosec G101 -- env var name, not a credential
+	EnvS3Endpoint          = "OLAKE_S3_ENDPOINT"
 	ConfigFolder           = "CONFIG_FOLDER"
 	StatePath              = "STATE_PATH"
 	StreamsPath            = "STREAMS_PATH"

go.mod

@@ -4,6 +4,7 @@ go 1.25.13
 
 require (
 	github.com/aws/aws-sdk-go-v2/config v1.29.17
+	github.com/aws/aws-sdk-go-v2/credentials v1.17.70
 	github.com/aws/aws-sdk-go-v2/service/kms v1.41.1
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.97.3
 	github.com/cockroachdb/pebble/v2 v2.1.6

protocol/clear.go

@@ -41,7 +41,9 @@ var clearCmd = &cobra.Command{
 		}
 		return nil
 	},
-	RunE: func(cmd *cobra.Command, _ []string) error {
+	RunE: func(cmd *cobra.Command, _ []string) (err error) {
+		defer utils.FinalizeS3Upload(cmd.Context(), &err, noSave)
+
 		selectedStreamsMetadata, err := classifyStreams(catalog, nil, state)
 		if err != nil {
 			return fmt.Errorf("failed to get selected streams for clearing: %w", err)

protocol/discover.go

@@ -44,12 +44,14 @@ var discoverCmd = &cobra.Command{
 
 		return nil
 	},
-	RunE: func(cmd *cobra.Command, _ []string) error {
+	RunE: func(cmd *cobra.Command, _ []string) (err error) {
+		defer utils.FinalizeS3Upload(cmd.Context(), &err, noSave)
+
 		if streamsPath != "" && differencePath != "" {
 			return compareStreams()
 		}
 
-		err := connector.Setup(cmd.Context())
+		err = connector.Setup(cmd.Context())
 		if err != nil {
 			return err
 		}

protocol/root.go

@@ -14,6 +14,7 @@ import (
 	"github.com/datazip-inc/olake/utils"
 	"github.com/datazip-inc/olake/utils/errs"
 	"github.com/datazip-inc/olake/utils/logger"
+	"github.com/datazip-inc/olake/utils/s3"
 	"github.com/datazip-inc/olake/utils/telemetry"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
@@ -44,14 +45,15 @@ var (
 var RootCmd = &cobra.Command{
 	Use:   "olake",
 	Short: "root command",
-	RunE: func(cmd *cobra.Command, args []string) error {
-
+	PersistentPreRunE: func(cmd *cobra.Command, _ []string) error {
+		// Resolve now as configPaths are needed by logger.Init(), but the error is handled later because the logger is not initialized yet.
+		s3Err := utils.ResolveS3Paths(cmd.Context(), []*string{&configPath, &destinationConfigPath, &streamsPath, &statePath, &differencePath})
 		// set global variables
 		viper.SetDefault(constants.ConfigFolder, os.TempDir())
 		viper.SetDefault(constants.StatePath, filepath.Join(os.TempDir(), "state.json"))
 		viper.SetDefault(constants.StreamsPath, filepath.Join(os.TempDir(), "streams.json"))
 		viper.SetDefault(constants.DifferencePath, filepath.Join(os.TempDir(), "difference_streams.json"))
-		if !noSave {
+		if s3Err == nil && !noSave {
 			configFolder := utils.Ternary(configPath == "not-set", filepath.Dir(destinationConfigPath), filepath.Dir(configPath)).(string)
 			streamsPathEnv := utils.Ternary(streamsPath == "", filepath.Join(configFolder, "streams.json"), streamsPath).(string)
 			differencePathEnv := utils.Ternary(streamsPath != "", filepath.Join(filepath.Dir(streamsPath), "difference_streams.json"), filepath.Join(configFolder, "difference_streams.json")).(string)
@@ -66,10 +68,15 @@ var RootCmd = &cobra.Command{
 			viper.Set(constants.EncryptionKey, encryptionKey)
 		}
 
-		// logger uses CONFIG_FOLDER
-		logger.Init()
+		// logger uses CONFIG_FOLDER; S3 jobs get JSON stdout matching olake.log
+		logger.Init(s3.IsS3Job())
 		telemetry.Init()
 
+		// Checked last so a resolution failure is reported through the
+		// now-initialized logger instead of being silently discarded.
+		return s3Err
+	},
+	RunE: func(cmd *cobra.Command, args []string) error {
 		if len(args) == 0 {
 			return cmd.Help()
 		}
@@ -130,6 +137,9 @@ func signalAwareRootContext(parent context.Context) context.Context {
 }
 
 func init() {
+	// Run root PersistentPreRunE (S3 path resolution) before child hooks like sync's.
+	cobra.EnableTraverseRunHooks = true
+
 	// TODO: replace --catalog flag with --streams
 	commands = append(commands, specCmd, checkCmd, discoverCmd, syncCmd, clearCmd)
 	RootCmd.PersistentFlags().StringVarP(&configPath, "config", "", "not-set", "(Required) Config for connector")

protocol/sync.go

@@ -91,6 +91,8 @@ var syncCmd = &cobra.Command{
 		return nil
 	},
 	RunE: func(cmd *cobra.Command, _ []string) (err error) {
+		defer utils.FinalizeS3Upload(cmd.Context(), &err, noSave)
+
 		// setup conector first
 		err = connector.Setup(cmd.Context())
 		if err != nil {

utils/logger/logger.go

@@ -10,13 +10,16 @@ import (
 	"net/http/httputil"
 	"os"
 	"os/exec"
+	"path"
 	"path/filepath"
 	"regexp"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/datazip-inc/olake/constants"
+	"github.com/datazip-inc/olake/utils/s3"
 	"github.com/rs/zerolog"
 	"github.com/shirou/gopsutil/v4/cpu"
 	"github.com/shirou/gopsutil/v4/mem"
@@ -174,25 +177,34 @@ func StatsLogger(ctx context.Context, statsFunc func() (int64, int64, int64, int
 				if err := FileLogger(stats, "stats", ".json"); err != nil {
 					Fatalf("failed to write stats in file: %s", err)
 				}
+				if s3.IsS3Job() {
+					statsPath := filepath.Join(viper.GetString(constants.ConfigFolder), "stats.json")
+					if err := s3.UploadFileToS3(ctx, statsPath, s3.JobBucket, path.Join(s3.JobPrefix, "stats.json")); err != nil {
+						Debugf("failed to upload stats.json: %s", err)
+					}
+				}
 			}
 		}
 	}()
 }
 
-func Init() {
+func Init(s3Mode bool) {
 	// Set up timestamp for log file names
 	currentTimestamp := time.Now().UTC()
 	timestamp := fmt.Sprintf("%d-%02d-%02d_%02d-%02d-%02d",
 		currentTimestamp.Year(), currentTimestamp.Month(), currentTimestamp.Day(),
 		currentTimestamp.Hour(), currentTimestamp.Minute(), currentTimestamp.Second())
 
 	// Configure rotating file logs
-	rotatingFile := &lumberjack.Logger{
-		Filename:   fmt.Sprintf("%s/logs/sync_%s/olake.log", viper.GetString(constants.ConfigFolder), timestamp),
-		MaxSize:    100, // Max size in MB
-		MaxBackups: 5,   // Number of old log files to retain
-		MaxAge:     30,  // Days to retain old log files
-		Compress:   true,
+	var rotatingFile *lumberjack.Logger
+	if !s3Mode {
+		rotatingFile = &lumberjack.Logger{
+			Filename:   fmt.Sprintf("%s/logs/sync_%s/olake.log", viper.GetString(constants.ConfigFolder), timestamp),
+			MaxSize:    100, // Max size in MB
+			MaxBackups: 5,   // Number of old log files to retain
+			MaxAge:     30,  // Days to retain old log files
+			Compress:   true,
+		}
 	}
 
 	zerolog.TimestampFunc = func() time.Time { return time.Now().UTC() }
@@ -236,6 +248,15 @@ func Init() {
 		}
 	}
 
+	if s3Mode {
+		// S3 jobs capture stdout; JSON matches olake.log format.
+		var seq uint64
+		logger = zerolog.New(os.Stdout).Hook(zerolog.HookFunc(func(e *zerolog.Event, _ zerolog.Level, _ string) {
+			e.Uint64("seq", atomic.AddUint64(&seq, 1))
+		})).With().Timestamp().Logger()
+		return
+	}
+
 	// MultiWriter (rotating file + console writer per goroutine)
 	multiWriter := zerolog.MultiLevelWriter(rotatingFile, newConsoleWriter())
 

utils/s3/s3.go

@@ -0,0 +1,112 @@
+package s3
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/datazip-inc/olake/constants"
+)
+
+var (
+	s3Client  *awss3.Client
+	JobBucket string
+	JobPrefix string
+)
+
+// IsS3Job reports whether OLAKE_STORAGE_MODE is s3.
+func IsS3Job() bool {
+	return os.Getenv(constants.EnvStorageMode) == constants.StorageModeS3
+}
+
+// Init initializes the shared S3 client when storage mode is S3. No-op for NFS.
+func Init(ctx context.Context) error {
+	if !IsS3Job() {
+		return nil
+	}
+
+	configOpts := []func(*config.LoadOptions) error{}
+	if region := os.Getenv(constants.EnvS3Region); region != "" {
+		configOpts = append(configOpts, config.WithRegion(region))
+	}
+
+	accessKey := os.Getenv(constants.EnvS3AccessKeyID)
+	secretKey := os.Getenv(constants.EnvS3SecretAccessKey)
+	if accessKey != "" && secretKey != "" {
+		configOpts = append(configOpts, config.WithCredentialsProvider(
+			credentials.NewStaticCredentialsProvider(accessKey, secretKey, os.Getenv(constants.EnvS3SessionToken)),
+		))
+	}
+
+	awsCfg, err := config.LoadDefaultConfig(ctx, configOpts...)
+	if err != nil {
+		return fmt.Errorf("failed to load AWS config: %s", err)
+	}
+
+	var s3Opts []func(*awss3.Options)
+	if endpoint := os.Getenv(constants.EnvS3Endpoint); endpoint != "" {
+		// Path-style is required for MinIO and other S3-compatible endpoints.
+		s3Opts = append(s3Opts, func(o *awss3.Options) {
+			o.BaseEndpoint = aws.String(endpoint)
+			o.UsePathStyle = true
+		})
+	}
+
+	s3Client = awss3.NewFromConfig(awsCfg, s3Opts...)
+	return nil
+}
+
+func getS3Client() (*awss3.Client, error) {
+	if s3Client == nil {
+		return nil, fmt.Errorf("s3 storage not initialized")
+	}
+	return s3Client, nil
+}
+
+// GetObject downloads an object and records the job prefix from the first key.
+func GetObject(ctx context.Context, bucket, key string) (*awss3.GetObjectOutput, error) {
+	if JobBucket == "" {
+		JobBucket = bucket
+		JobPrefix = path.Dir(key)
+	}
+
+	client, err := getS3Client()
+	if err != nil {
+		return nil, err
+	}
+	return client.GetObject(ctx, &awss3.GetObjectInput{
+		Bucket: aws.String(bucket),
+		Key:    aws.String(key),
+	})
+}
+
+// UploadFile uploads a local file to bucket/key.
+func UploadFileToS3(ctx context.Context, localPath, bucket, key string) error {
+	client, err := getS3Client()
+	if err != nil {
+		return err
+	}
+
+	file, err := os.Open(localPath)
+	if err != nil {
+		return fmt.Errorf("failed to open local file %s: %s", localPath, err)
+	}
+	defer file.Close()
+
+	// Upload the file to S3.
+	_, err = client.PutObject(ctx, &awss3.PutObjectInput{
+		Bucket: aws.String(bucket),
+		Key:    aws.String(key),
+		Body:   file,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to upload s3://%s/%s: %s", bucket, key, err)
+	}
+
+	return nil
+}