object_disk.CopyObjectStreaming can't copy files with size more 5Gb to S3, cause pass localSize 0 into PutFileAbsolute, fix #1176

Signed-off-by: Slach <[email protected]>

Author: Slach

ChangeLog.md

@@ -1,3 +1,7 @@
+# v2.6.26
+BUG FIXES
+- `object_disk.CopyObjectStreaming` can't copy files with size more 5Gb to S3, cause pass localSize 0 into PutFileAbsolute, fix [1176](https://github.com/Altinity/clickhouse-backup/issues/1176)
+
 # v2.6.25
 IMPROVEMENTS
 - change retries from constant to exponential backoff and add RETRIES_JITTER configuration option, to avoid same time retries from parallel operation

pkg/storage/azblob.go

@@ -229,8 +229,12 @@ func (a *AzureBlob) DeleteFileFromObjectDiskBackup(ctx context.Context, key stri
 }
 
 func (a *AzureBlob) StatFile(ctx context.Context, key string) (RemoteFile, error) {
-	a.logf("AZBLOB->StatFile %s", key)
-	blob := a.Container.NewBlockBlobURL(path.Join(a.Config.Path, key))
+	return a.StatFileAbsolute(ctx, path.Join(a.Config.Path, key))
+}
+
+func (a *AzureBlob) StatFileAbsolute(ctx context.Context, key string) (RemoteFile, error) {
+	a.logf("AZBLOB->StatFileAbsolute %s", key)
+	blob := a.Container.NewBlockBlobURL(key)
 	r, err := blob.GetProperties(ctx, azblob.BlobAccessConditions{}, a.CPK)
 	if err != nil {
 		var se azblob.StorageError

pkg/storage/cos.go

@@ -67,8 +67,12 @@ func (c *COS) Close(ctx context.Context) error {
 }
 
 func (c *COS) StatFile(ctx context.Context, key string) (RemoteFile, error) {
-	// file max size is 5Gb
-	resp, err := c.client.Object.Get(ctx, path.Join(c.Config.Path, key), nil)
+	return c.StatFileAbsolute(ctx, path.Join(c.Config.Path, key))
+}
+
+func (c *COS) StatFileAbsolute(ctx context.Context, key string) (RemoteFile, error) {
+	// @todo - COS Stat file max size is 5Gb
+	resp, err := c.client.Object.Get(ctx, key, nil)
 	if err != nil {
 		var cosErr *cos.ErrorResponse
 		ok := errors.As(err, &cosErr)

pkg/storage/ftp.go

@@ -83,8 +83,12 @@ func (f *FTP) returnConnectionToPool(ctx context.Context, where string, client *
 }
 
 func (f *FTP) StatFile(ctx context.Context, key string) (RemoteFile, error) {
-	// cant list files, so check the dir
-	dir := path.Dir(path.Join(f.Config.Path, key))
+	return f.StatFileAbsolute(ctx, path.Join(f.Config.Path, key))
+}
+
+func (f *FTP) StatFileAbsolute(ctx context.Context, key string) (RemoteFile, error) {
+	// can't list files, so check the dir
+	dir := path.Dir(key)
 	client, err := f.getConnectionFromPool(ctx, fmt.Sprintf("StatFile, key=%s", key))
 	if err != nil {
 		return nil, err
@@ -98,7 +102,7 @@ func (f *FTP) StatFile(ctx context.Context, key string) (RemoteFile, error) {
 		}
 		return nil, err
 	}
-	file := path.Base(path.Join(f.Config.Path, key))
+	file := path.Base(key)
 	for i := range entries {
 		if file == entries[i].Name {
 			// file found, return it

pkg/storage/gcs.go

@@ -292,7 +292,11 @@ func (gcs *GCS) PutFileAbsolute(ctx context.Context, key string, r io.ReadCloser
 }
 
 func (gcs *GCS) StatFile(ctx context.Context, key string) (RemoteFile, error) {
-	objAttr, err := gcs.client.Bucket(gcs.Config.Bucket).Object(path.Join(gcs.Config.Path, key)).Attrs(ctx)
+	return gcs.StatFileAbsolute(ctx, path.Join(gcs.Config.Path, key))
+}
+
+func (gcs *GCS) StatFileAbsolute(ctx context.Context, key string) (RemoteFile, error) {
+	objAttr, err := gcs.client.Bucket(gcs.Config.Bucket).Object(key).Attrs(ctx)
 	if err != nil {
 		if errors.Is(err, storage.ErrObjectNotExist) {
 			return nil, ErrNotFound

pkg/storage/object_disk/object_disk.go

@@ -727,6 +727,11 @@ func CopyObject(ctx context.Context, diskName string, srcSize int64, srcBucket,
 }
 
 func CopyObjectStreaming(ctx context.Context, srcStorage storage.RemoteStorage, dstStorage storage.RemoteStorage, srcKey, dstKey string) error {
+	srcInfo, statErr := srcStorage.StatFileAbsolute(ctx, srcKey)
+	if statErr != nil {
+		return fmt.Errorf("srcStorage.StatFileReaderAbsolute(%s) error: %v", srcKey, statErr)
+	}
+
 	srcReader, srcErr := srcStorage.GetFileReaderAbsolute(ctx, srcKey)
 	if srcErr != nil {
 		return fmt.Errorf("srcStorage.GetFileReaderAbsolute(%s) error: %v", srcKey, srcErr)
@@ -736,7 +741,7 @@ func CopyObjectStreaming(ctx context.Context, srcStorage storage.RemoteStorage,
 			log.Error().Msgf("srcReader.Close(%s) error: %v", srcKey, closeErr)
 		}
 	}()
-	if putErr := dstStorage.PutFileAbsolute(ctx, dstKey, srcReader, 0); putErr != nil {
+	if putErr := dstStorage.PutFileAbsolute(ctx, dstKey, srcReader, srcInfo.Size()); putErr != nil {
 		return fmt.Errorf("dstStorage.PutFileAbsolute(%s) error: %v", dstKey, putErr)
 	}
 	return nil

pkg/storage/s3.go

@@ -405,9 +405,13 @@ func (s *S3) getObjectAllVersions(ctx context.Context, key string) ([]string, er
 }
 
 func (s *S3) StatFile(ctx context.Context, key string) (RemoteFile, error) {
+	return s.StatFileAbsolute(ctx, path.Join(s.Config.Path, key))
+}
+
+func (s *S3) StatFileAbsolute(ctx context.Context, key string) (RemoteFile, error) {
 	params := &s3.HeadObjectInput{
 		Bucket: aws.String(s.Config.Bucket),
-		Key:    aws.String(path.Join(s.Config.Path, key)),
+		Key:    aws.String(key),
 	}
 	s.enrichHeadParams(params)
 	head, err := s.client.HeadObject(ctx, params)

pkg/storage/sftp.go

@@ -100,11 +100,13 @@ func (sftp *SFTP) Close(ctx context.Context) error {
 }
 
 func (sftp *SFTP) StatFile(ctx context.Context, key string) (RemoteFile, error) {
-	filePath := path.Join(sftp.Config.Path, key)
+	return sftp.StatFileAbsolute(ctx, path.Join(sftp.Config.Path, key))
+}
 
-	stat, err := sftp.sftpClient.Stat(filePath)
+func (sftp *SFTP) StatFileAbsolute(ctx context.Context, key string) (RemoteFile, error) {
+	stat, err := sftp.sftpClient.Stat(key)
 	if err != nil {
-		sftp.Debug("[SFTP_DEBUG] StatFile::STAT %s return error %v", filePath, err)
+		sftp.Debug("[SFTP_DEBUG] StatFile::STAT %s return error %v", key, err)
 		if strings.Contains(err.Error(), "not exist") {
 			return nil, ErrNotFound
 		}

pkg/storage/structs.go

@@ -25,6 +25,7 @@ type RemoteStorage interface {
 	Connect(ctx context.Context) error
 	Close(ctx context.Context) error
 	StatFile(ctx context.Context, key string) (RemoteFile, error)
+	StatFileAbsolute(ctx context.Context, key string) (RemoteFile, error)
 	DeleteFile(ctx context.Context, key string) error
 	DeleteFileFromObjectDiskBackup(ctx context.Context, key string) error
 	Walk(ctx context.Context, prefix string, recursive bool, fn func(context.Context, RemoteFile) error) error