feat: lenient decoding of CNPG resources

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)