GT-428 Add disallowConcurrent option to ArangoBackupPolicy (#1318)

Author: nikita-vanyasin

CHANGELOG.md

@@ -10,6 +10,7 @@
 - (Maintenance) Generics for type handling
 - (Bugfix) Fix creating sync components with EA type set to Managed and headless svc
 - (Feature) Check if Volume with LocalStorage is missing
+- (Feature) Add disallowConcurrent option to ArangoBackupPolicy
 
 ## [1.2.27](https://github.com/arangodb/kube-arangodb/tree/1.2.27) (2023-04-27)
 - (Feature) Add InSync Cache

pkg/apis/backup/v1/backup.go

@@ -1,7 +1,7 @@
 //
 // DISCLAIMER
 //
-// Copyright 2016-2022 ArangoDB GmbH, Cologne, Germany
+// Copyright 2016-2023 ArangoDB GmbH, Cologne, Germany
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -46,6 +46,10 @@ type ArangoBackupList struct {
 	Items []ArangoBackup `json:"items"`
 }
 
+func (a ArangoBackupList) GetItems() []ArangoBackup {
+	return a.Items
+}
+
 // +genclient
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 

pkg/apis/backup/v1/backup_policy_spec.go

@@ -1,7 +1,7 @@
 //
 // DISCLAIMER
 //
-// Copyright 2016-2022 ArangoDB GmbH, Cologne, Germany
+// Copyright 2016-2023 ArangoDB GmbH, Cologne, Germany
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -22,16 +22,26 @@ package v1
 
 import (
 	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/arangodb/kube-arangodb/pkg/util"
 )
 
 type ArangoBackupPolicySpec struct {
+	// Schedule is cron-compatible specification of backup schedule
 	Schedule string `json:"schedule"`
-
+	// AllowConcurrent if true, ArangoBackup will not be created when previous Backups are not finished. Defaults to true
+	AllowConcurrent *bool `json:"allowConcurrent,omitempty"`
+	// DeploymentSelector specifies which deployments should get a backup
 	DeploymentSelector *meta.LabelSelector `json:"selector,omitempty"`
-
+	// ArangoBackupTemplate specifies additional options for newly created ArangoBackup
 	BackupTemplate ArangoBackupTemplate `json:"template"`
 }
 
+// GetAllowConcurrent returns AllowConcurrent values. If AllowConcurrent is nil returns true
+func (a ArangoBackupPolicySpec) GetAllowConcurrent() bool {
+	return util.TypeOrDefault(a.AllowConcurrent, true)
+}
+
 type ArangoBackupTemplate struct {
 	Options *ArangoBackupSpecOptions `json:"options,omitempty"`
 

pkg/apis/backup/v1/zz_generated.deepcopy.go

@@ -1,7 +1,7 @@
 //
 // DISCLAIMER
 //
-// Copyright 2016-2022 ArangoDB GmbH, Cologne, Germany
+// Copyright 2016-2023 ArangoDB GmbH, Cologne, Germany
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -32,16 +32,21 @@ import (
 
 	"github.com/arangodb/kube-arangodb/pkg/apis/backup"
 	backupApi "github.com/arangodb/kube-arangodb/pkg/apis/backup/v1"
+	deployment "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
 	arangoClientSet "github.com/arangodb/kube-arangodb/pkg/generated/clientset/versioned"
 	operator "github.com/arangodb/kube-arangodb/pkg/operatorV2"
 	"github.com/arangodb/kube-arangodb/pkg/operatorV2/event"
 	"github.com/arangodb/kube-arangodb/pkg/operatorV2/operation"
+	"github.com/arangodb/kube-arangodb/pkg/util/errors"
+	"github.com/arangodb/kube-arangodb/pkg/util/globals"
+	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
 )
 
 const (
-	backupCreated = "ArangoBackupCreated"
-	policyError   = "Error"
-	rescheduled   = "Rescheduled"
+	backupCreated   = "ArangoBackupCreated"
+	policyError     = "Error"
+	rescheduled     = "Rescheduled"
+	scheduleSkipped = "ScheduleSkipped"
 )
 
 type handler struct {
@@ -153,8 +158,25 @@ func (h *handler) processBackupPolicy(policy *backupApi.ArangoBackupPolicy) back
 	}
 
 	for _, deployment := range deployments.Items {
-		b := policy.NewBackup(deployment.DeepCopy())
+		depl := deployment.DeepCopy()
+
+		if !policy.Spec.GetAllowConcurrent() {
+			previousBackupInProgress, err := h.isPreviousBackupInProgress(context.Background(), depl, policy.Name)
+			if err != nil {
+				h.eventRecorder.Warning(policy, policyError, "Policy Error: %s", err.Error())
+				return backupApi.ArangoBackupPolicyStatus{
+					Scheduled: policy.Status.Scheduled,
+					Message:   fmt.Sprintf("backup creation failed: %s", err.Error()),
+				}
+			}
+			if previousBackupInProgress {
+				eventMsg := fmt.Sprintf("Skipping ArangoBackup creation because earlier backup still running %s/%s", deployment.Namespace, deployment.Name)
+				h.eventRecorder.Normal(policy, scheduleSkipped, eventMsg)
+				continue
+			}
+		}
 
+		b := policy.NewBackup(depl)
 		if _, err := h.client.BackupV1().ArangoBackups(b.Namespace).Create(context.Background(), b, meta.CreateOptions{}); err != nil {
 			h.eventRecorder.Warning(policy, policyError, "Policy Error: %s", err.Error())
 
@@ -183,3 +205,60 @@ func (*handler) CanBeHandled(item operation.Item) bool {
 		item.Version == backupApi.SchemeGroupVersion.Version &&
 		item.Kind == backup.ArangoBackupPolicyResourceKind
 }
+
+func (h *handler) listAllBackupsForPolicy(ctx context.Context, d *deployment.ArangoDeployment, policyName string) ([]*backupApi.ArangoBackup, error) {
+	var r []*backupApi.ArangoBackup
+
+	if err := k8sutil.APIList[*backupApi.ArangoBackupList](ctx, h.client.BackupV1().ArangoBackups(d.Namespace), meta.ListOptions{
+		Limit: globals.GetGlobals().Kubernetes().RequestBatchSize().Get(),
+	}, func(result *backupApi.ArangoBackupList, err error) error {
+		if err != nil {
+			return err
+		}
+
+		for _, b := range result.Items {
+			if b.Spec.PolicyName == nil || *b.Spec.PolicyName != policyName {
+				continue
+			}
+			if b.Spec.Deployment.Name != d.Name {
+				continue
+			}
+			r = append(r, b.DeepCopy())
+		}
+
+		return nil
+	}); err != nil {
+		return nil, err
+	}
+
+	return r, nil
+}
+
+func (h *handler) isPreviousBackupInProgress(ctx context.Context, d *deployment.ArangoDeployment, policyName string) (bool, error) {
+	// It would be nice to List CRs with fieldSelector set, but this is not supported:
+	// https://github.com/kubernetes/kubernetes/issues/53459
+	// Instead we fetch all ArangoBackups:
+	backups, err := h.listAllBackupsForPolicy(ctx, d, policyName)
+	if err != nil {
+		return false, errors.Wrap(err, "Failed to list ArangoBackups")
+	}
+
+	for _, b := range backups {
+		// Check if we are in the failed state
+		switch b.Status.State {
+		case backupApi.ArangoBackupStateFailed:
+			continue
+		}
+
+		if b.Spec.Download != nil {
+			continue
+		}
+
+		// Backup is not yet done
+		if b.Status.Backup == nil {
+			return true, nil
+		}
+	}
+
+	return false, nil
+}

pkg/handlers/policy/handler.go

@@ -1,7 +1,7 @@
 //
 // DISCLAIMER
 //
-// Copyright 2016-2022 ArangoDB GmbH, Cologne, Germany
+// Copyright 2016-2023 ArangoDB GmbH, Cologne, Germany
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@
 package policy
 
 import (
+	"context"
 	"testing"
 	"time"
 
@@ -30,6 +31,7 @@ import (
 
 	backupApi "github.com/arangodb/kube-arangodb/pkg/apis/backup/v1"
 	"github.com/arangodb/kube-arangodb/pkg/operatorV2/operation"
+	"github.com/arangodb/kube-arangodb/pkg/util"
 )
 
 func Test_Scheduler_Schedule(t *testing.T) {
@@ -39,7 +41,7 @@ func Test_Scheduler_Schedule(t *testing.T) {
 	name := string(uuid.NewUUID())
 	namespace := string(uuid.NewUUID())
 
-	policy := newArangoBackupPolicy("* * * */2 *", namespace, name, map[string]string{}, backupApi.ArangoBackupTemplate{})
+	policy := newArangoBackupPolicy(namespace, name, newSimpleArangoBackupPolicySpec("* * * */2 *"))
 
 	database := newArangoDeployment(namespace, map[string]string{
 		"test": "me",
@@ -67,7 +69,7 @@ func Test_Scheduler_InvalidSchedule(t *testing.T) {
 	name := string(uuid.NewUUID())
 	namespace := string(uuid.NewUUID())
 
-	policy := newArangoBackupPolicy("", namespace, name, map[string]string{}, backupApi.ArangoBackupTemplate{})
+	policy := newArangoBackupPolicy(namespace, name, newSimpleArangoBackupPolicySpec(""))
 
 	database := newArangoDeployment(namespace, map[string]string{})
 
@@ -93,7 +95,7 @@ func Test_Scheduler_Valid_OneObject_SelectAll(t *testing.T) {
 	name := string(uuid.NewUUID())
 	namespace := string(uuid.NewUUID())
 
-	policy := newArangoBackupPolicy("* * * */2 *", namespace, name, map[string]string{}, backupApi.ArangoBackupTemplate{})
+	policy := newArangoBackupPolicy(namespace, name, newSimpleArangoBackupPolicySpec("* * * */2 *"))
 	policy.Status.Scheduled = meta.Time{
 		Time: time.Now().Add(-1 * time.Hour),
 	}
@@ -131,8 +133,9 @@ func Test_Scheduler_Valid_OneObject_Selector(t *testing.T) {
 	selectors := map[string]string{
 		"SELECTOR": string(uuid.NewUUID()),
 	}
-
-	policy := newArangoBackupPolicy("* * * */2 *", namespace, name, selectors, backupApi.ArangoBackupTemplate{})
+	spec := newSimpleArangoBackupPolicySpec("* * * */2 *")
+	spec.DeploymentSelector = &meta.LabelSelector{MatchLabels: selectors}
+	policy := newArangoBackupPolicy(namespace, name, spec)
 	policy.Status.Scheduled = meta.Time{
 		Time: time.Now().Add(-1 * time.Hour),
 	}
@@ -170,7 +173,9 @@ func Test_Scheduler_Valid_MultipleObject_Selector(t *testing.T) {
 		"SELECTOR": string(uuid.NewUUID()),
 	}
 
-	policy := newArangoBackupPolicy("* * * */2 *", namespace, name, selectors, backupApi.ArangoBackupTemplate{})
+	spec := newSimpleArangoBackupPolicySpec("* * * */2 *")
+	spec.DeploymentSelector = &meta.LabelSelector{MatchLabels: selectors}
+	policy := newArangoBackupPolicy(namespace, name, spec)
 	policy.Status.Scheduled = meta.Time{
 		Time: time.Now().Add(-1 * time.Hour),
 	}
@@ -211,7 +216,9 @@ func Test_Reschedule(t *testing.T) {
 		"SELECTOR": string(uuid.NewUUID()),
 	}
 
-	policy := newArangoBackupPolicy("* 13 * * *", namespace, name, selectors, backupApi.ArangoBackupTemplate{})
+	spec := newSimpleArangoBackupPolicySpec("* 13 * * *")
+	spec.DeploymentSelector = &meta.LabelSelector{MatchLabels: selectors}
+	policy := newArangoBackupPolicy(namespace, name, spec)
 
 	// Act
 	createArangoBackupPolicy(t, handler, policy)
@@ -269,8 +276,9 @@ func Test_Validate(t *testing.T) {
 			selectors := map[string]string{
 				"SELECTOR": string(uuid.NewUUID()),
 			}
-
-			policy := newArangoBackupPolicy(c, namespace, name, selectors, backupApi.ArangoBackupTemplate{})
+			spec := newSimpleArangoBackupPolicySpec(c)
+			spec.DeploymentSelector = &meta.LabelSelector{MatchLabels: selectors}
+			policy := newArangoBackupPolicy(namespace, name, spec)
 
 			require.NoError(t, policy.Validate())
 
@@ -286,3 +294,72 @@ func Test_Validate(t *testing.T) {
 		})
 	}
 }
+
+func Test_Concurrent(t *testing.T) {
+	testCase := func(t *testing.T, allowConcurrent *bool) {
+		handler := newFakeHandler()
+
+		name := string(uuid.NewUUID())
+		namespace := string(uuid.NewUUID())
+
+		spec := newSimpleArangoBackupPolicySpec("* * * */2 *")
+		spec.AllowConcurrent = allowConcurrent
+		policy := newArangoBackupPolicy(namespace, name, spec)
+		policy.Status.Scheduled = meta.Time{
+			Time: time.Now().Add(-1 * time.Hour),
+		}
+
+		database := newArangoDeployment(namespace, map[string]string{
+			"test": "me",
+		})
+
+		createArangoBackupPolicy(t, handler, policy)
+		createArangoDeployment(t, handler, database)
+
+		// "create" first backup
+		require.NoError(t, handler.Handle(newItemFromBackupPolicy(operation.Update, policy)))
+		backups := listArangoBackups(t, handler, namespace)
+		require.Len(t, backups, 1)
+
+		// "try create" second backup but first one still is not in TerminalState
+		policy = refreshArangoBackupPolicy(t, handler, policy)
+		policy.Status.Scheduled = meta.Time{
+			Time: time.Now().Add(-1 * time.Hour),
+		}
+		updateArangoBackupPolicy(t, handler, policy)
+		require.NoError(t, handler.Handle(newItemFromBackupPolicy(operation.Update, policy)))
+		backups = listArangoBackups(t, handler, namespace)
+		require.Len(t, backups, util.BoolSwitch(policy.Spec.GetAllowConcurrent(), 2, 1))
+
+		// mark previous backup as Ready
+		backup := backups[0].DeepCopy()
+		backup.Status.State = backupApi.ArangoBackupStateReady
+		backup.Status.Backup = &backupApi.ArangoBackupDetails{
+			ID: "SOME_ID",
+		}
+		_, err := handler.client.BackupV1().ArangoBackups(namespace).UpdateStatus(context.Background(), backup, meta.UpdateOptions{})
+		require.NoError(t, err)
+
+		// "try create" second backup again, should succeed
+		policy = refreshArangoBackupPolicy(t, handler, policy)
+		policy.Status.Scheduled = meta.Time{
+			Time: time.Now().Add(-1 * time.Hour),
+		}
+		updateArangoBackupPolicy(t, handler, policy)
+		require.NoError(t, handler.Handle(newItemFromBackupPolicy(operation.Update, policy)))
+		backups = listArangoBackups(t, handler, namespace)
+		require.Len(t, backups, util.BoolSwitch(policy.Spec.GetAllowConcurrent(), 3, 2))
+	}
+
+	t.Run("Explicit Allow", func(t *testing.T) {
+		testCase(t, util.NewType(true))
+	})
+
+	t.Run("Default Allow", func(t *testing.T) {
+		testCase(t, nil)
+	})
+
+	t.Run("Explicit Disallow", func(t *testing.T) {
+		testCase(t, util.NewType(false))
+	})
+}