Add helper functions for e2e tests (submariner-io#198)

* Add helper functions for e2e tests

Added some helper functions for upcoming e2e enhancements.

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

test/e2e/framework/api_errors.go

@@ -6,11 +6,17 @@ import (
 
 // identify API errors which could be considered transient/recoverable
 // due to server state.
-func IsTransientError(err error) bool {
-	return errors.IsInternalError(err) ||
+func IsTransientError(err error, opMsg string) bool {
+	if errors.IsInternalError(err) ||
 		errors.IsServerTimeout(err) ||
 		errors.IsTimeout(err) ||
 		errors.IsServiceUnavailable(err) ||
 		errors.IsUnexpectedServerError(err) ||
-		errors.IsTooManyRequests(err)
+		errors.IsTooManyRequests(err) {
+
+		Logf("Transient failure when attempting to %s: %v", opMsg, err)
+		return true
+	}
+
+	return false
 }

test/e2e/framework/framework.go

@@ -1,14 +1,17 @@
 package framework
 
 import (
+	"encoding/json"
 	"fmt"
 	"os"
 	"strings"
 	"time"
 
 	"github.com/onsi/ginkgo"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/uuid"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 
@@ -34,6 +37,22 @@ const (
 	ClusterC
 )
 
+const (
+	SubmarinerEngine = "submariner-engine"
+	GatewayLabel     = "submariner.io/gateway"
+)
+
+type PatchFunc func(pt types.PatchType, payload []byte) error
+
+type PatchStringValue struct {
+	Op    string `json:"op"`
+	Path  string `json:"path"`
+	Value string `json:"value"`
+}
+
+type DoOperationFunc func() (interface{}, error)
+type CheckResultFunc func(result interface{}) (bool, error)
+
 // Framework supports common operations used by e2e tests; it will keep a client & a namespace for you.
 // Eventual goal is to merge this with integration test framework.
 type Framework struct {
@@ -275,3 +294,53 @@ func createNamespace(client kubeclientset.Interface, name string, labels map[str
 	Expect(err).NotTo(HaveOccurred(), "Error creating namespace %v", namespaceObj)
 	return namespace
 }
+
+// DoPatchOperation performs a REST patch operation for the given path and value.
+func DoPatchOperation(path string, value string, patchFunc PatchFunc) {
+	payload := []PatchStringValue{{
+		Op:    "add",
+		Path:  path,
+		Value: value,
+	}}
+
+	payloadBytes, err := json.Marshal(payload)
+	Expect(err).NotTo(HaveOccurred())
+
+	AwaitUntil("perform patch operation", func() (interface{}, error) {
+		return nil, patchFunc(types.JSONPatchType, payloadBytes)
+	}, NoopCheckResult)
+}
+
+func NoopCheckResult(interface{}) (bool, error) {
+	return true, nil
+}
+
+// AwaitUntil periodically performs the given operation until the given CheckResultFunc returns true, an error, or a
+// timeout is reached.
+func AwaitUntil(opMsg string, doOperation DoOperationFunc, checkResult CheckResultFunc) interface{} {
+	var finalResult interface{}
+	err := wait.PollImmediate(5*time.Second, 1*time.Minute, func() (bool, error) {
+		result, err := doOperation()
+		if err != nil {
+			if IsTransientError(err, opMsg) {
+				return false, nil
+			}
+			return false, err
+		}
+
+		ok, err := checkResult(result)
+		if err != nil {
+			return false, err
+		}
+
+		if ok {
+			finalResult = result
+			return true, nil
+		}
+
+		return false, nil
+	})
+
+	Expect(err).NotTo(HaveOccurred(), "Failed to "+opMsg)
+	return finalResult
+}

test/e2e/framework/network_pods.go

@@ -3,12 +3,10 @@ package framework
 import (
 	"fmt"
 	"strconv"
-	"time"
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/uuid"
-	"k8s.io/apimachinery/pkg/util/wait"
 
 	. "github.com/onsi/gomega"
 )
@@ -79,43 +77,31 @@ func (f *Framework) NewNetworkPod(config *NetworkPodConfig) *NetworkPod {
 }
 
 func (np *NetworkPod) AwaitReady() {
-	var finalPod *v1.Pod
-	pc := np.framework.ClusterClients[np.Config.Cluster].CoreV1().Pods(np.framework.Namespace)
-	err := wait.PollImmediate(5*time.Second, 2*time.Minute, func() (bool, error) {
-		pod, err := pc.Get(np.Pod.Name, metav1.GetOptions{})
-		if err != nil {
-			if IsTransientError(err) {
-				Logf("Transient failure when attempting to list pods: %v", err)
-				return false, nil // return nil to avoid PollImmediate from stopping
-			}
-			return false, err
-		}
+	pods := np.framework.ClusterClients[np.Config.Cluster].CoreV1().Pods(np.framework.Namespace)
+
+	np.Pod = AwaitUntil("get pod", func() (interface{}, error) {
+		return pods.Get(np.Pod.Name, metav1.GetOptions{})
+	}, func(result interface{}) (bool, error) {
+		pod := result.(*v1.Pod)
 		if pod.Status.Phase != v1.PodRunning {
 			if pod.Status.Phase != v1.PodPending {
 				return false, fmt.Errorf("expected pod to be in phase \"Pending\" or \"Running\"")
 			}
 			return false, nil // pod is still pending
 		}
-		finalPod = pod
-		return true, nil // pods is running
-	})
-	Expect(err).NotTo(HaveOccurred())
-	np.Pod = finalPod
+
+		return true, nil // pod is running
+	}).(*v1.Pod)
 }
 
 func (np *NetworkPod) AwaitSuccessfulFinish() {
-	pc := np.framework.ClusterClients[np.Config.Cluster].CoreV1().Pods(np.framework.Namespace)
+	pods := np.framework.ClusterClients[np.Config.Cluster].CoreV1().Pods(np.framework.Namespace)
+
+	np.Pod = AwaitUntil("get pod", func() (interface{}, error) {
+		return pods.Get(np.Pod.Name, metav1.GetOptions{})
+	}, func(result interface{}) (bool, error) {
+		np.Pod = result.(*v1.Pod)
 
-	err := wait.PollImmediate(5*time.Second, 2*time.Minute, func() (bool, error) {
-		var err error
-		np.Pod, err = pc.Get(np.Pod.Name, metav1.GetOptions{})
-		if err != nil {
-			if IsTransientError(err) {
-				Logf("Transient failure when attempting to list pods: %v", err)
-				return false, nil // return nil to avoid PollImmediate from stopping
-			}
-			return false, err
-		}
 		switch np.Pod.Status.Phase {
 		case v1.PodSucceeded:
 			return true, nil
@@ -124,9 +110,7 @@ func (np *NetworkPod) AwaitSuccessfulFinish() {
 		default:
 			return false, nil
 		}
-	})
-
-	Expect(err).NotTo(HaveOccurred())
+	}).(*v1.Pod)
 
 	finished := np.Pod.Status.Phase == v1.PodSucceeded || np.Pod.Status.Phase == v1.PodFailed
 	Expect(finished).To(BeTrue())
@@ -219,18 +203,15 @@ func (np *NetworkPod) buildTCPCheckConnectorPod() {
 }
 
 func nodeAffinity(scheduling NetworkPodScheduling) *v1.Affinity {
-
-	const gatewayLabel = "submariner.io/gateway"
-
 	var nodeSelReqs []v1.NodeSelectorRequirement
 
 	switch scheduling {
 	case GatewayNode:
-		nodeSelReqs = addNodeSelectorRequirement(nodeSelReqs, gatewayLabel, v1.NodeSelectorOpIn, []string{"true"})
+		nodeSelReqs = addNodeSelectorRequirement(nodeSelReqs, GatewayLabel, v1.NodeSelectorOpIn, []string{"true"})
 
 	case NonGatewayNode:
-		nodeSelReqs = addNodeSelectorRequirement(nodeSelReqs, gatewayLabel, v1.NodeSelectorOpDoesNotExist, nil)
-		nodeSelReqs = addNodeSelectorRequirement(nodeSelReqs, gatewayLabel, v1.NodeSelectorOpNotIn, []string{"true"})
+		nodeSelReqs = addNodeSelectorRequirement(nodeSelReqs, GatewayLabel, v1.NodeSelectorOpDoesNotExist, nil)
+		nodeSelReqs = addNodeSelectorRequirement(nodeSelReqs, GatewayLabel, v1.NodeSelectorOpNotIn, []string{"true"})
 	}
 
 	affinity := v1.Affinity{

test/e2e/framework/nodes.go

@@ -0,0 +1,46 @@
+package framework
+
+import (
+	"strconv"
+	"strings"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+// FindNodesByGatewayLabel finds the nodes in a given cluster by matching 'submariner.io/gateway' value. A missing label
+// is treated as false. Note the control plane node labeled as master is ignored.
+func (f *Framework) FindNodesByGatewayLabel(cluster ClusterIndex, isGateway bool) []*v1.Node {
+	nodes := AwaitUntil("list nodes", func() (interface{}, error) {
+		// Ignore the control plane node labeled as master as it doesn't allow scheduling of pods
+		return f.ClusterClients[cluster].CoreV1().Nodes().List(metav1.ListOptions{
+			LabelSelector: "!node-role.kubernetes.io/master",
+		})
+	}, NoopCheckResult).(*v1.NodeList)
+
+	expLabelValue := strconv.FormatBool(isGateway)
+	retNodes := []*v1.Node{}
+	for i := range nodes.Items {
+		value, exists := nodes.Items[i].Labels[GatewayLabel]
+		if !exists {
+			value = "false"
+		}
+
+		if value == expLabelValue {
+			retNodes = append(retNodes, &nodes.Items[i])
+		}
+	}
+
+	return retNodes
+}
+
+// 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),
+		func(pt types.PatchType, payload []byte) error {
+			_, err := f.ClusterClients[cluster].CoreV1().Nodes().Patch(nodeName, pt, payload)
+			return err
+		})
+}

test/e2e/framework/pods.go

@@ -0,0 +1,37 @@
+package framework
+
+import (
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// AwaitPodsByAppLabel finds pods in a given cluster whose 'app' label value matches a specified value. If the specified
+// expectedCount >= 0, the function waits until the number of pods equals the expectedCount.
+func (f *Framework) AwaitPodsByAppLabel(cluster ClusterIndex, appName string, namespace string, expectedCount int) *v1.PodList {
+	return AwaitUntil("find pods for app "+appName, func() (interface{}, error) {
+		return f.ClusterClients[cluster].CoreV1().Pods(namespace).List(metav1.ListOptions{
+			LabelSelector: "app=" + appName,
+		})
+	}, func(result interface{}) (bool, error) {
+		pods := result.(*v1.PodList)
+		if expectedCount < 0 || len(pods.Items) == expectedCount {
+			return true, nil
+		}
+
+		Logf("Actual pod count %d does not match the expected pod count %d - retrying...", len(pods.Items), expectedCount)
+		return false, nil
+	}).(*v1.PodList)
+}
+
+// AwaitSubmarinerEnginePod finds the submariner engine pod in a given cluster, waiting if necessary for a period of time
+// for the pod to materialize.
+func (f *Framework) AwaitSubmarinerEnginePod(cluster ClusterIndex) *v1.Pod {
+	return &f.AwaitPodsByAppLabel(cluster, SubmarinerEngine, TestContext.SubmarinerNamespace, 1).Items[0]
+}
+
+// DeletePod deletes the pod for the given name and namespace.
+func (f *Framework) DeletePod(cluster ClusterIndex, podName string, namespace string) {
+	AwaitUntil("list pods", func() (interface{}, error) {
+		return nil, f.ClusterClients[cluster].CoreV1().Pods(namespace).Delete(podName, &metav1.DeleteOptions{})
+	}, NoopCheckResult)
+}

test/e2e/framework/test_context.go

@@ -11,10 +11,11 @@ import (
 type contextArray []string
 
 type TestContextType struct {
-	KubeConfig   string
-	KubeContexts contextArray
-	ReportDir    string
-	ReportPrefix string
+	KubeConfig          string
+	KubeContexts        contextArray
+	ReportDir           string
+	ReportPrefix        string
+	SubmarinerNamespace string
 }
 
 func (contexts *contextArray) String() string {
@@ -34,6 +35,7 @@ func registerFlags(t *TestContextType) {
 	flag.Var(&TestContext.KubeContexts, "dp-context", "kubeconfig context for dataplane clusters (use several times).")
 	flag.StringVar(&TestContext.ReportPrefix, "report-prefix", "", "Optional prefix for JUnit XML reports. Default is empty, which doesn't prepend anything to the default name.")
 	flag.StringVar(&TestContext.ReportDir, "report-dir", "", "Path to the directory where the JUnit XML reports should be saved. Default is empty, which doesn't generate these reports.")
+	flag.StringVar(&TestContext.SubmarinerNamespace, "submariner-namespace", "submariner", "Namespace in which the submariner components are deployed.")
 }
 
 func validateFlags(t *TestContextType) {