Merge pull request #1207 from percona/PBM-1605-fix-incomplete-backups-gcs-hmac

PBM-1605: PBM marks incomplete backups as successful without logging errors

Author: boris-ilijic

pbm/storage/gcs/hmac_client.go

@@ -2,6 +2,9 @@ package gcs
 
 import (
 	"context"
+	"encoding/base64"
+	"encoding/binary"
+	"hash/crc32"
 	"io"
 	"path"
 	"runtime"
@@ -73,22 +76,31 @@ func (h hmacClient) save(name string, data io.Reader, options ...storage.Option)
 			partSize, storage.PrettySize(partSize))
 	}
 
+	crc := crc32.New(crc32.MakeTable(crc32.Castagnoli))
+	dataWithCRC := io.TeeReader(data, crc)
+
 	putOpts := minio.PutObjectOptions{
 		PartSize:   uint64(partSize),
 		NumThreads: uint(max(runtime.NumCPU()/2, 1)),
 	}
-
-	_, err := h.client.PutObject(
+	putInfo, err := h.client.PutObject(
 		context.Background(),
 		h.opts.Bucket,
 		path.Join(h.opts.Prefix, name),
-		data,
+		dataWithCRC,
 		-1,
 		putOpts,
 	)
 	if err != nil {
 		return errors.Wrap(err, "PutObject")
 	}
+
+	localCRC := crcToBase64(crc.Sum32())
+	if putInfo.ChecksumCRC32C != localCRC {
+		return errors.Errorf("wrong CRC after uploading %s, GCS: %s, PBM: %s",
+			name, putInfo.ChecksumCRC32C, localCRC)
+	}
+
 	return nil
 }
 
@@ -215,3 +227,9 @@ func (h hmacClient) getPartialObject(name string, buf *storage.Arena, start, len
 
 	return ch, nil
 }
+
+func crcToBase64(v uint32) string {
+	buf := make([]byte, 4)
+	binary.BigEndian.PutUint32(buf, v)
+	return base64.StdEncoding.EncodeToString(buf)
+}

pbm/storage/mio/minio_test.go

@@ -2,8 +2,12 @@ package mio
 
 import (
 	"context"
+	"io"
 	"net/url"
+	"path"
+	"runtime"
 	"testing"
+	"time"
 
 	"github.com/minio/minio-go/v7"
 	"github.com/minio/minio-go/v7/pkg/credentials"
@@ -80,3 +84,174 @@ func TestMinio(t *testing.T) {
 		storage.RunSplitMergeMWTests(t, stg)
 	})
 }
+
+// TestUploadGCS shows how it's possible to upload corrupted file
+// without getting any error from the minio library.
+//
+// To simulate network interruption use:
+// tc qdisc add dev eth0 root netem loss 100%
+//
+// To revert it to normal use:
+// tc qdisc del dev eth0 root netem
+func TestUploadGCS(t *testing.T) {
+	t.Skip("for manual invocation, it will be deleted after GCS HMAC is deprecated")
+
+	ep := "storage.googleapis.com"
+	bucket := "gcs-bucket"
+	prefix := "test-prefix"
+	accessKeyID := "key-id"
+	secretAccessKey := "secret-key"
+
+	fname := time.Now().Format("2006-01-02T15:04:05")
+
+	mc, err := minio.New(ep, &minio.Options{
+		Creds:  credentials.NewStaticV2(accessKeyID, secretAccessKey, ""),
+		Secure: true,
+	})
+	if err != nil {
+		t.Fatalf("minio client creation for GCS: %v", err)
+	}
+	t.Log("minio client created")
+
+	t.Logf("uploading file: %s", fname)
+
+	infR := NewInfiniteCustomReader()
+	r := io.LimitReader(infR, targetSizeBytes)
+
+	putOpts := minio.PutObjectOptions{
+		PartSize:   uint64(defaultPartSize),
+		NumThreads: uint(max(runtime.NumCPU()/2, 1)),
+	}
+	info, err := mc.PutObject(
+		context.Background(),
+		bucket,
+		path.Join(prefix, fname),
+		r,
+		-1,
+		putOpts,
+	)
+	if err != nil {
+		t.Fatalf("put object: %v", err)
+	}
+
+	t.Logf("upload info: %#v", info)
+}
+
+func TestUploadAWSSigV2(t *testing.T) {
+	t.Skip("for manual invocation, it will be deleted after GCS HMAC is deprecated")
+
+	ep := "s3.amazonaws.com"
+	region := "eu-central-1"
+	bucket := "aws-bucket"
+	prefix := "test-prefix"
+	accessKeyID := "key-id"
+	secretAccessKey := "secret-key"
+
+	fname := time.Now().Format("2006-01-02T15:04:05")
+
+	mc, err := minio.New(ep, &minio.Options{
+		Region: region,
+		Creds:  credentials.NewStaticV2(accessKeyID, secretAccessKey, ""),
+		Secure: true,
+	})
+	if err != nil {
+		t.Fatalf("minio client creation for aws: %v", err)
+	}
+	t.Log("minio client created for aws with sigV2")
+
+	t.Logf("uploading file: %s", fname)
+
+	infR := NewInfiniteCustomReader()
+	r := io.LimitReader(infR, targetSizeBytes)
+
+	putOpts := minio.PutObjectOptions{
+		PartSize:   uint64(defaultPartSize),
+		NumThreads: uint(max(runtime.NumCPU()/2, 1)),
+	}
+	info, err := mc.PutObject(
+		context.Background(),
+		bucket,
+		path.Join(prefix, fname),
+		r,
+		-1,
+		putOpts,
+	)
+	if err != nil {
+		t.Fatalf("put object: %v", err)
+	}
+
+	t.Logf("upload info: %#v", info)
+}
+
+func TestUploadAWSSigV4(t *testing.T) {
+	t.Skip("for manual invocation, it will be deleted after GCS HMAC is deprecated")
+
+	ep := "s3.amazonaws.com"
+	region := "eu-central-1"
+	bucket := "aws-bucket"
+	prefix := "test-prefix"
+	accessKeyID := "key-id"
+	secretAccessKey := "secret-key"
+
+	fname := time.Now().Format("2006-01-02T15:04:05")
+
+	mc, err := minio.New(ep, &minio.Options{
+		Region: region,
+		Creds:  credentials.NewStaticV4(accessKeyID, secretAccessKey, ""),
+		Secure: true,
+	})
+	if err != nil {
+		t.Fatalf("minio client creation for aws: %v", err)
+	}
+	t.Log("minio client created for aws with sigV4")
+
+	t.Logf("uploading file: %s ....", fname)
+
+	infR := NewInfiniteCustomReader()
+	r := io.LimitReader(infR, targetSizeBytes)
+
+	putOpts := minio.PutObjectOptions{
+		PartSize:   uint64(defaultPartSize),
+		NumThreads: uint(max(runtime.NumCPU()/2, 1)),
+	}
+	info, err := mc.PutObject(
+		context.Background(),
+		bucket,
+		path.Join(prefix, fname),
+		r,
+		-1,
+		putOpts,
+	)
+	if err != nil {
+		t.Fatalf("put object: %v", err)
+	}
+
+	t.Logf("upload info: %#v", info)
+}
+
+const targetSizeBytes = 1000 * 1024 * 1024
+
+type InfiniteCustomReader struct {
+	pattern      []byte
+	patternIndex int
+}
+
+func NewInfiniteCustomReader() *InfiniteCustomReader {
+	pattern := []byte{0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF, 0x11, 0x22}
+
+	return &InfiniteCustomReader{
+		pattern:      pattern,
+		patternIndex: 0,
+	}
+}
+
+func (r *InfiniteCustomReader) Read(p []byte) (int, error) {
+	readLen := len(p)
+
+	for i := range readLen {
+		p[i] = r.pattern[r.patternIndex]
+		r.patternIndex = (r.patternIndex + 1) % len(r.pattern)
+	}
+
+	return readLen, nil
+}