autobackup-operator/internal/controller/cluster_controller.go

package controller

import (
	"context"
	"errors"

	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
)

// CNPG typed API is deliberately not imported: its module graph (barman-cloud,
// cnpg-i/gRPC, prometheus-operator) is heavy to vendor, so Clusters and
// ScheduledBackups are handled as unstructured objects.
var (
	clusterGVK         = schema.GroupVersionKind{Group: "postgresql.cnpg.io", Version: "v1", Kind: "Cluster"}
	scheduledBackupGVK = schema.GroupVersionKind{Group: "postgresql.cnpg.io", Version: "v1", Kind: "ScheduledBackup"}
)

func newCluster() *unstructured.Unstructured {
	u := &unstructured.Unstructured{}
	u.SetGroupVersionKind(clusterGVK)
	return u
}

func newScheduledBackup() *unstructured.Unstructured {
	u := &unstructured.Unstructured{}
	u.SetGroupVersionKind(scheduledBackupGVK)
	return u
}

// ClusterReconciler backs up annotated CloudNativePG Clusters by provisioning an
// S3 bucket (via cephrgw), patching spec.backup.barmanObjectStore and creating a
// ScheduledBackup.
type ClusterReconciler struct {
	baseReconciler
}

// +kubebuilder:rbac:groups=postgresql.cnpg.io,resources=clusters,verbs=get;list;watch;update;patch
// +kubebuilder:rbac:groups=postgresql.cnpg.io,resources=scheduledbackups,verbs=get;list;watch;create;update;patch;delete

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

	cluster := newCluster()
	if err := r.Get(ctx, req.NamespacedName, cluster); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	schedule := cluster.GetAnnotations()[annSchedule]
	if schedule == "" {
		return r.teardown(ctx, cluster)
	}

	cnpgSchedule, err := ScheduleForCNPG(schedule)
	if err != nil {
		r.Recorder.Event(cluster, corev1.EventTypeWarning, "InvalidSchedule", err.Error())
		return ctrl.Result{}, nil
	}

	destName := cluster.GetAnnotations()[annDestination]
	if destName == "" {
		r.Recorder.Event(cluster, 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(cluster, corev1.EventTypeWarning, "UnknownDestination", err.Error())
			return ctrl.Result{RequeueAfter: requeueShort}, nil
		}
		return ctrl.Result{}, err
	}

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

	if err := r.patchBarman(ctx, cluster, credSecret, dest); err != nil {
		return ctrl.Result{}, err
	}

	if err := r.ensureScheduledBackup(ctx, cluster, cnpgSchedule); err != nil {
		return ctrl.Result{}, err
	}

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

func (r *ClusterReconciler) teardown(ctx context.Context, cluster *unstructured.Unstructured) (ctrl.Result, error) {
	sb := newScheduledBackup()
	sb.SetName(scheduleName(cluster.GetName()))
	sb.SetNamespace(cluster.GetNamespace())
	return ctrl.Result{}, deleteIfExists(ctx, r.Client, sb)
}

// patchBarman idempotently sets spec.backup.barmanObjectStore on the Cluster. If
// the user already set a different destinationPath it is left untouched and a
// Warning event is emitted instead of clobbering it.
func (r *ClusterReconciler) patchBarman(ctx context.Context, cluster *unstructured.Unstructured, credSecret string, dest Destination) error {
	bucket := bucketName(cluster.GetNamespace(), cluster.GetName())
	desiredPath := "s3://" + bucket

	existing, _, _ := unstructured.NestedString(cluster.Object, "spec", "backup", "barmanObjectStore", "destinationPath")
	if existing != "" && existing != desiredPath {
		r.Recorder.Eventf(cluster, corev1.EventTypeWarning, "DestinationConflict",
			"cluster already configured with destinationPath %q; not overwriting", existing)
		return nil
	}

	patch := client.MergeFrom(cluster.DeepCopy())

	barman := map[string]interface{}{
		"destinationPath": desiredPath,
		"endpointURL":     dest.Endpoint,
		"serverName":      cluster.GetName(),
		"s3Credentials": map[string]interface{}{
			"accessKeyId":     map[string]interface{}{"name": credSecret, "key": "AWS_ACCESS_KEY_ID"},
			"secretAccessKey": map[string]interface{}{"name": credSecret, "key": "AWS_SECRET_ACCESS_KEY"},
		},
	}
	if dest.EndpointCASecret != "" {
		key := dest.EndpointCAKey
		if key == "" {
			key = "ca.crt"
		}
		barman["endpointCA"] = map[string]interface{}{"name": dest.EndpointCASecret, "key": key}
	}
	if err := unstructured.SetNestedMap(cluster.Object, barman, "spec", "backup", "barmanObjectStore"); err != nil {
		return err
	}
	if rp, _, _ := unstructured.NestedString(cluster.Object, "spec", "backup", "retentionPolicy"); rp == "" {
		if err := unstructured.SetNestedField(cluster.Object, "30d", "spec", "backup", "retentionPolicy"); err != nil {
			return err
		}
	}
	return r.Patch(ctx, cluster, patch)
}

func (r *ClusterReconciler) ensureScheduledBackup(ctx context.Context, cluster *unstructured.Unstructured, schedule string) error {
	sb := newScheduledBackup()
	sb.SetName(scheduleName(cluster.GetName()))
	sb.SetNamespace(cluster.GetNamespace())
	_, err := controllerutil.CreateOrUpdate(ctx, r.Client, sb, func() error {
		for _, f := range []struct {
			value string
			path  []string
		}{
			{cluster.GetName(), []string{"spec", "cluster", "name"}},
			{schedule, []string{"spec", "schedule"}},
			{"barmanObjectStore", []string{"spec", "method"}},
			{"self", []string{"spec", "backupOwnerReference"}},
		} {
			if err := unstructured.SetNestedField(sb.Object, f.value, f.path...); err != nil {
				return err
			}
		}
		return controllerutil.SetControllerReference(cluster, sb, r.Scheme)
	})
	return err
}

func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(newCluster()).
		Owns(newScheduledBackup()).
		Named("cluster-autobackup").
		Complete(r)
}