Merge pull request oracle#199 in OKE/oci-cloud-controller-manager from task/OKE-18125 to internal

* commit 'e1cb4074cbf30e2d895207abba29e4c95db799a0':
  Added missing break statement
  Reverted pod start timeout change
  Changed test to make deployment use a single node to spawn pods
  Changed test to exec into restarted pod to check data
  JIRA:task/OKE-18125 added test to check data integrity on pod restart

Author: YashasG98

test/e2e/cloud-provider-oci/csi_volume_creation.go

@@ -58,6 +58,15 @@ var _ = Describe("CSI Volume Creation", func() {
 
 			pvcJig.CheckVolumeCapacity("100Gi", pvc.Name, f.Namespace.Name)
 		})
+
+		It("Data should persist on CSI volume on pod restart", func() {
+			pvcJig := framework.NewPVCTestJig(f.ClientSet, "csi-pod-restart-data-persistence")
+
+			scName := f.CreateStorageClassOrFail(framework.ClassOCICSI, "blockvolume.csi.oraclecloud.com", nil, pvcJig.Labels, "WaitForFirstConsumer")
+			pvc := pvcJig.CreateAndAwaitPVCOrFailCSI(f.Namespace.Name, framework.MinVolumeBlock, scName, nil)
+
+			pvcJig.CheckDataPersistenceWithDeployment(pvc.Name, f.Namespace.Name)
+		})
 	})
 })
 

test/e2e/framework/deployment_util.go

@@ -0,0 +1,92 @@
+package framework
+
+import (
+	appsv1 "k8s.io/api/apps/v1"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/utils/pointer"
+	"time"
+)
+
+func (j *PVCTestJig) createDeploymentOnNodeAndWait(command string, pvcName string, ns string, name string, replicas int32, nodeSelectorLabels map[string]string) string{
+	deployment, err := j.KubeClient.AppsV1().Deployments(ns).Create(&appsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Spec: appsv1.DeploymentSpec{
+			Replicas: pointer.Int32Ptr(replicas),
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{
+					"app": name,
+				},
+			},
+			Template: v1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						"app": name,
+					},
+				},
+				Spec: v1.PodSpec{
+					Containers: []v1.Container{
+						{
+							Name:    name,
+							Image:   centos,
+							Command: []string{"/bin/sh"},
+							Args:    []string{"-c", command},
+							VolumeMounts: []v1.VolumeMount{
+								{
+									Name:      "persistent-storage",
+									MountPath: "/data",
+								},
+							},
+						},
+					},
+					Volumes: []v1.Volume{
+						{
+							Name: "persistent-storage",
+							VolumeSource: v1.VolumeSource{
+								PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+									ClaimName: pvcName,
+								},
+							},
+						},
+					},
+					NodeSelector: nodeSelectorLabels,
+				},
+			},
+		},
+	})
+
+	if err != nil{
+		Failf("Error creating deployment %v: %v", name, err)
+	}
+
+	// Waiting for deployment to be completed
+	Logf("Waiting up to %v for deployment %v to be completed", deploymentAvailableTimeout, deployment.Name)
+	err = j.waitTimeoutForDeploymentAvailable(deployment.Name, ns, deploymentAvailableTimeout, replicas)
+	if err != nil {
+		Failf("Deployment %q did not complete: %v", deployment.Name, err)
+	}
+
+	return deployment.Name
+}
+
+// waitTimeoutForDeploymentCompleted waits default amount of time (deploymentCompletionTimeout) for the specified deployment to complete
+//Returns an error if timeout occurs first, or pod goes in to failed state.
+func (j *PVCTestJig) waitTimeoutForDeploymentAvailable(deploymentName string, namespace string, timeout time.Duration, replicas int32) error {
+	return wait.PollImmediate(Poll, timeout, j.deploymentAvailable(deploymentName, namespace, replicas))
+}
+
+func (j *PVCTestJig) deploymentAvailable(deploymentName string, namespace string, replicas int32) wait.ConditionFunc {
+	return func() (bool, error) {
+		deployment, err := j.KubeClient.AppsV1().Deployments(namespace).Get(deploymentName, metav1.GetOptions{})
+		if err != nil {
+			return false, err
+		}
+		if deployment.Status.AvailableReplicas == replicas {
+			return true, nil
+		}
+		return false, nil
+	}
+}

test/e2e/framework/framework.go

@@ -32,6 +32,8 @@ const (
 	// Some pods can take much longer to get ready due to volume attach/detach latency.
 	slowPodStartTimeout = 15 * time.Minute
 
+	deploymentAvailableTimeout = 5 * time.Minute
+
 	DefaultClusterKubeconfig = "/tmp/clusterkubeconfig"
 	DefaultCloudConfig       = "/tmp/cloudconfig"
 

test/e2e/framework/pvc_util.go

@@ -15,7 +15,15 @@
 package framework
 
 import (
+	"bytes"
 	"fmt"
+	"io"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/tools/remotecommand"
+	"os"
+	"path/filepath"
 	"strings"
 	"time"
 
@@ -873,3 +881,143 @@ func (j *PVCTestJig) CheckMultiplePodReadWrite(namespace string, pvcName string,
 	By("Creating Pod that can read contents of existing file")
 	j.NewPodForCSIFSSRead(string(uuid2), namespace, pvcName, fileName)
 }
+
+func (j *PVCTestJig) CheckDataPersistenceWithDeployment(pvcName string, ns string){
+	nodes, err := j.KubeClient.CoreV1().Nodes().List(metav1.ListOptions{})
+
+	if err!= nil{
+		Failf("Error getting list of nodes: %v", err)
+	}
+
+	if len(nodes.Items) == 0{
+		Failf("No worker nodes are present in the cluster")
+	}
+
+	nodeSelectorLabels := map[string]string{}
+	schedulableNodeFound := false
+
+	for _, node := range nodes.Items {
+		if node.Spec.Unschedulable == false {
+			schedulableNodeFound = true
+			nodeSelectorLabels = node.Labels
+			break
+		}
+	}
+
+	if !schedulableNodeFound{
+		Failf("No schedulable nodes found")
+	}
+
+	podRunningCommand := " while true; do true; done;"
+
+	dataWritten := "Data written"
+
+	writeCommand := "echo \"" + dataWritten +"\" >> /data/out.txt;"
+	readCommand := "cat /data/out.txt"
+
+	By("Creating a deployment")
+	deploymentName := j.createDeploymentOnNodeAndWait(podRunningCommand, pvcName, ns, "data-persistence-deployment", 1, nodeSelectorLabels)
+
+	deployment, err := j.KubeClient.AppsV1().Deployments(ns).Get(deploymentName, metav1.GetOptions{})
+
+	if err!= nil {
+		Failf("Error while fetching deployment %v: %v", deploymentName, err)
+	}
+
+	set := labels.Set(deployment.Spec.Selector.MatchLabels)
+	pods, err := j.KubeClient.CoreV1().Pods(ns).List(metav1.ListOptions{LabelSelector: set.AsSelector().String()})
+
+	if err != nil {
+		Failf("Error getting list of pods: %v", err)
+	}
+
+	podName := pods.Items[0].Name
+
+	By("Writing to the volume using the pod")
+	_, _, err = j.ExecCommandOnPod(podName, writeCommand, nil, ns)
+
+	if err!= nil{
+		Failf("Error executing write command a pod: %v", err)
+	}
+
+	By("Deleting the pod used to write to the volume")
+	err = j.KubeClient.CoreV1().Pods(ns).Delete(podName, &metav1.DeleteOptions{})
+
+	if err!= nil{
+		Failf("Error deleting pod: %v", err)
+	}
+
+	By("Waiting for pod to be restarted")
+	err = j.waitTimeoutForDeploymentAvailable(deploymentName, ns, deploymentAvailableTimeout, 1)
+
+	if err!= nil {
+		Failf("Error waiting for deployment to become available again: %v", err)
+	}
+
+	pods, err = j.KubeClient.CoreV1().Pods(ns).List(metav1.ListOptions{LabelSelector: set.AsSelector().String()})
+
+	if err != nil {
+		Failf("Error getting list of pods: %v", err)
+	}
+
+	podName = pods.Items[0].Name
+
+	By("Reading from the volume using the pod and checking data integrity")
+	stdout, _, err := j.ExecCommandOnPod(podName, readCommand, nil, ns)
+
+	if err!= nil{
+		Failf("Error executing write command a pod: %v", err)
+	}
+
+	if dataWritten != strings.TrimSpace(stdout){
+		Failf("Written data not found on the volume, written: %v, found: %v", dataWritten, strings.TrimSpace(stdout))
+	}
+
+}
+
+func (j *PVCTestJig) ExecCommandOnPod(podName string, command string, stdin io.Reader, ns string) (string, string, error) {
+	cmd := []string{
+		"sh",
+		"-c",
+		command,
+	}
+
+	req := j.KubeClient.CoreV1().RESTClient().Post().Resource("pods").Name(podName).Namespace(ns).SubResource("exec")
+	option := &v1.PodExecOptions{
+		Command: cmd,
+		Stdin:   true,
+		Stdout:  true,
+		Stderr:  true,
+		TTY:     true,
+	}
+	if stdin == nil {
+		option.Stdin = false
+	}
+	req.VersionedParams(
+		option,
+		scheme.ParameterCodec,
+	)
+
+	kubeconfig := filepath.Join(os.Getenv("HOME"), ".kube", "config_amd")
+	config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
+
+	if err != nil {
+		return "", "", fmt.Errorf("error while retrieving kubeconfig: %v", err)
+	}
+
+	exec, err := remotecommand.NewSPDYExecutor(config, "POST", req.URL())
+	if err != nil {
+		return "", "", fmt.Errorf("error while creating Executor: %v", err)
+	}
+	var stdout, stderr bytes.Buffer
+	err = exec.Stream(remotecommand.StreamOptions{
+		Stdin:  stdin,
+		Stdout: &stdout,
+		Stderr: &stderr,
+	})
+	if err != nil {
+		return "", "", fmt.Errorf("error in Stream: %v", err)
+	}
+
+	return stdout.String(), stderr.String(), nil
+}