Repository: vshn/k8up
Pull request #169 (2020-12-01 15:28:11)

feat: ensure repository is always set
This is a reopening of #148 

This change ensures that the restic repository is always available and set on jobs which is required for the queue to be able to detect which jobs are running against which repository. It adds a test related to this case to document and ensure what the execution queue does works.

Additionally it refactors to have a type for the possible job types.


Patch:
diff --git a/cfg/config.go b/cfg/config.go
index 815eb531..e723b844 100644
--- a/cfg/config.go
+++ b/cfg/config.go
@@ -1,5 +1,7 @@
 package cfg
 
+import "fmt"
+
 const (
 	RestoreS3EndpointEnvName        = "RESTORE_S3ENDPOINT"
 	RestoreS3AccessKeyIDEnvName     = "RESTORE_ACCESSKEYID"
@@ -63,3 +65,8 @@ func NewDefaultConfig() *Configuration {
 		PodFilter:               "backupPod=true",
 	}
 }
+
+// GetGlobalRepository is a shortcut for building an S3 string "s3:<endpoint>/<bucket>"
+func GetGlobalRepository() string {
+	return fmt.Sprintf("s3:%s/%s", Config.GlobalS3Endpoint, Config.GlobalS3Bucket)
+}
diff --git a/controllers/archive_controller.go b/controllers/archive_controller.go
index 1cb0ed6e..cbe82eb9 100644
--- a/controllers/archive_controller.go
+++ b/controllers/archive_controller.go
@@ -11,6 +11,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	k8upv1alpha1 "github.com/vshn/k8up/api/v1alpha1"
+	"github.com/vshn/k8up/cfg"
 	"github.com/vshn/k8up/handler"
 	"github.com/vshn/k8up/job"
 )
@@ -39,7 +40,11 @@ func (r *ArchiveReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 		return ctrl.Result{}, err
 	}
 
-	config := job.NewConfig(ctx, r.Client, log, archive, r.Scheme)
+	repository := cfg.GetGlobalRepository()
+	if archive.Spec.Backend != nil {
+		repository = archive.Spec.Backend.String()
+	}
+	config := job.NewConfig(ctx, r.Client, log, archive, r.Scheme, repository)
 
 	archiveHandler := handler.NewHandler(config)
 	return ctrl.Result{RequeueAfter: time.Second * 30}, archiveHandler.Handle()
diff --git a/controllers/backup_controller.go b/controllers/backup_controller.go
index 200b0429..7517b119 100644
--- a/controllers/backup_controller.go
+++ b/controllers/backup_controller.go
@@ -11,6 +11,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	k8upv1alpha1 "github.com/vshn/k8up/api/v1alpha1"
+	"github.com/vshn/k8up/cfg"
 	"github.com/vshn/k8up/handler"
 	"github.com/vshn/k8up/job"
 )
@@ -41,7 +42,11 @@ func (r *BackupReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 		return ctrl.Result{}, err
 	}
 
-	config := job.NewConfig(ctx, r.Client, log, backup, r.Scheme)
+	repository := cfg.GetGlobalRepository()
+	if backup.Spec.Backend != nil {
+		repository = backup.Spec.Backend.String()
+	}
+	config := job.NewConfig(ctx, r.Client, log, backup, r.Scheme, repository)
 
 	backupHandler := handler.NewHandler(config)
 
diff --git a/controllers/check_controller.go b/controllers/check_controller.go
index 31064605..b423a849 100644
--- a/controllers/check_controller.go
+++ b/controllers/check_controller.go
@@ -11,6 +11,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	k8upv1alpha1 "github.com/vshn/k8up/api/v1alpha1"
+	"github.com/vshn/k8up/cfg"
 	"github.com/vshn/k8up/handler"
 	"github.com/vshn/k8up/job"
 )
@@ -39,7 +40,12 @@ func (r *CheckReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 		return ctrl.Result{}, err
 	}
 
-	config := job.NewConfig(ctx, r.Client, logger, check, r.Scheme)
+	repository := cfg.GetGlobalRepository()
+	if check.Spec.Backend != nil {
+		repository = check.Spec.Backend.String()
+	}
+
+	config := job.NewConfig(ctx, r.Client, logger, check, r.Scheme, repository)
 
 	checkHandler := handler.NewHandler(config)
 
diff --git a/controllers/job_controller.go b/controllers/job_controller.go
index fd8acadf..2e464b56 100644
--- a/controllers/job_controller.go
+++ b/controllers/job_controller.go
@@ -40,7 +40,7 @@ func (r *JobReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 		return reconcile.Result{}, err
 	}
 
-	config := job.NewConfig(ctx, r.Client, log, nil, r.Scheme)
+	config := job.NewConfig(ctx, r.Client, log, nil, r.Scheme, "")
 
 	return ctrl.Result{}, handler.NewJobHandler(config, jobObj).Handle()
 }
diff --git a/controllers/prune_controller.go b/controllers/prune_controller.go
index f32bb228..b35369cf 100644
--- a/controllers/prune_controller.go
+++ b/controllers/prune_controller.go
@@ -11,6 +11,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	k8upv1alpha1 "github.com/vshn/k8up/api/v1alpha1"
+	"github.com/vshn/k8up/cfg"
 	"github.com/vshn/k8up/handler"
 	"github.com/vshn/k8up/job"
 )
@@ -42,7 +43,11 @@ func (r *PruneReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 		return ctrl.Result{}, nil
 	}
 
-	config := job.NewConfig(ctx, r.Client, log, prune, r.Scheme)
+	repository := cfg.GetGlobalRepository()
+	if prune.Spec.Backend != nil {
+		repository = prune.Spec.Backend.String()
+	}
+	config := job.NewConfig(ctx, r.Client, log, prune, r.Scheme, repository)
 
 	pruneHandler := handler.NewHandler(config)
 
diff --git a/controllers/restore_controller.go b/controllers/restore_controller.go
index a3206d9a..9d3f64f0 100644
--- a/controllers/restore_controller.go
+++ b/controllers/restore_controller.go
@@ -11,6 +11,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	k8upv1alpha1 "github.com/vshn/k8up/api/v1alpha1"
+	"github.com/vshn/k8up/cfg"
 	"github.com/vshn/k8up/handler"
 	"github.com/vshn/k8up/job"
 )
@@ -42,7 +43,11 @@ func (r *RestoreReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 		return ctrl.Result{}, nil
 	}
 
-	config := job.NewConfig(ctx, r.Client, log, restore, r.Scheme)
+	repository := cfg.GetGlobalRepository()
+	if restore.Spec.Backend != nil {
+		repository = restore.Spec.Backend.String()
+	}
+	config := job.NewConfig(ctx, r.Client, log, restore, r.Scheme, repository)
 
 	restoreHandler := handler.NewHandler(config)
 
diff --git a/controllers/schedule_controller.go b/controllers/schedule_controller.go
index d2bffa36..192f5024 100644
--- a/controllers/schedule_controller.go
+++ b/controllers/schedule_controller.go
@@ -11,6 +11,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	k8upv1alpha1 "github.com/vshn/k8up/api/v1alpha1"
+	"github.com/vshn/k8up/cfg"
 	"github.com/vshn/k8up/handler"
 	"github.com/vshn/k8up/job"
 )
@@ -39,7 +40,11 @@ func (r *ScheduleReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 		return reconcile.Result{}, err
 	}
 
-	config := job.NewConfig(ctx, r.Client, log, schedule, r.Scheme)
+	repository := cfg.GetGlobalRepository()
+	if schedule.Spec.Backend != nil {
+		repository = schedule.Spec.Backend.String()
+	}
+	config := job.NewConfig(ctx, r.Client, log, schedule, r.Scheme, repository)
 
 	return ctrl.Result{}, handler.NewScheduleHandler(config, schedule).Handle()
 }
diff --git a/executor/worker.go b/executor/worker.go
index 00be6d6a..d6ae6e32 100644
--- a/executor/worker.go
+++ b/executor/worker.go
@@ -59,19 +59,22 @@ func (qe *QueueWorker) loopRepositoryJobs(repository string) {
 				!observer.GetObserver().IsConcurrentJobsLimitReached(jobType, jobLimit)
 		}
 
-		if shouldRun {
-			err := job.Execute()
-			if err != nil {
-				if !errors.IsAlreadyExists(err) {
-					job.Logger().Error(err, "cannot create job", "repository", repository)
-				}
-			}
+		if !shouldRun {
+			job.Logger().Info("skipping job due to exclusivity", "exclusive", job.Exclusive(), "repository", job.GetRepository())
+			continue
+		}
 
-			// Skip the rest for this repository if we just started an exclusive
-			// job.
-			if job.Exclusive() {
-				return
+		err := job.Execute()
+		if err != nil {
+			if !errors.IsAlreadyExists(err) {
+				job.Logger().Error(err, "cannot create job", "repository", repository)
 			}
 		}
+
+		// Skip the rest for this repository if we just started an exclusive
+		// job.
+		if job.Exclusive() {
+			return
+		}
 	}
 }
diff --git a/handler/generic.go b/handler/generic.go
index d39a2007..c51b470a 100644
--- a/handler/generic.go
+++ b/handler/generic.go
@@ -30,7 +30,9 @@ func (h *Handler) Handle() error {
 func (h *Handler) queueJob() error {
 	h.Log.Info("adding job to the queue")
 
-	queue.GetExecQueue().Add(executor.NewExecutor(h.Config))
+	e := executor.NewExecutor(h.Config)
+
+	queue.GetExecQueue().Add(e)
 
 	return nil
 }
diff --git a/job/job.go b/job/job.go
index 1018c6e9..0d02d45d 100644
--- a/job/job.go
+++ b/job/job.go
@@ -47,13 +47,14 @@ type Object interface {
 }
 
 // NewConfig returns a new configuration.
-func NewConfig(ctx context.Context, client client.Client, log logr.Logger, obj Object, scheme *runtime.Scheme) Config {
+func NewConfig(ctx context.Context, client client.Client, log logr.Logger, obj Object, scheme *runtime.Scheme, repository string) Config {
 	return Config{
-		Client: client,
-		Log:    log,
-		CTX:    ctx,
-		Obj:    obj,
-		Scheme: scheme,
+		Client:     client,
+		Log:        log,
+		CTX:        ctx,
+		Obj:        obj,
+		Scheme:     scheme,
+		Repository: repository,
 	}
 }
 


Test patch:
diff --git a/executor/restore_test.go b/executor/restore_test.go
index 76762781..73e06b45 100644
--- a/executor/restore_test.go
+++ b/executor/restore_test.go
@@ -298,6 +298,6 @@ func jobMatcher(restoreType string, additionalArgs []string, env Elements, volum
 }
 
 func newConfig() *job.Config {
-	cfg := job.NewConfig(context.TODO(), nil, nil, &k8upv1alpha1.Restore{}, scheme)
+	cfg := job.NewConfig(context.TODO(), nil, nil, &k8upv1alpha1.Restore{}, scheme, "")
 	return &cfg
 }
diff --git a/queue/execution_test.go b/queue/execution_test.go
new file mode 100644
index 00000000..e3421851
--- /dev/null
+++ b/queue/execution_test.go
@@ -0,0 +1,66 @@
+package queue
+
+import (
+	"errors"
+	"testing"
+
+	k8upv1alpha1 "github.com/vshn/k8up/api/v1alpha1"
+
+	"github.com/go-logr/logr"
+)
+
+type mockExecutor struct {
+	exclusive  bool
+	repository string
+}
+
+func (m *mockExecutor) Execute() error {
+	return errors.New("not implemented")
+}
+func (m *mockExecutor) Exclusive() bool {
+	return m.exclusive
+}
+func (m *mockExecutor) Logger() logr.Logger {
+	return nil
+}
+func (m *mockExecutor) GetRepository() string {
+	return m.repository
+}
+func (m *mockExecutor) GetJobNamespace() string {
+	return "test"
+}
+func (m *mockExecutor) GetJobType() k8upv1alpha1.JobType {
+	return k8upv1alpha1.ArchiveType
+}
+func (m *mockExecutor) GetConcurrencyLimit() int {
+	return 1
+}
+
+func TestExecutionQueue(t *testing.T) {
+	q := newExecutionQueue()
+
+	if !q.IsEmpty("repo1") || !q.IsEmpty("repo2") || !q.IsEmpty("") {
+		t.Fatal("queue is supposed to be empty")
+	}
+
+	m1 := &mockExecutor{false, "repo1"}
+	q.Add(m1)
+	m2 := &mockExecutor{true, "repo1"}
+	q.Add(m2)
+	m3 := &mockExecutor{true, "repo2"}
+	q.Add(m3)
+
+	a1 := q.Get("repo1")
+	a2 := q.Get("repo1")
+	a3 := q.Get("repo2")
+
+	if a1 != m2 {
+		t.Error("expected to retrieve exclusive executor first")
+	}
+	if a2 != m1 {
+		t.Error("expected to retrieve non-exclusive executor second")
+	}
+	if a3 != m3 {
+		t.Error("expected to retrieve repo2")
+	}
+}