Merge branch 'master' of github.com:Altinity/clickhouse-backup into bug1291

Author: Slach

.gitignore

@@ -1,18 +1,19 @@
-.vscode
+.vscode/
 *.exe
 debug
 /clickhouse-backup
 credentials.json
-.idea
+.idea/
 *.iml
 build/
 .env
 *.log
+vendor/
+.aider*
+AGENTS.md
 
 # TestFlows tests artifacts that shall not be stored in Git
 _instances/
 _coverage_/
 __pycache__/
-*.py[cod]
-vendor/
-.aider*
+*.py[cod]

pkg/backup/backup_shard.go

@@ -2,10 +2,11 @@ package backup
 
 import (
 	"context"
-	"errors"
 	"fmt"
-	"github.com/rs/zerolog/log"
 	"hash/fnv"
+
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog/log"
 )
 
 var (
@@ -170,7 +171,7 @@ func (rd *replicaDeterminer) getReplicaState(ctx context.Context) ([]tableReplic
 	// TODO: Change query to pull replica_is_active after upgrading to clickhouse-go v2
 	query := "SELECT t.database, t.name AS table, r.replica_name, arraySort(mapKeys(mapFilter((replica, active) -> (active == 1), r.replica_is_active))) AS active_replicas FROM system.tables t LEFT JOIN system.replicas r ON t.database = r.database AND t.name = r.table"
 	if err := rd.q.SelectContext(ctx, &md, query); err != nil {
-		return nil, fmt.Errorf("could not determine replication state: %w", err)
+		return nil, errors.Wrap(err, "could not determine replication state")
 	}
 
 	// Handle views and memory tables by putting in stand-in replication metadata

pkg/backup/backuper.go

@@ -4,18 +4,19 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/binary"
-	"errors"
 	"fmt"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/common"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/metadata"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/utils"
-	"github.com/eapache/go-resiliency/retrier"
 	"net/url"
 	"os"
 	"path"
 	"regexp"
 	"strings"
 
+	"github.com/Altinity/clickhouse-backup/v2/pkg/common"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/metadata"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/utils"
+	"github.com/eapache/go-resiliency/retrier"
+	"github.com/pkg/errors"
+
 	"github.com/Altinity/clickhouse-backup/v2/pkg/clickhouse"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/resumable"
@@ -117,7 +118,7 @@ func (b *Backuper) initDisksPathsAndBackupDestination(ctx context.Context, disks
 			return err
 		}
 		if err := b.dst.Connect(ctx); err != nil {
-			return fmt.Errorf("can't connect to %s: %v", b.dst.Kind(), err)
+			return errors.Wrapf(err, "can't connect to %s", b.dst.Kind())
 		}
 	}
 	return nil
@@ -166,7 +167,7 @@ func (b *Backuper) populateBackupShardField(ctx context.Context, tables []clickh
 		// Parse shard config here to avoid error return in NewBackuper
 		shardFunc, err := shardFuncByName(b.cfg.General.ShardedOperationMode)
 		if err != nil {
-			return fmt.Errorf("could not determine shards for tables: %w", err)
+			return errors.Wrap(err, "could not determine shards for tables")
 		}
 		b.bs = newReplicaDeterminer(b.ch, shardFunc)
 	}
@@ -417,7 +418,7 @@ func (b *Backuper) getTablesDiffFromRemote(ctx context.Context, diffFromRemote s
 	tablesForUploadFromDiff = make(map[metadata.TableTitle]metadata.TableMetadata)
 	backupList, err := b.dst.BackupList(ctx, true, diffFromRemote)
 	if err != nil {
-		return nil, fmt.Errorf("b.dst.BackupList return error: %v", err)
+		return nil, errors.Wrap(err, "b.dst.BackupList return error")
 	}
 	var diffRemoteMetadata *metadata.BackupMetadata
 	for _, backup := range backupList {
@@ -433,7 +434,7 @@ func (b *Backuper) getTablesDiffFromRemote(ctx context.Context, diffFromRemote s
 	if len(diffRemoteMetadata.Tables) != 0 {
 		diffTablesList, tableListErr := getTableListByPatternRemote(ctx, b, diffRemoteMetadata, tablePattern, false)
 		if tableListErr != nil {
-			return nil, fmt.Errorf("getTableListByPatternRemote return error: %v", tableListErr)
+			return nil, errors.Wrap(tableListErr, "getTableListByPatternRemote return error")
 		}
 		for _, t := range diffTablesList {
 			tablesForUploadFromDiff[metadata.TableTitle{
@@ -537,7 +538,7 @@ func (b *Backuper) calculateChecksum(disk *clickhouse.Disk, partName string) (ui
 	checksumsFilePath := path.Join(disk.Path, partName, "checksums.txt")
 	content, err := os.ReadFile(checksumsFilePath)
 	if err != nil {
-		return 0, fmt.Errorf("could not read %s: %w", checksumsFilePath, err)
+		return 0, errors.Wrapf(err, "could not read %s", checksumsFilePath)
 	}
 
 	hash := sha256.Sum256(content)

pkg/backup/create.go

@@ -3,7 +3,6 @@ package backup
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"os"
 	"path"
@@ -17,6 +16,7 @@ import (
 	"github.com/Altinity/clickhouse-backup/v2/pkg/pidlock"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/resumable"
 	"github.com/eapache/go-resiliency/retrier"
+	"github.com/pkg/errors"
 
 	"github.com/google/uuid"
 	recursiveCopy "github.com/otiai10/copy"
@@ -78,7 +78,7 @@ func (b *Backuper) CreateBackup(backupName, diffFromRemote, tablePattern string,
 	backupName = utils.CleanBackupNameRE.ReplaceAllString(backupName, "")
 
 	if err := b.ch.Connect(); err != nil {
-		return fmt.Errorf("can't connect to clickhouse: %v", err)
+		return errors.Wrap(err, "can't connect to clickhouse")
 	}
 	defer b.ch.Close()
 
@@ -106,11 +106,11 @@ func (b *Backuper) CreateBackup(backupName, diffFromRemote, tablePattern string,
 
 	allDatabases, err := b.ch.GetDatabases(ctx, b.cfg, tablePattern)
 	if err != nil {
-		return fmt.Errorf("can't get database engines from clickhouse: %v", err)
+		return errors.Wrap(err, "can't get database engines from clickhouse")
 	}
 	tables, err := b.GetTables(ctx, tablePattern)
 	if err != nil {
-		return fmt.Errorf("can't get tables from clickhouse: %v", err)
+		return errors.Wrap(err, "can't get tables from clickhouse")
 	}
 
 	if b.CalculateNonSkipTables(tables) == 0 && !b.cfg.General.AllowEmptyBackups {
@@ -119,7 +119,7 @@ func (b *Backuper) CreateBackup(backupName, diffFromRemote, tablePattern string,
 
 	allFunctions, err := b.ch.GetUserDefinedFunctions(ctx)
 	if err != nil {
-		return fmt.Errorf("GetUserDefinedFunctions return error: %v", err)
+		return errors.Wrap(err, "GetUserDefinedFunctions return error")
 	}
 
 	disks, err := b.ch.GetDisks(ctx, false)
@@ -274,7 +274,7 @@ func (b *Backuper) createBackupLocal(ctx context.Context, backupName, diffFromRe
 			return err
 		}
 		if err = b.dst.Connect(ctx); err != nil {
-			return fmt.Errorf("can't connect to %s: %v", b.dst.Kind(), err)
+			return errors.Wrapf(err, "can't connect to %s", b.dst.Kind())
 		}
 		defer func() {
 			if closeErr := b.dst.Close(ctx); closeErr != nil {
@@ -297,7 +297,7 @@ func (b *Backuper) createBackupLocal(ctx context.Context, backupName, diffFromRe
 		var diffFromRemoteErr error
 		tablesDiffFromRemote, diffFromRemoteErr = b.getTablesDiffFromRemote(ctx, diffFromRemote, tablePattern)
 		if diffFromRemoteErr != nil {
-			return fmt.Errorf("b.getTablesDiffFromRemote return error: %v", diffFromRemoteErr)
+			return errors.Wrap(diffFromRemoteErr, "b.getTablesDiffFromRemote return error")
 		}
 	}
 
@@ -378,12 +378,12 @@ func (b *Backuper) createBackupLocal(ctx context.Context, backupName, diffFromRe
 		})
 	}
 	if wgWaitErr := createBackupWorkingGroup.Wait(); wgWaitErr != nil {
-		return fmt.Errorf("one of createBackupLocal go-routine return error: %v", wgWaitErr)
+		return errors.Wrap(wgWaitErr, "one of createBackupLocal go-routine return error")
 	}
 
 	backupMetaFile := path.Join(b.DefaultDataPath, "backup", backupName, "metadata.json")
 	if err := b.createBackupMetadata(ctx, backupMetaFile, backupName, diffFromRemote, backupVersion, "regular", diskMap, diskTypes, disks, backupDataSize, backupObjectDiskSize, backupMetadataSize, backupRBACSize, backupConfigSize, backupNamedCollectionsSize, tableMetas, allDatabases, allFunctions); err != nil {
-		return fmt.Errorf("createBackupMetadata return error: %v", err)
+		return errors.Wrap(err, "createBackupMetadata return error")
 	}
 	log.Info().Str("version", backupVersion).Str("operation", "createBackupLocal").Str("duration", utils.HumanizeDuration(time.Since(startBackup))).Msg("done")
 	return nil
@@ -429,7 +429,7 @@ func (b *Backuper) createBackupEmbedded(ctx context.Context, backupName, baseBac
 		}
 		backupResult := make([]clickhouse.SystemBackups, 0)
 		if err := b.ch.SelectContext(ctx, &backupResult, backupSQL); err != nil {
-			return fmt.Errorf("backup error: %v", err)
+			return errors.Wrap(err, "backup error")
 		}
 		if len(backupResult) != 1 || (backupResult[0].Status != "BACKUP_COMPLETE" && backupResult[0].Status != "BACKUP_CREATED") {
 			return fmt.Errorf("backup return wrong results: %+v", backupResult)
@@ -474,7 +474,7 @@ func (b *Backuper) createBackupEmbedded(ctx context.Context, backupName, baseBac
 				return err
 			}
 			if err = b.dst.Connect(ctx); err != nil {
-				return fmt.Errorf("createBackupEmbedded: can't connect to %s: %v", b.dst.Kind(), err)
+				return errors.Wrapf(err, "createBackupEmbedded: can't connect to %s", b.dst.Kind())
 			}
 			defer func() {
 				if closeErr := b.dst.Close(ctx); closeErr != nil {
@@ -721,11 +721,11 @@ func (b *Backuper) createBackupRBAC(ctx context.Context, backupPath string, disk
 		rbacBackup := path.Join(backupPath, "access")
 		replicatedRBACDataSize, err := b.createBackupRBACReplicated(ctx, rbacBackup)
 		if err != nil {
-			return 0, fmt.Errorf("b.createBackupRBACReplicated error: %v", err)
+			return 0, errors.Wrap(err, "b.createBackupRBACReplicated error")
 		}
 		accessPath, err := b.ch.GetAccessManagementPath(ctx, disks)
 		if err != nil {
-			return 0, fmt.Errorf("b.ch.GetAccessManagementPath error: %v", err)
+			return 0, errors.Wrap(err, "b.ch.GetAccessManagementPath error")
 		}
 		accessPathInfo, err := os.Stat(accessPath)
 		if err != nil && !os.IsNotExist(err) {
@@ -851,7 +851,7 @@ func (b *Backuper) createBackupRBACReplicated(ctx context.Context, rbacBackup st
 		for _, userDirectory := range replicatedRBAC {
 			replicatedAccessPath, err := k.GetReplicatedAccessPath(userDirectory.Name)
 			if err != nil {
-				return 0, fmt.Errorf("k.GetReplicatedAccessPath(%s) error: %v", userDirectory.Name, err)
+				return 0, errors.Wrapf(err, "k.GetReplicatedAccessPath(%s) error", userDirectory.Name)
 			}
 			rbacUUIDObjectsCount, err := k.ChildCount(replicatedAccessPath, "uuid")
 			if err != nil {
@@ -1063,7 +1063,7 @@ func (b *Backuper) uploadObjectDiskParts(ctx context.Context, backupName string,
 						return nil
 					})
 					if copyObjectErr != nil {
-						return fmt.Errorf("b.dst.CopyObject in %s error: %v", backupShadowPath, copyObjectErr)
+						return errors.Wrapf(copyObjectErr, "b.dst.CopyObject in %s error", backupShadowPath)
 					}
 				} else {
 					if !isCopyFailed.Load() {
@@ -1079,7 +1079,7 @@ func (b *Backuper) uploadObjectDiskParts(ctx context.Context, backupName string,
 							return object_disk.CopyObjectStreaming(uploadCtx, srcDiskConnection.GetRemoteStorage(), b.dst, srcKey, path.Join(objectDiskPath, dstKey))
 						})
 						if copyObjectErr != nil {
-							return fmt.Errorf("object_disk.CopyObjectStreaming in %s error: %v", backupShadowPath, copyObjectErr)
+							return errors.Wrapf(copyObjectErr, "object_disk.CopyObjectStreaming in %s error", backupShadowPath)
 						}
 					}
 					objSize = storageObject.ObjectSize
@@ -1103,7 +1103,7 @@ func (b *Backuper) uploadObjectDiskParts(ctx context.Context, backupName string,
 	}
 
 	if wgWaitErr := uploadObjectDiskPartsWorkingGroup.Wait(); wgWaitErr != nil {
-		return 0, fmt.Errorf("one of uploadObjectDiskParts go-routine return error: %v", wgWaitErr)
+		return 0, errors.Wrap(wgWaitErr, "one of uploadObjectDiskParts go-routine return error")
 	}
 	return size, nil
 }
@@ -1140,7 +1140,7 @@ func (b *Backuper) createBackupMetadata(ctx context.Context, backupMetaFile, bac
 		}
 		content, err := json.MarshalIndent(&backupMetadata, "", "\t")
 		if err != nil {
-			return fmt.Errorf("can't marshal backup metafile json: %v", err)
+			return errors.Wrap(err, "can't marshal backup metafile json")
 		}
 		if err := os.WriteFile(backupMetaFile, content, 0640); err != nil {
 			return err
@@ -1164,10 +1164,10 @@ func (b *Backuper) createTableMetadata(metadataPath string, table metadata.Table
 	metadataFile := path.Join(metadataDatabasePath, fmt.Sprintf("%s.json", common.TablePathEncode(table.Table)))
 	metadataBody, err := json.MarshalIndent(&table, "", " ")
 	if err != nil {
-		return 0, fmt.Errorf("can't marshal %s: %v", MetaFileName, err)
+		return 0, errors.Wrapf(err, "can't marshal %s", MetaFileName)
 	}
 	if err := os.WriteFile(metadataFile, metadataBody, 0644); err != nil {
-		return 0, fmt.Errorf("can't create %s: %v", MetaFileName, err)
+		return 0, errors.Wrapf(err, "can't create %s", MetaFileName)
 	}
 	if err := filesystemhelper.Chown(metadataFile, b.ch, disks, false); err != nil {
 		return 0, err

pkg/backup/delete.go

@@ -3,14 +3,15 @@ package backup
 import (
 	"context"
 	"fmt"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/pidlock"
 	"io/fs"
 	"os"
 	"path"
 	"path/filepath"
 	"strings"
 	"time"
 
+	"github.com/Altinity/clickhouse-backup/v2/pkg/pidlock"
+
 	"github.com/Altinity/clickhouse-backup/v2/pkg/clickhouse"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/custom"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/status"
@@ -25,7 +26,7 @@ import (
 // Clean - removed all data in shadow folder
 func (b *Backuper) Clean(ctx context.Context) error {
 	if err := b.ch.Connect(); err != nil {
-		return fmt.Errorf("can't connect to clickhouse: %v", err)
+		return errors.Wrap(err, "can't connect to clickhouse")
 	}
 	defer b.ch.Close()
 
@@ -39,7 +40,7 @@ func (b *Backuper) Clean(ctx context.Context) error {
 		}
 		shadowDir := path.Join(disk.Path, "shadow")
 		if err := b.cleanDir(shadowDir); err != nil {
-			return fmt.Errorf("can't clean '%s': %v", shadowDir, err)
+			return errors.Wrapf(err, "can't clean '%s'", shadowDir)
 		}
 		log.Info().Msg(shadowDir)
 	}
@@ -117,7 +118,7 @@ func (b *Backuper) RemoveBackupLocal(ctx context.Context, backupName string, dis
 	start := time.Now()
 	backupName = utils.CleanBackupNameRE.ReplaceAllString(backupName, "")
 	if err = b.ch.Connect(); err != nil {
-		return fmt.Errorf("can't connect to clickhouse: %v", err)
+		return errors.Wrap(err, "can't connect to clickhouse")
 	}
 	defer b.ch.Close()
 	if disks == nil {
@@ -142,7 +143,7 @@ func (b *Backuper) RemoveBackupLocal(ctx context.Context, backupName string, dis
 				}
 				err = bd.Connect(ctx)
 				if err != nil {
-					return fmt.Errorf("can't connect to remote storage: %v", err)
+					return errors.Wrap(err, "can't connect to remote storage")
 				}
 				defer func() {
 					if err := bd.Close(ctx); err != nil {
@@ -279,7 +280,7 @@ func (b *Backuper) RemoveBackupRemote(ctx context.Context, backupName string) er
 		return custom.DeleteRemote(ctx, b.cfg, backupName)
 	}
 	if err := b.ch.Connect(); err != nil {
-		return fmt.Errorf("can't connect to clickhouse: %v", err)
+		return errors.Wrap(err, "can't connect to clickhouse")
 	}
 	defer b.ch.Close()
 
@@ -289,7 +290,7 @@ func (b *Backuper) RemoveBackupRemote(ctx context.Context, backupName string) er
 	}
 	err = bd.Connect(ctx)
 	if err != nil {
-		return fmt.Errorf("can't connect to remote storage: %v", err)
+		return errors.Wrap(err, "can't connect to remote storage")
 	}
 	defer func() {
 		if err := bd.Close(ctx); err != nil {
@@ -472,14 +473,14 @@ func (b *Backuper) cleanPartialRequiredBackup(ctx context.Context, disks []click
 		for _, localBackup := range localBackups {
 			if localBackup.BackupName != currentBackupName && localBackup.DataSize+localBackup.CompressedSize+localBackup.MetadataSize+localBackup.RBACSize == 0 {
 				if err = b.RemoveBackupLocal(ctx, localBackup.BackupName, disks); err != nil {
-					return fmt.Errorf("CleanPartialRequiredBackups %s -> RemoveBackupLocal cleaning error: %v", localBackup.BackupName, err)
+					return errors.Wrapf(err, "CleanPartialRequiredBackups %s -> RemoveBackupLocal cleaning error", localBackup.BackupName)
 				} else {
 					log.Info().Msgf("CleanPartialRequiredBackups %s deleted", localBackup.BackupName)
 				}
 			}
 		}
 	} else {
-		return fmt.Errorf("CleanPartialRequiredBackups -> GetLocalBackups cleaning error: %v", err)
+		return errors.Wrap(err, "CleanPartialRequiredBackups -> GetLocalBackups cleaning error")
 	}
 	return nil
 }