You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by di...@apache.org on 2020/09/17 15:41:00 UTC
[airflow] branch master updated: Simplify the K8sExecutor and
K8sPodOperator (#10393)
This is an automated email from the ASF dual-hosted git repository.
dimberman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/master by this push:
new cba51d4 Simplify the K8sExecutor and K8sPodOperator (#10393)
cba51d4 is described below
commit cba51d49eea6a0563044191c8111978836d697ef
Author: Daniel Imberman <da...@gmail.com>
AuthorDate: Thu Sep 17 08:40:20 2020 -0700
Simplify the K8sExecutor and K8sPodOperator (#10393)
* Simplify Airflow on Kubernetes Story
Removes thousands of lines of code that essentially ammount to us
re-creating the Kubernetes API. Will offer a faster, simpler
KubernetesExecutor for 2.0
* Fix podgen tests
* fix documentation
* simplify validate function
* @mik-laj comments
* spellcheck
* spellcheck
* Update airflow/executors/kubernetes_executor.py
Co-authored-by: Kaxil Naik <ka...@gmail.com>
Co-authored-by: Kaxil Naik <ka...@gmail.com>
---
UPDATING.md | 522 +++++++++++-
airflow/cli/commands/dag_command.py | 3 +-
airflow/config_templates/config.yml | 431 +---------
airflow/config_templates/default_airflow.cfg | 267 +-----
.../example_kubernetes_executor_config.py | 18 +-
airflow/executors/kubernetes_executor.py | 190 +----
airflow/kubernetes/pod.py | 127 ---
airflow/kubernetes/pod_generator.py | 292 ++-----
airflow/kubernetes/pod_generator_deprecated.py | 341 ++++++++
airflow/kubernetes/volume.py | 56 --
airflow/kubernetes/volume_mount.py | 72 --
airflow/kubernetes/worker_configuration.py | 448 ----------
.../basic_template.yaml | 79 ++
.../kubernetes/example_dags/example_kubernetes.py | 28 +-
.../cncf/kubernetes/operators/kubernetes_pod.py | 137 ++-
airflow/serialization/serialized_objects.py | 3 +
airflow/utils/json.py | 8 +
docs/howto/operator/kubernetes.rst | 20 +-
kubernetes_tests/test_kubernetes_pod_operator.py | 177 ++--
scripts/ci/libraries/_kind.sh | 2 +
tests/executors/test_kubernetes_executor.py | 104 +--
tests/kubernetes/models/test_pod.py | 77 --
tests/kubernetes/models/test_secret.py | 84 +-
tests/kubernetes/pod_generator_base.yaml | 52 ++
.../pod_generator_base_with_secrets.yaml | 61 ++
tests/kubernetes/test_pod_generator.py | 682 ++++++---------
tests/kubernetes/test_worker_configuration.py | 930 ---------------------
.../kubernetes/operators/test_kubernetes_pod.py | 2 +-
tests/test_config_templates.py | 6 +-
29 files changed, 1718 insertions(+), 3501 deletions(-)
diff --git a/UPDATING.md b/UPDATING.md
index ef6c035..faca832 100644
--- a/UPDATING.md
+++ b/UPDATING.md
@@ -138,7 +138,6 @@ with third party services to the ``airflow.providers`` package.
All changes made are backward compatible, but if you use the old import paths you will
see a deprecation warning. The old import paths can be abandoned in the future.
-
### Breaking Change in OAuth
The flask-ouathlib has been replaced with authlib because flask-outhlib has
@@ -154,6 +153,527 @@ The Old and New provider configuration keys that have changed are as follows
For more information, visit https://flask-appbuilder.readthedocs.io/en/latest/security.html#authentication-oauth
+### Changes to the KubernetesExecutor
+
+#### The KubernetesExecutor Will No Longer Read from the airflow.cfg for Base Pod Configurations
+
+In Airflow 2.0, the KubernetesExecutor will require a base pod template written in yaml. This file can exist
+anywhere on the host machine and will be linked using the `pod_template_file` configuration in the airflow.cfg.
+
+The `airflow.cfg` will still accept values for the `worker_container_repository`, the `worker_container_tag`, and
+the default namespace.
+
+The following `airflow.cfg` values will be deprecated:
+
+```
+worker_container_image_pull_policy
+airflow_configmap
+airflow_local_settings_configmap
+dags_in_image
+dags_volume_subpath
+dags_volume_mount_point
+dags_volume_claim
+logs_volume_subpath
+logs_volume_claim
+dags_volume_host
+logs_volume_host
+env_from_configmap_ref
+env_from_secret_ref
+git_repo
+git_branch
+git_sync_depth
+git_subpath
+git_sync_rev
+git_user
+git_password
+git_sync_root
+git_sync_dest
+git_dags_folder_mount_point
+git_ssh_key_secret_name
+git_ssh_known_hosts_configmap_name
+git_sync_credentials_secret
+git_sync_container_repository
+git_sync_container_tag
+git_sync_init_container_name
+git_sync_run_as_user
+worker_service_account_name
+image_pull_secrets
+gcp_service_account_keys
+affinity
+tolerations
+run_as_user
+fs_group
+[kubernetes_node_selectors]
+[kubernetes_annotations]
+[kubernetes_environment_variables]
+[kubernetes_secrets]
+[kubernetes_labels]
+```
+
+#### The `executor_config` Will Now Expect a `kubernetes.client.models.V1Pod` Class When Launching Tasks
+
+In Airflow 1.10.x, users could modify task pods at runtime by passing a dictionary to the `executor_config` variable.
+Users will now have full access the Kubernetes API via the `kubernetes.client.models.V1Pod`.
+
+While in the deprecated version a user would mount a volume using the following dictionary:
+
+```python
+second_task = PythonOperator(
+ task_id="four_task",
+ python_callable=test_volume_mount,
+ executor_config={
+ "KubernetesExecutor": {
+ "volumes": [
+ {
+ "name": "example-kubernetes-test-volume",
+ "hostPath": {"path": "/tmp/"},
+ },
+ ],
+ "volume_mounts": [
+ {
+ "mountPath": "/foo/",
+ "name": "example-kubernetes-test-volume",
+ },
+ ]
+ }
+ }
+)
+```
+
+In the new model a user can accomplish the same thing using the following code under the ``pod_override`` key:
+
+```python
+from kubernetes.client import models as k8s
+
+second_task = PythonOperator(
+ task_id="four_task",
+ python_callable=test_volume_mount,
+ executor_config={"pod_override": k8s.V1Pod(
+ spec=k8s.V1PodSpec(
+ containers=[
+ k8s.V1Container(
+ name="base",
+ volume_mounts=[
+ k8s.V1VolumeMount(
+ mount_path="/foo/",
+ name="example-kubernetes-test-volume"
+ )
+ ]
+ )
+ ],
+ volumes=[
+ k8s.V1Volume(
+ name="example-kubernetes-test-volume",
+ host_path=k8s.V1HostPathVolumeSource(
+ path="/tmp/"
+ )
+ )
+ ]
+ )
+ )
+ }
+)
+```
+For Airflow 2.0, the traditional `executor_config` will continue operation with a deprecation warning,
+but will be removed in a future version.
+
+### Changes to the KubernetesPodOperator
+
+Much like the `KubernetesExecutor`, the `KubernetesPodOperator` will no longer take Airflow custom classes and will
+instead expect either a pod_template yaml file, or `kubernetes.client.models` objects.
+
+The one notable exception is that we will continue to support the `airflow.kubernetes.secret.Secret` class.
+
+Whereas previously a user would import each individual class to build the pod as so:
+
+```python
+from airflow.kubernetes.pod import Port
+from airflow.kubernetes.volume import Volume
+from airflow.kubernetes.secret import Secret
+from airflow.kubernetes.volume_mount import VolumeMount
+
+
+volume_config = {
+ 'persistentVolumeClaim': {
+ 'claimName': 'test-volume'
+ }
+}
+volume = Volume(name='test-volume', configs=volume_config)
+volume_mount = VolumeMount('test-volume',
+ mount_path='/root/mount_file',
+ sub_path=None,
+ read_only=True)
+
+port = Port('http', 80)
+secret_file = Secret('volume', '/etc/sql_conn', 'airflow-secrets', 'sql_alchemy_conn')
+secret_env = Secret('env', 'SQL_CONN', 'airflow-secrets', 'sql_alchemy_conn')
+
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo", "10"],
+ labels={"foo": "bar"},
+ secrets=[secret_file, secret_env],
+ ports=[port],
+ volumes=[volume],
+ volume_mounts=[volume_mount],
+ name="airflow-test-pod",
+ task_id="task",
+ affinity=affinity,
+ is_delete_operator_pod=True,
+ hostnetwork=False,
+ tolerations=tolerations,
+ configmaps=configmaps,
+ init_containers=[init_container],
+ priority_class_name="medium",
+)
+```
+Now the user can use the `kubernetes.client.models` class as a single point of entry for creating all k8s objects.
+
+```python
+from kubernetes.client import models as k8s
+from airflow.kubernetes.secret import Secret
+
+
+configmaps = ['test-configmap-1', 'test-configmap-2']
+
+volume = k8s.V1Volume(
+ name='test-volume',
+ persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(claim_name='test-volume'),
+)
+
+port = k8s.V1ContainerPort(name='http', container_port=80)
+secret_file = Secret('volume', '/etc/sql_conn', 'airflow-secrets', 'sql_alchemy_conn')
+secret_env = Secret('env', 'SQL_CONN', 'airflow-secrets', 'sql_alchemy_conn')
+secret_all_keys = Secret('env', None, 'airflow-secrets-2')
+volume_mount = k8s.V1VolumeMount(
+ name='test-volume', mount_path='/root/mount_file', sub_path=None, read_only=True
+)
+
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo", "10"],
+ labels={"foo": "bar"},
+ secrets=[secret_file, secret_env],
+ ports=[port],
+ volumes=[volume],
+ volume_mounts=[volume_mount],
+ name="airflow-test-pod",
+ task_id="task",
+ is_delete_operator_pod=True,
+ hostnetwork=False)
+```
+We decided to keep the Secret class as users seem to really like that simplifies the complexity of mounting
+Kubernetes secrets into workers.
+
+### Changed Parameters for the KubernetesPodOperator
+
+#### port has migrated from a List[Port] to a List[V1ContainerPort]
+Before:
+```python
+from airflow.kubernetes.pod import Port
+port = Port('http', 80)
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ ports=[port],
+ task_id="task",
+)
+```
+
+After:
+```python
+from kubernetes.client import models as k8s
+port = k8s.V1ContainerPort(name='http', container_port=80)
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ ports=[port],
+ task_id="task",
+)
+```
+
+#### volume_mounts has migrated from a List[VolumeMount] to a List[V1VolumeMount]
+Before:
+```python
+from airflow.kubernetes.volume_mount import VolumeMount
+volume_mount = VolumeMount('test-volume',
+ mount_path='/root/mount_file',
+ sub_path=None,
+ read_only=True)
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ volume_mounts=[volume_mount],
+ task_id="task",
+)
+```
+
+After:
+```python
+from kubernetes.client import models as k8s
+volume_mount = k8s.V1VolumeMount(
+ name='test-volume', mount_path='/root/mount_file', sub_path=None, read_only=True
+)
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ volume_mounts=[volume_mount],
+ task_id="task",
+)
+```
+
+#### volumes has migrated from a List[Volume] to a List[V1Volume]
+Before:
+```python
+from airflow.kubernetes.volume import Volume
+
+volume_config = {
+ 'persistentVolumeClaim': {
+ 'claimName': 'test-volume'
+ }
+}
+volume = Volume(name='test-volume', configs=volume_config)
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ volumes=[volume],
+ task_id="task",
+)
+```
+
+After:
+```python
+from kubernetes.client import models as k8s
+volume = k8s.V1Volume(
+ name='test-volume',
+ persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(claim_name='test-volume'),
+)
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ volumes=[volume],
+ task_id="task",
+)
+```
+#### env_vars has migrated from a Dict to a List[V1EnvVar]
+Before:
+```python
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ env_vars={"ENV1": "val1", "ENV2": "val2"},
+ task_id="task",
+)
+```
+
+After:
+```python
+from kubernetes.client import models as k8s
+
+env_vars = [
+ k8s.V1EnvVar(
+ name="ENV1",
+ value="val1"
+ ),
+ k8s.V1EnvVar(
+ name="ENV2",
+ value="val2"
+ )]
+
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ env_vars=env_vars,
+ task_id="task",
+)
+```
+
+#### PodRuntimeInfoEnv has been removed
+
+PodRuntimeInfoEnv can now be added to the `env_vars` variable as a `V1EnvVarSource`
+
+Before:
+```python
+from airflow.kubernetes.pod_runtime_info_env import PodRuntimeInfoEnv
+
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ pod_runtime_info_envs=[PodRuntimeInfoEnv("ENV3", "status.podIP")],
+ task_id="task",
+)
+```
+
+After:
+```python
+from kubernetes.client import models as k8s
+
+env_vars = [
+ k8s.V1EnvVar(
+ name="ENV3",
+ value_from=k8s.V1EnvVarSource(
+ field_ref=k8s.V1ObjectFieldSelector(
+ field_path="status.podIP"
+ )
+ )
+ )
+]
+
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ env_vars=env_vars,
+ task_id="task",
+)
+```
+
+#### configmaps has been removed
+
+configmaps can now be added to the `env_from` variable as a `V1EnvVarSource`
+
+Before:
+```python
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ configmaps=['test-configmap'],
+ task_id="task"
+)
+```
+
+After:
+
+```python
+from kubernetes.client import models as k8s
+
+configmap ="test-configmap"
+env_from = [k8s.V1EnvFromSource(
+ config_map_ref=k8s.V1ConfigMapEnvSource(
+ name=configmap
+ )
+ )]
+
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ env_from=env_from,
+ task_id="task"
+)
+```
+
+
+#### resources has migrated from a Dict to a V1ResourceRequirements
+
+Before:
+```python
+resources = {
+ 'limit_cpu': 0.25,
+ 'limit_memory': '64Mi',
+ 'limit_ephemeral_storage': '2Gi',
+ 'request_cpu': '250m',
+ 'request_memory': '64Mi',
+ 'request_ephemeral_storage': '1Gi',
+}
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ labels={"foo": "bar"},
+ name="test",
+ task_id="task" + self.get_current_task_name(),
+ in_cluster=False,
+ do_xcom_push=False,
+ resources=resources,
+)
+```
+
+After:
+```python
+from kubernetes.client import models as k8s
+
+resources=k8s.V1ResourceRequirements(
+ requests={
+ 'memory': '64Mi',
+ 'cpu': '250m',
+ 'ephemeral-storage': '1Gi'
+ },
+ limits={
+ 'memory': '64Mi',
+ 'cpu': 0.25,
+ 'nvidia.com/gpu': None,
+ 'ephemeral-storage': '2Gi'
+ }
+)
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ labels={"foo": "bar"},
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
+ in_cluster=False,
+ do_xcom_push=False,
+ resources=resources,
+)
+```
+#### image_pull_secrets has migrated from a String to a List[k8s.V1LocalObjectReference]
+
+Before:
+```python
+k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ name="test",
+ task_id="task",
+ image_pull_secrets="fake-secret",
+ cluster_context='default')
+```
+
+After:
+```python
+quay_k8s = KubernetesPodOperator(
+ namespace='default',
+ image='quay.io/apache/bash',
+ image_pull_secrets=[k8s.V1LocalObjectReference('testquay')],
+ cmds=["bash", "-cx"],
+ name="airflow-private-image-pod",
+ task_id="task-two",
+)
+
+```
+
### Migration Guide from Experimental API to Stable API v1
In Airflow 2.0, we added the new REST API. Experimental API still works, but support may be dropped in the future.
If your application is still using the experimental API, you should consider migrating to the stable API.
diff --git a/airflow/cli/commands/dag_command.py b/airflow/cli/commands/dag_command.py
index b15aa76..02b8b74 100644
--- a/airflow/cli/commands/dag_command.py
+++ b/airflow/cli/commands/dag_command.py
@@ -389,7 +389,6 @@ def generate_pod_yaml(args):
from airflow.executors.kubernetes_executor import AirflowKubernetesScheduler, KubeConfig
from airflow.kubernetes import pod_generator
from airflow.kubernetes.pod_generator import PodGenerator
- from airflow.kubernetes.worker_configuration import WorkerConfiguration
from airflow.settings import pod_mutation_hook
execution_date = args.execution_date
@@ -410,7 +409,7 @@ def generate_pod_yaml(args):
pod_override_object=PodGenerator.from_obj(ti.executor_config),
worker_uuid="worker-config",
namespace=kube_config.executor_namespace,
- base_worker_pod=WorkerConfiguration(kube_config=kube_config).as_pod()
+ base_worker_pod=PodGenerator.deserialize_model_file(kube_config.pod_template_file)
)
pod_mutation_hook(pod)
api_client = ApiClient()
diff --git a/airflow/config_templates/config.yml b/airflow/config_templates/config.yml
index c1400ad..2d08f46 100644
--- a/airflow/config_templates/config.yml
+++ b/airflow/config_templates/config.yml
@@ -1849,17 +1849,17 @@
- name: kubernetes
description: ~
options:
- - name: worker_container_repository
+ - name: pod_template_file
description: |
- The repository of the Kubernetes Image for the Worker to Run
- version_added: ~
+ Path to the YAML pod file. If set, all other kubernetes-related fields are ignored.
+ version_added: 1.10.11
type: string
example: ~
default: ""
- - name: pod_template_file
+ - name: worker_container_repository
description: |
- Path to the YAML pod file. If set, all other kubernetes-related fields are ignored.
- version_added: 1.10.11
+ The repository of the Kubernetes Image for the Worker to Run
+ version_added: ~
type: string
example: ~
default: ""
@@ -1870,13 +1870,13 @@
type: string
example: ~
default: ""
- - name: worker_container_image_pull_policy
+ - name: namespace
description: |
- The imagePullPolicy of the Kubernetes Image for the Worker to Run
- version_added: 1.10.2
+ The Kubernetes namespace where airflow workers should be created. Defaults to ``default``
+ version_added: ~
type: string
example: ~
- default: "IfNotPresent"
+ default: "default"
- name: delete_worker_pods
description: |
If True, all worker pods will be deleted upon termination
@@ -1903,13 +1903,6 @@
type: string
example: ~
default: "1"
- - name: namespace
- description: |
- The Kubernetes namespace where airflow workers should be created. Defaults to ``default``
- version_added: ~
- type: string
- example: ~
- default: "default"
- name: multi_namespace_mode
description: |
Allows users to launch pods in multiple namespaces.
@@ -1918,317 +1911,6 @@
type: boolean
example: ~
default: "False"
- - name: airflow_configmap
- description: |
- The name of the Kubernetes ConfigMap containing ``airflow.cfg`` file.
-
- For example:
-
- ``airflow_configmap = "airflow-configmap"`` if you have the following ConfigMap.
-
- ``airflow-configmap.yaml``:
-
- .. code-block:: yaml
-
- ---
- apiVersion: v1
- kind: ConfigMap
- metadata:
- name: airflow-configmap
- data:
- airflow.cfg: |
- ...
-
- Note that the ``airflow_configmap`` and the ``airflow_local_settings_configmap``
- can be the same configMap
- version_added: ~
- type: string
- example: "airflow-configmap"
- default: ""
- - name: airflow_local_settings_configmap
- description: |
- The name of the Kubernetes ConfigMap containing ``airflow_local_settings.py`` file.
-
- For example:
-
- ``airflow_local_settings_configmap = "airflow-configmap"`` if you have the following ConfigMap.
-
- ``airflow-configmap.yaml``:
-
- .. code-block:: yaml
-
- ---
- apiVersion: v1
- kind: ConfigMap
- metadata:
- name: airflow-configmap
- data:
- airflow_local_settings.py: |
- def pod_mutation_hook(pod):
- ...
- airflow.cfg: |
- ...
- version_added: 1.10.8
- type: string
- example: "airflow-configmap"
- default: ""
- - name: dags_in_image
- description: |
- For docker image already contains DAGs, this is set to ``True``, and the worker will
- search for dags in dags_folder, otherwise use git sync or dags volume claim to mount DAGs
- version_added: 1.10.2
- type: string
- example: ~
- default: "False"
- - name: dags_volume_subpath
- description: |
- For either git sync or volume mounted DAGs, the worker will look in this subpath for DAGs
- version_added: ~
- type: string
- example: ~
- default: ""
- - name: dags_volume_mount_point
- description: |
- For either git sync or volume mounted DAGs, the worker will mount the volume in this path
- version_added: 1.10.11
- type: string
- example: ~
- default: ""
- - name: dags_volume_claim
- description: |
- For DAGs mounted via a volume claim (mutually exclusive with git-sync and host path)
- version_added: ~
- type: string
- example: ~
- default: ""
- - name: logs_volume_subpath
- description: |
- For volume mounted logs, the worker will look in this subpath for logs
- version_added: ~
- type: string
- example: ~
- default: ""
- - name: logs_volume_claim
- description: |
- A shared volume claim for the logs
- version_added: ~
- type: string
- example: ~
- default: ""
- - name: dags_volume_host
- description: |
- For DAGs mounted via a hostPath volume (mutually exclusive with volume claim and git-sync)
- Useful in local environment, discouraged in production
- version_added: 1.10.2
- type: string
- example: ~
- default: ""
- - name: logs_volume_host
- description: |
- A hostPath volume for the logs
- Useful in local environment, discouraged in production
- version_added: 1.10.2
- type: string
- example: ~
- default: ""
- - name: env_from_configmap_ref
- description: |
- A list of configMapsRefs to envFrom. If more than one configMap is
- specified, provide a comma separated list: configmap_a,configmap_b
- version_added: 1.10.3
- type: string
- example: ~
- default: ""
- - name: env_from_secret_ref
- description: |
- A list of secretRefs to envFrom. If more than one secret is
- specified, provide a comma separated list: secret_a,secret_b
- version_added: 1.10.3
- type: string
- example: ~
- default: ""
- - name: git_repo
- description: |
- Git credentials and repository for DAGs mounted via Git (mutually exclusive with volume claim)
- version_added: ~
- type: string
- example: ~
- default: ""
- - name: git_branch
- description: |
- Branch of the git repo containing DAGs. This can be useful for users who wish to version their DAGs
- using release branches.
- version_added: ~
- type: string
- example: ~
- default: ""
- - name: git_sync_depth
- description: |
- Use a shallow clone with a history truncated to the specified number of commits.
- 0 - do not use shallow clone.
- version_added: 1.10.11
- type: string
- example: ~
- default: "1"
- - name: git_subpath
- description: |
- Subpath of git repo to be pulled for DAGs. This allows users to maintain multiple DAG
- repositories in a single git repo.
- version_added: ~
- type: string
- example: ~
- default: ""
- - name: git_sync_rev
- description: |
- The specific rev or hash the git_sync init container will checkout
- This becomes GIT_SYNC_REV environment variable in the git_sync init container for worker pods
- version_added: 1.10.7
- type: string
- example: ~
- default: ""
- - name: git_user
- description: |
- Use git_user and git_password for user authentication or git_ssh_key_secret_name
- and git_ssh_key_secret_key for SSH authentication
- version_added: ~
- type: string
- example: ~
- default: ""
- - name: git_password
- description: ~
- version_added: ~
- type: string
- example: ~
- default: ""
- - name: git_sync_root
- description: ~
- version_added: 1.10.2
- type: string
- example: ~
- default: "/git"
- - name: git_sync_dest
- description: ~
- version_added: 1.10.2
- type: string
- example: ~
- default: "repo"
- - name: git_dags_folder_mount_point
- description: |
- Mount point of the volume if git-sync is being used.
- i.e. {AIRFLOW_HOME}/dags
- version_added: 1.10.2
- type: string
- example: ~
- default: ""
- - name: git_ssh_key_secret_name
- description: |
- To get Git-sync SSH authentication set up follow this format
-
- ``airflow-secrets.yaml``:
-
- .. code-block:: yaml
-
- ---
- apiVersion: v1
- kind: Secret
- metadata:
- name: airflow-secrets
- data:
- # key needs to be gitSshKey
- gitSshKey: <base64_encoded_data>
- version_added: 1.10.3
- type: string
- example: "airflow-secrets"
- default: ""
- - name: git_ssh_known_hosts_configmap_name
- description: |
- To get Git-sync SSH authentication set up follow this format
-
- ``airflow-configmap.yaml``:
-
- .. code-block:: yaml
-
- ---
- apiVersion: v1
- kind: ConfigMap
- metadata:
- name: airflow-configmap
- data:
- known_hosts: |
- github.com ssh-rsa <...>
- airflow.cfg: |
- ...
-
- version_added: 1.10.3
- type: string
- example: "airflow-configmap"
- default: ""
- - name: git_sync_credentials_secret
- description: |
- To give the git_sync init container credentials via a secret, create a secret
- with two fields: GIT_SYNC_USERNAME and GIT_SYNC_PASSWORD (example below) and
- add ``git_sync_credentials_secret = <secret_name>`` to your airflow config under the
- ``kubernetes`` section
-
- Secret Example:
-
- .. code-block:: yaml
-
- ---
- apiVersion: v1
- kind: Secret
- metadata:
- name: git-credentials
- data:
- GIT_SYNC_USERNAME: <base64_encoded_git_username>
- GIT_SYNC_PASSWORD: <base64_encoded_git_password>
- version_added: 1.10.5
- type: string
- example: ~
- default: ""
- - name: git_sync_container_repository
- description: |
- For cloning DAGs from git repositories into volumes: https://github.com/kubernetes/git-sync
- version_added: ~
- type: string
- example: ~
- default: "k8s.gcr.io/git-sync"
- - name: git_sync_container_tag
- description: ~
- version_added: ~
- type: string
- example: ~
- default: "v3.1.1"
- - name: git_sync_init_container_name
- description: ~
- version_added: ~
- type: string
- example: ~
- default: "git-sync-clone"
- - name: git_sync_run_as_user
- description: ~
- version_added: 1.10.5
- type: string
- example: ~
- default: "65533"
- - name: worker_service_account_name
- description: |
- The name of the Kubernetes service account to be associated with airflow workers, if any.
- Service accounts are required for workers that require access to secrets or cluster resources.
- See the Kubernetes RBAC documentation for more:
- https://kubernetes.io/docs/admin/authorization/rbac/
- version_added: ~
- type: string
- example: ~
- default: ""
- - name: image_pull_secrets
- description: |
- Any image pull secrets to be given to worker pods, If more than one secret is
- required, provide a comma separated list: secret_a,secret_b
- version_added: ~
- type: string
- example: ~
- default: ""
- name: in_cluster
description: |
Use the service account kubernetes gives to pods to connect to kubernetes cluster.
@@ -2253,24 +1935,6 @@
type: string
example: ~
default: ~
- - name: affinity
- description: |
- Affinity configuration as a single line formatted JSON object.
- See the affinity model for top-level key names (e.g. ``nodeAffinity``, etc.):
- https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.12/#affinity-v1-core
- version_added: 1.10.2
- type: string
- example: ~
- default: ""
- - name: tolerations
- description: |
- A list of toleration objects as a single line formatted JSON array
- See:
- https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.12/#toleration-v1-core
- version_added: 1.10.2
- type: string
- example: ~
- default: ""
- name: kube_client_request_args
description: |
Keyword parameters to pass while calling a kubernetes client core_v1_api methods
@@ -2293,81 +1957,6 @@
type: string
example: '{"grace_period_seconds": 10}'
default: ""
- - name: run_as_user
- description: |
- Specifies the uid to run the first process of the worker pods containers as
- version_added: 1.10.3
- type: string
- example: ~
- default: ""
- - name: fs_group
- description: |
- Specifies a gid to associate with all containers in the worker pods
- if using a git_ssh_key_secret_name use an fs_group
- that allows for the key to be read, e.g. 65533
- version_added: 1.10.3
- type: string
- example: ~
- default: ""
- - name: worker_annotations
- description: |
- Annotations configuration as a single line formatted JSON object.
- See the naming convention in:
- https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/
- version_added: ~
- type: string
- example: ~
- default: ""
- - name: worker_resources
- description: |
- Worker resources configuration as a single line formatted JSON object.
- version_added: ~
- type: string
- example: '{ "limit_cpu": 0.25, "limit_memory": "64Mi", "request_cpu": "250m", "request_memory": "64Mi"}'
- default: ""
-- name: kubernetes_node_selectors
- description: |
- The Key-value pairs to be given to worker pods.
- The worker pods will be scheduled to the nodes of the specified key-value pairs.
- Should be supplied in the format: key = value
- options: []
-- name: kubernetes_environment_variables
- description: |
- The scheduler sets the following environment variables into your workers. You may define as
- many environment variables as needed and the kubernetes launcher will set them in the launched workers.
- Environment variables in this section are defined as followsa
- ``<environment_variable_key> = <environment_variable_value>``
-
- For example if you wanted to set an environment variable with value `prod` and key
- ``ENVIRONMENT`` you would follow the following format:
- ENVIRONMENT = prod
-
- Additionally you may override worker airflow settings with the ``AIRFLOW__<SECTION>__<KEY>``
- formatting as supported by airflow normally.
- options: []
-- name: kubernetes_secrets
- description: |
- The scheduler mounts the following secrets into your workers as they are launched by the
- scheduler. You may define as many secrets as needed and the kubernetes launcher will parse the
- defined secrets and mount them as secret environment variables in the launched workers.
- Secrets in this section are defined as follows
- ``<environment_variable_mount> = <kubernetes_secret_object>=<kubernetes_secret_key>``
-
- For example if you wanted to mount a kubernetes secret key named ``postgres_password`` from the
- kubernetes secret object ``airflow-secret`` as the environment variable ``POSTGRES_PASSWORD`` into
- your workers you would follow the following format:
- ``POSTGRES_PASSWORD = airflow-secret=postgres_credentials``
-
- Additionally you may override worker airflow settings with the ``AIRFLOW__<SECTION>__<KEY>``
- formatting as supported by airflow normally.
- options: []
-- name: kubernetes_labels
- description: |
- The Key-value pairs to be given to worker pods.
- The worker pods will be given these static labels, as well as some additional dynamic labels
- to identify the task.
- Should be supplied in the format: ``key = value``
- options: []
- name: smart_sensor
description: ~
options:
diff --git a/airflow/config_templates/default_airflow.cfg b/airflow/config_templates/default_airflow.cfg
index bc1c62b..116fbd2 100644
--- a/airflow/config_templates/default_airflow.cfg
+++ b/airflow/config_templates/default_airflow.cfg
@@ -884,17 +884,17 @@ use_ssl = False
verify_certs = True
[kubernetes]
-# The repository of the Kubernetes Image for the Worker to Run
-worker_container_repository =
-
# Path to the YAML pod file. If set, all other kubernetes-related fields are ignored.
pod_template_file =
+# The repository of the Kubernetes Image for the Worker to Run
+worker_container_repository =
+
# The tag of the Kubernetes Image for the Worker to Run
worker_container_tag =
-# The imagePullPolicy of the Kubernetes Image for the Worker to Run
-worker_container_image_pull_policy = IfNotPresent
+# The Kubernetes namespace where airflow workers should be created. Defaults to ``default``
+namespace = default
# If True, all worker pods will be deleted upon termination
delete_worker_pods = True
@@ -910,197 +910,10 @@ delete_worker_pods_on_failure = False
# better performance.
worker_pods_creation_batch_size = 1
-# The Kubernetes namespace where airflow workers should be created. Defaults to ``default``
-namespace = default
-
# Allows users to launch pods in multiple namespaces.
# Will require creating a cluster-role for the scheduler
multi_namespace_mode = False
-# The name of the Kubernetes ConfigMap containing ``airflow.cfg`` file.
-#
-# For example:
-#
-# ``airflow_configmap = "airflow-configmap"`` if you have the following ConfigMap.
-#
-# ``airflow-configmap.yaml``:
-#
-# .. code-block:: yaml
-#
-# ---
-# apiVersion: v1
-# kind: ConfigMap
-# metadata:
-# name: airflow-configmap
-# data:
-# airflow.cfg: |
-# ...
-#
-# Note that the ``airflow_configmap`` and the ``airflow_local_settings_configmap``
-# can be the same configMap
-# Example: airflow_configmap = airflow-configmap
-airflow_configmap =
-
-# The name of the Kubernetes ConfigMap containing ``airflow_local_settings.py`` file.
-#
-# For example:
-#
-# ``airflow_local_settings_configmap = "airflow-configmap"`` if you have the following ConfigMap.
-#
-# ``airflow-configmap.yaml``:
-#
-# .. code-block:: yaml
-#
-# ---
-# apiVersion: v1
-# kind: ConfigMap
-# metadata:
-# name: airflow-configmap
-# data:
-# airflow_local_settings.py: |
-# def pod_mutation_hook(pod):
-# ...
-# airflow.cfg: |
-# ...
-# Example: airflow_local_settings_configmap = airflow-configmap
-airflow_local_settings_configmap =
-
-# For docker image already contains DAGs, this is set to ``True``, and the worker will
-# search for dags in dags_folder, otherwise use git sync or dags volume claim to mount DAGs
-dags_in_image = False
-
-# For either git sync or volume mounted DAGs, the worker will look in this subpath for DAGs
-dags_volume_subpath =
-
-# For either git sync or volume mounted DAGs, the worker will mount the volume in this path
-dags_volume_mount_point =
-
-# For DAGs mounted via a volume claim (mutually exclusive with git-sync and host path)
-dags_volume_claim =
-
-# For volume mounted logs, the worker will look in this subpath for logs
-logs_volume_subpath =
-
-# A shared volume claim for the logs
-logs_volume_claim =
-
-# For DAGs mounted via a hostPath volume (mutually exclusive with volume claim and git-sync)
-# Useful in local environment, discouraged in production
-dags_volume_host =
-
-# A hostPath volume for the logs
-# Useful in local environment, discouraged in production
-logs_volume_host =
-
-# A list of configMapsRefs to envFrom. If more than one configMap is
-# specified, provide a comma separated list: configmap_a,configmap_b
-env_from_configmap_ref =
-
-# A list of secretRefs to envFrom. If more than one secret is
-# specified, provide a comma separated list: secret_a,secret_b
-env_from_secret_ref =
-
-# Git credentials and repository for DAGs mounted via Git (mutually exclusive with volume claim)
-git_repo =
-
-# Branch of the git repo containing DAGs. This can be useful for users who wish to version their DAGs
-# using release branches.
-git_branch =
-
-# Use a shallow clone with a history truncated to the specified number of commits.
-# 0 - do not use shallow clone.
-git_sync_depth = 1
-
-# Subpath of git repo to be pulled for DAGs. This allows users to maintain multiple DAG
-# repositories in a single git repo.
-git_subpath =
-
-# The specific rev or hash the git_sync init container will checkout
-# This becomes GIT_SYNC_REV environment variable in the git_sync init container for worker pods
-git_sync_rev =
-
-# Use git_user and git_password for user authentication or git_ssh_key_secret_name
-# and git_ssh_key_secret_key for SSH authentication
-git_user =
-git_password =
-git_sync_root = /git
-git_sync_dest = repo
-
-# Mount point of the volume if git-sync is being used.
-# i.e. {AIRFLOW_HOME}/dags
-git_dags_folder_mount_point =
-
-# To get Git-sync SSH authentication set up follow this format
-#
-# ``airflow-secrets.yaml``:
-#
-# .. code-block:: yaml
-#
-# ---
-# apiVersion: v1
-# kind: Secret
-# metadata:
-# name: airflow-secrets
-# data:
-# # key needs to be gitSshKey
-# gitSshKey: <base64_encoded_data>
-# Example: git_ssh_key_secret_name = airflow-secrets
-git_ssh_key_secret_name =
-
-# To get Git-sync SSH authentication set up follow this format
-#
-# ``airflow-configmap.yaml``:
-#
-# .. code-block:: yaml
-#
-# ---
-# apiVersion: v1
-# kind: ConfigMap
-# metadata:
-# name: airflow-configmap
-# data:
-# known_hosts: |
-# github.com ssh-rsa <...>
-# airflow.cfg: |
-# ...
-# Example: git_ssh_known_hosts_configmap_name = airflow-configmap
-git_ssh_known_hosts_configmap_name =
-
-# To give the git_sync init container credentials via a secret, create a secret
-# with two fields: GIT_SYNC_USERNAME and GIT_SYNC_PASSWORD (example below) and
-# add ``git_sync_credentials_secret = <secret_name>`` to your airflow config under the
-# ``kubernetes`` section
-#
-# Secret Example:
-#
-# .. code-block:: yaml
-#
-# ---
-# apiVersion: v1
-# kind: Secret
-# metadata:
-# name: git-credentials
-# data:
-# GIT_SYNC_USERNAME: <base64_encoded_git_username>
-# GIT_SYNC_PASSWORD: <base64_encoded_git_password>
-git_sync_credentials_secret =
-
-# For cloning DAGs from git repositories into volumes: https://github.com/kubernetes/git-sync
-git_sync_container_repository = k8s.gcr.io/git-sync
-git_sync_container_tag = v3.1.1
-git_sync_init_container_name = git-sync-clone
-git_sync_run_as_user = 65533
-
-# The name of the Kubernetes service account to be associated with airflow workers, if any.
-# Service accounts are required for workers that require access to secrets or cluster resources.
-# See the Kubernetes RBAC documentation for more:
-# https://kubernetes.io/docs/admin/authorization/rbac/
-worker_service_account_name =
-
-# Any image pull secrets to be given to worker pods, If more than one secret is
-# required, provide a comma separated list: secret_a,secret_b
-image_pull_secrets =
-
# Use the service account kubernetes gives to pods to connect to kubernetes cluster.
# It's intended for clients that expect to be running inside a pod running on kubernetes.
# It will raise an exception if called from a process not running in a kubernetes environment.
@@ -1113,16 +926,6 @@ in_cluster = True
# Path to the kubernetes configfile to be used when ``in_cluster`` is set to False
# config_file =
-# Affinity configuration as a single line formatted JSON object.
-# See the affinity model for top-level key names (e.g. ``nodeAffinity``, etc.):
-# https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.12/#affinity-v1-core
-affinity =
-
-# A list of toleration objects as a single line formatted JSON array
-# See:
-# https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.12/#toleration-v1-core
-tolerations =
-
# Keyword parameters to pass while calling a kubernetes client core_v1_api methods
# from Kubernetes Executor provided as a single line formatted JSON dictionary string.
# List of supported params are similar for all core_v1_apis, hence a single config
@@ -1138,66 +941,6 @@ kube_client_request_args =
# Example: delete_option_kwargs = {{"grace_period_seconds": 10}}
delete_option_kwargs =
-# Specifies the uid to run the first process of the worker pods containers as
-run_as_user =
-
-# Specifies a gid to associate with all containers in the worker pods
-# if using a git_ssh_key_secret_name use an fs_group
-# that allows for the key to be read, e.g. 65533
-fs_group =
-
-# Annotations configuration as a single line formatted JSON object.
-# See the naming convention in:
-# https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/
-worker_annotations =
-
-# Worker resources configuration as a single line formatted JSON object.
-# Example: worker_resources = {{ "limit_cpu": 0.25, "limit_memory": "64Mi", "request_cpu": "250m", "request_memory": "64Mi"}}
-worker_resources =
-
-[kubernetes_node_selectors]
-
-# The Key-value pairs to be given to worker pods.
-# The worker pods will be scheduled to the nodes of the specified key-value pairs.
-# Should be supplied in the format: key = value
-
-[kubernetes_environment_variables]
-
-# The scheduler sets the following environment variables into your workers. You may define as
-# many environment variables as needed and the kubernetes launcher will set them in the launched workers.
-# Environment variables in this section are defined as followsa
-# ``<environment_variable_key> = <environment_variable_value>``
-#
-# For example if you wanted to set an environment variable with value `prod` and key
-# ``ENVIRONMENT`` you would follow the following format:
-# ENVIRONMENT = prod
-#
-# Additionally you may override worker airflow settings with the ``AIRFLOW__<SECTION>__<KEY>``
-# formatting as supported by airflow normally.
-
-[kubernetes_secrets]
-
-# The scheduler mounts the following secrets into your workers as they are launched by the
-# scheduler. You may define as many secrets as needed and the kubernetes launcher will parse the
-# defined secrets and mount them as secret environment variables in the launched workers.
-# Secrets in this section are defined as follows
-# ``<environment_variable_mount> = <kubernetes_secret_object>=<kubernetes_secret_key>``
-#
-# For example if you wanted to mount a kubernetes secret key named ``postgres_password`` from the
-# kubernetes secret object ``airflow-secret`` as the environment variable ``POSTGRES_PASSWORD`` into
-# your workers you would follow the following format:
-# ``POSTGRES_PASSWORD = airflow-secret=postgres_credentials``
-#
-# Additionally you may override worker airflow settings with the ``AIRFLOW__<SECTION>__<KEY>``
-# formatting as supported by airflow normally.
-
-[kubernetes_labels]
-
-# The Key-value pairs to be given to worker pods.
-# The worker pods will be given these static labels, as well as some additional dynamic labels
-# to identify the task.
-# Should be supplied in the format: ``key = value``
-
[smart_sensor]
# When `use_smart_sensor` is True, Airflow redirects multiple qualified sensor tasks to
# smart sensor task.
diff --git a/airflow/example_dags/example_kubernetes_executor_config.py b/airflow/example_dags/example_kubernetes_executor_config.py
index db64b31..cb71a0c 100644
--- a/airflow/example_dags/example_kubernetes_executor_config.py
+++ b/airflow/example_dags/example_kubernetes_executor_config.py
@@ -67,10 +67,11 @@ with DAG(
start_task = PythonOperator(
task_id="start_task",
python_callable=print_stuff,
- executor_config={
- "KubernetesExecutor": {
- "annotations": {"test": "annotation"}
- }
+ executor_config={"pod_override": k8s.V1Pod(
+ metadata=k8s.V1ObjectMeta(
+ annotations={"test": "annotation"}
+ )
+ )
}
)
@@ -148,12 +149,13 @@ with DAG(
third_task = PythonOperator(
task_id="non_root_task",
python_callable=print_stuff,
- executor_config={
- "KubernetesExecutor": {
- "labels": {
+ executor_config={"pod_override": k8s.V1Pod(
+ metadata=k8s.V1ObjectMeta(
+ labels={
"release": "stable"
}
- }
+ )
+ )
}
)
diff --git a/airflow/executors/kubernetes_executor.py b/airflow/executors/kubernetes_executor.py
index 92441d2..9080a3a 100644
--- a/airflow/executors/kubernetes_executor.py
+++ b/airflow/executors/kubernetes_executor.py
@@ -38,13 +38,12 @@ from urllib3.exceptions import ReadTimeoutError
from airflow import settings
from airflow.configuration import conf
-from airflow.exceptions import AirflowConfigException, AirflowException
+from airflow.exceptions import AirflowException
from airflow.executors.base_executor import NOT_STARTED_MESSAGE, BaseExecutor, CommandType
from airflow.kubernetes import pod_generator
from airflow.kubernetes.kube_client import get_kube_client
from airflow.kubernetes.pod_generator import MAX_POD_ID_LEN, PodGenerator
from airflow.kubernetes.pod_launcher import PodLauncher
-from airflow.kubernetes.worker_configuration import WorkerConfiguration
from airflow.models import KubeResourceVersion, KubeWorkerIdentifier, TaskInstance
from airflow.models.taskinstance import TaskInstanceKey
from airflow.utils.log.logging_mixin import LoggingMixin
@@ -70,116 +69,24 @@ class KubeConfig: # pylint: disable=too-many-instance-attributes
def __init__(self): # pylint: disable=too-many-statements
configuration_dict = conf.as_dict(display_sensitive=True)
self.core_configuration = configuration_dict['core']
- self.kube_secrets = configuration_dict.get('kubernetes_secrets', {})
- self.kube_env_vars = configuration_dict.get('kubernetes_environment_variables', {})
- self.env_from_configmap_ref = conf.get(self.kubernetes_section,
- 'env_from_configmap_ref')
- self.env_from_secret_ref = conf.get(self.kubernetes_section,
- 'env_from_secret_ref')
self.airflow_home = settings.AIRFLOW_HOME
self.dags_folder = conf.get(self.core_section, 'dags_folder')
self.parallelism = conf.getint(self.core_section, 'parallelism')
- self.worker_container_repository = conf.get(
- self.kubernetes_section, 'worker_container_repository')
- self.worker_container_tag = conf.get(
- self.kubernetes_section, 'worker_container_tag')
- self.kube_image = '{}:{}'.format(
- self.worker_container_repository, self.worker_container_tag)
- self.kube_image_pull_policy = conf.get(
- self.kubernetes_section, "worker_container_image_pull_policy"
- )
- self.kube_node_selectors = configuration_dict.get('kubernetes_node_selectors', {})
self.pod_template_file = conf.get(self.kubernetes_section, 'pod_template_file',
fallback=None)
- kube_worker_annotations = conf.get(self.kubernetes_section, 'worker_annotations')
- if kube_worker_annotations:
- self.kube_annotations = json.loads(kube_worker_annotations)
- else:
- self.kube_annotations = None
-
- self.kube_labels = configuration_dict.get('kubernetes_labels', {})
self.delete_worker_pods = conf.getboolean(
self.kubernetes_section, 'delete_worker_pods')
self.delete_worker_pods_on_failure = conf.getboolean(
self.kubernetes_section, 'delete_worker_pods_on_failure')
self.worker_pods_creation_batch_size = conf.getint(
self.kubernetes_section, 'worker_pods_creation_batch_size')
- self.worker_service_account_name = conf.get(
- self.kubernetes_section, 'worker_service_account_name')
- self.image_pull_secrets = conf.get(self.kubernetes_section, 'image_pull_secrets')
-
- # NOTE: user can build the dags into the docker image directly,
- # this will set to True if so
- self.dags_in_image = conf.getboolean(self.kubernetes_section, 'dags_in_image')
- # Run as user for pod security context
- self.worker_run_as_user = self._get_security_context_val('run_as_user')
- self.worker_fs_group = self._get_security_context_val('fs_group')
-
- kube_worker_resources = conf.get(self.kubernetes_section, 'worker_resources')
- if kube_worker_resources:
- self.worker_resources = json.loads(kube_worker_resources)
- else:
- self.worker_resources = None
-
- # NOTE: `git_repo` and `git_branch` must be specified together as a pair
- # The http URL of the git repository to clone from
- self.git_repo = conf.get(self.kubernetes_section, 'git_repo')
- # The branch of the repository to be checked out
- self.git_branch = conf.get(self.kubernetes_section, 'git_branch')
- # Clone depth for git sync
- self.git_sync_depth = conf.get(self.kubernetes_section, 'git_sync_depth')
- # Optionally, the directory in the git repository containing the dags
- self.git_subpath = conf.get(self.kubernetes_section, 'git_subpath')
- # Optionally, the root directory for git operations
- self.git_sync_root = conf.get(self.kubernetes_section, 'git_sync_root')
- # Optionally, the name at which to publish the checked-out files under --root
- self.git_sync_dest = conf.get(self.kubernetes_section, 'git_sync_dest')
- # Optionally, the tag or hash to checkout
- self.git_sync_rev = conf.get(self.kubernetes_section, 'git_sync_rev')
- # Optionally, if git_dags_folder_mount_point is set the worker will use
- # {git_dags_folder_mount_point}/{git_sync_dest}/{git_subpath} as dags_folder
- self.git_dags_folder_mount_point = conf.get(self.kubernetes_section,
- 'git_dags_folder_mount_point')
-
- # Optionally a user may supply a (`git_user` AND `git_password`) OR
- # (`git_ssh_key_secret_name` AND `git_ssh_key_secret_key`) for private repositories
- self.git_user = conf.get(self.kubernetes_section, 'git_user')
- self.git_password = conf.get(self.kubernetes_section, 'git_password')
- self.git_ssh_key_secret_name = conf.get(self.kubernetes_section, 'git_ssh_key_secret_name')
- self.git_ssh_known_hosts_configmap_name = conf.get(self.kubernetes_section,
- 'git_ssh_known_hosts_configmap_name')
- self.git_sync_credentials_secret = conf.get(self.kubernetes_section,
- 'git_sync_credentials_secret')
-
- # NOTE: The user may optionally use a volume claim to mount a PV containing
- # DAGs directly
- self.dags_volume_claim = conf.get(self.kubernetes_section, 'dags_volume_claim')
-
- self.dags_volume_mount_point = conf.get(self.kubernetes_section, 'dags_volume_mount_point')
-
- # This prop may optionally be set for PV Claims and is used to write logs
- self.logs_volume_claim = conf.get(self.kubernetes_section, 'logs_volume_claim')
-
- # This prop may optionally be set for PV Claims and is used to locate DAGs
- # on a SubPath
- self.dags_volume_subpath = conf.get(
- self.kubernetes_section, 'dags_volume_subpath')
-
- # This prop may optionally be set for PV Claims and is used to locate logs
- # on a SubPath
- self.logs_volume_subpath = conf.get(
- self.kubernetes_section, 'logs_volume_subpath')
-
- # Optionally, hostPath volume containing DAGs
- self.dags_volume_host = conf.get(self.kubernetes_section, 'dags_volume_host')
-
- # Optionally, write logs to a hostPath Volume
- self.logs_volume_host = conf.get(self.kubernetes_section, 'logs_volume_host')
-
- # This prop may optionally be set for PV Claims and is used to write logs
- self.base_log_folder = conf.get(self.logging_section, 'base_log_folder')
+ self.worker_container_repository = conf.get(
+ self.kubernetes_section, 'worker_container_repository')
+ self.worker_container_tag = conf.get(
+ self.kubernetes_section, 'worker_container_tag')
+ self.kube_image = f'{self.worker_container_repository}:{self.worker_container_tag}'
# The Kubernetes Namespace in which the Scheduler and Webserver reside. Note
# that if your
@@ -193,42 +100,6 @@ class KubeConfig: # pylint: disable=too-many-instance-attributes
# interact with cluster components.
self.executor_namespace = conf.get(self.kubernetes_section, 'namespace')
- # If the user is using the git-sync container to clone their repository via git,
- # allow them to specify repository, tag, and pod name for the init container.
- self.git_sync_container_repository = conf.get(
- self.kubernetes_section, 'git_sync_container_repository')
-
- self.git_sync_container_tag = conf.get(
- self.kubernetes_section, 'git_sync_container_tag')
- self.git_sync_container = '{}:{}'.format(
- self.git_sync_container_repository, self.git_sync_container_tag)
-
- self.git_sync_init_container_name = conf.get(
- self.kubernetes_section, 'git_sync_init_container_name')
-
- self.git_sync_run_as_user = self._get_security_context_val('git_sync_run_as_user')
-
- # The worker pod may optionally have a valid Airflow config loaded via a
- # configmap
- self.airflow_configmap = conf.get(self.kubernetes_section, 'airflow_configmap')
-
- # The worker pod may optionally have a valid Airflow local settings loaded via a
- # configmap
- self.airflow_local_settings_configmap = conf.get(
- self.kubernetes_section, 'airflow_local_settings_configmap')
-
- affinity_json = conf.get(self.kubernetes_section, 'affinity')
- if affinity_json:
- self.kube_affinity = json.loads(affinity_json)
- else:
- self.kube_affinity = None
-
- tolerations_json = conf.get(self.kubernetes_section, 'tolerations')
- if tolerations_json:
- self.kube_tolerations = json.loads(tolerations_json)
- else:
- self.kube_tolerations = None
-
kube_client_request_args = conf.get(self.kubernetes_section, 'kube_client_request_args')
if kube_client_request_args:
self.kube_client_request_args = json.loads(kube_client_request_args)
@@ -238,8 +109,6 @@ class KubeConfig: # pylint: disable=too-many-instance-attributes
tuple(self.kube_client_request_args['_request_timeout'])
else:
self.kube_client_request_args = {}
- self._validate()
-
delete_option_kwargs = conf.get(self.kubernetes_section, 'delete_option_kwargs')
if delete_option_kwargs:
self.delete_option_kwargs = json.loads(delete_option_kwargs)
@@ -255,32 +124,6 @@ class KubeConfig: # pylint: disable=too-many-instance-attributes
else:
return int(val)
- def _validate(self):
- if self.pod_template_file:
- return
- # TODO: use XOR for dags_volume_claim and git_dags_folder_mount_point
- # pylint: disable=too-many-boolean-expressions
- if not self.dags_volume_claim \
- and not self.dags_volume_host \
- and not self.dags_in_image \
- and (not self.git_repo or not self.git_branch or not self.git_dags_folder_mount_point):
- raise AirflowConfigException(
- 'In kubernetes mode the following must be set in the `kubernetes` '
- 'config section: `dags_volume_claim` '
- 'or `dags_volume_host` '
- 'or `dags_in_image` '
- 'or `git_repo and git_branch and git_dags_folder_mount_point`')
- if self.git_repo \
- and (self.git_user or self.git_password) \
- and self.git_ssh_key_secret_name:
- raise AirflowConfigException(
- 'In kubernetes mode, using `git_repo` to pull the DAGs: '
- 'for private repositories, either `git_user` and `git_password` '
- 'must be set for authentication through user credentials; '
- 'or `git_ssh_key_secret_name` must be set for authentication '
- 'through ssh key, but not both')
- # pylint: enable=too-many-boolean-expressions
-
class KubernetesJobWatcher(multiprocessing.Process, LoggingMixin):
"""Watches for Kubernetes jobs"""
@@ -443,7 +286,6 @@ class AirflowKubernetesScheduler(LoggingMixin):
self.log.debug("Kubernetes using namespace %s", self.namespace)
self.kube_client = kube_client
self.launcher = PodLauncher(kube_client=self.kube_client)
- self.worker_configuration_pod = WorkerConfiguration(kube_config=self.kube_config).as_pod()
self._manager = multiprocessing.Manager()
self.watcher_queue = self._manager.Queue()
self.worker_uuid = worker_uuid
@@ -483,18 +325,23 @@ class AirflowKubernetesScheduler(LoggingMixin):
if command[0:3] != ["airflow", "tasks", "run"]:
raise ValueError('The command must start with ["airflow", "tasks", "run"].')
+ base_worker_pod = PodGenerator.deserialize_model_file(self.kube_config.pod_template_file)
+ if not base_worker_pod:
+ raise AirflowException("could not find a valid worker template yaml at {}"
+ .format(self.kube_config.pod_template_file))
+
pod = PodGenerator.construct_pod(
namespace=self.namespace,
worker_uuid=self.worker_uuid,
pod_id=self._create_pod_id(dag_id, task_id),
dag_id=dag_id,
task_id=task_id,
- try_number=try_number,
kube_image=self.kube_config.kube_image,
+ try_number=try_number,
date=execution_date,
command=command,
pod_override_object=kube_executor_config,
- base_worker_pod=self.worker_configuration_pod
+ base_worker_pod=base_worker_pod
)
# Reconcile the pod generated by the Operator and the Pod
# generated by the .cfg file
@@ -801,9 +648,14 @@ class KubernetesExecutor(BaseExecutor, LoggingMixin):
try:
self.kube_scheduler.run_next(task)
except ApiException as e:
- self.log.warning('ApiException when attempting to run task, re-queueing. '
- 'Message: %s', json.loads(e.body)['message'])
- self.task_queue.put(task)
+ if e.reason == "BadRequest":
+ self.log.error("Request was invalid. Failing task")
+ key, _, _ = task
+ self.change_state(key, State.FAILED, e)
+ else:
+ self.log.warning('ApiException when attempting to run task, re-queueing. '
+ 'Message: %s', json.loads(e.body)['message'])
+ self.task_queue.put(task)
finally:
self.task_queue.task_done()
except Empty:
diff --git a/airflow/kubernetes/pod.py b/airflow/kubernetes/pod.py
deleted file mode 100644
index dcd904d..0000000
--- a/airflow/kubernetes/pod.py
+++ /dev/null
@@ -1,127 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-"""
-Classes for interacting with Kubernetes API
-"""
-import copy
-
-from kubernetes.client import models as k8s
-
-from airflow.kubernetes.k8s_model import K8SModel
-
-
-class Resources(K8SModel):
- """
- Stores information about resources used by the Pod.
-
- :param request_memory: requested memory
- :type request_memory: str
- :param request_cpu: requested CPU number
- :type request_cpu: float | str
- :param request_ephemeral_storage: requested ephermeral storage
- :type request_ephemeral_storage: str
- :param limit_memory: limit for memory usage
- :type limit_memory: str
- :param limit_cpu: Limit for CPU used
- :type limit_cpu: float | str
- :param limit_gpu: Limits for GPU used
- :type limit_gpu: int
- :param limit_ephemeral_storage: Limit for ephermeral storage
- :type limit_ephemeral_storage: float | str
- """
- __slots__ = ('request_memory', 'request_cpu', 'limit_memory', 'limit_cpu', 'limit_gpu')
-
- def __init__(
- self,
- request_memory=None,
- request_cpu=None,
- request_ephemeral_storage=None,
- limit_memory=None,
- limit_cpu=None,
- limit_gpu=None,
- limit_ephemeral_storage=None):
- self.request_memory = request_memory
- self.request_cpu = request_cpu
- self.request_ephemeral_storage = request_ephemeral_storage
- self.limit_memory = limit_memory
- self.limit_cpu = limit_cpu
- self.limit_gpu = limit_gpu
- self.limit_ephemeral_storage = limit_ephemeral_storage
-
- def is_empty_resource_request(self):
- """Whether resource is empty"""
- return not self.has_limits() and not self.has_requests()
-
- def has_limits(self):
- """Whether resource has limits"""
- return self.limit_cpu is not None or \
- self.limit_memory is not None or \
- self.limit_gpu is not None or \
- self.limit_ephemeral_storage is not None
-
- def has_requests(self):
- """Whether resource has requests"""
- return self.request_cpu is not None or \
- self.request_memory is not None or \
- self.request_ephemeral_storage is not None
-
- def to_k8s_client_obj(self) -> k8s.V1ResourceRequirements:
- """Converts to k8s client object"""
- return k8s.V1ResourceRequirements(
- limits={
- 'cpu': self.limit_cpu,
- 'memory': self.limit_memory,
- 'nvidia.com/gpu': self.limit_gpu,
- 'ephemeral-storage': self.limit_ephemeral_storage
- },
- requests={
- 'cpu': self.request_cpu,
- 'memory': self.request_memory,
- 'ephemeral-storage': self.request_ephemeral_storage}
- )
-
- def attach_to_pod(self, pod: k8s.V1Pod) -> k8s.V1Pod:
- """Attaches to pod"""
- cp_pod = copy.deepcopy(pod)
- resources = self.to_k8s_client_obj()
- cp_pod.spec.containers[0].resources = resources
- return cp_pod
-
-
-class Port(K8SModel):
- """POD port"""
- __slots__ = ('name', 'container_port')
-
- def __init__(
- self,
- name=None,
- container_port=None):
- """Creates port"""
- self.name = name
- self.container_port = container_port
-
- def to_k8s_client_obj(self) -> k8s.V1ContainerPort:
- """Converts to k8s client object"""
- return k8s.V1ContainerPort(name=self.name, container_port=self.container_port)
-
- def attach_to_pod(self, pod: k8s.V1Pod) -> k8s.V1Pod:
- """Attaches to pod"""
- cp_pod = copy.deepcopy(pod)
- port = self.to_k8s_client_obj()
- cp_pod.spec.containers[0].ports = cp_pod.spec.containers[0].ports or []
- cp_pod.spec.containers[0].ports.append(port)
- return cp_pod
diff --git a/airflow/kubernetes/pod_generator.py b/airflow/kubernetes/pod_generator.py
index d17f029..21d4ff2 100644
--- a/airflow/kubernetes/pod_generator.py
+++ b/airflow/kubernetes/pod_generator.py
@@ -23,13 +23,12 @@ is supported and no serialization need be written.
import copy
import datetime
import hashlib
-import inspect
import os
import re
import uuid
import warnings
from functools import reduce
-from typing import Dict, List, Optional, Union
+from typing import List, Optional, Union
import yaml
from dateutil import parser
@@ -37,6 +36,7 @@ from kubernetes.client import models as k8s
from kubernetes.client.api_client import ApiClient
from airflow.exceptions import AirflowConfigException
+from airflow.kubernetes.pod_generator_deprecated import PodGenerator as PodGeneratorDeprecated
from airflow.version import version as airflow_version
MAX_POD_ID_LEN = 253
@@ -123,173 +123,31 @@ class PodGenerator:
Any configuration that is container specific gets applied to
the first container in the list of containers.
- :param image: The docker image
- :type image: Optional[str]
- :param name: name in the metadata section (not the container name)
- :type name: Optional[str]
- :param namespace: pod namespace
- :type namespace: Optional[str]
- :param volume_mounts: list of kubernetes volumes mounts
- :type volume_mounts: Optional[List[Union[k8s.V1VolumeMount, dict]]]
- :param envs: A dict containing the environment variables
- :type envs: Optional[Dict[str, str]]
- :param cmds: The command to be run on the first container
- :type cmds: Optional[List[str]]
- :param args: The arguments to be run on the pod
- :type args: Optional[List[str]]
- :param labels: labels for the pod metadata
- :type labels: Optional[Dict[str, str]]
- :param node_selectors: node selectors for the pod
- :type node_selectors: Optional[Dict[str, str]]
- :param ports: list of ports. Applies to the first container.
- :type ports: Optional[List[Union[k8s.V1ContainerPort, dict]]]
- :param volumes: Volumes to be attached to the first container
- :type volumes: Optional[List[Union[k8s.V1Volume, dict]]]
- :param image_pull_policy: Specify a policy to cache or always pull an image
- :type image_pull_policy: str
- :param restart_policy: The restart policy of the pod
- :type restart_policy: str
- :param image_pull_secrets: Any image pull secrets to be given to the pod.
- If more than one secret is required, provide a comma separated list:
- secret_a,secret_b
- :type image_pull_secrets: str
- :param init_containers: A list of init containers
- :type init_containers: Optional[List[k8s.V1Container]]
- :param service_account_name: Identity for processes that run in a Pod
- :type service_account_name: Optional[str]
- :param resources: Resource requirements for the first containers
- :type resources: Optional[Union[k8s.V1ResourceRequirements, dict]]
- :param annotations: annotations for the pod
- :type annotations: Optional[Dict[str, str]]
- :param affinity: A dict containing a group of affinity scheduling rules
- :type affinity: Optional[dict]
- :param hostnetwork: If True enable host networking on the pod
- :type hostnetwork: bool
- :param tolerations: A list of kubernetes tolerations
- :type tolerations: Optional[list]
- :param security_context: A dict containing the security context for the pod
- :type security_context: Optional[Union[k8s.V1PodSecurityContext, dict]]
- :param configmaps: Any configmap refs to envfrom.
- If more than one configmap is required, provide a comma separated list
- configmap_a,configmap_b
- :type configmaps: List[str]
- :param dnspolicy: Specify a dnspolicy for the pod
- :type dnspolicy: Optional[str]
- :param schedulername: Specify a schedulername for the pod
- :type schedulername: Optional[str]
:param pod: The fully specified pod. Mutually exclusive with `path_or_string`
:type pod: Optional[kubernetes.client.models.V1Pod]
:param pod_template_file: Path to YAML file. Mutually exclusive with `pod`
:type pod_template_file: Optional[str]
:param extract_xcom: Whether to bring up a container for xcom
:type extract_xcom: bool
- :param priority_class_name: priority class name for the launched Pod
- :type priority_class_name: str
"""
def __init__( # pylint: disable=too-many-arguments,too-many-locals
self,
- image: Optional[str] = None,
- name: Optional[str] = None,
- namespace: Optional[str] = None,
- volume_mounts: Optional[List[Union[k8s.V1VolumeMount, dict]]] = None,
- envs: Optional[Dict[str, str]] = None,
- cmds: Optional[List[str]] = None,
- args: Optional[List[str]] = None,
- labels: Optional[Dict[str, str]] = None,
- node_selectors: Optional[Dict[str, str]] = None,
- ports: Optional[List[Union[k8s.V1ContainerPort, dict]]] = None,
- volumes: Optional[List[Union[k8s.V1Volume, dict]]] = None,
- image_pull_policy: Optional[str] = None,
- restart_policy: Optional[str] = None,
- image_pull_secrets: Optional[str] = None,
- init_containers: Optional[List[k8s.V1Container]] = None,
- service_account_name: Optional[str] = None,
- resources: Optional[Union[k8s.V1ResourceRequirements, dict]] = None,
- annotations: Optional[Dict[str, str]] = None,
- affinity: Optional[dict] = None,
- hostnetwork: bool = False,
- tolerations: Optional[list] = None,
- security_context: Optional[Union[k8s.V1PodSecurityContext, dict]] = None,
- configmaps: Optional[List[str]] = None,
- dnspolicy: Optional[str] = None,
- schedulername: Optional[str] = None,
pod: Optional[k8s.V1Pod] = None,
pod_template_file: Optional[str] = None,
- extract_xcom: bool = False,
- priority_class_name: Optional[str] = None,
+ extract_xcom: bool = True
):
- self.validate_pod_generator_args(locals())
+ if not pod_template_file and not pod:
+ raise AirflowConfigException("Podgenerator requires either a "
+ "`pod` or a `pod_template_file` argument")
+ if pod_template_file and pod:
+ raise AirflowConfigException("Cannot pass both `pod` "
+ "and `pod_template_file` arguments")
if pod_template_file:
self.ud_pod = self.deserialize_model_file(pod_template_file)
else:
self.ud_pod = pod
- self.pod = k8s.V1Pod()
- self.pod.api_version = 'v1'
- self.pod.kind = 'Pod'
-
- # Pod Metadata
- self.metadata = k8s.V1ObjectMeta()
- self.metadata.labels = labels
- self.metadata.name = name
- self.metadata.namespace = namespace
- self.metadata.annotations = annotations
-
- # Pod Container
- self.container = k8s.V1Container(name='base')
- self.container.image = image
- self.container.env = []
-
- if envs:
- if isinstance(envs, dict):
- for key, val in envs.items():
- self.container.env.append(k8s.V1EnvVar(
- name=key,
- value=val
- ))
- elif isinstance(envs, list):
- self.container.env.extend(envs)
-
- configmaps = configmaps or []
- self.container.env_from = []
- for configmap in configmaps:
- self.container.env_from.append(k8s.V1EnvFromSource(
- config_map_ref=k8s.V1ConfigMapEnvSource(
- name=configmap
- )
- ))
-
- self.container.command = cmds or []
- self.container.args = args or []
- self.container.image_pull_policy = image_pull_policy
- self.container.ports = ports or []
- self.container.resources = resources
- self.container.volume_mounts = volume_mounts or []
-
- # Pod Spec
- self.spec = k8s.V1PodSpec(containers=[])
- self.spec.security_context = security_context
- self.spec.tolerations = tolerations
- self.spec.dns_policy = dnspolicy
- self.spec.scheduler_name = schedulername
- self.spec.host_network = hostnetwork
- self.spec.affinity = affinity
- self.spec.service_account_name = service_account_name
- self.spec.init_containers = init_containers
- self.spec.volumes = volumes or []
- self.spec.node_selector = node_selectors
- self.spec.restart_policy = restart_policy
- self.spec.priority_class_name = priority_class_name
-
- self.spec.image_pull_secrets = []
-
- if image_pull_secrets:
- for image_pull_secret in image_pull_secrets.split(','):
- self.spec.image_pull_secrets.append(k8s.V1LocalObjectReference(
- name=image_pull_secret
- ))
-
# Attach sidecar
self.extract_xcom = extract_xcom
@@ -297,21 +155,15 @@ class PodGenerator:
"""Generates pod"""
result = self.ud_pod
- if result is None:
- result = self.pod
- result.spec = self.spec
- result.metadata = self.metadata
- result.spec.containers = [self.container]
-
result.metadata.name = self.make_unique_pod_id(result.metadata.name)
if self.extract_xcom:
- result = self.add_sidecar(result)
+ result = self.add_xcom_sidecar(result)
return result
@staticmethod
- def add_sidecar(pod: k8s.V1Pod) -> k8s.V1Pod:
+ def add_xcom_sidecar(pod: k8s.V1Pod) -> k8s.V1Pod:
"""Adds sidecar"""
pod_cp = copy.deepcopy(pod)
pod_cp.spec.volumes = pod.spec.volumes or []
@@ -323,7 +175,7 @@ class PodGenerator:
return pod_cp
@staticmethod
- def from_obj(obj) -> Optional[k8s.V1Pod]:
+ def from_obj(obj) -> Optional[Union[dict, k8s.V1Pod]]:
"""Converts to pod from obj"""
if obj is None:
@@ -359,14 +211,6 @@ class PodGenerator:
if obj is None:
return None
- if isinstance(obj, PodGenerator):
- return obj.gen_pod()
-
- if not isinstance(obj, dict):
- raise TypeError(
- 'Cannot convert a non-dictionary or non-PodGenerator '
- 'object into a KubernetesExecutorConfig')
-
# We do not want to extract constant here from ExecutorLoader because it is just
# A name in dictionary rather than executor selection mechanism and it causes cyclic import
namespaced = obj.get("KubernetesExecutor", {})
@@ -396,7 +240,7 @@ class PodGenerator:
limits=limits
)
namespaced['resources'] = resources
- return PodGenerator(**namespaced).gen_pod()
+ return PodGeneratorDeprecated(**namespaced).gen_pod()
@staticmethod
def reconcile_pods(base_pod: k8s.V1Pod, client_pod: Optional[k8s.V1Pod]) -> k8s.V1Pod:
@@ -520,27 +364,43 @@ class PodGenerator:
- executor_config
- dynamic arguments
"""
- dynamic_pod = PodGenerator(
- namespace=namespace,
- image=kube_image,
- labels={
- 'airflow-worker': worker_uuid,
- 'dag_id': make_safe_label_value(dag_id),
- 'task_id': make_safe_label_value(task_id),
- 'execution_date': datetime_to_label_safe_datestring(date),
- 'try_number': str(try_number),
- 'airflow_version': airflow_version.replace('+', '-'),
- 'kubernetes_executor': 'True',
- },
- annotations={
- 'dag_id': dag_id,
- 'task_id': task_id,
- 'execution_date': date.isoformat(),
- 'try_number': str(try_number),
- },
- cmds=command,
- name=pod_id
- ).gen_pod()
+ try:
+ image = pod_override_object.spec.containers[0].image # type: ignore
+ if not image:
+ image = kube_image
+ except Exception: # pylint: disable=W0703
+ image = kube_image
+
+ dynamic_pod = k8s.V1Pod(
+ metadata=k8s.V1ObjectMeta(
+ namespace=namespace,
+ annotations={
+ 'dag_id': dag_id,
+ 'task_id': task_id,
+ 'execution_date': date.isoformat(),
+ 'try_number': str(try_number),
+ },
+ name=PodGenerator.make_unique_pod_id(pod_id),
+ labels={
+ 'airflow-worker': worker_uuid,
+ 'dag_id': dag_id,
+ 'task_id': task_id,
+ 'execution_date': datetime_to_label_safe_datestring(date),
+ 'try_number': str(try_number),
+ 'airflow_version': airflow_version.replace('+', '-'),
+ 'kubernetes_executor': 'True',
+ }
+ ),
+ spec=k8s.V1PodSpec(
+ containers=[
+ k8s.V1Container(
+ name="base",
+ command=command,
+ image=image,
+ )
+ ]
+ )
+ )
# Reconcile the pods starting with the first chronologically,
# Pod from the pod_template_File -> Pod from executor_config arg -> Pod from the K8s executor
@@ -551,7 +411,9 @@ class PodGenerator:
@staticmethod
def serialize_pod(pod: k8s.V1Pod):
"""
+
Converts a k8s.V1Pod into a jsonified object
+
@param pod:
@return:
"""
@@ -589,65 +451,23 @@ class PodGenerator:
pod_dict, k8s.V1Pod)
@staticmethod
- def make_unique_pod_id(dag_id):
+ def make_unique_pod_id(pod_id):
r"""
Kubernetes pod names must be <= 253 chars and must pass the following regex for
validation
``^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$``
- :param dag_id: a dag_id with only alphanumeric characters
+ :param pod_id: a dag_id with only alphanumeric characters
:return: ``str`` valid Pod name of appropriate length
"""
- if not dag_id:
+ if not pod_id:
return None
safe_uuid = uuid.uuid4().hex
- safe_pod_id = dag_id[:MAX_POD_ID_LEN - len(safe_uuid) - 1] + "-" + safe_uuid
+ safe_pod_id = pod_id[:MAX_POD_ID_LEN - len(safe_uuid) - 1] + "-" + safe_uuid
return safe_pod_id
- @staticmethod
- def validate_pod_generator_args(given_args):
- """
- :param given_args: The arguments passed to the PodGenerator constructor.
- :type given_args: dict
- :return: None
-
- Validate that if `pod` or `pod_template_file` are set that the user is not attempting
- to configure the pod with the other arguments.
- """
- pod_args = list(inspect.signature(PodGenerator).parameters.items())
-
- def predicate(k, v):
- """
- :param k: an arg to PodGenerator
- :type k: string
- :param v: the parameter of the given arg
- :type v: inspect.Parameter
- :return: bool
-
- returns True if the PodGenerator argument has no default arguments
- or the default argument is None, and it is not one of the listed field
- in `non_empty_fields`.
- """
- non_empty_fields = {
- 'pod', 'pod_template_file', 'extract_xcom', 'service_account_name', 'image_pull_policy',
- 'restart_policy'
- }
-
- return (v.default is None or v.default is v.empty) and k not in non_empty_fields
-
- args_without_defaults = {k: given_args[k] for k, v in pod_args if predicate(k, v) and given_args[k]}
-
- if given_args['pod'] and given_args['pod_template_file']:
- raise AirflowConfigException("Cannot pass both `pod` and `pod_template_file` arguments")
- if args_without_defaults and (given_args['pod'] or given_args['pod_template_file']):
- raise AirflowConfigException(
- "Cannot configure pod and pass either `pod` or `pod_template_file`. Fields {} passed.".format(
- list(args_without_defaults.keys())
- )
- )
-
def merge_objects(base_obj, client_obj):
"""
diff --git a/airflow/kubernetes/pod_generator_deprecated.py b/airflow/kubernetes/pod_generator_deprecated.py
new file mode 100644
index 0000000..984fd96
--- /dev/null
+++ b/airflow/kubernetes/pod_generator_deprecated.py
@@ -0,0 +1,341 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+This module provides an interface between the previous Pod
+API and outputs a kubernetes.client.models.V1Pod.
+The advantage being that the full Kubernetes API
+is supported and no serialization need be written.
+"""
+import copy
+import hashlib
+import re
+import uuid
+from typing import Dict, List, Optional, Union
+
+from kubernetes.client import models as k8s
+
+MAX_POD_ID_LEN = 253
+
+MAX_LABEL_LEN = 63
+
+
+class PodDefaults:
+ """
+ Static defaults for Pods
+ """
+ XCOM_MOUNT_PATH = '/airflow/xcom'
+ SIDECAR_CONTAINER_NAME = 'airflow-xcom-sidecar'
+ XCOM_CMD = 'trap "exit 0" INT; while true; do sleep 30; done;'
+ VOLUME_MOUNT = k8s.V1VolumeMount(
+ name='xcom',
+ mount_path=XCOM_MOUNT_PATH
+ )
+ VOLUME = k8s.V1Volume(
+ name='xcom',
+ empty_dir=k8s.V1EmptyDirVolumeSource()
+ )
+ SIDECAR_CONTAINER = k8s.V1Container(
+ name=SIDECAR_CONTAINER_NAME,
+ command=['sh', '-c', XCOM_CMD],
+ image='alpine',
+ volume_mounts=[VOLUME_MOUNT],
+ resources=k8s.V1ResourceRequirements(
+ requests={
+ "cpu": "1m",
+ }
+ ),
+ )
+
+
+def make_safe_label_value(string):
+ """
+ Valid label values must be 63 characters or less and must be empty or begin and
+ end with an alphanumeric character ([a-z0-9A-Z]) with dashes (-), underscores (_),
+ dots (.), and alphanumerics between.
+
+ If the label value is greater than 63 chars once made safe, or differs in any
+ way from the original value sent to this function, then we need to truncate to
+ 53 chars, and append it with a unique hash.
+ """
+ safe_label = re.sub(r"^[^a-z0-9A-Z]*|[^a-zA-Z0-9_\-\.]|[^a-z0-9A-Z]*$", "", string)
+
+ if len(safe_label) > MAX_LABEL_LEN or string != safe_label:
+ safe_hash = hashlib.md5(string.encode()).hexdigest()[:9]
+ safe_label = safe_label[:MAX_LABEL_LEN - len(safe_hash) - 1] + "-" + safe_hash
+
+ return safe_label
+
+
+class PodGenerator:
+ """
+ Contains Kubernetes Airflow Worker configuration logic
+
+ Represents a kubernetes pod and manages execution of a single pod.
+ Any configuration that is container specific gets applied to
+ the first container in the list of containers.
+
+ :param image: The docker image
+ :type image: Optional[str]
+ :param name: name in the metadata section (not the container name)
+ :type name: Optional[str]
+ :param namespace: pod namespace
+ :type namespace: Optional[str]
+ :param volume_mounts: list of kubernetes volumes mounts
+ :type volume_mounts: Optional[List[Union[k8s.V1VolumeMount, dict]]]
+ :param envs: A dict containing the environment variables
+ :type envs: Optional[Dict[str, str]]
+ :param cmds: The command to be run on the first container
+ :type cmds: Optional[List[str]]
+ :param args: The arguments to be run on the pod
+ :type args: Optional[List[str]]
+ :param labels: labels for the pod metadata
+ :type labels: Optional[Dict[str, str]]
+ :param node_selectors: node selectors for the pod
+ :type node_selectors: Optional[Dict[str, str]]
+ :param ports: list of ports. Applies to the first container.
+ :type ports: Optional[List[Union[k8s.V1ContainerPort, dict]]]
+ :param volumes: Volumes to be attached to the first container
+ :type volumes: Optional[List[Union[k8s.V1Volume, dict]]]
+ :param image_pull_policy: Specify a policy to cache or always pull an image
+ :type image_pull_policy: str
+ :param restart_policy: The restart policy of the pod
+ :type restart_policy: str
+ :param image_pull_secrets: Any image pull secrets to be given to the pod.
+ If more than one secret is required, provide a comma separated list:
+ secret_a,secret_b
+ :type image_pull_secrets: str
+ :param init_containers: A list of init containers
+ :type init_containers: Optional[List[k8s.V1Container]]
+ :param service_account_name: Identity for processes that run in a Pod
+ :type service_account_name: Optional[str]
+ :param resources: Resource requirements for the first containers
+ :type resources: Optional[Union[k8s.V1ResourceRequirements, dict]]
+ :param annotations: annotations for the pod
+ :type annotations: Optional[Dict[str, str]]
+ :param affinity: A dict containing a group of affinity scheduling rules
+ :type affinity: Optional[dict]
+ :param hostnetwork: If True enable host networking on the pod
+ :type hostnetwork: bool
+ :param tolerations: A list of kubernetes tolerations
+ :type tolerations: Optional[list]
+ :param security_context: A dict containing the security context for the pod
+ :type security_context: Optional[Union[k8s.V1PodSecurityContext, dict]]
+ :param configmaps: Any configmap refs to envfrom.
+ If more than one configmap is required, provide a comma separated list
+ configmap_a,configmap_b
+ :type configmaps: List[str]
+ :param dnspolicy: Specify a dnspolicy for the pod
+ :type dnspolicy: Optional[str]
+ :param schedulername: Specify a schedulername for the pod
+ :type schedulername: Optional[str]
+ :param pod: The fully specified pod. Mutually exclusive with `path_or_string`
+ :type pod: Optional[kubernetes.client.models.V1Pod]
+ :param extract_xcom: Whether to bring up a container for xcom
+ :type extract_xcom: bool
+ :param priority_class_name: priority class name for the launched Pod
+ :type priority_class_name: str
+ """
+ def __init__( # pylint: disable=too-many-arguments,too-many-locals
+ self,
+ image: Optional[str] = None,
+ name: Optional[str] = None,
+ namespace: Optional[str] = None,
+ volume_mounts: Optional[List[Union[k8s.V1VolumeMount, dict]]] = None,
+ envs: Optional[Dict[str, str]] = None,
+ cmds: Optional[List[str]] = None,
+ args: Optional[List[str]] = None,
+ labels: Optional[Dict[str, str]] = None,
+ node_selectors: Optional[Dict[str, str]] = None,
+ ports: Optional[List[Union[k8s.V1ContainerPort, dict]]] = None,
+ volumes: Optional[List[Union[k8s.V1Volume, dict]]] = None,
+ image_pull_policy: Optional[str] = None,
+ restart_policy: Optional[str] = None,
+ image_pull_secrets: Optional[str] = None,
+ init_containers: Optional[List[k8s.V1Container]] = None,
+ service_account_name: Optional[str] = None,
+ resources: Optional[Union[k8s.V1ResourceRequirements, dict]] = None,
+ annotations: Optional[Dict[str, str]] = None,
+ affinity: Optional[dict] = None,
+ hostnetwork: bool = False,
+ tolerations: Optional[list] = None,
+ security_context: Optional[Union[k8s.V1PodSecurityContext, dict]] = None,
+ configmaps: Optional[List[str]] = None,
+ dnspolicy: Optional[str] = None,
+ schedulername: Optional[str] = None,
+ extract_xcom: bool = False,
+ priority_class_name: Optional[str] = None,
+ ):
+
+ self.pod = k8s.V1Pod()
+ self.pod.api_version = 'v1'
+ self.pod.kind = 'Pod'
+
+ # Pod Metadata
+ self.metadata = k8s.V1ObjectMeta()
+ self.metadata.labels = labels
+ self.metadata.name = name
+ self.metadata.namespace = namespace
+ self.metadata.annotations = annotations
+
+ # Pod Container
+ self.container = k8s.V1Container(name='base')
+ self.container.image = image
+ self.container.env = []
+
+ if envs:
+ if isinstance(envs, dict):
+ for key, val in envs.items():
+ self.container.env.append(k8s.V1EnvVar(
+ name=key,
+ value=val
+ ))
+ elif isinstance(envs, list):
+ self.container.env.extend(envs)
+
+ configmaps = configmaps or []
+ self.container.env_from = []
+ for configmap in configmaps:
+ self.container.env_from.append(k8s.V1EnvFromSource(
+ config_map_ref=k8s.V1ConfigMapEnvSource(
+ name=configmap
+ )
+ ))
+
+ self.container.command = cmds or []
+ self.container.args = args or []
+ self.container.image_pull_policy = image_pull_policy
+ self.container.ports = ports or []
+ self.container.resources = resources
+ self.container.volume_mounts = volume_mounts or []
+
+ # Pod Spec
+ self.spec = k8s.V1PodSpec(containers=[])
+ self.spec.security_context = security_context
+ self.spec.tolerations = tolerations
+ self.spec.dns_policy = dnspolicy
+ self.spec.scheduler_name = schedulername
+ self.spec.host_network = hostnetwork
+ self.spec.affinity = affinity
+ self.spec.service_account_name = service_account_name
+ self.spec.init_containers = init_containers
+ self.spec.volumes = volumes or []
+ self.spec.node_selector = node_selectors
+ self.spec.restart_policy = restart_policy
+ self.spec.priority_class_name = priority_class_name
+
+ self.spec.image_pull_secrets = []
+
+ if image_pull_secrets:
+ for image_pull_secret in image_pull_secrets.split(','):
+ self.spec.image_pull_secrets.append(k8s.V1LocalObjectReference(
+ name=image_pull_secret
+ ))
+
+ # Attach sidecar
+ self.extract_xcom = extract_xcom
+
+ def gen_pod(self) -> k8s.V1Pod:
+ """Generates pod"""
+ result = None
+
+ if result is None:
+ result = self.pod
+ result.spec = self.spec
+ result.metadata = self.metadata
+ result.spec.containers = [self.container]
+
+ result.metadata.name = self.make_unique_pod_id(result.metadata.name)
+
+ if self.extract_xcom:
+ result = self.add_sidecar(result)
+
+ return result
+
+ @staticmethod
+ def add_sidecar(pod: k8s.V1Pod) -> k8s.V1Pod:
+ """Adds sidecar"""
+ pod_cp = copy.deepcopy(pod)
+ pod_cp.spec.volumes = pod.spec.volumes or []
+ pod_cp.spec.volumes.insert(0, PodDefaults.VOLUME)
+ pod_cp.spec.containers[0].volume_mounts = pod_cp.spec.containers[0].volume_mounts or []
+ pod_cp.spec.containers[0].volume_mounts.insert(0, PodDefaults.VOLUME_MOUNT)
+ pod_cp.spec.containers.append(PodDefaults.SIDECAR_CONTAINER)
+
+ return pod_cp
+
+ @staticmethod
+ def from_obj(obj) -> Optional[k8s.V1Pod]:
+ """Converts to pod from obj"""
+ if obj is None:
+ return None
+
+ if isinstance(obj, PodGenerator):
+ return obj.gen_pod()
+
+ if not isinstance(obj, dict):
+ raise TypeError(
+ 'Cannot convert a non-dictionary or non-PodGenerator '
+ 'object into a KubernetesExecutorConfig')
+
+ # We do not want to extract constant here from ExecutorLoader because it is just
+ # A name in dictionary rather than executor selection mechanism and it causes cyclic import
+ namespaced = obj.get("KubernetesExecutor", {})
+
+ if not namespaced:
+ return None
+
+ resources = namespaced.get('resources')
+
+ if resources is None:
+ requests = {
+ 'cpu': namespaced.get('request_cpu'),
+ 'memory': namespaced.get('request_memory'),
+ 'ephemeral-storage': namespaced.get('ephemeral-storage')
+ }
+ limits = {
+ 'cpu': namespaced.get('limit_cpu'),
+ 'memory': namespaced.get('limit_memory'),
+ 'ephemeral-storage': namespaced.get('ephemeral-storage')
+ }
+ all_resources = list(requests.values()) + list(limits.values())
+ if all(r is None for r in all_resources):
+ resources = None
+ else:
+ resources = k8s.V1ResourceRequirements(
+ requests=requests,
+ limits=limits
+ )
+ namespaced['resources'] = resources
+ return PodGenerator(**namespaced).gen_pod()
+
+ @staticmethod
+ def make_unique_pod_id(dag_id):
+ r"""
+ Kubernetes pod names must be <= 253 chars and must pass the following regex for
+ validation
+ ``^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$``
+
+ :param dag_id: a dag_id with only alphanumeric characters
+ :return: ``str`` valid Pod name of appropriate length
+ """
+ if not dag_id:
+ return None
+
+ safe_uuid = uuid.uuid4().hex
+ safe_pod_id = dag_id[:MAX_POD_ID_LEN - len(safe_uuid) - 1] + "-" + safe_uuid
+
+ return safe_pod_id
diff --git a/airflow/kubernetes/volume.py b/airflow/kubernetes/volume.py
deleted file mode 100644
index 3383961..0000000
--- a/airflow/kubernetes/volume.py
+++ /dev/null
@@ -1,56 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-"""
-Classes for interacting with Kubernetes API
-"""
-import copy
-from typing import Dict
-
-from kubernetes.client import models as k8s
-
-from airflow.kubernetes.k8s_model import K8SModel
-
-
-class Volume(K8SModel):
- """
- Adds Kubernetes Volume to pod. allows pod to access features like ConfigMaps
- and Persistent Volumes
-
- :param name: the name of the volume mount
- :type name: str
- :param configs: dictionary of any features needed for volume.
- We purposely keep this vague since there are multiple volume types with changing
- configs.
- :type configs: dict
- """
- def __init__(self, name, configs):
- self.name = name
- self.configs = configs
-
- def to_k8s_client_obj(self) -> Dict[str, str]:
- """Converts to k8s object"""
- return {
- 'name': self.name,
- **self.configs
- }
-
- def attach_to_pod(self, pod: k8s.V1Pod) -> k8s.V1Pod:
- cp_pod = copy.deepcopy(pod)
- volume = self.to_k8s_client_obj()
- cp_pod.spec.volumes = pod.spec.volumes or []
- cp_pod.spec.volumes.append(volume)
- return cp_pod
diff --git a/airflow/kubernetes/volume_mount.py b/airflow/kubernetes/volume_mount.py
deleted file mode 100644
index 1e4bb8f..0000000
--- a/airflow/kubernetes/volume_mount.py
+++ /dev/null
@@ -1,72 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-"""
-Classes for interacting with Kubernetes API
-"""
-import copy
-
-from kubernetes.client import models as k8s
-
-from airflow.kubernetes.k8s_model import K8SModel
-
-
-class VolumeMount(K8SModel):
- """
- Initialize a Kubernetes Volume Mount. Used to mount pod level volumes to
- running container.
-
- :param name: the name of the volume mount
- :type name: str
- :param mount_path:
- :type mount_path: str
- :param sub_path: subpath within the volume mount
- :type sub_path: Optional[str]
- :param read_only: whether to access pod with read-only mode
- :type read_only: bool
- """
- def __init__(self, name, mount_path, sub_path, read_only):
- self.name = name
- self.mount_path = mount_path
- self.sub_path = sub_path
- self.read_only = read_only
-
- def to_k8s_client_obj(self) -> k8s.V1VolumeMount:
- """
- Converts to k8s object.
-
- :return Volume Mount k8s object
-
- """
- return k8s.V1VolumeMount(
- name=self.name,
- mount_path=self.mount_path,
- sub_path=self.sub_path,
- read_only=self.read_only
- )
-
- def attach_to_pod(self, pod: k8s.V1Pod) -> k8s.V1Pod:
- """
- Attaches to pod
-
- :return Copy of the Pod object
-
- """
- cp_pod = copy.deepcopy(pod)
- volume_mount = self.to_k8s_client_obj()
- cp_pod.spec.containers[0].volume_mounts = pod.spec.containers[0].volume_mounts or []
- cp_pod.spec.containers[0].volume_mounts.append(volume_mount)
- return cp_pod
diff --git a/airflow/kubernetes/worker_configuration.py b/airflow/kubernetes/worker_configuration.py
deleted file mode 100644
index 079cb82..0000000
--- a/airflow/kubernetes/worker_configuration.py
+++ /dev/null
@@ -1,448 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-"""Configuration of the worker"""
-import os
-from typing import Dict, List
-
-from kubernetes.client import models as k8s
-
-from airflow.configuration import conf
-from airflow.kubernetes.k8s_model import append_to_pod
-from airflow.kubernetes.pod_generator import PodGenerator
-from airflow.kubernetes.secret import Secret
-from airflow.utils.log.logging_mixin import LoggingMixin
-
-
-class WorkerConfiguration(LoggingMixin):
- """
- Contains Kubernetes Airflow Worker configuration logic
-
- :param kube_config: the kubernetes configuration from airflow.cfg
- :type kube_config: airflow.executors.kubernetes_executor.KubeConfig
- """
-
- dags_volume_name = 'airflow-dags'
- logs_volume_name = 'airflow-logs'
- git_sync_ssh_secret_volume_name = 'git-sync-ssh-key'
- git_ssh_key_secret_key = 'gitSshKey'
- git_sync_ssh_known_hosts_volume_name = 'git-sync-known-hosts'
-
- def __init__(self, kube_config):
- self.kube_config = kube_config
- self.worker_airflow_home = self.kube_config.airflow_home
- self.worker_airflow_dags = self.kube_config.dags_folder
- self.worker_airflow_logs = self.kube_config.base_log_folder
- super().__init__()
-
- def _get_init_containers(self) -> List[k8s.V1Container]:
- """When using git to retrieve the DAGs, use the GitSync Init Container"""
- # If we're using volume claims to mount the dags, no init container is needed
- if self.kube_config.dags_volume_claim or \
- self.kube_config.dags_volume_host or self.kube_config.dags_in_image:
- return []
-
- # Otherwise, define a git-sync init container
- init_environment = [k8s.V1EnvVar(
- name='GIT_SYNC_REPO',
- value=self.kube_config.git_repo
- ), k8s.V1EnvVar(
- name='GIT_SYNC_BRANCH',
- value=self.kube_config.git_branch
- ), k8s.V1EnvVar(
- name='GIT_SYNC_ROOT',
- value=self.kube_config.git_sync_root
- ), k8s.V1EnvVar(
- name='GIT_SYNC_DEST',
- value=self.kube_config.git_sync_dest
- ), k8s.V1EnvVar(
- name='GIT_SYNC_REV',
- value=self.kube_config.git_sync_rev
- ), k8s.V1EnvVar(
- name='GIT_SYNC_DEPTH',
- value=self.kube_config.git_sync_depth
- ), k8s.V1EnvVar(
- name='GIT_SYNC_ONE_TIME',
- value='true'
- )]
- if self.kube_config.git_user:
- init_environment.append(k8s.V1EnvVar(
- name='GIT_SYNC_USERNAME',
- value=self.kube_config.git_user
- ))
- if self.kube_config.git_password:
- init_environment.append(k8s.V1EnvVar(
- name='GIT_SYNC_PASSWORD',
- value=self.kube_config.git_password
- ))
-
- volume_mounts = [k8s.V1VolumeMount(
- mount_path=self.kube_config.git_sync_root,
- name=self.dags_volume_name,
- read_only=False
- )]
-
- if self.kube_config.git_sync_credentials_secret:
- init_environment.extend([
- k8s.V1EnvVar(
- name='GIT_SYNC_USERNAME',
- value_from=k8s.V1EnvVarSource(
- secret_key_ref=k8s.V1SecretKeySelector(
- name=self.kube_config.git_sync_credentials_secret,
- key='GIT_SYNC_USERNAME')
- )
- ),
- k8s.V1EnvVar(
- name='GIT_SYNC_PASSWORD',
- value_from=k8s.V1EnvVarSource(
- secret_key_ref=k8s.V1SecretKeySelector(
- name=self.kube_config.git_sync_credentials_secret,
- key='GIT_SYNC_PASSWORD')
- )
- )
- ])
-
- if self.kube_config.git_ssh_key_secret_name:
- volume_mounts.append(k8s.V1VolumeMount(
- name=self.git_sync_ssh_secret_volume_name,
- mount_path='/etc/git-secret/ssh',
- sub_path='ssh'
- ))
-
- init_environment.extend([
- k8s.V1EnvVar(
- name='GIT_SSH_KEY_FILE',
- value='/etc/git-secret/ssh'
- ),
- k8s.V1EnvVar(
- name='GIT_SYNC_ADD_USER',
- value='true'
- ),
- k8s.V1EnvVar(
- name='GIT_SYNC_SSH',
- value='true'
- )
- ])
-
- if self.kube_config.git_ssh_known_hosts_configmap_name:
- volume_mounts.append(k8s.V1VolumeMount(
- name=self.git_sync_ssh_known_hosts_volume_name,
- mount_path='/etc/git-secret/known_hosts',
- sub_path='known_hosts'
- ))
- init_environment.extend([k8s.V1EnvVar(
- name='GIT_KNOWN_HOSTS',
- value='true'
- ), k8s.V1EnvVar(
- name='GIT_SSH_KNOWN_HOSTS_FILE',
- value='/etc/git-secret/known_hosts'
- )])
- else:
- init_environment.append(k8s.V1EnvVar(
- name='GIT_KNOWN_HOSTS',
- value='false'
- ))
-
- init_containers = k8s.V1Container(
- name=self.kube_config.git_sync_init_container_name,
- image=self.kube_config.git_sync_container,
- env=init_environment,
- volume_mounts=volume_mounts
- )
-
- if self.kube_config.git_sync_run_as_user != "":
- init_containers.security_context = k8s.V1SecurityContext(
- run_as_user=self.kube_config.git_sync_run_as_user
- ) # git-sync user
-
- return [init_containers]
-
- def _get_environment(self) -> Dict[str, str]:
- """Defines any necessary environment variables for the pod executor"""
- env = {}
-
- for env_var_name, env_var_val in self.kube_config.kube_env_vars.items():
- env[env_var_name] = env_var_val
-
- env["AIRFLOW__CORE__EXECUTOR"] = "LocalExecutor"
-
- if self.kube_config.airflow_configmap:
- env['AIRFLOW_HOME'] = self.worker_airflow_home
- env['AIRFLOW__CORE__DAGS_FOLDER'] = self.worker_airflow_dags
- if (not self.kube_config.airflow_configmap and
- 'AIRFLOW__CORE__SQL_ALCHEMY_CONN' not in self.kube_config.kube_secrets):
- env['AIRFLOW__CORE__SQL_ALCHEMY_CONN'] = conf.get("core", "SQL_ALCHEMY_CONN")
- if self.kube_config.git_dags_folder_mount_point:
- # /root/airflow/dags/repo/dags
- dag_volume_mount_path = os.path.join(
- self.kube_config.git_dags_folder_mount_point,
- self.kube_config.git_sync_dest, # repo
- self.kube_config.git_subpath # dags
- )
- env['AIRFLOW__CORE__DAGS_FOLDER'] = dag_volume_mount_path
- return env
-
- def _get_env_from(self) -> List[k8s.V1EnvFromSource]:
- """Extracts any configmapRefs to envFrom"""
- env_from = []
-
- if self.kube_config.env_from_configmap_ref:
- for config_map_ref in self.kube_config.env_from_configmap_ref.split(','):
- env_from.append(
- k8s.V1EnvFromSource(config_map_ref=k8s.V1ConfigMapEnvSource(config_map_ref))
- )
-
- if self.kube_config.env_from_secret_ref:
- for secret_ref in self.kube_config.env_from_secret_ref.split(','):
- env_from.append(
- k8s.V1EnvFromSource(secret_ref=k8s.V1SecretEnvSource(secret_ref))
- )
-
- return env_from
-
- def _get_secrets(self):
- """Defines any necessary secrets for the pod executor"""
- worker_secrets = []
-
- for env_var_name, obj_key_pair in self.kube_config.kube_secrets.items():
- k8s_secret_obj, k8s_secret_key = obj_key_pair.split('=')
- worker_secrets.append(
- Secret('env', env_var_name, k8s_secret_obj, k8s_secret_key)
- )
-
- if self.kube_config.env_from_secret_ref:
- for secret_ref in self.kube_config.env_from_secret_ref.split(','):
- worker_secrets.append(
- Secret('env', None, secret_ref)
- )
-
- return worker_secrets
-
- def _get_security_context(self) -> k8s.V1PodSecurityContext:
- """Defines the security context"""
-
- security_context = k8s.V1PodSecurityContext()
-
- if self.kube_config.worker_run_as_user != "":
- security_context.run_as_user = self.kube_config.worker_run_as_user
-
- if self.kube_config.worker_fs_group != "":
- security_context.fs_group = self.kube_config.worker_fs_group
-
- # set fs_group to 65533 if not explicitly specified and using git ssh keypair auth
- if self.kube_config.git_ssh_key_secret_name and security_context.fs_group is None:
- security_context.fs_group = 65533
-
- return security_context
-
- def _get_labels(self, kube_executor_labels, labels) -> k8s.V1LabelSelector:
- copy = self.kube_config.kube_labels.copy()
- copy.update(kube_executor_labels)
- copy.update(labels)
- return copy
-
- def _get_volume_mounts(self) -> List[k8s.V1VolumeMount]:
- volume_mounts = {
- self.dags_volume_name: k8s.V1VolumeMount(
- name=self.dags_volume_name,
- mount_path=self.generate_dag_volume_mount_path(),
- read_only=True,
- ),
- self.logs_volume_name: k8s.V1VolumeMount(
- name=self.logs_volume_name,
- mount_path=self.worker_airflow_logs,
- )
- }
-
- if self.kube_config.dags_volume_subpath:
- volume_mounts[self.dags_volume_name].sub_path = self.kube_config.dags_volume_subpath
-
- if self.kube_config.logs_volume_subpath:
- volume_mounts[self.logs_volume_name].sub_path = self.kube_config.logs_volume_subpath
-
- if self.kube_config.dags_in_image:
- del volume_mounts[self.dags_volume_name]
-
- # Mount the airflow.cfg file via a configmap the user has specified
- if self.kube_config.airflow_configmap:
- config_volume_name = 'airflow-config'
- config_path = '{}/airflow.cfg'.format(self.worker_airflow_home)
- volume_mounts[config_volume_name] = k8s.V1VolumeMount(
- name=config_volume_name,
- mount_path=config_path,
- sub_path='airflow.cfg',
- read_only=True
- )
-
- # Mount the airflow_local_settings.py file via a configmap the user has specified
- if self.kube_config.airflow_local_settings_configmap:
- if self.kube_config.airflow_local_settings_configmap != self.kube_config.airflow_configmap:
- volume_mount_name = 'airflow-local-settings'
- else:
- volume_mount_name = 'airflow-config'
-
- config_path = '{}/config/airflow_local_settings.py'.format(self.worker_airflow_home)
- volume_mounts['airflow-local-settings'] = k8s.V1VolumeMount(
- name=volume_mount_name,
- mount_path=config_path,
- sub_path='airflow_local_settings.py',
- read_only=True
- )
-
- return list(volume_mounts.values())
-
- def _get_volumes(self) -> List[k8s.V1Volume]:
- def _construct_volume(name: str, claim: str = '', host: str = '') -> k8s.V1Volume:
- volume = k8s.V1Volume(name=name)
-
- if claim:
- volume.persistent_volume_claim = k8s.V1PersistentVolumeClaimVolumeSource(
- claim_name=claim
- )
- elif host:
- volume.host_path = k8s.V1HostPathVolumeSource(
- path=host,
- type=''
- )
- else:
- volume.empty_dir = {}
-
- return volume
-
- volumes = {
- self.dags_volume_name: _construct_volume(
- self.dags_volume_name,
- self.kube_config.dags_volume_claim,
- self.kube_config.dags_volume_host
- ),
- self.logs_volume_name: _construct_volume(
- self.logs_volume_name,
- self.kube_config.logs_volume_claim,
- self.kube_config.logs_volume_host
- )
- }
-
- if self.kube_config.dags_in_image:
- del volumes[self.dags_volume_name]
-
- # Get the SSH key from secrets as a volume
- if self.kube_config.git_ssh_key_secret_name:
- volumes[self.git_sync_ssh_secret_volume_name] = k8s.V1Volume(
- name=self.git_sync_ssh_secret_volume_name,
- secret=k8s.V1SecretVolumeSource(
- secret_name=self.kube_config.git_ssh_key_secret_name,
- items=[k8s.V1KeyToPath(
- key=self.git_ssh_key_secret_key,
- path='ssh',
- mode=0o440
- )]
- )
- )
-
- if self.kube_config.git_ssh_known_hosts_configmap_name:
- volumes[self.git_sync_ssh_known_hosts_volume_name] = k8s.V1Volume(
- name=self.git_sync_ssh_known_hosts_volume_name,
- config_map=k8s.V1ConfigMapVolumeSource(
- name=self.kube_config.git_ssh_known_hosts_configmap_name,
- default_mode=0o440
- )
- )
-
- # Mount the airflow_local_settings.py file via a configmap the user has specified
- if self.kube_config.airflow_local_settings_configmap:
- if self.kube_config.airflow_local_settings_configmap != self.kube_config.airflow_configmap:
- config_volume_name = 'airflow-local-settings'
- volumes[config_volume_name] = k8s.V1Volume(
- name=config_volume_name,
- config_map=k8s.V1ConfigMapVolumeSource(
- name=self.kube_config.airflow_local_settings_configmap
- )
- )
-
- # Mount the airflow.cfg file via a configmap the user has specified
- if self.kube_config.airflow_configmap:
- config_volume_name = 'airflow-config'
- volumes[config_volume_name] = k8s.V1Volume(
- name=config_volume_name,
- config_map=k8s.V1ConfigMapVolumeSource(
- name=self.kube_config.airflow_configmap
- )
- )
-
- return list(volumes.values())
-
- def generate_dag_volume_mount_path(self) -> str:
- """Generate path for DAG volume"""
-
- if self.kube_config.dags_volume_mount_point:
- return self.kube_config.dags_volume_mount_point
-
- if self.kube_config.dags_volume_claim or self.kube_config.dags_volume_host:
- return self.worker_airflow_dags
-
- return self.kube_config.git_dags_folder_mount_point
-
- def _get_resources(self) -> k8s.V1ResourceRequirements:
- if self.kube_config.worker_resources is None:
- return None
- resources_dict = self.kube_config.worker_resources
-
- requests = {}
- request_cpu = resources_dict.get('request_cpu', None)
- if request_cpu is not None:
- requests["cpu"] = request_cpu
- request_memory = resources_dict.get('request_memory', None)
- if request_memory is not None:
- requests["memory"] = request_memory
-
- limits = {}
- limit_cpu = resources_dict.get('limit_cpu', None)
- if limit_cpu is not None:
- limits["cpu"] = limit_cpu
- limit_memory = resources_dict.get('limit_memory', None)
- if limit_memory is not None:
- limits["memory"] = limit_memory
-
- return k8s.V1ResourceRequirements(requests=requests, limits=limits)
-
- def as_pod(self) -> k8s.V1Pod:
- """Creates POD."""
- if self.kube_config.pod_template_file:
- return PodGenerator(pod_template_file=self.kube_config.pod_template_file).gen_pod()
-
- pod = PodGenerator(
- image=self.kube_config.kube_image,
- image_pull_policy=self.kube_config.kube_image_pull_policy or 'IfNotPresent',
- image_pull_secrets=self.kube_config.image_pull_secrets,
- volumes=self._get_volumes(),
- volume_mounts=self._get_volume_mounts(),
- init_containers=self._get_init_containers(),
- labels=self.kube_config.kube_labels,
- annotations=self.kube_config.kube_annotations,
- affinity=self.kube_config.kube_affinity,
- tolerations=self.kube_config.kube_tolerations,
- envs=self._get_environment(),
- node_selectors=self.kube_config.kube_node_selectors,
- service_account_name=self.kube_config.worker_service_account_name or 'default',
- restart_policy='Never',
- resources=self._get_resources(),
- ).gen_pod()
-
- pod.spec.containers[0].env_from = pod.spec.containers[0].env_from or []
- pod.spec.containers[0].env_from.extend(self._get_env_from())
- pod.spec.security_context = self._get_security_context()
-
- return append_to_pod(pod, self._get_secrets())
diff --git a/airflow/kubernetes_executor_templates/basic_template.yaml b/airflow/kubernetes_executor_templates/basic_template.yaml
new file mode 100644
index 0000000..82c455b
--- /dev/null
+++ b/airflow/kubernetes_executor_templates/basic_template.yaml
@@ -0,0 +1,79 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+kind: Pod
+apiVersion: v1
+metadata:
+ name: dummy-name-dont-delete
+ namespace: dummy-name-dont-delete
+ labels:
+ mylabel: foo
+spec:
+ containers:
+ - name: base
+ image: dummy-name-dont-delete
+ env:
+ - name: AIRFLOW__CORE__EXECUTOR
+ value: LocalExecutor
+ - name: AIRFLOW_HOME
+ value: /opt/airflow
+ - name: AIRFLOW__CORE__DAGS_FOLDER
+ value: /opt/airflow/dags
+ - name: AIRFLOW__CORE__FERNET_KEY
+ valueFrom:
+ secretKeyRef:
+ name: airflow-fernet-key
+ key: fernet-key
+ - name: AIRFLOW__CORE__SQL_ALCHEMY_CONN
+ valueFrom:
+ secretKeyRef:
+ name: airflow-airflow-metadata
+ key: connection
+ - name: foo
+ value: bar
+ resources: {}
+ volumeMounts:
+ - name: airflow-logs
+ mountPath: /opt/airflow/logs
+ - name: airflow-config
+ readOnly: true
+ mountPath: /opt/airflow/airflow.cfg
+ subPath: airflow.cfg
+ - name: airflow-config
+ readOnly: true
+ mountPath: /opt/airflow/config/airflow_local_settings.py
+ subPath: airflow_local_settings.py
+ terminationMessagePath: /dev/termination-log
+ terminationMessagePolicy: File
+ imagePullPolicy: IfNotPresent
+ volumes:
+ - name: airflow-logs
+ emptyDir: {}
+ - name: airflow-config
+ configMap:
+ name: airflow-airflow-config
+ defaultMode: 420
+ restartPolicy: Never
+ terminationGracePeriodSeconds: 30
+ serviceAccountName: airflow-worker-serviceaccount
+ serviceAccount: airflow-worker-serviceaccount
+ securityContext:
+ runAsUser: 50000
+ fsGroup: 50000
+ imagePullSecrets:
+ - name: airflow-registry
+ schedulerName: default-scheduler
diff --git a/airflow/providers/cncf/kubernetes/example_dags/example_kubernetes.py b/airflow/providers/cncf/kubernetes/example_dags/example_kubernetes.py
index 708dd06..6135a6d 100644
--- a/airflow/providers/cncf/kubernetes/example_dags/example_kubernetes.py
+++ b/airflow/providers/cncf/kubernetes/example_dags/example_kubernetes.py
@@ -22,10 +22,7 @@ This is an example dag for using the KubernetesPodOperator.
from kubernetes.client import models as k8s
from airflow import DAG
-from airflow.kubernetes.pod import Port
from airflow.kubernetes.secret import Secret
-from airflow.kubernetes.volume import Volume
-from airflow.kubernetes.volume_mount import VolumeMount
from airflow.operators.bash import BashOperator
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from airflow.utils.dates import days_ago
@@ -34,13 +31,21 @@ from airflow.utils.dates import days_ago
secret_file = Secret('volume', '/etc/sql_conn', 'airflow-secrets', 'sql_alchemy_conn')
secret_env = Secret('env', 'SQL_CONN', 'airflow-secrets', 'sql_alchemy_conn')
secret_all_keys = Secret('env', None, 'airflow-secrets-2')
-volume_mount = VolumeMount('test-volume', mount_path='/root/mount_file', sub_path=None, read_only=True)
-configmaps = ['test-configmap-1', 'test-configmap-2']
-volume_config = {'persistentVolumeClaim': {'claimName': 'test-volume'}}
-volume = Volume(name='test-volume', configs=volume_config)
-# [END howto_operator_k8s_cluster_resources]
+volume_mount = k8s.V1VolumeMount(
+ name='test-volume', mount_path='/root/mount_file', sub_path=None, read_only=True
+)
-port = Port('http', 80)
+configmaps = [
+ k8s.V1EnvFromSource(config_map_ref=k8s.V1ConfigMapEnvSource(name='test-configmap-1')),
+ k8s.V1EnvFromSource(config_map_ref=k8s.V1ConfigMapEnvSource(name='test-configmap-2')),
+]
+
+volume = k8s.V1Volume(
+ name='test-volume',
+ persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(claim_name='test-volume'),
+)
+
+port = k8s.V1ContainerPort(name='http', container_port=80)
init_container_volume_mounts = [
k8s.V1VolumeMount(mount_path='/etc/foo', name='test-volume', sub_path=None, read_only=True)
@@ -89,6 +94,7 @@ affinity = {
}
tolerations = [{'key': "key", 'operator': 'Equal', 'value': 'value'}]
+# [END howto_operator_k8s_cluster_resources]
default_args = {
@@ -112,13 +118,13 @@ with DAG(
ports=[port],
volumes=[volume],
volume_mounts=[volume_mount],
+ env_from=configmaps,
name="airflow-test-pod",
task_id="task",
affinity=affinity,
is_delete_operator_pod=True,
hostnetwork=False,
tolerations=tolerations,
- configmaps=configmaps,
init_containers=[init_container],
priority_class_name="medium",
)
@@ -127,7 +133,7 @@ with DAG(
quay_k8s = KubernetesPodOperator(
namespace='default',
image='quay.io/apache/bash',
- image_pull_secrets='testquay',
+ image_pull_secrets=[k8s.V1LocalObjectReference('testquay')],
cmds=["bash", "-cx"],
arguments=["echo", "10", "echo pwd"],
labels={"foo": "bar"},
diff --git a/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py b/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py
index 7ef381f..4c07ad2 100644
--- a/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py
+++ b/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py
@@ -24,12 +24,7 @@ from kubernetes.client import CoreV1Api
from airflow.exceptions import AirflowException
from airflow.kubernetes import kube_client, pod_generator, pod_launcher
-from airflow.kubernetes.k8s_model import append_to_pod
-from airflow.kubernetes.pod import Port, Resources
-from airflow.kubernetes.pod_runtime_info_env import PodRuntimeInfoEnv
from airflow.kubernetes.secret import Secret
-from airflow.kubernetes.volume import Volume
-from airflow.kubernetes.volume_mount import VolumeMount
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults
from airflow.utils.helpers import validate_key
@@ -66,11 +61,11 @@ class KubernetesPodOperator(BaseOperator): # pylint: disable=too-many-instance-
The docker image's CMD is used if this is not provided.
:type arguments: list[str]
:param ports: ports for launched pod.
- :type ports: list[airflow.kubernetes.pod.Port]
+ :type ports: list[k8s.V1ContainerPort]
:param volume_mounts: volumeMounts for launched pod.
- :type volume_mounts: list[airflow.kubernetes.volume_mount.VolumeMount]
+ :type volume_mounts: list[k8s.V1VolumeMount]
:param volumes: volumes for launched pod. Includes ConfigMaps and PersistentVolumes.
- :type volumes: list[airflow.kubernetes.volume.Volume]
+ :type volumes: list[k8s.V1Volume]
:param env_vars: Environment variables initialized in the container. (templated)
:type env_vars: dict
:param secrets: Kubernetes secrets to inject in the container.
@@ -99,7 +94,7 @@ class KubernetesPodOperator(BaseOperator): # pylint: disable=too-many-instance-
Possible keys are request_memory, request_cpu, limit_memory, limit_cpu,
and limit_gpu, which will be used to generate airflow.kubernetes.pod.Resources.
See also kubernetes.io/docs/concepts/configuration/manage-compute-resources-container
- :type resources: dict
+ :type resources: k8s.V1ResourceRequirements
:param affinity: A dict containing a group of affinity scheduling rules.
:type affinity: dict
:param config_file: The path to the Kubernetes config file. (templated)
@@ -110,7 +105,7 @@ class KubernetesPodOperator(BaseOperator): # pylint: disable=too-many-instance-
:param image_pull_secrets: Any image pull secrets to be given to the pod.
If more than one secret is required, provide a
comma separated list: secret_a,secret_b
- :type image_pull_secrets: str
+ :type image_pull_secrets: List[k8s.V1LocalObjectReference]
:param service_account_name: Name of the service account
:type service_account_name: str
:param is_delete_operator_pod: What to do when the pod reaches its final
@@ -121,14 +116,8 @@ class KubernetesPodOperator(BaseOperator): # pylint: disable=too-many-instance-
:type hostnetwork: bool
:param tolerations: A list of kubernetes tolerations.
:type tolerations: list tolerations
- :param configmaps: A list of configmap names objects that we
- want mount as env variables.
- :type configmaps: list[str]
:param security_context: security options the pod should run with (PodSecurityContext).
:type security_context: dict
- :param pod_runtime_info_envs: environment variables about
- pod runtime information (ip, namespace, nodeName, podName).
- :type pod_runtime_info_envs: list[airflow.kubernetes.pod_runtime_info_env.PodRuntimeInfoEnv]
:param dnspolicy: dnspolicy for the pod.
:type dnspolicy: str
:param schedulername: Specify a schedulername for the pod
@@ -162,10 +151,11 @@ class KubernetesPodOperator(BaseOperator): # pylint: disable=too-many-instance-
name: Optional[str] = None,
cmds: Optional[List[str]] = None,
arguments: Optional[List[str]] = None,
- ports: Optional[List[Port]] = None,
- volume_mounts: Optional[List[VolumeMount]] = None,
- volumes: Optional[List[Volume]] = None,
- env_vars: Optional[Dict] = None,
+ ports: Optional[List[k8s.V1ContainerPort]] = None,
+ volume_mounts: Optional[List[k8s.V1VolumeMount]] = None,
+ volumes: Optional[List[k8s.V1Volume]] = None,
+ env_vars: Optional[List[k8s.V1EnvVar]] = None,
+ env_from: Optional[List[k8s.V1EnvFromSource]] = None,
secrets: Optional[List[Secret]] = None,
in_cluster: Optional[bool] = None,
cluster_context: Optional[str] = None,
@@ -175,18 +165,16 @@ class KubernetesPodOperator(BaseOperator): # pylint: disable=too-many-instance-
get_logs: bool = True,
image_pull_policy: str = 'IfNotPresent',
annotations: Optional[Dict] = None,
- resources: Optional[Dict] = None,
+ resources: Optional[k8s.V1ResourceRequirements] = None,
affinity: Optional[Dict] = None,
config_file: Optional[str] = None,
node_selectors: Optional[Dict] = None,
- image_pull_secrets: Optional[str] = None,
+ image_pull_secrets: Optional[List[k8s.V1LocalObjectReference]] = None,
service_account_name: str = 'default',
is_delete_operator_pod: bool = False,
hostnetwork: bool = False,
tolerations: Optional[List] = None,
- configmaps: Optional[List] = None,
security_context: Optional[Dict] = None,
- pod_runtime_info_envs: Optional[List[PodRuntimeInfoEnv]] = None,
dnspolicy: Optional[str] = None,
schedulername: Optional[str] = None,
full_pod_spec: Optional[k8s.V1Pod] = None,
@@ -208,7 +196,8 @@ class KubernetesPodOperator(BaseOperator): # pylint: disable=too-many-instance-
self.arguments = arguments or []
self.labels = labels or {}
self.startup_timeout_seconds = startup_timeout_seconds
- self.env_vars = env_vars or {}
+ self.env_vars = env_vars or []
+ self.env_from = env_from or []
self.ports = ports or []
self.volume_mounts = volume_mounts or []
self.volumes = volumes or []
@@ -221,16 +210,14 @@ class KubernetesPodOperator(BaseOperator): # pylint: disable=too-many-instance-
self.node_selectors = node_selectors or {}
self.annotations = annotations or {}
self.affinity = affinity or {}
- self.resources = self._set_resources(resources) # noqa
+ self.k8s_resources = resources or {}
self.config_file = config_file
- self.image_pull_secrets = image_pull_secrets
+ self.image_pull_secrets = image_pull_secrets or []
self.service_account_name = service_account_name
self.is_delete_operator_pod = is_delete_operator_pod
self.hostnetwork = hostnetwork
self.tolerations = tolerations or []
- self.configmaps = configmaps or []
self.security_context = security_context or {}
- self.pod_runtime_info_envs = pod_runtime_info_envs or []
self.dnspolicy = dnspolicy
self.schedulername = schedulername
self.full_pod_spec = full_pod_spec
@@ -347,12 +334,6 @@ class KubernetesPodOperator(BaseOperator): # pylint: disable=too-many-instance-
def _try_numbers_match(context, pod):
return pod.metadata.labels['try_number'] == context['ti'].try_number
- @staticmethod
- def _set_resources(resources):
- if not resources:
- return []
- return [Resources(**resources)]
-
def _set_name(self, name):
if self.pod_template_file or self.full_pod_spec:
return None
@@ -365,43 +346,55 @@ class KubernetesPodOperator(BaseOperator): # pylint: disable=too-many-instance-
will supersede all other values.
"""
- pod = pod_generator.PodGenerator(
- image=self.image,
- namespace=self.namespace,
- cmds=self.cmds,
- args=self.arguments,
- labels=self.labels,
- name=self.name,
- envs=self.env_vars,
- extract_xcom=self.do_xcom_push,
- image_pull_policy=self.image_pull_policy,
- node_selectors=self.node_selectors,
- annotations=self.annotations,
- affinity=self.affinity,
- image_pull_secrets=self.image_pull_secrets,
- service_account_name=self.service_account_name,
- hostnetwork=self.hostnetwork,
- tolerations=self.tolerations,
- configmaps=self.configmaps,
- security_context=self.security_context,
- dnspolicy=self.dnspolicy,
- schedulername=self.schedulername,
- init_containers=self.init_containers,
- restart_policy='Never',
- priority_class_name=self.priority_class_name,
- pod_template_file=self.pod_template_file,
- pod=self.full_pod_spec,
- ).gen_pod()
-
- pod = append_to_pod(
- pod,
- self.pod_runtime_info_envs +
- self.ports + # type: ignore
- self.resources +
- self.secrets + # type: ignore
- self.volumes + # type: ignore
- self.volume_mounts # type: ignore
- )
+ if self.pod_template_file:
+ pod = pod_generator.PodGenerator.deserialize_model_file(self.pod_template_file)
+ elif not self.pod:
+ pod = k8s.V1Pod(
+ api_version="v1",
+ kind="Pod",
+ metadata=k8s.V1ObjectMeta(
+ namespace=self.namespace,
+ labels=self.labels,
+ name=self.name,
+ annotations=self.annotations,
+
+ ),
+ spec=k8s.V1PodSpec(
+ node_selector=self.node_selectors,
+ affinity=self.affinity,
+ tolerations=self.tolerations,
+ init_containers=self.init_containers,
+ containers=[
+ k8s.V1Container(
+ image=self.image,
+ name="base",
+ command=self.cmds,
+ ports=self.ports,
+ resources=self.k8s_resources,
+ volume_mounts=self.volume_mounts,
+ args=self.arguments,
+ env=self.env_vars,
+ env_from=self.env_from,
+ )
+ ],
+ image_pull_secrets=self.image_pull_secrets,
+ service_account_name=self.service_account_name,
+ host_network=self.hostnetwork,
+ security_context=self.security_context,
+ dns_policy=self.dnspolicy,
+ scheduler_name=self.schedulername,
+ restart_policy='Never',
+ priority_class_name=self.priority_class_name,
+ volumes=self.volumes,
+ )
+ )
+ else:
+ pod = self.pod
+ for secret in self.secrets:
+ pod = secret.attach_to_pod(pod)
+ if self.do_xcom_push:
+ from airflow.kubernetes.pod_generator import PodGenerator
+ pod = PodGenerator.add_xcom_sidecar(pod)
return pod
def create_new_pod_for_operator(self, labels, launcher) -> Tuple[State, k8s.V1Pod, Optional[str]]:
diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py
index 62cee8c..6129a9d 100644
--- a/airflow/serialization/serialized_objects.py
+++ b/airflow/serialization/serialized_objects.py
@@ -185,6 +185,9 @@ class BaseSerialization:
{str(k): cls._serialize(v) for k, v in var.items()},
type_=DAT.DICT
)
+ elif isinstance(var, k8s.V1Pod):
+ json_pod = PodGenerator.serialize_pod(var)
+ return cls._encode(json_pod, type_=DAT.POD)
elif isinstance(var, list):
return [cls._serialize(v) for v in var]
elif isinstance(var, k8s.V1Pod):
diff --git a/airflow/utils/json.py b/airflow/utils/json.py
index b486cf0..c18e090 100644
--- a/airflow/utils/json.py
+++ b/airflow/utils/json.py
@@ -21,6 +21,11 @@ from datetime import date, datetime
import numpy as np
+try:
+ from kubernetes.client import models as k8s
+except ImportError:
+ k8s = None
+
# Dates and JSON encoding/decoding
@@ -50,5 +55,8 @@ class AirflowJsonEncoder(json.JSONEncoder):
elif isinstance(obj, (np.float_, np.float16, np.float32, np.float64,
np.complex_, np.complex64, np.complex128)):
return float(obj)
+ elif k8s is not None and isinstance(obj, k8s.V1Pod):
+ from airflow.kubernetes.pod_generator import PodGenerator
+ PodGenerator.serialize_pod(obj)
raise TypeError(f"Object of type '{obj.__class__.__name__}' is not JSON serializable")
diff --git a/docs/howto/operator/kubernetes.rst b/docs/howto/operator/kubernetes.rst
index 24b05bc..73d0253 100644
--- a/docs/howto/operator/kubernetes.rst
+++ b/docs/howto/operator/kubernetes.rst
@@ -53,15 +53,17 @@ Ultimately, it allows Airflow to act a job orchestrator - no matter the language
How to use cluster ConfigMaps, Secrets, and Volumes with Pod?
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-Kubernetes cluster resources such as ConfigMaps, Secrets, and Volumes can be used with a Pod to be launched.
-Utilize the Airflow Kubernetes model classes such as:
-:class:`~airflow.kubernetes.secret.Secret`
-or
-:class:`~airflow.kubernetes.volume.Volume`
-or
-:class:`~airflow.kubernetes.volume_mount.VolumeMount`
-to do this (as well as standard Python dictionaries). These can they be specified in the appropriate parameters
-when declaring the Pod task.
+
+To add ConfigMaps, Volumes, and other Kubernetes native objects, we recommend that you import the Kubernetes model API
+like this:
+.. code-block:: python
+
+ from kubernetes.client import models as k8s
+
+With this API object, you can have access to all Kubernetes API objects in the form of python classes.
+Using this method will ensure correctness
+and type safety. While we have removed almost all Kubernetes convenience classes, we have kept the
+:class:`~airflow.kubernetes.secret.Secret` class to simplify the process of generating secret volumes/env variables.
.. exampleinclude:: ../../../airflow/providers/cncf/kubernetes/example_dags/example_kubernetes.py
:language: python
diff --git a/kubernetes_tests/test_kubernetes_pod_operator.py b/kubernetes_tests/test_kubernetes_pod_operator.py
index d961a1c..017a9db 100644
--- a/kubernetes_tests/test_kubernetes_pod_operator.py
+++ b/kubernetes_tests/test_kubernetes_pod_operator.py
@@ -18,6 +18,7 @@
import json
import logging
import os
+import random
import shutil
import sys
import textwrap
@@ -32,13 +33,9 @@ from kubernetes.client.rest import ApiException
from airflow.exceptions import AirflowException
from airflow.kubernetes import kube_client
-from airflow.kubernetes.pod import Port
from airflow.kubernetes.pod_generator import PodDefaults
from airflow.kubernetes.pod_launcher import PodLauncher
-from airflow.kubernetes.pod_runtime_info_env import PodRuntimeInfoEnv
from airflow.kubernetes.secret import Secret
-from airflow.kubernetes.volume import Volume
-from airflow.kubernetes.volume_mount import VolumeMount
from airflow.models import DAG, TaskInstance
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from airflow.utils import timezone
@@ -90,8 +87,8 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
'args': ["echo 10"],
'command': ["bash", "-cx"],
'env': [],
- 'imagePullPolicy': 'IfNotPresent',
'envFrom': [],
+ 'resources': {},
'name': 'base',
'ports': [],
'volumeMounts': [],
@@ -111,6 +108,8 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
def tearDown(self) -> None:
client = kube_client.get_kube_client(in_cluster=False)
client.delete_collection_namespaced_pod(namespace="default")
+ import time
+ time.sleep(1)
def test_do_xcom_push_defaults_false(self):
new_config_path = '/tmp/kube_config'
@@ -123,7 +122,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -160,7 +159,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -178,7 +177,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -197,7 +196,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -218,7 +217,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -241,7 +240,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -263,7 +262,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -276,21 +275,26 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
self.assertEqual(self.expected_pod, actual_pod)
def test_pod_resources(self):
- resources = {
- 'limit_cpu': 0.25,
- 'limit_memory': '64Mi',
- 'limit_ephemeral_storage': '2Gi',
- 'request_cpu': '250m',
- 'request_memory': '64Mi',
- 'request_ephemeral_storage': '1Gi',
- }
+ resources = k8s.V1ResourceRequirements(
+ requests={
+ 'memory': '64Mi',
+ 'cpu': '250m',
+ 'ephemeral-storage': '1Gi'
+ },
+ limits={
+ 'memory': '64Mi',
+ 'cpu': 0.25,
+ 'nvidia.com/gpu': None,
+ 'ephemeral-storage': '2Gi'
+ }
+ )
k = KubernetesPodOperator(
namespace='default',
image="ubuntu:16.04",
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -338,7 +342,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -351,7 +355,10 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
self.assertEqual(self.expected_pod, actual_pod)
def test_port(self):
- port = Port('http', 80)
+ port = k8s.V1ContainerPort(
+ name='http',
+ container_port=80,
+ )
k = KubernetesPodOperator(
namespace='default',
@@ -359,7 +366,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -376,18 +383,20 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
def test_volume_mount(self):
with mock.patch.object(PodLauncher, 'log') as mock_logger:
- volume_mount = VolumeMount('test-volume',
- mount_path='/tmp/test_volume',
- sub_path=None,
- read_only=False)
-
- volume_config = {
- 'persistentVolumeClaim':
- {
- 'claimName': 'test-volume'
- }
- }
- volume = Volume(name='test-volume', configs=volume_config)
+ volume_mount = k8s.V1VolumeMount(
+ name='test-volume',
+ mount_path='/tmp/test_volume',
+ sub_path=None,
+ read_only=False
+ )
+
+ volume = k8s.V1Volume(
+ name='test-volume',
+ persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(
+ claim_name='test-volume'
+ )
+ )
+
args = ["echo \"retrieved from mount\" > /tmp/test_volume/test.txt "
"&& cat /tmp/test_volume/test.txt"]
k = KubernetesPodOperator(
@@ -398,7 +407,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
labels={"foo": "bar"},
volume_mounts=[volume_mount],
volumes=[volume],
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -433,7 +442,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -458,7 +467,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -483,7 +492,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-fs-group",
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -503,7 +512,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -524,7 +533,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -549,7 +558,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=bad_internal_command,
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -570,7 +579,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=args,
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=True,
@@ -594,7 +603,10 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
# GIVEN
from airflow.utils.state import State
- configmap = 'test-configmap'
+ configmap_name = "test-config-map"
+ env_from = [k8s.V1EnvFromSource(config_map_ref=k8s.V1ConfigMapEnvSource(
+ name=configmap_name
+ ))]
# WHEN
k = KubernetesPodOperator(
namespace='default',
@@ -602,21 +614,18 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
- configmaps=[configmap],
+ env_from=env_from
)
# THEN
mock_monitor.return_value = (State.SUCCESS, None)
context = create_context(k)
k.execute(context)
self.assertEqual(
- mock_start.call_args[0][0].spec.containers[0].env_from,
- [k8s.V1EnvFromSource(config_map_ref=k8s.V1ConfigMapEnvSource(
- name=configmap
- ))]
+ mock_start.call_args[0][0].spec.containers[0].env_from, env_from
)
@mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.start_pod")
@@ -635,7 +644,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
arguments=["echo 10"],
secrets=secrets,
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -653,15 +662,33 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
def test_env_vars(self):
# WHEN
+ env_vars = [
+ k8s.V1EnvVar(
+ name="ENV1",
+ value="val1"
+ ),
+ k8s.V1EnvVar(
+ name="ENV2",
+ value="val2"
+ ),
+ k8s.V1EnvVar(
+ name="ENV3",
+ value_from=k8s.V1EnvVarSource(
+ field_ref=k8s.V1ObjectFieldSelector(
+ field_path="status.podIP"
+ )
+ )
+ ),
+ ]
+
k = KubernetesPodOperator(
namespace='default',
image="ubuntu:16.04",
cmds=["bash", "-cx"],
arguments=["echo 10"],
- env_vars={"ENV1": "val1", "ENV2": "val2", },
- pod_runtime_info_envs=[PodRuntimeInfoEnv("ENV3", "status.podIP")],
+ env_vars=env_vars,
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
@@ -726,14 +753,12 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
args=["echo 10"]
)
- volume_config = {
- 'persistentVolumeClaim':
- {
- 'claimName': 'test-volume'
- }
- }
- volume = Volume(name='test-volume', configs=volume_config)
-
+ volume = k8s.V1Volume(
+ name='test-volume',
+ persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(
+ claim_name='test-volume'
+ )
+ )
expected_init_container = {
'name': 'init-container',
'image': 'ubuntu:16.04',
@@ -759,7 +784,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
volumes=[volume],
init_containers=[init_container],
@@ -782,16 +807,19 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
@mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.monitor_pod")
@mock.patch("airflow.kubernetes.kube_client.get_kube_client")
def test_pod_template_file(
- self,
- mock_client,
- monitor_mock,
- start_mock): # pylint: disable=unused-argument
+ self,
+ mock_client,
+ monitor_mock,
+ start_mock # pylint: disable=unused-argument
+ ):
from airflow.utils.state import State
+ path = sys.path[0] + '/tests/kubernetes/pod.yaml'
k = KubernetesPodOperator(
task_id="task" + self.get_current_task_name(),
- pod_template_file='tests/kubernetes/pod.yaml',
+ pod_template_file=path,
do_xcom_push=True
)
+
monitor_mock.return_value = (State.SUCCESS, None)
context = create_context(k)
with self.assertLogs(k.log, level=logging.DEBUG) as cm:
@@ -847,10 +875,11 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
@mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.monitor_pod")
@mock.patch("airflow.kubernetes.kube_client.get_kube_client")
def test_pod_priority_class_name(
- self,
- mock_client,
- monitor_mock,
- start_mock): # pylint: disable=unused-argument
+ self,
+ mock_client,
+ monitor_mock,
+ start_mock # pylint: disable=unused-argument
+ ):
"""Test ability to assign priorityClassName to pod
"""
@@ -863,7 +892,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
+ name="test-" + str(random.randint(0, 1000000)),
task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
diff --git a/scripts/ci/libraries/_kind.sh b/scripts/ci/libraries/_kind.sh
index eff8c56..21703d2 100644
--- a/scripts/ci/libraries/_kind.sh
+++ b/scripts/ci/libraries/_kind.sh
@@ -253,6 +253,8 @@ USER root
COPY --chown=airflow:root airflow/example_dags/ \${AIRFLOW_HOME}/dags/
+COPY --chown=airflow:root airflow/kubernetes_executor_templates/ \${AIRFLOW_HOME}/pod_templates/
+
USER airflow
EOF
diff --git a/tests/executors/test_kubernetes_executor.py b/tests/executors/test_kubernetes_executor.py
index 9cf2465..9bd1868 100644
--- a/tests/executors/test_kubernetes_executor.py
+++ b/tests/executors/test_kubernetes_executor.py
@@ -30,9 +30,7 @@ from tests.test_utils.config import conf_vars
try:
from kubernetes.client.rest import ApiException
- from airflow.executors.kubernetes_executor import (
- AirflowKubernetesScheduler, KubeConfig, KubernetesExecutor,
- )
+ from airflow.executors.kubernetes_executor import AirflowKubernetesScheduler, KubernetesExecutor
from airflow.kubernetes import pod_generator
from airflow.kubernetes.pod_generator import PodGenerator
from airflow.utils.state import State
@@ -118,55 +116,6 @@ class TestAirflowKubernetesScheduler(unittest.TestCase):
self.assertEqual(datetime_obj, new_datetime_obj)
-class TestKubeConfig(unittest.TestCase):
- def setUp(self):
- if AirflowKubernetesScheduler is None:
- self.skipTest("kubernetes python package is not installed")
-
- @conf_vars({
- ('kubernetes', 'git_ssh_known_hosts_configmap_name'): 'airflow-configmap',
- ('kubernetes', 'git_ssh_key_secret_name'): 'airflow-secrets',
- ('kubernetes', 'worker_annotations'): '{ "iam.com/role" : "role-arn", "other/annotation" : "value" }'
- })
- def test_kube_config_worker_annotations_properly_parsed(self):
- annotations = KubeConfig().kube_annotations
- self.assertEqual({'iam.com/role': 'role-arn', 'other/annotation': 'value'}, annotations)
-
- @conf_vars({
- ('kubernetes', 'git_ssh_known_hosts_configmap_name'): 'airflow-configmap',
- ('kubernetes', 'git_ssh_key_secret_name'): 'airflow-secrets'
- })
- def test_kube_config_no_worker_annotations(self):
- annotations = KubeConfig().kube_annotations
- self.assertIsNone(annotations)
-
- @conf_vars({
- ('kubernetes', 'git_repo'): 'foo',
- ('kubernetes', 'git_branch'): 'foo',
- ('kubernetes', 'git_dags_folder_mount_point'): 'foo',
- ('kubernetes', 'git_sync_run_as_user'): '0',
- })
- def test_kube_config_git_sync_run_as_user_root(self):
- self.assertEqual(KubeConfig().git_sync_run_as_user, 0)
-
- @conf_vars({
- ('kubernetes', 'git_repo'): 'foo',
- ('kubernetes', 'git_branch'): 'foo',
- ('kubernetes', 'git_dags_folder_mount_point'): 'foo',
- })
- def test_kube_config_git_sync_run_as_user_not_present(self):
- self.assertEqual(KubeConfig().git_sync_run_as_user, 65533)
-
- @conf_vars({
- ('kubernetes', 'git_repo'): 'foo',
- ('kubernetes', 'git_branch'): 'foo',
- ('kubernetes', 'git_dags_folder_mount_point'): 'foo',
- ('kubernetes', 'git_sync_run_as_user'): '',
- })
- def test_kube_config_git_sync_run_as_user_empty_string(self):
- self.assertEqual(KubeConfig().git_sync_run_as_user, '')
-
-
class TestKubernetesExecutor(unittest.TestCase):
"""
Tests if an ApiException from the Kube Client will cause the task to
@@ -177,6 +126,8 @@ class TestKubernetesExecutor(unittest.TestCase):
@mock.patch('airflow.executors.kubernetes_executor.KubernetesJobWatcher')
@mock.patch('airflow.executors.kubernetes_executor.get_kube_client')
def test_run_next_exception(self, mock_get_kube_client, mock_kubernetes_job_watcher):
+ import sys
+ path = sys.path[0] + '/tests/kubernetes/pod_generator_base_with_secrets.yaml'
# When a quota is exceeded this is the ApiException we get
response = HTTPResponse(
@@ -195,29 +146,32 @@ class TestKubernetesExecutor(unittest.TestCase):
mock_api_client = mock.MagicMock()
mock_api_client.sanitize_for_serialization.return_value = {}
mock_kube_client.api_client = mock_api_client
-
- kubernetes_executor = KubernetesExecutor()
- kubernetes_executor.start()
-
- # Execute a task while the Api Throws errors
- try_number = 1
- kubernetes_executor.execute_async(key=('dag', 'task', datetime.utcnow(), try_number),
- queue=None,
- command=['airflow', 'tasks', 'run', 'true', 'some_parameter'],
- executor_config={})
- kubernetes_executor.sync()
- kubernetes_executor.sync()
-
- assert mock_kube_client.create_namespaced_pod.called
- self.assertFalse(kubernetes_executor.task_queue.empty())
-
- # Disable the ApiException
- mock_kube_client.create_namespaced_pod.side_effect = None
-
- # Execute the task without errors should empty the queue
- kubernetes_executor.sync()
- assert mock_kube_client.create_namespaced_pod.called
- self.assertTrue(kubernetes_executor.task_queue.empty())
+ config = {
+ ('kubernetes', 'pod_template_file'): path,
+ }
+ with conf_vars(config):
+
+ kubernetes_executor = KubernetesExecutor()
+ kubernetes_executor.start()
+ # Execute a task while the Api Throws errors
+ try_number = 1
+ kubernetes_executor.execute_async(key=('dag', 'task', datetime.utcnow(), try_number),
+ queue=None,
+ command=['airflow', 'tasks', 'run', 'true', 'some_parameter'],
+ )
+ kubernetes_executor.sync()
+ kubernetes_executor.sync()
+
+ assert mock_kube_client.create_namespaced_pod.called
+ self.assertFalse(kubernetes_executor.task_queue.empty())
+
+ # Disable the ApiException
+ mock_kube_client.create_namespaced_pod.side_effect = None
+
+ # Execute the task without errors should empty the queue
+ kubernetes_executor.sync()
+ assert mock_kube_client.create_namespaced_pod.called
+ self.assertTrue(kubernetes_executor.task_queue.empty())
@mock.patch('airflow.executors.kubernetes_executor.KubeConfig')
@mock.patch('airflow.executors.kubernetes_executor.KubernetesExecutor.sync')
diff --git a/tests/kubernetes/models/test_pod.py b/tests/kubernetes/models/test_pod.py
deleted file mode 100644
index 67659b5..0000000
--- a/tests/kubernetes/models/test_pod.py
+++ /dev/null
@@ -1,77 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-import unittest
-import uuid
-from unittest import mock
-
-from kubernetes.client import ApiClient, models as k8s
-
-from airflow.kubernetes.k8s_model import append_to_pod
-from airflow.kubernetes.pod import Port
-from airflow.kubernetes.pod_generator import PodGenerator
-
-
-class TestPod(unittest.TestCase):
-
- def test_port_to_k8s_client_obj(self):
- port = Port('http', 80)
- self.assertEqual(
- port.to_k8s_client_obj(),
- k8s.V1ContainerPort(
- name='http',
- container_port=80
- )
- )
-
- @mock.patch('uuid.uuid4')
- def test_port_attach_to_pod(self, mock_uuid):
- static_uuid = uuid.UUID('cf4a56d2-8101-4217-b027-2af6216feb48')
- mock_uuid.return_value = static_uuid
- pod = PodGenerator(image='airflow-worker:latest', name='base').gen_pod()
- ports = [
- Port('https', 443),
- Port('http', 80)
- ]
- k8s_client = ApiClient()
- result = append_to_pod(pod, ports)
- result = k8s_client.sanitize_for_serialization(result)
- self.assertEqual({
- 'apiVersion': 'v1',
- 'kind': 'Pod',
- 'metadata': {'name': 'base-' + static_uuid.hex},
- 'spec': {
- 'containers': [{
- 'args': [],
- 'command': [],
- 'env': [],
- 'envFrom': [],
- 'image': 'airflow-worker:latest',
- 'name': 'base',
- 'ports': [{
- 'name': 'https',
- 'containerPort': 443
- }, {
- 'name': 'http',
- 'containerPort': 80
- }],
- 'volumeMounts': [],
- }],
- 'hostNetwork': False,
- 'imagePullSecrets': [],
- 'volumes': []
- }
- }, result)
diff --git a/tests/kubernetes/models/test_secret.py b/tests/kubernetes/models/test_secret.py
index e4bbda5..2906272 100644
--- a/tests/kubernetes/models/test_secret.py
+++ b/tests/kubernetes/models/test_secret.py
@@ -14,6 +14,7 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+import sys
import unittest
import uuid
from unittest import mock
@@ -67,8 +68,8 @@ class TestSecret(unittest.TestCase):
def test_attach_to_pod(self, mock_uuid):
static_uuid = uuid.UUID('cf4a56d2-8101-4217-b027-2af6216feb48')
mock_uuid.return_value = static_uuid
- pod = PodGenerator(image='airflow-worker:latest',
- name='base').gen_pod()
+ path = sys.path[0] + '/tests/kubernetes/pod_generator_base.yaml'
+ pod = PodGenerator(pod_template_file=path).gen_pod()
secrets = [
# This should be a secretRef
Secret('env', None, 'secret_a'),
@@ -78,39 +79,46 @@ class TestSecret(unittest.TestCase):
Secret('env', 'TARGET', 'secret_b', 'source_b'),
]
k8s_client = ApiClient()
- result = append_to_pod(pod, secrets)
- result = k8s_client.sanitize_for_serialization(result)
- self.assertEqual(result, {
- 'apiVersion': 'v1',
- 'kind': 'Pod',
- 'metadata': {'name': 'base-' + static_uuid.hex},
- 'spec': {
- 'containers': [{
- 'args': [],
- 'command': [],
- 'env': [{
- 'name': 'TARGET',
- 'valueFrom': {
- 'secretKeyRef': {
- 'key': 'source_b',
- 'name': 'secret_b'
- }
- }
- }],
- 'envFrom': [{'secretRef': {'name': 'secret_a'}}],
- 'image': 'airflow-worker:latest',
- 'name': 'base',
- 'ports': [],
- 'volumeMounts': [{
- 'mountPath': '/etc/foo',
- 'name': 'secretvol' + str(static_uuid),
- 'readOnly': True}]
- }],
- 'hostNetwork': False,
- 'imagePullSecrets': [],
- 'volumes': [{
- 'name': 'secretvol' + str(static_uuid),
- 'secret': {'secretName': 'secret_b'}
- }]
- }
- })
+ pod = append_to_pod(pod, secrets)
+ result = k8s_client.sanitize_for_serialization(pod)
+ self.assertEqual(result,
+ {'apiVersion': 'v1',
+ 'kind': 'Pod',
+ 'metadata': {'labels': {'app': 'myapp'},
+ 'name': 'myapp-pod-cf4a56d281014217b0272af6216feb48',
+ 'namespace': 'default'},
+ 'spec': {'containers': [{'command': ['sh', '-c', 'echo Hello Kubernetes!'],
+ 'env': [{'name': 'ENVIRONMENT', 'value': 'prod'},
+ {'name': 'LOG_LEVEL', 'value': 'warning'},
+ {'name': 'TARGET',
+ 'valueFrom':
+ {'secretKeyRef': {'key': 'source_b',
+ 'name': 'secret_b'}}}],
+ 'envFrom': [{'configMapRef': {'name': 'configmap_a'}},
+ {'secretRef': {'name': 'secret_a'}}],
+ 'image': 'busybox',
+ 'name': 'base',
+ 'ports': [{'containerPort': 1234, 'name': 'foo'}],
+ 'resources': {'limits': {'memory': '200Mi'},
+ 'requests': {'memory': '100Mi'}},
+ 'volumeMounts': [{'mountPath': '/airflow/xcom',
+ 'name': 'xcom'},
+ {'mountPath': '/etc/foo',
+ 'name': 'secretvol' + str(static_uuid),
+ 'readOnly': True}]},
+ {'command': ['sh',
+ '-c',
+ 'trap "exit 0" INT; while true; do sleep '
+ '30; done;'],
+ 'image': 'alpine',
+ 'name': 'airflow-xcom-sidecar',
+ 'resources': {'requests': {'cpu': '1m'}},
+ 'volumeMounts': [{'mountPath': '/airflow/xcom',
+ 'name': 'xcom'}]}],
+ 'hostNetwork': True,
+ 'imagePullSecrets': [{'name': 'pull_secret_a'},
+ {'name': 'pull_secret_b'}],
+ 'securityContext': {'fsGroup': 2000, 'runAsUser': 1000},
+ 'volumes': [{'emptyDir': {}, 'name': 'xcom'},
+ {'name': 'secretvol' + str(static_uuid),
+ 'secret': {'secretName': 'secret_b'}}]}})
diff --git a/tests/kubernetes/pod_generator_base.yaml b/tests/kubernetes/pod_generator_base.yaml
new file mode 100644
index 0000000..2a8c8c5
--- /dev/null
+++ b/tests/kubernetes/pod_generator_base.yaml
@@ -0,0 +1,52 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+---
+apiVersion: v1
+kind: Pod
+metadata:
+ labels:
+ app: myapp
+ name: myapp-pod
+ namespace: default
+spec:
+ containers:
+ - name: base
+ image: busybox
+ env:
+ - name: ENVIRONMENT
+ value: 'prod'
+ - name: LOG_LEVEL
+ value: 'warning'
+ envFrom:
+ - configMapRef:
+ name: configmap_a
+ ports:
+ - name: foo
+ containerPort: 1234
+ resources:
+ limits:
+ memory: "200Mi"
+ requests:
+ memory: "100Mi"
+ command: ['sh', '-c', 'echo Hello Kubernetes!']
+ securityContext:
+ fsGroup: 2000
+ runAsUser: 1000
+ hostNetwork: False,
+ imagePullSecrets:
+ - name: pull_secret_a
+ - name: pull_secret_b
diff --git a/tests/kubernetes/pod_generator_base_with_secrets.yaml b/tests/kubernetes/pod_generator_base_with_secrets.yaml
new file mode 100644
index 0000000..3e71214
--- /dev/null
+++ b/tests/kubernetes/pod_generator_base_with_secrets.yaml
@@ -0,0 +1,61 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+---
+apiVersion: v1
+kind: Pod
+metadata:
+ labels:
+ app: myapp
+ name: myapp-pod
+ namespace: default
+spec:
+ containers:
+ - name: base
+ image: busybox
+ env:
+ - name: ENVIRONMENT
+ value: 'prod'
+ - name: LOG_LEVEL
+ value: 'warning'
+ - name: TARGET
+ valueFrom:
+ secretKeyRef:
+ name: secret_b
+ key: source_b
+ envFrom:
+ - configMapRef:
+ name: configmap_a
+ - configMapRef:
+ name: configmap_b
+ - secretRef:
+ name: secret_a
+ ports:
+ - name: foo
+ containerPort: 1234
+ resources:
+ limits:
+ memory: "200Mi"
+ requests:
+ memory: "100Mi"
+ command: ['sh', '-c', 'echo Hello Kubernetes!']
+ securityContext:
+ fsGroup: 2000
+ runAsUser: 1000
+ hostNetwork: False,
+ imagePullSecrets:
+ - name: pull_secret_a
+ - name: pull_secret_b
diff --git a/tests/kubernetes/test_pod_generator.py b/tests/kubernetes/test_pod_generator.py
index 9e96af9..be961e4 100644
--- a/tests/kubernetes/test_pod_generator.py
+++ b/tests/kubernetes/test_pod_generator.py
@@ -23,8 +23,6 @@ from dateutil import parser
from kubernetes.client import ApiClient, models as k8s
from airflow.exceptions import AirflowConfigException
-from airflow.kubernetes.k8s_model import append_to_pod
-from airflow.kubernetes.pod import Resources
from airflow.kubernetes.pod_generator import (
PodDefaults, PodGenerator, datetime_to_label_safe_datestring, extend_object_field, merge_objects,
)
@@ -77,7 +75,7 @@ class TestPodGenerator(unittest.TestCase):
'execution_date': self.execution_date_label,
'task_id': self.task_id,
'try_number': str(self.try_number),
- 'airflow_version': mock.ANY,
+ 'airflow_version': '2.0.0.dev0',
'kubernetes_executor': 'True'
}
self.annotations = {
@@ -93,141 +91,115 @@ class TestPodGenerator(unittest.TestCase):
'annotations': self.annotations,
}
- self.resources = Resources('1Gi', 1, '2Gi', '2Gi', 2, 1, '4Gi')
- self.k8s_client = ApiClient()
- self.expected = {
- 'apiVersion': 'v1',
- 'kind': 'Pod',
- 'metadata': {
- 'name': 'myapp-pod-' + self.static_uuid.hex,
- 'labels': {'app': 'myapp'},
- 'namespace': 'default'
+ self.resources = k8s.V1ResourceRequirements(
+ requests={
+ "cpu": 1,
+ "memory": "1Gi",
+ "ephemeral-storage": "2Gi",
},
- 'spec': {
- 'containers': [{
- 'name': 'base',
- 'image': 'busybox',
- 'args': [],
- 'command': [
- 'sh', '-c', 'echo Hello Kubernetes!'
- ],
- 'env': [{
- 'name': 'ENVIRONMENT',
- 'value': 'prod'
- }, {
- 'name': 'LOG_LEVEL',
- 'value': 'warning'
- }, {
- 'name': 'TARGET',
- 'valueFrom': {
- 'secretKeyRef': {
- 'name': 'secret_b',
- 'key': 'source_b'
- }
- }
- }],
- 'envFrom': [{
- 'configMapRef': {
- 'name': 'configmap_a'
- }
- }, {
- 'configMapRef': {
- 'name': 'configmap_b'
- }
- }, {
- 'secretRef': {
- 'name': 'secret_a'
- }
- }],
- 'resources': {
- 'requests': {
- 'memory': '1Gi',
- 'cpu': 1,
- 'ephemeral-storage': '2Gi'
- },
- 'limits': {
- 'memory': '2Gi',
- 'cpu': 2,
- 'nvidia.com/gpu': 1,
- 'ephemeral-storage': '4Gi'
- },
- },
- 'ports': [{'name': 'foo', 'containerPort': 1234}],
- 'volumeMounts': [{
- 'mountPath': '/etc/foo',
- 'name': 'secretvol' + str(self.static_uuid),
- 'readOnly': True
- }]
- }],
- 'volumes': [{
- 'name': 'secretvol' + str(self.static_uuid),
- 'secret': {
- 'secretName': 'secret_b'
- }
- }],
- 'hostNetwork': False,
- 'imagePullSecrets': [
- {'name': 'pull_secret_a'},
- {'name': 'pull_secret_b'}
- ],
- 'securityContext': {
- 'runAsUser': 1000,
- 'fsGroup': 2000,
- },
+ limits={
+ "cpu": 2,
+ "memory": "2Gi",
+ "ephemeral-storage": "4Gi",
+ 'nvidia.com/gpu': 1
}
- }
+ )
- @mock.patch('uuid.uuid4')
- def test_gen_pod(self, mock_uuid):
- mock_uuid.return_value = self.static_uuid
- pod_generator = PodGenerator(
- labels={'app': 'myapp'},
- name='myapp-pod',
- image_pull_secrets='pull_secret_a,pull_secret_b',
- image='busybox',
- envs=self.envs,
- cmds=['sh', '-c', 'echo Hello Kubernetes!'],
- security_context=k8s.V1PodSecurityContext(
- run_as_user=1000,
- fs_group=2000,
+ self.k8s_client = ApiClient()
+ self.expected = k8s.V1Pod(
+ api_version="v1",
+ kind="Pod",
+ metadata=k8s.V1ObjectMeta(
+ namespace="default",
+ name='myapp-pod-' + self.static_uuid.hex,
+ labels={'app': 'myapp'},
+ ),
+ spec=k8s.V1PodSpec(
+ containers=[
+ k8s.V1Container(
+ name='base',
+ image='busybox',
+ command=[
+ 'sh', '-c', 'echo Hello Kubernetes!'
+ ],
+ env=[
+ k8s.V1EnvVar(
+ name='ENVIRONMENT',
+ value='prod'
+ ),
+ k8s.V1EnvVar(
+ name="LOG_LEVEL",
+ value='warning',
+ ),
+ k8s.V1EnvVar(
+ name='TARGET',
+ value_from=k8s.V1EnvVarSource(
+ secret_key_ref=k8s.V1SecretKeySelector(
+ name='secret_b',
+ key='source_b'
+ )
+ ),
+ )
+ ],
+ env_from=[
+ k8s.V1EnvFromSource(
+ config_map_ref=k8s.V1ConfigMapEnvSource(
+ name='configmap_a'
+ )
+ ),
+ k8s.V1EnvFromSource(
+ config_map_ref=k8s.V1ConfigMapEnvSource(
+ name='configmap_b'
+ )
+ ),
+ k8s.V1EnvFromSource(
+ secret_ref=k8s.V1SecretEnvSource(
+ name='secret_a'
+ )
+ ),
+ ],
+ ports=[
+ k8s.V1ContainerPort(
+ name="foo",
+ container_port=1234
+ )
+ ],
+ resources=k8s.V1ResourceRequirements(
+ requests={
+ 'memory': '100Mi'
+ },
+ limits={
+ 'memory': '200Mi',
+ }
+ )
+ )
+ ],
+ security_context=k8s.V1PodSecurityContext(
+ fs_group=2000,
+ run_as_user=1000,
+ ),
+ host_network=True,
+ image_pull_secrets=[
+ k8s.V1LocalObjectReference(
+ name="pull_secret_a"
+ ),
+ k8s.V1LocalObjectReference(
+ name="pull_secret_b"
+ )
+ ]
),
- namespace='default',
- ports=[k8s.V1ContainerPort(name='foo', container_port=1234)],
- configmaps=['configmap_a', 'configmap_b']
- )
- result = pod_generator.gen_pod()
- result = append_to_pod(result, self.secrets)
- result = self.resources.attach_to_pod(result)
- result_dict = self.k8s_client.sanitize_for_serialization(result)
- # sort
- result_dict['spec']['containers'][0]['env'].sort(key=lambda x: x['name'])
- result_dict['spec']['containers'][0]['envFrom'].sort(
- key=lambda x: list(x.values())[0]['name']
)
- self.assertDictEqual(self.expected, result_dict)
@mock.patch('uuid.uuid4')
def test_gen_pod_extract_xcom(self, mock_uuid):
mock_uuid.return_value = self.static_uuid
+ path = sys.path[0] + '/tests/kubernetes/pod_generator_base_with_secrets.yaml'
+
pod_generator = PodGenerator(
- labels={'app': 'myapp'},
- name='myapp-pod',
- image_pull_secrets='pull_secret_a,pull_secret_b',
- image='busybox',
- envs=self.envs,
- cmds=['sh', '-c', 'echo Hello Kubernetes!'],
- namespace='default',
- security_context=k8s.V1PodSecurityContext(
- run_as_user=1000,
- fs_group=2000,
- ),
- ports=[k8s.V1ContainerPort(name='foo', container_port=1234)],
- configmaps=['configmap_a', 'configmap_b'],
+ pod_template_file=path,
extract_xcom=True
)
result = pod_generator.gen_pod()
- result = append_to_pod(result, self.secrets)
- result = self.resources.attach_to_pod(result)
result_dict = self.k8s_client.sanitize_for_serialization(result)
container_two = {
'name': 'airflow-xcom-sidecar',
@@ -241,18 +213,83 @@ class TestPodGenerator(unittest.TestCase):
],
'resources': {'requests': {'cpu': '1m'}},
}
- self.expected['spec']['containers'].append(container_two)
- self.expected['spec']['containers'][0]['volumeMounts'].insert(0, {
- 'name': 'xcom',
- 'mountPath': '/airflow/xcom'
- })
- self.expected['spec']['volumes'].insert(0, {
- 'name': 'xcom', 'emptyDir': {}
- })
- result_dict['spec']['containers'][0]['env'].sort(key=lambda x: x['name'])
- self.assertEqual(result_dict, self.expected)
+ self.expected.spec.containers.append(container_two)
+ base_container: k8s.V1Container = self.expected.spec.containers[0]
+ base_container.volume_mounts = base_container.volume_mounts or []
+ base_container.volume_mounts.append(k8s.V1VolumeMount(
+ name="xcom",
+ mount_path="/airflow/xcom"
+ ))
+ self.expected.spec.containers[0] = base_container
+ self.expected.spec.volumes = self.expected.spec.volumes or []
+ self.expected.spec.volumes.append(
+ k8s.V1Volume(
+ name='xcom',
+ empty_dir={},
+ )
+ )
+ result_dict = self.k8s_client.sanitize_for_serialization(result)
+ expected_dict = self.k8s_client.sanitize_for_serialization(self.expected)
+
+ self.assertEqual(result_dict, expected_dict)
def test_from_obj(self):
+ result = PodGenerator.from_obj(
+ {
+ "pod_override": k8s.V1Pod(
+ api_version="v1",
+ kind="Pod",
+ metadata=k8s.V1ObjectMeta(
+ name="foo",
+ annotations={"test": "annotation"}
+ ),
+ spec=k8s.V1PodSpec(
+ containers=[
+ k8s.V1Container(
+ name="base",
+ volume_mounts=[
+ k8s.V1VolumeMount(
+ mount_path="/foo/",
+ name="example-kubernetes-test-volume"
+ )
+ ]
+ )
+ ],
+ volumes=[
+ k8s.V1Volume(
+ name="example-kubernetes-test-volume",
+ host_path=k8s.V1HostPathVolumeSource(
+ path="/tmp/"
+ )
+ )
+ ]
+ )
+ )
+ }
+ )
+ result = self.k8s_client.sanitize_for_serialization(result)
+
+ self.assertEqual({
+ 'apiVersion': 'v1',
+ 'kind': 'Pod',
+ 'metadata': {
+ 'name': 'foo',
+ 'annotations': {'test': 'annotation'},
+ },
+ 'spec': {
+ 'containers': [{
+ 'name': 'base',
+ 'volumeMounts': [{
+ 'mountPath': '/foo/',
+ 'name': 'example-kubernetes-test-volume'
+ }],
+ }],
+ 'volumes': [{
+ 'hostPath': {'path': '/tmp/'},
+ 'name': 'example-kubernetes-test-volume'
+ }],
+ }
+ }, result)
result = PodGenerator.from_obj({
"KubernetesExecutor": {
"annotations": {"test": "annotation"},
@@ -343,22 +380,13 @@ class TestPodGenerator(unittest.TestCase):
@mock.patch('uuid.uuid4')
def test_reconcile_pods_empty_mutator_pod(self, mock_uuid):
mock_uuid.return_value = self.static_uuid
- base_pod = PodGenerator(
- image='image1',
- name='name1',
- envs={'key1': 'val1'},
- cmds=['/bin/command1.sh', 'arg1'],
- ports=[k8s.V1ContainerPort(name='port', container_port=2118)],
- volumes=[{
- 'hostPath': {'path': '/tmp/'},
- 'name': 'example-kubernetes-test-volume1'
- }],
- volume_mounts=[{
- 'mountPath': '/foo/',
- 'name': 'example-kubernetes-test-volume1'
- }],
- ).gen_pod()
+ path = sys.path[0] + '/tests/kubernetes/pod_generator_base_with_secrets.yaml'
+ pod_generator = PodGenerator(
+ pod_template_file=path,
+ extract_xcom=True
+ )
+ base_pod = pod_generator.gen_pod()
mutator_pod = None
name = 'name1-' + self.static_uuid.hex
@@ -374,147 +402,64 @@ class TestPodGenerator(unittest.TestCase):
@mock.patch('uuid.uuid4')
def test_reconcile_pods(self, mock_uuid):
mock_uuid.return_value = self.static_uuid
- base_pod = PodGenerator(
- image='image1',
- name='name1',
- envs={'key1': 'val1'},
- cmds=['/bin/command1.sh', 'arg1'],
- ports=[k8s.V1ContainerPort(name='port', container_port=2118)],
- volumes=[{
- 'hostPath': {'path': '/tmp/'},
- 'name': 'example-kubernetes-test-volume1'
- }],
- labels={"foo": "bar"},
- volume_mounts=[{
- 'mountPath': '/foo/',
- 'name': 'example-kubernetes-test-volume1'
- }],
- ).gen_pod()
+ path = sys.path[0] + '/tests/kubernetes/pod_generator_base_with_secrets.yaml'
- mutator_pod = PodGenerator(
- envs={'key2': 'val2'},
- image='',
- name='name2',
- labels={"bar": "baz"},
- cmds=['/bin/command2.sh', 'arg2'],
- volumes=[{
- 'hostPath': {'path': '/tmp/'},
- 'name': 'example-kubernetes-test-volume2'
- }],
- volume_mounts=[{
- 'mountPath': '/foo/',
- 'name': 'example-kubernetes-test-volume2'
- }]
+ base_pod = PodGenerator(
+ pod_template_file=path,
+ extract_xcom=False
).gen_pod()
- result = PodGenerator.reconcile_pods(base_pod, mutator_pod)
- result = self.k8s_client.sanitize_for_serialization(result)
- self.assertEqual({
- 'apiVersion': 'v1',
- 'kind': 'Pod',
- 'metadata': {'name': 'name2-' + self.static_uuid.hex,
- 'labels': {'foo': 'bar', "bar": "baz"}},
- 'spec': {
- 'containers': [{
- 'args': [],
- 'command': ['/bin/command2.sh', 'arg2'],
- 'env': [
- {'name': 'key1', 'value': 'val1'},
- {'name': 'key2', 'value': 'val2'}
- ],
- 'envFrom': [],
- 'image': 'image1',
- 'name': 'base',
- 'ports': [{
- 'containerPort': 2118,
- 'name': 'port',
- }],
- 'volumeMounts': [{
- 'mountPath': '/foo/',
- 'name': 'example-kubernetes-test-volume1'
- }, {
- 'mountPath': '/foo/',
- 'name': 'example-kubernetes-test-volume2'
- }]
- }],
- 'hostNetwork': False,
- 'imagePullSecrets': [],
- 'volumes': [{
- 'hostPath': {'path': '/tmp/'},
- 'name': 'example-kubernetes-test-volume1'
- }, {
- 'hostPath': {'path': '/tmp/'},
- 'name': 'example-kubernetes-test-volume2'
- }]
- }
- }, result)
-
- @mock.patch('uuid.uuid4')
- def test_construct_pod_empty_worker_config(self, mock_uuid):
- mock_uuid.return_value = self.static_uuid
- executor_config = k8s.V1Pod(
+ mutator_pod = k8s.V1Pod(
+ metadata=k8s.V1ObjectMeta(
+ name="name2",
+ labels={"bar": "baz"},
+ ),
spec=k8s.V1PodSpec(
- containers=[
- k8s.V1Container(
- name='',
- resources=k8s.V1ResourceRequirements(
- limits={
- 'cpu': '1m',
- 'memory': '1G'
- }
- )
- )
+ containers=[k8s.V1Container(
+ image='',
+ name='name',
+ command=['/bin/command2.sh', 'arg2'],
+ volume_mounts=[k8s.V1VolumeMount(mount_path="/foo/",
+ name="example-kubernetes-test-volume2")]
+ )],
+ volumes=[
+ k8s.V1Volume(host_path=k8s.V1HostPathVolumeSource(path="/tmp/"),
+ name="example-kubernetes-test-volume2")
]
)
+
)
- worker_config = k8s.V1Pod()
- result = PodGenerator.construct_pod(
- self.dag_id,
- self.task_id,
- 'pod_id',
- self.try_number,
- "kube_image",
- self.execution_date,
- ['command'],
- executor_config,
- worker_config,
- 'namespace',
- 'uuid',
+ result = PodGenerator.reconcile_pods(base_pod, mutator_pod)
+ expected: k8s.V1Pod = self.expected
+ expected.metadata.name = "name2"
+ expected.metadata.labels['bar'] = 'baz'
+ expected.spec.volumes = expected.spec.volumes or []
+ expected.spec.volumes.append(
+ k8s.V1Volume(host_path=k8s.V1HostPathVolumeSource(path="/tmp/"),
+ name="example-kubernetes-test-volume2")
)
- 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': [],
- 'name': 'base',
- 'image': 'kube_image',
- 'ports': [],
- 'resources': {
- 'limits': {
- 'cpu': '1m',
- 'memory': '1G'
- }
- },
- 'volumeMounts': []
- }],
- 'hostNetwork': False,
- 'imagePullSecrets': [],
- 'volumes': []
- }
- }, sanitized_result)
+ base_container: k8s.V1Container = expected.spec.containers[0]
+ base_container.command = ['/bin/command2.sh', 'arg2']
+ base_container.volume_mounts = [
+ k8s.V1VolumeMount(mount_path="/foo/",
+ name="example-kubernetes-test-volume2")
+ ]
+ base_container.name = "name"
+ expected.spec.containers[0] = base_container
+
+ result_dict = self.k8s_client.sanitize_for_serialization(result)
+ expected_dict = self.k8s_client.sanitize_for_serialization(expected)
+
+ self.assertEqual(result_dict, expected_dict)
@mock.patch('uuid.uuid4')
- def test_construct_pod_empty_execuctor_config(self, mock_uuid):
+ def test_construct_pod(self, mock_uuid):
+ path = sys.path[0] + '/tests/kubernetes/pod_generator_base_with_secrets.yaml'
+ worker_config = PodGenerator.deserialize_model_file(path)
mock_uuid.return_value = self.static_uuid
- worker_config = k8s.V1Pod(
+ executor_config = k8s.V1Pod(
spec=k8s.V1PodSpec(
containers=[
k8s.V1Container(
@@ -529,138 +474,66 @@ class TestPodGenerator(unittest.TestCase):
]
)
)
- executor_config = None
result = PodGenerator.construct_pod(
- self.dag_id,
- self.task_id,
- 'pod_id',
- self.try_number,
- "kube_image",
- self.execution_date,
- ['command'],
- executor_config,
- worker_config,
- 'namespace',
- 'uuid',
+ dag_id=self.dag_id,
+ task_id=self.task_id,
+ pod_id='pod_id',
+ kube_image='airflow_image',
+ try_number=self.try_number,
+ date=self.execution_date,
+ command=['command'],
+ pod_override_object=executor_config,
+ base_worker_pod=worker_config,
+ namespace='test_namespace',
+ worker_uuid='uuid',
)
- sanitized_result = self.k8s_client.sanitize_for_serialization(result)
+ expected = self.expected
+ expected.metadata.labels = self.labels
+ expected.metadata.labels['app'] = 'myapp'
+ expected.metadata.annotations = self.annotations
+ expected.metadata.name = 'pod_id-' + self.static_uuid.hex
+ expected.metadata.namespace = 'test_namespace'
+ expected.spec.containers[0].command = ['command']
+ expected.spec.containers[0].image = 'airflow_image'
+ expected.spec.containers[0].resources = {'limits': {'cpu': '1m',
+ 'memory': '1G'}
+ }
+ result_dict = self.k8s_client.sanitize_for_serialization(result)
+ expected_dict = self.k8s_client.sanitize_for_serialization(self.expected)
- self.assertEqual({
- 'apiVersion': 'v1',
- 'kind': 'Pod',
- 'metadata': self.metadata,
- 'spec': {
- 'containers': [{
- 'args': [],
- 'command': ['command'],
- 'env': [],
- 'envFrom': [],
- 'name': 'base',
- 'image': 'kube_image',
- 'ports': [],
- 'resources': {
- 'limits': {
- 'cpu': '1m',
- 'memory': '1G'
- }
- },
- 'volumeMounts': []
- }],
- 'hostNetwork': False,
- 'imagePullSecrets': [],
- 'volumes': []
- }
- }, sanitized_result)
+ self.assertEqual(expected_dict, result_dict)
@mock.patch('uuid.uuid4')
- def test_construct_pod(self, mock_uuid):
+ def test_construct_pod_empty_executor_config(self, mock_uuid):
+ path = sys.path[0] + '/tests/kubernetes/pod_generator_base_with_secrets.yaml'
+ worker_config = PodGenerator.deserialize_model_file(path)
mock_uuid.return_value = self.static_uuid
- worker_config = k8s.V1Pod(
- metadata=k8s.V1ObjectMeta(
- name='gets-overridden-by-dynamic-args',
- annotations={
- 'should': 'stay'
- }
- ),
- spec=k8s.V1PodSpec(
- containers=[
- k8s.V1Container(
- name='doesnt-override',
- resources=k8s.V1ResourceRequirements(
- limits={
- 'cpu': '1m',
- 'memory': '1G'
- }
- ),
- security_context=k8s.V1SecurityContext(
- run_as_user=1
- )
- )
- ]
- )
- )
- executor_config = k8s.V1Pod(
- spec=k8s.V1PodSpec(
- containers=[
- k8s.V1Container(
- name='doesnt-override-either',
- resources=k8s.V1ResourceRequirements(
- limits={
- 'cpu': '2m',
- 'memory': '2G'
- }
- )
- )
- ]
- )
- )
+ executor_config = None
result = PodGenerator.construct_pod(
- self.dag_id,
- self.task_id,
- 'pod_id',
- self.try_number,
- "kube_image",
- self.execution_date,
- ['command'],
- executor_config,
- worker_config,
- 'namespace',
- 'uuid',
+ dag_id='dag_id',
+ task_id='task_id',
+ pod_id='pod_id',
+ kube_image='test-image',
+ try_number=3,
+ date=self.execution_date,
+ command=['command'],
+ pod_override_object=executor_config,
+ base_worker_pod=worker_config,
+ namespace='namespace',
+ worker_uuid='uuid',
)
sanitized_result = self.k8s_client.sanitize_for_serialization(result)
-
- expected_metadata = dict(self.metadata)
- expected_metadata['annotations'].update({'should': 'stay'})
-
- self.assertEqual({
- 'apiVersion': 'v1',
- 'kind': 'Pod',
- 'metadata': expected_metadata,
- 'spec': {
- 'containers': [{
- 'args': [],
- 'command': ['command'],
- 'env': [],
- 'envFrom': [],
- 'name': 'base',
- 'image': 'kube_image',
- 'ports': [],
- 'resources': {
- 'limits': {
- 'cpu': '2m',
- 'memory': '2G'
- }
- },
- 'volumeMounts': [],
- 'securityContext': {'runAsUser': 1}
- }],
- 'hostNetwork': False,
- 'imagePullSecrets': [],
- 'volumes': []
- }
- }, sanitized_result)
+ worker_config.spec.containers[0].image = "test-image"
+ worker_config.spec.containers[0].command = ["command"]
+ worker_config.metadata.annotations = self.annotations
+ worker_config.metadata.labels = self.labels
+ worker_config.metadata.labels['app'] = 'myapp'
+ worker_config.metadata.name = 'pod_id-' + self.static_uuid.hex
+ worker_config.metadata.namespace = 'namespace'
+ worker_config_result = self.k8s_client.sanitize_for_serialization(worker_config)
+ self.assertEqual(worker_config_result, sanitized_result)
def test_merge_objects_empty(self):
annotations = {'foo1': 'bar1'}
@@ -793,9 +666,8 @@ class TestPodGenerator(unittest.TestCase):
self.assertEqual(client_spec, res)
def test_deserialize_model_file(self):
-
- fixture = sys.path[0] + '/tests/kubernetes/pod.yaml'
- result = PodGenerator.deserialize_model_file(fixture)
+ path = sys.path[0] + '/tests/kubernetes/pod.yaml'
+ result = PodGenerator.deserialize_model_file(path)
sanitized_res = self.k8s_client.sanitize_for_serialization(result)
self.assertEqual(sanitized_res, self.deserialize_result)
@@ -824,12 +696,8 @@ spec:
def test_validate_pod_generator(self):
with self.assertRaises(AirflowConfigException):
- PodGenerator(image='k', pod=k8s.V1Pod())
- with self.assertRaises(AirflowConfigException):
PodGenerator(pod=k8s.V1Pod(), pod_template_file='k')
with self.assertRaises(AirflowConfigException):
- PodGenerator(image='k', pod_template_file='k')
-
- PodGenerator(image='k')
+ PodGenerator()
PodGenerator(pod_template_file='tests/kubernetes/pod.yaml')
PodGenerator(pod=k8s.V1Pod())
diff --git a/tests/kubernetes/test_worker_configuration.py b/tests/kubernetes/test_worker_configuration.py
deleted file mode 100644
index a74b811..0000000
--- a/tests/kubernetes/test_worker_configuration.py
+++ /dev/null
@@ -1,930 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-#
-import sys
-import unittest
-from unittest.mock import ANY
-
-import mock
-from dateutil import parser
-from parameterized import parameterized
-
-from tests.test_utils.config import conf_vars
-
-try:
- from kubernetes.client import models as k8s
- from kubernetes.client.api_client import ApiClient
-
- from airflow.exceptions import AirflowConfigException
- from airflow.executors.kubernetes_executor import AirflowKubernetesScheduler, KubeConfig
- from airflow.kubernetes.pod_generator import PodGenerator, datetime_to_label_safe_datestring
- from airflow.kubernetes.secret import Secret
- from airflow.kubernetes.worker_configuration import WorkerConfiguration
- from airflow.version import version as airflow_version
-except ImportError:
- AirflowKubernetesScheduler = None # type: ignore
-
-
-class TestKubernetesWorkerConfiguration(unittest.TestCase):
- """
- Tests that if dags_volume_subpath/logs_volume_subpath configuration
- options are passed to worker pod config
- """
-
- affinity_config = {
- 'podAntiAffinity': {
- 'requiredDuringSchedulingIgnoredDuringExecution': [
- {
- 'topologyKey': 'kubernetes.io/hostname',
- 'labelSelector': {
- 'matchExpressions': [
- {
- 'key': 'app',
- 'operator': 'In',
- 'values': ['airflow']
- }
- ]
- }
- }
- ]
- }
- }
-
- tolerations_config = [
- {
- 'key': 'dedicated',
- 'operator': 'Equal',
- 'value': 'airflow'
- },
- {
- 'key': 'prod',
- 'operator': 'Exists'
- }
- ]
-
- worker_annotations_config = {
- 'iam.amazonaws.com/role': 'role-arn',
- 'other/annotation': 'value'
- }
-
- def setUp(self):
- if AirflowKubernetesScheduler is None:
- self.skipTest("kubernetes python package is not installed")
-
- self.kube_config = mock.MagicMock()
- self.kube_config.airflow_home = '/'
- self.kube_config.airflow_dags = 'dags'
- self.kube_config.airflow_logs = 'logs'
- self.kube_config.dags_volume_subpath = None
- self.kube_config.dags_volume_mount_point = None
- self.kube_config.logs_volume_subpath = None
- self.kube_config.dags_in_image = False
- self.kube_config.dags_folder = None
- self.kube_config.git_dags_folder_mount_point = None
- self.kube_config.kube_labels = {'dag_id': 'original_dag_id', 'my_label': 'label_id'}
- self.kube_config.pod_template_file = ''
- self.kube_config.restart_policy = ''
- self.kube_config.image_pull_policy = ''
- self.api_client = ApiClient()
-
- def tearDown(self) -> None:
- self.kube_config = None
-
- def test_worker_configuration_no_subpaths(self):
- self.kube_config.dags_volume_claim = 'airflow-dags'
- self.kube_config.dags_folder = 'dags'
- worker_config = WorkerConfiguration(self.kube_config)
- volumes = worker_config._get_volumes()
- volume_mounts = worker_config._get_volume_mounts()
- for volume_or_mount in volumes + volume_mounts:
- if volume_or_mount.name not in ['airflow-config', 'airflow-local-settings']:
- self.assertNotIn(
- 'subPath', self.api_client.sanitize_for_serialization(volume_or_mount),
- "subPath shouldn't be defined"
- )
-
- @conf_vars({
- ('kubernetes', 'git_ssh_known_hosts_configmap_name'): 'airflow-configmap',
- ('kubernetes', 'git_ssh_key_secret_name'): 'airflow-secrets',
- ('kubernetes', 'git_user'): 'some-user',
- ('kubernetes', 'git_password'): 'some-password',
- ('kubernetes', 'git_repo'): 'git@github.com:apache/airflow.git',
- ('kubernetes', 'git_branch'): 'master',
- ('kubernetes', 'git_dags_folder_mount_point'): '/usr/local/airflow/dags',
- ('kubernetes', 'delete_worker_pods'): 'True',
- ('kubernetes', 'kube_client_request_args'): '{"_request_timeout" : [60,360]}',
- })
- def test_worker_configuration_auth_both_ssh_and_user(self):
- with self.assertRaisesRegex(AirflowConfigException,
- 'either `git_user` and `git_password`.*'
- 'or `git_ssh_key_secret_name`.*'
- 'but not both$'):
- KubeConfig()
-
- @parameterized.expand([
- ('{"grace_period_seconds": 10}', {"grace_period_seconds": 10}),
- ("", {})
- ])
- def test_delete_option_kwargs_config(self, config, expected_value):
- with conf_vars({
- ('kubernetes', 'delete_option_kwargs'): config,
- }):
- self.assertEqual(KubeConfig().delete_option_kwargs, expected_value)
-
- def test_worker_with_subpaths(self):
- self.kube_config.dags_volume_subpath = 'dags'
- self.kube_config.logs_volume_subpath = 'logs'
- self.kube_config.dags_volume_claim = 'dags'
- self.kube_config.dags_folder = 'dags'
- worker_config = WorkerConfiguration(self.kube_config)
- volumes = worker_config._get_volumes()
- volume_mounts = worker_config._get_volume_mounts()
-
- for volume in volumes:
- self.assertNotIn(
- 'subPath', self.api_client.sanitize_for_serialization(volume),
- "subPath isn't valid configuration for a volume"
- )
-
- for volume_mount in volume_mounts:
- if volume_mount.name != 'airflow-config':
- self.assertIn(
- 'subPath', self.api_client.sanitize_for_serialization(volume_mount),
- "subPath should've been passed to volumeMount configuration"
- )
-
- def test_worker_generate_dag_volume_mount_path(self):
- self.kube_config.git_dags_folder_mount_point = '/root/airflow/git/dags'
- self.kube_config.dags_folder = '/root/airflow/dags'
- worker_config = WorkerConfiguration(self.kube_config)
-
- self.kube_config.dags_volume_claim = 'airflow-dags'
- self.kube_config.dags_volume_host = ''
- dag_volume_mount_path = worker_config.generate_dag_volume_mount_path()
- self.assertEqual(dag_volume_mount_path, self.kube_config.dags_folder)
-
- self.kube_config.dags_volume_mount_point = '/root/airflow/package'
- dag_volume_mount_path = worker_config.generate_dag_volume_mount_path()
- self.assertEqual(dag_volume_mount_path, '/root/airflow/package')
- self.kube_config.dags_volume_mount_point = ''
-
- self.kube_config.dags_volume_claim = ''
- self.kube_config.dags_volume_host = '/host/airflow/dags'
- dag_volume_mount_path = worker_config.generate_dag_volume_mount_path()
- self.assertEqual(dag_volume_mount_path, self.kube_config.dags_folder)
-
- self.kube_config.dags_volume_claim = ''
- self.kube_config.dags_volume_host = ''
- dag_volume_mount_path = worker_config.generate_dag_volume_mount_path()
- self.assertEqual(dag_volume_mount_path,
- self.kube_config.git_dags_folder_mount_point)
-
- def test_worker_environment_no_dags_folder(self):
- self.kube_config.airflow_configmap = ''
- self.kube_config.git_dags_folder_mount_point = ''
- self.kube_config.dags_folder = ''
- worker_config = WorkerConfiguration(self.kube_config)
- env = worker_config._get_environment()
-
- self.assertNotIn('AIRFLOW__CORE__DAGS_FOLDER', env)
-
- def test_worker_environment_when_dags_folder_specified(self):
- self.kube_config.airflow_configmap = 'airflow-configmap'
- self.kube_config.git_dags_folder_mount_point = ''
- dags_folder = '/workers/path/to/dags'
- self.kube_config.dags_folder = dags_folder
-
- worker_config = WorkerConfiguration(self.kube_config)
- env = worker_config._get_environment()
-
- self.assertEqual(dags_folder, env['AIRFLOW__CORE__DAGS_FOLDER'])
-
- def test_worker_environment_dags_folder_using_git_sync(self):
- self.kube_config.airflow_configmap = 'airflow-configmap'
- self.kube_config.git_sync_dest = 'repo'
- self.kube_config.git_subpath = 'dags'
- self.kube_config.git_dags_folder_mount_point = '/workers/path/to/dags'
-
- dags_folder = '{}/{}/{}'.format(self.kube_config.git_dags_folder_mount_point,
- self.kube_config.git_sync_dest,
- self.kube_config.git_subpath)
-
- worker_config = WorkerConfiguration(self.kube_config)
- env = worker_config._get_environment()
-
- self.assertEqual(dags_folder, env['AIRFLOW__CORE__DAGS_FOLDER'])
-
- def test_init_environment_using_git_sync_ssh_without_known_hosts(self):
- # Tests the init environment created with git-sync SSH authentication option is correct
- # without known hosts file
- self.kube_config.airflow_configmap = 'airflow-configmap'
- self.kube_config.git_ssh_secret_name = 'airflow-secrets'
- self.kube_config.git_ssh_known_hosts_configmap_name = None
- self.kube_config.dags_volume_claim = None
- self.kube_config.dags_volume_host = None
- self.kube_config.dags_in_image = None
-
- worker_config = WorkerConfiguration(self.kube_config)
- init_containers = worker_config._get_init_containers()
-
- self.assertTrue(init_containers) # check not empty
- env = init_containers[0].env
-
- self.assertIn(k8s.V1EnvVar(name='GIT_SSH_KEY_FILE', value='/etc/git-secret/ssh'), env)
- self.assertIn(k8s.V1EnvVar(name='GIT_SYNC_ADD_USER', value='true'), env)
- self.assertIn(k8s.V1EnvVar(name='GIT_KNOWN_HOSTS', value='false'), env)
- self.assertIn(k8s.V1EnvVar(name='GIT_SYNC_SSH', value='true'), env)
-
- def test_init_environment_using_git_sync_ssh_with_known_hosts(self):
- # Tests the init environment created with git-sync SSH authentication option is correct
- # with known hosts file
- self.kube_config.airflow_configmap = 'airflow-configmap'
- self.kube_config.git_ssh_key_secret_name = 'airflow-secrets'
- self.kube_config.git_ssh_known_hosts_configmap_name = 'airflow-configmap'
- self.kube_config.dags_volume_claim = None
- self.kube_config.dags_volume_host = None
- self.kube_config.dags_in_image = None
-
- worker_config = WorkerConfiguration(self.kube_config)
- init_containers = worker_config._get_init_containers()
-
- self.assertTrue(init_containers) # check not empty
- env = init_containers[0].env
-
- self.assertIn(k8s.V1EnvVar(name='GIT_SSH_KEY_FILE', value='/etc/git-secret/ssh'), env)
- self.assertIn(k8s.V1EnvVar(name='GIT_SYNC_ADD_USER', value='true'), env)
- self.assertIn(k8s.V1EnvVar(name='GIT_KNOWN_HOSTS', value='true'), env)
- self.assertIn(k8s.V1EnvVar(
- name='GIT_SSH_KNOWN_HOSTS_FILE',
- value='/etc/git-secret/known_hosts'
- ), env)
- self.assertIn(k8s.V1EnvVar(name='GIT_SYNC_SSH', value='true'), env)
-
- def test_init_environment_using_git_sync_user_without_known_hosts(self):
- # Tests the init environment created with git-sync User authentication option is correct
- # without known hosts file
- self.kube_config.airflow_configmap = 'airflow-configmap'
- self.kube_config.git_user = 'git_user'
- self.kube_config.git_password = 'git_password'
- self.kube_config.git_ssh_known_hosts_configmap_name = None
- self.kube_config.git_ssh_key_secret_name = None
- self.kube_config.dags_volume_claim = None
- self.kube_config.dags_volume_host = None
- self.kube_config.dags_in_image = None
-
- worker_config = WorkerConfiguration(self.kube_config)
- init_containers = worker_config._get_init_containers()
-
- self.assertTrue(init_containers) # check not empty
- env = init_containers[0].env
-
- self.assertNotIn(k8s.V1EnvVar(name='GIT_SSH_KEY_FILE', value='/etc/git-secret/ssh'), env)
- self.assertNotIn(k8s.V1EnvVar(name='GIT_SYNC_ADD_USER', value='true'), env)
- self.assertIn(k8s.V1EnvVar(name='GIT_SYNC_USERNAME', value='git_user'), env)
- self.assertIn(k8s.V1EnvVar(name='GIT_SYNC_PASSWORD', value='git_password'), env)
- self.assertIn(k8s.V1EnvVar(name='GIT_KNOWN_HOSTS', value='false'), env)
- self.assertNotIn(k8s.V1EnvVar(
- name='GIT_SSH_KNOWN_HOSTS_FILE',
- value='/etc/git-secret/known_hosts'
- ), env)
- self.assertNotIn(k8s.V1EnvVar(name='GIT_SYNC_SSH', value='true'), env)
-
- def test_init_environment_using_git_sync_user_with_known_hosts(self):
- # Tests the init environment created with git-sync User authentication option is correct
- # with known hosts file
- self.kube_config.airflow_configmap = 'airflow-configmap'
- self.kube_config.git_user = 'git_user'
- self.kube_config.git_password = 'git_password'
- self.kube_config.git_ssh_known_hosts_configmap_name = 'airflow-configmap'
- self.kube_config.git_ssh_key_secret_name = None
- self.kube_config.dags_volume_claim = None
- self.kube_config.dags_volume_host = None
- self.kube_config.dags_in_image = None
-
- worker_config = WorkerConfiguration(self.kube_config)
- init_containers = worker_config._get_init_containers()
-
- self.assertTrue(init_containers) # check not empty
- env = init_containers[0].env
-
- self.assertNotIn(k8s.V1EnvVar(name='GIT_SSH_KEY_FILE', value='/etc/git-secret/ssh'), env)
- self.assertNotIn(k8s.V1EnvVar(name='GIT_SYNC_ADD_USER', value='true'), env)
- self.assertIn(k8s.V1EnvVar(name='GIT_SYNC_USERNAME', value='git_user'), env)
- self.assertIn(k8s.V1EnvVar(name='GIT_SYNC_PASSWORD', value='git_password'), env)
- self.assertIn(k8s.V1EnvVar(name='GIT_KNOWN_HOSTS', value='true'), env)
- self.assertIn(k8s.V1EnvVar(
- name='GIT_SSH_KNOWN_HOSTS_FILE',
- value='/etc/git-secret/known_hosts'
- ), env)
- self.assertNotIn(k8s.V1EnvVar(name='GIT_SYNC_SSH', value='true'), env)
-
- def test_init_environment_using_git_sync_run_as_user_empty(self):
- # Tests if git_syn_run_as_user is none, then no securityContext created in init container
-
- self.kube_config.dags_volume_claim = None
- self.kube_config.dags_volume_host = None
- self.kube_config.dags_in_image = None
- self.kube_config.git_sync_run_as_user = ''
-
- worker_config = WorkerConfiguration(self.kube_config)
- init_containers = worker_config._get_init_containers()
- self.assertTrue(init_containers) # check not empty
-
- self.assertIsNone(init_containers[0].security_context)
-
- def test_init_environment_using_git_sync_run_as_user_root(self):
- # Tests if git_syn_run_as_user is '0', securityContext is created with
- # the right uid
-
- self.kube_config.dags_volume_claim = None
- self.kube_config.dags_volume_host = None
- self.kube_config.dags_in_image = None
- self.kube_config.git_sync_run_as_user = 0
-
- worker_config = WorkerConfiguration(self.kube_config)
- init_containers = worker_config._get_init_containers()
- self.assertTrue(init_containers) # check not empty
-
- self.assertEqual(0, init_containers[0].security_context.run_as_user)
-
- def test_make_pod_run_as_user_0(self):
- # Tests the pod created with run-as-user 0 actually gets that in it's config
- self.kube_config.worker_run_as_user = 0
- self.kube_config.dags_volume_claim = None
- self.kube_config.dags_volume_host = None
- self.kube_config.dags_in_image = None
- self.kube_config.worker_fs_group = None
- self.kube_config.git_dags_folder_mount_point = 'dags'
- self.kube_config.git_sync_dest = 'repo'
- self.kube_config.git_subpath = 'path'
-
- worker_config = WorkerConfiguration(self.kube_config)
- pod = worker_config.as_pod()
-
- self.assertEqual(0, pod.spec.security_context.run_as_user)
-
- def test_make_pod_assert_labels(self):
- # Tests the pod created has all the expected labels set
- self.kube_config.dags_folder = 'dags'
-
- worker_config = WorkerConfiguration(self.kube_config)
- execution_date = parser.parse('2019-11-21 11:08:22.920875')
- pod = PodGenerator.construct_pod(
- "test_dag_id",
- "test_task_id",
- "test_pod_id",
- 1,
- 'kube_image',
- execution_date,
- ["bash -c 'ls /'"],
- None,
- worker_config.as_pod(),
- "default",
- "sample-uuid",
- )
- expected_labels = {
- 'airflow-worker': 'sample-uuid',
- 'airflow_version': airflow_version.replace('+', '-'),
- 'dag_id': 'test_dag_id',
- 'execution_date': datetime_to_label_safe_datestring(execution_date),
- 'kubernetes_executor': 'True',
- 'my_label': 'label_id',
- 'task_id': 'test_task_id',
- 'try_number': '1'
- }
- self.assertEqual(pod.metadata.labels, expected_labels)
-
- def test_make_pod_git_sync_ssh_without_known_hosts(self):
- # Tests the pod created with git-sync SSH authentication option is correct without known hosts
- self.kube_config.airflow_configmap = 'airflow-configmap'
- self.kube_config.git_ssh_key_secret_name = 'airflow-secrets'
- self.kube_config.dags_volume_claim = None
- self.kube_config.dags_volume_host = None
- self.kube_config.dags_in_image = None
- self.kube_config.worker_fs_group = None
- self.kube_config.git_dags_folder_mount_point = 'dags'
- self.kube_config.git_sync_dest = 'repo'
- self.kube_config.git_subpath = 'path'
-
- worker_config = WorkerConfiguration(self.kube_config)
-
- pod = worker_config.as_pod()
-
- init_containers = worker_config._get_init_containers()
- git_ssh_key_file = next((x.value for x in init_containers[0].env
- if x.name == 'GIT_SSH_KEY_FILE'), None)
- volume_mount_ssh_key = next((x.mount_path for x in init_containers[0].volume_mounts
- if x.name == worker_config.git_sync_ssh_secret_volume_name),
- None)
- self.assertTrue(git_ssh_key_file)
- self.assertTrue(volume_mount_ssh_key)
- self.assertEqual(65533, pod.spec.security_context.fs_group)
- self.assertEqual(git_ssh_key_file,
- volume_mount_ssh_key,
- 'The location where the git ssh secret is mounted'
- ' needs to be the same as the GIT_SSH_KEY_FILE path')
-
- def test_make_pod_git_sync_credentials_secret(self):
- # Tests the pod created with git_sync_credentials_secret will get into the init container
- self.kube_config.git_sync_credentials_secret = 'airflow-git-creds-secret'
- self.kube_config.dags_volume_claim = None
- self.kube_config.dags_volume_host = None
- self.kube_config.dags_in_image = None
- self.kube_config.worker_fs_group = None
- self.kube_config.git_dags_folder_mount_point = 'dags'
- self.kube_config.git_sync_dest = 'repo'
- self.kube_config.git_subpath = 'path'
-
- worker_config = WorkerConfiguration(self.kube_config)
-
- pod = worker_config.as_pod()
-
- username_env = k8s.V1EnvVar(
- name='GIT_SYNC_USERNAME',
- value_from=k8s.V1EnvVarSource(
- secret_key_ref=k8s.V1SecretKeySelector(
- name=self.kube_config.git_sync_credentials_secret,
- key='GIT_SYNC_USERNAME')
- )
- )
- password_env = k8s.V1EnvVar(
- name='GIT_SYNC_PASSWORD',
- value_from=k8s.V1EnvVarSource(
- secret_key_ref=k8s.V1SecretKeySelector(
- name=self.kube_config.git_sync_credentials_secret,
- key='GIT_SYNC_PASSWORD')
- )
- )
-
- self.assertIn(username_env, pod.spec.init_containers[0].env,
- 'The username env for git credentials did not get into the init container')
-
- self.assertIn(password_env, pod.spec.init_containers[0].env,
- 'The password env for git credentials did not get into the init container')
-
- def test_make_pod_git_sync_rev(self):
- # Tests the pod created with git_sync_credentials_secret will get into the init container
- self.kube_config.git_sync_rev = 'sampletag'
- self.kube_config.dags_volume_claim = None
- self.kube_config.dags_volume_host = None
- self.kube_config.dags_in_image = None
- self.kube_config.worker_fs_group = None
- self.kube_config.git_dags_folder_mount_point = 'dags'
- self.kube_config.git_sync_dest = 'repo'
- self.kube_config.git_subpath = 'path'
-
- worker_config = WorkerConfiguration(self.kube_config)
-
- pod = worker_config.as_pod()
-
- rev_env = k8s.V1EnvVar(
- name='GIT_SYNC_REV',
- value=self.kube_config.git_sync_rev,
- )
-
- self.assertIn(rev_env, pod.spec.init_containers[0].env,
- 'The git_sync_rev env did not get into the init container')
-
- def test_make_pod_git_sync_ssh_with_known_hosts(self):
- # Tests the pod created with git-sync SSH authentication option is correct with known hosts
- self.kube_config.airflow_configmap = 'airflow-configmap'
- self.kube_config.git_ssh_secret_name = 'airflow-secrets'
- self.kube_config.dags_volume_claim = None
- self.kube_config.dags_volume_host = None
- self.kube_config.dags_in_image = None
-
- worker_config = WorkerConfiguration(self.kube_config)
-
- init_containers = worker_config._get_init_containers()
- git_ssh_known_hosts_file = next((x.value for x in init_containers[0].env
- if x.name == 'GIT_SSH_KNOWN_HOSTS_FILE'), None)
-
- volume_mount_ssh_known_hosts_file = next(
- (x.mount_path for x in init_containers[0].volume_mounts
- if x.name == worker_config.git_sync_ssh_known_hosts_volume_name),
- None)
- self.assertTrue(git_ssh_known_hosts_file)
- self.assertTrue(volume_mount_ssh_known_hosts_file)
- self.assertEqual(git_ssh_known_hosts_file,
- volume_mount_ssh_known_hosts_file,
- 'The location where the git known hosts file is mounted'
- ' needs to be the same as the GIT_SSH_KNOWN_HOSTS_FILE path')
-
- def test_make_pod_with_empty_executor_config(self):
- self.kube_config.kube_affinity = self.affinity_config
- self.kube_config.kube_tolerations = self.tolerations_config
- self.kube_config.kube_annotations = self.worker_annotations_config
- self.kube_config.dags_folder = 'dags'
- worker_config = WorkerConfiguration(self.kube_config)
- pod = worker_config.as_pod()
-
- self.assertTrue(pod.spec.affinity['podAntiAffinity'] is not None)
- self.assertEqual('app',
- pod.spec.affinity['podAntiAffinity']
- ['requiredDuringSchedulingIgnoredDuringExecution'][0]
- ['labelSelector']
- ['matchExpressions'][0]
- ['key'])
-
- self.assertEqual(2, len(pod.spec.tolerations))
- self.assertEqual('prod', pod.spec.tolerations[1]['key'])
- self.assertEqual('role-arn', pod.metadata.annotations['iam.amazonaws.com/role'])
- self.assertEqual('value', pod.metadata.annotations['other/annotation'])
-
- def test_make_pod_with_executor_config(self):
- self.kube_config.dags_folder = 'dags'
- worker_config = WorkerConfiguration(self.kube_config)
- config_pod = PodGenerator(
- image='',
- affinity=self.affinity_config,
- tolerations=self.tolerations_config,
- ).gen_pod()
-
- pod = worker_config.as_pod()
-
- result = PodGenerator.reconcile_pods(pod, config_pod)
-
- self.assertTrue(result.spec.affinity['podAntiAffinity'] is not None)
- self.assertEqual('app',
- result.spec.affinity['podAntiAffinity']
- ['requiredDuringSchedulingIgnoredDuringExecution'][0]
- ['labelSelector']
- ['matchExpressions'][0]
- ['key'])
-
- self.assertEqual(2, len(result.spec.tolerations))
- self.assertEqual('prod', result.spec.tolerations[1]['key'])
-
- def test_worker_pvc_dags(self):
- # Tests persistence volume config created when `dags_volume_claim` is set
- self.kube_config.dags_volume_claim = 'airflow-dags'
- self.kube_config.dags_folder = 'dags'
- worker_config = WorkerConfiguration(self.kube_config)
- volumes = worker_config._get_volumes()
- volume_mounts = worker_config._get_volume_mounts()
-
- init_containers = worker_config._get_init_containers()
-
- dag_volume = [volume for volume in volumes if volume.name == 'airflow-dags']
- dag_volume_mount = [mount for mount in volume_mounts if mount.name == 'airflow-dags']
-
- self.assertEqual('airflow-dags', dag_volume[0].persistent_volume_claim.claim_name)
- self.assertEqual(1, len(dag_volume_mount))
- self.assertTrue(dag_volume_mount[0].read_only)
- self.assertEqual(0, len(init_containers))
-
- def test_worker_git_dags(self):
- # Tests persistence volume config created when `git_repo` is set
- self.kube_config.dags_volume_claim = None
- self.kube_config.dags_volume_host = None
- self.kube_config.dags_folder = '/usr/local/airflow/dags'
- self.kube_config.worker_dags_folder = '/usr/local/airflow/dags'
-
- self.kube_config.git_sync_container_repository = 'gcr.io/google-containers/git-sync-amd64'
- self.kube_config.git_sync_container_tag = 'v2.0.5'
- self.kube_config.git_sync_container = 'gcr.io/google-containers/git-sync-amd64:v2.0.5'
- self.kube_config.git_sync_init_container_name = 'git-sync-clone'
- self.kube_config.git_subpath = 'dags_folder'
- self.kube_config.git_sync_root = '/git'
- self.kube_config.git_sync_run_as_user = 65533
- self.kube_config.git_dags_folder_mount_point = '/usr/local/airflow/dags/repo/dags_folder'
-
- worker_config = WorkerConfiguration(self.kube_config)
- volumes = worker_config._get_volumes()
- volume_mounts = worker_config._get_volume_mounts()
-
- dag_volume = [volume for volume in volumes if volume.name == 'airflow-dags']
- dag_volume_mount = [mount for mount in volume_mounts if mount.name == 'airflow-dags']
-
- self.assertIsNotNone(dag_volume[0].empty_dir)
- self.assertEqual(self.kube_config.git_dags_folder_mount_point, dag_volume_mount[0].mount_path)
- self.assertTrue(dag_volume_mount[0].read_only)
-
- init_container = worker_config._get_init_containers()[0]
- init_container_volume_mount = [mount for mount in init_container.volume_mounts
- if mount.name == 'airflow-dags']
-
- self.assertEqual('git-sync-clone', init_container.name)
- self.assertEqual('gcr.io/google-containers/git-sync-amd64:v2.0.5', init_container.image)
- self.assertEqual(1, len(init_container_volume_mount))
- self.assertFalse(init_container_volume_mount[0].read_only)
- self.assertEqual(65533, init_container.security_context.run_as_user)
-
- def test_worker_container_dags(self):
- # Tests that the 'airflow-dags' persistence volume is NOT created when `dags_in_image` is set
- self.kube_config.dags_in_image = True
- self.kube_config.dags_folder = 'dags'
- worker_config = WorkerConfiguration(self.kube_config)
- volumes = worker_config._get_volumes()
- volume_mounts = worker_config._get_volume_mounts()
-
- dag_volume = [volume for volume in volumes if volume.name == 'airflow-dags']
- dag_volume_mount = [mount for mount in volume_mounts if mount.name == 'airflow-dags']
-
- init_containers = worker_config._get_init_containers()
-
- self.assertEqual(0, len(dag_volume))
- self.assertEqual(0, len(dag_volume_mount))
- self.assertEqual(0, len(init_containers))
-
- def test_set_airflow_config_configmap(self):
- """
- Test that airflow.cfg can be set via configmap by
- checking volume & volume-mounts are set correctly.
- """
- self.kube_config.airflow_home = '/usr/local/airflow'
- self.kube_config.airflow_configmap = 'airflow-configmap'
- self.kube_config.airflow_local_settings_configmap = None
- self.kube_config.dags_folder = '/workers/path/to/dags'
-
- worker_config = WorkerConfiguration(self.kube_config)
- pod = worker_config.as_pod()
-
- pod_spec_dict = pod.spec.to_dict()
-
- airflow_config_volume = [
- volume for volume in pod_spec_dict['volumes'] if volume["name"] == 'airflow-config'
- ]
- # Test that volume_name is found
- self.assertEqual(1, len(airflow_config_volume))
-
- # Test that config map exists
- self.assertEqual(
- {'default_mode': None, 'items': None, 'name': 'airflow-configmap', 'optional': None},
- airflow_config_volume[0]['config_map']
- )
-
- # Test that only 1 Volume Mounts exists with 'airflow-config' name
- # One for airflow.cfg
- volume_mounts = [
- volume_mount for volume_mount in pod_spec_dict['containers'][0]['volume_mounts']
- if volume_mount['name'] == 'airflow-config'
- ]
-
- self.assertEqual(
- [
- {
- 'mount_path': '/usr/local/airflow/airflow.cfg',
- 'mount_propagation': None,
- 'name': 'airflow-config',
- 'read_only': True,
- 'sub_path': 'airflow.cfg',
- 'sub_path_expr': None
- }
- ],
- volume_mounts
- )
-
- def test_set_airflow_local_settings_configmap(self):
- """
- Test that airflow_local_settings.py can be set via configmap by
- checking volume & volume-mounts are set correctly.
- """
- self.kube_config.airflow_home = '/usr/local/airflow'
- self.kube_config.airflow_configmap = 'airflow-configmap'
- self.kube_config.airflow_local_settings_configmap = 'airflow-configmap'
- self.kube_config.dags_folder = '/workers/path/to/dags'
-
- worker_config = WorkerConfiguration(self.kube_config)
- pod = worker_config.as_pod()
-
- pod_spec_dict = pod.spec.to_dict()
-
- airflow_config_volume = [
- volume for volume in pod_spec_dict['volumes'] if volume["name"] == 'airflow-config'
- ]
- # Test that volume_name is found
- self.assertEqual(1, len(airflow_config_volume))
-
- # Test that config map exists
- self.assertEqual(
- {'default_mode': None, 'items': None, 'name': 'airflow-configmap', 'optional': None},
- airflow_config_volume[0]['config_map']
- )
-
- # Test that 2 Volume Mounts exists and has 2 different mount-paths
- # One for airflow.cfg
- # Second for airflow_local_settings.py
- volume_mounts = [
- volume_mount for volume_mount in pod_spec_dict['containers'][0]['volume_mounts']
- if volume_mount['name'] == 'airflow-config'
- ]
- self.assertEqual(2, len(volume_mounts))
-
- self.assertEqual(
- [
- {
- 'mount_path': '/usr/local/airflow/airflow.cfg',
- 'mount_propagation': None,
- 'name': 'airflow-config',
- 'read_only': True,
- 'sub_path': 'airflow.cfg',
- 'sub_path_expr': None
- },
- {
- 'mount_path': '/usr/local/airflow/config/airflow_local_settings.py',
- 'mount_propagation': None,
- 'name': 'airflow-config',
- 'read_only': True,
- 'sub_path': 'airflow_local_settings.py',
- 'sub_path_expr': None
- }
- ],
- volume_mounts
- )
-
- def test_set_airflow_configmap_different_for_local_setting(self):
- """
- Test that airflow_local_settings.py can be set via configmap by
- checking volume & volume-mounts are set correctly when using a different
- configmap than airflow_configmap (airflow.cfg)
- """
- self.kube_config.airflow_home = '/usr/local/airflow'
- self.kube_config.airflow_configmap = 'airflow-configmap'
- self.kube_config.airflow_local_settings_configmap = 'airflow-ls-configmap'
- self.kube_config.dags_folder = '/workers/path/to/dags'
-
- worker_config = WorkerConfiguration(self.kube_config)
- pod = worker_config.as_pod()
-
- pod_spec_dict = pod.spec.to_dict()
-
- airflow_local_settings_volume = [
- volume for volume in pod_spec_dict['volumes'] if volume["name"] == 'airflow-local-settings'
- ]
- # Test that volume_name is found
- self.assertEqual(1, len(airflow_local_settings_volume))
-
- # Test that config map exists
- self.assertEqual(
- {'default_mode': None, 'items': None, 'name': 'airflow-ls-configmap', 'optional': None},
- airflow_local_settings_volume[0]['config_map']
- )
-
- # Test that 2 Volume Mounts exists and has 2 different mount-paths
- # One for airflow.cfg
- # Second for airflow_local_settings.py
- airflow_cfg_volume_mount = [
- volume_mount for volume_mount in pod_spec_dict['containers'][0]['volume_mounts']
- if volume_mount['name'] == 'airflow-config'
- ]
-
- local_setting_volume_mount = [
- volume_mount for volume_mount in pod_spec_dict['containers'][0]['volume_mounts']
- if volume_mount['name'] == 'airflow-local-settings'
- ]
- self.assertEqual(1, len(airflow_cfg_volume_mount))
- self.assertEqual(1, len(local_setting_volume_mount))
-
- self.assertEqual(
- [
- {
- 'mount_path': '/usr/local/airflow/config/airflow_local_settings.py',
- 'mount_propagation': None,
- 'name': 'airflow-local-settings',
- 'read_only': True,
- 'sub_path': 'airflow_local_settings.py',
- 'sub_path_expr': None
- }
- ],
- local_setting_volume_mount
- )
-
- self.assertEqual(
- [
- {
- 'mount_path': '/usr/local/airflow/airflow.cfg',
- 'mount_propagation': None,
- 'name': 'airflow-config',
- 'read_only': True,
- 'sub_path': 'airflow.cfg',
- 'sub_path_expr': None
- }
- ],
- airflow_cfg_volume_mount
- )
-
- def test_kubernetes_environment_variables(self):
- # Tests the kubernetes environment variables get copied into the worker pods
- input_environment = {
- 'ENVIRONMENT': 'prod',
- 'LOG_LEVEL': 'warning'
- }
- self.kube_config.kube_env_vars = input_environment
- worker_config = WorkerConfiguration(self.kube_config)
- env = worker_config._get_environment()
- for key in input_environment:
- self.assertIn(key, env)
- self.assertIn(input_environment[key], env.values())
-
- core_executor = 'AIRFLOW__CORE__EXECUTOR'
- input_environment = {
- core_executor: 'NotLocalExecutor'
- }
- self.kube_config.kube_env_vars = input_environment
- worker_config = WorkerConfiguration(self.kube_config)
- env = worker_config._get_environment()
- self.assertEqual(env[core_executor], 'LocalExecutor')
-
- def test_get_secrets(self):
- # Test when secretRef is None and kube_secrets is not empty
- self.kube_config.kube_secrets = {
- 'AWS_SECRET_KEY': 'airflow-secret=aws_secret_key',
- 'POSTGRES_PASSWORD': 'airflow-secret=postgres_credentials'
- }
- self.kube_config.env_from_secret_ref = None
- worker_config = WorkerConfiguration(self.kube_config)
- secrets = worker_config._get_secrets()
- secrets.sort(key=lambda secret: secret.deploy_target)
- expected = [
- Secret('env', 'AWS_SECRET_KEY', 'airflow-secret', 'aws_secret_key'),
- Secret('env', 'POSTGRES_PASSWORD', 'airflow-secret', 'postgres_credentials')
- ]
- self.assertListEqual(expected, secrets)
-
- # Test when secret is not empty and kube_secrets is empty dict
- self.kube_config.kube_secrets = {}
- self.kube_config.env_from_secret_ref = 'secret_a,secret_b'
- worker_config = WorkerConfiguration(self.kube_config)
- secrets = worker_config._get_secrets()
- expected = [Secret('env', None, 'secret_a'), Secret('env', None, 'secret_b')]
- self.assertListEqual(expected, secrets)
-
- def test_get_env_from(self):
- # Test when configmap is empty
- self.kube_config.env_from_configmap_ref = ''
- worker_config = WorkerConfiguration(self.kube_config)
- configmaps = worker_config._get_env_from()
- self.assertListEqual([], configmaps)
-
- # test when configmap is not empty
- self.kube_config.env_from_configmap_ref = 'configmap_a,configmap_b'
- self.kube_config.env_from_secret_ref = 'secretref_a,secretref_b'
- worker_config = WorkerConfiguration(self.kube_config)
- configmaps = worker_config._get_env_from()
- self.assertListEqual([
- k8s.V1EnvFromSource(config_map_ref=k8s.V1ConfigMapEnvSource(name='configmap_a')),
- k8s.V1EnvFromSource(config_map_ref=k8s.V1ConfigMapEnvSource(name='configmap_b')),
- k8s.V1EnvFromSource(secret_ref=k8s.V1SecretEnvSource(name='secretref_a')),
- k8s.V1EnvFromSource(secret_ref=k8s.V1SecretEnvSource(name='secretref_b'))
- ], configmaps)
-
- def test_pod_template_file(self):
- fixture = sys.path[0] + '/tests/kubernetes/pod.yaml'
- self.kube_config.pod_template_file = fixture
- worker_config = WorkerConfiguration(self.kube_config)
- result = worker_config.as_pod()
- expected = PodGenerator.deserialize_model_file(fixture)
- expected.metadata.name = ANY
- self.assertEqual(expected, result)
-
- def test_get_labels(self):
- worker_config = WorkerConfiguration(self.kube_config)
- labels = worker_config._get_labels({'my_kube_executor_label': 'kubernetes'}, {
- 'dag_id': 'override_dag_id',
- })
- self.assertEqual({
- 'my_label': 'label_id',
- 'dag_id': 'override_dag_id',
- 'my_kube_executor_label': 'kubernetes',
- }, labels)
-
- def test_make_pod_with_image_pull_secrets(self):
- # Tests the pod created with image_pull_secrets actually gets that in it's config
- self.kube_config.dags_volume_claim = None
- self.kube_config.dags_volume_host = None
- self.kube_config.dags_in_image = None
- self.kube_config.git_dags_folder_mount_point = 'dags'
- self.kube_config.git_sync_dest = 'repo'
- self.kube_config.git_subpath = 'path'
- self.kube_config.image_pull_secrets = 'image_pull_secret1,image_pull_secret2'
-
- worker_config = WorkerConfiguration(self.kube_config)
- pod = worker_config.as_pod()
-
- self.assertEqual(2, len(pod.spec.image_pull_secrets))
-
- def test_get_resources(self):
- self.kube_config.worker_resources = {'limit_cpu': 0.25, 'limit_memory': '64Mi', 'request_cpu': '250m',
- 'request_memory': '64Mi'}
-
- worker_config = WorkerConfiguration(self.kube_config)
- resources = worker_config._get_resources()
- self.assertEqual(resources.limits["cpu"], 0.25)
- self.assertEqual(resources.limits["memory"], "64Mi")
- self.assertEqual(resources.requests["cpu"], "250m")
- self.assertEqual(resources.requests["memory"], "64Mi")
diff --git a/tests/providers/cncf/kubernetes/operators/test_kubernetes_pod.py b/tests/providers/cncf/kubernetes/operators/test_kubernetes_pod.py
index 1d4dc01..b2c92c0 100644
--- a/tests/providers/cncf/kubernetes/operators/test_kubernetes_pod.py
+++ b/tests/providers/cncf/kubernetes/operators/test_kubernetes_pod.py
@@ -86,7 +86,7 @@ class TestKubernetesPodOperator(unittest.TestCase):
task_id="task",
in_cluster=False,
do_xcom_push=False,
- image_pull_secrets=fake_pull_secrets,
+ image_pull_secrets=[k8s.V1LocalObjectReference(fake_pull_secrets)],
cluster_context='default',
)
monitor_mock.return_value = (State.SUCCESS, None)
diff --git a/tests/test_config_templates.py b/tests/test_config_templates.py
index 374e8c2..af6cc4c 100644
--- a/tests/test_config_templates.py
+++ b/tests/test_config_templates.py
@@ -51,11 +51,7 @@ DEFAULT_AIRFLOW_SECTIONS = [
'elasticsearch',
'elasticsearch_configs',
'kubernetes',
- 'kubernetes_node_selectors',
- 'kubernetes_environment_variables',
- 'kubernetes_secrets',
- 'kubernetes_labels',
- 'smart_sensor'
+ 'smart_sensor',
]
DEFAULT_TEST_SECTIONS = [