Support for EnvFromSource in cluster spec (#283)

Author: jaredstehler

api/v1beta1/flinkcluster_types.go

@@ -413,6 +413,10 @@ type FlinkClusterSpec struct {
 	// containers.
 	EnvVars []corev1.EnvVar `json:"envVars,omitempty"`
 
+	// Environment variables injected from a source, shared by all JobManager,
+	// TaskManager and job containers.
+	EnvFrom []corev1.EnvFromSource `json:"envFrom,omitempty"`
+
 	// Flink properties which are appened to flink-conf.yaml.
 	FlinkProperties map[string]string `json:"flinkProperties,omitempty"`
 

api/v1beta1/zz_generated.deepcopy.go

@@ -30,6 +30,27 @@ spec:
           properties:
             batchSchedulerName:
               type: string
+            envFrom:
+              items:
+                properties:
+                  configMapRef:
+                    properties:
+                      name:
+                        type: string
+                      optional:
+                        type: boolean
+                    type: object
+                  prefix:
+                    type: string
+                  secretRef:
+                    properties:
+                      name:
+                        type: string
+                      optional:
+                        type: boolean
+                    type: object
+                type: object
+              type: array
             envVars:
               items:
                 properties:

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

@@ -187,6 +187,7 @@ func getDesiredJobManagerDeployment(
 		ReadinessProbe:  &readinessProbe,
 		Resources:       jobManagerSpec.Resources,
 		Env:             envVars,
+		EnvFrom:         flinkCluster.Spec.EnvFrom,
 		VolumeMounts:    volumeMounts,
 	}}
 
@@ -477,6 +478,7 @@ func getDesiredTaskManagerDeployment(
 		ReadinessProbe:  &readinessProbe,
 		Resources:       taskManagerSpec.Resources,
 		Env:             envVars,
+		EnvFrom:         flinkCluster.Spec.EnvFrom,
 		VolumeMounts:    volumeMounts,
 	}}
 	containers = append(containers, taskManagerSpec.Sidecars...)

controllers/flinkcluster_converter.go

@@ -239,6 +239,10 @@ func TestGetDesiredClusterState(t *testing.T) {
 			},
 			FlinkProperties: map[string]string{"taskmanager.numberOfTaskSlots": "1"},
 			EnvVars:         []corev1.EnvVar{{Name: "FOO", Value: "abc"}},
+			EnvFrom:        []corev1.EnvFromSource{{ConfigMapRef: &corev1.ConfigMapEnvSource{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: "FOOMAP",
+			}}}},
 			HadoopConfig: &v1beta1.HadoopConfig{
 				ConfigMapName: "hadoop-configmap",
 				MountPath:     "/etc/hadoop/conf",
@@ -347,6 +351,15 @@ func TestGetDesiredClusterState(t *testing.T) {
 									Value: "abc",
 								},
 							},
+							EnvFrom: []corev1.EnvFromSource{
+								{
+									ConfigMapRef: &corev1.ConfigMapEnvSource{
+										LocalObjectReference: corev1.LocalObjectReference{
+											Name: "FOOMAP",
+										},
+									},
+								},
+							},
 							Resources: corev1.ResourceRequirements{
 								Requests: map[corev1.ResourceName]resource.Quantity{
 									corev1.ResourceCPU:    resource.MustParse("100m"),
@@ -601,6 +614,15 @@ func TestGetDesiredClusterState(t *testing.T) {
 									Value: "abc",
 								},
 							},
+							EnvFrom: []corev1.EnvFromSource{
+								{
+									ConfigMapRef: &corev1.ConfigMapEnvSource{
+										LocalObjectReference: corev1.LocalObjectReference{
+											Name: "FOOMAP",
+										},
+									},
+								},
+							},
 							Resources: corev1.ResourceRequirements{
 								Requests: map[corev1.ResourceName]resource.Quantity{
 									corev1.ResourceCPU:    resource.MustParse("200m"),

controllers/flinkcluster_converter_test.go

@@ -78,6 +78,7 @@ FlinkCluster
         |__ cancelRequested
         |__ podAnnotations
     |__ envVars
+    |__ envFrom
     |__ flinkProperties
     |__ hadoopConfig
         |__ configMapName
@@ -266,6 +267,7 @@ FlinkCluster
       * **podAnnotations** (optional): Pod template annotations for the job.
         See [more info](https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/) about annotations.
     * **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.
     * **hadoopConfig** (optional): Configs for Hadoop.
       * **configMapName**: The name of the ConfigMap which holds the Hadoop config files. The ConfigMap must be in the