refactoring of job type from string to JobType type

api/v1alpha1/archive_types.go

@@ -1,6 +1,7 @@
 package v1alpha1
 
 import (
+	"github.com/vshn/k8up/cfg"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 )
@@ -58,8 +59,8 @@ func (a *Archive) GetMetaObject() metav1.Object {
 	return a
 }
 
-func (*Archive) GetType() string {
-	return "archive"
+func (*Archive) GetType() cfg.JobType {
+	return cfg.Archive
 }
 
 func (a *Archive) GetK8upStatus() *K8upStatus {

api/v1alpha1/backup_types.go

@@ -1,6 +1,7 @@
 package v1alpha1
 
 import (
+	"github.com/vshn/k8up/cfg"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 )
@@ -81,8 +82,8 @@ func (b *Backup) GetMetaObject() metav1.Object {
 	return b
 }
 
-func (*Backup) GetType() string {
-	return "backup"
+func (*Backup) GetType() cfg.JobType {
+	return cfg.Backup
 }
 
 func (b *Backup) GetK8upStatus() *K8upStatus {

api/v1alpha1/check_types.go

@@ -1,6 +1,7 @@
 package v1alpha1
 
 import (
+	"github.com/vshn/k8up/cfg"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 )
@@ -51,8 +52,8 @@ func (c *Check) GetMetaObject() metav1.Object {
 	return c
 }
 
-func (c *Check) GetType() string {
-	return "check"
+func (c *Check) GetType() cfg.JobType {
+	return cfg.Check
 }
 
 func (c *Check) GetK8upStatus() *K8upStatus {

api/v1alpha1/prune_types.go

@@ -1,6 +1,7 @@
 package v1alpha1
 
 import (
+	"github.com/vshn/k8up/cfg"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 )
@@ -64,8 +65,8 @@ func (p *Prune) GetMetaObject() metav1.Object {
 	return p
 }
 
-func (p *Prune) GetType() string {
-	return "prune"
+func (p *Prune) GetType() cfg.JobType {
+	return cfg.Prune
 }
 
 func (p *Prune) GetK8upStatus() *K8upStatus {

api/v1alpha1/restore_types.go

@@ -1,6 +1,7 @@
 package v1alpha1
 
 import (
+	"github.com/vshn/k8up/cfg"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -65,8 +66,8 @@ func (r *Restore) GetMetaObject() metav1.Object {
 	return r
 }
 
-func (r *Restore) GetType() string {
-	return "restore"
+func (r *Restore) GetType() cfg.JobType {
+	return cfg.Restore
 }
 
 func (r *Restore) GetK8upStatus() *K8upStatus {

api/v1alpha1/schedule_types.go

@@ -3,6 +3,7 @@ package v1alpha1
 import (
 	"fmt"
 
+	"github.com/vshn/k8up/cfg"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 )
@@ -100,8 +101,8 @@ func (s *Schedule) GetMetaObject() metav1.Object {
 	return s
 }
 
-func (*Schedule) GetType() string {
-	return "schedule"
+func (*Schedule) GetType() cfg.JobType {
+	return cfg.Schedule
 }
 
 func (s *Schedule) GetK8upStatus() *K8upStatus {

cfg/config.go

@@ -1,5 +1,8 @@
 package cfg
 
+// JobType is the type the jobs can have
+type JobType string
+
 const (
 	RestoreS3EndpointEnvName        = "RESTORE_S3ENDPOINT"
 	RestoreS3AccessKeyIDEnvName     = "RESTORE_ACCESSKEYID"
@@ -8,6 +11,14 @@ const (
 	ResticPasswordEnvName           = "RESTIC_PASSWORD"
 	AwsAccessKeyIDEnvName           = "AWS_ACCESS_KEY_ID"
 	AwsSecretAccessKeyEnvName       = "AWS_SECRET_ACCESS_KEY"
+
+	Archive  JobType = "archive"
+	Backup   JobType = "backup"
+	Check    JobType = "check"
+	Prune    JobType = "prune"
+	Restore  JobType = "restore"
+	Schedule JobType = "schedule"
+	Job      JobType = ""
 )
 
 // Configuration holds a strongly-typed tree of the configuration

executor/archive.go

@@ -2,6 +2,7 @@ package executor
 
 import (
 	stderrors "errors"
+
 	"github.com/vshn/k8up/cfg"
 
 	k8upv1alpha1 "github.com/vshn/k8up/api/v1alpha1"
@@ -28,6 +29,11 @@ func NewArchiveExecutor(config job.Config) *ArchiveExecutor {
 	}
 }
 
+// GetConcurrencyLimit returns the concurrent jobs limit
+func (a *ArchiveExecutor) GetConcurrencyLimit() int {
+	return cfg.Config.GlobalConcurrentArchiveJobsLimit
+}
+
 // Execute creates the actual batch.job on the k8s api.
 func (a *ArchiveExecutor) Execute() error {
 	archive, ok := a.Obj.(*k8upv1alpha1.Archive)

executor/backup.go

@@ -2,10 +2,11 @@ package executor
 
 import (
 	stderrors "errors"
-	"github.com/vshn/k8up/cfg"
 	"path"
 	"strconv"
 
+	"github.com/vshn/k8up/cfg"
+
 	k8upv1alpha1 "github.com/vshn/k8up/api/v1alpha1"
 	"github.com/vshn/k8up/job"
 	"github.com/vshn/k8up/observer"
@@ -39,6 +40,11 @@ func NewBackupExecutor(config job.Config) *BackupExecutor {
 	}
 }
 
+// GetConcurrencyLimit returns the concurrent jobs limit
+func (b *BackupExecutor) GetConcurrencyLimit() int {
+	return cfg.Config.GlobalConcurrentBackupJobsLimit
+}
+
 // Execute triggers the actual batch.job creation on the cluster. It will also register
 // a callback function on the observer so the prebackup pods can be removed after the backup
 // has finished.

executor/check.go

@@ -2,6 +2,7 @@ package executor
 
 import (
 	stderrors "errors"
+
 	"github.com/vshn/k8up/cfg"
 
 	k8upv1alpha1 "github.com/vshn/k8up/api/v1alpha1"
@@ -22,6 +23,11 @@ func NewCheckExecutor(config job.Config) *CheckExecutor {
 	}
 }
 
+// GetConcurrencyLimit returns the concurrent jobs limit
+func (c *CheckExecutor) GetConcurrencyLimit() int {
+	return cfg.Config.GlobalConcurrentCheckJobsLimit
+}
+
 // Execute creates the actual batch.job on the k8s api.
 func (c *CheckExecutor) Execute() error {
 	checkObject, ok := c.Obj.(*k8upv1alpha1.Check)

executor/generic.go

@@ -6,9 +6,10 @@ package executor
 
 import (
 	"fmt"
-	"github.com/vshn/k8up/cfg"
 	"sort"
 
+	"github.com/vshn/k8up/cfg"
+
 	"github.com/go-logr/logr"
 	"github.com/imdario/mergo"
 	"github.com/vshn/k8up/job"
@@ -112,22 +113,22 @@ func (g *generic) GetJobNamespace() string {
 	return g.Obj.GetMetaObject().GetNamespace()
 }
 
-func (g *generic) GetJobType() string {
+func (g *generic) GetJobType() cfg.JobType {
 	return g.Obj.GetType()
 }
 
 // NewExecutor will return the right Executor for the given job object.
 func NewExecutor(config job.Config) queue.Executor {
 	switch config.Obj.GetType() {
-	case "backup":
+	case cfg.Backup:
 		return NewBackupExecutor(config)
-	case "check":
+	case cfg.Check:
 		return NewCheckExecutor(config)
-	case "archive":
+	case cfg.Archive:
 		return NewArchiveExecutor(config)
-	case "prune":
+	case cfg.Prune:
 		return NewPruneExecutor(config)
-	case "restore":
+	case cfg.Restore:
 		return NewRestoreExecutor(config)
 	}
 	return nil

executor/prune.go

@@ -2,10 +2,11 @@ package executor
 
 import (
 	"errors"
-	"github.com/vshn/k8up/cfg"
 	"strconv"
 	"strings"
 
+	"github.com/vshn/k8up/cfg"
+
 	k8upv1alpha1 "github.com/vshn/k8up/api/v1alpha1"
 	"github.com/vshn/k8up/job"
 	"github.com/vshn/k8up/observer"
@@ -28,6 +29,11 @@ func NewPruneExecutor(config job.Config) *PruneExecutor {
 	}
 }
 
+// GetConcurrencyLimit returns the concurrent jobs limit
+func (p *PruneExecutor) GetConcurrencyLimit() int {
+	return cfg.Config.GlobalConcurrentPruneJobsLimit
+}
+
 // Execute creates the actual batch.job on the k8s api.
 func (p *PruneExecutor) Execute() error {
 	prune, ok := p.Obj.(*k8upv1alpha1.Prune)

executor/restore.go

@@ -2,9 +2,10 @@ package executor
 
 import (
 	"errors"
-	"github.com/vshn/k8up/cfg"
 	"strconv"
 
+	"github.com/vshn/k8up/cfg"
+
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -22,12 +23,19 @@ type RestoreExecutor struct {
 	generic
 }
 
+// NewRestoreExecutor will return a new executor for Restore jobs.
 func NewRestoreExecutor(config job.Config) *RestoreExecutor {
 	return &RestoreExecutor{
 		generic: generic{config},
 	}
 }
 
+// GetConcurrencyLimit returns the concurrent jobs limit
+func (r *RestoreExecutor) GetConcurrencyLimit() int {
+	return cfg.Config.GlobalConcurrentRestoreJobsLimit
+}
+
+// Execute creates the actual batch.job on the k8s api.
 func (r *RestoreExecutor) Execute() error {
 	restore, ok := r.Obj.(*k8upv1alpha1.Restore)
 	if !ok {
@@ -43,6 +51,7 @@ func (r *RestoreExecutor) Execute() error {
 	return nil
 }
 
+// Exclusive should return true for jobs that can't run while other jobs run.
 func (r *RestoreExecutor) Exclusive() bool {
 	return true
 }

executor/worker.go

@@ -47,6 +47,7 @@ func (qe *QueueWorker) loopRepositoryJobs(repository string) {
 	for !queue.GetExecQueue().IsEmpty(repository) {
 		job := queue.GetExecQueue().Get(repository)
 		jobType := job.GetJobType()
+		jobLimit := job.GetConcurrencyLimit()
 
 		var shouldRun bool
 		if job.Exclusive() {
@@ -55,7 +56,7 @@ func (qe *QueueWorker) loopRepositoryJobs(repository string) {
 			shouldRun = !observer.GetObserver().IsAnyJobRunning(repository)
 		} else {
 			shouldRun = !observer.GetObserver().IsExclusiveJobRunning(repository) &&
-				!observer.GetObserver().IsLimitConcurrentJobsReached(jobType, repository)
+				!observer.GetObserver().IsLimitConcurrentJobsReached(jobType, jobLimit, repository)
 		}
 
 		if shouldRun {

handler/job.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"strconv"
 
+	"github.com/vshn/k8up/cfg"
 	"github.com/vshn/k8up/job"
 	"github.com/vshn/k8up/observer"
 	batchv1 "k8s.io/api/batch/v1"
@@ -75,7 +76,7 @@ func (j *JobHandler) Handle() error {
 		exclusive = false
 	}
 
-	jobType := ""
+	jobType := cfg.Job
 	if j.Config.Obj != nil {
 		jobType = j.Config.Obj.GetType()
 	}

job/job.go

@@ -4,6 +4,7 @@ package job
 
 import (
 	"context"
+
 	"github.com/vshn/k8up/cfg"
 
 	"github.com/go-logr/logr"
@@ -41,7 +42,7 @@ type Object interface {
 	GetMetaObject() metav1.Object
 	GetRuntimeObject() runtime.Object
 	GetK8upStatus() *k8upv1alpha1.K8upStatus
-	GetType() string
+	GetType() cfg.JobType
 }
 
 // NewConfig returns a new configuration.