Use StatefulSet instead of Deployment (GoogleCloudPlatform#354)

Author: shashken

api/v1alpha1/flinkcluster_types.go

@@ -372,7 +372,7 @@ type FlinkClusterComponentsStatus struct {
 	// The state of configMap.
 	ConfigMap FlinkClusterComponentState `json:"configMap"`
 
-	// The state of JobManager deployment.
+	// The state of JobManager Deployment.
 	JobManagerDeployment FlinkClusterComponentState `json:"jobManagerDeployment"`
 
 	// The state of JobManager service.
@@ -381,7 +381,7 @@ type FlinkClusterComponentsStatus struct {
 	// The state of JobManager ingress.
 	JobManagerIngress *JobManagerIngressStatus `json:"jobManagerIngress,omitempty"`
 
-	// The state of TaskManager deployment.
+	// The state of TaskManager Deployment.
 	TaskManagerDeployment FlinkClusterComponentState `json:"taskManagerDeployment"`
 
 	// The status of the job, available only when JobSpec is provided.

api/v1beta1/flinkcluster_types.go

@@ -205,6 +205,8 @@ type JobManagerSpec struct {
 	// Volume mounts in the JobManager container.
 	VolumeMounts []corev1.VolumeMount `json:"volumeMounts,omitempty"`
 
+	VolumeClaimTemplates []corev1.PersistentVolumeClaim `json:"volumeClaimTemplates,omitempty"`
+
 	// Init containers of the Job Manager pod.
 	InitContainers []corev1.Container `json:"initContainers,omitempty"`
 
@@ -221,13 +223,13 @@ type JobManagerSpec struct {
 	// pod.
 	Sidecars []corev1.Container `json:"sidecars,omitempty"`
 
-	// JobManager Deployment pod template annotations.
+	// JobManager StatefulSet pod template annotations.
 	PodAnnotations map[string]string `json:"podAnnotations,omitempty"`
 
 	// SecurityContext of the JM pod.
 	SecurityContext *corev1.PodSecurityContext `json:"securityContext,omitempty"`
 
-	// JobManager Deployment pod template labels.
+	// JobManager StatefulSet pod template labels.
 	PodLabels map[string]string `json:"podLabels,omitempty"`
 }
 
@@ -278,6 +280,8 @@ type TaskManagerSpec struct {
 	// More info: https://kubernetes.io/docs/concepts/storage/volumes/
 	VolumeMounts []corev1.VolumeMount `json:"volumeMounts,omitempty"`
 
+	VolumeClaimTemplates []corev1.PersistentVolumeClaim `json:"volumeClaimTemplates,omitempty"`
+
 	// Init containers of the Task Manager pod.
 	InitContainers []corev1.Container `json:"initContainers,omitempty"`
 
@@ -294,13 +298,13 @@ type TaskManagerSpec struct {
 	// pod.
 	Sidecars []corev1.Container `json:"sidecars,omitempty"`
 
-	// TaskManager Deployment pod template annotations.
+	// TaskManager StatefulSet pod template annotations.
 	PodAnnotations map[string]string `json:"podAnnotations,omitempty"`
 
 	// SecurityContext of the TM pod.
 	SecurityContext *corev1.PodSecurityContext `json:"securityContext,omitempty"`
 
-	// TaskManager Deployment pod template labels.
+	// TaskManager StatefulSet pod template labels.
 	PodLabels map[string]string `json:"podLabels,omitempty"`
 }
 
@@ -502,17 +506,17 @@ type FlinkClusterComponentsStatus struct {
 	// The state of configMap.
 	ConfigMap FlinkClusterComponentState `json:"configMap"`
 
-	// The state of JobManager deployment.
-	JobManagerDeployment FlinkClusterComponentState `json:"jobManagerDeployment"`
+	// The state of JobManager StatefulSet.
+	JobManagerStatefulSet FlinkClusterComponentState `json:"jobManagerStatefulSet"`
 
 	// The state of JobManager service.
 	JobManagerService JobManagerServiceStatus `json:"jobManagerService"`
 
 	// The state of JobManager ingress.
 	JobManagerIngress *JobManagerIngressStatus `json:"jobManagerIngress,omitempty"`
 
-	// The state of TaskManager deployment.
-	TaskManagerDeployment FlinkClusterComponentState `json:"taskManagerDeployment"`
+	// The state of TaskManager StatefulSet.
+	TaskManagerStatefulSet FlinkClusterComponentState `json:"taskManagerStatefulSet"`
 
 	// The status of the job, available only when JobSpec is provided.
 	Job *JobStatus `json:"job,omitempty"`

api/v1beta1/zz_generated.deepcopy.go

@@ -5165,7 +5165,7 @@ spec:
                   required:
                   - state
                   type: object
-                jobManagerDeployment:
+                jobManagerStatefulSet:
                   properties:
                     name:
                       type: string
@@ -5202,7 +5202,7 @@ spec:
                   - name
                   - state
                   type: object
-                taskManagerDeployment:
+                taskManagerStatefulSet:
                   properties:
                     name:
                       type: string
@@ -5214,9 +5214,9 @@ spec:
                   type: object
               required:
               - configMap
-              - jobManagerDeployment
+              - jobManagerStatefulSet
               - jobManagerService
-              - taskManagerDeployment
+              - taskManagerStatefulSet
               type: object
             control:
               properties:

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

@@ -30,6 +30,7 @@ rules:
   - apps
   resources:
   - deployments
+  - statefulsets
   verbs:
   - get
   - list
@@ -42,6 +43,7 @@ rules:
   - apps
   resources:
   - deployments/status
+  - statefulsets/status
   verbs:
   - get
 - apiGroups:

config/rbac/role.yaml

@@ -74,19 +74,19 @@ func (v *VolcanoBatchScheduler) Schedule(cluster *v1beta1.FlinkCluster, state *m
 
 func (v *VolcanoBatchScheduler) setSchedulerMeta(cluster *v1beta1.FlinkCluster, state *model.DesiredClusterState) {
 	podgroupName := v.getPodGroupName(cluster)
-	if state.TmDeployment != nil {
-		state.TmDeployment.Spec.Template.Spec.SchedulerName = v.Name()
-		if state.TmDeployment.Spec.Template.Annotations == nil {
-			state.TmDeployment.Spec.Template.Annotations = make(map[string]string)
+	if state.TmStatefulSet != nil {
+		state.TmStatefulSet.Spec.Template.Spec.SchedulerName = v.Name()
+		if state.TmStatefulSet.Spec.Template.Annotations == nil {
+			state.TmStatefulSet.Spec.Template.Annotations = make(map[string]string)
 		}
-		state.TmDeployment.Spec.Template.Annotations[scheduling.KubeGroupNameAnnotationKey] = podgroupName
+		state.TmStatefulSet.Spec.Template.Annotations[scheduling.KubeGroupNameAnnotationKey] = podgroupName
 	}
-	if state.JmDeployment != nil {
-		state.JmDeployment.Spec.Template.Spec.SchedulerName = v.Name()
-		if state.JmDeployment.Spec.Template.Annotations == nil {
-			state.JmDeployment.Spec.Template.Annotations = make(map[string]string)
+	if state.JmStatefulSet != nil {
+		state.JmStatefulSet.Spec.Template.Spec.SchedulerName = v.Name()
+		if state.JmStatefulSet.Spec.Template.Annotations == nil {
+			state.JmStatefulSet.Spec.Template.Annotations = make(map[string]string)
 		}
-		state.JmDeployment.Spec.Template.Annotations[scheduling.KubeGroupNameAnnotationKey] = podgroupName
+		state.JmStatefulSet.Spec.Template.Annotations[scheduling.KubeGroupNameAnnotationKey] = podgroupName
 	}
 	if state.Job != nil {
 		state.Job.Spec.Template.Spec.SchedulerName = v.Name()
@@ -145,18 +145,18 @@ func getClusterResource(state *model.DesiredClusterState) (corev1.ResourceList,
 	resource := corev1.ResourceList{}
 	var size int32
 
-	if state.JmDeployment != nil {
-		size += *state.JmDeployment.Spec.Replicas
-		for i := int32(0); i < *state.JmDeployment.Spec.Replicas; i++ {
-			jmResource := getPodResource(&state.JmDeployment.Spec.Template.Spec)
+	if state.JmStatefulSet != nil {
+		size += *state.JmStatefulSet.Spec.Replicas
+		for i := int32(0); i < *state.JmStatefulSet.Spec.Replicas; i++ {
+			jmResource := getPodResource(&state.JmStatefulSet.Spec.Template.Spec)
 			addResourceList(resource, jmResource, nil)
 		}
 	}
 
-	if state.TmDeployment != nil {
-		size += *state.TmDeployment.Spec.Replicas
-		for i := int32(0); i < *state.TmDeployment.Spec.Replicas; i++ {
-			tmResource := getPodResource(&state.TmDeployment.Spec.Template.Spec)
+	if state.TmStatefulSet != nil {
+		size += *state.TmStatefulSet.Spec.Replicas
+		for i := int32(0); i < *state.TmStatefulSet.Spec.Replicas; i++ {
+			tmResource := getPodResource(&state.TmStatefulSet.Spec.Template.Spec)
 			addResourceList(resource, tmResource, nil)
 		}
 	}

controllers/batchscheduler/volcano/volcano.go

@@ -32,7 +32,7 @@ func TestGetClusterResource(t *testing.T) {
 	jmRep := int32(1)
 	replicas := int32(4)
 	desiredState := &model.DesiredClusterState{
-		JmDeployment: &appsv1.Deployment{
+		JmStatefulSet: &appsv1.StatefulSet{
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      "flinkjobcluster-sample-jobmanager",
 				Namespace: "default",
@@ -42,8 +42,9 @@ func TestGetClusterResource(t *testing.T) {
 					"component": "jobmanager",
 				},
 			},
-			Spec: appsv1.DeploymentSpec{
+			Spec: appsv1.StatefulSetSpec{
 				Replicas: &jmRep,
+				ServiceName: "flinkjobcluster-sample-jobmanager",
 				Selector: &metav1.LabelSelector{
 					MatchLabels: map[string]string{
 						"app":       "flink",
@@ -161,7 +162,7 @@ func TestGetClusterResource(t *testing.T) {
 				},
 			},
 		},
-		TmDeployment: &appsv1.Deployment{
+		TmStatefulSet: &appsv1.StatefulSet{
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      "flinkjobcluster-sample-taskmanager",
 				Namespace: "default",
@@ -171,8 +172,10 @@ func TestGetClusterResource(t *testing.T) {
 					"component": "taskmanager",
 				},
 			},
-			Spec: appsv1.DeploymentSpec{
+			Spec: appsv1.StatefulSetSpec{
 				Replicas: &replicas,
+				ServiceName: "flinkjobcluster-sample-taskmanager",
+				PodManagementPolicy: "Parallel",
 				Selector: &metav1.LabelSelector{
 					MatchLabels: map[string]string{
 						"app":       "flink",

controllers/batchscheduler/volcano/volcano_test.go

@@ -48,6 +48,8 @@ type FlinkClusterReconciler struct {
 // +kubebuilder:rbac:groups=flinkoperator.k8s.io,resources=flinkclusters/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=apps,resources=deployments/status,verbs=get
+// +kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups=apps,resources=statefulsets/status,verbs=get
 // +kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch
 // +kubebuilder:rbac:groups=core,resources=pods/status,verbs=get
 // +kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;create;update;patch;delete
@@ -88,6 +90,7 @@ func (reconciler *FlinkClusterReconciler) SetupWithManager(
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&v1beta1.FlinkCluster{}).
 		Owns(&appsv1.Deployment{}).
+		Owns(&appsv1.StatefulSet{}).
 		Owns(&corev1.Service{}).
 		Owns(&batchv1.Job{}).
 		Complete(reconciler)
@@ -177,10 +180,10 @@ func (handler *FlinkClusterHandler) reconcile(
 	} else {
 		log.Info("Desired state", "ConfigMap", "nil")
 	}
-	if desired.JmDeployment != nil {
-		log.Info("Desired state", "JobManager deployment", *desired.JmDeployment)
+	if desired.JmStatefulSet != nil {
+		log.Info("Desired state", "JobManager StatefulSet", *desired.JmStatefulSet)
 	} else {
-		log.Info("Desired state", "JobManager deployment", "nil")
+		log.Info("Desired state", "JobManager StatefulSet", "nil")
 	}
 	if desired.JmService != nil {
 		log.Info("Desired state", "JobManager service", *desired.JmService)
@@ -192,10 +195,10 @@ func (handler *FlinkClusterHandler) reconcile(
 	} else {
 		log.Info("Desired state", "JobManager ingress", "nil")
 	}
-	if desired.TmDeployment != nil {
-		log.Info("Desired state", "TaskManager deployment", *desired.TmDeployment)
+	if desired.TmStatefulSet != nil {
+		log.Info("Desired state", "TaskManager StatefulSet", *desired.TmStatefulSet)
 	} else {
-		log.Info("Desired state", "TaskManager deployment", "nil")
+		log.Info("Desired state", "TaskManager StatefulSet", "nil")
 	}
 	if desired.Job != nil {
 		log.Info("Desired state", "Job", *desired.Job)