Merge pull request #784 from k8up-io/remove-obs-other

Free Schedule, Restore and Archive controllers from internal observer

Author: ccremer

api/v1/archive_types.go

@@ -1,6 +1,8 @@
 package v1
 
 import (
+	"reflect"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -137,3 +139,7 @@ func (in *ArchiveSchedule) GetSchedule() ScheduleDefinition {
 func (in *ArchiveSchedule) GetObjectCreator() ObjectCreator {
 	return in
 }
+
+var (
+	ArchiveKind = reflect.TypeOf(Archive{}).Name()
+)

api/v1/restore_types.go

@@ -1,6 +1,8 @@
 package v1
 
 import (
+	"reflect"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -168,3 +170,7 @@ type RestoreList struct {
 func init() {
 	SchemeBuilder.Register(&Restore{}, &RestoreList{})
 }
+
+var (
+	RestoreKind = reflect.TypeOf(Restore{}).Name()
+)

cmd/operator/main.go

@@ -4,9 +4,12 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/k8up-io/k8up/v2/operator/archivecontroller"
 	"github.com/k8up-io/k8up/v2/operator/backupcontroller"
 	"github.com/k8up-io/k8up/v2/operator/jobcontroller"
 	"github.com/k8up-io/k8up/v2/operator/locker"
+	"github.com/k8up-io/k8up/v2/operator/restorecontroller"
+	"github.com/k8up-io/k8up/v2/operator/schedulecontroller"
 	"k8s.io/apimachinery/pkg/api/resource"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 
@@ -117,10 +120,10 @@ func operatorMain(c *cli.Context) error {
 	executor.StartExecutor(lock)
 
 	for name, reconciler := range map[string]controllers.ReconcilerSetup{
-		"Schedule": &controllers.ScheduleReconciler{},
+		"Schedule": &schedulecontroller.ScheduleReconciler{},
 		"Backup":   &backupcontroller.BackupReconciler{},
-		"Restore":  &controllers.RestoreReconciler{},
-		"Archive":  &controllers.ArchiveReconciler{},
+		"Restore":  &restorecontroller.RestoreReconciler{},
+		"Archive":  &archivecontroller.ArchiveReconciler{},
 		"Check":    &controllers.CheckReconciler{},
 		"Prune":    &controllers.PruneReconciler{},
 		"Job":      &jobcontroller.JobReconciler{},

controllers/archive_controller.go

@@ -6,8 +6,6 @@ image:
   repository: $E2E_REPO
   tag: $E2E_TAG
 k8up:
-  env:
+  envVars:
     - name: K8UP_DEBUG
       value: "true"
-    - name: BACKUP_ENABLE_LEADER_ELECTION
-      value: $BACKUP_ENABLE_LEADER_ELECTION

controllers/restore_controller.go

@@ -96,7 +96,6 @@ prepare() {
 
 	replace_in_file "${target_file}" E2E_IMAGE "'${E2E_IMAGE}'"
 	replace_in_file "${target_file}" ID "$(id -u)"
-	replace_in_file "${target_file}" BACKUP_ENABLE_LEADER_ELECTION "'${BACKUP_ENABLE_LEADER_ELECTION}'"
 	replace_in_file "${target_file}" BACKUP_FILE_NAME "${BACKUP_FILE_NAME}"
 	replace_in_file "${target_file}" BACKUP_FILE_CONTENT "${BACKUP_FILE_CONTENT}"
 }

controllers/schedule_controller.go

@@ -0,0 +1,54 @@
+package archivecontroller
+
+import (
+	"context"
+	"time"
+
+	k8upv1 "github.com/k8up-io/k8up/v2/api/v1"
+	"github.com/k8up-io/k8up/v2/operator/cfg"
+	"github.com/k8up-io/k8up/v2/operator/job"
+	"github.com/k8up-io/k8up/v2/operator/queue"
+	"k8s.io/apimachinery/pkg/api/errors"
+	controllerruntime "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// ArchiveReconciler reconciles a Archive object
+type ArchiveReconciler struct {
+	Kube client.Client
+}
+
+// Reconcile is the entrypoint to manage the given resource.
+func (r *ArchiveReconciler) Reconcile(ctx context.Context, req controllerruntime.Request) (controllerruntime.Result, error) {
+	log := controllerruntime.LoggerFrom(ctx)
+
+	archive := &k8upv1.Archive{}
+	err := r.Kube.Get(ctx, req.NamespacedName, archive)
+	if err != nil {
+		if errors.IsNotFound(err) {
+			return controllerruntime.Result{}, nil
+		}
+		log.Error(err, "Failed to get Archive")
+		return controllerruntime.Result{}, err
+	}
+
+	repository := cfg.Config.GetGlobalRepository()
+	if archive.Spec.Backend != nil {
+		repository = archive.Spec.Backend.String()
+	}
+	if archive.Spec.RestoreSpec == nil {
+		archive.Spec.RestoreSpec = &k8upv1.RestoreSpec{}
+	}
+	config := job.NewConfig(ctx, r.Kube, log, archive, repository)
+
+	executor := NewArchiveExecutor(config)
+
+	if archive.Status.HasFinished() {
+		executor.cleanupOldArchives(executor.GetJobNamespacedName(), archive)
+		return controllerruntime.Result{}, nil
+	}
+
+	log.V(1).Info("adding job to the queue")
+	queue.GetExecQueue().Add(executor)
+	return controllerruntime.Result{RequeueAfter: time.Second * 30}, nil
+}