Implement streaming S3 facility

Author: martin31821

README.md

@@ -117,6 +117,8 @@ Examples:
 ```s3sync --tk KEY2 --ts SECRET2 --sk KEY1 --ss SECRET1 --se "http://127.0.0.1:7484" --te "http://127.0.0.1:7484" -w 128 s3://shared s3://shared_new```
 * Sync one Amazon bucket directory to another Amazon bucket:  
 ```s3sync --tk KEY2 --ts SECRET2 --sk KEY1 --ss SECRET1 -w 128 s3://shared/test/ s3://shared_new```
+* Use streaming S3 transfers (reduces memory, might be slower for small files):
+```s3sync --sk KEY --ss SECRET --tk KEY --ts SECRET -w 128 s3s://shared/test/ s3s://shared_new```
 
 SOURCE and TARGET should be a directory. Syncing of single file are not supported (This will not work `s3sync --sk KEY --ss SECRET s3://shared/megafile.zip fs:///opt/backups/s3/`)  
 

cli/cli.go

@@ -2,15 +2,16 @@ package main
 
 import (
 	"fmt"
-	"github.com/alexflint/go-arg"
-	"github.com/larrabee/s3sync/storage"
-	"github.com/mattn/go-isatty"
 	"net/url"
 	"os"
 	"strconv"
 	"strings"
 	"time"
 	"unicode"
+
+	"github.com/alexflint/go-arg"
+	"github.com/larrabee/s3sync/storage"
+	"github.com/mattn/go-isatty"
 )
 
 var (
@@ -202,6 +203,10 @@ func parseConn(cStr string) (conn connect, err error) {
 		conn.Type = storage.TypeS3
 		conn.Bucket = u.Host
 		conn.Path = strings.TrimPrefix(u.Path, "/")
+  case "s3s":
+    conn.Type = storage.TypeS3Stream
+    conn.Bucket = u.Host
+    conn.Path = strings.TrimPrefix(u.Path, "/")
 	case "fs":
 		conn.Type = storage.TypeFS
 		conn.Path = strings.TrimPrefix(cStr, "fs://")

cli/setup.go

@@ -3,12 +3,14 @@ package main
 import (
 	"context"
 	"fmt"
+	"os"
+
 	"github.com/larrabee/s3sync/pipeline"
 	"github.com/larrabee/s3sync/pipeline/collection"
 	"github.com/larrabee/s3sync/storage"
 	"github.com/larrabee/s3sync/storage/fs"
 	"github.com/larrabee/s3sync/storage/s3"
-	"os"
+	"github.com/larrabee/s3sync/storage/s3stream"
 )
 
 func setupStorages(ctx context.Context, syncGroup *pipeline.Group, cli *argsParsed) error {
@@ -18,6 +20,10 @@ func setupStorages(ctx context.Context, syncGroup *pipeline.Group, cli *argsPars
 		sourceStorage = s3.NewS3Storage(cli.SourceNoSign, cli.SourceKey, cli.SourceSecret, cli.SourceToken, cli.SourceRegion, cli.SourceEndpoint,
 			cli.Source.Bucket, cli.Source.Path, cli.S3KeysPerReq, cli.S3Retry, cli.S3RetryInterval,
 		)
+  case storage.TypeS3Stream:
+    sourceStorage = s3stream.NewS3StreamStorage(cli.SourceNoSign, cli.SourceKey, cli.SourceSecret, cli.SourceToken, cli.SourceRegion, cli.SourceEndpoint,
+			cli.Source.Bucket, cli.Source.Path, cli.S3KeysPerReq, cli.S3Retry, cli.S3RetryInterval,
+		)
 	case storage.TypeFS:
 		sourceStorage = fs.NewFSStorage(cli.Source.Path, cli.FSFilePerm, cli.FSDirPerm, os.Getpagesize()*256*32, !cli.FSDisableXattr, cli.ErrorHandlingMask, cli.FSAtomicWrite)
 	}
@@ -27,6 +33,10 @@ func setupStorages(ctx context.Context, syncGroup *pipeline.Group, cli *argsPars
 		targetStorage = s3.NewS3Storage(cli.TargetNoSign, cli.TargetKey, cli.TargetSecret, cli.TargetToken, cli.TargetRegion, cli.TargetEndpoint,
 			cli.Target.Bucket, cli.Target.Path, cli.S3KeysPerReq, cli.S3Retry, cli.S3RetryInterval,
 		)
+  case storage.TypeS3Stream:
+		targetStorage = s3stream.NewS3StreamStorage(cli.TargetNoSign, cli.TargetKey, cli.TargetSecret, cli.TargetToken, cli.TargetRegion, cli.TargetEndpoint,
+			cli.Target.Bucket, cli.Target.Path, cli.S3KeysPerReq, cli.S3Retry, cli.S3RetryInterval,
+		)
 	case storage.TypeFS:
 		targetStorage = fs.NewFSStorage(cli.Target.Path, cli.FSFilePerm, cli.FSDirPerm, 0, !cli.FSDisableXattr, cli.ErrorHandlingMask, cli.FSAtomicWrite)
 	}

pipeline/collection/misc.go

@@ -28,7 +28,7 @@ var Logger pipeline.StepFn = func(group *pipeline.Group, stepNum int, input <-ch
 		if ok {
 			cfg.WithFields(logrus.Fields{
 				"key":          *obj.Key,
-				"size":         len(*obj.Content),
+				"size":         *obj.ContentLength,
 				"Content-Type": *obj.ContentType,
 			}).Infof("Sync file")
 			output <- obj

storage/fs/fs.go

@@ -5,16 +5,17 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"github.com/karrick/godirwalk"
-	"github.com/larrabee/ratelimit"
-	"github.com/larrabee/s3sync/storage"
-	"github.com/pkg/xattr"
 	"io"
 	"io/ioutil"
 	"mime"
 	"os"
 	"path/filepath"
 	"strings"
+
+	"github.com/karrick/godirwalk"
+	"github.com/larrabee/ratelimit"
+	"github.com/larrabee/s3sync/storage"
+	"github.com/pkg/xattr"
 )
 
 const tempFileSuffixLen = 8
@@ -188,7 +189,10 @@ func (st *FSStorage) GetObjectContent(obj *storage.Object) error {
 		return err
 	}
 
+  dataSize := int64(len(data))
+
 	obj.Content = &data
+  obj.ContentLength = &dataSize
 
 	if err := st.GetObjectMeta(obj); err != nil {
 		return err
@@ -256,4 +260,4 @@ func (st *FSStorage) DeleteObject(obj *storage.Object) error {
 	}
 
 	return nil
-}
+}

storage/s3/s3.go

@@ -3,17 +3,18 @@ package s3
 import (
 	"bytes"
 	"context"
+	"io"
+	"net/url"
+	"strings"
+	"time"
+
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/defaults"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/larrabee/ratelimit"
 	"github.com/larrabee/s3sync/storage"
-	"io"
-	"net/url"
-	"strings"
-	"time"
 )
 
 // S3Storage configuration.
@@ -186,6 +187,7 @@ func (st *S3Storage) GetObjectContent(obj *storage.Object) error {
 	data := buf.Bytes()
 	obj.Content = &data
 	obj.ContentType = result.ContentType
+  obj.ContentLength = result.ContentLength
 	obj.ContentDisposition = result.ContentDisposition
 	obj.ContentEncoding = result.ContentEncoding
 	obj.ContentLanguage = result.ContentLanguage