Add podLabels to JM, TM, Job (#321)

Author: shashken

api/v1beta1/flinkcluster_types.go

@@ -221,6 +221,9 @@ type JobManagerSpec struct {
 
 	// JobManager Deployment pod template annotations.
 	PodAnnotations map[string]string `json:"podAnnotations,omitempty"`
+
+	// JobManager Deployment pod template labels.
+	PodLabels map[string]string `json:"podLabels,omitempty"`
 }
 
 // TaskManagerPorts defines ports of TaskManager.
@@ -288,6 +291,9 @@ type TaskManagerSpec struct {
 
 	// TaskManager Deployment pod template annotations.
 	PodAnnotations map[string]string `json:"podAnnotations,omitempty"`
+
+	// TaskManager Deployment pod template labels.
+	PodLabels map[string]string `json:"podLabels,omitempty"`
 }
 
 // CleanupAction defines the action to take after job finishes.
@@ -382,6 +388,9 @@ type JobSpec struct {
 	// Job pod template annotations.
 	PodAnnotations map[string]string `json:"podAnnotations,omitempty"`
 
+	// Job pod template labels.
+	PodLabels map[string]string `json:"podLabels,omitempty"`
+
 	// Compute resources required by each Job container.
 	// If omitted, a default value will be used.
 	// Cannot be updated.

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

@@ -723,6 +723,10 @@ spec:
                   additionalProperties:
                     type: string
                   type: object
+                podLabels:
+                  additionalProperties:
+                    type: string
+                  type: object
                 resources:
                   properties:
                     limits:
@@ -1956,6 +1960,10 @@ spec:
                   additionalProperties:
                     type: string
                   type: object
+                podLabels:
+                  additionalProperties:
+                    type: string
+                  type: object
                 ports:
                   properties:
                     blob:
@@ -3743,6 +3751,10 @@ spec:
                   additionalProperties:
                     type: string
                   type: object
+                podLabels:
+                  additionalProperties:
+                    type: string
+                  type: object
                 ports:
                   properties:
                     data:

controllers/flinkcluster_converter.go

@@ -98,6 +98,7 @@ func getDesiredJobManagerDeployment(
 	}
 	var jobManagerDeploymentName = getJobManagerDeploymentName(clusterName)
 	var podLabels = getComponentLabels(*flinkCluster, "jobmanager")
+	podLabels = mergeLabels(podLabels, jobManagerSpec.PodLabels)
 	var deploymentLabels = mergeLabels(podLabels, getRevisionHashLabels(flinkCluster.Status))
 	// Make Volume, VolumeMount to use configMap data for flink-conf.yaml, if flinkProperties is provided.
 	var volumes []corev1.Volume
@@ -253,6 +254,7 @@ func getDesiredJobManagerService(
 		TargetPort: intstr.FromString("ui")}
 	var jobManagerServiceName = getJobManagerServiceName(clusterName)
 	var podLabels = getComponentLabels(*flinkCluster, "jobmanager")
+	podLabels = mergeLabels(podLabels, jobManagerSpec.PodLabels)
 	var serviceLabels = mergeLabels(podLabels, getRevisionHashLabels(flinkCluster.Status))
 	var jobManagerService = &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{
@@ -388,6 +390,7 @@ func getDesiredTaskManagerDeployment(
 	}
 	var taskManagerDeploymentName = getTaskManagerDeploymentName(clusterName)
 	var podLabels = getComponentLabels(*flinkCluster, "taskmanager")
+	podLabels = mergeLabels(podLabels, taskManagerSpec.PodLabels)
 	var deploymentLabels = mergeLabels(podLabels, getRevisionHashLabels(flinkCluster.Status))
 	// Make Volume, VolumeMount to use configMap data for flink-conf.yaml
 	var volumes []corev1.Volume
@@ -598,6 +601,7 @@ func getDesiredJob(
 	var jobManagerAddress = fmt.Sprintf(
 		"%s:%d", jobManagerServiceName, *jobManagerSpec.Ports.UI)
 	var podLabels = getClusterLabels(*flinkCluster)
+	podLabels = mergeLabels(podLabels, jobManagerSpec.PodLabels)
 	var jobLabels = mergeLabels(podLabels, getRevisionHashLabels(flinkCluster.Status))
 	var jobArgs = []string{"bash", "/opt/flink-operator/submit-job.sh"}
 	jobArgs = append(jobArgs, "--jobmanager", jobManagerAddress)

docs/crd.md

@@ -39,6 +39,7 @@ FlinkCluster
         |__ tolerations
         |__ sidecars
         |__ podAnnotations
+        |__ podLabels
     |__ taskManager
         |__ replicas
         |__ ports
@@ -56,6 +57,7 @@ FlinkCluster
         |__ tolerations
         |__ sidecars
         |__ podAnnotations
+        |__ podLabels
     |__ job
         |__ jarFile
         |__ className
@@ -77,6 +79,7 @@ FlinkCluster
             |__ afterJobCancelled
         |__ cancelRequested
         |__ podAnnotations
+        |__ podLabels
     |__ envVars
     |__ envFrom
     |__ flinkProperties
@@ -191,6 +194,7 @@ FlinkCluster
         See [more info](https://kubernetes.io/docs/concepts/containers/) about containers.
       * **podAnnotations** (optional): Pod template annotations for the JobManager deployment.
         See [more info](https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/) about annotations.
+      * **podLabels** (optional): Pod template labels for the JobManager deployment.
     * **taskManager** (required): TaskManager spec.
       * **replicas** (required): The number of TaskManager replicas.
       * **ports** (optional): Ports that TaskManager listening on.
@@ -227,6 +231,7 @@ FlinkCluster
         See [more info](https://kubernetes.io/docs/concepts/containers/) about containers.
       * **podAnnotations** (optional): Pod template annotations for the TaskManager deployment.
         See [more info](https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/) about annotations.
+      * **podLabels** (optional): Pod template labels for the TaskManager deployment.
     * **job** (optional): Job spec. If specified, the cluster is a Flink job cluster; otherwise, it is a Flink
       session cluster.
       * **jarFile** (required): JAR file of the job. It could be a local file or remote URI, depending on which
@@ -267,6 +272,7 @@ FlinkCluster
         `savePointsDir` is provided, a savepoint will be taken before stopping the job.
       * **podAnnotations** (optional): Pod template annotations for the job.
         See [more info](https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/) about annotations.
+      * **podLabels** (optional): Pod template labels for the job.
     * **envVars** (optional): Environment variables shared by all JobManager, TaskManager and job containers.
     * **envFrom** (optional): Environment variables from ConfigMaps or Secrets shared by all JobManager, TaskManager and job containers.
     * **flinkProperties** (optional): Flink properties which are appened to flink-conf.yaml.