marco/plugin-barman-cloud
1
0
Fork 0
mirror of https://github.com/cloudnative-pg/plugin-barman-cloud.git synced 2026-01-11 13:23:09 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: lenient decoding of CNPG resources (#192)

Browse Source 
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 <leonardo.cecchi@enterprisedb.com>
Signed-off-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>
Co-authored-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>
This commit is contained in:
Leonardo Cecchi 2025-03-14 12:23:23 +01:00 committed by GitHub
parent fcbc472092
commit 13e3fab268
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 109 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
go.mod
View File

@ -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

4
go.sum
View File

@ -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=

2
internal/cmd/instance/main.go
View File

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

44
internal/cnpgi/instance/manager.go
View File

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

11
internal/cnpgi/operator/config/config.go
View File

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

29
internal/cnpgi/operator/lifecycle.go
View File

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

39
internal/cnpgi/operator/ownership.go Normal file
View File

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

13
internal/cnpgi/operator/reconciler.go
View File

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