You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by bo...@apache.org on 2018/04/23 16:43:30 UTC
incubator-airflow git commit: [AIRFLOW-2357] Add persistent volume
for the logs
Repository: incubator-airflow
Updated Branches:
refs/heads/master 1d3bb5470 -> e30a1f451
[AIRFLOW-2357] Add persistent volume for the logs
The logs are kept inside of the worker pod. By
attaching a persistent
disk we keep the logs and make them available for
the webserver.
- Remove the requirements.txt since we dont want
to maintain another
dependency file
- Fix some small casing stuff
- Removed some unused code
- Add missing shebang lines
- Started on some docs
- Fixed the logging
Closes #3252 from Fokko/airflow-2357-pd-for-logs
Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/e30a1f45
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/e30a1f45
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/e30a1f45
Branch: refs/heads/master
Commit: e30a1f451aa5ec5aca4c886067ba8946a3d33395
Parents: 1d3bb54
Author: Fokko Driesprong <fo...@godatadriven.com>
Authored: Mon Apr 23 18:43:24 2018 +0200
Committer: Bolke de Bruin <bo...@xs4all.nl>
Committed: Mon Apr 23 18:43:24 2018 +0200
----------------------------------------------------------------------
airflow/config_templates/default_airflow.cfg | 3 +
.../contrib/executors/kubernetes_executor.py | 216 +++++++++--------
airflow/contrib/kubernetes/pod.py | 8 +-
airflow/contrib/kubernetes/pod_generator.py | 150 +-----------
airflow/contrib/kubernetes/pod_launcher.py | 22 +-
.../contrib/kubernetes/worker_configuration.py | 116 +++++----
.../operators/kubernetes_pod_operator.py | 21 +-
airflow/utils/cli.py | 6 +-
docs/kubernetes.rst | 12 +-
scripts/ci/kubernetes/docker/Dockerfile | 15 +-
scripts/ci/kubernetes/docker/Dockerfile_zip | 20 --
scripts/ci/kubernetes/docker/bootstrap.sh | 3 +-
scripts/ci/kubernetes/docker/build.sh | 8 +-
scripts/ci/kubernetes/docker/requirements.txt | 103 --------
scripts/ci/kubernetes/kube/airflow.yaml | 221 +++++++++++++++++
.../ci/kubernetes/kube/airflow.yaml.template | 240 -------------------
scripts/ci/kubernetes/kube/deploy.sh | 9 +-
scripts/ci/kubernetes/kube/postgres.yaml | 28 +--
scripts/ci/kubernetes/kube/volumes.yaml | 64 +++++
19 files changed, 531 insertions(+), 734 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/airflow/config_templates/default_airflow.cfg
----------------------------------------------------------------------
diff --git a/airflow/config_templates/default_airflow.cfg b/airflow/config_templates/default_airflow.cfg
index 6a966cb..845342c 100644
--- a/airflow/config_templates/default_airflow.cfg
+++ b/airflow/config_templates/default_airflow.cfg
@@ -518,6 +518,9 @@ dags_volume_subpath =
# For DAGs mounted via a volume claim (mutually exclusive with volume claim)
dags_volume_claim =
+# A shared volume claim for the logs
+logs_volume_claim =
+
# Git credentials and repository for DAGs mounted via Git (mutually exclusive with volume claim)
git_repo =
git_branch =
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/airflow/contrib/executors/kubernetes_executor.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/executors/kubernetes_executor.py b/airflow/contrib/executors/kubernetes_executor.py
index 1a50d85..cdce95f 100644
--- a/airflow/contrib/executors/kubernetes_executor.py
+++ b/airflow/contrib/executors/kubernetes_executor.py
@@ -47,16 +47,10 @@ class KubernetesExecutorConfig:
def __repr__(self):
return "{}(image={}, request_memory={} ,request_cpu={}, limit_memory={}, " \
- "limit_cpu={}, gcp_service_account_key={})"\
- .format(
- KubernetesExecutorConfig.__name__,
- self.image,
- self.request_memory,
- self.request_cpu,
- self.limit_memory,
- self.limit_cpu,
- self.gcp_service_account_key
- )
+ "limit_cpu={}, gcp_service_account_key={})" \
+ .format(KubernetesExecutorConfig.__name__, self.image, self.request_memory,
+ self.request_cpu, self.limit_memory, self.limit_cpu,
+ self.gcp_service_account_key)
@staticmethod
def from_dict(obj):
@@ -65,33 +59,33 @@ class KubernetesExecutorConfig:
if not isinstance(obj, dict):
raise TypeError(
- "Cannot convert a non-dictionary object into a KubernetesExecutorConfig")
+ 'Cannot convert a non-dictionary object into a KubernetesExecutorConfig')
namespaced = obj.get(Executors.KubernetesExecutor, {})
return KubernetesExecutorConfig(
- image=namespaced.get("image", None),
- request_memory=namespaced.get("request_memory", None),
- request_cpu=namespaced.get("request_cpu", None),
- limit_memory=namespaced.get("limit_memory", None),
- limit_cpu=namespaced.get("limit_cpu", None),
- gcp_service_account_key=namespaced.get("gcp_service_account_key", None)
+ image=namespaced.get('image', None),
+ request_memory=namespaced.get('request_memory', None),
+ request_cpu=namespaced.get('request_cpu', None),
+ limit_memory=namespaced.get('limit_memory', None),
+ limit_cpu=namespaced.get('limit_cpu', None),
+ gcp_service_account_key=namespaced.get('gcp_service_account_key', None)
)
def as_dict(self):
return {
- "image": self.image,
- "request_memory": self.request_memory,
- "request_cpu": self.request_cpu,
- "limit_memory": self.limit_memory,
- "limit_cpu": self.limit_cpu,
- "gcp_service_account_key": self.gcp_service_account_key
+ 'image': self.image,
+ 'request_memory': self.request_memory,
+ 'request_cpu': self.request_cpu,
+ 'limit_memory': self.limit_memory,
+ 'limit_cpu': self.limit_cpu,
+ 'gcp_service_account_key': self.gcp_service_account_key
}
class KubeConfig:
- core_section = "core"
- kubernetes_section = "kubernetes"
+ core_section = 'core'
+ kubernetes_section = 'kubernetes'
@staticmethod
def safe_get(section, option, default):
@@ -146,11 +140,18 @@ class KubeConfig:
self.dags_volume_claim = self.safe_get(self.kubernetes_section,
'dags_volume_claim', None)
- # This prop may optionally be set for PV Claims and is used to locate DAGs on a
- # SubPath
+ # This prop may optionally be set for PV Claims and is used to write logs
+ self.logs_volume_claim = self.safe_get(
+ self.kubernetes_section, 'logs_volume_claim', None)
+
+ # 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)
+ # This prop may optionally be set for PV Claims and is used to write logs
+ self.base_log_folder = configuration.get(self.core_section, 'base_log_folder')
+
# 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
@@ -191,9 +192,8 @@ class KubeConfig:
def _validate(self):
if not self.dags_volume_claim and (not self.git_repo or not self.git_branch):
raise AirflowConfigException(
- "In kubernetes mode the following must be set in the `kubernetes` "
- "config section: "
- "`dags_volume_claim` or `git_repo and git_branch` ")
+ 'In kubernetes mode the following must be set in the `kubernetes` '
+ 'config section: `dags_volume_claim` or `git_repo and git_branch`')
class KubernetesJobWatcher(multiprocessing.Process, LoggingMixin, object):
@@ -211,29 +211,31 @@ class KubernetesJobWatcher(multiprocessing.Process, LoggingMixin, object):
self.resource_version = self._run(kube_client, self.resource_version,
self.worker_uuid)
except Exception:
- self.log.exception("Unknown error in KubernetesJobWatcher. Failing")
+ self.log.exception('Unknown error in KubernetesJobWatcher. Failing')
raise
else:
- self.log.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: %s', self.resource_version)
def _run(self, kube_client, resource_version, worker_uuid):
self.log.info(
- "Event: and now my watch begins starting at resource_version: {}"
- .format(resource_version))
+ 'Event: and now my watch begins starting at resource_version: %s',
+ resource_version
+ )
watcher = watch.Watch()
- kwargs = {"label_selector": "airflow-worker={}".format(worker_uuid)}
+ kwargs = {'label_selector': 'airflow-worker={}'.format(worker_uuid)}
if resource_version:
- kwargs["resource_version"] = resource_version
+ kwargs['resource_version'] = resource_version
last_resource_version = None
for event in watcher.stream(kube_client.list_namespaced_pod, self.namespace,
**kwargs):
task = event['object']
self.log.info(
- "Event: {} had an event of type {}".format(task.metadata.name,
- event['type']))
+ 'Event: %s had an event of type %s',
+ task.metadata.name, event['type']
+ )
if event['type'] == 'ERROR':
return self.process_error(event)
self.process_status(
@@ -245,44 +247,50 @@ class KubernetesJobWatcher(multiprocessing.Process, LoggingMixin, object):
return last_resource_version
def process_error(self, event):
- self.log.error("Encountered Error response from k8s list namespaced pod "
- "stream => {}".format(event))
+ self.log.error(
+ 'Encountered Error response from k8s list namespaced pod stream => %s',
+ event
+ )
raw_object = event['raw_object']
if raw_object['code'] == 410:
- self.log.info('Kubernetes resource version is too old, must '
- 'reset to 0 => {}'.format(raw_object['message']))
+ self.log.info(
+ 'Kubernetes resource version is too old, must reset to 0 => %s',
+ raw_object['message']
+ )
# Return resource version 0
return '0'
raise AirflowException(
- 'Kubernetes failure for {} with code {} and message: {}'
- .format(raw_object['reason'], raw_object['code'], raw_object['message']))
+ 'Kubernetes failure for %s with code %s and message: %s',
+ raw_object['reason'], raw_object['code'], raw_object['message']
+ )
def process_status(self, pod_id, status, labels, resource_version):
if status == 'Pending':
- self.log.info("Event: {} Pending".format(pod_id))
+ self.log.info('Event: %s Pending', pod_id)
elif status == 'Failed':
- self.log.info("Event: {} Failed".format(pod_id))
+ self.log.info('Event: %s Failed', pod_id)
self.watcher_queue.put((pod_id, State.FAILED, labels, resource_version))
elif status == 'Succeeded':
- self.log.info("Event: {} Succeeded".format(pod_id))
+ self.log.info('Event: %s Succeeded', pod_id)
self.watcher_queue.put((pod_id, State.SUCCESS, labels, resource_version))
elif status == 'Running':
- self.log.info("Event: {} is Running".format(pod_id))
+ self.log.info('Event: %s is Running', pod_id)
else:
- self.log.warn("Event: Invalid state: {} on pod: {} with labels: {} "
- "with resource_version: {}"
- .format(status, pod_id, labels, resource_version))
+ self.log.warn(
+ 'Event: Invalid state: %s on pod: %s with labels: %s with '
+ 'resource_version: %s', status, pod_id, labels, resource_version
+ )
-class AirflowKubernetesScheduler(LoggingMixin, object):
- def __init__(self, kube_config, task_queue, result_queue,
- session, kube_client, worker_uuid):
- self.log.debug("creating kubernetes executor")
+class AirflowKubernetesScheduler(LoggingMixin):
+ def __init__(self, kube_config, task_queue, result_queue, session,
+ kube_client, worker_uuid):
+ self.log.debug("Creating Kubernetes executor")
self.kube_config = kube_config
self.task_queue = task_queue
self.result_queue = result_queue
self.namespace = self.kube_config.kube_namespace
- self.log.debug("k8s: using namespace {}".format(self.namespace))
+ 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 = WorkerConfiguration(kube_config=self.kube_config)
@@ -303,8 +311,8 @@ class AirflowKubernetesScheduler(LoggingMixin, object):
pass
else:
self.log.error(
- "Error while health checking kube watcher process. "
- "Process died for unknown reasons")
+ '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):
@@ -312,14 +320,14 @@ class AirflowKubernetesScheduler(LoggingMixin, object):
The run_next command will check the task_queue for any un-run jobs.
It will then create a unique job-id, launch that job in the cluster,
- and store relevent info in the current_jobs map so we can track the job's
+ and store relevant info in the current_jobs map so we can track the job's
status
"""
- self.log.info('k8s: job is {}'.format(str(next_job)))
+ self.log.info('Kubernetes job is %s', str(next_job))
key, command, kube_executor_config = next_job
dag_id, task_id, execution_date = key
- self.log.debug("k8s: running for command {}".format(command))
- self.log.debug("k8s: launching image {}".format(self.kube_config.kube_image))
+ self.log.debug("Kubernetes running for command %s", command)
+ self.log.debug("Kubernetes launching image %s", self.kube_config.kube_image)
pod = self.worker_configuration.make_pod(
namespace=self.namespace, worker_uuid=self.worker_uuid,
pod_id=self._create_pod_id(dag_id, task_id),
@@ -329,7 +337,7 @@ class AirflowKubernetesScheduler(LoggingMixin, object):
)
# the watcher will monitor pods, so we do not block.
self.launcher.run_pod_async(pod)
- self.log.debug("k8s: Job created!")
+ self.log.debug("Kubernetes Job created!")
def delete_pod(self, pod_id):
if self.kube_config.delete_worker_pods:
@@ -337,12 +345,12 @@ class AirflowKubernetesScheduler(LoggingMixin, object):
self.kube_client.delete_namespaced_pod(
pod_id, self.namespace, body=client.V1DeleteOptions())
except ApiException as e:
+ # If the pod is already deleted
if e.status != 404:
raise
def sync(self):
"""
-
The sync function checks the status of all currently running kubernetes jobs.
If a job is completed, it's status is placed in the result queue to
be sent back to the scheduler.
@@ -357,11 +365,12 @@ class AirflowKubernetesScheduler(LoggingMixin, object):
def process_watcher_task(self):
pod_id, state, labels, resource_version = self.watcher_queue.get()
self.log.info(
- "Attempting to finish pod; pod_id: {}; state: {}; labels: {}"
- .format(pod_id, state, labels))
+ 'Attempting to finish pod; pod_id: %s; state: %s; labels: %s',
+ pod_id, state, labels
+ )
key = self._labels_to_key(labels=labels)
if key:
- self.log.debug("finishing job {} - {} ({})".format(key, state, pod_id))
+ self.log.debug('finishing job %s - %s (%s)', key, state, pod_id)
self.result_queue.put((key, state, pod_id, resource_version))
@staticmethod
@@ -420,8 +429,6 @@ class AirflowKubernetesScheduler(LoggingMixin, object):
:param string: string
:return: datetime.datetime object
"""
- print("unparsing date!")
-
return parser.parse(string.replace('_plus_', '+').replace("_", ":"))
@staticmethod
@@ -438,13 +445,13 @@ class AirflowKubernetesScheduler(LoggingMixin, object):
def _labels_to_key(self, labels):
try:
return (
- labels["dag_id"], labels["task_id"],
- self._label_safe_datestring_to_datetime(labels["execution_date"]))
+ labels['dag_id'], labels['task_id'],
+ self._label_safe_datestring_to_datetime(labels['execution_date']))
except Exception as e:
self.log.warn(
- "Error while converting labels to key; labels: {}; exception: {}".format(
- labels, e
- ))
+ 'Error while converting labels to key; labels: %s; exception: %s',
+ labels, e
+ )
return None
@@ -478,26 +485,27 @@ class KubernetesExecutor(BaseExecutor, LoggingMixin):
queued_tasks = self._session.query(
TaskInstance).filter(TaskInstance.state == State.QUEUED).all()
self.log.info(
- "When executor started up, found {} queued task instances".format(
- len(queued_tasks)))
+ 'When executor started up, found %s queued task instances',
+ len(queued_tasks)
+ )
- for t in queued_tasks:
- dict_string = "dag_id={},task_id={},execution_date={},airflow-worker={}"\
- .format(t.dag_id, t.task_id,
+ for task in queued_tasks:
+ dict_string = "dag_id={},task_id={},execution_date={},airflow-worker={}" \
+ .format(task.dag_id, task.task_id,
AirflowKubernetesScheduler._datetime_to_label_safe_datestring(
- t.execution_date), self.worker_uuid)
+ task.execution_date), self.worker_uuid)
kwargs = dict(label_selector=dict_string)
pod_list = self.kube_client.list_namespaced_pod(
self.kube_config.kube_namespace, **kwargs)
if len(pod_list.items) == 0:
self.log.info(
- "TaskInstance: {} found in queued state but was not launched, "
- "rescheduling"
- .format(t))
+ 'TaskInstance: %s found in queued state but was not launched, '
+ 'rescheduling', task
+ )
self._session.query(TaskInstance).filter(
- TaskInstance.dag_id == t.dag_id,
- TaskInstance.task_id == t.task_id,
- TaskInstance.execution_date == t.execution_date
+ TaskInstance.dag_id == task.dag_id,
+ TaskInstance.task_id == task.task_id,
+ TaskInstance.execution_date == task.execution_date
).update({TaskInstance.state: State.NONE})
self._session.commit()
@@ -518,9 +526,11 @@ class KubernetesExecutor(BaseExecutor, LoggingMixin):
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))
+ self.log.exception(
+ 'Exception while trying to inject secret. '
+ 'Secret name: %s, error details: %s',
+ secret_name, e
+ )
raise
# For each GCP service account key, inject it as a secret in executor
@@ -535,11 +545,11 @@ class KubernetesExecutor(BaseExecutor, LoggingMixin):
_create_or_update_secret(service_account['name'], service_account['path'])
def start(self):
- self.log.info('k8s: starting kubernetes executor')
+ self.log.info('Start Kubernetes executor')
self._session = settings.Session()
self.worker_uuid = KubeWorkerIdentifier.get_or_create_current_kube_worker_uuid(
self._session)
- self.log.debug('k8s: starting with worker_uuid: {}'.format(self.worker_uuid))
+ self.log.debug('Start with worker_uuid: %s', self.worker_uuid)
# always need to reset resource version since we don't know
# when we last started, note for behavior below
# https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#list_namespaced_pod
@@ -556,15 +566,17 @@ class KubernetesExecutor(BaseExecutor, LoggingMixin):
def execute_async(self, key, command, queue=None, executor_config=None):
self.log.info(
- "k8s: adding task {} with command {} with executor_config {}".format(
- key, command, executor_config
- ))
+ 'Add task %s with command %s with executor_config %s',
+ key, command, executor_config
+ )
kube_executor_config = KubernetesExecutorConfig.from_dict(executor_config)
self.task_queue.put((key, command, kube_executor_config))
def sync(self):
- self.log.info("self.running: {}".format(self.running))
- self.log.info("self.queued: {}".format(self.queued_tasks))
+ if self.running:
+ self.log.info('self.running: %s', self.running)
+ if self.queued_tasks:
+ self.log.info('self.queued: %s', self.queued_tasks)
self.kube_scheduler.sync()
last_resource_version = None
@@ -572,7 +584,7 @@ class KubernetesExecutor(BaseExecutor, LoggingMixin):
results = self.result_queue.get()
key, state, pod_id, resource_version = results
last_resource_version = resource_version
- self.log.info("Changing state of {} to {}".format(results, state))
+ self.log.info('Changing state of %s to %s', results, state)
self._change_state(key, state, pod_id)
KubeResourceVersion.checkpoint_resource_version(
@@ -586,10 +598,10 @@ class KubernetesExecutor(BaseExecutor, LoggingMixin):
if state != State.RUNNING:
self.kube_scheduler.delete_pod(pod_id)
try:
- self.log.info("popping: {}".format(str(key)))
+ self.log.info('Deleted pod: %s', str(key))
self.running.pop(key)
- except KeyError as k:
- print("{}".format(k))
+ except KeyError as _:
+ self.log.debug('Could not find key: %s', str(key))
pass
self.event_buffer[key] = state
(dag_id, task_id, ex_time) = key
@@ -605,5 +617,5 @@ class KubernetesExecutor(BaseExecutor, LoggingMixin):
self._session.commit()
def end(self):
- self.log.info('ending kube executor')
+ self.log.info('Shutting down Kubernetes executor')
self.task_queue.join()
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/airflow/contrib/kubernetes/pod.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/kubernetes/pod.py b/airflow/contrib/kubernetes/pod.py
index 8dbc947..df60c8a 100644
--- a/airflow/contrib/kubernetes/pod.py
+++ b/airflow/contrib/kubernetes/pod.py
@@ -44,10 +44,10 @@ class Pod:
Represents a kubernetes pod and manages execution of a single pod.
:param image: The docker image
:type image: str
- :param env: A dict containing the environment variables
- :type env: dict
+ :param envs: A dict containing the environment variables
+ :type envs:s dict
:param cmds: The command to be run on the pod
- :type cmd: list str
+ :type cmds: list str
:param secrets: Secrets to be launched to the pod
:type secrets: list Secret
:param result: The result that will be returned to the operator after
@@ -68,7 +68,7 @@ class Pod:
volume_mounts=None,
namespace='default',
result=None,
- image_pull_policy="IfNotPresent",
+ image_pull_policy='IfNotPresent',
image_pull_secrets=None,
init_containers=None,
service_account_name=None,
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/airflow/contrib/kubernetes/pod_generator.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/kubernetes/pod_generator.py b/airflow/contrib/kubernetes/pod_generator.py
index 82ac7c4..d75ba22 100644
--- a/airflow/contrib/kubernetes/pod_generator.py
+++ b/airflow/contrib/kubernetes/pod_generator.py
@@ -35,7 +35,7 @@ class PodGenerator:
def add_init_container(self,
name,
image,
- securityContext,
+ security_context,
init_environment,
volume_mounts
):
@@ -46,7 +46,7 @@ class PodGenerator:
Args:
name (str):
image (str):
- securityContext (dict):
+ security_context (dict):
init_environment (dict):
volume_mounts (dict):
@@ -57,7 +57,7 @@ class PodGenerator:
{
'name': name,
'image': image,
- 'securityContext': securityContext,
+ 'securityContext': security_context,
'env': init_environment,
'volumeMounts': volume_mounts
}
@@ -118,18 +118,10 @@ class PodGenerator:
return []
return self.kube_config.image_pull_secrets.split(',')
- def make_pod(self, namespace, image, pod_id, cmds,
- arguments, labels, kube_executor_config=None):
+ def make_pod(self, namespace, image, pod_id, cmds, arguments, labels):
volumes, volume_mounts = self._get_volumes_and_mounts()
worker_init_container_spec = self._get_init_containers()
- # resources = Resources(
- # request_memory=kube_executor_config.request_memory,
- # request_cpu=kube_executor_config.request_cpu,
- # limit_memory=kube_executor_config.limit_memory,
- # limit_cpu=kube_executor_config.limit_cpu
- # )
-
return Pod(
namespace=namespace,
name=pod_id + "-" + str(uuid.uuid1())[:8],
@@ -146,137 +138,3 @@ class PodGenerator:
volume_mounts=volume_mounts,
resources=None
)
-
-
-'''
-This class is a necessary building block to the kubernetes executor, which will be PR'd
-shortly
-'''
-
-
-class WorkerGenerator(PodGenerator):
- def __init__(self, kube_config):
- PodGenerator.__init__(self, kube_config)
- self.volumes, self.volume_mounts = self._init_volumes_and_mounts()
- self.init_containers = self._init_init_containers()
-
- def _init_volumes_and_mounts(self):
- 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 _init_labels(self, dag_id, task_id, execution_date, worker_uuid):
- return {
- "airflow-worker": worker_uuid,
- "dag_id": dag_id,
- "task_id": task_id,
- "execution_date": execution_date
- },
-
- def _get_environment(self):
- env = super(self, WorkerGenerator).env_vars
- """Defines any necessary environment variables for the pod executor"""
- env['AIRFLOW__CORE__DAGS_FOLDER'] = '/tmp/dags'
- env['AIRFLOW__CORE__EXECUTOR'] = 'LocalExecutor'
-
- if self.kube_config.airflow_configmap:
- env['AIRFLOW__CORE__AIRFLOW_HOME'] = self.kube_config.airflow_home
- return env
-
- def _init_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 make_worker_pod(self,
- namespace,
- worker_uuid,
- pod_id,
- dag_id,
- task_id,
- execution_date,
- airflow_command,
- kube_executor_config):
- cmds = ["bash", "-cx", "--"]
- labels = self._init_labels(dag_id, task_id, execution_date, worker_uuid)
- PodGenerator.make_pod(self,
- namespace=namespace,
- pod_id=pod_id,
- cmds=cmds,
- arguments=airflow_command,
- labels=labels,
- kube_executor_config=kube_executor_config)
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/airflow/contrib/kubernetes/pod_launcher.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/kubernetes/pod_launcher.py b/airflow/contrib/kubernetes/pod_launcher.py
index 23d3a7c..37f0a04 100644
--- a/airflow/contrib/kubernetes/pod_launcher.py
+++ b/airflow/contrib/kubernetes/pod_launcher.py
@@ -45,10 +45,10 @@ class PodLauncher(LoggingMixin):
def run_pod_async(self, pod):
req = self.kube_req_factory.create(pod)
- self.log.debug('Pod Creation Request: \n{}'.format(json.dumps(req, indent=2)))
+ self.log.debug('Pod Creation Request: \n%s', 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))
+ self.log.debug('Pod Creation Response: %s', resp)
except ApiException:
self.log.exception('Exception when attempting to create Namespaced Pod.')
raise
@@ -90,15 +90,14 @@ class PodLauncher(LoggingMixin):
self.log.info(line)
else:
while self.pod_is_running(pod):
- self.log.info("Pod {} has state {}".format(pod.name, State.RUNNING))
+ self.log.info('Pod %s has state %s', pod.name, State.RUNNING)
time.sleep(2)
return self._task_status(self.read_pod(pod))
def _task_status(self, event):
- # type: (V1Pod) -> State
self.log.info(
- "Event: {} had an event of type {}".format(event.metadata.name,
- event.status.phase))
+ 'Event: %s had an event of type %s',
+ event.metadata.name, event.status.phase)
status = self.process_status(event.metadata.name, event.status.phase)
return status
@@ -114,21 +113,22 @@ class PodLauncher(LoggingMixin):
try:
return self._client.read_namespaced_pod(pod.name, pod.namespace)
except HTTPError as e:
- raise AirflowException("There was an error reading the kubernetes API: {}"
- .format(e))
+ raise AirflowException(
+ 'There was an error reading the kubernetes API: {}'.format(e)
+ )
def process_status(self, job_id, status):
status = status.lower()
if status == PodStatus.PENDING:
return State.QUEUED
elif status == PodStatus.FAILED:
- self.log.info("Event: {} Failed".format(job_id))
+ self.log.info('Event with job id %s Failed', job_id)
return State.FAILED
elif status == PodStatus.SUCCEEDED:
- self.log.info("Event: {} Succeeded".format(job_id))
+ self.log.info('Event with job id %s Succeeded', job_id)
return State.SUCCESS
elif status == PodStatus.RUNNING:
return State.RUNNING
else:
- self.log.info("Event: Invalid state {} on job {}".format(status, job_id))
+ self.log.info('Event: Invalid state %s on job %s', status, job_id)
return State.FAILED
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/airflow/contrib/kubernetes/worker_configuration.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/kubernetes/worker_configuration.py b/airflow/contrib/kubernetes/worker_configuration.py
index e97f5a3..cd2cb9f 100644
--- a/airflow/contrib/kubernetes/worker_configuration.py
+++ b/airflow/contrib/kubernetes/worker_configuration.py
@@ -72,48 +72,6 @@ class WorkerConfiguration:
'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 = {
@@ -139,9 +97,69 @@ class WorkerConfiguration:
return []
return self.kube_config.image_pull_secrets.split(',')
+ def init_volumes_and_mounts(self):
+ dags_volume_name = 'airflow-dags'
+ logs_volume_name = 'airflow-logs'
+
+ def _construct_volume(name, claim, subpath=None):
+ vo = {
+ 'name': name
+ }
+ if claim:
+ vo['persistentVolumeClaim'] = {
+ 'claimName': claim
+ }
+ if subpath:
+ vo['subPath'] = subpath
+ else:
+ vo['emptyDir'] = {}
+ return vo
+
+ volumes = [
+ _construct_volume(
+ dags_volume_name,
+ self.kube_config.dags_volume_claim,
+ self.kube_config.dags_volume_subpath
+ ),
+ _construct_volume(
+ logs_volume_name,
+ self.kube_config.logs_volume_claim
+ )
+ ]
+ volume_mounts = [{
+ 'name': dags_volume_name,
+ 'mountPath': os.path.join(
+ self.kube_config.dags_folder,
+ self.kube_config.git_subpath
+ ),
+ 'readOnly': True
+ }, {
+ 'name': logs_volume_name,
+ 'mountPath': self.kube_config.base_log_folder
+ }]
+
+ # 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
+ })
+
+ return volumes, volume_mounts
+
def make_pod(self, namespace, worker_uuid, pod_id, dag_id, task_id, execution_date,
airflow_command, kube_executor_config):
- volumes, volume_mounts = self._get_volumes_and_mounts()
+ volumes, volume_mounts = self.init_volumes_and_mounts()
worker_init_container_spec = self._get_init_containers(
copy.deepcopy(volume_mounts))
resources = Resources(
@@ -152,20 +170,20 @@ class WorkerConfiguration:
)
gcp_sa_key = kube_executor_config.gcp_service_account_key
annotations = {
- "iam.cloud.google.com/service-account": gcp_sa_key
+ 'iam.cloud.google.com/service-account': gcp_sa_key
} if gcp_sa_key else {}
return Pod(
namespace=namespace,
name=pod_id,
image=kube_executor_config.image or self.kube_config.kube_image,
- cmds=["bash", "-cx", "--"],
+ cmds=['bash', '-cx', '--'],
args=[airflow_command],
labels={
- "airflow-worker": worker_uuid,
- "dag_id": dag_id,
- "task_id": task_id,
- "execution_date": execution_date
+ 'airflow-worker': worker_uuid,
+ 'dag_id': dag_id,
+ 'task_id': task_id,
+ 'execution_date': execution_date
},
envs=self._get_environment(),
secrets=self._get_secrets(),
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/airflow/contrib/operators/kubernetes_pod_operator.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/operators/kubernetes_pod_operator.py b/airflow/contrib/operators/kubernetes_pod_operator.py
index 82dfa52..bcc3cad 100644
--- a/airflow/contrib/operators/kubernetes_pod_operator.py
+++ b/airflow/contrib/operators/kubernetes_pod_operator.py
@@ -33,14 +33,14 @@ class KubernetesPodOperator(BaseOperator):
client = kube_client.get_kube_client(in_cluster=self.in_cluster)
gen = pod_generator.PodGenerator()
- pod = gen.make_pod(namespace=self.namespace,
- image=self.image,
- pod_id=self.name,
- cmds=self.cmds,
- arguments=self.arguments,
- labels=self.labels,
- kube_executor_config=self.kube_executor_config
- )
+ pod = gen.make_pod(
+ namespace=self.namespace,
+ image=self.image,
+ pod_id=self.name,
+ cmds=self.cmds,
+ arguments=self.arguments,
+ labels=self.labels
+ )
launcher = pod_launcher.PodLauncher(client)
final_state = launcher.run_pod(
@@ -48,10 +48,9 @@ class KubernetesPodOperator(BaseOperator):
startup_timeout=self.startup_timeout_seconds,
get_logs=self.get_logs)
if final_state != State.SUCCESS:
- raise AirflowException("Pod returned a failure")
+ raise AirflowException('Pod returned a failure')
except AirflowException as ex:
- raise AirflowException("Pod Launching failed: {error}".format(error=ex))
-
+ raise AirflowException('Pod Launching failed: {error}'.format(error=ex))
@apply_defaults
def __init__(self,
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/airflow/utils/cli.py
----------------------------------------------------------------------
diff --git a/airflow/utils/cli.py b/airflow/utils/cli.py
index 746e9ba..4a1e57a 100644
--- a/airflow/utils/cli.py
+++ b/airflow/utils/cli.py
@@ -7,9 +7,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
@@ -74,7 +74,7 @@ def action_logging(f):
return f(*args, **kwargs)
except Exception as e:
metrics['error'] = e
- raise e
+ raise
finally:
metrics['end_datetime'] = datetime.utcnow()
cli_action_loggers.on_post_execution(**metrics)
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/docs/kubernetes.rst
----------------------------------------------------------------------
diff --git a/docs/kubernetes.rst b/docs/kubernetes.rst
index 3305a1e..1dd77cb 100644
--- a/docs/kubernetes.rst
+++ b/docs/kubernetes.rst
@@ -1,3 +1,13 @@
+Kubernetes Executor
+===================
+
+The kubernetes executor is introduced in Apache Airflow 1.10.0. The Kubernetes executor will create a new pod for every task instance.
+
+Example helm charts are available at `scripts/ci/kubernetes/kube/{airflow,volumes,postgres}.yaml` in the source distribution. The volumes are optional and depend on your configuration. There are two volumes available:
+- Dags: by storing all the dags onto the persistent disks, all the workers can read the dags from there. Another option is using git-sync, before starting the container, a git pull of the dags repository will be performed and used throughout the lifecycle of the pod/
+- Logs: by storing the logs onto a persistent disk, all the logs will be available for all the workers and the webserver itself. If you don't configure this, the logs will be lost after the worker pods shuts down. Another option is to use S3/GCS/etc to store the logs.
+
+
Kubernetes Operator
===================
@@ -22,7 +32,7 @@ Variable Description
================================= ====================================
``@namespace`` The namespace is your isolated work environment within kubernetes
``@image`` docker image you wish to launch. Defaults to dockerhub.io, but fully qualified URLS will point to custom repositories
- ``@cmds`` To start a task in a docker image, we need to tell it what to do. the cmds array is the space seperated bash command that will define the task completed by the container
+``@cmds`` To start a task in a docker image, we need to tell it what to do. the cmds array is the space seperated bash command that will define the task completed by the container
``arguments`` arguments for your bash command
``@labels`` Labels are an important element of launching kubernetes pods, as it tells kubernetes what pods a service can route to. For example, if you launch 5 postgres pods with the label {'postgres':'foo'} and create a postgres service with the same label, kubernetes will know that any time that service is queried, it can pick any of those 5 postgres instances as the endpoint for that service.
``@name`` name of the task you want to run, will be used to generate a pod id
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/scripts/ci/kubernetes/docker/Dockerfile
----------------------------------------------------------------------
diff --git a/scripts/ci/kubernetes/docker/Dockerfile b/scripts/ci/kubernetes/docker/Dockerfile
index ad7919f..967b698 100644
--- a/scripts/ci/kubernetes/docker/Dockerfile
+++ b/scripts/ci/kubernetes/docker/Dockerfile
@@ -28,17 +28,16 @@ RUN apt-get update -y && apt-get install -y \
libssl-dev \
git \
inetutils-telnet \
- bind9utils
+ bind9utils \
+ zip \
+ unzip \
+ && apt-get clean
+# Since we install vanilla Airflow, we also want to have support for Postgres and Kubernetes
RUN pip install -U setuptools && \
- pip install -U pip
-
-COPY requirements.txt /tmp/requirements.txt
-RUN pip install -r /tmp/requirements.txt
-
-RUN pip install kubernetes && \
+ pip install kubernetes && \
pip install cryptography && \
- pip install psycopg2==2.7.3.1 # I had issues with older versions of psycopg2, just a warning
+ pip install psycopg2-binary==2.7.4 # I had issues with older versions of psycopg2, just a warning
# install airflow
COPY airflow.tar.gz /tmp/airflow.tar.gz
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/scripts/ci/kubernetes/docker/Dockerfile_zip
----------------------------------------------------------------------
diff --git a/scripts/ci/kubernetes/docker/Dockerfile_zip b/scripts/ci/kubernetes/docker/Dockerfile_zip
deleted file mode 100644
index 494b16a..0000000
--- a/scripts/ci/kubernetes/docker/Dockerfile_zip
+++ /dev/null
@@ -1,20 +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. *
-
-FROM airflow/ci:latest
-
-RUN apt-get -y update && apt-get -y install zip unzip
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/scripts/ci/kubernetes/docker/bootstrap.sh
----------------------------------------------------------------------
diff --git a/scripts/ci/kubernetes/docker/bootstrap.sh b/scripts/ci/kubernetes/docker/bootstrap.sh
index b8e54e6..8dabd79 100644
--- a/scripts/ci/kubernetes/docker/bootstrap.sh
+++ b/scripts/ci/kubernetes/docker/bootstrap.sh
@@ -1,4 +1,5 @@
-#!/bin/bash
+#!/usr/bin/env bash
+#
# 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 *
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/scripts/ci/kubernetes/docker/build.sh
----------------------------------------------------------------------
diff --git a/scripts/ci/kubernetes/docker/build.sh b/scripts/ci/kubernetes/docker/build.sh
index 7f0fe61..6f14c4d 100755
--- a/scripts/ci/kubernetes/docker/build.sh
+++ b/scripts/ci/kubernetes/docker/build.sh
@@ -1,3 +1,5 @@
+#!/usr/bin/env bash
+#
# 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 *
@@ -15,7 +17,7 @@
# specific language governing permissions and limitations *
# under the License. *
-IMAGE=${1:-airflow/ci}
+IMAGE=${1:-airflow}
TAG=${2:-latest}
DIRNAME=$(cd "$(dirname "$0")"; pwd)
AIRFLOW_ROOT="$DIRNAME/../../../.."
@@ -27,5 +29,5 @@ fi
cd $AIRFLOW_ROOT && python setup.py sdist && cp $AIRFLOW_ROOT/dist/*.tar.gz $DIRNAME/airflow.tar.gz && \
cd $DIRNAME && \
-docker build -f Dockerfile $DIRNAME --tag=${IMAGE}:${TAG} && \
-docker build -f Dockerfile_zip $DIRNAME --tag=${IMAGE}_zip:${TAG}
+docker build --pull $DIRNAME --tag=${IMAGE}:${TAG} && \
+rm $DIRNAME/airflow.tar.gz
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/scripts/ci/kubernetes/docker/requirements.txt
----------------------------------------------------------------------
diff --git a/scripts/ci/kubernetes/docker/requirements.txt b/scripts/ci/kubernetes/docker/requirements.txt
deleted file mode 100644
index aa136e0..0000000
--- a/scripts/ci/kubernetes/docker/requirements.txt
+++ /dev/null
@@ -1,103 +0,0 @@
-alembic==0.8.10
-asn1crypto==0.24.0
-aws-xray-sdk==0.95
-Babel==2.5.3
-bleach==2.1.2
-boto==2.48.0
-boto3==1.7.4
-botocore==1.10.4
-cachetools==2.0.1
-certifi==2018.4.16
-cffi==1.11.5
-chardet==3.0.4
-click==6.7
-colorama==0.3.9
-configparser==3.5.0
-cookies==2.2.1
-croniter==0.3.20
-cryptography==2.2.2
-defusedxml==0.5.0
-dill==0.2.7.1
-docker==3.2.1
-docker-pycreds==0.2.2
-docutils==0.14
-Flask==0.12.2
-Flask-Admin==1.4.1
-Flask-AppBuilder==1.10.0
-Flask-Babel==0.11.1
-Flask-Caching==1.3.3
-Flask-Login==0.2.11
-Flask-OpenID==1.2.5
-Flask-SQLAlchemy==2.1
-flask-swagger==0.2.13
-Flask-WTF==0.14.2
-freezegun==0.3.10
-funcsigs==1.0.0
-future==0.16.0
-gitdb2==2.0.3
-GitPython==2.1.9
-google-auth==1.4.1
-gunicorn==19.7.1
-html5lib==1.0.1
-idna==2.6
-ipaddress==1.0.22
-iso8601==0.1.12
-itsdangerous==0.24
-Jinja2==2.8.1
-jmespath==0.9.3
-jsondiff==1.1.1
-jsonpickle==0.9.6
-kubernetes==6.0.0
-lockfile==0.12.2
-lxml==3.8.0
-Mako==1.0.7
-Markdown==2.6.11
-MarkupSafe==1.0
-mock==2.0.0
-moto==1.3.3
-nose==1.3.7
-numpy==1.14.2
-oauthlib==2.0.7
-ordereddict==1.1
-pandas==0.22.0
-parameterized==0.6.1
-pbr==4.0.2
-pendulum==1.4.4
-psutil==4.4.2
-pyaml==17.12.1
-pyasn1==0.4.2
-pyasn1-modules==0.2.1
-pycparser==2.18
-Pygments==2.2.0
-python-daemon==2.1.2
-python-dateutil==2.6.1
-python-editor==1.0.3
-python-nvd3==0.15.0
-python-slugify==1.2.5
-python3-openid==3.1.0
-pytz==2018.4
-pytzdata==2018.4
-PyYAML==3.12
-requests==2.18.4
-requests-oauthlib==0.8.0
-responses==0.9.0
-rsa==3.4.2
-s3transfer==0.1.13
-setproctitle==1.1.10
-six==1.11.0
-smmap2==2.0.3
-SQLAlchemy==1.1.18
-SQLAlchemy-Utc==0.10.0
-tabulate==0.7.7
-thrift==0.11.0
-tzlocal==1.5.1
-unicodecsv==0.14.1
-Unidecode==1.0.22
-urllib3==1.22
-webencodings==0.5.1
-websocket-client==0.47.0
-Werkzeug==0.14.1
-wrapt==1.10.11
-WTForms==2.1
-xmltodict==0.11.0
-zope.deprecation==4.3.0
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/scripts/ci/kubernetes/kube/airflow.yaml
----------------------------------------------------------------------
diff --git a/scripts/ci/kubernetes/kube/airflow.yaml b/scripts/ci/kubernetes/kube/airflow.yaml
new file mode 100644
index 0000000..77566ae
--- /dev/null
+++ b/scripts/ci/kubernetes/kube/airflow.yaml
@@ -0,0 +1,221 @@
+# 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: rbac.authorization.k8s.io/v1beta1
+kind: ClusterRoleBinding
+metadata:
+ name: admin-rbac
+subjects:
+ - kind: ServiceAccount
+ # Reference to upper's `metadata.name`
+ name: default
+ # Reference to upper's `metadata.namespace`
+ namespace: default
+roleRef:
+ kind: ClusterRole
+ name: cluster-admin
+ apiGroup: rbac.authorization.k8s.io
+---
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+ name: airflow
+spec:
+ replicas: 1
+ template:
+ metadata:
+ labels:
+ name: airflow
+ spec:
+ initContainers:
+ - name: "init"
+ image: airflow
+ 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 heads"
+ containers:
+ - name: webserver
+ image: airflow
+ imagePullPolicy: IfNotPresent
+ ports:
+ - name: webserver
+ containerPort: 8080
+ args: ["webserver"]
+ env:
+ - name: AIRFLOW_KUBE_NAMESPACE
+ valueFrom:
+ fieldRef:
+ fieldPath: metadata.namespace
+ - 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
+ - name: airflow-logs
+ mountPath: /root/airflow/logs
+ readinessProbe:
+ initialDelaySeconds: 5
+ timeoutSeconds: 5
+ periodSeconds: 5
+ httpGet:
+ path: /admin
+ port: 8080
+ livenessProbe:
+ initialDelaySeconds: 5
+ timeoutSeconds: 5
+ failureThreshold: 5
+ httpGet:
+ path: /admin
+ port: 8080
+ - name: scheduler
+ image: airflow
+ imagePullPolicy: IfNotPresent
+ args: ["scheduler"]
+ env:
+ - name: AIRFLOW_KUBE_NAMESPACE
+ valueFrom:
+ fieldRef:
+ fieldPath: metadata.namespace
+ - 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
+ - name: airflow-logs
+ mountPath: /root/airflow/logs
+ volumes:
+ - name: airflow-dags
+ persistentVolumeClaim:
+ claimName: airflow-dags
+ - name: airflow-logs
+ persistentVolumeClaim:
+ claimName: airflow-logs
+ - name: airflow-configmap
+ configMap:
+ name: airflow-configmap
+---
+apiVersion: v1
+kind: Service
+metadata:
+ name: airflow
+spec:
+ type: NodePort
+ ports:
+ - port: 8080
+ 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 = 300
+ child_process_log_directory = /root/airflow/logs/scheduler
+ # Task instances listen for external kill signal (when you clear tasks
+ # from the CLI or the UI), this defines the frequency at which they should
+ # listen (in seconds).
+ job_heartbeat_sec = 5
+ max_threads = 16
+
+ # The scheduler constantly tries to trigger new tasks (look at the
+ # scheduler section in the docs for more information). This defines
+ # how often the scheduler should run (in seconds).
+ scheduler_heartbeat_sec = 5
+
+ # after how much time should the scheduler terminate in seconds
+ # -1 indicates to run continuously (see also num_runs)
+ run_duration = -1
+
+ # after how much time a new DAGs should be picked up from the filesystem
+ min_file_process_interval = 0
+
+ statsd_on = False
+ statsd_host = localhost
+ statsd_port = 8125
+ statsd_prefix = airflow
+
+ # How many seconds to wait between file-parsing loops to prevent the logs from being spammed.
+ min_file_parsing_loop_time = 1
+
+ print_stats_interval = 30
+ scheduler_zombie_task_threshold = 300
+ max_tis_per_query = 0
+ authenticate = False
+
+ [kubernetes]
+ airflow_configmap = airflow-configmap
+ worker_container_repository = airflow
+ worker_container_tag = latest
+ delete_worker_pods = True
+ git_repo = https://github.com/grantnicholas/testdags.git
+ git_branch = master
+ dags_volume_claim = airflow-dags
+ logs_volume_claim = airflow-logs
+
+ [kubernetes_secrets]
+ SQL_ALCHEMY_CONN = airflow-secrets=sql_alchemy_conn
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/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
deleted file mode 100644
index 6a57c24..0000000
--- a/scripts/ci/kubernetes/kube/airflow.yaml.template
+++ /dev/null
@@ -1,240 +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. *
-
-# The backing volume can be anything you want, it just needs to be `ReadWriteOnce`
-# I'm using hostPath since minikube is nice for testing, but any (non-local) volume will work on a real cluster
-kind: PersistentVolume
-apiVersion: v1
-metadata:
- name: airflow-dags
- labels:
- type: local
-spec:
- capacity:
- storage: 10Gi
- accessModes:
- - ReadWriteOnce
- hostPath:
- path: "/data/airflow-dags"
----
-kind: PersistentVolumeClaim
-apiVersion: v1
-metadata:
- name: airflow-dags
-spec:
- accessModes:
- - ReadWriteOnce
- resources:
- requests:
- storage: 10Gi
----
-apiVersion: rbac.authorization.k8s.io/v1beta1
-kind: ClusterRoleBinding
-metadata:
- name: admin-rbac
-subjects:
- - kind: ServiceAccount
- # Reference to upper's `metadata.name`
- name: default
- # Reference to upper's `metadata.namespace`
- namespace: default
-roleRef:
- kind: ClusterRole
- name: cluster-admin
- apiGroup: rbac.authorization.k8s.io
----
-apiVersion: extensions/v1beta1
-kind: Deployment
-metadata:
- name: airflow
-spec:
- replicas: 1
- template:
- metadata:
- labels:
- name: airflow
- 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 heads"
- containers:
- - name: webserver
- image: {{docker_image}}:{{docker_tag}}
- imagePullPolicy: IfNotPresent
- ports:
- - name: webserver
- containerPort: 8080
- args: ["webserver"]
- env:
- - name: AIRFLOW_KUBE_NAMESPACE
- valueFrom:
- fieldRef:
- fieldPath: metadata.namespace
- - 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:
- initialDelaySeconds: 5
- timeoutSeconds: 5
- periodSeconds: 5
- httpGet:
- path: /admin
- port: 8080
- livenessProbe:
- initialDelaySeconds: 5
- timeoutSeconds: 5
- failureThreshold: 5
- httpGet:
- path: /admin
- port: 8080
- - name: scheduler
- image: {{docker_image}}:{{docker_tag}}
- imagePullPolicy: IfNotPresent
- args: ["scheduler"]
- env:
- - name: AIRFLOW_KUBE_NAMESPACE
- valueFrom:
- fieldRef:
- fieldPath: metadata.namespace
- - 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
-metadata:
- name: airflow
-spec:
- type: NodePort
- ports:
- - port: 8080
- 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 = 300
- child_process_log_directory = /root/airflow/logs/scheduler
- # Task instances listen for external kill signal (when you clear tasks
- # from the CLI or the UI), this defines the frequency at which they should
- # listen (in seconds).
- job_heartbeat_sec = 5
- max_threads = 16
-
- # The scheduler constantly tries to trigger new tasks (look at the
- # scheduler section in the docs for more information). This defines
- # how often the scheduler should run (in seconds).
- scheduler_heartbeat_sec = 5
-
- # after how much time should the scheduler terminate in seconds
- # -1 indicates to run continuously (see also num_runs)
- run_duration = -1
-
- # after how much time a new DAGs should be picked up from the filesystem
- min_file_process_interval = 0
-
- statsd_on = False
- statsd_host = localhost
- statsd_port = 8125
- statsd_prefix = airflow
-
- # How many seconds to wait between file-parsing loops to prevent the logs from being spammed.
- min_file_parsing_loop_time = 1
-
- print_stats_interval = 30
- scheduler_zombie_task_threshold = 300
- max_tis_per_query = 0
- authenticate = False
-
- [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/e30a1f45/scripts/ci/kubernetes/kube/deploy.sh
----------------------------------------------------------------------
diff --git a/scripts/ci/kubernetes/kube/deploy.sh b/scripts/ci/kubernetes/kube/deploy.sh
index 32bed83..a5adcf8 100755
--- a/scripts/ci/kubernetes/kube/deploy.sh
+++ b/scripts/ci/kubernetes/kube/deploy.sh
@@ -1,3 +1,5 @@
+#!/usr/bin/env bash
+#
# 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 *
@@ -19,12 +21,9 @@ IMAGE=${1:-airflow/ci}
TAG=${2:-latest}
DIRNAME=$(cd "$(dirname "$0")"; pwd)
-# create an emptydir for postgres to store it's volume data in
-#sudo mkdir -p /data/postgres-airflow
-
-mkdir -p $DIRNAME/.generated
kubectl apply -f $DIRNAME/postgres.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
+kubectl apply -f $DIRNAME/volumes.yaml
+kubectl apply -f $DIRNAME/airflow.yaml
# wait for up to 10 minutes for everything to be deployed
for i in {1..150}
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/scripts/ci/kubernetes/kube/postgres.yaml
----------------------------------------------------------------------
diff --git a/scripts/ci/kubernetes/kube/postgres.yaml b/scripts/ci/kubernetes/kube/postgres.yaml
index e0bbdff..67a0635 100644
--- a/scripts/ci/kubernetes/kube/postgres.yaml
+++ b/scripts/ci/kubernetes/kube/postgres.yaml
@@ -14,31 +14,7 @@
# KIND, either express or implied. See the License for the *
# specific language governing permissions and limitations *
# under the License. *
-#
-#apiVersion: v1
-#kind: PersistentVolume
-#metadata:
-# name: postgres-airflow
-#spec:
-# accessModes:
-# - ReadWriteOnce
-# capacity:
-# storage: 5Gi
-# hostPath:
-# path: /data/postgres-airflow
-#
-#---
-#kind: PersistentVolumeClaim
-#apiVersion: v1
-#metadata:
-# name: postgres-airflow
-#spec:
-# accessModes:
-# - ReadWriteOnce
-# resources:
-# requests:
-# storage: 5Gi
-#---
+
kind: Deployment
apiVersion: extensions/v1beta1
metadata:
@@ -97,8 +73,6 @@ spec:
volumes:
- name: dbvol
emptyDir: {}
-# persistentVolumeClaim:
-# claimName: postgres-airflow
---
apiVersion: v1
kind: Service
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e30a1f45/scripts/ci/kubernetes/kube/volumes.yaml
----------------------------------------------------------------------
diff --git a/scripts/ci/kubernetes/kube/volumes.yaml b/scripts/ci/kubernetes/kube/volumes.yaml
new file mode 100644
index 0000000..073e98c
--- /dev/null
+++ b/scripts/ci/kubernetes/kube/volumes.yaml
@@ -0,0 +1,64 @@
+# 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. *
+
+# The backing volume can be anything you want, it just needs to be `ReadWriteOnce`
+# I'm using hostPath since minikube is nice for testing, but any (non-local) volume will work on a real cluster
+kind: PersistentVolume
+apiVersion: v1
+metadata:
+ name: airflow-dags
+spec:
+ accessModes:
+ - ReadWriteOnce
+ capacity:
+ storage: 2Gi
+ hostPath:
+ path: /airflow-dags/
+---
+kind: PersistentVolumeClaim
+apiVersion: v1
+metadata:
+ name: airflow-dags
+spec:
+ accessModes:
+ - ReadWriteMany
+ resources:
+ requests:
+ storage: 2Gi
+---
+kind: PersistentVolume
+apiVersion: v1
+metadata:
+ name: airflow-logs
+spec:
+ accessModes:
+ - ReadWriteMany
+ capacity:
+ storage: 2Gi
+ hostPath:
+ path: /airflow-logs/
+---
+kind: PersistentVolumeClaim
+apiVersion: v1
+metadata:
+ name: airflow-logs
+spec:
+ accessModes:
+ - ReadWriteMany
+ resources:
+ requests:
+ storage: 2Gi