Add third gateway fail-over E2E scenario test (submariner-io#264)

Two gateway nodes configured with 2 submariner engine replicas. This tests
the leader election fail-over.

Signed-off-by: Tom Pantelis <tompantelis@gmail.com>

test/e2e/framework/deployments.go

@@ -0,0 +1,34 @@
+package framework
+
+import (
+	"fmt"
+
+	. "github.com/onsi/gomega"
+	appsv1 "k8s.io/api/apps/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+func (f *Framework) FindDeployment(cluster ClusterIndex, appName string, namespace string) *appsv1.Deployment {
+	deployments := AwaitUntil("list deployments", func() (interface{}, error) {
+		return f.ClusterClients[cluster].AppsV1().Deployments(namespace).List(metav1.ListOptions{
+			LabelSelector: "app=" + appName,
+		})
+	}, NoopCheckResult).(*appsv1.DeploymentList)
+	Expect(deployments.Items).To(HaveLen(1), fmt.Sprintf("Expected one %q deployment on %q",
+		appName, TestContext.KubeContexts[cluster]))
+
+	return &deployments.Items[0]
+}
+
+func (f *Framework) FindSubmarinerEngineDeployment(cluster ClusterIndex) *appsv1.Deployment {
+	return f.FindDeployment(cluster, SubmarinerEngine, TestContext.SubmarinerNamespace)
+}
+
+func (f *Framework) SetReplicas(cluster ClusterIndex, deployment *appsv1.Deployment, replicas uint32) {
+	PatchInt("/spec/replicas", replicas,
+		func(pt types.PatchType, payload []byte) error {
+			_, err := f.ClusterClients[cluster].AppsV1().Deployments(deployment.Namespace).Patch(deployment.Name, pt, payload)
+			return err
+		})
+}

test/e2e/framework/framework.go

@@ -50,6 +50,12 @@ type PatchStringValue struct {
 	Value string `json:"value"`
 }
 
+type PatchUInt32Value struct {
+	Op    string `json:"op"`
+	Path  string `json:"path"`
+	Value uint32 `json:"value"`
+}
+
 type DoOperationFunc func() (interface{}, error)
 type CheckResultFunc func(result interface{}) (bool, string, error)
 
@@ -300,14 +306,29 @@ func createNamespace(client kubeclientset.Interface, name string, labels map[str
 	return namespace
 }
 
-// DoPatchOperation performs a REST patch operation for the given path and value.
-func DoPatchOperation(path string, value string, patchFunc PatchFunc) {
+// PatchString performs a REST patch operation for the given path and string value.
+func PatchString(path string, value string, patchFunc PatchFunc) {
 	payload := []PatchStringValue{{
 		Op:    "add",
 		Path:  path,
 		Value: value,
 	}}
 
+	doPatchOperation(payload, patchFunc)
+}
+
+// PatchInt performs a REST patch operation for the given path and int value.
+func PatchInt(path string, value uint32, patchFunc PatchFunc) {
+	payload := []PatchUInt32Value{{
+		Op:    "add",
+		Path:  path,
+		Value: value,
+	}}
+
+	doPatchOperation(payload, patchFunc)
+}
+
+func doPatchOperation(payload interface{}, patchFunc PatchFunc) {
 	payloadBytes, err := json.Marshal(payload)
 	Expect(err).NotTo(HaveOccurred())
 

test/e2e/framework/nodes.go

@@ -38,7 +38,7 @@ func (f *Framework) FindNodesByGatewayLabel(cluster ClusterIndex, isGateway bool
 // SetGatewayLabelOnNode sets the 'submariner.io/gateway' value for a node to the specified value.
 func (f *Framework) SetGatewayLabelOnNode(cluster ClusterIndex, nodeName string, isGateway bool) {
 	// Escape the '/' char in the label name with the special sequence "~1" so it isn't treated as part of the path
-	DoPatchOperation("/metadata/labels/"+strings.Replace(GatewayLabel, "/", "~1", -1), strconv.FormatBool(isGateway),
+	PatchString("/metadata/labels/"+strings.Replace(GatewayLabel, "/", "~1", -1), strconv.FormatBool(isGateway),
 		func(pt types.PatchType, payload []byte) error {
 			_, err := f.ClusterClients[cluster].CoreV1().Nodes().Patch(nodeName, pt, payload)
 			return err

test/e2e/framework/submariner_resources.go

@@ -1,10 +1,13 @@
 package framework
 
 import (
+	"fmt"
+
 	. "github.com/onsi/gomega"
 	submarinerv1 "github.com/submariner-io/submariner/pkg/apis/submariner.io/v1"
 	submarinerClientset "github.com/submariner-io/submariner/pkg/client/clientset/versioned"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 )
 
 type CheckEndpointFunc func(endpoint *submarinerv1.Endpoint) (bool, string, error)
@@ -47,3 +50,13 @@ func (f *Framework) AwaitSubmarinerEndpoint(cluster ClusterIndex, checkEndpoint
 
 	return retEndpoint
 }
+
+func (f *Framework) AwaitNewSubmarinerEndpoint(cluster ClusterIndex, prevEndpointUID types.UID) *submarinerv1.Endpoint {
+	return f.AwaitSubmarinerEndpoint(cluster, func(endpoint *submarinerv1.Endpoint) (bool, string, error) {
+		if endpoint.ObjectMeta.UID != prevEndpointUID {
+			return true, "", nil
+		}
+
+		return false, fmt.Sprintf("Expecting new Endpoint instance (UUID %q matches previous instance)", endpoint.ObjectMeta.UID), nil
+	})
+}

test/e2e/redundancy/gateway_failover.go

@@ -5,9 +5,9 @@ import (
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
-	submarinerv1 "github.com/submariner-io/submariner/pkg/apis/submariner.io/v1"
 	"github.com/submariner-io/submariner/test/e2e/dataplane"
 	"github.com/submariner-io/submariner/test/e2e/framework"
+	appsv1 "k8s.io/api/apps/v1"
 )
 
 var _ = Describe("[redundancy] Gateway fail-over tests", func() {
@@ -24,6 +24,22 @@ var _ = Describe("[redundancy] Gateway fail-over tests", func() {
 			testTwoGatewayNodesWithOneReplica(f)
 		})
 	})
+
+	When("two gateway nodes are configured with two submariner engine replicas and the active gateway node fails", func() {
+		var deployment *appsv1.Deployment
+
+		BeforeEach(func() {
+			deployment = f.FindSubmarinerEngineDeployment(framework.ClusterA)
+		})
+
+		It("should fail over to the second submariner engine running on the second gateway node and be able to connect from another cluster", func() {
+			testTwoGatewayNodesWithTwoReplicas(f, deployment)
+		})
+
+		AfterEach(func() {
+			f.SetReplicas(framework.ClusterA, deployment, 1)
+		})
+	})
 })
 
 func testOneGatewayNode(f *framework.Framework) {
@@ -93,13 +109,54 @@ func testTwoGatewayNodesWithOneReplica(f *framework.Framework) {
 
 	// Verify a new Endpoint instance is created by the new engine instance. This is a bit whitebox but it's a ssanity check
 	// and also gives it a bit more of a cushion to avoid premature timeout in the connectivity test.
-	newSubmEndpoint := f.AwaitSubmarinerEndpoint(framework.ClusterA, func(endpoint *submarinerv1.Endpoint) (bool, string, error) {
-		if endpoint.ObjectMeta.UID != submEndpoint.ObjectMeta.UID {
-			return true, "", nil
-		}
+	newSubmEndpoint := f.AwaitNewSubmarinerEndpoint(framework.ClusterA, submEndpoint.ObjectMeta.UID)
+	By(fmt.Sprintf("Found new submariner endpoint for %q: %#v", clusterAName, newSubmEndpoint))
 
-		return false, fmt.Sprintf("Expecting new Endpoint instance (UUID %q matches previous instance)", endpoint.ObjectMeta.UID), nil
-	})
+	By(fmt.Sprintf("Verifying TCP connectivity from gateway node on %q to gateway node on %q", clusterBName, clusterAName))
+	dataplane.RunConnectivityTest(f, false, framework.GatewayNode, framework.GatewayNode, framework.ClusterA, framework.ClusterB)
+
+	By(fmt.Sprintf("Verifying TCP connectivity from non-gateway node on %q to non-gateway node on %q", clusterBName, clusterAName))
+	dataplane.RunConnectivityTest(f, false, framework.NonGatewayNode, framework.NonGatewayNode, framework.ClusterA, framework.ClusterB)
+}
+
+func testTwoGatewayNodesWithTwoReplicas(f *framework.Framework, deployment *appsv1.Deployment) {
+	clusterAName := framework.TestContext.KubeContexts[framework.ClusterA]
+	clusterBName := framework.TestContext.KubeContexts[framework.ClusterB]
+
+	gatewayNodes := f.FindNodesByGatewayLabel(framework.ClusterA, true)
+	Expect(gatewayNodes).To(HaveLen(1), fmt.Sprintf("Expected only one gateway node on %q", clusterAName))
+	initGatewayNode := gatewayNodes[0]
+	By(fmt.Sprintf("Found gateway node %q on %q", initGatewayNode.Name, clusterAName))
+
+	nonGatewayNodes := f.FindNodesByGatewayLabel(framework.ClusterA, false)
+	Expect(nonGatewayNodes).ToNot(BeZero(), fmt.Sprintf("Expected at least one non-gateway node on %q", clusterAName))
+	nonGatewayNode := nonGatewayNodes[0]
+	By(fmt.Sprintf("Found non-gateway node %q on %q", nonGatewayNode.Name, clusterAName))
+
+	firstEnginePod := f.AwaitSubmarinerEnginePod(framework.ClusterA)
+	By(fmt.Sprintf("Found active submariner engine pod %q on %q", firstEnginePod.Name, clusterAName))
+
+	submEndpoint := f.AwaitSubmarinerEndpoint(framework.ClusterA, framework.NoopCheckEndpoint)
+	By(fmt.Sprintf("Found submariner endpoint for %q: %#v", clusterAName, submEndpoint))
+
+	By(fmt.Sprintf("Setting the replicas to 2 for submariner engine deployment %q on %q", deployment.Name, clusterAName))
+	f.SetReplicas(framework.ClusterA, deployment, 2)
+
+	By(fmt.Sprintf("Setting the gateway label for node %q to true", nonGatewayNode.Name))
+	f.SetGatewayLabelOnNode(framework.ClusterA, nonGatewayNode.Name, true)
+
+	By(fmt.Sprintf("Awaiting second submariner engine pod running on %q...", clusterAName))
+	enginePods := f.AwaitPodsByAppLabel(framework.ClusterA, framework.SubmarinerEngine, framework.TestContext.SubmarinerNamespace, 2)
+	By(fmt.Sprintf("2 submariner engine pods now running on %q: %q and %q", clusterAName, enginePods.Items[0].Name, enginePods.Items[1].Name))
+
+	By(fmt.Sprintf("Setting the gateway label for node %q to false", initGatewayNode.Name))
+	f.SetGatewayLabelOnNode(framework.ClusterA, initGatewayNode.Name, false)
+
+	By(fmt.Sprintf("Deleting active submariner engine pod %q", firstEnginePod.Name))
+	f.DeletePod(framework.ClusterA, firstEnginePod.Name, framework.TestContext.SubmarinerNamespace)
+
+	By(fmt.Sprintf("Awaiting new submariner endpoint on %q...", clusterAName))
+	newSubmEndpoint := f.AwaitNewSubmarinerEndpoint(framework.ClusterA, submEndpoint.ObjectMeta.UID)
 	By(fmt.Sprintf("Found new submariner endpoint for %q: %#v", clusterAName, newSubmEndpoint))
 
 	By(fmt.Sprintf("Verifying TCP connectivity from gateway node on %q to gateway node on %q", clusterBName, clusterAName))