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) }