[Bugfix] Allow ArangoBackup Creation during Upload state

CHANGELOG.md

@@ -12,6 +12,7 @@
 - (Feature) Set a leader in active fail-over mode
 - (Feature) Use policy/v1 instead policy/v1beta1
 - (Feature) OPS CLI with Arango Task
+- (Bugfix) Allow ArangoBackup Creation during Upload state
 
 ## [1.2.13](https://github.com/arangodb/kube-arangodb/tree/1.2.13) (2022-06-07)
 - (Bugfix) Fix arangosync members state inspection

pkg/handlers/backup/state_pending.go

@@ -30,12 +30,12 @@ func statePendingHandler(h *handler, backup *backupApi.ArangoBackup) (*backupApi
 		return nil, err
 	}
 
-	running, err := isBackupRunning(backup, h.client.BackupV1().ArangoBackups(backup.Namespace))
+	states, err := countBackupStates(backup, h.client.BackupV1().ArangoBackups(backup.Namespace))
 	if err != nil {
 		return nil, err
 	}
 
-	if running {
+	if l := states.get(backupApi.ArangoBackupStateScheduled, backupApi.ArangoBackupStateCreate, backupApi.ArangoBackupStateDownload, backupApi.ArangoBackupStateDownloading); len(l) > 0 {
 		return wrapUpdateStatus(backup,
 			updateStatusState(backupApi.ArangoBackupStatePending, "backup already in process"))
 	}

pkg/handlers/backup/state_pending_test.go

@@ -78,6 +78,44 @@ func Test_State_Pending_OneBackupObject(t *testing.T) {
 	checkBackup(t, newObj, backupApi.ArangoBackupStateScheduled, false)
 }
 
+func Test_State_Pending_WithUploadRunning(t *testing.T) {
+	// Arrange
+	handler, _ := newErrorsFakeHandler(mockErrorsArangoClientBackup{})
+
+	obj, deployment := newObjectSet(backupApi.ArangoBackupStatePending)
+
+	uploading := newArangoBackup(deployment.GetName(), deployment.GetNamespace(), string(uuid.NewUUID()), backupApi.ArangoBackupStateUploading)
+
+	// Act
+	createArangoDeployment(t, handler, deployment)
+	createArangoBackup(t, handler, obj, uploading)
+
+	require.NoError(t, handler.Handle(newItemFromBackup(operation.Update, obj)))
+
+	// Assert
+	newObj := refreshArangoBackup(t, handler, obj)
+	checkBackup(t, newObj, backupApi.ArangoBackupStateScheduled, false)
+}
+
+func Test_State_Pending_WithScheduled(t *testing.T) {
+	// Arrange
+	handler, _ := newErrorsFakeHandler(mockErrorsArangoClientBackup{})
+
+	obj, deployment := newObjectSet(backupApi.ArangoBackupStatePending)
+
+	uploading := newArangoBackup(deployment.GetName(), deployment.GetNamespace(), string(uuid.NewUUID()), backupApi.ArangoBackupStateScheduled)
+
+	// Act
+	createArangoDeployment(t, handler, deployment)
+	createArangoBackup(t, handler, obj, uploading)
+
+	require.NoError(t, handler.Handle(newItemFromBackup(operation.Update, obj)))
+
+	// Assert
+	newObj := refreshArangoBackup(t, handler, obj)
+	checkBackup(t, newObj, backupApi.ArangoBackupStatePending, false)
+}
+
 func Test_State_Pending_MultipleBackupObjectWithLimitation(t *testing.T) {
 	// Arrange
 	handler, _ := newErrorsFakeHandler(mockErrorsArangoClientBackup{})

pkg/handlers/backup/state_ready.go

@@ -23,6 +23,7 @@ package backup
 import (
 	"github.com/arangodb/go-driver"
 	backupApi "github.com/arangodb/kube-arangodb/pkg/apis/backup/v1"
+	"github.com/arangodb/kube-arangodb/pkg/util/globals"
 )
 
 func stateReadyHandler(h *handler, backup *backupApi.ArangoBackup) (*backupApi.ArangoBackupStatus, error) {
@@ -71,21 +72,33 @@ func stateReadyHandler(h *handler, backup *backupApi.ArangoBackup) (*backupApi.A
 	if backup.Spec.Upload != nil &&
 		(backup.Status.Backup.Uploaded == nil || (backup.Status.Backup.Uploaded != nil && !*backup.Status.Backup.Uploaded)) {
 		// Ensure that we can start upload process
-		running, err := isBackupRunning(backup, h.client.BackupV1().ArangoBackups(backup.Namespace))
+		states, err := countBackupStates(backup, h.client.BackupV1().ArangoBackups(backup.Namespace))
 		if err != nil {
 			return nil, err
 		}
 
-		if running {
-			return wrapUpdateStatus(backup,
-				updateStatusState(backupApi.ArangoBackupStateReady, "Upload process queued"),
-				updateStatusBackup(backupMeta),
-				updateStatusAvailable(true),
-			)
+		if l := states.get(backupApi.ArangoBackupStateUpload, backupApi.ArangoBackupStateUploading); globals.GetGlobals().Backup().ConcurrentUploads().Get() > len(l) {
+			// Check if there is no upload with same id
+			if len(l.filter(func(b *backupApi.ArangoBackup) bool {
+				if a1 := b.Status.Backup; a1 != nil {
+					if a2 := backup.Status.Backup; a2 != nil {
+						return a1.ID == a2.ID
+					}
+				}
+
+				return false
+			})) == 0 {
+
+				return wrapUpdateStatus(backup,
+					updateStatusState(backupApi.ArangoBackupStateUpload, ""),
+					updateStatusBackup(backupMeta),
+					updateStatusAvailable(true),
+				)
+			}
 		}
 
 		return wrapUpdateStatus(backup,
-			updateStatusState(backupApi.ArangoBackupStateUpload, ""),
+			updateStatusState(backupApi.ArangoBackupStateReady, "Upload process queued"),
 			updateStatusBackup(backupMeta),
 			updateStatusAvailable(true),
 		)

pkg/handlers/backup/util.go

@@ -22,9 +22,6 @@ package backup
 
 import (
 	"context"
-	"strings"
-
-	"github.com/arangodb/kube-arangodb/pkg/util/globals"
 
 	clientBackup "github.com/arangodb/kube-arangodb/pkg/generated/clientset/versioned/typed/backup/v1"
 
@@ -34,66 +31,68 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/handlers/backup/state"
 )
 
-var (
-	progressStates = []state.State{
-		backupApi.ArangoBackupStateScheduled,
-		backupApi.ArangoBackupStateCreate,
-		backupApi.ArangoBackupStateDownload,
-		backupApi.ArangoBackupStateDownloading,
-		backupApi.ArangoBackupStateUpload,
-		backupApi.ArangoBackupStateUploading,
+type backupStates []*backupApi.ArangoBackup
+
+func (b backupStates) filter(f func(b *backupApi.ArangoBackup) bool) backupStates {
+	if f == nil {
+		return nil
 	}
-)
 
-func inProgress(backup *backupApi.ArangoBackup) bool {
-	for _, state := range progressStates {
-		if state == backup.Status.State {
-			return true
+	r := make(backupStates, 0, len(b))
+
+	for id := range b {
+		if f(b[id]) {
+			r = append(r, b[id])
 		}
 	}
 
-	return false
+	return r
 }
 
-func isBackupRunning(backup *backupApi.ArangoBackup, client clientBackup.ArangoBackupInterface) (bool, error) {
+type backupStatesCount map[state.State]backupStates
+
+func (b backupStatesCount) get(states ...state.State) backupStates {
+	i := 0
+
+	for _, s := range states {
+		i += len(b[s])
+	}
+
+	if i == 0 {
+		return nil
+	}
+
+	r := make(backupStates, 0, i)
+
+	for _, s := range states {
+		r = append(r, b[s]...)
+	}
+
+	return r
+}
+
+func countBackupStates(backup *backupApi.ArangoBackup, client clientBackup.ArangoBackupInterface) (backupStatesCount, error) {
 	backups, err := client.List(context.Background(), meta.ListOptions{})
 
 	if err != nil {
-		return false, newTemporaryError(err)
+		return nil, newTemporaryError(err)
 	}
 
-	currentUploads := 0
+	ret := map[state.State]backupStates{}
 
 	for _, existingBackup := range backups.Items {
+		// Skip same backup from count
 		if existingBackup.Name == backup.Name {
 			continue
 		}
 
-		// We can upload multiple uploads from same deployment in same time
-		if backup.Status.State == backupApi.ArangoBackupStateReady &&
-			(existingBackup.Status.State == backupApi.ArangoBackupStateUpload || existingBackup.Status.State == backupApi.ArangoBackupStateUploading) {
-			currentUploads++
-			if backupUpload := backup.Status.Backup; backupUpload != nil {
-				if existingBackupUpload := existingBackup.Status.Backup; existingBackupUpload != nil {
-					if strings.EqualFold(backupUpload.ID, existingBackupUpload.ID) {
-						return true, nil
-					}
-				}
-			}
-		} else {
-			if existingBackup.Spec.Deployment.Name != backup.Spec.Deployment.Name {
-				continue
-			}
-
-			if inProgress(&existingBackup) {
-				return true, nil
-			}
+		// Skip backups which are not on same deployment from count
+		if existingBackup.Spec.Deployment.Name != backup.Spec.Deployment.Name {
+			continue
 		}
-	}
 
-	if backup.Status.State == backupApi.ArangoBackupStateReady {
-		return currentUploads >= globals.GetGlobals().Backup().ConcurrentUploads().Get(), nil
+		ret[existingBackup.Status.State] = append(ret[existingBackup.Status.State], existingBackup.DeepCopy())
 	}
 
-	return false, nil
+	return ret, nil
 }