test: e2e tests for replication slots (cloudnative-pg#760)

refine replication slots tests. Test failover/switchover/scaling with
replication slots on or off

Signed-off-by: Jitendra Wadle <jitendra.wadle@enterprisedb.com>
Signed-off-by: Danish Khan <danish.khan@enterprisedb.com>
Signed-off-by: Niccolò Fei <niccolo.fei@enterprisedb.com>
Signed-off-by: Jaime Silvela <jaime.silvela@enterprisedb.com>
Signed-off-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>
Co-authored-by: Danish Khan <danish.khan@enterprisedb.com>
Co-authored-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>
Co-authored-by: Niccolò Fei <niccolo.fei@enterprisedb.com>
Co-authored-by: Jaime Silvela <jaime.silvela@enterprisedb.com>

Author: 5 people

docs/src/e2e.md

@@ -57,3 +57,4 @@ and the following suite of E2E tests are performed on that cluster:
 * Physical replica clusters;
 * Storage expansion;
 * Data corruption;
+* Replication Slots;

tests/e2e/asserts_test.go

@@ -2493,70 +2493,114 @@ func AssertPvcHasLabels(
 	})
 }
 
-// AssertRepSlotsOnPod checks if all the required replication slot exists in a given pod,
+// AssertReplicationSlotsOnPod checks that all the required replication slots exist in a given pod,
 // and that obsolete slots are correctly deleted (post management operations).
-// In case we are targeting the primary, it will also check if the slot is active.
-func AssertRepSlotsOnPod(
+// In the primary, it will also check if the slots are active.
+func AssertReplicationSlotsOnPod(
 	namespace,
 	clusterName string,
 	pod corev1.Pod,
 ) {
-	expectedSlots, err := testsUtils.GetExpectedRepSlotsOnPod(namespace, clusterName, pod.GetName(), env)
+	// Replication slot high availability requires PostgreSQL 11 or above
+	if env.PostgresVersion == 10 {
+		Skip("Ignoring replication slots verification for postgres 10")
+		return
+	}
+
+	expectedSlots, err := testsUtils.GetExpectedReplicationSlotsOnPod(namespace, clusterName, pod.GetName(), env)
 	Expect(err).ToNot(HaveOccurred())
 
 	Eventually(func() ([]string, error) {
-		currentSlots, err := testsUtils.GetRepSlotsOnPod(namespace, pod.GetName(), env)
+		currentSlots, err := testsUtils.GetReplicationSlotsOnPod(namespace, pod.GetName(), env)
 		return currentSlots, err
 	}, 300).Should(BeEquivalentTo(expectedSlots),
-		fmt.Sprintf(
-			"List of expected replication slots on %v pod %v",
-			pod.Labels["role"],
-			pod.GetName()))
+		func() string {
+			return testsUtils.PrintReplicationSlots(namespace, clusterName, env)
+		})
 
 	for _, slot := range expectedSlots {
 		query := fmt.Sprintf(
 			"SELECT EXISTS (SELECT 1 FROM pg_replication_slots "+
 				"WHERE slot_name = '%v' AND active = 'f' "+
 				"AND temporary = 'f' AND slot_type = 'physical')", slot)
-		description := fmt.Sprintf(
-			"On %v pod %v, expect replication slot %v to exist and be inactive",
-			pod.Labels["role"],
-			pod.GetName(),
-			slot)
 		if specs.IsPodPrimary(pod) {
 			query = fmt.Sprintf(
 				"SELECT EXISTS (SELECT 1 FROM pg_replication_slots "+
 					"WHERE slot_name = '%v' AND active = 't' "+
 					"AND temporary = 'f' AND slot_type = 'physical')", slot)
-			description = fmt.Sprintf(
-				"On %v pod %v, expect replication slot %v to exist and be active",
-				pod.Labels["role"],
-				pod.GetName(),
-				slot)
 		}
 		Eventually(func() (string, error) {
 			stdout, _, err := testsUtils.RunQueryFromPod(&pod, testsUtils.PGLocalSocketDir,
 				"app", "postgres", "''", query, env)
 			return strings.TrimSpace(stdout), err
-		}, 300).Should(BeEquivalentTo("t"), description)
+		}, 300).Should(BeEquivalentTo("t"),
+			func() string {
+				return testsUtils.PrintReplicationSlots(namespace, clusterName, env)
+			})
 	}
 }
 
-// AssertClusterRepSlotsAligned will compare all the replication slot restart_lsn
+// AssertClusterReplicationSlotsAligned will compare all the replication slot restart_lsn
 // in a cluster. The assertion will succeed if they are all equivalent.
-func AssertClusterRepSlotsAligned(
+func AssertClusterReplicationSlotsAligned(
 	namespace,
 	clusterName string,
 ) {
+	// Replication slot high availability requires PostgreSQL 11 or above
+	if env.PostgresVersion == 10 {
+		Skip("Ignoring replication slots verification for postgres 10")
+	}
+
 	podList, err := env.GetClusterPodList(namespace, clusterName)
 	Expect(err).ToNot(HaveOccurred())
 	Eventually(func() bool {
 		var lsnList []string
 		for _, pod := range podList.Items {
-			out, err := testsUtils.GetRepSlotsLsnOnPod(namespace, clusterName, pod, env)
+			out, err := testsUtils.GetReplicationSlotLsnsOnPod(namespace, clusterName, pod, env)
 			Expect(err).ToNot(HaveOccurred())
 			lsnList = append(lsnList, out...)
 		}
-		return testsUtils.CompareLsn(lsnList)
-	}, 300).Should(BeEquivalentTo(true))
+		return testsUtils.AreSameLsn(lsnList)
+	}, 300).Should(BeEquivalentTo(true),
+		func() string {
+			return testsUtils.PrintReplicationSlots(namespace, clusterName, env)
+		})
+}
+
+// AssertClusterReplicationSlots will verify if the replication slots of each pod
+// of the cluster exist and are aligned.
+func AssertClusterReplicationSlots(namespace, clusterName string) {
+	By("verifying all cluster's replication slots exist and are aligned", func() {
+		podList, err := env.GetClusterPodList(namespace, clusterName)
+		Expect(err).ToNot(HaveOccurred())
+		for _, pod := range podList.Items {
+			AssertReplicationSlotsOnPod(namespace, clusterName, pod)
+		}
+		AssertClusterReplicationSlotsAligned(namespace, clusterName)
+	})
+}
+
+// AssertClusterRollingRestart restart given cluster
+func AssertClusterRollingRestart(namespace, clusterName string) {
+	By(fmt.Sprintf("restarting cluster %v", clusterName), func() {
+		cluster, err := env.GetCluster(namespace, clusterName)
+		Expect(err).ToNot(HaveOccurred())
+		clusterRestarted := cluster.DeepCopy()
+		if clusterRestarted.Annotations == nil {
+			clusterRestarted.Annotations = make(map[string]string)
+		}
+		clusterRestarted.Annotations[specs.ClusterRestartAnnotationName] = time.Now().Format(time.RFC3339)
+		clusterRestarted.ManagedFields = nil
+		err = env.Client.Patch(env.Ctx, clusterRestarted, ctrlclient.MergeFrom(cluster))
+		Expect(err).ToNot(HaveOccurred())
+	})
+
+	By("waiting for the cluster to end up in upgrading state", func() {
+		// waiting for cluster phase to end up in "Upgrading cluster" state after restarting the cluster.
+		Eventually(func() (bool, error) {
+			cluster, err := env.GetCluster(namespace, clusterName)
+			return cluster.Status.Phase == apiv1.PhaseUpgrade, err
+		}, 120, 3).Should(BeTrue())
+	})
+	AssertClusterIsReady(namespace, clusterName, 300, env)
 }

tests/e2e/fastfailover_test.go

@@ -27,12 +27,13 @@ import (
 
 var _ = Describe("Fast failover", Serial, Label(tests.LabelPerformance), func() {
 	const (
-		sampleFile             = fixturesDir + "/fastfailover/cluster-fast-failover.yaml.template"
-		sampleFileSyncReplicas = fixturesDir + "/fastfailover/cluster-syncreplicas-fast-failover.yaml.template"
-		webTestFile            = fixturesDir + "/fastfailover/webtest.yaml"
-		webTestSyncReplicas    = fixturesDir + "/fastfailover/webtest-syncreplicas.yaml"
-		webTestJob             = fixturesDir + "/fastfailover/apache-benchmark-webtest.yaml"
-		level                  = tests.Highest
+		sampleFileWithoutReplicationSlots = fixturesDir + "/fastfailover/cluster-fast-failover.yaml.template"
+		sampleFileWithReplicationSlots    = fixturesDir + "/fastfailover/cluster-fast-failover-with-repl-slots.yaml.template"
+		sampleFileSyncReplicas            = fixturesDir + "/fastfailover/cluster-syncreplicas-fast-failover.yaml.template"
+		webTestFile                       = fixturesDir + "/fastfailover/webtest.yaml"
+		webTestSyncReplicas               = fixturesDir + "/fastfailover/webtest-syncreplicas.yaml"
+		webTestJob                        = fixturesDir + "/fastfailover/apache-benchmark-webtest.yaml"
+		level                             = tests.Highest
 	)
 	var (
 		namespace       string
@@ -82,7 +83,7 @@ var _ = Describe("Fast failover", Serial, Label(tests.LabelPerformance), func()
 	})
 
 	AfterEach(func() {
-		err := env.DeleteNamespace(namespace)
+		err := env.DeleteNamespaceAndWait(namespace, 120)
 		Expect(err).ToNot(HaveOccurred())
 	})
 
@@ -95,7 +96,26 @@ var _ = Describe("Fast failover", Serial, Label(tests.LabelPerformance), func()
 		It("can do a fast failover", func() {
 			namespace = "primary-failover-time"
 			clusterName = "cluster-fast-failover"
-			AssertFastFailOver(namespace, sampleFile, clusterName, webTestFile, webTestJob, maxReattachTime, maxFailoverTime)
+			AssertFastFailOver(namespace, sampleFileWithoutReplicationSlots, clusterName,
+				webTestFile, webTestJob, maxReattachTime, maxFailoverTime)
+		})
+	})
+
+	Context("with async replicas cluster and HA Replication Slots", func() {
+		// Confirm that a standby closely following the primary doesn't need more
+		// than 10 seconds to be promoted and be able to start inserting records.
+		// We test this setting up an application pointing to the rw service,
+		// forcing a failover and measuring how much time passes between the
+		// last row written on timeline 1 and the first one on timeline 2.
+		It("can do a fast failover", func() {
+			if env.PostgresVersion == 10 {
+				Skip("replication slots not available in PostgreSQL 10 or older")
+			}
+			namespace = "primary-failover-time"
+			clusterName = "cluster-fast-failover"
+			AssertFastFailOver(namespace, sampleFileWithReplicationSlots,
+				clusterName, webTestFile, webTestJob, maxReattachTime, maxFailoverTime)
+			AssertClusterReplicationSlots(namespace, clusterName)
 		})
 	})