Test S3 gateway multipart copy in Esti

esti/s3_gateway_test.go

@@ -2,6 +2,7 @@ package esti
 
 import (
 	"bytes"
+	"hash/crc32"
 	"io"
 	"math/rand"
 	"net/http"
@@ -328,6 +329,87 @@ func TestS3HeadBucket(t *testing.T) {
 	})
 }
 
+func TestS3CopyObjectMultipart(t *testing.T) {
+	const minDataContentLengthForMultipart = 5 << 20
+	ctx, _, repo := setupTest(t)
+	defer tearDownTest(repo)
+
+	// additional repository for copy between repos
+	const destRepoName = "tests3copyobjectdest"
+	destRepo := createRepositoryByName(ctx, t, destRepoName)
+	defer deleteRepositoryIfAskedTo(ctx, destRepoName)
+
+	// content
+	r := rand.New(rand.NewSource(17))
+	objContent := testutil.RandomReader(r, largeDataContentLength)
+	srcPath := gatewayTestPrefix + "source-file"
+	destPath := gatewayTestPrefix + "dest-file"
+
+	// upload data
+	s3lakefsClient := newMinioClient(t, credentials.NewStaticV2)
+	_, err := s3lakefsClient.PutObject(ctx, repo, srcPath, objContent, largeDataContentLength,
+		minio.PutObjectOptions{})
+	require.NoError(t, err)
+
+	dest := minio.CopyDestOptions{
+		Bucket: destRepo,
+		Object: destPath,
+	}
+
+	srcs := []minio.CopySrcOptions{
+		{
+			Bucket: srcRepo,
+			Object: srcPath,
+			Start:  0,
+			End:    minDataContentLengthForMultipart,
+		}, {
+			Bucket: srcRepo,
+			Object: srcPath,
+			Start:  minDataContentLengthForMultipart,
+			End:    largeDataContentLength,
+		},
+	}
+
+	ui, err := s3lakefsClient.ComposeObject(ctx, dst, srcs...)
+	if err != nil {
+		t.Fatalf("minio.Client.ComposeObject from(%+v) to(%+v): %s", srcs, dest, err)
+	}
+
+	if ui.Size != largeDataContentLength {
+		t.Errorf("Copied %d bytes when expecting %d", ui.Size, largeDataContentLength)
+	}
+
+	// Comparing 2 readers is too much work.  Instead just hash them.
+	// This will fail for malicious bad S3 gateways, but otherwise is
+	// fine.
+	uploadedReader, err := s3lakefsClient.GetObject(ctx, repo, srcPath, minio.GetObjectOptions{})
+	if err != nil {
+		t.Fatalf("Get uploaded object: %s", err)
+	}
+	defer uploadedReader.Close()
+	uploadedCRC, err := testutil.ChecksumReader(uploadedReader)
+	if err != nil {
+		t.Fatalf("Read uploaded object: %s", err)
+	}
+	if uploadedCRC == 0 {
+		t.Fatal("Impossibly bad luck: uploaded data with CRC64 == 0!")
+	}
+
+	copiedReader, err := s3lakefsClient.GetObject(ctx, repo, srcPath, minio.GetObjectOptions{})
+	if err != nil {
+		t.Fatalf("Get copied object: %s", err)
+	}
+	defer copiedReader.Close()
+	copiedCRC, err := testutil.ChecksumReader(copiedReader)
+	if err != nil {
+		t.Fatalf("Read copied object: %s", err)
+	}
+
+	if uploadedCRC != copiedCRC {
+		t.Fatal("Copy not equal")
+	}
+}
+
 func TestS3CopyObject(t *testing.T) {
 	ctx, _, repo := setupTest(t)
 	defer tearDownTest(repo)

esti/system_test.go

@@ -146,6 +146,10 @@ func deleteRepositoryIfAskedTo(ctx context.Context, repositoryName string) {
 
 const randomDataContentLength = 16
 
+// largeDataContentLength is >5 MiB, which is large enough to require
+// multipart operations.
+const largeDataContentLength = 6 << 20
+
 func uploadFileRandomDataAndReport(ctx context.Context, repo, branch, objPath string, direct bool) (checksum, content string, err error) {
 	objContent := randstr.String(randomDataContentLength)
 	checksum, err = uploadFileAndReport(ctx, repo, branch, objPath, objContent, direct)

pkg/testutil/checksum.go

@@ -0,0 +1,29 @@
+package testutil
+
+import (
+	"hash/crc64"
+	"io"
+)
+
+const bufSize = 4 << 16
+
+var polynomial = crc64.ECMA
+
+// ChecksumReader returns the checksum (CRC-64) of the contents of reader.
+func ChecksumReader(reader io.Reader) (uint64, error) {
+	buf := make([]byte, bufSize)
+	var val uint64
+	for {
+		n, err := r.Read(b)
+		if err != nil {
+			if err == io.EOF {
+				return val, nil
+			}
+			return val, err
+		}
+		if n == 0 {
+			return val, nil
+		}
+		val = crc64.Update(val, polynomial, b[:n])
+	}
+}

pkg/testutil/random.go

@@ -1,6 +1,7 @@
 package testutil
 
 import (
+	"io"
 	"math/rand"
 	"strings"
 	"unicode/utf8"
@@ -33,3 +34,23 @@ func RandomString(rand *rand.Rand, size int) string {
 	_, lastRuneSize := utf8.DecodeLastRuneInString(ret)
 	return ret[0 : len(ret)-lastRuneSize]
 }
+
+type randomReader struct {
+	rand      *rand.Rand
+	remaining int64
+}
+
+func (r *randomReader) Read(p []byte) (n int, err error) {
+	n := len(p)
+	if n > r.remaining {
+		n = r.remaining
+	}
+	r.rand.Read(p[:n])
+	r.remaining -= n
+	return n, nil
+}
+
+// RandomReader returns a reader that will return size bytes from rand.
+func RandomReader(rand *rand.Rand, size int64) io.Reader {
+	return &randomReader{rand: rand, remaining: size}
+}