Fix merging of secrets and configmaps (#14090)

closes #13023
apache · Mar 9, 2021 · 7e7657b · 7e7657b
1 parent 08c015f
commit 7e7657b
Show file tree

Hide file tree

Showing 2 changed files with 98 additions and 4 deletions.
diff --git a/airflow/kubernetes/pod_generator.py b/airflow/kubernetes/pod_generator.py
@@ -506,7 +506,7 @@ def reconcile_containers(base_containers,
             'volume_mounts',
             'mount_path')
         client_container = extend_object_field(base_container, client_container, 'env')
-        client_container = extend_object_field(base_container, client_container, 'env_from')
+        client_container = extend_object_field(base_container, client_container, 'env_from', None)
         client_container = extend_object_field(base_container, client_container, 'ports')
         client_container = extend_object_field(base_container, client_container, 'volume_devices')
         client_container = merge_objects(base_container, client_container)
@@ -650,10 +650,14 @@ def extend_object_field(base_obj, client_obj, field_name, field_to_merge="name")
         setattr(client_obj_cp, field_name, base_obj_field)
         return client_obj_cp
 
-    base_obj_set = _get_dict_from_list(base_obj_field, field_to_merge)
-    client_obj_set = _get_dict_from_list(client_obj_field, field_to_merge)
+    if field_to_merge is None:
+        # no merge, just append
+        appended_fields = base_obj_field + client_obj_field
+    else:
+        base_obj_set = _get_dict_from_list(base_obj_field, field_to_merge)
+        client_obj_set = _get_dict_from_list(client_obj_field, field_to_merge)
 
-    appended_fields = _merge_list_of_objects(base_obj_set, client_obj_set)
+        appended_fields = _merge_list_of_objects(base_obj_set, client_obj_set)
 
     setattr(client_obj_cp, field_name, appended_fields)
     return client_obj_cp

diff --git a/tests/kubernetes/test_pod_generator.py b/tests/kubernetes/test_pod_generator.py
@@ -695,6 +695,96 @@ def test_construct_pod_empty_worker_config(self, mock_uuid):
             }
         }, sanitized_result)
 
+    @mock.patch('uuid.uuid4')
+    def test_construct_pod_env_from_merge_secretref_and_configmapref(self, mock_uuid):
+        mock_uuid.return_value = self.static_uuid
+        executor_config = k8s.V1Pod(
+            spec=k8s.V1PodSpec(
+                containers=[
+                    k8s.V1Container(
+                        name='',
+                        resources=k8s.V1ResourceRequirements(
+                            limits={
+                                'cpu': '1m',
+                                'memory': '1G'
+                            }
+                        ),
+                        env_from=[k8s.V1EnvFromSource(
+                            config_map_ref=k8s.V1ConfigMapEnvSource("test_configmap"))
+                        ]
+                    )
+                ]
+            )
+        )
+        worker_config = k8s.V1Pod(
+            spec=k8s.V1PodSpec(
+                containers=[
+                    k8s.V1Container(
+                        name='',
+                        resources=k8s.V1ResourceRequirements(
+                            limits={
+                                'cpu': '1m',
+                                'memory': '1G'
+                            }
+                        ),
+                        env_from=[
+                            k8s.V1EnvFromSource(secret_ref=k8s.V1SecretEnvSource("test_secretref"))
+                        ]
+                    )
+                ]
+            )
+
+        )
+
+        result = PodGenerator.construct_pod(
+            'dag_id',
+            'task_id',
+            'pod_id',
+            self.try_number,
+            "kube_image",
+            self.execution_date,
+            ['command'],
+            executor_config,
+            worker_config,
+            'namespace',
+            'uuid',
+        )
+        sanitized_result = self.k8s_client.sanitize_for_serialization(result)
+
+        self.assertEqual({
+            'apiVersion': 'v1',
+            'kind': 'Pod',
+            'metadata': self.metadata,
+            'spec': {
+                'containers': [{
+                    'args': [],
+                    'command': ['command'],
+                    'env': [],
+                    'envFrom': [
+                        {'secretRef': {
+                            'name': 'test_secretref'
+                        }},
+                        {'configMapRef': {
+                            'name': 'test_configmap'
+                        }}
+                    ],
+                    'name': 'base',
+                    'image': 'kube_image',
+                    'ports': [],
+                    'resources': {
+                        'limits': {
+                            'cpu': '1m',
+                            'memory': '1G'
+                        }
+                    },
+                    'volumeMounts': []
+                }],
+                'hostNetwork': False,
+                'imagePullSecrets': [],
+                'volumes': []
+            }
+        }, sanitized_result)
+
     @mock.patch('uuid.uuid4')
     def test_construct_pod_empty_executor_config(self, mock_uuid):
         mock_uuid.return_value = self.static_uuid