Add recreateOnUpdate options when update cluster (#365)

related-issue: #360

Author: DeliangFan

api/v1beta1/flinkcluster_default.go

@@ -23,6 +23,10 @@ import (
 
 // Sets default values for unspecified FlinkCluster properties.
 func _SetDefault(cluster *FlinkCluster) {
+	if cluster.Spec.RecreateOnUpdate == nil {
+		cluster.Spec.RecreateOnUpdate = new(bool)
+		*cluster.Spec.RecreateOnUpdate = true
+	}
 	_SetImageDefault(&cluster.Spec.Image)
 	_SetJobManagerDefault(&cluster.Spec.JobManager)
 	_SetTaskManagerDefault(&cluster.Spec.TaskManager)

api/v1beta1/flinkcluster_default_test.go

@@ -55,6 +55,8 @@ func TestSetDefault(t *testing.T) {
 	var defatulJobManagerIngressTLSUse = false
 	var defaultMemoryOffHeapRatio = int32(25)
 	var defaultMemoryOffHeapMin = resource.MustParse("600M")
+	defaultRecreateOnUpdate := new(bool)
+	*defaultRecreateOnUpdate = true
 	var expectedCluster = FlinkCluster{
 		TypeMeta:   metav1.TypeMeta{},
 		ObjectMeta: metav1.ObjectMeta{},
@@ -112,7 +114,8 @@ func TestSetDefault(t *testing.T) {
 			HadoopConfig: &HadoopConfig{
 				MountPath: "/etc/hadoop/conf",
 			},
-			EnvVars: nil,
+			EnvVars:          nil,
+			RecreateOnUpdate: defaultRecreateOnUpdate,
 		},
 		Status: FlinkClusterStatus{},
 	}
@@ -146,6 +149,8 @@ func TestSetNonDefault(t *testing.T) {
 		RunAsUser:  &securityContextUserGroup,
 		RunAsGroup: &securityContextUserGroup,
 	}
+	defaultRecreateOnUpdate := new(bool)
+	*defaultRecreateOnUpdate = true
 	var cluster = FlinkCluster{
 		TypeMeta:   metav1.TypeMeta{},
 		ObjectMeta: metav1.ObjectMeta{},
@@ -267,7 +272,8 @@ func TestSetNonDefault(t *testing.T) {
 			HadoopConfig: &HadoopConfig{
 				MountPath: "/opt/flink/hadoop/conf",
 			},
-			EnvVars: nil,
+			EnvVars:          nil,
+			RecreateOnUpdate: defaultRecreateOnUpdate,
 		},
 		Status: FlinkClusterStatus{},
 	}

api/v1beta1/flinkcluster_types.go

@@ -459,6 +459,9 @@ type FlinkClusterSpec struct {
 
 	// The maximum number of revision history to keep, default: 10.
 	RevisionHistoryLimit *int32 `json:"revisionHistoryLimit,omitempty"`
+
+	// Recreate components when updating flinkcluster, default: true.
+	RecreateOnUpdate *bool `json:"recreateOnUpdate,omitempty"`
 }
 
 // HadoopConfig defines configs for Hadoop.

api/v1beta1/zz_generated.deepcopy.go

@@ -3279,6 +3279,8 @@ spec:
               additionalProperties:
                 type: string
               type: object
+            recreateOnUpdate:
+              type: boolean
             revisionHistoryLimit:
               format: int32
               type: integer

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

@@ -140,7 +140,12 @@ func (reconciler *ClusterReconciler) reconcileStatefulSet(
 	if desiredStatefulSet != nil && observedStatefulSet != nil {
 		if getUpdateState(reconciler.observed) == UpdateStateInProgress {
 			updateComponent := fmt.Sprintf("%v StatefulSet", component)
-			err := reconciler.deleteOldComponent(desiredStatefulSet, observedStatefulSet, updateComponent)
+			var err error
+			if *reconciler.observed.cluster.Spec.RecreateOnUpdate {
+				err = reconciler.deleteOldComponent(desiredStatefulSet, observedStatefulSet, updateComponent)
+			} else {
+				err = reconciler.updateComponent(desiredStatefulSet, updateComponent)
+			}
 			if err != nil {
 				return err
 			}
@@ -192,6 +197,21 @@ func (reconciler *ClusterReconciler) deleteOldComponent(desired runtime.Object,
 	return nil
 }
 
+func (reconciler *ClusterReconciler) updateComponent(desired runtime.Object, component string) error {
+	var log = reconciler.log.WithValues("component", component)
+	var context = reconciler.context
+	var k8sClient = reconciler.k8sClient
+
+	log.Info("Update component", "component", desired)
+	err := k8sClient.Update(context, desired)
+	if err != nil {
+		log.Error(err, "Failed to update component for update")
+		return err
+	}
+	log.Info("Component update successfully")
+	return nil
+}
+
 func (reconciler *ClusterReconciler) updateStatefulSet(
 	statefulSet *appsv1.StatefulSet, component string) error {
 	var context = reconciler.context
@@ -238,7 +258,12 @@ func (reconciler *ClusterReconciler) reconcileJobManagerService() error {
 			// v1.Service API does not handle update correctly when below values are empty.
 			desiredJmService.SetResourceVersion(observedJmService.GetResourceVersion())
 			desiredJmService.Spec.ClusterIP = observedJmService.Spec.ClusterIP
-			err := reconciler.deleteOldComponent(desiredJmService, observedJmService, "JobManager service")
+			var err error
+			if *reconciler.observed.cluster.Spec.RecreateOnUpdate {
+				err = reconciler.deleteOldComponent(desiredJmService, observedJmService, "JobManager service")
+			} else {
+				err = reconciler.updateComponent(desiredJmService, "JobManager service")
+			}
 			if err != nil {
 				return err
 			}
@@ -298,7 +323,12 @@ func (reconciler *ClusterReconciler) reconcileJobManagerIngress() error {
 
 	if desiredJmIngress != nil && observedJmIngress != nil {
 		if getUpdateState(reconciler.observed) == UpdateStateInProgress {
-			err := reconciler.deleteOldComponent(desiredJmIngress, observedJmIngress, "JobManager ingress")
+			var err error
+			if *reconciler.observed.cluster.Spec.RecreateOnUpdate {
+				err = reconciler.deleteOldComponent(desiredJmIngress, observedJmIngress, "JobManager ingress")
+			} else {
+				err = reconciler.updateComponent(desiredJmIngress, "JobManager ingress")
+			}
 			if err != nil {
 				return err
 			}
@@ -358,7 +388,12 @@ func (reconciler *ClusterReconciler) reconcileConfigMap() error {
 
 	if desiredConfigMap != nil && observedConfigMap != nil {
 		if getUpdateState(reconciler.observed) == UpdateStateInProgress {
-			err := reconciler.deleteOldComponent(desiredConfigMap, observedConfigMap, "ConfigMap")
+			var err error
+			if *reconciler.observed.cluster.Spec.RecreateOnUpdate {
+				err = reconciler.deleteOldComponent(desiredConfigMap, observedConfigMap, "ConfigMap")
+			} else {
+				err = reconciler.updateComponent(desiredConfigMap, "ConfigMap")
+			}
 			if err != nil {
 				return err
 			}

controllers/flinkcluster_reconciler.go

@@ -99,6 +99,7 @@ FlinkCluster
             |__ mountPath
     |__ logConfig
     |__ revisionHistoryLimit
+    |__ recreateOnUpdate
 |__ status
     |__ state
     |__ components
@@ -313,6 +314,7 @@ FlinkCluster
         to stdout will be provided.
       * Other arbitrary keys are also allowed, and will become part of the ConfigMap.
     * **revisionHistoryLimit** (optional): The maximum number of revision history to keep, default: 10.
+    * **recreateOnUpdate** (optional): Recreate components when updating flinkcluster, default: true.
   * **status**: Flink job or session cluster status.
     * **state**: The overall state of the Flink cluster.
     * **components**: The status of the components.