Add a ForceRollback option to CRD

docs/crd.md

@@ -105,3 +105,6 @@ Below is the list of fields in the custom resource and their description
 
   * **VolumeMounts** `type:[]v1.VolumeMount`
     Describes a mounting of a Volume within a container.
+
+  * **CancelDeploy** `type:bool`
+    Can be set to true to force cancel a deploy/update. Resetting to false or un-setting implies the deploy will continue.

integ/simple_test.go

@@ -4,6 +4,8 @@ import (
 	"encoding/json"
 	"fmt"
 
+	"k8s.io/apimachinery/pkg/api/resource"
+
 	"os"
 	"time"
 
@@ -170,6 +172,60 @@ func (s *IntegSuite) TestSimple(c *C) {
 		}, "")
 	}
 
+	// Test cancelling a deploy
+
+	{
+		newApp, err := s.Util.GetFlinkApplication(config.Name)
+		c.Assert(err, IsNil)
+		// User sets large (bad) value for cluster update
+		var TaskManagerDefaultResources = corev1.ResourceRequirements{
+			Requests: corev1.ResourceList{
+				corev1.ResourceCPU:    resource.MustParse("2"),
+				corev1.ResourceMemory: resource.MustParse("5Gi"),
+			},
+			Limits: corev1.ResourceList{
+				corev1.ResourceCPU:    resource.MustParse("2"),
+				corev1.ResourceMemory: resource.MustParse("5Gi"),
+			},
+		}
+		newApp.Spec.TaskManagerConfig.Resources = &TaskManagerDefaultResources
+
+		_, _ = s.Util.FlinkApps().Update(newApp)
+		c.Assert(s.Util.WaitForPhase(newApp.Name, v1alpha1.FlinkApplicationClusterStarting, ""), IsNil)
+
+		// User realizes error and  cancels the deploy
+		log.Infof("Cancelling deploy...")
+		newApp.Spec.CancelDeploy = true
+		_, _ = s.Util.FlinkApps().Update(newApp)
+
+		// we should end up in the DeployFailed phase
+		c.Assert(s.Util.WaitForPhase(newApp.Name, v1alpha1.FlinkApplicationDeployFailed, ""), IsNil)
+		c.Assert(newApp.Spec.CancelDeploy, Equals, true)
+		log.Info("User cancelled deploy. Job is in deploy failed, waiting for tasks to start")
+
+		// but the job should still be running
+		c.Assert(newApp.Status.JobStatus.State, Equals, v1alpha1.Running)
+
+		log.Info("Attempting to roll forward with fix")
+
+		// Fixed resources
+		var TaskManagerFixedResources = corev1.ResourceRequirements{
+			Requests: corev1.ResourceList{
+				corev1.ResourceCPU:    resource.MustParse("0.2"),
+				corev1.ResourceMemory: resource.MustParse("200Mi"),
+			},
+			Limits: corev1.ResourceList{
+				corev1.ResourceCPU:    resource.MustParse("0.2"),
+				corev1.ResourceMemory: resource.MustParse("200Mi"),
+			},
+		}
+		// and we should be able to roll forward by resubmitting with a fixed config
+		updateAndValidate(c, s, config.Name, func(app *v1alpha1.FlinkApplication) {
+			app.Spec.TaskManagerConfig.Resources = &TaskManagerFixedResources
+			app.Spec.CancelDeploy = false
+		}, "")
+	}
+
 	// delete the application and ensure everything is cleaned up successfully
 	c.Assert(s.Util.FlinkApps().Delete(config.Name, &v1.DeleteOptions{}), IsNil)
 

integ/test.sh

@@ -9,5 +9,5 @@ export OPERATOR_IMAGE=127.0.0.1:32000/flinkk8soperator:local
 umask 000
 
 cd $(dirname "$0")
-go test -timeout 20m -check.vv IntegSuite
+go test -timeout 22m -check.vv IntegSuite
 

pkg/apis/app/v1alpha1/types.go

@@ -53,6 +53,7 @@ type FlinkApplicationSpec struct {
 	RestartNonce          string                       `json:"restartNonce"`
 	DeleteMode            DeleteMode                   `json:"deleteMode,omitempty"`
 	AllowNonRestoredState bool                         `json:"allowNonRestoredState,omitempty"`
+	CancelDeploy          bool                         `json:"cancelDeploy"`
 }
 
 type FlinkConfig map[string]interface{}

pkg/controller/flinkapplication/flink_state_machine.go

@@ -80,6 +80,9 @@ func (s *FlinkStateMachine) updateApplicationPhase(ctx context.Context, applicat
 }
 
 func (s *FlinkStateMachine) shouldRollback(ctx context.Context, application *v1alpha1.FlinkApplication) bool {
+	if application.Spec.CancelDeploy {
+		return true
+	}
 	if application.Status.DeployHash == "" {
 		// TODO: we may want some more sophisticated way of handling this case
 		// there's no previous deploy for this application, so nothing to roll back to

pkg/controller/flinkapplication/flink_state_machine_test.go

@@ -1130,3 +1130,30 @@ func TestErrorHandlingInRunningPhase(t *testing.T) {
 	assert.Nil(t, app.Status.LastSeenError)
 
 }
+
+func TestCancelDeploy(t *testing.T) {
+	app := v1alpha1.FlinkApplication{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-app",
+			Namespace: "flink",
+		},
+		Spec: v1alpha1.FlinkApplicationSpec{
+			JarName:      "job.jar",
+			Parallelism:  5,
+			EntryClass:   "com.my.Class",
+			ProgramArgs:  "--test",
+			CancelDeploy: true,
+		},
+		Status: v1alpha1.FlinkApplicationStatus{
+			Phase:      v1alpha1.FlinkApplicationClusterStarting,
+			DeployHash: "old-hash-retry-err",
+		},
+	}
+
+	stateMachineForTest := getTestStateMachine()
+	err := stateMachineForTest.Handle(context.Background(), &app)
+	assert.Nil(t, err)
+	// cancelled deploy while cluster is starting
+	assert.Equal(t, v1alpha1.FlinkApplicationDeployFailed, app.Status.Phase)
+	assert.True(t, app.Spec.CancelDeploy)
+}