Repository: cloudnative-pg/cloudnative-pg

Issue:
[Bug]: operator isn’t reloading the certificates of the (cnpg-i) plugins
### Is there an existing issue already for this bug?

- [x] I have searched for an existing issue, and could not find anything. I believe this is a new bug.

### I have read the troubleshooting guide

- [x] I have read the troubleshooting guide and I think this is a new bug.

### I am running a supported version of CloudNativePG

- [x] I have read the troubleshooting guide and I think this is a new bug.

### Contact Details

hh937599103@gmail.com

### Version

1.25 (latest patch)

### What version of Kubernetes are you using?

1.32

### What is your Kubernetes environment?

Self-managed: kind (evaluation)

### How did you install the operator?

YAML manifest

### What happened?

We have implemented a cnpg-i plugin which is deployed as deployment in the operator's namespace. Mostly, it has been working well. However, we have observed several times that when the plugin's TLS certificates got expired, we can find the error logs like below in the operator's log:

```
\"reconcileID\":\"962e1f6d-3e96-4ba1-9a90-ce9510f44f9f\",\"error\":\"while getting plugin connection: while querying plugin identity: rpc error: code = Unavailable desc = connection error: desc = \\\"transport: authentication handshake failed: tls: failed to verify certificate: x509: certificate signed by unknown authority\\\"\"
```

And by checking the code base, it looks like we don't refresh the plugin configuration (from the operator) if it's changed

### Cluster resource

```shell

```

### Relevant log output

```shell

```

### Code of Conduct

- [x] I agree to follow this project's Code of Conduct

Hints from issue comments:
@agelwarg: I'm (still) getting this error, using cnpg 1.26.0 and plugin-barman-cloud v0.5.0.  Is this supposed to be fixed?@anjuls: @anishbista60 can you look into it?@anishbista60: ok
@klaudworks: Same issue on our side using cnpg 1.26.1 and plugin-barman-cloud v0.5.0.@lrstanley: Same here, cnpg 1.27 & plugin-barman-cloud 0.5.x. I just upgraded to 0.6.0 and it seemed to fix it, but I'm not sure if it was because of a restart of the barman cloud plugin pod, or due to the upgrade, so it may happen again after a period of time.@klaudworks: restarting the barman cloud pod actually fixed it for me, at least for the time being.@alexklibisz: Still seeing the issue. cnpg 1.26 and plugin-barman cloud 0.5.0. Resolved by rebooting the plugin-barman-cloud pod in the operator namespace. It seems like this issue should be re-opened.@hh24k: Hi @alexklibisz @Irstanley and @klaudworks, for barman plugin, we have the issue fixed in v0.6.0 (see to https://github.com/cloudnative-pg/plugin-barman-cloud/pull/439) ... plz. check it out ... thanks!@alexklibisz: Thanks. Just summarizing the changes/fixes:

* Actual fixes in cnpg-i-machinery: https://github.com/cloudnative-pg/cnpg-i-machinery/pull/174/files, https://github.com/cloudnative-pg/cnpg-i-machinery/pull/197/files
* Dependency update in barman cloud plugin: https://github.com/cloudnative-pg/plugin-barman-cloud/pull/439
* Released in 0.6.0: https://github.com/cloudnative-pg/plugin-barman-cloud/releases/tag/v0.6.0

Solution patch:
diff --git a/internal/cmd/manager/controller/controller.go b/internal/cmd/manager/controller/controller.go
index 01ccec51..93d8315c 100644
--- a/internal/cmd/manager/controller/controller.go
+++ b/internal/cmd/manager/controller/controller.go
@@ -236,8 +236,8 @@ func RunController(
 		return err
 	}
 
-	if err = controller.NewPluginReconciler(mgr, pluginRepository).
-		SetupWithManager(mgr, configuration.Current.OperatorNamespace, maxConcurrentReconciles); err != nil {
+	if err = controller.NewPluginReconciler(mgr, conf.OperatorNamespace, pluginRepository).
+		SetupWithManager(mgr, maxConcurrentReconciles); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Plugin")
 		return err
 	}
diff --git a/internal/cnpi/plugin/repository/setup.go b/internal/cnpi/plugin/repository/setup.go
index ca068240..b47db042 100644
--- a/internal/cnpi/plugin/repository/setup.go
+++ b/internal/cnpi/plugin/repository/setup.go
@@ -34,11 +34,12 @@ import (
 type Interface interface {
 	// ForgetPlugin closes every connection to the plugin with the passed name
 	// and forgets its discovery info.
-	// If the plug in was not available in the repository, this is a no-op
+	// This operation is synchronous and blocks until every connection is closed.
+	// If the plugin was not available in the repository, this is a no-op.
 	ForgetPlugin(name string)
 
 	// RegisterRemotePlugin registers a plugin available on a remote
-	// TCP entrypoint
+	// TCP entrypoint.
 	RegisterRemotePlugin(name string, address string, tlsConfig *tls.Config) error
 
 	// RegisterUnixSocketPluginsInPath scans the passed directory
@@ -60,71 +61,84 @@ type data struct {
 	pluginConnectionPool map[string]*puddle.Pool[connection.Interface]
 }
 
+// pluginSetupOptions are the options to be used when setting up
+// a plugin connection
+type pluginSetupOptions struct {
+	// forceRegistration forces the creation of a new plugin connection
+	// even if one already exists. The existing connection will be closed.
+	forceRegistration bool
+}
+
 // maxPoolSize is the maximum number of connections in a plugin's connection
 // pool
 const maxPoolSize = 5
 
-func (r *data) setPluginProtocol(name string, protocol connection.Protocol) error {
-	r.mux.Lock()
-	defer r.mux.Unlock()
-
-	if r.pluginConnectionPool == nil {
-		r.pluginConnectionPool = make(map[string]*puddle.Pool[connection.Interface])
-	}
-
-	_, ok := r.pluginConnectionPool[name]
-	if ok {
-		return &ErrPluginAlreadyRegistered{
-			Name: name,
-		}
-	}
-
-	constructor := func(ctx context.Context) (res connection.Interface, err error) {
-		var handler connection.Handler
-
-		defer func() {
-			if err != nil && handler != nil {
-				_ = handler.Close()
-			}
-		}()
-
-		constructorLogger := log.
+func pluginConnectionConstructor(name string, protocol connection.Protocol) puddle.Constructor[connection.Interface] {
+	return func(ctx context.Context) (connection.Interface, error) {
+		logger := log.
 			FromContext(ctx).
 			WithName("setPluginProtocol").
 			WithValues("pluginName", name)
-		ctx = log.IntoContext(ctx, constructorLogger)
+		ctx = log.IntoContext(ctx, logger)
 
-		constructorLogger.Trace("Acquired physical plugin connection")
+		logger.Trace("Connecting to plugin")
+		var (
+			result  connection.Interface
+			handler connection.Handler
+			err     error
+		)
 
 		if handler, err = protocol.Dial(ctx); err != nil {
-			constructorLogger.Error(err, "Got error while connecting to plugin")
+			logger.Error(err, "Error while connecting to plugin (physical)")
+			return nil, err
+		}
+
+		if result, err = connection.LoadPlugin(ctx, handler); err != nil {
+			logger.Error(err, "Error while connecting to plugin (logical)")
+			_ = handler.Close()
 			return nil, err
 		}
 
-		return connection.LoadPlugin(ctx, handler)
+		return result, err
 	}
+}
 
-	destructor := func(res connection.Interface) {
-		constructorLogger := log.
-			FromContext(context.Background()).
-			WithName("setPluginProtocol").
-			WithValues("pluginName", name)
-		constructorLogger.Trace("Released physical plugin connection")
-
-		err := res.Close()
-		if err != nil {
-			destructorLogger := log.FromContext(context.Background()).
-				WithName("setPluginProtocol").
-				WithValues("pluginName", res.Name())
-			destructorLogger.Warning("Error while closing plugin connection", "err", err)
+func pluginConnectionDestructor(res connection.Interface) {
+	logger := log.FromContext(context.Background()).
+		WithName("pluginConnectionDestructor").
+		WithValues("pluginName", res.Name())
+
+	logger.Trace("Released physical plugin connection")
+
+	err := res.Close()
+	if err != nil {
+		logger.Warning("Error while closing plugin connection", "err", err)
+	}
+}
+
+func (r *data) setPluginProtocol(name string, protocol connection.Protocol, opts pluginSetupOptions) error {
+	r.mux.Lock()
+	defer r.mux.Unlock()
+
+	if r.pluginConnectionPool == nil {
+		r.pluginConnectionPool = make(map[string]*puddle.Pool[connection.Interface])
+	}
+
+	if oldPool, alreadyRegistered := r.pluginConnectionPool[name]; alreadyRegistered {
+		if opts.forceRegistration {
+			oldPool.Close()
+		} else {
+			return &ErrPluginAlreadyRegistered{
+				Name: name,
+			}
 		}
 	}
 
 	var err error
 	r.pluginConnectionPool[name], err = puddle.NewPool(
 		&puddle.Config[connection.Interface]{
-			Constructor: constructor,
-			Destructor:  destructor,
+			Constructor: pluginConnectionConstructor(name, protocol),
+			Destructor:  pluginConnectionDestructor,
 			MaxSize:     maxPoolSize,
 		},
 	)
@@ -143,22 +157,34 @@ func (r *data) ForgetPlugin(name string) {
 		return
 	}
 
-	// TODO(leonardoce): should we really wait for all the plugin connections
-	// to be closed?
 	pool.Close()
+	delete(r.pluginConnectionPool, name)
 }
 
 // registerUnixSocketPlugin registers a plugin available at the passed
 // unix socket path
 func (r *data) registerUnixSocketPlugin(name, path string) error {
-	return r.setPluginProtocol(name, connection.ProtocolUnix(path))
+	return r.setPluginProtocol(name, connection.ProtocolUnix(path), pluginSetupOptions{
+		// Forcing the registration of a Unix socket plugin has no meaning
+		// because they can be installed and started only when the Pod is created.
+		forceRegistration: false,
+	})
 }
 
 func (r *data) RegisterRemotePlugin(name string, address string, tlsConfig *tls.Config) error {
-	return r.setPluginProtocol(name, &connection.ProtocolTCP{
+	protocol := &connection.ProtocolTCP{
 		TLSConfig: tlsConfig,
 		Address:   address,
-	})
+	}
+
+	// The RegisterRemotePlugin function is called when the plugin is registered for
+	// the first time and when the certificates of an existing plugin get refreshed.
+	// In the second case, the plugin loading will be forced and all existing
+	// connections will be dropped and recreated.
+	opts := pluginSetupOptions{
+		forceRegistration: true,
+	}
+	return r.setPluginProtocol(name, protocol, opts)
 }
 
 func (r *data) RegisterUnixSocketPluginsInPath(pluginsPath string) ([]string, error) {
diff --git a/internal/controller/plugin_controller.go b/internal/controller/plugin_controller.go
index 0e393f15..83efdef4 100644
--- a/internal/controller/plugin_controller.go
+++ b/internal/controller/plugin_controller.go
@@ -32,9 +32,9 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
-	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
-	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	"github.com/cloudnative-pg/cloudnative-pg/internal/cnpi/plugin/repository"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
@@ -46,14 +46,21 @@ type PluginReconciler struct {
 
 	Scheme  *runtime.Scheme
 	Plugins repository.Interface
+
+	OperatorNamespace string
 }
 
 // NewPluginReconciler creates a new PluginReconciler initializing it
-func NewPluginReconciler(mgr manager.Manager, plugins repository.Interface) *PluginReconciler {
+func NewPluginReconciler(
+	mgr manager.Manager,
+	operatorNamespace string,
+	plugins repository.Interface,
+) *PluginReconciler {
 	return &PluginReconciler{
-		Client:  mgr.GetClient(),
-		Scheme:  mgr.GetScheme(),
-		Plugins: plugins,
+		Client:            mgr.GetClient(),
+		Scheme:            mgr.GetScheme(),
+		Plugins:           plugins,
+		OperatorNamespace: operatorNamespace,
 	}
 }
 
@@ -61,9 +68,9 @@ func NewPluginReconciler(mgr manager.Manager, plugins repository.Interface) *Plu
 func (r *PluginReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	contextLogger, ctx := log.SetupLogger(ctx)
 
-	contextLogger.Debug("Plugin reconciliation loop start")
+	contextLogger.Trace("Plugin reconciliation loop start")
 	defer func() {
-		contextLogger.Debug("Plugin reconciliation loop end")
+		contextLogger.Trace("Plugin reconciliation loop end")
 	}()
 
 	var service corev1.Service
@@ -80,6 +87,11 @@ func (r *PluginReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 		return ctrl.Result{}, fmt.Errorf("cannot get the resource: %w", err)
 	}
 
+	if !isPluginService(&service, r.OperatorNamespace) {
+		contextLogger.Trace("Skipping reconciliation for a non-cnpg-i service")
+		return ctrl.Result{}, nil
+	}
+
 	// Process label and annotations
 	pluginName := service.Labels[utils.PluginNameLabelName]
 	if len(pluginName) == 0 {
@@ -207,31 +219,55 @@ func (r *PluginReconciler) getSecret(
 	return &secret, nil
 }
 
+func (r *PluginReconciler) mapSecretToPlugin(ctx context.Context, obj client.Object) []reconcile.Request {
+	// We only consider the secrets that are installed in the
+	// operator namespace because plugins need to be deployed
+	// in the same namespace as the operator.
+	if obj.GetNamespace() != r.OperatorNamespace {
+		return nil
+	}
+
+	logger := log.FromContext(ctx)
+
+	var services corev1.ServiceList
+	if err := r.Client.List(
+		ctx,
+		&services,
+		client.HasLabels{utils.PluginNameLabelName},
+		client.InNamespace(r.OperatorNamespace),
+	); err != nil {
+		logger.Error(
+			err,
+			"Error while listing CNPG-I services in the operator namespace",
+		)
+		return nil
+	}
+
+	var result []reconcile.Request
+	for i := range services.Items {
+		service := &services.Items[i]
+		if isSecretUsedByPluginService(service, obj.GetName()) {
+			result = append(result, reconcile.Request{
+				NamespacedName: client.ObjectKeyFromObject(service),
+			})
+		}
+	}
+
+	return result
+}
+
 // SetupWithManager adds this PluginReconciler to the passed controller manager
 func (r *PluginReconciler) SetupWithManager(
 	mgr ctrl.Manager,
-	operatorNamespace string,
 	maxConcurrentReconciles int,
 ) error {
-	pluginServicesPredicate := predicate.Funcs{
-		CreateFunc: func(e event.CreateEvent) bool {
-			return isPluginService(e.Object, operatorNamespace)
-		},
-		DeleteFunc: func(e event.DeleteEvent) bool {
-			return isPluginService(e.Object, operatorNamespace)
-		},
-		GenericFunc: func(e event.GenericEvent) bool {
-			return isPluginService(e.Object, operatorNamespace)
-		},
-		UpdateFunc: func(e event.UpdateEvent) bool {
-			return isPluginService(e.ObjectNew, operatorNamespace)
-		},
-	}
-
 	return ctrl.NewControllerManagedBy(mgr).
 		WithOptions(controller.Options{MaxConcurrentReconciles: maxConcurrentReconciles}).
 		For(&corev1.Service{}).
 		Named("plugin").
-		WithEventFilter(pluginServicesPredicate).
+		Watches(
+			&corev1.Secret{},
+			handler.EnqueueRequestsFromMapFunc(r.mapSecretToPlugin),
+		).
 		Complete(r)
 }
diff --git a/internal/controller/plugin_predicates.go b/internal/controller/plugin_predicates.go
index 7f6e381a..119eb2d2 100644
--- a/internal/controller/plugin_predicates.go
+++ b/internal/controller/plugin_predicates.go
@@ -22,7 +22,7 @@ import (
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
 )
 
-var isPluginService = func(object client.Object, operatorNamespace string) bool {
+func isPluginService(object client.Object, operatorNamespace string) bool {
 	if object.GetNamespace() != operatorNamespace {
 		// Only consider the services that are in the same
 		// namespace where the operator is installed
@@ -50,3 +50,14 @@ var isPluginService = func(object client.Object, operatorNamespace string) bool
 
 	return true
 }
+
+// isSecretUsedByPluginService returns true when the passed service
+// uses the secret with the passed name
+func isSecretUsedByPluginService(service client.Object, secretName string) bool {
+	annotations := service.GetAnnotations()
+
+	clientSecretName := annotations[utils.PluginClientSecretAnnotationName]
+	serverSecretName := annotations[utils.PluginServerSecretAnnotationName]
+
+	return clientSecretName == secretName || serverSecretName == secretName
+}



Test patch:
diff --git a/internal/cnpi/plugin/repository/setup_test.go b/internal/cnpi/plugin/repository/setup_test.go
new file mode 100644
index 00000000..d3030a57
--- /dev/null
+++ b/internal/cnpi/plugin/repository/setup_test.go
@@ -0,0 +1,62 @@
+package repository
+
+import (
+	"context"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Set Plugin Protocol", func() {
+	var repository *data
+
+	BeforeEach(func() {
+		repository = &data{}
+	})
+
+	It("creates connection pool for new plugin", func() {
+		err := repository.setPluginProtocol("plugin1", newUnitTestProtocol("test"), pluginSetupOptions{})
+		Expect(err).NotTo(HaveOccurred())
+		Expect(repository.pluginConnectionPool).To(HaveKey("plugin1"))
+	})
+
+	It("fails when adding same plugin name without forceRegistration", func() {
+		err := repository.setPluginProtocol("plugin1", newUnitTestProtocol("/tmp/socket"), pluginSetupOptions{})
+		Expect(err).NotTo(HaveOccurred())
+
+		err = repository.setPluginProtocol("plugin1", newUnitTestProtocol("/tmp/socket2"), pluginSetupOptions{})
+		Expect(err).To(BeEquivalentTo(&ErrPluginAlreadyRegistered{Name: "plugin1"}))
+	})
+
+	It("overwrites existing plugin when forceRegistration is true", func() {
+		first := newUnitTestProtocol("/tmp/socket")
+		err := repository.setPluginProtocol("plugin1", first, pluginSetupOptions{})
+		Expect(err).NotTo(HaveOccurred())
+		pool1 := repository.pluginConnectionPool["plugin1"]
+
+		ctx1, cancel := context.WithCancel(context.Background())
+		conn1, err := pool1.Acquire(ctx1)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(conn1).NotTo(BeNil())
+		cancel()
+		conn1.Release()
+
+		second := newUnitTestProtocol("/tmp/socket2")
+		err = repository.setPluginProtocol("plugin1", second, pluginSetupOptions{forceRegistration: true})
+		Expect(err).NotTo(HaveOccurred())
+		pool2 := repository.pluginConnectionPool["plugin1"]
+
+		ctx2, cancel := context.WithCancel(context.Background())
+		conn2, err := pool2.Acquire(ctx2)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(conn2).NotTo(BeNil())
+		cancel()
+		conn2.Release()
+
+		Expect(pool1).NotTo(Equal(pool2))
+		Expect(first.mockHandlers).To(HaveLen(1))
+		Expect(first.mockHandlers[0].closed).To(BeTrue())
+		Expect(second.mockHandlers).To(HaveLen(1))
+		Expect(second.mockHandlers[0].closed).To(BeFalse())
+	})
+})
diff --git a/internal/cnpi/plugin/repository/suite_test.go b/internal/cnpi/plugin/repository/suite_test.go
new file mode 100644
index 00000000..4421aecd
--- /dev/null
+++ b/internal/cnpi/plugin/repository/suite_test.go
@@ -0,0 +1,119 @@
+package repository
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"github.com/cloudnative-pg/cnpg-i/pkg/identity"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/test/bufconn"
+
+	"github.com/cloudnative-pg/cloudnative-pg/internal/cnpi/plugin/connection"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestRepository(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Repository Suite")
+}
+
+type identityImplementation struct {
+	identity.UnimplementedIdentityServer
+}
+
+// GetPluginMetadata implements Identity
+func (i identityImplementation) GetPluginMetadata(
+	_ context.Context,
+	_ *identity.GetPluginMetadataRequest,
+) (*identity.GetPluginMetadataResponse, error) {
+	return &identity.GetPluginMetadataResponse{
+		Name:          "testing-service",
+		Version:       "0.0.1",
+		DisplayName:   "testing-service",
+		ProjectUrl:    "https://github.com/cloudnative-pg/cloudnative-pg",
+		RepositoryUrl: "https://github.com/cloudnative-pg/cloudnative-pg",
+		License:       "APACHE 2.0",
+		Maturity:      "alpha",
+	}, nil
+}
+
+// GetPluginCapabilities implements identity
+func (i identityImplementation) GetPluginCapabilities(
+	_ context.Context,
+	_ *identity.GetPluginCapabilitiesRequest,
+) (*identity.GetPluginCapabilitiesResponse, error) {
+	return &identity.GetPluginCapabilitiesResponse{
+		Capabilities: []*identity.PluginCapability{},
+	}, nil
+}
+
+// Probe implements Identity
+func (i identityImplementation) Probe(
+	_ context.Context,
+	_ *identity.ProbeRequest,
+) (*identity.ProbeResponse, error) {
+	return &identity.ProbeResponse{
+		Ready: true,
+	}, nil
+}
+
+type unitTestProtocol struct {
+	name         string
+	mockHandlers []*mockHandler
+	server       *grpc.Server
+}
+
+type mockHandler struct {
+	*grpc.ClientConn
+	closed bool
+}
+
+func newUnitTestProtocol(name string) *unitTestProtocol {
+	return &unitTestProtocol{name: name}
+}
+
+func (h *mockHandler) Close() error {
+	_ = h.ClientConn.Close()
+	h.closed = true
+	return nil
+}
+
+func (p *unitTestProtocol) Dial(ctx context.Context) (connection.Handler, error) {
+	listener := bufconn.Listen(1024 * 1024)
+
+	if len(p.mockHandlers) == 0 {
+		p.server = grpc.NewServer()
+
+		identity.RegisterIdentityServer(p.server, &identityImplementation{})
+
+		go func() {
+			<-ctx.Done()
+			p.server.Stop()
+		}()
+
+		go func() {
+			_ = p.server.Serve(listener)
+		}()
+	}
+
+	dialer := func(_ context.Context, _ string) (net.Conn, error) {
+		return listener.Dial()
+	}
+
+	conn, err := grpc.NewClient("passthrough://bufnet",
+		grpc.WithContextDialer(dialer),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	if err != nil {
+		return nil, err
+	}
+	mh := &mockHandler{
+		ClientConn: conn,
+	}
+	p.mockHandlers = append(p.mockHandlers, mh)
+	return mh, nil
+}