fix(backup): clean unused backup connections (cloudnative-pg#6882)

An error while taking a snapshot backup could leave an unused connection
open until the next backup. This patch ensures that backup connections
are explicitly closed when no longer needed, preventing potential
connection leaks.

Partially closes cloudnative-pg#6761

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

Author: 3 people

pkg/management/postgres/webserver/remote.go

@@ -27,10 +27,12 @@ import (
 	"os"
 	"os/exec"
 	"path"
+	"time"
 
 	"github.com/cloudnative-pg/machinery/pkg/execlog"
 	"github.com/cloudnative-pg/machinery/pkg/fileutils"
 	"github.com/cloudnative-pg/machinery/pkg/log"
+	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
@@ -111,7 +113,84 @@ func NewRemoteWebServer(
 		}
 	}
 
-	return NewWebServer(server), nil
+	srv := NewWebServer(server)
+
+	srv.routines = append(srv.routines, endpoints.cleanupStaleCollections)
+
+	return srv, nil
+}
+
+func (ws *remoteWebserverEndpoints) cleanupStaleCollections(ctx context.Context) {
+	closeBackupConnection := func(bc *backupConnection) {
+		log := log.WithValues(
+			"backupName", bc.data.BackupName,
+			"phase", bc.data.Phase,
+		)
+		log.Warning("Closing stale PostgreSQL backup connection")
+
+		if err := bc.conn.Close(); err != nil {
+			log.Error(err, "Error while closing stale PostgreSQL backup connection")
+		}
+		bc.data.Phase = Completed
+	}
+
+	innerRoutine := func() {
+		if ws == nil {
+			return
+		}
+		bc := ws.currentBackup
+		if bc == nil || bc.conn == nil {
+			return
+		}
+
+		if bc.data.Phase == Completed || bc.data.BackupName == "" {
+			return
+		}
+
+		bc.sync.Lock()
+		defer bc.sync.Unlock()
+
+		if bc.err != nil {
+			closeBackupConnection(bc)
+			return
+		}
+
+		if err := bc.conn.PingContext(ctx); err != nil {
+			bc.err = fmt.Errorf("error while pinging: %w", err)
+			closeBackupConnection(bc)
+			return
+		}
+
+		var backup apiv1.Backup
+
+		err := ws.typedClient.Get(ctx, client.ObjectKey{
+			Namespace: ws.instance.GetNamespaceName(),
+			Name:      bc.data.BackupName,
+		}, &backup)
+		if apierrs.IsNotFound(err) {
+			bc.err = fmt.Errorf("backup %s not found", bc.data.BackupName)
+			closeBackupConnection(bc)
+			return
+		}
+		if err != nil {
+			return
+		}
+
+		if backup.Status.IsDone() {
+			bc.err = fmt.Errorf("backup %s is done", bc.data.BackupName)
+			closeBackupConnection(bc)
+			return
+		}
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(1 * time.Minute):
+			innerRoutine()
+		}
+	}
 }
 
 func (ws *remoteWebserverEndpoints) isServerHealthy(w http.ResponseWriter, _ *http.Request) {

pkg/management/postgres/webserver/webserver.go

@@ -67,7 +67,8 @@ func (body Response[T]) EnsureDataIsPresent() error {
 
 // Webserver wraps a webserver to make it a kubernetes Runnable
 type Webserver struct {
-	server *http.Server
+	server   *http.Server
+	routines []func(ctx context.Context)
 }
 
 // NewWebServer creates a Webserver as a Kubernetes Runnable, given a http.Server
@@ -96,6 +97,13 @@ func (ws *Webserver) Start(ctx context.Context) error {
 		}
 	}()
 
+	subCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	for _, routine := range ws.routines {
+		routine(subCtx)
+	}
+
 	select {
 	// we exit with error code, potentially we could do a retry logic, but rarely a webserver that doesn't start will run
 	// on subsequent tries

tests/e2e/volume_snapshot_test.go

@@ -20,19 +20,22 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 
 	volumesnapshot "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/retry"
 	k8client "sigs.k8s.io/controller-runtime/pkg/client"
 
 	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
 	"github.com/cloudnative-pg/cloudnative-pg/tests"
 	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/backups"
 	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/clusterutils"
+	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/exec"
 	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/minio"
 	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/postgres"
 	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/secrets"
@@ -66,6 +69,18 @@ var _ = Describe("Verify Volume Snapshot",
 			return snapshotList, nil
 		}
 
+		updateClusterSnapshotClass := func(namespace, clusterName, className string) {
+			cluster := &apiv1.Cluster{}
+			err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
+				var err error
+				cluster, err = clusterutils.Get(env.Ctx, env.Client, namespace, clusterName)
+				Expect(err).ToNot(HaveOccurred())
+				cluster.Spec.Backup.VolumeSnapshot.ClassName = className
+				return env.Client.Update(env.Ctx, cluster)
+			})
+			Expect(err).ToNot(HaveOccurred())
+		}
+
 		var namespace string
 
 		Context("using the kubectl cnpg plugin", Ordered, func() {
@@ -842,5 +857,66 @@ var _ = Describe("Verify Volume Snapshot",
 					AssertDataExpectedCount(env, tableLocator, 6)
 				})
 			})
+
+			It("should clean up unused backup connections", func() {
+				By("setting a non-existing snapshotClass", func() {
+					updateClusterSnapshotClass(namespace, clusterToSnapshotName, "wrongSnapshotClass")
+				})
+
+				By("starting a new backup that will fail", func() {
+					backupName := fmt.Sprintf("%s-failed", clusterToSnapshotName)
+					failedBackup, err := backups.Create(
+						env.Ctx, env.Client,
+						apiv1.Backup{
+							ObjectMeta: metav1.ObjectMeta{
+								Namespace: namespace,
+								Name:      backupName,
+							},
+							Spec: apiv1.BackupSpec{
+								Target:  apiv1.BackupTargetPrimary,
+								Method:  apiv1.BackupMethodVolumeSnapshot,
+								Cluster: apiv1.LocalObjectReference{Name: clusterToSnapshotName},
+							},
+						},
+					)
+					Expect(err).ToNot(HaveOccurred())
+
+					Eventually(func(g Gomega) {
+						err = env.Client.Get(env.Ctx, types.NamespacedName{
+							Namespace: namespace,
+							Name:      backupName,
+						}, failedBackup)
+						g.Expect(err).ToNot(HaveOccurred())
+						g.Expect(failedBackup.Status.Phase).To(BeEquivalentTo(apiv1.BackupPhaseFailed))
+						g.Expect(failedBackup.Status.Error).To(ContainSubstring("Failed to get snapshot class"))
+					}, RetryTimeout).Should(Succeed())
+				})
+
+				By("verifying that the backup connection is cleaned up", func() {
+					primaryPod, err := clusterutils.GetPrimary(env.Ctx, env.Client, namespace,
+						clusterToSnapshotName)
+					Expect(err).ToNot(HaveOccurred())
+					query := "SELECT count(*) FROM pg_stat_activity WHERE query ILIKE '%pg_backup_start%' " +
+						"AND application_name = 'cnpg-instance-manager'"
+
+					Eventually(func() (int, error, error) {
+						stdout, _, err := exec.QueryInInstancePod(
+							env.Ctx, env.Client, env.Interface, env.RestClientConfig,
+							exec.PodLocator{
+								Namespace: primaryPod.Namespace,
+								PodName:   primaryPod.Name,
+							},
+							postgres.PostgresDBName,
+							query)
+						value, atoiErr := strconv.Atoi(strings.TrimSpace(stdout))
+						return value, err, atoiErr
+					}, RetryTimeout).Should(BeEquivalentTo(0),
+						"Stale backup connection should have been dropped")
+				})
+
+				By("resetting the snapshotClass value", func() {
+					updateClusterSnapshotClass(namespace, clusterToSnapshotName, os.Getenv("E2E_CSI_STORAGE_CLASS"))
+				})
+			})
 		})
 	})