feat: lenient decoding of CNPG resources (#192)

This patch enables the barman-cloud plugin to function with an operator
that is structurally identical to CNPG but works with a different API group.

It achieves this through lenient decoding of the provided CNPG resources
and injecting the detected GVK into the sidecar, enabling it to correctly
encode and decode the Kubernetes resources.

Signed-off-by: Leonardo Cecchi <[email protected]>
Signed-off-by: Marco Nenciarini <[email protected]>
Co-authored-by: Marco Nenciarini <[email protected]>

Author: leonardoce

go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/cloudnative-pg/barman-cloud v0.2.0
 	github.com/cloudnative-pg/cloudnative-pg v1.25.1
 	github.com/cloudnative-pg/cnpg-i v0.1.0
-	github.com/cloudnative-pg/cnpg-i-machinery v0.1.2
+	github.com/cloudnative-pg/cnpg-i-machinery v0.2.0
 	github.com/cloudnative-pg/machinery v0.1.0
 	github.com/onsi/ginkgo/v2 v2.23.0
 	github.com/onsi/gomega v1.36.2

go.sum

@@ -26,8 +26,8 @@ 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/cnpg-i-machinery v0.2.0 h1:htNuKirdAOYrc7Hu5mLDoOES+nKSyPaXNDLgbV5dLSI=
+github.com/cloudnative-pg/cnpg-i-machinery v0.2.0/go.mod h1:MHVxMMbLeCRnEM8PLWW4C2CsHqOeAU2OsrwWMKy3tPA=
 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=

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
 }

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

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
 	}
 

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 ""
+}

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

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)