feat: implement CNPG-I plugin infrastructure (cloudnative-pg#3719)

Introduce foundational changes to support dynamic plugins in
adherence to the CloudNativePG Interface (CNPG-I).
This commit establishes the necessary infrastructure for
seamless plugin management.

Partially closes: cloudnative-pg#3699

Signed-off-by: Leonardo Cecchi <leonardo.cecchi@enterprisedb.com>
Signed-off-by: Armando Ruocco <armando.ruocco@enterprisedb.com>
Co-authored-by: Armando Ruocco <armando.ruocco@enterprisedb.com>

.wordlist-en-custom.txt

@@ -246,6 +246,7 @@ OnlineUpgrading
 OpenSSL
 OpenShift
 Openshift
+OperatorCapabilities
 OperatorGroup
 OperatorHub
 PGAudit
@@ -272,6 +273,8 @@ PgBouncerSecrets
 PgBouncerSecretsVersions
 PgBouncerSpec
 Philippe
+PluginConfigurationList
+PluginStatus
 PoLA
 PodAffinity
 PodAntiAffinity
@@ -893,6 +896,7 @@ openldap
 openshift
 operability
 operativity
+operatorCapabilities
 operatorframework
 operatorgorup
 operatorgroup
@@ -931,6 +935,7 @@ pid
 pitr
 plpgsql
 pluggable
+pluginStatus
 png
 podAffinityTerm
 podAntiAffinity
@@ -990,6 +995,7 @@ readService
 readinessProbe
 readthedocs
 readyInstances
+reconciler
 reconciliationLoop
 recoverability
 recoveredCluster

api/v1/cluster_plugins.go

@@ -0,0 +1,39 @@
+/*
+Copyright The CloudNativePG Contributors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1
+
+import (
+	"context"
+
+	"github.com/cloudnative-pg/cloudnative-pg/internal/cnpi/plugin/client"
+	"github.com/cloudnative-pg/cloudnative-pg/internal/configuration"
+)
+
+// NewLoader creates a new plugin client, loading the plugins that are required
+// by this cluster
+func (plugins PluginConfigurationList) NewLoader(ctx context.Context) (client.Client, error) {
+	pluginLoader := client.NewUnixSocketClient(configuration.Current.PluginSocketDir)
+
+	// Load the plugins
+	for _, pluginDeclaration := range plugins {
+		if err := pluginLoader.Load(ctx, pluginDeclaration.Name); err != nil {
+			return nil, err
+		}
+	}
+
+	return pluginLoader, nil
+}

api/v1/cluster_types.go

@@ -489,8 +489,16 @@ type ClusterSpec struct {
 	// The tablespaces configuration
 	// +optional
 	Tablespaces []TablespaceConfiguration `json:"tablespaces,omitempty"`
+
+	// The plugins configuration, containing
+	// any plugin to be loaded with the corresponding configuration
+	Plugins PluginConfigurationList `json:"plugins,omitempty"`
 }
 
+// PluginConfigurationList represent a set of plugin with their
+// configuration parameters
+type PluginConfigurationList []PluginConfiguration
+
 const (
 	// PhaseSwitchover when a cluster is changing the primary node
 	PhaseSwitchover = "Switchover in progress"
@@ -896,6 +904,9 @@ type ClusterStatus struct {
 	// Image contains the image name used by the pods
 	// +optional
 	Image string `json:"image,omitempty"`
+
+	// PluginStatus is the status of the loaded plugins
+	PluginStatus []PluginStatus `json:"pluginStatus,omitempty"`
 }
 
 // InstanceReportedState describes the last reported state of an instance during a reconciliation loop
@@ -2343,6 +2354,34 @@ type ManagedConfiguration struct {
 	Roles []RoleConfiguration `json:"roles,omitempty"`
 }
 
+// PluginConfiguration specifies a plugin that need to be loaded for this
+// cluster to be reconciled
+type PluginConfiguration struct {
+	// Name is the plugin name
+	Name string `json:"name"`
+
+	// Parameters is the configuration of the plugin
+	Parameters map[string]string `json:"parameters,omitempty"`
+}
+
+// PluginStatus is the status of a loaded plugin
+type PluginStatus struct {
+	// Name is the name of the plugin
+	Name string `json:"name"`
+
+	// Version is the version of the plugin loaded by the
+	// latest reconciliation loop
+	Version string `json:"version"`
+
+	// Capabilities are the list of capabilities of the
+	// plugin
+	Capabilities []string `json:"capabilities,omitempty"`
+
+	// OperatorCapabilities are the list of capabilities of the
+	// plugin regarding the reconciler
+	OperatorCapabilities []string `json:"operatorCapabilities,omitempty"`
+}
+
 // RoleConfiguration is the representation, in Kubernetes, of a PostgreSQL role
 // with the additional field Ensure specifying whether to ensure the presence or
 // absence of the role in the database

api/v1/cluster_webhook.go

@@ -17,6 +17,7 @@ limitations under the License.
 package v1
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"strconv"
@@ -162,6 +163,27 @@ func (r *Cluster) setDefaults(preserveUserSettings bool) {
 	if len(r.Spec.Tablespaces) > 0 {
 		r.defaultTablespaces()
 	}
+
+	ctx := context.Background()
+
+	// Call the plugins to help with defaulting this cluster
+	contextLogger := log.FromContext(ctx)
+	pluginClient, err := r.Spec.Plugins.NewLoader(ctx)
+	if err != nil {
+		contextLogger.Error(err, "Error invoking plugin in the defaulting webhook, skipping")
+		return
+	}
+	defer func() {
+		pluginClient.Close(ctx)
+	}()
+
+	var mutatedCluster Cluster
+	if err := pluginClient.MutateCluster(ctx, r, &mutatedCluster); err != nil {
+		contextLogger.Error(err, "Error invoking plugin in the defaulting webhook, skipping")
+		return
+	}
+
+	mutatedCluster.DeepCopyInto(r)
 }
 
 // defaultTablespaces adds the tablespace owner where the
@@ -286,6 +308,26 @@ var _ webhook.Validator = &Cluster{}
 func (r *Cluster) ValidateCreate() (admission.Warnings, error) {
 	clusterLog.Info("validate create", "name", r.Name, "namespace", r.Namespace)
 	allErrs := r.Validate()
+
+	// Call the plugins to help validating this cluster creation
+	ctx := context.Background()
+	contextLogger := log.FromContext(ctx)
+	pluginClient, err := r.Spec.Plugins.NewLoader(ctx)
+	if err != nil {
+		contextLogger.Error(err, "Error invoking plugin in the validate/create webhook")
+		return nil, err
+	}
+	defer func() {
+		pluginClient.Close(ctx)
+	}()
+
+	pluginValidationResult, err := pluginClient.ValidateClusterCreate(ctx, r)
+	if err != nil {
+		contextLogger.Error(err, "Error invoking plugin in the validate/update webhook")
+		return nil, err
+	}
+	allErrs = append(allErrs, pluginValidationResult...)
+
 	if len(allErrs) == 0 {
 		return nil, nil
 	}
@@ -356,6 +398,25 @@ func (r *Cluster) ValidateUpdate(old runtime.Object) (admission.Warnings, error)
 		r.ValidateChanges(oldCluster)...,
 	)
 
+	// Call the plugins to help validating this cluster update
+	ctx := context.Background()
+	contextLogger := log.FromContext(ctx)
+	pluginClient, err := r.Spec.Plugins.NewLoader(ctx)
+	if err != nil {
+		contextLogger.Error(err, "Error invoking plugin in the validate/create webhook")
+		return nil, err
+	}
+	defer func() {
+		pluginClient.Close(ctx)
+	}()
+
+	pluginValidationResult, err := pluginClient.ValidateClusterUpdate(ctx, oldCluster, r)
+	if err != nil {
+		contextLogger.Error(err, "Error invoking plugin in the validate/update webhook")
+		return nil, err
+	}
+	allErrs = append(allErrs, pluginValidationResult...)
+
 	if len(allErrs) == 0 {
 		return nil, nil
 	}

config/crd/bases/postgresql.cnpg.io_clusters.yaml

@@ -3210,6 +3210,25 @@ spec:
                       up again) or not (recreate it elsewhere - when `instances` >1)
                     type: boolean
                 type: object
+              plugins:
+                description: The plugins configuration, containing any plugin to be
+                  loaded with the corresponding configuration
+                items:
+                  description: PluginConfiguration specifies a plugin that need to
+                    be loaded for this cluster to be reconciled
+                  properties:
+                    name:
+                      description: Name is the plugin name
+                      type: string
+                    parameters:
+                      additionalProperties:
+                        type: string
+                      description: Parameters is the configuration of the plugin
+                      type: object
+                  required:
+                  - name
+                  type: object
+                type: array
               postgresGID:
                 default: 26
                 description: The GID of the `postgres` user inside the image, defaults
@@ -5141,6 +5160,35 @@ spec:
               phaseReason:
                 description: Reason for the current phase
                 type: string
+              pluginStatus:
+                description: PluginStatus is the status of the loaded plugins
+                items:
+                  description: PluginStatus is the status of a loaded plugin
+                  properties:
+                    capabilities:
+                      description: Capabilities are the list of capabilities of the
+                        plugin
+                      items:
+                        type: string
+                      type: array
+                    name:
+                      description: Name is the name of the plugin
+                      type: string
+                    operatorCapabilities:
+                      description: OperatorCapabilities are the list of capabilities
+                        of the plugin regarding the reconciler
+                      items:
+                        type: string
+                      type: array
+                    version:
+                      description: Version is the version of the plugin loaded by
+                        the latest reconciliation loop
+                      type: string
+                  required:
+                  - name
+                  - version
+                  type: object
+                type: array
               poolerIntegrations:
                 description: The integration needed by poolers referencing the cluster
                 properties: