autobackup-operator/internal/controller/pvc_controller.go

package controller

import (
	"context"
	"errors"
	"strings"

	k8upv1 "github.com/k8up-io/k8up/v2/api/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/event"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
)

// PVCReconciler backs up annotated PersistentVolumeClaims by provisioning an S3
// bucket (via cephrgw) and a k8up Schedule scoped to the PVC.
type PVCReconciler struct {
	baseReconciler
}

// +kubebuilder:rbac:groups="",resources=persistentvolumeclaims,verbs=get;list;watch
// +kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch;create;update;patch
// +kubebuilder:rbac:groups="",resources=configmaps,verbs=get;list;watch
// +kubebuilder:rbac:groups="",resources=events,verbs=create;patch
// +kubebuilder:rbac:groups=ceph.unkin.net,resources=objectstoreusers;buckets;bucketaccesses,verbs=get;list;watch;create;update;patch
// +kubebuilder:rbac:groups=ceph.unkin.net,resources=bucketaccesses/status,verbs=get
// +kubebuilder:rbac:groups=k8up.io,resources=schedules,verbs=get;list;watch;create;update;patch;delete

func (r *PVCReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	var pvc corev1.PersistentVolumeClaim
	if err := r.Get(ctx, req.NamespacedName, &pvc); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	schedule := pvc.Annotations[annSchedule]
	if schedule == "" {
		return r.teardown(ctx, &pvc)
	}

	k8upSchedule, err := ScheduleForK8up(schedule)
	if err != nil {
		r.Recorder.Event(&pvc, corev1.EventTypeWarning, "InvalidSchedule", err.Error())
		return ctrl.Result{}, nil
	}

	destName := pvc.Annotations[annDestination]
	if destName == "" {
		r.Recorder.Event(&pvc, corev1.EventTypeWarning, "MissingDestination", "backups.unkin.net/destination annotation is required")
		return ctrl.Result{}, nil
	}
	dest, err := r.resolveDest(ctx, destName)
	if err != nil {
		if errors.Is(err, errDestinationNotFound) {
			r.Recorder.Event(&pvc, corev1.EventTypeWarning, "UnknownDestination", err.Error())
			return ctrl.Result{RequeueAfter: requeueShort}, nil
		}
		return ctrl.Result{}, err
	}

	ready, credSecret, err := r.ensureBucketStack(ctx, &pvc, dest)
	if err != nil {
		return ctrl.Result{}, err
	}
	if !ready {
		return ctrl.Result{RequeueAfter: requeueShort}, nil
	}

	resticSecret, err := r.ensureResticSecret(ctx, &pvc)
	if err != nil {
		return ctrl.Result{}, err
	}

	if err := r.ensureSchedule(ctx, &pvc, k8upSchedule, credSecret, resticSecret, dest); err != nil {
		return ctrl.Result{}, err
	}

	logger.Info("pvc backup reconciled", "pvc", pvc.Name, "destination", destName)
	return ctrl.Result{RequeueAfter: requeueSteady}, nil
}

// teardown removes the k8up Schedule when the annotation is gone, and (only on
// explicit opt-in) the bucket stack. Bucket data is retained by default.
func (r *PVCReconciler) teardown(ctx context.Context, pvc *corev1.PersistentVolumeClaim) (ctrl.Result, error) {
	if err := deleteIfExists(ctx, r.Client, &k8upv1.Schedule{
		ObjectMeta: metav1.ObjectMeta{Name: scheduleName(pvc.Name), Namespace: pvc.Namespace},
	}); err != nil {
		return ctrl.Result{}, err
	}
	if !strings.EqualFold(pvc.Annotations[annPurge], "true") {
		return ctrl.Result{}, nil
	}
	for _, obj := range bucketStackObjects(pvc.Namespace, pvc.Name) {
		if err := deleteIfExists(ctx, r.Client, obj); err != nil {
			return ctrl.Result{}, err
		}
	}
	return ctrl.Result{}, nil
}

func (r *PVCReconciler) ensureSchedule(ctx context.Context, pvc *corev1.PersistentVolumeClaim, schedule, credSecret, resticSecret string, dest Destination) error {
	sched := &k8upv1.Schedule{ObjectMeta: metav1.ObjectMeta{Name: scheduleName(pvc.Name), Namespace: pvc.Namespace}}
	_, err := controllerutil.CreateOrUpdate(ctx, r.Client, sched, func() error {
		sched.Spec.Backend = &k8upv1.Backend{
			RepoPasswordSecretRef: secretKeyRef(resticSecret, "password"),
			S3: &k8upv1.S3Spec{
				Endpoint:                 dest.Endpoint,
				Bucket:                   bucketName(pvc.Namespace, pvc.Name),
				AccessKeyIDSecretRef:     secretKeyRef(credSecret, "AWS_ACCESS_KEY_ID"),
				SecretAccessKeySecretRef: secretKeyRef(credSecret, "AWS_SECRET_ACCESS_KEY"),
			},
		}
		sched.Spec.Backup = &k8upv1.BackupSchedule{
			ScheduleCommon: &k8upv1.ScheduleCommon{Schedule: k8upv1.ScheduleDefinition(schedule)},
			BackupSpec: k8upv1.BackupSpec{
				RunnableSpec: k8upv1.RunnableSpec{
					Volumes: &[]k8upv1.RunnableVolumeSpec{{
						Name:                  "backup-source",
						PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: pvc.Name},
					}},
				},
			},
		}
		sched.Spec.Prune = &k8upv1.PruneSchedule{
			ScheduleCommon: &k8upv1.ScheduleCommon{Schedule: k8upv1.ScheduleDefinition("@weekly")},
			PruneSpec:      k8upv1.PruneSpec{Retention: k8upv1.RetentionPolicy{KeepDaily: 7, KeepWeekly: 4}},
		}
		return controllerutil.SetControllerReference(pvc, sched, r.Scheme)
	})
	return err
}

func secretKeyRef(name, key string) *corev1.SecretKeySelector {
	return &corev1.SecretKeySelector{
		LocalObjectReference: corev1.LocalObjectReference{Name: name},
		Key:                  key,
	}
}

// hasScheduleAnnotation reports whether an object carries the schedule
// annotation that marks it for management.
func hasScheduleAnnotation(o client.Object) bool {
	_, ok := o.GetAnnotations()[annSchedule]
	return ok
}

// schedulePredicate limits reconciles to PVCs that carry (or, on update, used to
// carry) the schedule annotation, so unannotated PVCs no longer churn the queue
// on every resync while the annotation-removed teardown path still fires.
var schedulePredicate = predicate.Funcs{
	CreateFunc:  func(e event.CreateEvent) bool { return hasScheduleAnnotation(e.Object) },
	DeleteFunc:  func(e event.DeleteEvent) bool { return hasScheduleAnnotation(e.Object) },
	GenericFunc: func(e event.GenericEvent) bool { return hasScheduleAnnotation(e.Object) },
	UpdateFunc: func(e event.UpdateEvent) bool {
		return hasScheduleAnnotation(e.ObjectOld) || hasScheduleAnnotation(e.ObjectNew)
	},
}

func (r *PVCReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&corev1.PersistentVolumeClaim{}, builder.WithPredicates(schedulePredicate)).
		Owns(&k8upv1.Schedule{}).
		Named("pvc-autobackup").
		Complete(r)
}