This is an automated email from the ASF dual-hosted git repository.
dimberman pushed a commit to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v1-10-test by this push:
new 589d907 Fix pod_mutation_hook for 1.10.13 (#10850)
589d907 is described below
commit 589d907039c80950d0c8d71b7ca0e41d29428cb6
Author: Daniel Imberman <[email protected]>
AuthorDate: Thu Sep 10 11:27:08 2020 -0700
Fix pod_mutation_hook for 1.10.13 (#10850)
* Fix breaking changes in Pod conversion for 1.10.13
* fix tests
* fix flake8
* fix test
* fix image secrets
* Update airflow/kubernetes/pod_launcher.py
Co-authored-by: Kaxil Naik <[email protected]>
Co-authored-by: Kaxil Naik <[email protected]>
---
airflow/bin/cli.py | 2 ++
airflow/contrib/kubernetes/pod.py | 20 ++++++++++++++--
airflow/kubernetes/pod_launcher.py | 3 ++-
tests/kubernetes/models/test_pod.py | 12 +++++++---
tests/kubernetes/test_pod_launcher.py | 2 ++
tests/test_local_settings/test_local_settings.py | 29 ++++++++++++------------
6 files changed, 48 insertions(+), 20 deletions(-)
diff --git a/airflow/bin/cli.py b/airflow/bin/cli.py
index 45f1c5a..cc8372a 100644
--- a/airflow/bin/cli.py
+++ b/airflow/bin/cli.py
@@ -1255,6 +1255,7 @@ def _serve_logs(env, skip_serve_logs=False):
def kubernetes_generate_dag_yaml(args):
from airflow.executors.kubernetes_executor import
AirflowKubernetesScheduler, KubeConfig
from airflow.kubernetes.pod_generator import PodGenerator
+ from airflow.kubernetes.pod_launcher import PodLauncher
from airflow.kubernetes.worker_configuration import WorkerConfiguration
from kubernetes.client.api_client import ApiClient
dag = get_dag(args)
@@ -1276,6 +1277,7 @@ def kubernetes_generate_dag_yaml(args):
worker_config=WorkerConfiguration(kube_config=kube_config).as_pod()
)
api_client = ApiClient()
+ pod = PodLauncher._mutate_pod_backcompat(pod)
date_string =
AirflowKubernetesScheduler._datetime_to_label_safe_datestring( # pylint:
disable=W0212
args.execution_date)
yaml_file_name = "{}_{}_{}.yml".format(args.dag_id, ti.task_id,
date_string)
diff --git a/airflow/contrib/kubernetes/pod.py
b/airflow/contrib/kubernetes/pod.py
index 9dc11cc..d1f30a8 100644
--- a/airflow/contrib/kubernetes/pod.py
+++ b/airflow/contrib/kubernetes/pod.py
@@ -118,7 +118,7 @@ class Pod(object):
self.node_selectors = node_selectors or {}
self.namespace = namespace
self.image_pull_policy = image_pull_policy
- self.image_pull_secrets = image_pull_secrets
+ self.image_pull_secrets = image_pull_secrets or ""
self.init_containers = init_containers
self.service_account_name = service_account_name
self.resources = resources or Resources()
@@ -142,21 +142,29 @@ class Pod(object):
labels=self.labels,
name=self.name,
namespace=self.namespace,
+ annotations=self.annotations,
)
+ if self.image_pull_secrets:
+ image_pull_secrets = [k8s.V1LocalObjectReference(i)
+ for i in self.image_pull_secrets.split(",")]
+ else:
+ image_pull_secrets = []
spec = k8s.V1PodSpec(
init_containers=self.init_containers,
containers=[
k8s.V1Container(
image=self.image,
command=self.cmds,
+ env_from=[],
name="base",
env=[k8s.V1EnvVar(name=key, value=val) for key, val in
self.envs.items()],
args=self.args,
image_pull_policy=self.image_pull_policy,
)
],
- image_pull_secrets=self.image_pull_secrets,
+ image_pull_secrets=image_pull_secrets,
service_account_name=self.service_account_name,
+ node_selector=self.node_selectors,
dns_policy=self.dnspolicy,
host_network=self.hostnetwork,
tolerations=self.tolerations,
@@ -168,6 +176,14 @@ class Pod(object):
spec=spec,
metadata=meta,
)
+ for configmap_name in self.configmaps:
+ env_var = k8s.V1EnvFromSource(
+ config_map_ref=k8s.V1ConfigMapEnvSource(
+ name=configmap_name,
+ )
+ )
+ pod.spec.containers[0].env_from.append(env_var)
+
for port in _extract_ports(self.ports):
pod = port.attach_to_pod(pod)
volumes = _extract_volumes(self.volumes)
diff --git a/airflow/kubernetes/pod_launcher.py
b/airflow/kubernetes/pod_launcher.py
index e8d4d22..704a77e 100644
--- a/airflow/kubernetes/pod_launcher.py
+++ b/airflow/kubernetes/pod_launcher.py
@@ -303,6 +303,7 @@ def _convert_to_airflow_pod(pod):
volumes = _extract_volumes(pod.spec.volumes)
api_client = ApiClient()
init_containers = pod.spec.init_containers
+ image_pull_secrets = pod.spec.image_pull_secrets or []
if pod.spec.init_containers is not None:
init_containers = [api_client.sanitize_for_serialization(i) for i in
pod.spec.init_containers]
dummy_pod = Pod(
@@ -321,7 +322,7 @@ def _convert_to_airflow_pod(pod):
image_pull_policy=base_container.image_pull_policy or 'IfNotPresent',
tolerations=pod.spec.tolerations,
init_containers=init_containers,
- image_pull_secrets=pod.spec.image_pull_secrets,
+ image_pull_secrets=",".join([i.name for i in image_pull_secrets]),
resources=base_container.resources,
service_account_name=pod.spec.service_account_name,
secrets=secrets,
diff --git a/tests/kubernetes/models/test_pod.py
b/tests/kubernetes/models/test_pod.py
index 8a89da0..f8df28a 100644
--- a/tests/kubernetes/models/test_pod.py
+++ b/tests/kubernetes/models/test_pod.py
@@ -120,7 +120,10 @@ class TestPod(unittest.TestCase):
result = k8s_client.sanitize_for_serialization(result)
expected = \
- {'metadata': {'labels': {}, 'name': 'bar', 'namespace': 'baz'},
+ {'metadata': {'annotations': {},
+ 'labels': {},
+ 'name': 'bar',
+ 'namespace': 'baz'},
'spec': {'affinity': {},
'containers': [{'args': [],
'command': ['airflow'],
@@ -128,6 +131,7 @@ class TestPod(unittest.TestCase):
{'name': 'AIRFLOW_SECRET',
'valueFrom': {'secretKeyRef':
{'key': 'airflow_config',
'name': 'secret_name'}}}],
+ 'envFrom': [],
'image': 'foo',
'imagePullPolicy': 'Never',
'name': 'base',
@@ -139,18 +143,20 @@ class TestPod(unittest.TestCase):
'readOnly': True,
'subPath': '/'},
{'mountPath':
'/opt/airflow',
- 'name': 'secretvol' +
str(static_uuid),
+ 'name': 'secretvol' +
str(static_uuid),
'readOnly': True}]}],
'hostNetwork': False,
+ 'imagePullSecrets': [],
'initContainers': {'name': 'test-container',
'volumeMounts': {'mountPath':
'/foo/bar',
'name':
'init-volume-secret'}},
+ 'nodeSelector': {},
'securityContext': {},
'tolerations': [],
'volumes': [{'name': 'foo'},
{'name': 'bar',
'secret': {'secretName': 'volume-secret'}},
- {'name':
'secretvolcf4a56d2-8101-4217-b027-2af6216feb48',
+ {'name': 'secretvol' + str(static_uuid),
'secret': {'secretName':
'init-volume-secret'}},
{'name': 'secretvol' + str(static_uuid),
'secret': {'secretName': 'volume-secret'}}
diff --git a/tests/kubernetes/test_pod_launcher.py
b/tests/kubernetes/test_pod_launcher.py
index f56615f..63169ae 100644
--- a/tests/kubernetes/test_pod_launcher.py
+++ b/tests/kubernetes/test_pod_launcher.py
@@ -228,6 +228,7 @@ class TestPodLauncherHelper(unittest.TestCase):
run_as_user=0,
fs_group=0,
),
+ image_pull_secrets=[k8s.V1LocalObjectReference("my-secret")],
volumes=[
k8s.V1Volume(
name="myvolume"
@@ -286,6 +287,7 @@ class TestPodLauncherHelper(unittest.TestCase):
sub_path=None,
read_only=True
)],
+ image_pull_secrets="my-secret",
secrets=[Secret("env", "AIRFLOW_SECRET", "ai", "secret_key")],
security_context={'fsGroup': 0, 'runAsUser': 0},
volumes=[Volume(name="myvolume", configs={'name': 'myvolume'}),
diff --git a/tests/test_local_settings/test_local_settings.py
b/tests/test_local_settings/test_local_settings.py
index 9fab355..e69bd05 100644
--- a/tests/test_local_settings/test_local_settings.py
+++ b/tests/test_local_settings/test_local_settings.py
@@ -318,9 +318,10 @@ class LocalSettingsTest(unittest.TestCase):
self.assertEqual(
sanitized_pod_post_mutation,
- {"apiVersion": "v1",
- "kind": "Pod",
- 'metadata': {'labels': {'test_label': 'test_value'},
+ {'apiVersion': 'v1',
+ 'kind': 'Pod',
+ 'metadata': {'annotations': {},
+ 'labels': {'test_label': 'test_value'},
'name': mock.ANY,
'namespace': 'airflow-tests'},
'spec': {'affinity': {'nodeAffinity':
{'requiredDuringSchedulingIgnoredDuringExecution': {
@@ -330,6 +331,7 @@ class LocalSettingsTest(unittest.TestCase):
'containers': [{'args': ['/bin/sh', '-c', 'touch
/tmp/healthy2'],
'command': ['foo'],
'env': [{'name': 'TEST_USER',
'value': 'ADMIN'}],
+ 'envFrom': [],
'image': 'my_image',
'imagePullPolicy': 'Never',
'name': 'base',
@@ -338,14 +340,13 @@ class LocalSettingsTest(unittest.TestCase):
'resources': {'limits':
{'nvidia.com/gpu': '200G'},
'requests': {'cpu':
'200Mi',
'memory':
'2G'}},
- 'volumeMounts': [
- {'mountPath': '/mnt',
- 'name': 'foo',
- 'readOnly': True,
- 'subPath': '/'},
- {'mountPath':
'/opt/airflow/secrets/',
- 'name': 'airflow-secrets-mount',
- 'readOnly': True}]}],
+ 'volumeMounts': [{'mountPath':
'/mnt',
+ 'name': 'foo',
+ 'readOnly': True,
+ 'subPath': '/'},
+ {'mountPath':
'/opt/airflow/secrets/',
+ 'name':
'airflow-secrets-mount',
+ 'readOnly':
True}]}],
'hostNetwork': False,
'imagePullSecrets': [],
'initContainers': [{'name': 'init-container',
@@ -353,6 +354,8 @@ class LocalSettingsTest(unittest.TestCase):
'runAsUser':
50000},
'volumeMounts': [{'mountPath':
'/tmp',
'name':
'init-secret'}]}],
+ 'nodeSelector': {},
+ 'securityContext': {'runAsUser': 1},
'tolerations': [{'effect': 'NoSchedule',
'key': 'static-pods',
'operator': 'Equal',
@@ -364,9 +367,7 @@ class LocalSettingsTest(unittest.TestCase):
'volumes': [{'name': 'airflow-secrets-mount',
'secret': {'secretName':
'airflow-test-secrets'}},
{'name': 'bar'},
- {'name': 'foo'},
- ],
- 'securityContext': {'runAsUser': 1}}}
+ {'name': 'foo'}]}}
)
def test_pod_mutation_v1_pod(self):
