Merge pull request #1310 from xavierleune/master

fix S3 multipart operations to respect S3_MAX_PARTS_COUNT

Author: Slach

ChangeLog.md

@@ -1,3 +1,7 @@
+# v2.6.42
+BUG FIXES
+- fix S3 multipart operations (upload, download, copy) to respect `S3_MAX_PARTS_COUNT` instead of hardcoded 10000 value, allows S3-compatible backends with stricter limits
+
 # v2.6.41
 BUG FIXES
 - improve restore long RBAC which have length more 64k, fix [1305](https://github.com/Altinity/clickhouse-backup/issues/1305) 

pkg/backup/backuper.go

@@ -372,9 +372,9 @@ func (b *Backuper) getObjectDiskPath() (string, error) {
 		return b.cfg.FTP.ObjectDiskPath, nil
 	} else if b.cfg.General.RemoteStorage == "sftp" {
 		return b.cfg.SFTP.ObjectDiskPath, nil
-	} else {
-		return "", fmt.Errorf("cleanBackupObjectDisks: requesst object disks path but have unsupported remote_storage: %s", b.cfg.General.RemoteStorage)
 	}
+
+	return "", errors.WithStack(fmt.Errorf("cleanBackupObjectDisks: requesst object disks path but have unsupported remote_storage: %s", b.cfg.General.RemoteStorage))
 }
 
 func (b *Backuper) getTablesDiffFromLocal(ctx context.Context, diffFrom string, tablePattern string) (tablesForUploadFromDiff map[metadata.TableTitle]metadata.TableMetadata, err error) {

pkg/backup/upload.go

@@ -89,9 +89,9 @@ func (b *Backuper) Upload(backupName string, deleteSource bool, diffFrom, diffFr
 		if backupName == remoteBackups[i].BackupName {
 			if !b.resume {
 				return fmt.Errorf("'%s' already exists on remote storage", backupName)
-			} else {
-				log.Warn().Msgf("'%s' already exists on remote, will try to resume upload", backupName)
 			}
+
+			log.Warn().Msgf("'%s' already exists on remote, will try to resume upload", backupName)
 		}
 	}
 	backupMetadata, err := b.ReadBackupMetadataLocal(ctx, backupName)
@@ -582,14 +582,16 @@ func (b *Backuper) uploadTableData(ctx context.Context, backupName string, delet
 						}
 					}
 					log.Debug().Msgf("start upload %d files to %s", len(partFiles), remotePath)
-					if uploadPathBytes, err := b.dst.UploadPath(ctx, backupPath, partFiles, remotePath, b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter, b, b.cfg.General.UploadMaxBytesPerSecond); err != nil {
+					var uploadPathBytes int64
+					var err error
+					if uploadPathBytes, err = b.dst.UploadPath(ctx, backupPath, partFiles, remotePath, b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter, b, b.cfg.General.UploadMaxBytesPerSecond); err != nil {
 						log.Error().Msgf("UploadPath return error: %v", err)
 						return errors.Wrap(err, "can't upload")
-					} else {
-						atomic.AddInt64(&uploadedBytes, uploadPathBytes)
-						if b.resume {
-							b.resumableState.AppendToState(remotePathFull, uploadPathBytes)
-						}
+					}
+
+					atomic.AddInt64(&uploadedBytes, uploadPathBytes)
+					if b.resume {
+						b.resumableState.AppendToState(remotePathFull, uploadPathBytes)
 					}
 					// https://github.com/Altinity/clickhouse-backup/issues/777
 					if deleteSource {
@@ -659,12 +661,16 @@ func (b *Backuper) uploadTableData(ctx context.Context, backupName string, delet
 
 func (b *Backuper) uploadTableMetadata(ctx context.Context, backupName string, requiredBackupName string, tableMetadata *metadata.TableMetadata) (int64, error) {
 	if b.isEmbedded {
-		if sqlSize, err := b.uploadTableMetadataEmbedded(ctx, backupName, requiredBackupName, *tableMetadata); err != nil {
-			return sqlSize, err
-		} else {
-			jsonSize, err := b.uploadTableMetadataRegular(ctx, backupName, *tableMetadata)
-			return sqlSize + jsonSize, err
+		var sqlSize, jsonSize int64
+		var err error
+		if sqlSize, err = b.uploadTableMetadataEmbedded(ctx, backupName, requiredBackupName, *tableMetadata); err != nil {
+			return sqlSize, errors.WithStack(err)
 		}
+		jsonSize, err = b.uploadTableMetadataRegular(ctx, backupName, *tableMetadata)
+		if err != nil {
+			err = errors.WithStack(err)
+		}
+		return sqlSize + jsonSize, err
 	}
 	return b.uploadTableMetadataRegular(ctx, backupName, *tableMetadata)
 }
@@ -711,11 +717,11 @@ func (b *Backuper) uploadTableMetadataEmbedded(ctx context.Context, backupName s
 	if err != nil {
 		err = errors.Wrapf(err, "can't open %s", localTableMetaFile)
 		if requiredBackupName != "" {
-			log.Warn().Err(err).Send()
+			log.Warn().Stack().Err(err).Send()
 			return 0, nil
-		} else {
-			return 0, err
 		}
+
+		return 0, err
 	}
 	defer func() {
 		if closeErr := localReader.Close(); closeErr != nil {
@@ -812,9 +818,8 @@ bodyRead:
 func (b *Backuper) splitPartFiles(basePath string, parts []metadata.Part, database, table string, skipProjections []string) ([]metadata.SplitPartFiles, error) {
 	if b.cfg.General.UploadByPart {
 		return b.splitFilesByName(basePath, parts, database, table, skipProjections)
-	} else {
-		return b.splitFilesBySize(basePath, parts, database, table, skipProjections)
 	}
+	return b.splitFilesBySize(basePath, parts, database, table, skipProjections)
 }
 
 func (b *Backuper) splitFilesByName(basePath string, parts []metadata.Part, database, table string, skipProjections []string) ([]metadata.SplitPartFiles, error) {

pkg/storage/s3.go

@@ -258,7 +258,7 @@ func (s *S3) GetFileReaderWithLocalPath(ctx context.Context, key, localPath stri
 		downloader.Concurrency = s.Concurrency
 		downloader.BufferProvider = s3manager.NewPooledBufferedWriterReadFromProvider(s.BufferSize)
 		var partSize int64
-		if s.Config.ChunkSize > 0 && (remoteSize+s.Config.ChunkSize-1)/s.Config.ChunkSize < 10000 {
+		if s.Config.ChunkSize > 0 && (remoteSize+s.Config.ChunkSize-1)/s.Config.ChunkSize < s.Config.MaxPartsCount {
 			// Use configured chunk size
 			partSize = s.Config.ChunkSize
 		} else {
@@ -334,7 +334,7 @@ func (s *S3) PutFileAbsolute(ctx context.Context, key string, r io.ReadCloser, l
 	uploader.Concurrency = s.Concurrency
 	uploader.BufferProvider = s3manager.NewBufferedReadSeekerWriteToPool(s.BufferSize)
 	var partSize int64
-	if s.Config.ChunkSize > 0 && (localSize+s.Config.ChunkSize-1)/s.Config.ChunkSize < 10000 {
+	if s.Config.ChunkSize > 0 && (localSize+s.Config.ChunkSize-1)/s.Config.ChunkSize < s.Config.MaxPartsCount {
 		partSize = s.Config.ChunkSize
 	} else {
 		partSize = localSize / s.Config.MaxPartsCount
@@ -546,7 +546,7 @@ func (s *S3) CopyObject(ctx context.Context, srcSize int64, srcBucket, srcKey, d
 
 	// Set the part size (128 MB minimum for CopyObject, or use configured chunk size)
 	var partSize int64
-	if s.Config.ChunkSize > 0 && (srcSize+s.Config.ChunkSize-1)/s.Config.ChunkSize < 10000 {
+	if s.Config.ChunkSize > 0 && (srcSize+s.Config.ChunkSize-1)/s.Config.ChunkSize < s.Config.MaxPartsCount {
 		partSize = s.Config.ChunkSize
 	} else {
 		partSize = srcSize / s.Config.MaxPartsCount