feat: cluster-managed replication slots for High Availability (cloudnative-pg#740)

Introduce the concept of cluster-managed replication slots, starting with
high availability clusters. This feature automatically manages physical
replication slots for each hot standby replica in the High Availability
cluster, both in the primary and the standby.

We are introducing these two terms:

- "Primary HA slot": a physical replication slot whose lifecycle is entirely
  managed by the current primary of the cluster and whose purpose is to map to
  a specific standby in streaming replication. Such a slot lives on the primary
  only.

- "Standby HA slot": a physical replication slot for a standby whose lifecycle is
  entirely managed by another standby in the cluster, based on the content of
  the `pg_replication_slots` view in the primary, and updated at regular
  intervals using `pg_replication_slot_advance()`.

This feature is controlled through the following options:

- `.spec.replicationSlots.highAvailability.enabled`: if true, the feature is
  enabled (`false` is the default)

- `.spec.replicationSlots.highAvailability.slotPrefix`: the prefix that
  identifies replication slots managed by the operator for this feature
  (default: `_cnpg_`)

- `.spec.replicationSlots.updateInterval`: how often the standby synchronizes
  the position of the local copy of the replication slots with the position on
  the current primary, expressed in seconds (default: 30)

IMPORTANT: this capability requires PostgreSQL 11 or higher, as it relies on
the `pg_replication_slot_advance()` administration function to directly
manipulate the position of a replication slot. In PostgreSQL 11, enabling
replication slots if initially disabled, or conversely disabling them if
initially enabled, will require a rolling update of the cluster (due to the
presence of the `recovery.conf` file that is only read at startup).

Closes cloudnative-pg#8

ABOUT REPLICATION SLOTS IN POSTGRESQL: replication slots are a native
PostgreSQL feature introduced in 9.4 that provides an automated way to ensure
that the primary does not remove WAL segments until all the attached streaming
replication clients have received them, and that the primary does not remove
rows which could cause a recovery conflict even when the standby is
(temporarily) disconnected. A replication slot exists solely on the instance
that created it, and PostgreSQL does not replicate it on the standby servers.
As a result, after a failover or a switchover, the new primary does not contain
the replication slot from the old primary. This can create problems for the
streaming replication clients that were connected to the old primary and have
lost their slot. More information at:
https://www.postgresql.org/docs/current/warm-standby.html#STREAMING-REPLICATION-SLOTS

Signed-off-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>
Signed-off-by: Niccolò Fei <niccolo.fei@enterprisedb.com>
Signed-off-by: Jitendra Wadle <jitendra.wadle@enterprisedb.com>
Signed-off-by: Jonathan Gonzalez V <jonathan.gonzalez@enterprisedb.com>
Signed-off-by: Gabriele Bartolini <gabriele.bartolini@enterprisedb.com>
Signed-off-by: Jaime Silvela <jaime.silvela@enterprisedb.com>
Signed-off-by: Armando Ruocco <armando.ruocco@enterprisedb.com>
Signed-off-by: Tao Li <tao.li@enterprisedb.com>
Signed-off-by: Leonardo Cecchi <leonardo.cecchi@enterprisedb.com>
Signed-off-by: Philippe Scorsolini <philippe.scorsolini@enterprisedb.com>
Co-authored-by: Niccolò Fei <niccolo.fei@enterprisedb.com>
Co-authored-by: Jitendra Wadle <jitendra.wadle@enterprisedb.com>
Co-authored-by: Jonathan Gonzalez V <jonathan.gonzalez@enterprisedb.com>
Co-authored-by: Jaime Silvela <jaime.silvela@enterprisedb.com>
Co-authored-by: Armando Ruocco <armando.ruocco@enterprisedb.com>
Co-authored-by: Gabriele Bartolini <gabriele.bartolini@enterprisedb.com>
Co-authored-by: Tao Li <tao.li@enterprisedb.com>
Co-authored-by: Leonardo Cecchi <leonardo.cecchi@enterprisedb.com>
Co-authored-by: Philippe Scorsolini <philippe.scorsolini@enterprisedb.com>

Author: 10 people

.wordlist-en-custom.txt

@@ -243,6 +243,8 @@ RedHat
 RedHat's
 ReplicaClusterConfiguration
 ReplicaSet
+ReplicationSlotsConfiguration
+ReplicationSlotsHAConfiguration
 ReplicationTLSSecret
 ResizingPVC
 ResourceRequirements
@@ -528,6 +530,7 @@ gzip
 hba
 hdr
 healthz
+highAvailability
 historyTags
 horikyota
 hostPort
@@ -749,6 +752,7 @@ recoverytarget
 recv
 redhat
 relatime
+replicationSlots
 replicationTLSSecret
 repmgr
 reportNonRedacted

api/v1/cluster_types.go

@@ -19,6 +19,9 @@ package v1
 import (
 	"context"
 	"fmt"
+	"regexp"
+	"strings"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -157,6 +160,9 @@ type ClusterSpec struct {
 	// +optional
 	PostgresConfiguration PostgresConfiguration `json:"postgresql,omitempty"`
 
+	// Replication slots management configuration
+	ReplicationSlots *ReplicationSlotsConfiguration `json:"replicationSlots,omitempty"`
+
 	// Instructions to bootstrap this cluster
 	// +optional
 	Bootstrap *BootstrapConfiguration `json:"bootstrap,omitempty"`
@@ -423,6 +429,9 @@ type ClusterStatus struct {
 
 	// Conditions for cluster object
 	Conditions []metav1.Condition `json:"conditions,omitempty"`
+
+	// List of instance names in the cluster
+	InstanceNames []string `json:"instanceNames,omitempty"`
 }
 
 // InstanceReportedState describes the last reported state of an instance during a reconciliation loop
@@ -523,6 +532,86 @@ type ReplicaClusterConfiguration struct {
 	Source string `json:"source"`
 }
 
+// DefaultReplicationSlotsUpdateInterval is the default in seconds for the replication slots update interval
+const DefaultReplicationSlotsUpdateInterval = 30
+
+// DefaultReplicationSlotsHASlotPrefix is the default prefix for names of replication slots used for HA.
+const DefaultReplicationSlotsHASlotPrefix = "_cnpg_"
+
+// ReplicationSlotsConfiguration encapsulates the configuration
+// of replication slots
+type ReplicationSlotsConfiguration struct {
+	// Replication slots for high availability configuration
+	HighAvailability *ReplicationSlotsHAConfiguration `json:"highAvailability,omitempty"`
+
+	// Standby will update the status of the local replication slots
+	// every `updateInterval` seconds (default 30).
+	//+kubebuilder:default:=30
+	//+kubebuilder:validation:Minimum=1
+	UpdateInterval int `json:"updateInterval,omitempty"`
+}
+
+// GetUpdateInterval returns the update interval, defaulting to DefaultReplicationSlotsUpdateInterval if empty
+func (r *ReplicationSlotsConfiguration) GetUpdateInterval() time.Duration {
+	if r == nil || r.UpdateInterval <= 0 {
+		return DefaultReplicationSlotsUpdateInterval
+	}
+	return time.Duration(r.UpdateInterval) * time.Second
+}
+
+// ReplicationSlotsHAConfiguration encapsulates the configuration
+// of the replication slots that are automatically managed by
+// the operator to control the streaming replication connections
+// with the standby instances for high availability (HA) purposes.
+// Replication slots are a PostgreSQL feature that makes sure
+// that PostgreSQL automatically keeps WAL files in the primary
+// when a streaming client (in this specific case a replica that
+// is part of the HA cluster) gets disconnected.
+type ReplicationSlotsHAConfiguration struct {
+	// If enabled, the operator will automatically manage replication slots
+	// on the primary instance and use them in streaming replication
+	// connections with all the standby instances that are part of the HA
+	// cluster. If disabled (default), the operator will not take advantage
+	// of replication slots in streaming connections with the replicas.
+	// This feature also controls replication slots in replica cluster,
+	// from the designated primary to its cascading replicas. This can only
+	// be set at creation time.
+	// +optional
+	Enabled bool `json:"enabled"`
+
+	// Prefix for replication slots managed by the operator for HA.
+	// It may only contain lower case letters, numbers, and the underscore character.
+	// This can only be set at creation time. By default set to `_cnpg_`.
+	//+kubebuilder:default:=_cnpg_
+	//+kubebuilder:validation:Pattern=^[0-9a-z_]*$
+	SlotPrefix string `json:"slotPrefix,omitempty"`
+}
+
+// GetSlotPrefix returns the HA slot prefix, defaulting to DefaultReplicationSlotsHASlotPrefix if empty
+func (r *ReplicationSlotsHAConfiguration) GetSlotPrefix() string {
+	if r == nil || r.SlotPrefix == "" {
+		return DefaultReplicationSlotsHASlotPrefix
+	}
+	return r.SlotPrefix
+}
+
+// GetSlotNameFromInstanceName returns the slot name, given the instance name.
+// It returns an empty string if High Availability Replication Slots are disabled
+func (r *ReplicationSlotsHAConfiguration) GetSlotNameFromInstanceName(instanceName string) string {
+	if r == nil || !r.Enabled {
+		return ""
+	}
+
+	slotName := fmt.Sprintf(
+		"%s%s",
+		r.GetSlotPrefix(),
+		instanceName,
+	)
+	sanitizedName := slotNameNegativeRegex.ReplaceAllString(strings.ToLower(slotName), "_")
+
+	return sanitizedName
+}
+
 // KubernetesUpgradeStrategy tells the operator if the user want to
 // allocate more space while upgrading a k8s node which is hosting
 // the PostgreSQL Pods or just wait for the node to come up
@@ -999,7 +1088,7 @@ type StorageConfiguration struct {
 //
 // In future synchronous replica election restriction by name will be supported.
 type SyncReplicaElectionConstraints struct {
-	// This flag enabled the constraints for sync replicas
+	// This flag enables the constraints for sync replicas
 	Enabled bool `json:"enabled"`
 
 	// A list of node labels values to extract and compare to evaluate if the pods reside in the same topology or not
@@ -1894,6 +1983,20 @@ func (cluster Cluster) IsReplica() bool {
 	return cluster.Spec.ReplicaCluster != nil && cluster.Spec.ReplicaCluster.Enabled
 }
 
+var slotNameNegativeRegex = regexp.MustCompile("[^a-z0-9_]+")
+
+// GetSlotNameFromInstanceName returns the slot name, given the instance name.
+// It returns an empty string if High Availability Replication Slots are disabled
+func (cluster Cluster) GetSlotNameFromInstanceName(instanceName string) string {
+	if cluster.Spec.ReplicationSlots == nil ||
+		cluster.Spec.ReplicationSlots.HighAvailability == nil ||
+		!cluster.Spec.ReplicationSlots.HighAvailability.Enabled {
+		return ""
+	}
+
+	return cluster.Spec.ReplicationSlots.HighAvailability.GetSlotNameFromInstanceName(instanceName)
+}
+
 // GetBarmanEndpointCAForReplicaCluster checks if this is a replica cluster which needs barman endpoint CA
 func (cluster Cluster) GetBarmanEndpointCAForReplicaCluster() *SecretKeySelector {
 	if !cluster.IsReplica() {

api/v1/cluster_types_test.go

@@ -795,3 +795,51 @@ var _ = Describe("Barman credentials", func() {
 		}.ArePopulated()).To(BeTrue())
 	})
 })
+
+var _ = Describe("Replication slots names for instances", func() {
+	It("returns an empty name when no replication slots are configured", func() {
+		cluster := Cluster{}
+		Expect(cluster.GetSlotNameFromInstanceName("cluster-example-1")).To(BeEmpty())
+
+		cluster = Cluster{
+			Spec: ClusterSpec{
+				ReplicationSlots: &ReplicationSlotsConfiguration{
+					HighAvailability: nil,
+					UpdateInterval:   0,
+				},
+			},
+		}
+		Expect(cluster.GetSlotNameFromInstanceName("cluster-example-1")).To(BeEmpty())
+	})
+
+	It("returns the name of the slot for an instance when they are configured", func() {
+		cluster := Cluster{
+			Spec: ClusterSpec{
+				ReplicationSlots: &ReplicationSlotsConfiguration{
+					HighAvailability: &ReplicationSlotsHAConfiguration{
+						Enabled: true,
+					},
+					UpdateInterval: 0,
+				},
+			},
+		}
+		Expect(cluster.GetSlotNameFromInstanceName("cluster-example-1")).To(Equal(
+			"_cnpg_cluster_example_1"))
+	})
+
+	It("sanitizes the name of replication slots", func() {
+		cluster := Cluster{
+			Spec: ClusterSpec{
+				ReplicationSlots: &ReplicationSlotsConfiguration{
+					HighAvailability: &ReplicationSlotsHAConfiguration{
+						Enabled:    true,
+						SlotPrefix: "%232'test_",
+					},
+					UpdateInterval: 0,
+				},
+			},
+		}
+		Expect(cluster.GetSlotNameFromInstanceName("cluster-example-1")).To(Equal(
+			"_232_test_cluster_example_1"))
+	})
+})

api/v1/cluster_webhook.go

@@ -292,6 +292,7 @@ func (r *Cluster) Validate() (allErrs field.ErrorList) {
 		r.validateBackupConfiguration,
 		r.validateConfiguration,
 		r.validateLDAP,
+		r.validateReplicationSlots,
 	}
 
 	for _, validate := range validations {
@@ -337,6 +338,7 @@ func (r *Cluster) ValidateChanges(old *Cluster) (allErrs field.ErrorList) {
 	allErrs = append(allErrs, r.validateWalStorageChange(old)...)
 	allErrs = append(allErrs, r.validateReplicaModeChange(old)...)
 	allErrs = append(allErrs, r.validateUnixPermissionIdentifierChange(old)...)
+	allErrs = append(allErrs, r.validateReplicationSlotsChange(old)...)
 	return allErrs
 }
 
@@ -1535,6 +1537,69 @@ func (r *Cluster) validateRecoveryAndBackupTarget() field.ErrorList {
 	return allErrors
 }
 
+func (r *Cluster) validateReplicationSlots() field.ErrorList {
+	replicationSlots := r.Spec.ReplicationSlots
+	if replicationSlots == nil ||
+		replicationSlots.HighAvailability == nil ||
+		!replicationSlots.HighAvailability.Enabled {
+		return nil
+	}
+
+	psqlVersion, err := r.GetPostgresqlVersion()
+	if err != nil {
+		// The validation error will be already raised by the
+		// validateImageName function
+		return nil
+	}
+
+	if psqlVersion >= 110000 {
+		return nil
+	}
+
+	return field.ErrorList{
+		field.Invalid(
+			field.NewPath("spec", "replicationSlots", "highAvailability", "enabled"),
+			replicationSlots.HighAvailability.Enabled,
+			"Cannot enable replication slot high availability. It requires PostgreSQL 11 or above"),
+	}
+}
+
+func (r *Cluster) validateReplicationSlotsChange(old *Cluster) field.ErrorList {
+	newReplicationSlots := r.Spec.ReplicationSlots
+	oldReplicationSlots := old.Spec.ReplicationSlots
+
+	if oldReplicationSlots == nil || oldReplicationSlots.HighAvailability == nil ||
+		!oldReplicationSlots.HighAvailability.Enabled {
+		return nil
+	}
+
+	var errs field.ErrorList
+
+	// when disabling we should check that the prefix it's not removed, and it doesn't change to
+	// properly execute the cleanup logic
+	if newReplicationSlots == nil || newReplicationSlots.HighAvailability == nil {
+		path := field.NewPath("spec", "replicationSlots")
+		if newReplicationSlots != nil {
+			path = path.Child("highAvailability")
+		}
+		errs = append(errs,
+			field.Invalid(
+				path,
+				nil,
+				fmt.Sprintf("Cannot remove %v section while highAvailability is enabled", path)),
+		)
+	} else if oldReplicationSlots.HighAvailability.SlotPrefix != newReplicationSlots.HighAvailability.SlotPrefix {
+		errs = append(errs,
+			field.Invalid(
+				field.NewPath("spec", "replicationSlots", "highAvailability", "slotPrefix"),
+				newReplicationSlots.HighAvailability.SlotPrefix,
+				"Cannot change replication slot prefix while highAvailability is enabled"),
+		)
+	}
+
+	return errs
+}
+
 // validateAzureCredentials checks and validates the azure credentials
 func (azure *AzureCredentials) validateAzureCredentials(path *field.Path) field.ErrorList {
 	allErrors := field.ErrorList{}