support service account name (#334)

Author: yanghui16355

api/v1beta1/flinkcluster_types.go

@@ -411,6 +411,9 @@ type FlinkClusterSpec struct {
 	// Flink image spec for the cluster's components.
 	Image ImageSpec `json:"image"`
 
+	// The service account assigned to JobManager, TaskManager and Job submitter Pods. If empty, the default service account in the namespace will be used.
+	ServiceAccountName *string `json:"serviceAccountName,omitempty"`
+
 	// BatchSchedulerName specifies the batch scheduler name for JobManager, TaskManager.
 	// If empty, no batch scheduling is enabled.
 	BatchSchedulerName *string `json:"batchSchedulerName,omitempty"`

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

@@ -30,6 +30,8 @@ spec:
           properties:
             batchSchedulerName:
               type: string
+            serviceAccountName:
+              type: string
             envFrom:
               items:
                 properties:

controllers/flinkcluster_converter.go

@@ -87,6 +87,7 @@ func getDesiredJobManagerDeployment(
 	var clusterName = flinkCluster.ObjectMeta.Name
 	var clusterSpec = flinkCluster.Spec
 	var imageSpec = clusterSpec.Image
+	var serviceAccount = clusterSpec.ServiceAccountName
 	var jobManagerSpec = clusterSpec.JobManager
 	var rpcPort = corev1.ContainerPort{Name: "rpc", ContainerPort: *jobManagerSpec.Ports.RPC}
 	var blobPort = corev1.ContainerPort{Name: "blob", ContainerPort: *jobManagerSpec.Ports.Blob}
@@ -196,13 +197,14 @@ func getDesiredJobManagerDeployment(
 	containers = append(containers, jobManagerSpec.Sidecars...)
 
 	var podSpec = corev1.PodSpec{
-		InitContainers:   convertJobManagerInitContainers(&jobManagerSpec),
-		Containers:       containers,
-		Volumes:          volumes,
-		NodeSelector:     jobManagerSpec.NodeSelector,
-		Tolerations:      jobManagerSpec.Tolerations,
-		ImagePullSecrets: imageSpec.PullSecrets,
-		SecurityContext:  securityContext,
+		InitContainers:     convertJobManagerInitContainers(&jobManagerSpec),
+		Containers:         containers,
+		Volumes:            volumes,
+		NodeSelector:       jobManagerSpec.NodeSelector,
+		Tolerations:        jobManagerSpec.Tolerations,
+		ImagePullSecrets:   imageSpec.PullSecrets,
+		SecurityContext:    securityContext,
+		ServiceAccountName: getServiceAccountName(serviceAccount),
 	}
 	var jobManagerDeployment = &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
@@ -381,6 +383,7 @@ func getDesiredTaskManagerDeployment(
 	var clusterName = flinkCluster.ObjectMeta.Name
 	var clusterSpec = flinkCluster.Spec
 	var imageSpec = flinkCluster.Spec.Image
+	var serviceAccount = clusterSpec.ServiceAccountName
 	var taskManagerSpec = flinkCluster.Spec.TaskManager
 	var dataPort = corev1.ContainerPort{Name: "data", ContainerPort: *taskManagerSpec.Ports.Data}
 	var rpcPort = corev1.ContainerPort{Name: "rpc", ContainerPort: *taskManagerSpec.Ports.RPC}
@@ -490,13 +493,14 @@ func getDesiredTaskManagerDeployment(
 	}}
 	containers = append(containers, taskManagerSpec.Sidecars...)
 	var podSpec = corev1.PodSpec{
-		InitContainers:   convertTaskManagerInitContainers(&taskManagerSpec),
-		Containers:       containers,
-		Volumes:          volumes,
-		NodeSelector:     taskManagerSpec.NodeSelector,
-		Tolerations:      taskManagerSpec.Tolerations,
-		ImagePullSecrets: imageSpec.PullSecrets,
-		SecurityContext:  securityContext,
+		InitContainers:     convertTaskManagerInitContainers(&taskManagerSpec),
+		Containers:         containers,
+		Volumes:            volumes,
+		NodeSelector:       taskManagerSpec.NodeSelector,
+		Tolerations:        taskManagerSpec.Tolerations,
+		ImagePullSecrets:   imageSpec.PullSecrets,
+		SecurityContext:    securityContext,
+		ServiceAccountName: getServiceAccountName(serviceAccount),
 	}
 	var taskManagerDeployment = &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
@@ -598,6 +602,7 @@ func getDesiredJob(
 
 	var clusterSpec = flinkCluster.Spec
 	var imageSpec = clusterSpec.Image
+	var serviceAccount = clusterSpec.ServiceAccountName
 	var jobManagerSpec = clusterSpec.JobManager
 	var clusterNamespace = flinkCluster.ObjectMeta.Namespace
 	var clusterName = flinkCluster.ObjectMeta.Name
@@ -704,10 +709,11 @@ func getDesiredJob(
 				Resources:       jobSpec.Resources,
 			},
 		},
-		RestartPolicy:    corev1.RestartPolicyNever,
-		Volumes:          volumes,
-		ImagePullSecrets: imageSpec.PullSecrets,
-		SecurityContext:  securityContext,
+		RestartPolicy:      corev1.RestartPolicyNever,
+		Volumes:            volumes,
+		ImagePullSecrets:   imageSpec.PullSecrets,
+		SecurityContext:    securityContext,
+		ServiceAccountName: getServiceAccountName(serviceAccount),
 	}
 
 	// Disable the retry mechanism of k8s Job, all retires should be initiated
@@ -1035,6 +1041,14 @@ func getClusterLabels(cluster v1beta1.FlinkCluster) map[string]string {
 	}
 }
 
+func getServiceAccountName(serviceAccount *string) string {
+	if serviceAccount != nil {
+		return *serviceAccount
+	}
+
+	return ""
+}
+
 func getComponentLabels(cluster v1beta1.FlinkCluster, component string) map[string]string {
 	return mergeLabels(getClusterLabels(cluster), map[string]string{
 		"component": component,

controllers/flinkcluster_converter_test.go

@@ -50,6 +50,7 @@ func TestGetDesiredClusterState(t *testing.T) {
 	var tolerationSeconds int64 = 30
 	var restartPolicy = v1beta1.JobRestartPolicyFromSavepointOnFailure
 	var className = "org.apache.flink.examples.java.wordcount.WordCount"
+	var serviceAccount = "default"
 	var hostFormat = "{{$clusterName}}.example.com"
 	var memoryOffHeapRatio int32 = 25
 	var memoryOffHeapMin = resource.MustParse("600M")
@@ -131,7 +132,8 @@ func TestGetDesiredClusterState(t *testing.T) {
 			Namespace: "default",
 		},
 		Spec: v1beta1.FlinkClusterSpec{
-			Image: v1beta1.ImageSpec{Name: "flink:1.8.1"},
+			Image:              v1beta1.ImageSpec{Name: "flink:1.8.1"},
+			ServiceAccountName: &serviceAccount,
 			Job: &v1beta1.JobSpec{
 				Args:        []string{"--input", "./README.txt"},
 				ClassName:   &className,
@@ -407,6 +409,7 @@ func TestGetDesiredClusterState(t *testing.T) {
 						RunAsUser:  &userAndGroupId,
 						RunAsGroup: &userAndGroupId,
 					},
+					ServiceAccountName: serviceAccount,
 					Volumes: []corev1.Volume{
 						{
 							Name: "flink-config-volume",
@@ -709,6 +712,7 @@ func TestGetDesiredClusterState(t *testing.T) {
 						RunAsUser:  &userAndGroupId,
 						RunAsGroup: &userAndGroupId,
 					},
+					ServiceAccountName: serviceAccount,
 				},
 			},
 		},
@@ -871,6 +875,7 @@ func TestGetDesiredClusterState(t *testing.T) {
 						RunAsUser:  &userAndGroupId,
 						RunAsGroup: &userAndGroupId,
 					},
+					ServiceAccountName: serviceAccount,
 				},
 			},
 			BackoffLimit: &jobBackoffLimit,

docs/crd.md

@@ -16,6 +16,7 @@ FlinkCluster
         |__ pullPolicy
         |__ pullSecrets
     |__ batchSchedulerName
+    |__ serviceAccountName
     |__ jobManager
         |__ accessScope
         |__ ports
@@ -151,6 +152,8 @@ FlinkCluster
       * **pullSecrets** (optional): Secrets for image pull.
     * **batchSchedulerName** (optional): BatchSchedulerName specifies the batch scheduler name for JobManager, TaskManager.
       If empty, no batch scheduling is enabled.
+    * **serviceAccountName** (optional): the name of the service account(which must already exist in the namespace).
+      If empty, the default service account in the namespace will be used.
     * **jobManager** (required): JobManager spec.
       * **accessScope** (optional): Access scope of the JobManager service. `enum("Cluster", "VPC", "External", 
       "NodePort", "Headless")`. `Cluster`: accessible from within the same cluster; `VPC`: accessible from within the same VPC;