Fix storage identification during inc backup

Author: boris-ilijic

cmd/pbm-agent/pitr.go

@@ -160,7 +160,7 @@ func canSlicingNow(ctx context.Context, conn connect.Client, stgCfg *config.Stor
 			return errors.Wrap(err, "get backup metadata")
 		}
 
-		if bcp.Type == defs.LogicalBackup && bcp.Store.Equal(stgCfg) {
+		if bcp.Type == defs.LogicalBackup && bcp.Store.IsSameStorage(stgCfg) {
 			return lock.ConcurrentOpError{l.LockHeader}
 		}
 	}

pbm/backup/physical.go

@@ -232,7 +232,7 @@ func (b *Backup) doPhysical(
 				}
 			}
 
-			if !b.config.Storage.Equal(&src.Store.StorageConf) {
+			if !b.config.Storage.IsSameStorage(&src.Store.StorageConf) {
 				return errors.New("cannot use the configured storage: " +
 					"source backup is stored on a different storage")
 			}

pbm/config/config.go

@@ -274,6 +274,32 @@ func (s *StorageConf) Equal(other *StorageConf) bool {
 	return false
 }
 
+// IsSameStorage returns true if specified config params describes
+// the same instance of the storage.
+// It ignores storage properties, and just compare parts that specifies
+// the storage instance.
+func (s *StorageConf) IsSameStorage(other *StorageConf) bool {
+	if s.Type != other.Type {
+		return false
+	}
+
+	switch s.Type {
+	case storage.S3:
+		return s.S3.IsSameStorage(other.S3)
+	case storage.Azure:
+		return s.Azure.IsSameStorage(other.Azure)
+	case storage.GCS:
+		return s.GCS.IsSameStorage(other.GCS)
+	case storage.Filesystem:
+		// FS is the same if it's equal
+		return s.Filesystem.Equal(other.Filesystem)
+	case storage.Blackhole:
+		return true
+	}
+
+	return false
+}
+
 func (s *StorageConf) Cast() error {
 	switch s.Type {
 	case storage.Filesystem:

pbm/storage/azure/azure.go

@@ -79,6 +79,24 @@ func (cfg *Config) Equal(other *Config) bool {
 	return true
 }
 
+// IsSameStorage identifies the same instance of the Azure storage.
+func (cfg *Config) IsSameStorage(other *Config) bool {
+	if cfg == nil || other == nil {
+		return cfg == other
+	}
+
+	if cfg.Account != other.Account {
+		return false
+	}
+	if cfg.Container != other.Container {
+		return false
+	}
+	if cfg.Prefix != other.Prefix {
+		return false
+	}
+	return true
+}
+
 // resolveEndpointURL returns endpoint url based on provided
 // EndpointURL or associated EndpointURLMap configuration fields.
 // If specified EndpointURLMap overrides EndpointURL field.

pbm/storage/gcs/gcs.go

@@ -108,6 +108,22 @@ func (cfg *Config) Equal(other *Config) bool {
 	return true
 }
 
+// IsSameStorage identifies the same instance of the GCS storage.
+func (cfg *Config) IsSameStorage(other *Config) bool {
+	if cfg == nil || other == nil {
+		return cfg == other
+	}
+
+	if cfg.Bucket != other.Bucket {
+		return false
+	}
+	if cfg.Prefix != other.Prefix {
+		return false
+	}
+
+	return true
+}
+
 func New(opts *Config, node string, l log.LogEvent) (*GCS, error) {
 	g := &GCS{
 		opts: opts,

pbm/storage/s3/s3.go

@@ -188,6 +188,24 @@ func (cfg *Config) Equal(other *Config) bool {
 	return true
 }
 
+// IsSameStorage identifies the same instance of the S3 storage.
+func (cfg *Config) IsSameStorage(other *Config) bool {
+	if cfg == nil || other == nil {
+		return cfg == other
+	}
+
+	if cfg.Region != other.Region {
+		return false
+	}
+	if cfg.Bucket != other.Bucket {
+		return false
+	}
+	if cfg.Prefix != other.Prefix {
+		return false
+	}
+	return true
+}
+
 func (cfg *Config) Cast() error {
 	if cfg.Region == "" {
 		cfg.Region = defaultS3Region