From d7c0d4defd21306007a959fc1c43e47165e95063 Mon Sep 17 00:00:00 2001
From: Slach <bloodjazman@gmail.com>
Date: Mon, 24 Jul 2023 12:55:01 +0500
Subject: [PATCH] move object_disk_path validation to `create` and `restore` to
 avoid unnecessary failures, fix testflows

---
 pkg/backup/create.go |  4 ++++
 pkg/config/config.go | 17 ++++++++++-------
 2 files changed, 14 insertions(+), 7 deletions(-)

diff --git a/pkg/backup/create.go b/pkg/backup/create.go
index 08067c66..ea4f1c74 100644
--- a/pkg/backup/create.go
+++ b/pkg/backup/create.go
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/Altinity/clickhouse-backup/pkg/config"
 	"github.com/Altinity/clickhouse-backup/pkg/partition"
 	"github.com/Altinity/clickhouse-backup/pkg/status"
 	"github.com/Altinity/clickhouse-backup/pkg/storage"
@@ -510,6 +511,9 @@ func (b *Backuper) AddTableToBackup(ctx context.Context, backupName, shadowBacku
 			disksToPartsMap[disk.Name] = parts
 			log.WithField("disk", disk.Name).Debug("shadow moved")
 			if disk.Type == "s3" || disk.Type == "azure_blob_storage" && len(parts) > 0 {
+				if err = config.ValidateObjectDiskConfig(b.cfg); err != nil {
+					return nil, nil, err
+				}
 				start := time.Now()
 				if b.dst == nil {
 					b.dst, err = storage.NewBackupDestination(ctx, b.cfg, b.ch, false, backupName)
diff --git a/pkg/config/config.go b/pkg/config/config.go
index f420f320..0a16c084 100644
--- a/pkg/config/config.go
+++ b/pkg/config/config.go
@@ -418,20 +418,23 @@ func ValidateConfig(cfg *Config) error {
 			cfg.General.FullDuration = duration
 		}
 	}
-	// @TODO add all other storage types
+	return nil
+}
+
+func ValidateObjectDiskConfig(cfg *Config) error {
 	if !cfg.ClickHouse.UseEmbeddedBackupRestore {
 		switch cfg.General.RemoteStorage {
 		case "s3":
-			if cfg.S3.ObjectDiskPath == "" || strings.HasPrefix(cfg.S3.Path, cfg.S3.ObjectDiskPath) {
-				return fmt.Errorf("invalid s3->object_disk_path, shall be not empty and shall not be prefix for s3->path")
+			if cfg.S3.Path != "" && (cfg.S3.ObjectDiskPath == "" || strings.HasPrefix(cfg.S3.Path, cfg.S3.ObjectDiskPath)) {
+				return fmt.Errorf("data in objects disks, invalid s3->object_disk_path config section, shall be not empty and shall not be prefix for s3->path")
 			}
 		case "gcs":
-			if cfg.GCS.ObjectDiskPath == "" || strings.HasPrefix(cfg.GCS.Path, cfg.GCS.ObjectDiskPath) {
-				return fmt.Errorf("invalid gcs->object_disk_path, shall be not empty and shall not be prefix for gcs->path")
+			if cfg.GCS.Path != "" && (cfg.GCS.ObjectDiskPath == "" || strings.HasPrefix(cfg.GCS.Path, cfg.GCS.ObjectDiskPath)) {
+				return fmt.Errorf("data in objects disks, invalid gcs->object_disk_path config section, shall be not empty and shall not be prefix for gcs->path")
 			}
 		case "azblob":
-			if cfg.AzureBlob.ObjectDiskPath == "" || strings.HasPrefix(cfg.AzureBlob.Path, cfg.AzureBlob.ObjectDiskPath) {
-				return fmt.Errorf("invalid azblob->object_disk_path, shall be not empty and shall not be prefix for gcs->path")
+			if cfg.AzureBlob.Path != "" && (cfg.AzureBlob.ObjectDiskPath == "" || strings.HasPrefix(cfg.AzureBlob.Path, cfg.AzureBlob.ObjectDiskPath)) {
+				return fmt.Errorf("data in objects disks, invalid azblob->object_disk_path config section, shall be not empty and shall not be prefix for gcs->path")
 			}
 		}
 	}