From f936bf128fb5e00204f616834466e1a81a0e4421 Mon Sep 17 00:00:00 2001 From: Leonardo Cecchi Date: Fri, 7 Mar 2025 16:37:50 +0100 Subject: [PATCH] feat: lenient decoding of CNPG resources This patch allows the barman-cloud plugin to work with operator being structurally identical with CNPG but with a different API group. It does that by using lenient decoding of the passed CNPG resources and by injecting the detected GVK to the sidecar, that uses it to properly encode and decode the Kubernetes resources. Signed-off-by: Leonardo Cecchi --- go.mod | 2 ++ go.sum | 4 +-- internal/cmd/instance/main.go | 2 ++ internal/cnpgi/instance/manager.go | 44 +++++++++++++++++++----- internal/cnpgi/operator/config/config.go | 11 +----- internal/cnpgi/operator/lifecycle.go | 29 +++++++++++++--- internal/cnpgi/operator/ownership.go | 39 +++++++++++++++++++++ internal/cnpgi/operator/reconciler.go | 13 +++---- 8 files changed, 110 insertions(+), 34 deletions(-) create mode 100644 internal/cnpgi/operator/ownership.go diff --git a/go.mod b/go.mod index f99838c..233661a 100644 --- a/go.mod +++ b/go.mod @@ -136,3 +136,5 @@ require ( sigs.k8s.io/structured-merge-diff/v4 v4.5.0 // indirect sigs.k8s.io/yaml v1.4.0 // indirect ) + +replace github.com/cloudnative-pg/cnpg-i-machinery => github.com/leonardoce/cnpg-i-machinery v0.0.0-20250307135524-f7d0a34b5b3e diff --git a/go.sum b/go.sum index 5bfb28f..df19d9d 100644 --- a/go.sum +++ b/go.sum @@ -26,8 +26,6 @@ github.com/cloudnative-pg/cloudnative-pg v1.25.1 h1:Yc6T7ikQ1AiWXBQht+6C3DoihrIp github.com/cloudnative-pg/cloudnative-pg v1.25.1/go.mod h1:96b9bRFLSr3uFWHjhytPdcvKIKwy9H6AG7cH0O6jefs= github.com/cloudnative-pg/cnpg-i v0.1.0 h1:QH2xTsrODMhEEc6B25GbOYe7ZIttDmSkYvXotfU5dfs= github.com/cloudnative-pg/cnpg-i v0.1.0/go.mod h1:G28BhgUEHqrxEyyQeHz8BbpMVAsGuLhJm/tHUbDi8Sw= -github.com/cloudnative-pg/cnpg-i-machinery v0.1.2 h1:yY8tBkN8l8ENNWDMK0ZewK+nNzsxuSvxbSfkwJoSSZ0= -github.com/cloudnative-pg/cnpg-i-machinery v0.1.2/go.mod h1:4Lf5Vfl8tvCsgs7H38+JMkvFhUMIDiNoZtzfwqyFE+E= github.com/cloudnative-pg/machinery v0.1.0 h1:tjRmsqQmsO/OlaT0uFmkEtVqgr+SGPM88cKZOHYKLBo= github.com/cloudnative-pg/machinery v0.1.0/go.mod h1:0V3vm44FaIsY+x4pm8ORry7xCC3AJiO+ebfPNxeP5Ck= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= @@ -116,6 +114,8 @@ github.com/kubernetes-csi/external-snapshotter/client/v8 v8.2.0 h1:Q3jQ1NkFqv5o+ github.com/kubernetes-csi/external-snapshotter/client/v8 v8.2.0/go.mod h1:E3vdYxHj2C2q6qo8/Da4g7P+IcwqRZyy3gJBzYybV9Y= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= +github.com/leonardoce/cnpg-i-machinery v0.0.0-20250307135524-f7d0a34b5b3e h1:j8lLGGMIvNqPKIWYAVUtz+OspxXwxK/s/jdg18ZuaGU= +github.com/leonardoce/cnpg-i-machinery v0.0.0-20250307135524-f7d0a34b5b3e/go.mod h1:4Lf5Vfl8tvCsgs7H38+JMkvFhUMIDiNoZtzfwqyFE+E= github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY= diff --git a/internal/cmd/instance/main.go b/internal/cmd/instance/main.go index 7066290..49ae294 100644 --- a/internal/cmd/instance/main.go +++ b/internal/cmd/instance/main.go @@ -36,6 +36,8 @@ func NewCmd() *cobra.Command { _ = viper.BindEnv("pod-name", "POD_NAME") _ = viper.BindEnv("pgdata", "PGDATA") _ = viper.BindEnv("spool-directory", "SPOOL_DIRECTORY") + _ = viper.BindEnv("custom-cnpg-group", "CUSTOM_CNPG_GROUP") + _ = viper.BindEnv("custom-cnpg-version", "CUSTOM_CNPG_VERSIONXS") return cmd } diff --git a/internal/cnpgi/instance/manager.go b/internal/cnpgi/instance/manager.go index 757c937..fbd3602 100644 --- a/internal/cnpgi/instance/manager.go +++ b/internal/cnpgi/instance/manager.go @@ -8,26 +8,22 @@ import ( "github.com/spf13/viper" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" utilruntime "k8s.io/apimachinery/pkg/util/runtime" clientgoscheme "k8s.io/client-go/kubernetes/scheme" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/scheme" barmancloudv1 "github.com/cloudnative-pg/plugin-barman-cloud/api/v1" extendedclient "github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/instance/internal/client" ) -var scheme = runtime.NewScheme() - -func init() { - utilruntime.Must(barmancloudv1.AddToScheme(scheme)) - utilruntime.Must(cnpgv1.AddToScheme(scheme)) - utilruntime.Must(clientgoscheme.AddToScheme(scheme)) -} - // Start starts the sidecar informers and CNPG-i server func Start(ctx context.Context) error { + scheme := generateScheme(ctx) + setupLog := log.FromContext(ctx) setupLog.Info("Starting barman cloud instance plugin") podName := viper.GetString("pod-name") @@ -70,3 +66,35 @@ func Start(ctx context.Context) error { return nil } + +// generateScheme creates a runtime.Scheme object with all the +// definition needed to support the sidecar. This allows +// the plugin to be used in every CNPG-based operator. +func generateScheme(ctx context.Context) *runtime.Scheme { + result := runtime.NewScheme() + + utilruntime.Must(barmancloudv1.AddToScheme(result)) + utilruntime.Must(clientgoscheme.AddToScheme(result)) + + cnpgGroup := viper.GetString("custom-cnpg-group") + cnpgVersion := viper.GetString("custom-cnpg-version") + if len(cnpgGroup) == 0 { + cnpgGroup = cnpgv1.SchemeGroupVersion.Group + } + if len(cnpgVersion) == 0 { + cnpgVersion = cnpgv1.SchemeGroupVersion.Version + } + + // Proceed with custom registration of the CNPG scheme + schemeGroupVersion := schema.GroupVersion{Group: cnpgGroup, Version: cnpgVersion} + schemeBuilder := &scheme.Builder{GroupVersion: schemeGroupVersion} + schemeBuilder.Register(&cnpgv1.Cluster{}, &cnpgv1.ClusterList{}) + schemeBuilder.Register(&cnpgv1.Backup{}, &cnpgv1.BackupList{}) + schemeBuilder.Register(&cnpgv1.ScheduledBackup{}, &cnpgv1.ScheduledBackupList{}) + utilruntime.Must(schemeBuilder.AddToScheme(result)) + + schemeLog := log.FromContext(ctx) + schemeLog.Info("CNPG types registration", "schemeGroupVersion", schemeGroupVersion) + + return result +} diff --git a/internal/cnpgi/operator/config/config.go b/internal/cnpgi/operator/config/config.go index 1140d03..4c21ae8 100644 --- a/internal/cnpgi/operator/config/config.go +++ b/internal/cnpgi/operator/config/config.go @@ -5,7 +5,6 @@ import ( cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1" "github.com/cloudnative-pg/cnpg-i-machinery/pkg/pluginhelper/decoder" - "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/metadata" @@ -101,19 +100,11 @@ func (config *PluginConfiguration) GetReferredBarmanObjectsKey() []types.Namespa return result } -func getClusterGVK() schema.GroupVersionKind { - return schema.GroupVersionKind{ - Group: cnpgv1.SchemeGroupVersion.Group, - Version: cnpgv1.SchemeGroupVersion.Version, - Kind: cnpgv1.ClusterKind, - } -} - // NewFromClusterJSON decodes a JSON representation of a cluster. func NewFromClusterJSON(clusterJSON []byte) (*PluginConfiguration, error) { var result cnpgv1.Cluster - if err := decoder.DecodeObject(clusterJSON, &result, getClusterGVK()); err != nil { + if err := decoder.DecodeObjectLenient(clusterJSON, &result); err != nil { return nil, err } diff --git a/internal/cnpgi/operator/lifecycle.go b/internal/cnpgi/operator/lifecycle.go index 040048c..9ff6bde 100644 --- a/internal/cnpgi/operator/lifecycle.go +++ b/internal/cnpgi/operator/lifecycle.go @@ -4,9 +4,9 @@ import ( "context" "errors" "fmt" + "strings" cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1" - "github.com/cloudnative-pg/cloudnative-pg/pkg/utils" "github.com/cloudnative-pg/cnpg-i-machinery/pkg/pluginhelper/decoder" "github.com/cloudnative-pg/cnpg-i-machinery/pkg/pluginhelper/object" "github.com/cloudnative-pg/cnpg-i/pkg/lifecycle" @@ -77,10 +77,9 @@ func (impl LifecycleImplementation) LifecycleHook( } var cluster cnpgv1.Cluster - if err := decoder.DecodeObject( + if err := decoder.DecodeObjectLenient( request.GetClusterDefinition(), &cluster, - cnpgv1.SchemeGroupVersion.WithKind("Cluster"), ); err != nil { return nil, err } @@ -138,7 +137,7 @@ func reconcileJob( } var job batchv1.Job - if err := decoder.DecodeObject( + if err := decoder.DecodeObjectStrict( request.GetObjectDefinition(), &job, batchv1.SchemeGroupVersion.WithKind("Job"), @@ -151,7 +150,7 @@ func reconcileJob( WithValues("jobName", job.Name) contextLogger.Debug("starting job reconciliation") - if job.Spec.Template.Labels[utils.JobRoleLabelName] != "full-recovery" { + if getCNPGJobRole(&job) != "full-recovery" { contextLogger.Debug("job is not a recovery job, skipping") return nil, nil } @@ -270,6 +269,14 @@ func reconcilePodSpec( Name: "SPOOL_DIRECTORY", Value: "/controller/wal-restore-spool", }, + { + Name: "CUSTOM_CNPG_GROUP", + Value: cluster.GetObjectKind().GroupVersionKind().Group, + }, + { + Name: "CUSTOM_CNPG_VERSION", + Value: cluster.GetObjectKind().GroupVersionKind().Version, + }, } envs = append(envs, additionalEnvs...) @@ -445,3 +452,15 @@ func volumeListHasVolume(volumes []corev1.Volume, name string) bool { return false } + +// getCNPGJobRole gets the role associated to a CNPG job +func getCNPGJobRole(job *batchv1.Job) string { + const jobRoleLabelSuffix = "/jobRole" + for k, v := range job.Spec.Template.Labels { + if strings.HasSuffix(k, jobRoleLabelSuffix) { + return v + } + } + + return "" +} diff --git a/internal/cnpgi/operator/ownership.go b/internal/cnpgi/operator/ownership.go new file mode 100644 index 0000000..2fb73a5 --- /dev/null +++ b/internal/cnpgi/operator/ownership.go @@ -0,0 +1,39 @@ +package operator + +import ( + "fmt" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/utils/ptr" +) + +// setOwnerReference explicitly set the owner reference between an +// owner object and a controller one. +// +// Important: this function won't use any registered scheme and will +// fail unless the metadata has been correctly set into the owner +// object. +func setOwnerReference(owner, controlled metav1.Object) error { + ro, ok := owner.(runtime.Object) + if !ok { + return fmt.Errorf("%T is not a runtime.Object, cannot call setOwnerReference", owner) + } + + if len(ro.DeepCopyObject().GetObjectKind().GroupVersionKind().Group) == 0 { + return fmt.Errorf("%T metadata have not been set, cannot call setOwnerReference", owner) + } + + controlled.SetOwnerReferences([]metav1.OwnerReference{ + { + APIVersion: ro.GetObjectKind().GroupVersionKind().GroupVersion().String(), + Kind: ro.GetObjectKind().GroupVersionKind().Kind, + Name: owner.GetName(), + UID: owner.GetUID(), + BlockOwnerDeletion: ptr.To(true), + Controller: ptr.To(true), + }, + }) + + return nil +} diff --git a/internal/cnpgi/operator/reconciler.go b/internal/cnpgi/operator/reconciler.go index bcb3a0f..610b9a7 100644 --- a/internal/cnpgi/operator/reconciler.go +++ b/internal/cnpgi/operator/reconciler.go @@ -11,7 +11,6 @@ import ( rbacv1 "k8s.io/api/rbac/v1" "k8s.io/apimachinery/pkg/api/equality" apierrs "k8s.io/apimachinery/pkg/api/errors" - ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" barmancloudv1 "github.com/cloudnative-pg/plugin-barman-cloud/api/v1" @@ -61,10 +60,10 @@ func (r ReconcilerImplementation) Pre( contextLogger.Debug("parsing cluster definition") var cluster cnpgv1.Cluster - if err := decoder.DecodeObject( + if err := decoder.DecodeObjectLenient( request.GetResourceDefinition(), &cluster, - cnpgv1.SchemeGroupVersion.WithKind("Cluster")); err != nil { + ); err != nil { return nil, err } @@ -142,11 +141,7 @@ func (r ReconcilerImplementation) ensureRole( "namespace", newRole.Namespace, ) - if err := ctrl.SetControllerReference( - cluster, - newRole, - r.Client.Scheme(), - ); err != nil { + if err := setOwnerReference(cluster, newRole); err != nil { return err } @@ -193,7 +188,7 @@ func (r ReconcilerImplementation) createRoleBinding( cluster *cnpgv1.Cluster, ) error { roleBinding := specs.BuildRoleBinding(cluster) - if err := ctrl.SetControllerReference(cluster, roleBinding, r.Client.Scheme()); err != nil { + if err := setOwnerReference(cluster, roleBinding); err != nil { return err } return r.Client.Create(ctx, roleBinding)