change retries from constant to exponential backoff and add RETRIES_JITTER configuration option, to avoid same time retries from parallel operation

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

Author: Slach

ChangeLog.md

@@ -1,3 +1,7 @@
+# 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
+
 # v2.6.24
 IMPROVEMENTS
 - add logs for retries, to allow figure out with blackbaze2 s3 compatible provider, rate limit errors

ReadMe.md

@@ -141,6 +141,7 @@ general:
 
   retries_on_failure: 3          # RETRIES_ON_FAILURE, how many times to retry after a failure during upload or download
   retries_pause: 5s              # RETRIES_PAUSE, duration time to pause after each download or upload failure
+  retries_jitter: 30             # RETRIES_JITTER, percent of RETRIES_PAUSE for jitter to avoid same time retries from parallel operations 
 
   watch_interval: 1h       # WATCH_INTERVAL, use only for `watch` command, backup will create every 1h
   full_interval: 24h       # FULL_INTERVAL, use only for `watch` command, full backup will create every 24h

pkg/backup/create.go

@@ -922,7 +922,7 @@ func (b *Backuper) uploadObjectDiskParts(ctx context.Context, backupName string,
 				}
 				dstKey := path.Join(backupName, disk.Name, storageObject.ObjectRelativePath)
 				if !b.cfg.General.AllowObjectDiskStreaming {
-					retry := retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+					retry := retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 					copyObjectErr = retry.RunCtx(uploadCtx, func(ctx context.Context) error {
 						if objSize, err = b.dst.CopyObject(ctx, storageObject.ObjectSize, srcBucket, srcKey, dstKey); err != nil {
 							return err
@@ -941,7 +941,7 @@ func (b *Backuper) uploadObjectDiskParts(ctx context.Context, backupName string,
 						}
 					}
 					if isCopyFailed.Load() {
-						retry := retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+						retry := retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 						copyObjectErr = retry.RunCtx(uploadCtx, func(ctx context.Context) error {
 							return object_disk.CopyObjectStreaming(uploadCtx, srcDiskConnection.GetRemoteStorage(), b.dst, srcKey, path.Join(objectDiskPath, dstKey))
 						})

pkg/backup/download.go

@@ -439,7 +439,7 @@ func (b *Backuper) downloadTableMetadata(ctx context.Context, backupName string,
 			}
 		}
 		var tmBody []byte
-		retry := retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+		retry := retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 		err := retry.RunCtx(ctx, func(ctx context.Context) error {
 			tmReader, err := b.dst.GetFileReader(ctx, remoteMetadataFile)
 			if err != nil {
@@ -573,7 +573,7 @@ func (b *Backuper) downloadBackupRelatedDir(ctx context.Context, remoteBackup st
 	var downloadErr error
 	downloadedBytes := int64(0)
 	if remoteBackup.DataFormat == DirectoryFormat {
-		if downloadedBytes, downloadErr = b.dst.DownloadPath(ctx, remoteSource, localDir, b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration, b, b.cfg.General.DownloadMaxBytesPerSecond); downloadErr != nil {
+		if downloadedBytes, downloadErr = b.dst.DownloadPath(ctx, remoteSource, localDir, b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter, b, b.cfg.General.DownloadMaxBytesPerSecond); downloadErr != nil {
 			//SFTP can't walk on non exists paths and return error
 			if !strings.Contains(downloadErr.Error(), "not exist") {
 				return 0, downloadErr
@@ -592,7 +592,7 @@ func (b *Backuper) downloadBackupRelatedDir(ctx context.Context, remoteBackup st
 		log.Debug().Msgf("%s not exists on remote storage, skip download", remoteSource)
 		return 0, nil
 	}
-	retry := retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+	retry := retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 	err = retry.RunCtx(ctx, func(ctx context.Context) error {
 		downloadedBytes, downloadErr = b.dst.DownloadCompressedStream(ctx, remoteSource, localDir, b.cfg.General.DownloadMaxBytesPerSecond)
 		return downloadErr
@@ -643,7 +643,7 @@ func (b *Backuper) downloadTableData(ctx context.Context, remoteBackup metadata.
 						}
 
 					}
-					retry := retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+					retry := retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 					var downloadedBytes int64
 					var downloadErr error
 					err := retry.RunCtx(dataCtx, func(dataCtx context.Context) error {
@@ -701,7 +701,7 @@ func (b *Backuper) downloadTableData(ctx context.Context, remoteBackup metadata.
 						}
 					}
 
-					pathSize, downloadErr := b.dst.DownloadPath(dataCtx, partRemotePath, partLocalPath, b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration, b, b.cfg.General.DownloadMaxBytesPerSecond)
+					pathSize, downloadErr := b.dst.DownloadPath(dataCtx, partRemotePath, partLocalPath, b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter, b, b.cfg.General.DownloadMaxBytesPerSecond)
 					if downloadErr != nil {
 						return downloadErr
 					}
@@ -879,7 +879,7 @@ func (b *Backuper) downloadDiffRemoteFile(ctx context.Context, diffRemoteFilesLo
 		namedLock.Lock()
 		diffRemoteFilesLock.Unlock()
 		if path.Ext(tableRemoteFile) != "" {
-			retry := retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+			retry := retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 			err := retry.RunCtx(ctx, func(ctx context.Context) error {
 				var downloadErr error
 				downloadedBytes, downloadErr = b.dst.DownloadCompressedStream(ctx, tableRemoteFile, tableLocalDir, b.cfg.General.DownloadMaxBytesPerSecond)
@@ -891,7 +891,7 @@ func (b *Backuper) downloadDiffRemoteFile(ctx context.Context, diffRemoteFilesLo
 			}
 		} else {
 			// remoteFile could be a directory
-			if pathSize, err := b.dst.DownloadPath(ctx, tableRemoteFile, tableLocalDir, b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration, b, b.cfg.General.DownloadMaxBytesPerSecond); err != nil {
+			if pathSize, err := b.dst.DownloadPath(ctx, tableRemoteFile, tableLocalDir, b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter, b, b.cfg.General.DownloadMaxBytesPerSecond); err != nil {
 				log.Warn().Msgf("DownloadPath %s -> %s return error: %v", tableRemoteFile, tableLocalDir, err)
 				return 0, err
 			} else {
@@ -1122,8 +1122,7 @@ func (b *Backuper) downloadSingleBackupFile(ctx context.Context, remoteFile stri
 			return size, nil
 		}
 	}
-	retry := retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
-
+	retry := retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 	err := retry.RunCtx(ctx, func(ctx context.Context) error {
 		remoteReader, err := b.dst.GetFileReader(ctx, remoteFile)
 		if err != nil {

pkg/backup/restore.go

@@ -1657,7 +1657,7 @@ func (b *Backuper) downloadObjectDiskParts(ctx context.Context, backupName strin
 							copiedSize := int64(0)
 							var copyObjectErr error
 							if !b.cfg.General.AllowObjectDiskStreaming {
-								retry := retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+								retry := retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 								copyObjectErr = retry.RunCtx(downloadCtx, func(ctx context.Context) error {
 									var retryErr error
 									copiedSize, retryErr = object_disk.CopyObject(downloadCtx, dstDiskName, storageObject.ObjectSize, srcBucket, srcKey, storageObject.ObjectRelativePath)
@@ -1684,7 +1684,7 @@ func (b *Backuper) downloadObjectDiskParts(ctx context.Context, backupName strin
 									}
 									dstStorage := dstConnection.GetRemoteStorage()
 									dstKey := path.Join(dstConnection.GetRemoteObjectDiskPath(), storageObject.ObjectRelativePath)
-									retry := retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+									retry := retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 									copyObjectErr = retry.RunCtx(downloadCtx, func(ctx context.Context) error {
 										return object_disk.CopyObjectStreaming(downloadCtx, srcStorage, dstStorage, srcKey, dstKey)
 									})

pkg/backup/upload.go

@@ -235,7 +235,7 @@ func (b *Backuper) Upload(backupName string, deleteSource bool, diffFrom, diffFr
 	}
 	remoteBackupMetaFile := path.Join(backupName, "metadata.json")
 	if !b.resume || (b.resume && !b.resumableState.IsAlreadyProcessedBool(remoteBackupMetaFile)) {
-		retry := retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+		retry := retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 		err = retry.RunCtx(ctx, func(ctx context.Context) error {
 			return b.dst.PutFile(ctx, remoteBackupMetaFile, io.NopCloser(bytes.NewReader(newBackupMetadataBody)), 0)
 		})
@@ -324,7 +324,7 @@ func (b *Backuper) uploadSingleBackupFile(ctx context.Context, localFile, remote
 			log.Warn().Msgf("can't close %v: %v", f, err)
 		}
 	}()
-	retry := retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+	retry := retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 	err = retry.RunCtx(ctx, func(ctx context.Context) error {
 		return b.dst.PutFile(ctx, remoteFile, f, 0)
 	})
@@ -460,15 +460,15 @@ func (b *Backuper) uploadBackupRelatedDir(ctx context.Context, localBackupRelate
 	}
 	if b.cfg.GetCompressionFormat() == "none" {
 		remoteUploadedBytes := int64(0)
-		if remoteUploadedBytes, err = b.dst.UploadPath(ctx, localBackupRelatedDir, localFiles, destinationRemote, b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration, b, b.cfg.General.UploadMaxBytesPerSecond); err != nil {
+		if remoteUploadedBytes, err = b.dst.UploadPath(ctx, localBackupRelatedDir, localFiles, destinationRemote, b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter, b, b.cfg.General.UploadMaxBytesPerSecond); err != nil {
 			return 0, fmt.Errorf("can't RBAC or config upload %s: %v", destinationRemote, err)
 		}
 		if b.resume {
 			b.resumableState.AppendToState(destinationRemote, remoteUploadedBytes)
 		}
 		return uint64(remoteUploadedBytes), nil
 	}
-	retry := retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+	retry := retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 	err = retry.RunCtx(ctx, func(ctx context.Context) error {
 		return b.dst.UploadCompressedStream(ctx, localBackupRelatedDir, localFiles, destinationRemote, b.cfg.General.UploadMaxBytesPerSecond)
 	})
@@ -477,7 +477,7 @@ func (b *Backuper) uploadBackupRelatedDir(ctx context.Context, localBackupRelate
 	}
 
 	var remoteUploaded storage.RemoteFile
-	retry = retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+	retry = retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 	err = retry.RunCtx(ctx, func(ctx context.Context) error {
 		remoteUploaded, err = b.dst.StatFile(ctx, destinationRemote)
 		return err
@@ -553,7 +553,7 @@ 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, b.cfg.General.UploadMaxBytesPerSecond); err != nil {
+					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 fmt.Errorf("can't upload: %v", err)
 					} else {
@@ -585,7 +585,7 @@ func (b *Backuper) uploadTableData(ctx context.Context, backupName string, delet
 						}
 					}
 					log.Debug().Msgf("start upload %d files to %s", len(localFiles), remoteDataFile)
-					retry := retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+					retry := retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 					err := retry.RunCtx(ctx, func(ctx context.Context) error {
 						return b.dst.UploadCompressedStream(ctx, backupPath, localFiles, remoteDataFile, b.cfg.General.UploadMaxBytesPerSecond)
 					})
@@ -595,7 +595,7 @@ func (b *Backuper) uploadTableData(ctx context.Context, backupName string, delet
 					}
 
 					var remoteFile storage.RemoteFile
-					retry = retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+					retry = retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 					err = retry.RunCtx(ctx, func(ctx context.Context) error {
 						remoteFile, err = b.dst.StatFile(ctx, remoteDataFile)
 						return err
@@ -651,7 +651,7 @@ func (b *Backuper) uploadTableMetadataRegular(ctx context.Context, backupName st
 			return processedSize, nil
 		}
 	}
-	retry := retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+	retry := retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 	err = retry.RunCtx(ctx, func(ctx context.Context) error {
 		return b.dst.PutFile(ctx, remoteTableMetaFile, io.NopCloser(bytes.NewReader(content)), 0)
 	})
@@ -696,7 +696,7 @@ func (b *Backuper) uploadTableMetadataEmbedded(ctx context.Context, backupName s
 			log.Warn().Msgf("can't close %v: %v", localReader, err)
 		}
 	}()
-	retry := retrier.New(retrier.ConstantBackoff(b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration), b)
+	retry := retrier.New(retrier.ExponentialBackoff(b.cfg.General.RetriesOnFailure, common.AddRandomJitter(b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter)), b)
 	err = retry.RunCtx(ctx, func(ctx context.Context) error {
 		return b.dst.PutFile(ctx, remoteTableMetaFile, localReader, 0)
 	})

pkg/common/common.go

@@ -1,9 +1,11 @@
 package common
 
 import (
+	"math/rand"
 	"net/url"
 	"reflect"
 	"strings"
+	"time"
 )
 
 func TablePathEncode(str string) string {
@@ -108,3 +110,12 @@ func deepEqual(a, b interface{}) bool {
 	// В остальных случаях используем reflect.DeepEqual
 	return reflect.DeepEqual(a, b)
 }
+
+func AddRandomJitter(duration time.Duration, jitterPercent int8) time.Duration {
+	if jitterPercent <= 0 {
+		return duration
+	}
+	maxJitter := duration * time.Duration(jitterPercent) / 100
+	jitter := time.Duration(rand.Int63n(int64(maxJitter + 1)))
+	return duration + jitter
+}

pkg/config/config.go

@@ -59,6 +59,7 @@ type GeneralConfig struct {
 	RestoreTableMapping                 map[string]string `yaml:"restore_table_mapping" envconfig:"RESTORE_TABLE_MAPPING"`
 	RetriesOnFailure                    int               `yaml:"retries_on_failure" envconfig:"RETRIES_ON_FAILURE"`
 	RetriesPause                        string            `yaml:"retries_pause" envconfig:"RETRIES_PAUSE"`
+	RetriesJitter                       int8              `yaml:"retries_jitter" envconfig:"RETRIES_JITTER"`
 	WatchInterval                       string            `yaml:"watch_interval" envconfig:"WATCH_INTERVAL"`
 	FullInterval                        string            `yaml:"full_interval" envconfig:"FULL_INTERVAL"`
 	WatchBackupNameTemplate             string            `yaml:"watch_backup_name_template" envconfig:"WATCH_BACKUP_NAME_TEMPLATE"`

pkg/custom/download_custom.go

@@ -3,6 +3,7 @@ package custom
 import (
 	"context"
 	"fmt"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/common"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/utils"
 	"github.com/eapache/go-resiliency/retrier"
@@ -34,7 +35,7 @@ func Download(ctx context.Context, retrierClassifier retrier.Classifier, cfg *co
 		"schema":        schemaOnly,
 	}
 	args := ApplyCommandTemplate(cfg.Custom.DownloadCommand, templateData)
-	retry := retrier.New(retrier.ConstantBackoff(cfg.General.RetriesOnFailure, cfg.General.RetriesDuration), retrierClassifier)
+	retry := retrier.New(retrier.ExponentialBackoff(cfg.General.RetriesOnFailure, common.AddRandomJitter(cfg.General.RetriesDuration, cfg.General.RetriesJitter)), retrierClassifier)
 	err := retry.RunCtx(ctx, func(ctx context.Context) error {
 		return utils.ExecCmd(ctx, cfg.Custom.CommandTimeoutDuration, args[0], args[1:]...)
 	})

pkg/custom/upload_custom.go

@@ -3,6 +3,7 @@ package custom
 import (
 	"context"
 	"fmt"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/common"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/utils"
 	"github.com/eapache/go-resiliency/retrier"
@@ -40,7 +41,7 @@ func Upload(ctx context.Context, retrierClassifier retrier.Classifier, cfg *conf
 		"schema":           schemaOnly,
 	}
 	args := ApplyCommandTemplate(cfg.Custom.UploadCommand, templateData)
-	retry := retrier.New(retrier.ConstantBackoff(cfg.General.RetriesOnFailure, cfg.General.RetriesDuration), retrierClassifier)
+	retry := retrier.New(retrier.ExponentialBackoff(cfg.General.RetriesOnFailure, common.AddRandomJitter(cfg.General.RetriesDuration, cfg.General.RetriesJitter)), retrierClassifier)
 	err := retry.RunCtx(ctx, func(ctx context.Context) error {
 		return utils.ExecCmd(ctx, cfg.Custom.CommandTimeoutDuration, args[0], args[1:]...)
 	})