Improve submitting/tracking job and fix (#379)

Terminate the job submitter as soon as it has submitted the job. The controller is responsible for getting the Flink job ID from the job submitter pod termination log and monitoring the job status.

Author: elanv

api/v1beta1/flinkcluster_types.go

@@ -41,15 +41,17 @@ const (
 	ComponentStateDeleted  = "Deleted"
 )
 
-// JobState defines states for a Flink job.
+// JobState defines states for a Flink job deployment.
 const (
 	JobStatePending   = "Pending"
 	JobStateRunning   = "Running"
 	JobStateUpdating  = "Updating"
 	JobStateSucceeded = "Succeeded"
 	JobStateFailed    = "Failed"
 	JobStateCancelled = "Cancelled"
+	JobStateSuspended = "Suspended"
 	JobStateUnknown   = "Unknown"
+	JobStateLost      = "Lost"
 )
 
 // AccessScope defines the access scope of JobManager service.
@@ -537,10 +539,10 @@ type FlinkClusterControlStatus struct {
 // JobStatus defines the status of a job.
 type JobStatus struct {
 	// The name of the Kubernetes job resource.
-	Name string `json:"name"`
+	Name string `json:"name,omitempty"`
 
 	// The ID of the Flink job.
-	ID string `json:"id"`
+	ID string `json:"id,omitempty"`
 
 	// The state of the Kubernetes job.
 	State string `json:"state"`

api/v1beta1/zz_generated.deepcopy.go

@@ -30,8 +30,6 @@ spec:
           properties:
             batchSchedulerName:
               type: string
-            serviceAccountName:
-              type: string
             envFrom:
               items:
                 properties:
@@ -3284,6 +3282,8 @@ spec:
             revisionHistoryLimit:
               format: int32
               type: integer
+            serviceAccountName:
+              type: string
             taskManager:
               properties:
                 extraPorts:
@@ -5163,8 +5163,6 @@ spec:
                     state:
                       type: string
                   required:
-                  - id
-                  - name
                   - state
                   type: object
                 jobManagerDeployment:

config/crd/bases/flinkoperator.k8s.io_flinkclusters.yaml

@@ -171,7 +171,7 @@ func (handler *FlinkClusterHandler) reconcile(
 
 	log.Info("---------- 3. Compute the desired state ----------")
 
-	*desired = getDesiredClusterState(observed.cluster, time.Now())
+	*desired = getDesiredClusterState(observed, time.Now())
 	if desired.ConfigMap != nil {
 		log.Info("Desired state", "ConfigMap", *desired.ConfigMap)
 	} else {
@@ -211,7 +211,8 @@ func (handler *FlinkClusterHandler) reconcile(
 		context:     context,
 		log:         log,
 		observed:    handler.observed,
-		desired:     handler.desired, recorder: handler.recorder,
+		desired:     handler.desired,
+		recorder:    handler.recorder,
 	}
 	result, err := reconciler.reconcile()
 	if err != nil {

controllers/flinkcluster_controller.go

@@ -59,8 +59,10 @@ var flinkSysProps = map[string]struct{}{
 
 // Gets the desired state of a cluster.
 func getDesiredClusterState(
-	cluster *v1beta1.FlinkCluster,
+	observed *ObservedClusterState,
 	now time.Time) model.DesiredClusterState {
+	var cluster = observed.cluster
+
 	// The cluster has been deleted, all resources should be cleaned up.
 	if cluster == nil {
 		return model.DesiredClusterState{}
@@ -71,7 +73,7 @@ func getDesiredClusterState(
 		JmService:    getDesiredJobManagerService(cluster),
 		JmIngress:    getDesiredJobManagerIngress(cluster),
 		TmDeployment: getDesiredTaskManagerDeployment(cluster),
-		Job:          getDesiredJob(cluster),
+		Job:          getDesiredJob(observed),
 	}
 }
 
@@ -584,18 +586,19 @@ func getDesiredConfigMap(
 }
 
 // Gets the desired job spec from a cluster spec.
-func getDesiredJob(
-	flinkCluster *v1beta1.FlinkCluster) *batchv1.Job {
+func getDesiredJob(observed *ObservedClusterState) *batchv1.Job {
+	var flinkCluster = observed.cluster
 	var jobSpec = flinkCluster.Spec.Job
+	var jobStatus = flinkCluster.Status.Components.Job
 
 	if jobSpec == nil {
 		return nil
 	}
 
-	if !isUpdateTriggered(flinkCluster.Status) {
-		// We need to watch whether job is cancelled already if jobSpec.CancelRequested is deprecated
-		var jobStatus = flinkCluster.Status.Components.Job
-		if isJobCancelRequested(*flinkCluster) || (jobStatus != nil && jobStatus.State == v1beta1.JobStateCancelled) {
+	// Unless update has been triggered or the job needs to be restarted, keep the job to be stopped in that state.
+	if !(isUpdateTriggered(flinkCluster.Status) || shouldRestartJob(jobSpec.RestartPolicy, jobStatus)) {
+		// Job cancel requested or stopped already
+		if isJobCancelRequested(*flinkCluster) || isJobStopped(jobStatus) {
 			return nil
 		}
 	}
@@ -619,8 +622,8 @@ func getDesiredJob(
 		jobArgs = append(jobArgs, "--class", *jobSpec.ClassName)
 	}
 
-	var fromSavepoint = convertFromSavepoint(jobSpec, &flinkCluster.Status)
-	if fromSavepoint != nil && *fromSavepoint != "" {
+	var fromSavepoint = convertFromSavepoint(jobSpec, flinkCluster.Status.Components.Job)
+	if fromSavepoint != nil {
 		jobArgs = append(jobArgs, "--fromSavepoint", *fromSavepoint)
 	}
 
@@ -640,7 +643,7 @@ func getDesiredJob(
 
 	var securityContext = jobSpec.SecurityContext
 
-	var envVars = []corev1.EnvVar{}
+	var envVars []corev1.EnvVar
 
 	// If the JAR file is remote, put the URI in the env variable
 	// FLINK_JOB_JAR_URI and rewrite the JAR path to a local path. The entrypoint
@@ -654,6 +657,12 @@ func getDesiredJob(
 			Value: jobSpec.JarFile,
 		})
 	}
+	envVars = append(envVars,
+		corev1.EnvVar{
+			Name:  "FLINK_JM_ADDR",
+			Value: jobManagerAddress,
+		})
+
 	jobArgs = append(jobArgs, jarPath)
 	jobArgs = append(jobArgs, jobSpec.Args...)
 
@@ -716,7 +725,7 @@ func getDesiredJob(
 		ServiceAccountName: getServiceAccountName(serviceAccount),
 	}
 
-	// Disable the retry mechanism of k8s Job, all retires should be initiated
+	// Disable the retry mechanism of k8s Job, all retries should be initiated
 	// by the operator based on the job restart policy. This is because Flink
 	// jobs are stateful, if a job fails after running for 10 hours, we probably
 	// don't want to start over from the beginning, instead we want to resume
@@ -752,28 +761,28 @@ func getDesiredJob(
 // When FlinkCluster is created or updated, if spec.job.fromSavepoint is specified, Flink job will be restored from it.
 //
 // case 2) Restore Flink job from the latest savepoint.
-// When FlinkCluster is updated not specifying spec.job.fromSavepoint, or job is restarted from the failed state,
-// Flink job will be restored from the latest savepoint created by the operator or the savepoint from which current job was restored.
-func convertFromSavepoint(jobSpec *v1beta1.JobSpec, clusterStatus *v1beta1.FlinkClusterStatus) *string {
-	var jobStatus = clusterStatus.Components.Job
+// When FlinkCluster is updated with no spec.job.fromSavepoint, or job is restarted from the failed state,
+// Flink job will be restored from the latest savepoint created by the operator.
+//
+// case 3) When latest created savepoint is unavailable, use the savepoint from which current job was restored.
+func convertFromSavepoint(jobSpec *v1beta1.JobSpec, jobStatus *v1beta1.JobStatus) *string {
 	switch {
-	case shouldRestartJob(jobSpec.RestartPolicy, jobStatus):
-		return &jobStatus.SavepointLocation
-	case isUpdateTriggered(*clusterStatus) && (jobSpec.FromSavepoint == nil || *jobSpec.FromSavepoint == ""):
-		if jobStatus == nil {
-			return nil
-		}
-		// Latest savepoint created by Flink operator
-		if jobStatus.SavepointLocation != "" {
-			return &jobStatus.SavepointLocation
-		}
-		// The savepoint from which current running job was restored
-		if jobStatus.FromSavepoint != "" {
-			return &jobStatus.FromSavepoint
+	// Creating for the first time
+	case jobStatus == nil:
+		if jobSpec.FromSavepoint != nil && *jobSpec.FromSavepoint != "" {
+			return jobSpec.FromSavepoint
 		}
-		return nil
+	// Updating with FromSavepoint provided
+	case jobStatus.State == v1beta1.JobStateUpdating && jobSpec.FromSavepoint != nil && *jobSpec.FromSavepoint != "":
+		return jobSpec.FromSavepoint
+	// Latest savepoint
+	case jobStatus.SavepointLocation != "":
+		return &jobStatus.SavepointLocation
+	// The savepoint from which current job was restored
+	case jobStatus.FromSavepoint != "":
+		return &jobStatus.FromSavepoint
 	}
-	return jobSpec.FromSavepoint
+	return nil
 }
 
 // Copy any non-duplicate volume mounts to the specified initContainers