Add enable-backup-compaction flag

controllers/compaction_lease_controller.go

@@ -174,13 +174,16 @@ func (lc *CompactionLeaseController) reconcileJob(ctx context.Context, logger lo
 				RequeueAfter: 10 * time.Second,
 			}, fmt.Errorf("error while fetching compaction job: %v", err)
 		}
-		// Required job doesn't exist. Create new
-		job, err = lc.createCompactJob(ctx, logger, etcd)
-		logger.Info("Job Creation")
-		if err != nil {
-			return ctrl.Result{
-				RequeueAfter: 10 * time.Second,
-			}, fmt.Errorf("error during compaction job creation: %v", err)
+
+		if lc.config.CompactionEnabled {
+			// Required job doesn't exist. Create new
+			job, err = lc.createCompactJob(ctx, logger, etcd)
+			logger.Info("Job Creation")
+			if err != nil {
+				return ctrl.Result{
+					RequeueAfter: 10 * time.Second,
+				}, fmt.Errorf("error during compaction job creation: %v", err)
+			}
 		}
 	}
 
@@ -529,9 +532,10 @@ func (lc *CompactionLeaseController) SetupWithManager(mgr ctrl.Manager, workers
 		MaxConcurrentReconciles: workers,
 	})
 
-	builder = builder.WithEventFilter(buildPredicateForLC()).For(&druidv1alpha1.Etcd{})
-	builder = builder.Owns(&coordinationv1.Lease{})
-	return builder.Complete(lc)
+	return builder.
+		WithEventFilter(buildPredicateForLC()).For(&druidv1alpha1.Etcd{}).
+		Owns(&coordinationv1.Lease{}).
+		Complete(lc)
 }
 
 func buildPredicateForLC() predicate.Predicate {

controllers/config/compaction.go

@@ -18,6 +18,8 @@ import "time"
 
 // CompactionLeaseConfig contains configuration for the compaction controller.
 type CompactionLeaseConfig struct {
+	// CompactionEnabled defines of compaction jobs should be created.
+	CompactionEnabled bool
 	// ActiveDeadlineDuration is the duration after which a running compaction job will be killed (Ex: "300ms", "20s", "-1.5h" or "2h45m")
 	ActiveDeadlineDuration time.Duration
 	// EventsThreshold is total number of etcd events that can be allowed before a backup compaction job is triggered

controllers/controllers_suite_test.go

@@ -127,6 +127,7 @@ var _ = BeforeSuite(func(done Done) {
 	Expect(err).NotTo(HaveOccurred())
 
 	lc, err := NewCompactionLeaseControllerWithImageVector(mgr, controllersconfig.CompactionLeaseConfig{
+		CompactionEnabled:      true,
 		EventsThreshold:        1000000,
 		ActiveDeadlineDuration: activeDeadlineDuration,
 	})

main.go

@@ -52,6 +52,7 @@ func main() {
 	var (
 		metricsAddr                string
 		enableLeaderElection       bool
+		enableBackupCompaction     bool
 		leaderElectionID           string
 		leaderElectionResourceLock string
 		etcdWorkers                int
@@ -75,6 +76,8 @@ func main() {
 	flag.IntVar(&custodianWorkers, "custodian-workers", 3, "Number of worker threads of the custodian controller.")
 	flag.IntVar(&etcdCopyBackupsTaskWorkers, "etcd-copy-backups-task-workers", 3, "Number of worker threads of the EtcdCopyBackupsTask controller.")
 	flag.DurationVar(&custodianSyncPeriod, "custodian-sync-period", 30*time.Second, "Sync period of the custodian controller.")
+	flag.BoolVar(&enableBackupCompaction, "enable-backup-compaction", false,
+		"Enable automatic compaction of etcd backups.")
 	flag.IntVar(&compactionWorkers, "compaction-workers", 3, "Number of worker threads of the CompactionJob controller. The controller creates a backup compaction job if a certain etcd event threshold is reached. Setting this flag to 0 disabled the controller.")
 	flag.Int64Var(&eventsThreshold, "etcd-events-threshold", 1000000, "Total number of etcd events that can be allowed before a backup compaction job is triggered.")
 	flag.DurationVar(&activeDeadlineDuration, "active-deadline-duration", 3*time.Hour, "Duration after which a running backup compaction job will be killed (Ex: \"300ms\", \"20s\", \"-1.5h\" or \"2h45m\").")
@@ -148,6 +151,7 @@ func main() {
 	}
 
 	lc, err := controllers.NewCompactionLeaseControllerWithImageVector(mgr, controllersconfig.CompactionLeaseConfig{
+		CompactionEnabled:      enableBackupCompaction,
 		EventsThreshold:        eventsThreshold,
 		ActiveDeadlineDuration: activeDeadlineDuration,
 	})