You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by fo...@apache.org on 2018/04/22 08:32:45 UTC
[12/20] incubator-airflow git commit: [AIRFLOW-1314] Improve k8s
support
[AIRFLOW-1314] Improve k8s support
Add kubernetes config section in airflow.cfg and Inject GCP secrets upon executor start. (#17)
Update Airflow to Pass configuration to k8s containers, add some Py3 … (#9)
* Update Airflow to Pass configuration to k8s containers, add some Py3 compat., create git-sync pod
* Undo changes to display-source config setter for to_dict
* WIP Secrets and Configmaps
* Improve secrets support for multiple secrets. Add support for registry secrets. Add support for RBAC service accounts.
* Swap order of variables, overlooked very basic issue
* Secret env var names must be upper
* Update logging
* Revert spothero test code in setup.py
* WIP Fix tests
* Worker should be using local executor
* Consolidate worker setup and address code review comments
* reconfigure airflow script to use new secrets method
Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/ad4e67ce
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/ad4e67ce
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/ad4e67ce
Branch: refs/heads/master
Commit: ad4e67ce1be114f248ecd5ea1468ccea79366590
Parents: a9d90dc
Author: fenglu-g <fe...@google.com>
Authored: Mon Oct 23 09:52:06 2017 -0700
Committer: Fokko Driesprong <fo...@godatadriven.com>
Committed: Sun Apr 22 10:23:06 2018 +0200
----------------------------------------------------------------------
airflow/config_templates/default_airflow.cfg | 63 ++++-
.../contrib/executors/kubernetes_executor.py | 239 ++++++++++---------
.../contrib/kubernetes/kubernetes_factory.py | 2 -
.../kubernetes_request_factory.py | 25 +-
.../pod_request_factory.py | 28 ++-
airflow/contrib/kubernetes/pod.py | 13 +-
airflow/contrib/kubernetes/pod_launcher.py | 28 ++-
airflow/contrib/kubernetes/secret.py | 33 ++-
.../contrib/kubernetes/worker_configuration.py | 158 ++++++++++++
airflow/jobs.py | 5 +-
airflow/utils/dag_processing.py | 2 +-
scripts/ci/kubernetes/docker/Dockerfile | 6 -
.../ci/kubernetes/kube/airflow.yaml.template | 134 +++++++----
scripts/ci/kubernetes/kube/deploy.sh | 7 +-
14 files changed, 513 insertions(+), 230 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/ad4e67ce/airflow/config_templates/default_airflow.cfg
----------------------------------------------------------------------
diff --git a/airflow/config_templates/default_airflow.cfg b/airflow/config_templates/default_airflow.cfg
index 400bcc0..6a966cb 100644
--- a/airflow/config_templates/default_airflow.cfg
+++ b/airflow/config_templates/default_airflow.cfg
@@ -5,9 +5,9 @@
# 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
@@ -498,3 +498,62 @@ hide_sensitive_variable_fields = True
[elasticsearch]
elasticsearch_host =
+[kubernetes]
+# The repository and tag of the Kubernetes Image for the Worker to Run
+worker_container_repository =
+worker_container_tag =
+
+# If True (default), worker pods will be deleted upon termination
+delete_worker_pods = True
+
+# The Kubernetes namespace where airflow workers should be created. Defaults to `default`
+namespace = default
+
+# The name of the Kubernetes ConfigMap Containing the Airflow Configuration (this file)
+airflow_configmap =
+
+# For either git sync or volume mounted DAGs, the worker will look in this subpath for DAGs
+dags_volume_subpath =
+
+# For DAGs mounted via a volume claim (mutually exclusive with volume claim)
+dags_volume_claim =
+
+# Git credentials and repository for DAGs mounted via Git (mutually exclusive with volume claim)
+git_repo =
+git_branch =
+git_user =
+git_password =
+
+# For cloning DAGs from git repositories into volumes: https://github.com/kubernetes/git-sync
+git_sync_container_repository = gcr.io/google-containers/git-sync-amd64
+git_sync_container_tag = v2.0.5
+git_sync_init_container_name = git-sync-clone
+
+# 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 =
+
+# GCP Service Account Keys to be provided to tasks run on Kubernetes Executors
+# Should be supplied in the format: key-name-1:key-path-1,key-name-2:key-path-2
+gcp_service_account_keys =
+
+[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.
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/ad4e67ce/airflow/contrib/executors/kubernetes_executor.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/executors/kubernetes_executor.py b/airflow/contrib/executors/kubernetes_executor.py
index 8989add..9675e81 100644
--- a/airflow/contrib/executors/kubernetes_executor.py
+++ b/airflow/contrib/executors/kubernetes_executor.py
@@ -12,10 +12,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import calendar
-import logging
+import base64
import os
import multiprocessing
+import six
from queue import Queue
from dateutil import parser
from uuid import uuid4
@@ -23,13 +23,13 @@ from kubernetes import watch, client
from kubernetes.client.rest import ApiException
from airflow.contrib.kubernetes.pod_launcher import PodLauncher
from airflow.contrib.kubernetes.kube_client import get_kube_client
+from airflow.contrib.kubernetes.worker_configuration import WorkerConfiguration
from airflow.executors.base_executor import BaseExecutor
from airflow.models import TaskInstance, KubeResourceVersion
from airflow.utils.state import State
from airflow import configuration, settings
from airflow.exceptions import AirflowConfigException
-from airflow.contrib.kubernetes.pod import Pod
-
+from airflow.utils.log.logging_mixin import LoggingMixin
class KubeConfig:
core_section = "core"
@@ -50,98 +50,84 @@ class KubeConfig:
return default
def __init__(self):
+ configuration_dict = configuration.as_dict(display_sensitive=True)
+ self.core_configuration = configuration_dict['core']
+ self.kube_secrets = configuration_dict.get('kubernetes_secrets', {})
+ self.airflow_home = configuration.get(self.core_section, 'airflow_home')
self.dags_folder = configuration.get(self.core_section, 'dags_folder')
self.parallelism = configuration.getint(self.core_section, 'PARALLELISM')
- self.kube_image = configuration.get(self.kubernetes_section, 'container_image')
+ self.worker_container_repository = configuration.get(
+ self.kubernetes_section, 'worker_container_repository')
+ self.worker_container_tag = configuration.get(
+ self.kubernetes_section, 'worker_container_tag')
+ self.kube_image = '{}:{}'.format(
+ self.worker_container_repository, self.worker_container_tag)
self.delete_worker_pods = self.safe_getboolean(self.kubernetes_section, 'delete_worker_pods', True)
- self.kube_namespace = os.environ.get('AIRFLOW_KUBE_NAMESPACE', 'default')
- # These two props must be set together
+ self.worker_service_account_name = self.safe_get(
+ self.kubernetes_section, 'worker_service_account_name', 'default')
+ self.image_pull_secrets = self.safe_get(
+ self.kubernetes_section, 'image_pull_secrets', '')
+
+ # 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 = self.safe_get(self.kubernetes_section, 'git_repo', None)
+ # The branch of the repository to be checked out
self.git_branch = self.safe_get(self.kubernetes_section, 'git_branch', None)
+ # Optionally, the directory in the git repository containing the dags
+ self.git_subpath = self.safe_get(self.kubernetes_section, 'git_subpath', '')
+
+ # Optionally a user may supply a `git_user` and `git_password` for private repositories
+ self.git_user = self.safe_get(self.kubernetes_section, 'git_user', None)
+ self.git_password = self.safe_get(self.kubernetes_section, 'git_password', None)
- # Or this one prop
+ # NOTE: The user may optionally use a volume claim to mount a PV containing DAGs directly
self.dags_volume_claim = self.safe_get(self.kubernetes_section, 'dags_volume_claim', None)
- # And optionally this prop
+
+ # This prop may optionally be set for PV Claims and is used to locate DAGs on a SubPath
self.dags_volume_subpath = self.safe_get(self.kubernetes_section, 'dags_volume_subpath', None)
+ # The Kubernetes Namespace in which the Scheduler and Webserver reside. Note that if your
+ # cluster has RBAC enabled, your scheduler may need service account permissions to
+ # create, watch, get, and delete pods in this namespace.
+ self.kube_namespace = self.safe_get(self.kubernetes_section, 'namespace', 'default')
+ # The Kubernetes Namespace in which pods will be created by the executor. Note that if your
+ # cluster has RBAC enabled, your workers may need service account permissions to
+ # interact with cluster components.
+ self.executor_namespace = self.safe_get(self.kubernetes_section, 'namespace', 'default')
+ # Task secrets managed by KubernetesExecutor.
+ self.gcp_service_account_keys = self.safe_get(self.kubernetes_section, 'gcp_service_account_keys', None)
+
+ # 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 = self.safe_get(
+ self.kubernetes_section, 'git_sync_container_repository',
+ 'gcr.io/google-containers/git-sync-amd64')
+
+ self.git_sync_container_tag = self.safe_get(
+ self.kubernetes_section, 'git_sync_container_tag', 'v2.0.5')
+ self.git_sync_container = '{}:{}'.format(
+ self.git_sync_container_repository, self.git_sync_container_tag)
+
+ self.git_sync_init_container_name = self.safe_get(
+ self.kubernetes_section, 'git_sync_init_container_name', 'git-sync-clone')
+
+ # The worker pod may optionally have a valid Airflow config loaded via a configmap
+ self.airflow_configmap = self.safe_get(self.kubernetes_section, 'airflow_configmap', None)
+
self._validate()
def _validate(self):
- if self.dags_volume_claim:
- # do volume things
- pass
- elif self.git_repo and self.git_branch:
- # do git things
- pass
- else:
+ if not self.dags_volume_claim and (not self.git_repo or not self.git_branch):
raise AirflowConfigException(
"In kubernetes mode you must set the following configs in the `kubernetes` section: "
"`dags_volume_claim` or "
- "`git_repo and git_branch`"
+ "`git_repo and git_branch` "
)
-class PodMaker:
- def __init__(self, kube_config):
- self.logger = logging.getLogger(__name__)
- self.kube_config = kube_config
-
- def _get_volumes_and_mounts(self):
- volume_name = "airflow-dags"
-
- if self.kube_config.dags_volume_claim:
- volumes = [{
- "name": volume_name, "persistentVolumeClaim": {"claimName": self.kube_config.dags_volume_claim}
- }]
- volume_mounts = [{
- "name": volume_name, "mountPath": self.kube_config.dags_folder,
- "readOnly": True
- }]
- if self.kube_config.dags_volume_subpath:
- volume_mounts[0]["subPath"] = self.kube_config.dags_volume_subpath
-
- return volumes, volume_mounts
- else:
- return [], []
-
- def _get_args(self, airflow_command):
- if self.kube_config.dags_volume_claim:
- self.logger.info("Using k8s_dags_volume_claim for airflow dags")
- return [airflow_command]
- else:
- self.logger.info("Using git-syncher for airflow dags")
- cmd_args = "mkdir -p {dags_folder} && cd {dags_folder} &&" \
- "git init && git remote add origin {git_repo} && git pull origin {git_branch} --depth=1 &&" \
- "{command}".format(dags_folder=self.kube_config.dags_folder, git_repo=self.kube_config.git_repo,
- git_branch=self.kube_config.git_branch, command=airflow_command)
- return [cmd_args]
-
- def make_pod(self, namespace, pod_id, dag_id, task_id, execution_date, airflow_command):
- volumes, volume_mounts = self._get_volumes_and_mounts()
-
- pod = Pod(
- namespace=namespace,
- name=pod_id,
- image=self.kube_config.kube_image,
- cmds=["bash", "-cx", "--"],
- args=self._get_args(airflow_command),
- labels={
- "airflow-slave": "",
- "dag_id": dag_id,
- "task_id": task_id,
- "execution_date": execution_date
- },
- envs={"AIRFLOW__CORE__EXECUTOR": "LocalExecutor"},
- volumes=volumes,
- volume_mounts=volume_mounts
- )
- return pod
-
-
-class KubernetesJobWatcher(multiprocessing.Process, object):
+class KubernetesJobWatcher(multiprocessing.Process, LoggingMixin, object):
def __init__(self, namespace, watcher_queue, resource_version):
- self.logger = logging.getLogger(__name__)
multiprocessing.Process.__init__(self)
self.namespace = namespace
self.watcher_queue = watcher_queue
@@ -153,14 +139,14 @@ class KubernetesJobWatcher(multiprocessing.Process, object):
try:
self.resource_version = self._run(kube_client, self.resource_version)
except Exception:
- self.logger.exception("Unknown error in KubernetesJobWatcher. Failing")
+ self.log.exception("Unknown error in KubernetesJobWatcher. Failing")
raise
else:
- self.logger.warn("Watch died gracefully, starting back up with: "
- "last resource_version: {}".format(self.resource_version))
+ self.log.warn("Watch died gracefully, starting back up with: "
+ "last resource_version: {}".format(self.resource_version))
def _run(self, kube_client, resource_version):
- self.logger.info("Event: and now my watch begins starting at resource_version: {}".format(resource_version))
+ self.log.info("Event: and now my watch begins starting at resource_version: {}".format(resource_version))
watcher = watch.Watch()
kwargs = {"label_selector": "airflow-slave"}
@@ -170,7 +156,7 @@ class KubernetesJobWatcher(multiprocessing.Process, object):
last_resource_version = None
for event in watcher.stream(kube_client.list_namespaced_pod, self.namespace, **kwargs):
task = event['object']
- self.logger.info("Event: {} had an event of type {}".format(task.metadata.name, event['type']))
+ self.log.info("Event: {} had an event of type {}".format(task.metadata.name, event['type']))
self.process_status(
task.metadata.name, task.status.phase, task.metadata.labels, task.metadata.resource_version
)
@@ -180,32 +166,31 @@ class KubernetesJobWatcher(multiprocessing.Process, object):
def process_status(self, pod_id, status, labels, resource_version):
if status == 'Pending':
- self.logger.info("Event: {} Pending".format(pod_id))
+ self.log.info("Event: {} Pending".format(pod_id))
elif status == 'Failed':
- self.logger.info("Event: {} Failed".format(pod_id))
+ self.log.info("Event: {} Failed".format(pod_id))
self.watcher_queue.put((pod_id, State.FAILED, labels, resource_version))
elif status == 'Succeeded':
- self.logger.info("Event: {} Succeeded".format(pod_id))
+ self.log.info("Event: {} Succeeded".format(pod_id))
self.watcher_queue.put((pod_id, None, labels, resource_version))
elif status == 'Running':
- self.logger.info("Event: {} is Running".format(pod_id))
+ self.log.info("Event: {} is Running".format(pod_id))
else:
- self.logger.warn("Event: Invalid state: {} on pod: {} with labels: {} "
+ self.log.warn("Event: Invalid state: {} on pod: {} with labels: {} "
"with resource_version: {}".format(status, pod_id, labels, resource_version))
-class AirflowKubernetesScheduler(object):
+class AirflowKubernetesScheduler(LoggingMixin, object):
def __init__(self, kube_config, task_queue, result_queue, session, kube_client):
- self.logger = logging.getLogger(__name__)
- self.logger.info("creating kubernetes executor")
- self.kube_config = KubeConfig()
+ self.log.debug("creating kubernetes executor")
+ self.kube_config = kube_config
self.task_queue = task_queue
- self.pending_jobs = set()
- self.namespace = os.environ['k8s_POD_NAMESPACE']
- self.logger.info("k8s: using namespace {}".format(self.namespace))
+ self.result_queue = result_queue
+ self.namespace = self.kube_config.kube_namespace
+ self.log.debug("k8s: using namespace {}".format(self.namespace))
self.kube_client = kube_client
self.launcher = PodLauncher(kube_client=self.kube_client)
- self.pod_maker = PodMaker(kube_config=self.kube_config)
+ self.worker_configuration = WorkerConfiguration(kube_config=self.kube_config)
self.watcher_queue = multiprocessing.Queue()
self._session = session
self.kube_watcher = self._make_kube_watcher()
@@ -220,7 +205,7 @@ class AirflowKubernetesScheduler(object):
if self.kube_watcher.is_alive():
pass
else:
- self.logger.error("Error while health checking kube watcher process. Process died for unknown reasons")
+ self.log.error("Error while health checking kube watcher process. Process died for unknown reasons")
self.kube_watcher = self._make_kube_watcher()
def run_next(self, next_job):
@@ -234,19 +219,19 @@ class AirflowKubernetesScheduler(object):
:return:
"""
- self.logger.info('k8s: job is {}'.format(str(next_job)))
+ self.log.debug('k8s: job is {}'.format(str(next_job)))
key, command = next_job
dag_id, task_id, execution_date = key
- self.logger.info("k8s: running for command {}".format(command))
- self.logger.info("k8s: launching image {}".format(self.kube_config.kube_image))
- pod = self.pod_maker.make_pod(
+ self.log.debug("k8s: running for command {}".format(command))
+ self.log.debug("k8s: launching image {}".format(self.kube_config.kube_image))
+ pod = self.worker_configuration.make_pod(
namespace=self.namespace, pod_id=self._create_pod_id(dag_id, task_id),
dag_id=dag_id, task_id=task_id, execution_date=self._datetime_to_label_safe_datestring(execution_date),
airflow_command=command
)
# the watcher will monitor pods, so we do not block.
self.launcher.run_pod_async(pod)
- self.logger.info("k8s: Job created!")
+ self.log.debug("k8s: Job created!")
def delete_pod(self, pod_id):
if self.kube_config.delete_worker_pods:
@@ -272,10 +257,10 @@ class AirflowKubernetesScheduler(object):
def process_watcher_task(self):
pod_id, state, labels, resource_version = self.watcher_queue.get()
- logging.info("Attempting to finish pod; pod_id: {}; state: {}; labels: {}".format(pod_id, state, labels))
+ self.log.info("Attempting to finish pod; pod_id: {}; state: {}; labels: {}".format(pod_id, state, labels))
key = self._labels_to_key(labels)
if key:
- self.logger.info("finishing job {}".format(key))
+ self.log.debug("finishing job {} - {} ({})".format(key, state, pod_id))
self.result_queue.put((key, state, pod_id, resource_version))
@staticmethod
@@ -336,13 +321,13 @@ class AirflowKubernetesScheduler(object):
try:
return labels["dag_id"], labels["task_id"], self._label_safe_datestring_to_datetime(labels["execution_date"])
except Exception as e:
- self.logger.warn("Error while converting labels to key; labels: {}; exception: {}".format(
+ self.log.warn("Error while converting labels to key; labels: {}; exception: {}".format(
labels, e
))
return None
-class KubernetesExecutor(BaseExecutor):
+class KubernetesExecutor(BaseExecutor, LoggingMixin):
def __init__(self):
super(KubernetesExecutor, self).__init__(parallelism=PARALLELISM)
self.task_queue = None
@@ -362,7 +347,7 @@ class KubernetesExecutor(BaseExecutor):
:return: None
"""
queued_tasks = self._session.query(TaskInstance).filter(TaskInstance.state == State.QUEUED).all()
- self.logger.info("When executor started up, found {} queued task instances".format(len(queued_tasks)))
+ self.log.info("When executor started up, found {} queued task instances".format(len(queued_tasks)))
for t in queued_tasks:
kwargs = dict(label_selector="dag_id={},task_id={},execution_date={}".format(
@@ -370,7 +355,7 @@ class KubernetesExecutor(BaseExecutor):
))
pod_list = self.kube_client.list_namespaced_pod(self.kube_config.kube_namespace, **kwargs)
if len(pod_list.items) == 0:
- self.logger.info("TaskInstance: {} found in queued state but was not launched, rescheduling".format(t))
+ self.log.info("TaskInstance: {} found in queued state but was not launched, rescheduling".format(t))
self._session.query(TaskInstance).filter(
TaskInstance.dag_id == t.dag_id,
TaskInstance.task_id == t.task_id,
@@ -379,8 +364,37 @@ class KubernetesExecutor(BaseExecutor):
self._session.commit()
+ def _inject_secrets(self):
+ def _create_or_update_secret(secret_name, secret_path):
+ try:
+ return self.kube_client.create_namespaced_secret(
+ self.kube_config.executor_namespace, kubernetes.client.V1Secret(
+ data={'key.json' : base64.b64encode(open(secret_path, 'r').read())},
+ metadata=kubernetes.client.V1ObjectMeta(name=secret_name)))
+ except ApiException as e:
+ if e.status == 409:
+ return self.kube_client.replace_namespaced_secret(
+ secret_name, self.kube_config.executor_namespace,
+ kubernetes.client.V1Secret(
+ data={'key.json' : base64.b64encode(open(secret_path, 'r').read())},
+ metadata=kubernetes.client.V1ObjectMeta(name=secret_name)))
+ self.log.exception("Exception while trying to inject secret."
+ "Secret name: {}, error details: {}.".format(secret_name, e))
+ raise
+
+ # For each GCP service account key, inject it as a secret in executor
+ # namespace with the specific secret name configured in the airflow.cfg.
+ # We let exceptions to pass through to users.
+ if self.kube_config.gcp_service_account_keys:
+ name_path_pair_list = [
+ {'name' : account_spec.strip().split('=')[0],
+ 'path' : account_spec.strip().split('=')[1]}
+ for account_spec in self.kube_config.gcp_service_account_keys.split(',')]
+ for service_account in name_path_pair_list:
+ _create_or_update_secret(service_account['name'], service_account['path'])
+
def start(self):
- self.logger.info('k8s: starting kubernetes executor')
+ self.log.info('k8s: starting kubernetes executor')
self._session = settings.Session()
self.task_queue = Queue()
self.result_queue = Queue()
@@ -388,15 +402,16 @@ class KubernetesExecutor(BaseExecutor):
self.kube_scheduler = AirflowKubernetesScheduler(
self.kube_config, self.task_queue, self.result_queue, self._session, self.kube_client
)
+ self._inject_secrets()
self.clear_not_launched_queued_tasks()
def execute_async(self, key, command, queue=None):
- self.logger.info("k8s: adding task {} with command {}".format(key, command))
+ self.log.info("k8s: adding task {} with command {}".format(key, command))
self.task_queue.put((key, command))
def sync(self):
- self.logger.info("self.running: {}".format(self.running))
- self.logger.info("self.queued: {}".format(self.queued_tasks))
+ self.log.info("self.running: {}".format(self.running))
+ self.log.info("self.queued: {}".format(self.queued_tasks))
self.kube_scheduler.sync()
last_resource_version = None
@@ -404,7 +419,7 @@ class KubernetesExecutor(BaseExecutor):
results = self.result_queue.get()
key, state, pod_id, resource_version = results
last_resource_version = resource_version
- self.logger.info("Changing state of {}".format(results))
+ self.log.info("Changing state of {}".format(results))
self._change_state(key, state, pod_id)
KubeResourceVersion.checkpoint_resource_version(last_resource_version, session=self._session)
@@ -434,7 +449,7 @@ class KubernetesExecutor(BaseExecutor):
self._session.commit()
def end(self):
- self.logger.info('ending kube executor')
+ self.log.info('ending kube executor')
self.task_queue.join()
def execute_async(self, key, command, queue=None):
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/ad4e67ce/airflow/contrib/kubernetes/kubernetes_factory.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/kubernetes/kubernetes_factory.py b/airflow/contrib/kubernetes/kubernetes_factory.py
index 90b6f6c..715075a 100644
--- a/airflow/contrib/kubernetes/kubernetes_factory.py
+++ b/airflow/contrib/kubernetes/kubernetes_factory.py
@@ -12,7 +12,6 @@
# See the License for the specific language governing permissions and
from airflow.contrib.kubernetes.kubernetes_request_factory import KubernetesRequestFactory
-import logging
class KubernetesResourceBuilder:
@@ -31,7 +30,6 @@ class KubernetesResourceBuilder:
self.cmds = cmds
self.kub_req_factory = kub_req_factory
self.namespace = namespace
- self.logger = logging.getLogger(self.__class__.__name__)
self.envs = {}
self.labels = {}
self.secrets = {}
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/ad4e67ce/airflow/contrib/kubernetes/kubernetes_request_factory/kubernetes_request_factory.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/kubernetes/kubernetes_request_factory/kubernetes_request_factory.py b/airflow/contrib/kubernetes/kubernetes_request_factory/kubernetes_request_factory.py
index 7ea9bc2..9cfd77f 100644
--- a/airflow/contrib/kubernetes/kubernetes_request_factory/kubernetes_request_factory.py
+++ b/airflow/contrib/kubernetes/kubernetes_request_factory/kubernetes_request_factory.py
@@ -13,6 +13,7 @@
# limitations under the License.
from abc import ABCMeta, abstractmethod
+import six
class KubernetesRequestFactory():
"""
@@ -69,8 +70,9 @@ class KubernetesRequestFactoryHelper(object):
@staticmethod
def extract_labels(pod, req):
- for k in pod.labels.keys():
- req['metadata']['labels'][k] = pod.labels[k]
+ req['metadata']['labels'] = req['metadata'].get('labels', {})
+ for k, v in six.iteritems(pod.labels):
+ req['metadata']['labels'][k] = v
@staticmethod
def extract_cmds(pod, req):
@@ -114,7 +116,7 @@ class KubernetesRequestFactoryHelper(object):
})
@staticmethod
- def extract_secrets(pod, req):
+ def extract_env_and_secrets(pod, req):
env_secrets = [s for s in pod.secrets if s.deploy_type == 'env']
if len(pod.envs) > 0 or len(env_secrets) > 0:
env = []
@@ -123,3 +125,20 @@ class KubernetesRequestFactoryHelper(object):
for secret in env_secrets:
KubernetesRequestFactory.add_secret_to_env(env, secret)
req['spec']['containers'][0]['env'] = env
+
+ @staticmethod
+ def extract_init_containers(pod, req):
+ if pod.init_containers:
+ req['spec']['initContainers'] = pod.init_containers
+
+ @staticmethod
+ def extract_service_account_name(pod, req):
+ if pod.service_account_name:
+ req['spec']['serviceAccountName'] = pod.service_account_name
+
+ @staticmethod
+ def extract_image_pull_secrets(pod, req):
+ if pod.image_pull_secrets:
+ req['spec']['imagePullSecrets'] = [{
+ 'name': pull_secret
+ } for pull_secret in pod.image_pull_secrets.split(',')]
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/ad4e67ce/airflow/contrib/kubernetes/kubernetes_request_factory/pod_request_factory.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/kubernetes/kubernetes_request_factory/pod_request_factory.py b/airflow/contrib/kubernetes/kubernetes_request_factory/pod_request_factory.py
index 89631e0..dfa247f 100644
--- a/airflow/contrib/kubernetes/kubernetes_request_factory/pod_request_factory.py
+++ b/airflow/contrib/kubernetes/kubernetes_request_factory/pod_request_factory.py
@@ -12,11 +12,11 @@
# See the License for the specific language governing permissions and
import yaml
-import airflow.contrib.kubernetes.kubernetes_request_factory.kubernetes_request_factory as kreq
+from airflow.contrib.kubernetes.kubernetes_request_factory.kubernetes_request_factory import KubernetesRequestFactory
from airflow.contrib.kubernetes.pod import Pod
-class SimplePodRequestFactory(kreq.KubernetesRequestFactory):
+class SimplePodRequestFactory(KubernetesRequestFactory):
"""
Request generator for a simple pod.
"""
@@ -41,16 +41,18 @@ spec:
def create(self, pod):
# type: (Pod) -> dict
req = yaml.load(self._yaml)
- kreq.extract_name(pod, req)
- kreq.extract_labels(pod, req)
- kreq.extract_image(pod, req)
- if pod.image_pull_policy:
- kreq.extract_image_pull_policy(pod, req)
- kreq.extract_cmds(pod, req)
- kreq.extract_args(pod, req)
- if len(pod.node_selectors) > 0:
- self.extract_node_selector(pod, req)
- self.extract_secrets(pod, req)
+ self.extract_name(pod, req)
+ self.extract_labels(pod, req)
+ self.extract_image(pod, req)
+ self.extract_image_pull_policy(pod, req)
+ self.extract_cmds(pod, req)
+ self.extract_args(pod, req)
+ self.extract_node_selector(pod, req)
+ self.extract_env_and_secrets(pod, req)
self.extract_volume_secrets(pod, req)
- self.attach_volume_mounts(req=req, pod=pod)
+ self.attach_volumes(pod, req)
+ self.attach_volume_mounts(pod, req)
+ self.extract_service_account_name(pod, req)
+ self.extract_init_containers(pod, req)
+ self.extract_image_pull_secrets(pod, req)
return req
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/ad4e67ce/airflow/contrib/kubernetes/pod.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/kubernetes/pod.py b/airflow/contrib/kubernetes/pod.py
index 0200afa..be99bbf 100644
--- a/airflow/contrib/kubernetes/pod.py
+++ b/airflow/contrib/kubernetes/pod.py
@@ -12,7 +12,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import logging
class Pod:
"""
@@ -43,7 +42,11 @@ class Pod:
name=None,
volumes = [],
namespace='default',
- result=None):
+ result=None,
+ image_pull_policy="IfNotPresent",
+ image_pull_secrets=None,
+ init_containers=None,
+ service_account_name=None):
self.image = image
self.envs = envs if envs else {}
self.cmds = cmds
@@ -54,5 +57,7 @@ class Pod:
self.volumes = volumes
self.node_selectors = node_selectors if node_selectors else []
self.namespace = namespace
- self.logger = logging.getLogger(self.__class__.__name__)
-
+ self.image_pull_policy = image_pull_policy
+ self.image_pull_secrets = image_pull_secrets
+ self.init_containers = init_containers
+ self.service_account_name = service_account_name
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/ad4e67ce/airflow/contrib/kubernetes/pod_launcher.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/kubernetes/pod_launcher.py b/airflow/contrib/kubernetes/pod_launcher.py
index e435a12..1fcdb10 100644
--- a/airflow/contrib/kubernetes/pod_launcher.py
+++ b/airflow/contrib/kubernetes/pod_launcher.py
@@ -11,28 +11,34 @@
# 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 json
+
from airflow.contrib.kubernetes.pod import Pod
from airflow.contrib.kubernetes.kubernetes_request_factory.pod_request_factory import SimplePodRequestFactory
+from airflow.utils.log.logging_mixin import LoggingMixin
+from airflow.utils.state import State
from kubernetes import watch
from kubernetes.client import V1Pod
-from airflow.utils.state import State
-import json
-import logging
+from kubernetes.client.rest import ApiException
from .kube_client import get_kube_client
-class PodLauncher:
+class PodLauncher(LoggingMixin):
def __init__(self, kube_client=None):
self.kube_req_factory = SimplePodRequestFactory()
self._client = kube_client or get_kube_client()
self._watch = watch.Watch()
- self.logger = logging.getLogger(__name__)
def run_pod_async(self, pod):
req = self.kube_req_factory.create(pod)
- print(json.dumps(req))
- resp = self._client.create_namespaced_pod(body=req, namespace=pod.namespace)
+ self.log.debug('Pod Creation Request: \n{}'.format(json.dumps(req, indent=2)))
+ try:
+ resp = self._client.create_namespaced_pod(body=req, namespace=pod.namespace)
+ self.log.debug('Pod Creation Response: {}'.format(resp))
+ except ApiException:
+ self.log.exception('Exception when attempting to create Namespaced Pod.')
+ raise
return resp
def run_pod(self, pod):
@@ -54,7 +60,7 @@ class PodLauncher:
def _task_status(self, event):
# type: (V1Pod) -> State
task = event['object']
- self.logger.info(
+ self.log.info(
"Event: {} had an event of type {}".format(task.metadata.name,
event['type']))
status = self.process_status(task.metadata.name, task.status.phase)
@@ -67,13 +73,13 @@ class PodLauncher:
if status == 'Pending':
return State.QUEUED
elif status == 'Failed':
- self.logger.info("Event: {} Failed".format(job_id))
+ self.log.info("Event: {} Failed".format(job_id))
return State.FAILED
elif status == 'Succeeded':
- self.logger.info("Event: {} Succeeded".format(job_id))
+ self.log.info("Event: {} Succeeded".format(job_id))
return State.SUCCESS
elif status == 'Running':
return State.RUNNING
else:
- self.logger.info("Event: Invalid state {} on job {}".format(status, job_id))
+ self.log.info("Event: Invalid state {} on job {}".format(status, job_id))
return State.FAILED
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/ad4e67ce/airflow/contrib/kubernetes/secret.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/kubernetes/secret.py b/airflow/contrib/kubernetes/secret.py
index ec5d51c..a798a24 100644
--- a/airflow/contrib/kubernetes/secret.py
+++ b/airflow/contrib/kubernetes/secret.py
@@ -1,30 +1,25 @@
-# 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
+# -*- coding: utf-8 -*-
#
-# http://www.apache.org/licenses/LICENSE-2.0
+# Licensed 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
#
-# 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.
+# 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.
class Secret:
- """Defines Kubernetes Secret Volume"""
+ """Defines Kubernetes Secret Containers"""
def __init__(self, deploy_type, deploy_target, secret, key):
- """Initialize a Kubernetes Secret Object. Used to track requested secrets from
- the user.
+ """Initialize a Kubernetes Secret Object. Used to track requested secrets from the user.
- :param deploy_type: The type of secret deploy in Kubernetes, either `env` or
- `volume`
+ :param deploy_type: The type of secret deploy in Kubernetes, either `env` or `volume`
:type deploy_type: ``str``
:param deploy_target: The environment variable to be created in the worker.
:type deploy_target: ``str``
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/ad4e67ce/airflow/contrib/kubernetes/worker_configuration.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/kubernetes/worker_configuration.py b/airflow/contrib/kubernetes/worker_configuration.py
new file mode 100644
index 0000000..5e87941
--- /dev/null
+++ b/airflow/contrib/kubernetes/worker_configuration.py
@@ -0,0 +1,158 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed 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 copy
+import os
+import six
+
+from airflow.contrib.kubernetes.pod import Pod
+from airflow.contrib.kubernetes.secret import Secret
+
+
+class WorkerConfiguration:
+ """Contains Kubernetes Airflow Worker configuration logic"""
+
+ def __init__(self, kube_config):
+ self.kube_config = kube_config
+
+ def _get_init_containers(self, volume_mounts):
+ """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:
+ return []
+
+ # Otherwise, define a git-sync init container
+ init_environment = [{
+ 'name': 'GIT_SYNC_REPO',
+ 'value': self.kube_config.git_repo
+ }, {
+ 'name': 'GIT_SYNC_BRANCH',
+ 'value': self.kube_config.git_branch
+ }, {
+ 'name': 'GIT_SYNC_ROOT',
+ 'value': '/tmp'
+ }, {
+ 'name': 'GIT_SYNC_DEST',
+ 'value': 'dags'
+ }, {
+ 'name': 'GIT_SYNC_ONE_TIME',
+ 'value': 'true'
+ }]
+ if self.kube_config.git_user:
+ init_environment.append({
+ 'name': 'GIT_SYNC_USERNAME',
+ 'value': self.kube_config.git_user
+ })
+ if self.kube_config.git_password:
+ init_environment.append({
+ 'name': 'GIT_SYNC_PASSWORD',
+ 'value': self.kube_config.git_password
+ })
+
+ volume_mounts[0]['readOnly'] = False
+ return [{
+ 'name': self.kube_config.git_sync_init_container_name,
+ 'image': self.kube_config.git_sync_container,
+ 'securityContext': {'runAsUser': 0},
+ 'env': init_environment,
+ 'volumeMounts': volume_mounts
+ }]
+
+ def _get_volumes_and_mounts(self):
+ """Determine volumes and volume mounts for Airflow workers"""
+ dags_volume_name = "airflow-dags"
+ dags_path = os.path.join(self.kube_config.dags_folder, self.kube_config.git_subpath)
+ volumes = [{
+ 'name': dags_volume_name
+ }]
+ volume_mounts = [{
+ 'name': dags_volume_name,
+ 'mountPath': dags_path,
+ 'readOnly': True
+ }]
+
+ # 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.kube_config.airflow_home)
+ volumes.append({
+ 'name': config_volume_name,
+ 'configMap': {
+ 'name': self.kube_config.airflow_configmap
+ }
+ })
+ volume_mounts.append({
+ 'name': config_volume_name,
+ 'mountPath': config_path,
+ 'subPath': 'airflow.cfg',
+ 'readOnly': True
+ })
+
+ # A PV with the DAGs should be mounted
+ if self.kube_config.dags_volume_claim:
+ volumes[0]['persistentVolumeClaim'] = {"claimName": self.kube_config.dags_volume_claim}
+ if self.kube_config.dags_volume_subpath:
+ volume_mounts[0]["subPath"] = self.kube_config.dags_volume_subpath
+ else:
+ # Create a Shared Volume for the Git-Sync module to populate
+ volumes[0]["emptyDir"] = {}
+ return volumes, volume_mounts
+
+ def _get_environment(self):
+ """Defines any necessary environment variables for the pod executor"""
+ env = {
+ 'AIRFLOW__CORE__DAGS_FOLDER': '/tmp/dags',
+ 'AIRFLOW__CORE__EXECUTOR': 'LocalExecutor'
+ }
+ if self.kube_config.airflow_configmap:
+ env['AIRFLOW__CORE__AIRFLOW_HOME'] = self.kube_config.airflow_home
+ return env
+
+ def _get_secrets(self):
+ """Defines any necessary secrets for the pod executor"""
+ worker_secrets = []
+ for env_var_name, obj_key_pair in six.iteritems(self.kube_config.kube_secrets):
+ k8s_secret_obj, k8s_secret_key = obj_key_pair.split('=')
+ worker_secrets.append(Secret('env', env_var_name, k8s_secret_obj, k8s_secret_key))
+ return worker_secrets
+
+ def _get_image_pull_secrets(self):
+ """Extracts any image pull secrets for fetching container(s)"""
+ if not self.kube_config.image_pull_secrets:
+ return []
+ return self.kube_config.image_pull_secrets.split(',')
+
+ def make_pod(self, namespace, pod_id, dag_id, task_id, execution_date, airflow_command):
+ volumes, volume_mounts = self._get_volumes_and_mounts()
+ worker_init_container_spec = self._get_init_containers(copy.deepcopy(volume_mounts))
+ return Pod(
+ namespace=namespace,
+ name=pod_id,
+ image=self.kube_config.kube_image,
+ cmds=["bash", "-cx", "--"],
+ args=[airflow_command],
+ labels={
+ "airflow-slave": "",
+ "dag_id": dag_id,
+ "task_id": task_id,
+ "execution_date": execution_date
+ },
+ envs=self._get_environment(),
+ secrets=self._get_secrets(),
+ service_account_name=self.kube_config.worker_service_account_name,
+ image_pull_secrets=self.kube_config.image_pull_secrets,
+ init_containers=worker_init_container_spec,
+ volumes=volumes,
+ volume_mounts=volume_mounts
+ )
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/ad4e67ce/airflow/jobs.py
----------------------------------------------------------------------
diff --git a/airflow/jobs.py b/airflow/jobs.py
index 9754415..fe76fbd 100644
--- a/airflow/jobs.py
+++ b/airflow/jobs.py
@@ -1587,7 +1587,9 @@ class SchedulerJob(BaseJob):
timeout = 5
self.log.info("Waiting up to %s seconds for processes to exit...", timeout)
try:
- psutil.wait_procs(child_processes, timeout)
+ psutil.wait_procs(
+ child_processes, timeout=timeout,
+ callback=lambda x: self.log.info('Terminated PID %s', x.pid))
except psutil.TimeoutExpired:
self.log.debug("Ran out of time while waiting for processes to exit")
@@ -1595,6 +1597,7 @@ class SchedulerJob(BaseJob):
child_processes = [x for x in this_process.children(recursive=True)
if x.is_running() and x.pid in pids_to_kill]
if len(child_processes) > 0:
+ self.log.info("SIGKILL processes that did not terminate gracefully")
for child in child_processes:
self.log.info("Killing child PID: %s", child.pid)
child.kill()
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/ad4e67ce/airflow/utils/dag_processing.py
----------------------------------------------------------------------
diff --git a/airflow/utils/dag_processing.py b/airflow/utils/dag_processing.py
index f75b756..543eb41 100644
--- a/airflow/utils/dag_processing.py
+++ b/airflow/utils/dag_processing.py
@@ -461,7 +461,7 @@ class DagFileProcessorManager(LoggingMixin):
def heartbeat(self):
"""
This should be periodically called by the scheduler. This method will
- kick of new processes to process DAG definition files and read the
+ kick off new processes to process DAG definition files and read the
results from the finished processors.
:return: a list of SimpleDags that were produced by processors that
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/ad4e67ce/scripts/ci/kubernetes/docker/Dockerfile
----------------------------------------------------------------------
diff --git a/scripts/ci/kubernetes/docker/Dockerfile b/scripts/ci/kubernetes/docker/Dockerfile
index a3b05b0..b1bc493 100644
--- a/scripts/ci/kubernetes/docker/Dockerfile
+++ b/scripts/ci/kubernetes/docker/Dockerfile
@@ -41,12 +41,6 @@ RUN pip install kubernetes && \
COPY airflow.tar.gz /tmp/airflow.tar.gz
RUN pip install /tmp/airflow.tar.gz
-# prep airflow
-ENV AIRFLOW_HOME=/root/airflow
-ENV AIRFLOW__CORE__EXECUTOR=KubernetesExecutor
-ENV AIRFLOW__CORE__SQL_ALCHEMY_CONN=postgresql+psycopg2://root:root@postgres-airflow:5432/airflow
-
-
COPY bootstrap.sh /bootstrap.sh
RUN chmod +x /bootstrap.sh
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/ad4e67ce/scripts/ci/kubernetes/kube/airflow.yaml.template
----------------------------------------------------------------------
diff --git a/scripts/ci/kubernetes/kube/airflow.yaml.template b/scripts/ci/kubernetes/kube/airflow.yaml.template
index a297b95..af54175 100644
--- a/scripts/ci/kubernetes/kube/airflow.yaml.template
+++ b/scripts/ci/kubernetes/kube/airflow.yaml.template
@@ -52,29 +52,31 @@ spec:
metadata:
labels:
name: airflow
- annotations:
- pod.beta.kubernetes.io/init-containers: '[
- {
- "name": "init",
- "image": "{{docker_image}}",
- "imagePullPolicy": "IfNotPresent",
- "command": [
- "bash", "-cx", "cd /usr/local/lib/python2.7/dist-packages/airflow && cp -R example_dags/* $AIRFLOW_HOME/dags/ && airflow initdb && alembic upgrade head"
- ],
- "env": [
- {"name": "AIRFLOW__KUBERNETES__CONTAINER_IMAGE", "value": ""},
- {"name": "AIRFLOW__KUBERNETES__DAGS_VOLUME_CLAIM", "value": "airflow-dags"},
- {"name": "AIRFLOW__KUBERNETES__DAGS_VOLUME_SUBPATH", "value": "git"}
- ],
- "volumeMounts": [
- {"name": "airflow-dags", "mountPath": "/root/airflow/dags"}
- ]
- }
- ]'
spec:
+ initContainers:
+ - name: "init"
+ image: "{{docker_image}}:{{docker_tag}}"
+ imagePullPolicy: "IfNotPresent"
+ volumeMounts:
+ - name: airflow-configmap
+ mountPath: /root/airflow/airflow.cfg
+ subPath: airflow.cfg
+ - name: airflow-dags
+ mountPath: /root/airflow/dags
+ env:
+ - name: SQL_ALCHEMY_CONN
+ valueFrom:
+ secretKeyRef:
+ name: airflow-secrets
+ key: sql_alchemy_conn
+ command:
+ - "bash"
+ args:
+ - "-cx"
+ - "cd /usr/local/lib/python2.7/dist-packages/airflow && cp -R example_dags/* /root/airflow/dags/ && airflow initdb && alembic upgrade head"
containers:
- name: web
- image: {{docker_image}}
+ image: {{docker_image}}:{{docker_tag}}
imagePullPolicy: IfNotPresent
ports:
- name: web
@@ -85,22 +87,15 @@ spec:
valueFrom:
fieldRef:
fieldPath: metadata.namespace
- - name: AIRFLOW__CORE__EXECUTOR
- value: KubernetesExecutor
- - name: AIRFLOW__KUBERNETES__CONTAINER_IMAGE
- value: {{docker_image}}
- - name: AIRFLOW__KUBERNETES__DELETE_WORKER_PODS
- value: "True"
- # set these two confs
- - name: AIRFLOW__KUBERNETES__GIT_REPO
- value: https://github.com/grantnicholas/testdags.git
- - name: AIRFLOW__KUBERNETES__GIT_BRANCH
- value: master
- # or this one
- - name: AIRFLOW__KUBERNETES__DAGS_VOLUME_CLAIM
- value: airflow-dags
- #
+ - name: SQL_ALCHEMY_CONN
+ valueFrom:
+ secretKeyRef:
+ name: airflow-secrets
+ key: sql_alchemy_conn
volumeMounts:
+ - name: airflow-configmap
+ mountPath: /root/airflow/airflow.cfg
+ subPath: airflow.cfg
- name: airflow-dags
mountPath: /root/airflow/dags
readinessProbe:
@@ -126,30 +121,24 @@ spec:
valueFrom:
fieldRef:
fieldPath: metadata.namespace
- - name: AIRFLOW__CORE__EXECUTOR
- value: KubernetesExecutor
- - name: AIRFLOW__KUBERNETES__CONTAINER_IMAGE
- value: {{docker_image}}
- - name: AIRFLOW__KUBERNETES__DELETE_WORKER_PODS
- value: "True"
- # set these two confs
- - name: AIRFLOW__KUBERNETES__GIT_REPO
- value: https://github.com/grantnicholas/testdags.git
- - name: AIRFLOW__KUBERNETES__GIT_BRANCH
- value: master
- # or set this one
- - name: AIRFLOW__KUBERNETES__DAGS_VOLUME_CLAIM
- value: airflow-dags
- #
- - name: AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL
- value: "60"
+ - name: SQL_ALCHEMY_CONN
+ valueFrom:
+ secretKeyRef:
+ name: airflow-secrets
+ key: sql_alchemy_conn
volumeMounts:
+ - name: airflow-configmap
+ mountPath: /root/airflow/airflow.cfg
+ subPath: airflow.cfg
- name: airflow-dags
mountPath: /root/airflow/dags
volumes:
- name: airflow-dags
persistentVolumeClaim:
claimName: airflow-dags
+ - name: airflow-configmap
+ configMap:
+ name: airflow-configmap
---
apiVersion: v1
kind: Service
@@ -162,4 +151,45 @@ spec:
nodePort: 30809
selector:
name: airflow
+---
+apiVersion: v1
+kind: Secret
+metadata:
+ name: airflow-secrets
+type: Opaque
+data:
+ # The sql_alchemy_conn value is a base64 encoded represenation of this connection string:
+ # postgresql+psycopg2://root:root@postgres-airflow:5432/airflow
+ sql_alchemy_conn: cG9zdGdyZXNxbCtwc3ljb3BnMjovL3Jvb3Q6cm9vdEBwb3N0Z3Jlcy1haXJmbG93OjU0MzIvYWlyZmxvdwo=
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+ name: airflow-configmap
+data:
+ airflow.cfg: |
+ [core]
+ airflow_home = /root/airflow
+ dags_folder = /root/airflow/dags
+ base_log_folder = /root/airflow/logs
+ logging_level = INFO
+ executor = KubernetesExecutor
+ parallelism = 32
+ plugins_folder = /root/airflow/plugins
+ sql_alchemy_conn = $SQL_ALCHEMY_CONN
+
+ [scheduler]
+ dag_dir_list_interval = 60
+ child_process_log_directory = /root/airflow/logs/scheduler
+
+ [kubernetes]
+ airflow_configmap = airflow-configmap
+ worker_container_repository = {{docker_image}}
+ worker_container_tag = {{docker_tag}}
+ delete_worker_pods = True
+ git_repo = https://github.com/grantnicholas/testdags.git
+ git_branch = master
+ dags_volume_claim = airflow-dags
+ [kubernetes_secrets]
+ SQL_ALCHEMY_CONN = airflow-secrets=sql_alchemy_conn
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/ad4e67ce/scripts/ci/kubernetes/kube/deploy.sh
----------------------------------------------------------------------
diff --git a/scripts/ci/kubernetes/kube/deploy.sh b/scripts/ci/kubernetes/kube/deploy.sh
index 2532d83..953f50f 100755
--- a/scripts/ci/kubernetes/kube/deploy.sh
+++ b/scripts/ci/kubernetes/kube/deploy.sh
@@ -24,8 +24,7 @@ sudo mkdir -p /data/postgres-airflow
mkdir -p $DIRNAME/.generated
kubectl apply -f $DIRNAME/postgres.yaml
-sed "s#{{docker_image}}#$IMAGE:$TAG#g" $DIRNAME/airflow.yaml.template > $DIRNAME/.generated/airflow.yaml && kubectl apply -f $DIRNAME/.generated/airflow.yaml
-
+sed -e "s#{{docker_image}}#$IMAGE#g" -e "s#{{docker_tag}}#$TAG#g" $DIRNAME/airflow.yaml.template > $DIRNAME/.generated/airflow.yaml && kubectl apply -f $DIRNAME/.generated/airflow.yaml
# wait for up to 10 minutes for everything to be deployed
for i in {1..150}
@@ -33,8 +32,8 @@ do
echo "------- Running kubectl get pods -------"
PODS=$(kubectl get pods | awk 'NR>1 {print $0}')
echo "$PODS"
- NUM_AIRFLOW_READY=$(echo $PODS | grep airflow | awk '{print $2}' | grep -E '([0-9])\/(\1)' | wc -l)
- NUM_POSTGRES_READY=$(echo $PODS | grep postgres | awk '{print $2}' | grep -E '([0-9])\/(\1)' | wc -l)
+ NUM_AIRFLOW_READY=$(echo $PODS | grep airflow | awk '{print $2}' | grep -E '([0-9])\/(\1)' | wc -l | xargs)
+ NUM_POSTGRES_READY=$(echo $PODS | grep postgres | awk '{print $2}' | grep -E '([0-9])\/(\1)' | wc -l | xargs)
if [ "$NUM_AIRFLOW_READY" == "1" ] && [ "$NUM_POSTGRES_READY" == "1" ]; then
break
fi