Repository: C2FO/vfs
Pull request #95 (2021-09-10 10:57:54)

Ep 7935 - S3 Multipart Download
Implemented S3 Downloader to multipart download files on initial read calls. Updated test suite to mock S3 Downloader

```
Going to go with 32MB Chunks with 5 Processes Default. Results Below:
1 - 226/366 = 38% Faster Than S3 Get Objects (Single Socket Download)
366/226 = S3 Get Object 162% slower than S3 Downloader (Parallel Multipart Download)

Master Results S3 Get Object:
BenchmarkRead_AWS_File_100MB-16               10        4595231254 ns/op
BenchmarkRead_AWS_File_1GB-16                 10        36590394191 ns/op

Read with S3 Download Manager 5 MB chunks 5 Processes:
BenchmarkRead_AWS_File_100MB-16               10        4113697995 ns/op
BenchmarkRead_AWS_File_1GB-16                 10        22814272873 ns/op

Read with S# Download Manager 32 MB chunks 5 Processes:
BenchmarkRead_AWS_File_100MB-16               10        4151013363 ns/op
BenchmarkRead_AWS_File_1GB-16                 10        22638071967 ns/op
```


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index caba742..a1d7cd7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,9 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 ## [Unreleased]
+### Changed
+- Modified S3 file.go so that on the initial read when the remote file is downloaded, a temporary file is locally created
+using concurrent go routines to download parts of the file vs. a single request to download the whole object
 
 ## [5.9.0] - 2021-09-07
 ### Added
diff --git a/backend/s3/file.go b/backend/s3/file.go
index 95888f8..4dbbe18 100644
--- a/backend/s3/file.go
+++ b/backend/s3/file.go
@@ -10,8 +10,10 @@ import (
 	"path"
 	"time"
 
+	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/aws/aws-sdk-go/service/s3/s3iface"
 	"github.com/aws/aws-sdk-go/service/s3/s3manager"
 
 	"github.com/c2fo/vfs/v5"
@@ -29,6 +31,15 @@ type File struct {
 	writeBuffer *bytes.Buffer
 }
 
+// Downloader interface needed to mock S3 Downloader data access object in tests
+type Downloader interface {
+	Download(w io.WriterAt, input *s3.GetObjectInput, options ...func(downloader *s3manager.Downloader)) (n int64, err error)
+
+	DownloadWithContext(ctx aws.Context, w io.WriterAt, input *s3.GetObjectInput, options ...func(*s3manager.Downloader)) (n int64, err error)
+
+	DownloadWithIterator(ctx aws.Context, iter s3manager.BatchDownloadIterator, opts ...func(*s3manager.Downloader)) error
+}
+
 // Info Functions
 
 // LastModified returns the LastModified property of a HEAD request to the s3 object.
@@ -381,27 +392,32 @@ func (f *File) checkTempFile() error {
 }
 
 func (f *File) copyToLocalTempReader() (*os.File, error) {
+	// Create temp file
 	tmpFile, err := ioutil.TempFile("", fmt.Sprintf("%s.%d", f.Name(), time.Now().UnixNano()))
 	if err != nil {
 		return nil, err
 	}
 
-	outputReader, err := f.getObject()
-	if err != nil {
-		return nil, err
+	// Create S3 Downloader, get client, and set partition size for multipart download
+	var partSize int64
+	if opts, ok := f.Location().FileSystem().(*FileSystem).options.(Options); ok {
+		if partSize = opts.DownloadPartitionSize; partSize == 0 {
+			partSize = 32 * 1024 * 1024 // 32 MB per partition default if opts.DownloadPartitionSize is 0
+		}
 	}
 
-	buffer := make([]byte, utils.TouchCopyMinBufferSize)
-	if _, err := io.CopyBuffer(tmpFile, outputReader, buffer); err != nil {
+	client, err := f.fileSystem.Client()
+	if err != nil {
 		return nil, err
 	}
 
-	// Return cursor to the beginning of the new temp file
-	if _, err := tmpFile.Seek(0, 0); err != nil {
+	// Download file
+	_, err = getDownloader(client, partSize).Download(tmpFile, f.getObjectInput())
+	if err != nil {
 		return nil, err
 	}
 
-	// initialize temp ReadCloser
+	// Return temp file
 	return tmpFile, nil
 }
 
@@ -409,19 +425,6 @@ func (f *File) getObjectInput() *s3.GetObjectInput {
 	return new(s3.GetObjectInput).SetBucket(f.bucket).SetKey(f.key)
 }
 
-func (f *File) getObject() (io.ReadCloser, error) {
-	client, err := f.fileSystem.Client()
-	if err != nil {
-		return nil, err
-	}
-	getOutput, err := client.GetObject(f.getObjectInput())
-	if err != nil {
-		return nil, err
-	}
-
-	return getOutput.Body, nil
-}
-
 //TODO: need to provide an implementation-agnostic container for providing config options such as SSE
 func uploadInput(f *File) *s3manager.UploadInput {
 	sseType := "AES256"
@@ -481,3 +484,9 @@ func waitUntilFileExists(file vfs.File, retries int) error {
 
 	return nil
 }
+
+var getDownloader = func(client s3iface.S3API, partSize int64) Downloader {
+	return s3manager.NewDownloaderWithClient(client, func(d *s3manager.Downloader) {
+		d.PartSize = partSize
+	})
+}
diff --git a/backend/s3/options.go b/backend/s3/options.go
index 1b39502..9a895a8 100644
--- a/backend/s3/options.go
+++ b/backend/s3/options.go
@@ -18,15 +18,16 @@ import (
 
 // Options holds s3-specific options.  Currently only client options are used.
 type Options struct {
-	AccessKeyID     string `json:"accessKeyId,omitempty"`
-	SecretAccessKey string `json:"secretAccessKey,omitempty"`
-	SessionToken    string `json:"sessionToken,omitempty"`
-	Region          string `json:"region,omitempty"`
-	Endpoint        string `json:"endpoint,omitempty"`
-	ACL             string `json:"acl,omitempty"`
-	Retry           request.Retryer
-	MaxRetries      int
-	FileBufferSize  int // Buffer Size In Bytes Used with utils.TouchCopyBuffered
+	AccessKeyID           string `json:"accessKeyId,omitempty"`
+	SecretAccessKey       string `json:"secretAccessKey,omitempty"`
+	SessionToken          string `json:"sessionToken,omitempty"`
+	Region                string `json:"region,omitempty"`
+	Endpoint              string `json:"endpoint,omitempty"`
+	ACL                   string `json:"acl,omitempty"`
+	Retry                 request.Retryer
+	MaxRetries            int
+	FileBufferSize        int   // Buffer size in bytes used with utils.TouchCopyBuffered
+	DownloadPartitionSize int64 // Partition size in bytes used to multipart download large files using S3 Downloader
 }
 
 // getClient setup S3 client
diff --git a/mocks/S3Downloader.go b/mocks/S3Downloader.go
new file mode 100644
index 0000000..7f2c015
--- /dev/null
+++ b/mocks/S3Downloader.go
@@ -0,0 +1,33 @@
+package mocks
+
+import (
+	"io"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/aws/aws-sdk-go/service/s3/s3manager"
+)
+
+type S3MockDownloader struct {
+	SideEffect       error
+	Downloaded       int64
+	IsCalled         bool
+	ExpectedContents string
+}
+
+func (m S3MockDownloader) Download(w io.WriterAt, input *s3.GetObjectInput, options ...func(downloader *s3manager.Downloader)) (n int64, err error) {
+	m.IsCalled = true
+	written, err := w.WriteAt([]byte(m.ExpectedContents), 0)
+	if err != nil {
+		return 0, err
+	}
+	return int64(written), m.SideEffect
+}
+
+func (m S3MockDownloader) DownloadWithContext(ctx aws.Context, w io.WriterAt, input *s3.GetObjectInput, options ...func(*s3manager.Downloader)) (n int64, err error) {
+	return m.Downloaded, m.SideEffect
+}
+
+func (m S3MockDownloader) DownloadWithIterator(ctx aws.Context, iter s3manager.BatchDownloadIterator, opts ...func(*s3manager.Downloader)) error {
+	return m.SideEffect
+}


Test patch:
diff --git a/backend/s3/file_test.go b/backend/s3/file_test.go
index f808498..256b390 100644
--- a/backend/s3/file_test.go
+++ b/backend/s3/file_test.go
@@ -14,6 +14,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/aws/request"
 	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/aws/aws-sdk-go/service/s3/s3iface"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/suite"
@@ -25,6 +26,7 @@ import (
 
 type fileTestSuite struct {
 	suite.Suite
+	getDownloader func(client s3iface.S3API, partSize int64) Downloader
 }
 
 var (
@@ -40,22 +42,27 @@ func (ts *fileTestSuite) SetupTest() {
 	defaultOptions = Options{AccessKeyID: "abc"}
 	fs = FileSystem{client: s3apiMock, options: defaultOptions}
 	testFile, err = fs.NewFile("bucket", "/some/path/to/file.txt")
+	getDownloader = func(client s3iface.S3API, partSize int64) Downloader {
+		return mocks.S3MockDownloader{}
+	}
+
 	if err != nil {
 		ts.Fail("Shouldn't return error creating test s3.File instance.")
 	}
 }
 
-type nopCloser struct {
-	io.Reader
+func (ts *fileTestSuite) TearDownTest() {
+	getDownloader = ts.getDownloader
 }
 
-func (nopCloser) Close() error { return nil }
-
 func (ts *fileTestSuite) TestRead() {
 	contents := "hello world!"
-	s3apiMock.On("GetObject", mock.AnythingOfType("*s3.GetObjectInput")).Return(&s3.GetObjectOutput{
-		Body: nopCloser{bytes.NewBufferString(contents)},
-	}, nil)
+	downloader := mocks.S3MockDownloader{
+		ExpectedContents: contents,
+	}
+	getDownloader = func(client s3iface.S3API, partSize int64) Downloader {
+		return downloader
+	}
 
 	file, err := fs.NewFile("bucket", "/some/path/file.txt")
 	if err != nil {
@@ -66,12 +73,10 @@ func (ts *fileTestSuite) TestRead() {
 
 	buffer := make([]byte, utils.TouchCopyMinBufferSize)
 	_, copyErr := io.CopyBuffer(localFile, file, buffer)
-	assert.NoError(ts.T(), copyErr, "no error expected")
+	ts.NoError(copyErr, "no error expected")
 	closeErr := file.Close()
-	assert.NoError(ts.T(), closeErr, "no error expected")
-
-	s3apiMock.AssertExpectations(ts.T())
-	ts.Equal(localFile.String(), contents, "Copying an s3 file to a buffer should fill buffer with file's contents")
+	ts.NoError(closeErr, "no error expected")
+	ts.Equal(contents, localFile.String(), "Copying an s3 file to a buffer should fill buffer with file's contents")
 }
 
 // TODO: Write on Close() (actual s3 calls wait until file is closed to be made.)
@@ -91,9 +96,12 @@ func (ts *fileTestSuite) TestSeek() {
 	file, err := fs.NewFile("bucket", "/tmp/hello.txt")
 	ts.NoError(err, "Shouldn't fail creating new file")
 
-	s3apiMock.On("GetObject", mock.AnythingOfType("*s3.GetObjectInput")).Return(&s3.GetObjectOutput{
-		Body: nopCloser{bytes.NewBufferString(contents)},
-	}, nil)
+	downloader := mocks.S3MockDownloader{
+		ExpectedContents: contents,
+	}
+	getDownloader = func(client s3iface.S3API, partSize int64) Downloader {
+		return downloader
+	}
 
 	_, seekErr := file.Seek(6, 0)
 	assert.NoError(ts.T(), seekErr, "no error expected")
@@ -117,7 +125,6 @@ func (ts *fileTestSuite) TestSeek() {
 
 	closeErr := file.Close()
 	assert.NoError(ts.T(), closeErr, "no error expected")
-	s3apiMock.AssertExpectations(ts.T())
 }
 
 func (ts *fileTestSuite) TestGetLocation() {
@@ -167,8 +174,6 @@ func (ts *fileTestSuite) TestCopyToFile() {
 		key:    "testKey.txt",
 	}
 
-	fooReader := ioutil.NopCloser(strings.NewReader("blah"))
-	s3apiMock.On("GetObject", mock.AnythingOfType("*s3.GetObjectInput")).Return(&s3.GetObjectOutput{Body: fooReader}, nil)
 	s3apiMock.On("CopyObject", mock.AnythingOfType("*s3.CopyObjectInput")).Return(&s3.CopyObjectOutput{}, nil)
 
 	err := testFile.CopyToFile(targetFile)
@@ -188,8 +193,6 @@ func (ts *fileTestSuite) TestCopyToFile() {
 	}
 	defaultOptions.FileBufferSize = originalBufferSize
 
-	fooReader = ioutil.NopCloser(strings.NewReader("blah"))
-	s3apiMock.On("GetObject", mock.AnythingOfType("*s3.GetObjectInput")).Return(&s3.GetObjectOutput{Body: fooReader}, nil)
 	s3apiMock.On("CopyObject", mock.AnythingOfType("*s3.CopyObjectInput")).Return(&s3.CopyObjectOutput{}, nil)
 
 	err = testFile.CopyToFile(targetFile)
@@ -198,23 +201,34 @@ func (ts *fileTestSuite) TestCopyToFile() {
 }
 
 func (ts *fileTestSuite) TestEmptyCopyToFile() {
+	contents := ""
+	downloader := mocks.S3MockDownloader{
+		ExpectedContents: contents,
+	}
+	getDownloader = func(client s3iface.S3API, partSize int64) Downloader {
+		return downloader
+	}
+
 	targetFile := &mocks.File{}
 	targetFile.On("Write", mock.Anything).Return(0, nil)
 	targetFile.On("Close").Return(nil)
 
-	// TODO: note that ioutils.NopCloser is deprecated with 1.16 and will be moved to io.NopCloser
-	rc := ioutil.NopCloser(strings.NewReader(""))
-	s3apiMock.On("GetObject", mock.AnythingOfType("*s3.GetObjectInput")).Return(&s3.GetObjectOutput{Body: rc}, nil, nil)
-
 	err := testFile.CopyToFile(targetFile)
 	ts.Nil(err, "Error shouldn't be returned from successful call to CopyToFile")
-	s3apiMock.AssertExpectations(ts.T())
 
 	// Assert that file was still written to and closed when the reader size is 0 bytes.
 	targetFile.AssertExpectations(ts.T())
 }
 
 func (ts *fileTestSuite) TestMoveToFile() {
+	contents := "hello world!"
+	downloader := mocks.S3MockDownloader{
+		ExpectedContents: contents,
+	}
+	getDownloader = func(client s3iface.S3API, partSize int64) Downloader {
+		return downloader
+	}
+
 	targetFile := &File{
 		fileSystem: &FileSystem{
 			client:  s3apiMock,
@@ -224,8 +238,6 @@ func (ts *fileTestSuite) TestMoveToFile() {
 		key:    "testKey.txt",
 	}
 
-	fooReader := ioutil.NopCloser(strings.NewReader("blah"))
-	s3apiMock.On("GetObject", mock.AnythingOfType("*s3.GetObjectInput")).Return(&s3.GetObjectOutput{Body: fooReader}, nil)
 	s3apiMock.On("CopyObject", mock.AnythingOfType("*s3.CopyObjectInput")).Return(&s3.CopyObjectOutput{}, nil)
 	s3apiMock.On("DeleteObject", mock.AnythingOfType("*s3.DeleteObjectInput")).Return(&s3.DeleteObjectOutput{}, nil)
 
@@ -316,6 +328,14 @@ func (ts *fileTestSuite) TestGetCopyObject() {
 }
 
 func (ts *fileTestSuite) TestMoveToFile_CopyError() {
+	contents := "hello world!"
+	downloader := mocks.S3MockDownloader{
+		ExpectedContents: contents,
+	}
+	getDownloader = func(client s3iface.S3API, partSize int64) Downloader {
+		return downloader
+	}
+
 	targetFile := &File{
 		fileSystem: &FileSystem{
 			client:  s3apiMock,
@@ -325,8 +345,6 @@ func (ts *fileTestSuite) TestMoveToFile_CopyError() {
 		key:    "testKey.txt",
 	}
 
-	fooReader := ioutil.NopCloser(strings.NewReader("blah"))
-	s3apiMock.On("GetObject", mock.AnythingOfType("*s3.GetObjectInput")).Return(&s3.GetObjectOutput{Body: fooReader}, nil)
 	s3apiMock.On("CopyObject", mock.AnythingOfType("*s3.CopyObjectInput")).Return(nil, errors.New("some copy error"))
 
 	err := testFile.MoveToFile(targetFile)
@@ -373,12 +391,19 @@ func (ts *fileTestSuite) TestCopyToLocation() {
 func (ts *fileTestSuite) TestTouch() {
 	// Copy portion tested through CopyToLocation, just need to test whether or not Delete happens
 	// in addition to CopyToLocation
+	contents := "hello world!"
+	downloader := mocks.S3MockDownloader{
+		ExpectedContents: contents,
+	}
+	getDownloader = func(client s3iface.S3API, partSize int64) Downloader {
+		return downloader
+	}
+
 	s3Mock1 := &mocks.S3API{}
-	fooReader := ioutil.NopCloser(strings.NewReader("blah"))
-	s3Mock1.On("GetObject", mock.AnythingOfType("*s3.GetObjectInput")).Return(&s3.GetObjectOutput{Body: fooReader}, nil)
 	s3Mock1.On("HeadObject", mock.AnythingOfType("*s3.HeadObjectInput")).Return(&s3.HeadObjectOutput{}, nil)
 	s3Mock1.On("CopyObject", mock.AnythingOfType("*s3.CopyObjectInput")).Return(nil, nil)
 	s3Mock1.On("DeleteObject", mock.AnythingOfType("*s3.DeleteObjectInput")).Return(&s3.DeleteObjectOutput{}, nil)
+
 	file := &File{
 		fileSystem: &FileSystem{
 			client:  s3Mock1,
@@ -387,6 +412,7 @@ func (ts *fileTestSuite) TestTouch() {
 		bucket: "newBucket",
 		key:    "/new/file/path/hello.txt",
 	}
+
 	terr := file.Touch()
 	ts.NoError(terr, "Shouldn't return error creating test s3.File instance.")
 
@@ -414,12 +440,19 @@ func (ts *fileTestSuite) TestTouch() {
 	s3Mock2.AssertExpectations(ts.T())
 
 }
+
 func (ts *fileTestSuite) TestMoveToLocation() {
 	// Copy portion tested through CopyToLocation, just need to test whether or not Delete happens
 	// in addition to CopyToLocation
+	contents := "hello world!"
+	downloader := mocks.S3MockDownloader{
+		ExpectedContents: contents,
+	}
+	getDownloader = func(client s3iface.S3API, partSize int64) Downloader {
+		return downloader
+	}
+
 	s3Mock1 := &mocks.S3API{}
-	fooReader := ioutil.NopCloser(strings.NewReader("blah"))
-	s3Mock1.On("GetObject", mock.AnythingOfType("*s3.GetObjectInput")).Return(&s3.GetObjectOutput{Body: fooReader}, nil)
 	s3Mock1.On("CopyObject", mock.AnythingOfType("*s3.CopyObjectInput")).Return(nil, nil)
 	s3Mock1.On("HeadObject", mock.AnythingOfType("*s3.HeadObjectInput")).Return(&s3.HeadObjectOutput{}, nil)
 	f := &File{
@@ -433,8 +466,6 @@ func (ts *fileTestSuite) TestMoveToLocation() {
 	location := new(mocks.Location)
 	location.On("NewFile", mock.Anything).Return(f, nil)
 
-	fooReader2 := ioutil.NopCloser(strings.NewReader("blah"))
-	s3apiMock.On("GetObject", mock.AnythingOfType("*s3.GetObjectInput")).Return(&s3.GetObjectOutput{Body: fooReader2}, nil)
 	s3apiMock.On("CopyObject", mock.AnythingOfType("*s3.CopyObjectInput")).Return(&s3.CopyObjectOutput{}, nil)
 	s3apiMock.On("DeleteObject", mock.AnythingOfType("*s3.DeleteObjectInput")).Return(&s3.DeleteObjectOutput{}, nil)
 
@@ -457,8 +488,6 @@ func (ts *fileTestSuite) TestMoveToLocation() {
 		Return(&File{fileSystem: &FileSystem{client: s3Mock1}, bucket: "bucket", key: "/new/hello.txt"}, nil)
 
 	s3apiMock2 := &mocks.S3API{}
-	fooReader3 := ioutil.NopCloser(strings.NewReader("blah"))
-	s3apiMock2.On("GetObject", mock.AnythingOfType("*s3.GetObjectInput")).Return(&s3.GetObjectOutput{Body: fooReader3}, nil)
 	s3apiMock2.On("CopyObject", mock.AnythingOfType("*s3.CopyObjectInput")).Return(&s3.CopyObjectOutput{}, nil)
 
 	fs = FileSystem{client: s3apiMock2}
@@ -476,13 +505,19 @@ func (ts *fileTestSuite) TestMoveToLocation() {
 }
 
 func (ts *fileTestSuite) TestMoveToLocationFail() {
+	contents := "hello world!"
+	downloader := mocks.S3MockDownloader{
+		ExpectedContents: contents,
+	}
+	getDownloader = func(client s3iface.S3API, partSize int64) Downloader {
+		return downloader
+	}
+
 	// If CopyToLocation fails we need to ensure DeleteObject isn't called.
 	otherFs := new(mocks.FileSystem)
 	location := new(mocks.Location)
 	location.On("NewFile", mock.Anything).Return(&File{fileSystem: &fs, bucket: "bucket", key: "/new/hello.txt"}, nil)
 
-	fooReader := ioutil.NopCloser(strings.NewReader("blah"))
-	s3apiMock.On("GetObject", mock.AnythingOfType("*s3.GetObjectInput")).Return(&s3.GetObjectOutput{Body: fooReader}, nil)
 	s3apiMock.On("CopyObject", mock.AnythingOfType("*s3.CopyObjectInput")).Return(nil, errors.New("didn't copy, oh noes"))
 
 	file, err := fs.NewFile("bucket", "/hello.txt")