Repository: vshn/k8up (go)

Issue:
Limit max concurrent jobs
The operator should have the ability to limit how many jobs should get run concurrently.

This limit should be configurable by job type (prune, backup, check, etc.)

Pull request:
feat: limit max concurrent jobs by type (prune, backup, check, etc.)
Fixes #120 

Patch:
diff --git a/api/v1alpha1/archive_types.go b/api/v1alpha1/archive_types.go
index cc8a1f43..4ed57ab9 100644
--- a/api/v1alpha1/archive_types.go
+++ b/api/v1alpha1/archive_types.go
@@ -58,8 +58,8 @@ func (a *Archive) GetMetaObject() metav1.Object {
 	return a
 }
 
-func (*Archive) GetType() string {
-	return "archive"
+func (*Archive) GetType() JobType {
+	return ArchiveType
 }
 
 func (a *Archive) GetK8upStatus() *K8upStatus {
diff --git a/api/v1alpha1/backup_types.go b/api/v1alpha1/backup_types.go
index 397a5e45..e05e9284 100644
--- a/api/v1alpha1/backup_types.go
+++ b/api/v1alpha1/backup_types.go
@@ -81,8 +81,8 @@ func (b *Backup) GetMetaObject() metav1.Object {
 	return b
 }
 
-func (*Backup) GetType() string {
-	return "backup"
+func (*Backup) GetType() JobType {
+	return BackupType
 }
 
 func (b *Backup) GetK8upStatus() *K8upStatus {
diff --git a/api/v1alpha1/check_types.go b/api/v1alpha1/check_types.go
index 1fec9ca8..c6c921ca 100644
--- a/api/v1alpha1/check_types.go
+++ b/api/v1alpha1/check_types.go
@@ -51,8 +51,8 @@ func (c *Check) GetMetaObject() metav1.Object {
 	return c
 }
 
-func (c *Check) GetType() string {
-	return "check"
+func (c *Check) GetType() JobType {
+	return CheckType
 }
 
 func (c *Check) GetK8upStatus() *K8upStatus {
diff --git a/api/v1alpha1/job_types.go b/api/v1alpha1/job_types.go
new file mode 100644
index 00000000..dae68181
--- /dev/null
+++ b/api/v1alpha1/job_types.go
@@ -0,0 +1,13 @@
+package v1alpha1
+
+// JobType defines what job type this is.
+type JobType string
+
+const (
+	BackupType   JobType = "backup"
+	CheckType    JobType = "check"
+	ArchiveType  JobType = "archive"
+	RestoreType  JobType = "restore"
+	PruneType    JobType = "prune"
+	ScheduleType JobType = "schedule"
+)
diff --git a/api/v1alpha1/prune_types.go b/api/v1alpha1/prune_types.go
index 10c9ae89..62376224 100644
--- a/api/v1alpha1/prune_types.go
+++ b/api/v1alpha1/prune_types.go
@@ -64,8 +64,8 @@ func (p *Prune) GetMetaObject() metav1.Object {
 	return p
 }
 
-func (p *Prune) GetType() string {
-	return "prune"
+func (p *Prune) GetType() JobType {
+	return PruneType
 }
 
 func (p *Prune) GetK8upStatus() *K8upStatus {
diff --git a/api/v1alpha1/restore_types.go b/api/v1alpha1/restore_types.go
index 2dae0525..c86d4438 100644
--- a/api/v1alpha1/restore_types.go
+++ b/api/v1alpha1/restore_types.go
@@ -65,8 +65,8 @@ func (r *Restore) GetMetaObject() metav1.Object {
 	return r
 }
 
-func (r *Restore) GetType() string {
-	return "restore"
+func (r *Restore) GetType() JobType {
+	return RestoreType
 }
 
 func (r *Restore) GetK8upStatus() *K8upStatus {
diff --git a/api/v1alpha1/schedule_types.go b/api/v1alpha1/schedule_types.go
index e03f7818..6355ff32 100644
--- a/api/v1alpha1/schedule_types.go
+++ b/api/v1alpha1/schedule_types.go
@@ -100,8 +100,8 @@ func (s *Schedule) GetMetaObject() metav1.Object {
 	return s
 }
 
-func (*Schedule) GetType() string {
-	return "schedule"
+func (*Schedule) GetType() JobType {
+	return ScheduleType
 }
 
 func (s *Schedule) GetK8upStatus() *K8upStatus {
diff --git a/cfg/config.go b/cfg/config.go
index 8a8c8331..815eb531 100644
--- a/cfg/config.go
+++ b/cfg/config.go
@@ -12,29 +12,34 @@ const (
 
 // Configuration holds a strongly-typed tree of the configuration
 type Configuration struct {
-	MountPath                      string `koanf:"datapath"`
-	BackupAnnotation               string `koanf:"annotation"`
-	BackupCommandAnnotation        string `koanf:"backupcommandannotation"`
-	FileExtensionAnnotation        string `koanf:"fileextensionannotation"`
-	ServiceAccount                 string `koanf:"podexecaccountname"`
-	BackupCheckSchedule            string `koanf:"checkschedule"`
-	GlobalAccessKey                string `koanf:"globalaccesskeyid"`
-	GlobalKeepJobs                 int    `koanf:"globalkeepjobs"`
-	GlobalRepoPassword             string `koanf:"globalrepopassword"`
-	GlobalRestoreS3AccessKey       string `koanf:"globalrestores3accesskeyid"`
-	GlobalRestoreS3Bucket          string `koanf:"globalrestores3bucket"`
-	GlobalRestoreS3Endpoint        string `koanf:"globalrestores3endpoint"`
-	GlobalRestoreS3SecretAccessKey string `koanf:"globalrestores3secretaccesskeyid"`
-	GlobalS3Bucket                 string `koanf:"globals3bucket"`
-	GlobalS3Endpoint               string `koanf:"globals3endpoint"`
-	GlobalSecretAccessKey          string `koanf:"globalsecretaccesskeyid"`
-	GlobalStatsURL                 string `koanf:"globalstatsurl"`
-	BackupImage                    string `koanf:"image"`
-	MetricsBindAddress             string `koanf:"metrics-bindaddress"`
-	PodExecRoleName                string `koanf:"podexecrolename"`
-	PodFilter                      string `koanf:"podfilter"`
-	PromURL                        string `koanf:"promurl"`
-	RestartPolicy                  string `koanf:"restartpolicy"`
+	MountPath                        string `koanf:"datapath"`
+	BackupAnnotation                 string `koanf:"annotation"`
+	BackupCommandAnnotation          string `koanf:"backupcommandannotation"`
+	FileExtensionAnnotation          string `koanf:"fileextensionannotation"`
+	ServiceAccount                   string `koanf:"podexecaccountname"`
+	BackupCheckSchedule              string `koanf:"checkschedule"`
+	GlobalAccessKey                  string `koanf:"globalaccesskeyid"`
+	GlobalKeepJobs                   int    `koanf:"globalkeepjobs"`
+	GlobalRepoPassword               string `koanf:"globalrepopassword"`
+	GlobalRestoreS3AccessKey         string `koanf:"globalrestores3accesskeyid"`
+	GlobalRestoreS3Bucket            string `koanf:"globalrestores3bucket"`
+	GlobalRestoreS3Endpoint          string `koanf:"globalrestores3endpoint"`
+	GlobalRestoreS3SecretAccessKey   string `koanf:"globalrestores3secretaccesskeyid"`
+	GlobalS3Bucket                   string `koanf:"globals3bucket"`
+	GlobalS3Endpoint                 string `koanf:"globals3endpoint"`
+	GlobalSecretAccessKey            string `koanf:"globalsecretaccesskeyid"`
+	GlobalStatsURL                   string `koanf:"globalstatsurl"`
+	GlobalConcurrentArchiveJobsLimit int    `koanf:"globalconcurrentarchivejobslimit"`
+	GlobalConcurrentBackupJobsLimit  int    `koanf:"globalconcurrentbackupjobslimit"`
+	GlobalConcurrentCheckJobsLimit   int    `koanf:"globalconcurrentcheckjobslimit"`
+	GlobalConcurrentPruneJobsLimit   int    `koanf:"globalconcurrentprunejobslimit"`
+	GlobalConcurrentRestoreJobsLimit int    `koanf:"globalconcurrentrestorejobslimit"`
+	BackupImage                      string `koanf:"image"`
+	MetricsBindAddress               string `koanf:"metrics-bindaddress"`
+	PodExecRoleName                  string `koanf:"podexecrolename"`
+	PodFilter                        string `koanf:"podfilter"`
+	PromURL                          string `koanf:"promurl"`
+	RestartPolicy                    string `koanf:"restartpolicy"`
 }
 
 var (
diff --git a/executor/archive.go b/executor/archive.go
index a41229fb..5a1be2fc 100644
--- a/executor/archive.go
+++ b/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)
diff --git a/executor/backup.go b/executor/backup.go
index cafb57de..f66b77b8 100644
--- a/executor/backup.go
+++ b/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.
diff --git a/executor/check.go b/executor/check.go
index 757603a0..d0eecf42 100644
--- a/executor/check.go
+++ b/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)
diff --git a/executor/generic.go b/executor/generic.go
index f0c61bf0..f0ce2eb9 100644
--- a/executor/generic.go
+++ b/executor/generic.go
@@ -6,9 +6,11 @@ package executor
 
 import (
 	"fmt"
-	"github.com/vshn/k8up/cfg"
 	"sort"
 
+	"github.com/vshn/k8up/api/v1alpha1"
+	"github.com/vshn/k8up/cfg"
+
 	"github.com/go-logr/logr"
 	"github.com/imdario/mergo"
 	"github.com/vshn/k8up/job"
@@ -112,22 +114,22 @@ func (g *generic) GetJobNamespace() string {
 	return g.Obj.GetMetaObject().GetNamespace()
 }
 
-func (g *generic) GetJobType() string {
+func (g *generic) GetJobType() v1alpha1.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 v1alpha1.BackupType:
 		return NewBackupExecutor(config)
-	case "check":
+	case v1alpha1.CheckType:
 		return NewCheckExecutor(config)
-	case "archive":
+	case v1alpha1.ArchiveType:
 		return NewArchiveExecutor(config)
-	case "prune":
+	case v1alpha1.PruneType:
 		return NewPruneExecutor(config)
-	case "restore":
+	case v1alpha1.RestoreType:
 		return NewRestoreExecutor(config)
 	}
 	return nil
diff --git a/executor/prune.go b/executor/prune.go
index 08a912d2..caf57c5f 100644
--- a/executor/prune.go
+++ b/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)
diff --git a/executor/restore.go b/executor/restore.go
index c2c1790d..ffd4ce37 100644
--- a/executor/restore.go
+++ b/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
 }
diff --git a/executor/worker.go b/executor/worker.go
index a7bd3634..00be6d6a 100644
--- a/executor/worker.go
+++ b/executor/worker.go
@@ -46,6 +46,8 @@ func (qe *QueueWorker) executeQueue() {
 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() {
@@ -53,7 +55,8 @@ func (qe *QueueWorker) loopRepositoryJobs(repository string) {
 			// and mark that in the status. So it is skippable.
 			shouldRun = !observer.GetObserver().IsAnyJobRunning(repository)
 		} else {
-			shouldRun = !observer.GetObserver().IsExclusiveJobRunning(repository)
+			shouldRun = !observer.GetObserver().IsExclusiveJobRunning(repository) &&
+				!observer.GetObserver().IsConcurrentJobsLimitReached(jobType, jobLimit)
 		}
 
 		if shouldRun {
diff --git a/go.mod b/go.mod
index 0908f699..820c2173 100644
--- a/go.mod
+++ b/go.mod
@@ -10,6 +10,7 @@ require (
 	github.com/onsi/gomega v1.10.1
 	github.com/prometheus/client_golang v1.8.0
 	github.com/robfig/cron/v3 v3.0.1
+	github.com/stretchr/testify v1.6.1
 	k8s.io/api v0.18.10
 	k8s.io/apimachinery v0.18.10
 	k8s.io/client-go v0.18.10
diff --git a/handler/job.go b/handler/job.go
index 4816d572..d98da90d 100644
--- a/handler/job.go
+++ b/handler/job.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"strconv"
 
+	"github.com/vshn/k8up/api/v1alpha1"
 	"github.com/vshn/k8up/job"
 	"github.com/vshn/k8up/observer"
 	batchv1 "k8s.io/api/batch/v1"
@@ -75,8 +76,14 @@ func (j *JobHandler) Handle() error {
 		exclusive = false
 	}
 
+	jobType := v1alpha1.ScheduleType
+	if j.Config.Obj != nil {
+		jobType = j.Config.Obj.GetType()
+	}
+
 	oj := observer.ObservableJob{
 		Job:       j.job,
+		JobType:   jobType,
 		Exclusive: exclusive,
 		Event:     jobEvent,
 	}
diff --git a/job/job.go b/job/job.go
index aad249e9..1018c6e9 100644
--- a/job/job.go
+++ b/job/job.go
@@ -4,9 +4,11 @@ package job
 
 import (
 	"context"
+
 	"github.com/vshn/k8up/cfg"
 
 	"github.com/go-logr/logr"
+	"github.com/vshn/k8up/api/v1alpha1"
 	k8upv1alpha1 "github.com/vshn/k8up/api/v1alpha1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -41,7 +43,7 @@ type Object interface {
 	GetMetaObject() metav1.Object
 	GetRuntimeObject() runtime.Object
 	GetK8upStatus() *k8upv1alpha1.K8upStatus
-	GetType() string
+	GetType() v1alpha1.JobType
 }
 
 // NewConfig returns a new configuration.
diff --git a/observer/observer.go b/observer/observer.go
index 1113615a..73fb57b1 100644
--- a/observer/observer.go
+++ b/observer/observer.go
@@ -9,18 +9,22 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/vshn/k8up/api/v1alpha1"
 	batchv1 "k8s.io/api/batch/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/metrics"
 )
 
-var (
+const (
 	Update   EventType = "update"
 	Delete   EventType = "delete"
 	Create   EventType = "create"
 	Failed   EventType = "failed"
 	Suceeded EventType = "suceeded"
 	Running  EventType = "running"
+)
+
+var (
 	observer *Observer
 
 	promLabels = []string{
@@ -56,6 +60,7 @@ type Observer struct {
 // ObservableJob defines a batchv1.job that is being observed by the Observer.
 type ObservableJob struct {
 	Job        *batchv1.Job
+	JobType    v1alpha1.JobType
 	Event      EventType
 	Exclusive  bool
 	Repository string
@@ -202,6 +207,32 @@ func (o *Observer) IsAnyJobRunning(repository string) bool {
 	return false
 }
 
+// IsConcurrentJobsLimitReached checks if the limit of concurrent jobs by type (backup, check, etc)
+// has been reached
+func (o *Observer) IsConcurrentJobsLimitReached(jobType v1alpha1.JobType, limit int) bool {
+	if limit <= 0 {
+		return false
+	}
+
+	o.mutex.Lock()
+	listOfJobs := o.observedJobs
+	o.mutex.Unlock()
+
+	if len(listOfJobs) == 0 {
+		return false
+	}
+	count := 0
+	for _, job := range listOfJobs {
+		if job.JobType == jobType {
+			count++
+			if count >= limit {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // RegisterCallback will register a function to the given observed job.
 // The callback will be executed if the job is, successful, failed or deleted.
 func (o *Observer) RegisterCallback(name string, callback func()) {
diff --git a/queue/execution.go b/queue/execution.go
index 7af4b18e..7a04a4b6 100644
--- a/queue/execution.go
+++ b/queue/execution.go
@@ -10,6 +10,7 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/vshn/k8up/api/v1alpha1"
 	"sigs.k8s.io/controller-runtime/pkg/metrics"
 )
 
@@ -39,9 +40,11 @@ type Executor interface {
 	// Associate the logs with the actual job.
 	Logger() logr.Logger
 	// GetJobType() returns the type of the CDR that the job will execute
-	GetJobType() string
+	GetJobType() v1alpha1.JobType
 	// GetJobNamespace() returns the namespace of the CDR that the job will execute
 	GetJobNamespace() string
+	// GetConcurrencyLimit
+	GetConcurrencyLimit() int
 	// GetName() string
 	GetRepository() string
 	// TODO: ability to mark job as skipped && metric for that
@@ -113,10 +116,10 @@ func (eq *ExecutionQueue) GetRepositories() []string {
 	return repositories
 }
 
-func (eq *ExecutionQueue) incQueueGauge(namespace, jobType string) {
-	queueGauge.WithLabelValues(namespace, jobType).Inc()
+func (eq *ExecutionQueue) incQueueGauge(namespace string, jobType v1alpha1.JobType) {
+	queueGauge.WithLabelValues(namespace, string(jobType)).Inc()
 }
 
-func (eq *ExecutionQueue) decQueueGauge(namespace, jobType string) {
-	queueGauge.WithLabelValues(namespace, jobType).Dec()
+func (eq *ExecutionQueue) decQueueGauge(namespace string, jobType v1alpha1.JobType) {
+	queueGauge.WithLabelValues(namespace, string(jobType)).Dec()
 }


Test patch:
diff --git a/observer/observer_test.go b/observer/observer_test.go
new file mode 100644
index 00000000..a70845d6
--- /dev/null
+++ b/observer/observer_test.go
@@ -0,0 +1,58 @@
+package observer
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/vshn/k8up/api/v1alpha1"
+	batchv1 "k8s.io/api/batch/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	ctrl "sigs.k8s.io/controller-runtime"
+)
+
+func TestObserver_IsConcurrentJobsLimitReached(t *testing.T) {
+
+	o := &Observer{
+		events:       make(chan ObservableJob, 10),
+		observedJobs: make(map[string]ObservableJob),
+		log:          ctrl.Log.WithName("observer-test"),
+		mutex:        sync.Mutex{},
+	}
+
+	// empty observedJobs
+	isLimitReached := o.IsConcurrentJobsLimitReached(v1alpha1.BackupType, 1)
+	assert.False(t, isLimitReached)
+
+	oj := ObservableJob{
+		Job: &batchv1.Job{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "some-name",
+			},
+		},
+		JobType:    v1alpha1.BackupType,
+		Event:      Running,
+		Exclusive:  true,
+		Repository: "some-repo",
+		callback:   func() {},
+	}
+
+	ojs := make(map[string]ObservableJob)
+	ojs["some-name"] = oj
+	fmt.Println(len(ojs))
+
+	o.observedJobs = ojs
+
+	// 1 observedJob and limit = 1
+	isLimitReached = o.IsConcurrentJobsLimitReached(v1alpha1.BackupType, 1)
+	assert.True(t, isLimitReached)
+
+	// another job type not present in observableJobs
+	isLimitReached = o.IsConcurrentJobsLimitReached(v1alpha1.ArchiveType, 1)
+	assert.False(t, isLimitReached)
+
+	// limit is 0
+	isLimitReached = o.IsConcurrentJobsLimitReached(v1alpha1.ArchiveType, 0)
+	assert.False(t, isLimitReached)
+}