[STRMCMP-1639] Integration Test Fix

.github/workflows/actions.yml

@@ -6,7 +6,7 @@ on:
     branches: [ master ]
 jobs:
   unit-tests:
-    runs-on: ubuntu-18.04
+    runs-on: ubuntu-20.04
     defaults:
       run:
         working-directory: go/src/github.com/lyft/flinkk8soperator
@@ -27,7 +27,7 @@ jobs:
       - name: test
         run: make test_unit
   lint:
-    runs-on: ubuntu-18.04
+    runs-on: ubuntu-20.04
     defaults:
       run:
         working-directory: go/src/github.com/lyft/flinkk8soperator
@@ -47,27 +47,26 @@ jobs:
         run: make install
       - name: test
         run: make lint
-  # TODO: restore this test
-  # integration-tests:
-  #   runs-on: ubuntu-18.04
-  #   defaults:
-  #     run:
-  #       working-directory: go/src/github.com/lyft/flinkk8soperator
-  #   env:
-  #     GOPATH: "/home/runner/work/flinkk8soperator/flinkk8soperator/go/"
-  #   steps:
-  #     - name: checkout
-  #       uses: actions/checkout@v2
-  #       with:
-  #         fetch-depth: 1
-  #         path: go/src/github.com/lyft/flinkk8soperator
-  #     - name: install go
-  #       uses: actions/setup-go@v2
-  #       with:
-  #         go-version: 1.12
-  #     - name: install
-  #       run: integ/install.sh
-  #     - name: setup
-  #       run: integ/setup.sh
-  #     - name: test
-  #       run: sudo "PATH=$PATH" "GOPATH=$GOPATH" integ/test.sh
+  integration-tests:
+    runs-on: ubuntu-20.04
+    defaults:
+      run:
+        working-directory: go/src/github.com/lyft/flinkk8soperator
+    env:
+      GOPATH: "/home/runner/work/flinkk8soperator/flinkk8soperator/go/"
+    steps:
+      - name: checkout
+        uses: actions/checkout@v2
+        with:
+          fetch-depth: 1
+          path: go/src/github.com/lyft/flinkk8soperator
+      - name: install go
+        uses: actions/setup-go@v2
+        with:
+          go-version: 1.12
+      - name: install
+        run: integ/install.sh
+      - name: setup
+        run: integ/setup.sh
+      - name: test
+        run: sudo "PATH=$PATH" "GOPATH=$GOPATH" integ/test.sh

docs/local_dev.md

@@ -7,11 +7,11 @@ to develop their applications locally.
 
 ## Run the operator
 
-### Install [Docker for Mac](https://docs.docker.com/docker-for-mac/install/)
+### Install [Minikube](https://minikube.sigs.k8s.io/docs/start/#what-youll-need)
+
+You will want to start minikube on <=1.20, for example:
+`minikube start --kubernetes-version=v1.20.15`
 
-Once installed and running, enabled Kuberenetes in settings (from the
-docker icon in the menu bar, click Preferences -> Kubernetes -> Enable
-Kubernetes).
 
 ### (Optional) Setup kubernetes dashboard
 
@@ -46,6 +46,12 @@ $ cd flinkk8soperator
 $ kubectl create -f deploy/crd.yaml
 ```
 
+### Install permissions
+``` bash
+$ kubectl create -f deploy/role.yaml
+$ kubectl create -f deploy/role-binding.yaml
+```
+
 ### Start the operator
 
 #### Option 1: run outside the kubernetes cluster

integ/README.md

@@ -79,3 +79,40 @@ variables. Supported options include:
 You can also pass [gocheck](http://labix.org/gocheck) options to the
 test runner. Particularly useful is `-check.vv` which will output logs
 from the operator and Flink pods to help debugging test failures.
+
+### Minikube Setup
+
+Ideally we'd use k8s 1.16 to match the deployed k8s version, however, this
+is non-trivial due to cgroup configurations. Instead, we will use a version
+that is compatible with v1beta1 CRD's which corresponds to <1.22. CRD's v1
+is only available with client >=1.16, however, the client used here is 1.14
+and the upgrade is non-trivial.
+
+
+1. Install Dependencies
+   Run dep ensure -vendor-only
+
+3. Start minikube
+   minikube start --kubernetes-version=v1.20.15
+
+4. Proxy minikube
+   kubectl proxy --port 8001 &
+
+5. Set up test app images and operator image
+   integ/setup.sh
+
+8. Set the following for the Go test:
+   Package path: github.com/lyft/flinkk8soperator/integ
+   Env: INTEGRATION=true;OPERATOR_IMAGE=flinkk8soperator:local;RUN_DIRECT=true
+   Program Args: -timeout 40m -check.vv IntegTest
+
+
+Helpers:
+- Kill kube proxy
+  ps -ef | grep "kubectl proxy"
+  kill -9 <process_id>
+- Kill stuck flink app
+  kubectl patch FlinkApplication invalidcanceljob -p '{"metadata":{"finalizers":[]}}' --type=merge
+- Set default namespace
+  kubectl config set-context --current --namespace=flinkoperatortest
+

integ/blue_green_deployment_test.go

@@ -33,6 +33,7 @@ func WaitForUpdate(c *C, s *IntegSuite, name string, updateFn func(app *v1beta1.
 }
 
 func (s *IntegSuite) TestUpdateWithBlueGreenDeploymentMode(c *C) {
+	log.Info("Starting test TestUpdateWithBlueGreenDeploymentMode")
 
 	testName := "bluegreenupdate"
 	const finalizer = "bluegreen.finalizers.test.com"
@@ -55,7 +56,7 @@ func (s *IntegSuite) TestUpdateWithBlueGreenDeploymentMode(c *C) {
 	pods, err := s.Util.KubeClient.CoreV1().Pods(s.Util.Namespace.Name).
 		List(v1.ListOptions{LabelSelector: "integTest=" + testName})
 	c.Assert(err, IsNil)
-	c.Assert(len(pods.Items), Equals, 3)
+	c.Assert(len(pods.Items), Equals, 2)
 	for _, pod := range pods.Items {
 		c.Assert(pod.Spec.Containers[0].Image, Equals, config.Spec.Image)
 	}
@@ -72,7 +73,7 @@ func (s *IntegSuite) TestUpdateWithBlueGreenDeploymentMode(c *C) {
 		List(v1.ListOptions{LabelSelector: "integTest=" + testName})
 	c.Assert(err, IsNil)
 	// We have 2 applications running
-	c.Assert(len(pods.Items), Equals, 6)
+	c.Assert(len(pods.Items), Equals, 4)
 	c.Assert(s.Util.WaitForPhase(config.Name, v1beta1.FlinkApplicationDualRunning, v1beta1.FlinkApplicationDeployFailed), IsNil)
 	c.Assert(s.Util.GetJobID(newApp), NotNil)
 	c.Assert(newApp.Status.UpdatingVersion, Equals, v1beta1.BlueFlinkApplication)
@@ -153,4 +154,5 @@ func (s *IntegSuite) TestUpdateWithBlueGreenDeploymentMode(c *C) {
 		}
 	}
 	log.Info("All pods torn down")
+	log.Info("Completed test TestUpdateWithBlueGreenDeploymentMode")
 }

integ/checkpoint_failure_test.go

@@ -2,8 +2,6 @@ package integ
 
 import (
 	"fmt"
-	"io/ioutil"
-	"os"
 	"time"
 
 	"github.com/lyft/flinkk8soperator/pkg/apis/app/v1beta1"
@@ -67,18 +65,24 @@ func failingJobTest(s *IntegSuite, c *C, testName string, causeFailure func()) {
 
 // Tests that we correctly handle updating a job with task failures
 func (s *IntegSuite) TestJobWithTaskFailures(c *C) {
+	log.Info("Starting test TestJobWithTaskFailures")
+
 	failingJobTest(s, c, "taskfailure", func() {
-		f, err := os.OpenFile(s.Util.CheckpointDir+"/fail", os.O_RDONLY|os.O_CREATE, 0666)
+		err := s.Util.ExecuteCommand("minikube", "ssh", "touch /tmp/checkpoints/fail && chmod 0644 /tmp/checkpoints/fail")
 		c.Assert(err, IsNil)
-		c.Assert(f.Close(), IsNil)
 	})
+	log.Info("Completed test TestJobWithTaskFailures")
 }
 
 // Tests that we correctly handle updating a job with a checkpoint timeout
 func (s *IntegSuite) TestCheckpointTimeout(c *C) {
+	log.Info("Starting test TestCheckpointTimeout")
+
 	failingJobTest(s, c, "checkpointtimeout", func() {
 		// cause checkpoints to take 120 seconds
-		err := ioutil.WriteFile(s.Util.CheckpointDir+"/checkpoint_delay", []byte("120000"), 0644)
+		err := s.Util.ExecuteCommand("minikube", "ssh", "echo 120000 >> /tmp/checkpoints/checkpoint_delay && sudo chmod 0644 /tmp/checkpoints/checkpoint_delay")
 		c.Assert(err, IsNil)
 	})
+	log.Info("Completed test TestCheckpointTimeout")
+
 }

integ/install.sh

@@ -2,10 +2,15 @@
 
 set -e
 
-sudo snap install microk8s --classic --channel=1.13/stable
-microk8s.status --wait-ready
-microk8s.enable dns
-microk8s.enable registry
+curl -LO -s https://storage.googleapis.com/minikube/releases/latest/minikube-linux-amd64
+sudo install minikube-linux-amd64 /usr/local/bin/minikube
+
+minikube config set memory 6800
+minikube start --kubernetes-version=v1.20.15
+
+ls /etc/kubernetes/
+cat /etc/kubernetes/admin.conf
+cp /etc/kubernetes/admin.conf $HOME/.kube/config
 
 sh boilerplate/lyft/golang_test_targets/dep_install.sh
 

integ/job_cancellation_test.go

@@ -58,7 +58,7 @@ func WaitUpdateAndValidate(c *C, s *IntegSuite, name string, updateFn func(app *
 
 // tests the workflow of job cancellation without savepoint
 func (s *IntegSuite) TestJobCancellationWithoutSavepoint(c *C) {
-
+	log.Info("Starting test TestJobCancellationWithoutSavepoint")
 	testName := "cancelsuccess"
 	const finalizer = "simple.finalizers.test.com"
 
@@ -81,7 +81,7 @@ func (s *IntegSuite) TestJobCancellationWithoutSavepoint(c *C) {
 	pods, err := s.Util.KubeClient.CoreV1().Pods(s.Util.Namespace.Name).
 		List(v1.ListOptions{LabelSelector: "integTest=" + testName})
 	c.Assert(err, IsNil)
-	c.Assert(len(pods.Items), Equals, 3)
+	c.Assert(len(pods.Items), Equals, 2)
 	for _, pod := range pods.Items {
 		c.Assert(pod.Spec.Containers[0].Image, Equals, config.Spec.Image)
 	}
@@ -97,7 +97,7 @@ func (s *IntegSuite) TestJobCancellationWithoutSavepoint(c *C) {
 	pods, err = s.Util.KubeClient.CoreV1().Pods(s.Util.Namespace.Name).
 		List(v1.ListOptions{LabelSelector: "integTest=" + testName})
 	c.Assert(err, IsNil)
-	c.Assert(len(pods.Items), Equals, 3)
+	c.Assert(len(pods.Items), Equals, 2)
 	for _, pod := range pods.Items {
 		c.Assert(pod.Spec.Containers[0].Image, Equals, NewImage)
 	}
@@ -131,11 +131,13 @@ func (s *IntegSuite) TestJobCancellationWithoutSavepoint(c *C) {
 		}
 	}
 	log.Info("All pods torn down")
+	log.Info("Completed test TestJobCancellationWithoutSavepoint")
 }
 
 // tests a job update with the existing job already in cancelled state.
 // here, the new submitted job starts without a savepoint.
 func (s *IntegSuite) TestCancelledJobWithoutSavepoint(c *C) {
+	log.Info("Starting test TestCancelledJobWithoutSavepoint")
 
 	testName := "invalidcancel"
 	config, err := s.Util.ReadFlinkApplication("test_app.yaml")
@@ -150,6 +152,7 @@ func (s *IntegSuite) TestCancelledJobWithoutSavepoint(c *C) {
 		Commentf("Failed to create flink application"))
 
 	c.Assert(s.Util.WaitForPhase(config.Name, v1beta1.FlinkApplicationRunning, v1beta1.FlinkApplicationDeployFailed), IsNil)
+
 	c.Assert(s.Util.WaitForAllTasksRunning(config.Name), IsNil)
 
 	currApp, _ := s.Util.GetFlinkApplication(config.Name)
@@ -163,7 +166,7 @@ func (s *IntegSuite) TestCancelledJobWithoutSavepoint(c *C) {
 	c.Assert(err, IsNil)
 
 	// wait a bit
-	time.Sleep(1 * time.Second)
+	time.Sleep(5 * time.Second)
 
 	job = s.Util.GetJobOverview(currApp)
 	c.Assert(job["status"], Equals, "CANCELED")
@@ -205,10 +208,12 @@ func (s *IntegSuite) TestCancelledJobWithoutSavepoint(c *C) {
 		}
 	}
 	log.Info("All pods torn down")
+	log.Info("Completed test TestCancelledJobWithoutSavepoint")
 }
 
 // tests the recovery workflow of the job when savepoint is disabled.
 func (s *IntegSuite) TestJobRecoveryWithoutSavepoint(c *C) {
+	log.Info("Starting test TestJobRecoveryWithoutSavepoint")
 
 	const finalizer = "simple.finalizers.test.com"
 	const testName = "cancelrecovery"
@@ -300,4 +305,5 @@ func (s *IntegSuite) TestJobRecoveryWithoutSavepoint(c *C) {
 		time.Sleep(100 * time.Millisecond)
 	}
 	log.Info("All pods torn down")
+	log.Info("Completed test TestJobRecoveryWithoutSavepoint")
 }

integ/main_test.go

@@ -79,7 +79,7 @@ func (s *IntegSuite) SetUpSuite(c *C) {
 			LimitNamespace: namespace,
 			UseProxy:       true,
 			ResyncPeriod:   flyteConfig.Duration{Duration: 3 * time.Second},
-			MaxErrDuration: flyteConfig.Duration{Duration: 30 * time.Second},
+			MaxErrDuration: flyteConfig.Duration{Duration: 60 * time.Second},
 			MetricsPrefix:  "flinkk8soperator",
 			ProxyPort:      flyteConfig.Port{Port: 8001},
 		}
@@ -92,6 +92,18 @@ func (s *IntegSuite) SetUpSuite(c *C) {
 			}
 		}()
 	} else {
+		if err = s.Util.CreateClusterRole(); err != nil && !k8sErrors.IsAlreadyExists(err) {
+			c.Fatalf("Failed to create role: %v", err)
+		}
+
+		if err = s.Util.CreateServiceAccount(); err != nil && !k8sErrors.IsAlreadyExists(err) {
+			c.Fatalf("Failed to create service account: %v", err)
+		}
+
+		if err = s.Util.CreateClusterRoleBinding(); err != nil && !k8sErrors.IsAlreadyExists(err) {
+			c.Fatalf("Failed to create cluster role binding: %v", err)
+		}
+
 		if err = s.Util.CreateOperator(); err != nil {
 			c.Fatalf("Failed to create operator: %v", err)
 		}
@@ -111,18 +123,12 @@ func (s *IntegSuite) TearDownSuite(c *C) {
 
 func (s *IntegSuite) SetUpTest(c *C) {
 	// create checkpoint directory
-	if _, err := os.Stat(s.Util.CheckpointDir); os.IsNotExist(err) {
-		c.Assert(os.Mkdir(s.Util.CheckpointDir, 0777), IsNil)
+	if err := s.Util.ExecuteCommand("minikube", "ssh", "sudo mkdir /tmp/checkpoints && sudo chmod -R 0777 /tmp/checkpoints"); err != nil {
+		c.Fatalf("Failed to create checkpoint directory: %v", err)
 	}
 }
 
 func (s *IntegSuite) TearDownTest(c *C) {
-	jm, err := s.Util.GetJobManagerPod()
-	if err == nil {
-		fmt.Printf("\n\n######### JobManager logs for debugging #########\n---------------------------\n")
-		_ = s.Util.GetLogs(jm, nil)
-	}
-
 	tms, err := s.Util.GetTaskManagerPods()
 	if err == nil {
 		for i, tm := range tms {
@@ -132,13 +138,34 @@ func (s *IntegSuite) TearDownTest(c *C) {
 		}
 	}
 
+	jm, err := s.Util.GetJobManagerPod()
+	if err == nil {
+		fmt.Printf("\n\n######### JobManager logs for debugging #########\n---------------------------\n")
+		_ = s.Util.GetLogs(jm, nil)
+	}
+
+	fmt.Printf("\n\n######### Nodes for debugging #########\n---------------------------\n")
+	err = s.Util.ExecuteCommand("kubectl", "describe", "nodes")
+	c.Assert(err, IsNil)
+
+	fmt.Printf("\n\n######### Pods for debugging #########\n---------------------------\n")
+	err = s.Util.ExecuteCommand("kubectl", "get", "pods", "-n", "flinkoperatortest")
+	c.Assert(err, IsNil)
+
+	fmt.Printf("\n\n######### Pod details for debugging #########\n---------------------------\n")
+	err = s.Util.ExecuteCommand("kubectl", "describe", "pods", "-n", "flinkoperatortest")
+	c.Assert(err, IsNil)
+
+	fmt.Printf("\n\n######### Flink Applications for debugging #########\n---------------------------\n")
+	err = s.Util.ExecuteCommand("kubectl", "describe", "flinkapplications", "-n", "flinkoperatortest")
+	c.Assert(err, IsNil)
+
 	err = s.Util.FlinkApps().DeleteCollection(nil, v1.ListOptions{})
 	if err != nil {
-		log.Fatalf("Failed to clean up flink applications")
+		log.Fatalf("Failed to clean up flink applications: %v", err)
 	}
 
-	err = os.RemoveAll(s.Util.CheckpointDir)
-	if err != nil {
-		log.Fatalf("Failed to clean up checkpoints directory: %v", err)
+	if err := s.Util.ExecuteCommand("minikube", "ssh", "sudo rm -rf /tmp/checkpoints"); err != nil {
+		c.Fatalf("Failed to delete checkpoint directory: %v", err)
 	}
 }