feat: reload CNPG-i plugins automatically when their pods are rolled (cloudnative-pg#10836)

Previously, after upgrading a CNPG-i plugin (re-applying its manifest
with a new image), the operator kept talking to the plugin through the
connection it had opened during the most recent reconciliation. Clusters
using that plugin would not see the new version until something else
triggered a reconcile (a spec change, the periodic resync, or an
operator restart), which could leave clusters running against an old
plugin implementation for a long time (up to 10 hours, the default cache
resync period).

With this change, the operator watches the EndpointSlices that back
CNPG-i plugin Services in the operator namespace. When the plugin
Deployment finishes rolling out and the new Pods become Ready, the
EndpointSlice changes and every cluster using that plugin is enqueued
for reconciliation, ensuring the operator interacts with the updated
plugin promptly.

A new field index on Cluster (".spec.usedPlugins") is used to look up
all clusters affected by a given plugin without scanning the full
cluster list on every EndpointSlice event.

Signed-off-by: Leonardo Cecchi <leonardo.cecchi@enterprisedb.com>
Signed-off-by: Armando Ruocco <armando.ruocco@enterprisedb.com>
Signed-off-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>
Co-authored-by: Armando Ruocco <armando.ruocco@enterprisedb.com>
Co-authored-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>
Co-authored-by: Gabriele Fedi <gabriele.fedi@enterprisedb.com>
(cherry picked from commit ec3305d)
Signed-off-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>

Author: 4 people

config/rbac/role.yaml

@@ -111,6 +111,14 @@ rules:
   - create
   - get
   - update
+- apiGroups:
+  - discovery.k8s.io
+  resources:
+  - endpointslices
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - monitoring.coreos.com
   resources:

internal/cmd/manager/controller/controller.go

@@ -28,9 +28,11 @@ import (
 
 	"github.com/cloudnative-pg/machinery/pkg/log"
 	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/server"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
@@ -132,6 +134,31 @@ func RunController(
 		LeaderElectionReleaseOnCancel: true,
 	}
 
+	// EndpointSlices are watched only to detect CNPG-i plugin rollouts, and
+	// plugins are deployed in the operator namespace. Restricting the informer
+	// to that namespace avoids listing and watching every EndpointSlice in the
+	// cluster (one or more per Service in every namespace), which would
+	// otherwise be cached just to be dropped by the controller predicate.
+	//
+	// The restriction is only applied when the operator namespace is known: an
+	// empty namespace key is interpreted by controller-runtime as "all
+	// namespaces", which would silently widen the cache instead of narrowing
+	// it. When the namespace is unknown (e.g. local development without
+	// OPERATOR_NAMESPACE set) we simply fall back to the default caching.
+	if conf.OperatorNamespace != "" {
+		managerOptions.Cache = cache.Options{
+			ByObject: map[client.Object]cache.ByObject{
+				&discoveryv1.EndpointSlice{}: {
+					Namespaces: map[string]cache.Config{
+						conf.OperatorNamespace: {},
+					},
+				},
+			},
+		}
+	} else {
+		setupLog.Info("Plugin EndpointSlice watch disabled: OPERATOR_NAMESPACE not set")
+	}
+
 	if conf.WatchNamespace != "" {
 		namespaces := conf.WatchedNamespaces()
 		managerOptions.NewCache = multicache.DelegatingMultiNamespacedCacheBuilder(

internal/controller/cluster_controller.go

@@ -31,6 +31,7 @@ import (
 	"github.com/cloudnative-pg/machinery/pkg/stringset"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	policyv1 "k8s.io/api/policy/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -71,6 +72,9 @@ const (
 	poolerClusterKey              = ".spec.cluster.name"
 	disableDefaultQueriesSpecPath = ".spec.monitoring.disableDefaultQueries"
 	imageCatalogKey               = ".spec.imageCatalog.name"
+	// usedPluginsClusterKey is a synthetic index key, not a real Cluster spec field;
+	// it is populated by getPluginsNeededForReconcile.
+	usedPluginsClusterKey = ".spec.usedPlugins"
 )
 
 var apiSGVString = apiv1.SchemeGroupVersion.String()
@@ -121,6 +125,7 @@ var ErrNextLoop = utils.ErrNextLoop
 // +kubebuilder:rbac:groups=admissionregistration.k8s.io,resources=validatingwebhookconfigurations,verbs=get;patch
 // +kubebuilder:rbac:groups=batch,resources=jobs,verbs=get;list;delete;patch;create;watch
 // +kubebuilder:rbac:groups=coordination.k8s.io,resources=leases,verbs=get;create;update
+// +kubebuilder:rbac:groups=discovery.k8s.io,resources=endpointslices,verbs=get;list;watch
 // +kubebuilder:rbac:groups=monitoring.coreos.com,resources=podmonitors,verbs=get;create;list;watch;delete;patch
 // +kubebuilder:rbac:groups=policy,resources=poddisruptionbudgets,verbs=create;delete;get;list;watch;update;patch
 // +kubebuilder:rbac:groups=postgresql.cnpg.io,resources=clusters,verbs=get;list;watch;create;update;patch;delete
@@ -190,11 +195,7 @@ func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	ctx = cluster.SetInContext(ctx)
 
 	// Load the plugins required to bootstrap and reconcile this cluster
-	enabledPluginNames := apiv1.GetPluginConfigurationEnabledPluginNames(cluster.Spec.Plugins)
-	enabledPluginNames = append(
-		enabledPluginNames,
-		apiv1.GetExternalClustersEnabledPluginNames(cluster.Spec.ExternalClusters)...,
-	)
+	enabledPluginNames := getPluginsNeededForReconcile(cluster)
 
 	pluginLoadingContext, cancelPluginLoading := context.WithTimeout(ctx, 5*time.Second)
 	defer cancelPluginLoading()
@@ -238,6 +239,18 @@ func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	return result, nil
 }
 
+// getPluginsNeededForReconcile returns the names of the plugins that must be
+// loaded for the reconciliation loop to interact with this cluster, collected
+// from both the cluster's plugin configuration and its external clusters.
+func getPluginsNeededForReconcile(cluster *apiv1.Cluster) []string {
+	names := apiv1.GetPluginConfigurationEnabledPluginNames(cluster.Spec.Plugins)
+	names = append(
+		names,
+		apiv1.GetExternalClustersEnabledPluginNames(cluster.Spec.ExternalClusters)...,
+	)
+	return stringset.From(names).ToSortedList()
+}
+
 // Inner reconcile loop. Anything inside can require the reconciliation loop to stop by returning ErrNextLoop
 //
 //nolint:gocognit,gocyclo
@@ -1176,7 +1189,7 @@ func (r *ClusterReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manag
 		return err
 	}
 
-	return ctrl.NewControllerManagedBy(mgr).
+	ctrlBuilder := ctrl.NewControllerManagedBy(mgr).
 		WithOptions(controller.Options{
 			MaxConcurrentReconciles: maxConcurrentReconciles,
 		}).
@@ -1215,8 +1228,22 @@ func (r *ClusterReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manag
 			&apiv1.ClusterImageCatalog{},
 			handler.EnqueueRequestsFromMapFunc(r.mapClusterImageCatalogsToClusters()),
 			builder.WithPredicates(predicate.ResourceVersionChangedPredicate{}),
-		).
-		Complete(r)
+		)
+
+	if configuration.Current.OperatorNamespace != "" {
+		ctrlBuilder = ctrlBuilder.Watches(
+			&discoveryv1.EndpointSlice{},
+			handler.EnqueueRequestsFromMapFunc(
+				r.mapPluginEndpointSlicesToClusters(configuration.Current.OperatorNamespace),
+			),
+			builder.WithPredicates(
+				predicate.GenerationChangedPredicate{},
+				operatorNamespaceServiceEndpointSlicePredicate(configuration.Current.OperatorNamespace),
+			),
+		)
+	}
+
+	return ctrlBuilder.Complete(r)
 }
 
 // createFieldIndexes creates the indexes needed by this controller
@@ -1256,6 +1283,19 @@ func (r *ClusterReconciler) createFieldIndexes(ctx context.Context, mgr ctrl.Man
 		return err
 	}
 
+	// Create a new index field that allows mapping a cluster to all the
+	// plugins that are required to reconcile it
+	if err := mgr.GetFieldIndexer().IndexField(
+		ctx,
+		&apiv1.Cluster{},
+		usedPluginsClusterKey, func(rawObj client.Object) []string {
+			cluster := rawObj.(*apiv1.Cluster)
+			return getPluginsNeededForReconcile(cluster)
+		},
+	); err != nil {
+		return err
+	}
+
 	// Create a new indexed field on Pods. This field will be used to easily
 	// find all the Pods created by node
 	if err := mgr.GetFieldIndexer().IndexField(

internal/controller/cluster_controller_test.go

@@ -507,3 +507,112 @@ var _ = Describe("evaluatePodReadinessGuards", func() {
 			"kubelet-stale branch must stay event-less to avoid noise")
 	})
 })
+
+var _ = Describe("getPluginsNeededForReconcile", func() {
+	ptrBool := func(b bool) *bool { return &b }
+
+	It("returns an empty slice when no plugins or external clusters are configured", func() {
+		cluster := &apiv1.Cluster{}
+		Expect(getPluginsNeededForReconcile(cluster)).To(BeEmpty())
+	})
+
+	It("returns the names of enabled plugins from Spec.Plugins", func() {
+		cluster := &apiv1.Cluster{
+			Spec: apiv1.ClusterSpec{
+				Plugins: []apiv1.PluginConfiguration{
+					{Name: "plugin-a"},
+					{Name: "plugin-b", Enabled: ptrBool(true)},
+				},
+			},
+		}
+		Expect(getPluginsNeededForReconcile(cluster)).
+			To(ConsistOf("plugin-a", "plugin-b"))
+	})
+
+	It("skips plugins that are explicitly disabled", func() {
+		cluster := &apiv1.Cluster{
+			Spec: apiv1.ClusterSpec{
+				Plugins: []apiv1.PluginConfiguration{
+					{Name: "plugin-a"},
+					{Name: "plugin-b", Enabled: ptrBool(false)},
+				},
+			},
+		}
+		Expect(getPluginsNeededForReconcile(cluster)).
+			To(ConsistOf("plugin-a"))
+	})
+
+	It("includes plugins referenced by external clusters", func() {
+		cluster := &apiv1.Cluster{
+			Spec: apiv1.ClusterSpec{
+				ExternalClusters: []apiv1.ExternalCluster{
+					{
+						Name:                "source",
+						PluginConfiguration: &apiv1.PluginConfiguration{Name: "plugin-ext"},
+					},
+					{Name: "no-plugin"},
+				},
+			},
+		}
+		Expect(getPluginsNeededForReconcile(cluster)).
+			To(ConsistOf("plugin-ext"))
+	})
+
+	It("merges plugins from Spec.Plugins and Spec.ExternalClusters", func() {
+		cluster := &apiv1.Cluster{
+			Spec: apiv1.ClusterSpec{
+				Plugins: []apiv1.PluginConfiguration{
+					{Name: "plugin-a"},
+					{Name: "plugin-disabled", Enabled: ptrBool(false)},
+				},
+				ExternalClusters: []apiv1.ExternalCluster{
+					{
+						Name:                "source",
+						PluginConfiguration: &apiv1.PluginConfiguration{Name: "plugin-ext"},
+					},
+				},
+			},
+		}
+		Expect(getPluginsNeededForReconcile(cluster)).
+			To(ConsistOf("plugin-a", "plugin-ext"))
+	})
+
+	It("includes external-cluster plugins even when explicitly disabled", func() {
+		// GetExternalClustersEnabledPluginNames does not consult
+		// PluginConfiguration.Enabled — any non-nil PluginConfiguration on an
+		// external cluster contributes its name. This test locks that in.
+		cluster := &apiv1.Cluster{
+			Spec: apiv1.ClusterSpec{
+				ExternalClusters: []apiv1.ExternalCluster{
+					{
+						Name: "source",
+						PluginConfiguration: &apiv1.PluginConfiguration{
+							Name:    "plugin-ext",
+							Enabled: ptrBool(false),
+						},
+					},
+				},
+			},
+		}
+		Expect(getPluginsNeededForReconcile(cluster)).
+			To(ConsistOf("plugin-ext"))
+	})
+
+	It("deduplicates plugin names that appear in both Spec.Plugins and external clusters", func() {
+		cluster := &apiv1.Cluster{
+			Spec: apiv1.ClusterSpec{
+				Plugins: []apiv1.PluginConfiguration{
+					{Name: "plugin-shared"},
+				},
+				ExternalClusters: []apiv1.ExternalCluster{
+					{
+						Name:                "source",
+						PluginConfiguration: &apiv1.PluginConfiguration{Name: "plugin-shared"},
+					},
+				},
+			},
+		}
+		Expect(getPluginsNeededForReconcile(cluster)).
+			To(Equal([]string{"plugin-shared"}))
+	})
+})

internal/controller/cluster_plugins.go

@@ -24,10 +24,18 @@ import (
 	"context"
 	"reflect"
 
+	"github.com/cloudnative-pg/machinery/pkg/log"
+	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	apierrs "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
 	cnpgiclient "github.com/cloudnative-pg/cloudnative-pg/internal/cnpi/plugin/client"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
 )
 
 // updatePluginsStatus ensures that we load the plugins that are required to reconcile
@@ -57,3 +65,76 @@ func (r *ClusterReconciler) updatePluginsStatus(ctx context.Context, cluster *ap
 
 	return r.Client.Status().Patch(ctx, cluster, client.MergeFrom(oldCluster))
 }
+
+// mapPluginEndpointSlicesToClusters enqueues a reconcile request for every
+// cluster that uses a plugin whose backing Service has its endpoints changing.
+// EndpointSlice events fire when Pods behind the Service become Ready or
+// NotReady, so the operator picks up plugin rollouts after the new Pods are
+// actually serving traffic.
+func (r *ClusterReconciler) mapPluginEndpointSlicesToClusters(
+	operatorNamespace string,
+) handler.MapFunc {
+	return func(ctx context.Context, obj client.Object) []reconcile.Request {
+		logger := log.FromContext(ctx)
+
+		slice, ok := obj.(*discoveryv1.EndpointSlice)
+		if !ok {
+			return nil
+		}
+
+		serviceName := slice.Labels[discoveryv1.LabelServiceName]
+		if serviceName == "" {
+			return nil
+		}
+
+		var service corev1.Service
+		err := r.Get(
+			ctx,
+			types.NamespacedName{Namespace: slice.Namespace, Name: serviceName},
+			&service,
+		)
+		if apierrs.IsNotFound(err) {
+			return nil
+		}
+		if err != nil {
+			logger.Error(
+				err,
+				"Error while resolving the Service owning a plugin EndpointSlice, skipping",
+				"endpointSlice", client.ObjectKeyFromObject(slice),
+				"serviceName", serviceName,
+			)
+			return nil
+		}
+
+		if !isPluginService(&service, operatorNamespace) {
+			return nil
+		}
+
+		pluginName := service.Labels[utils.PluginNameLabelName]
+		if pluginName == "" {
+			return nil
+		}
+
+		var clusterList apiv1.ClusterList
+		if err := r.List(
+			ctx,
+			&clusterList,
+			client.MatchingFields{usedPluginsClusterKey: pluginName},
+		); err != nil {
+			logger.Error(
+				err,
+				"Error while looking up clusters using a plugin whose endpoints changed, skipping",
+				"endpointSlice", client.ObjectKeyFromObject(slice),
+				"pluginName", pluginName,
+			)
+			return nil
+		}
+
+		result := make([]reconcile.Request, len(clusterList.Items))
+		for i := range clusterList.Items {
+			result[i].Name = clusterList.Items[i].Name
+			result[i].Namespace = clusterList.Items[i].Namespace
+		}
+		return result
+	}
+}