Repository: incubator-airflow
Updated Branches:
  refs/heads/master 726dc9a21 -> 2fa155fe8


[AIRFLOW-2617] add imagePullPolicy config for kubernetes executor

Closes #3500 from Cplo/k8sexecutor


Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/2fa155fe
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/2fa155fe
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/2fa155fe

Branch: refs/heads/master
Commit: 2fa155fe8bd98cef5aef50f0c414f9b6f12c32c0
Parents: 726dc9a
Author: pengchen <[email protected]>
Authored: Fri Jun 15 11:38:19 2018 +0200
Committer: Fokko Driesprong <[email protected]>
Committed: Fri Jun 15 11:38:19 2018 +0200

----------------------------------------------------------------------
 .../contrib/executors/kubernetes_executor.py    | 21 +++++++++++++-------
 .../contrib/kubernetes/worker_configuration.py  |  2 ++
 scripts/ci/kubernetes/kube/configmaps.yaml      |  1 +
 3 files changed, 17 insertions(+), 7 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/2fa155fe/airflow/contrib/executors/kubernetes_executor.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/executors/kubernetes_executor.py 
b/airflow/contrib/executors/kubernetes_executor.py
index 74e4ae5..65053bd 100644
--- a/airflow/contrib/executors/kubernetes_executor.py
+++ b/airflow/contrib/executors/kubernetes_executor.py
@@ -37,9 +37,11 @@ from airflow.utils.log.logging_mixin import LoggingMixin
 
 
 class KubernetesExecutorConfig:
-    def __init__(self, image=None, request_memory=None, request_cpu=None,
-                 limit_memory=None, limit_cpu=None, 
gcp_service_account_key=None):
+    def __init__(self, image=None, image_pull_policy=None, request_memory=None,
+                 request_cpu=None, limit_memory=None, limit_cpu=None,
+                 gcp_service_account_key=None):
         self.image = image
+        self.image_pull_policy = image_pull_policy
         self.request_memory = request_memory
         self.request_cpu = request_cpu
         self.limit_memory = limit_memory
@@ -47,11 +49,11 @@ class KubernetesExecutorConfig:
         self.gcp_service_account_key = gcp_service_account_key
 
     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)
+        return "{}(image={}, image_pull_policy={}, request_memory={}, 
request_cpu={}, " \
+               "limit_memory={}, limit_cpu={}, gcp_service_account_key={})" \
+            .format(KubernetesExecutorConfig.__name__, self.image, 
self.image_pull_policy,
+                    self.request_memory, self.request_cpu, self.limit_memory,
+                    self.limit_cpu, self.gcp_service_account_key)
 
     @staticmethod
     def from_dict(obj):
@@ -66,6 +68,7 @@ class KubernetesExecutorConfig:
 
         return KubernetesExecutorConfig(
             image=namespaced.get('image', None),
+            image_pull_policy=namespaced.get('image_pull_policy', None),
             request_memory=namespaced.get('request_memory', None),
             request_cpu=namespaced.get('request_cpu', None),
             limit_memory=namespaced.get('limit_memory', None),
@@ -76,6 +79,7 @@ class KubernetesExecutorConfig:
     def as_dict(self):
         return {
             'image': self.image,
+            'image_pull_policy': self.image_pull_policy,
             'request_memory': self.request_memory,
             'request_cpu': self.request_cpu,
             'limit_memory': self.limit_memory,
@@ -101,6 +105,9 @@ class KubeConfig:
             self.kubernetes_section, 'worker_container_tag')
         self.kube_image = '{}:{}'.format(
             self.worker_container_repository, self.worker_container_tag)
+        self.kube_image_pull_policy = configuration.get(
+            self.kubernetes_section, "worker_container_image_pull_policy"
+        )
         self.delete_worker_pods = conf.getboolean(
             self.kubernetes_section, 'delete_worker_pods')
 

http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/2fa155fe/airflow/contrib/kubernetes/worker_configuration.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/kubernetes/worker_configuration.py 
b/airflow/contrib/kubernetes/worker_configuration.py
index f333f80..3ac830d 100644
--- a/airflow/contrib/kubernetes/worker_configuration.py
+++ b/airflow/contrib/kubernetes/worker_configuration.py
@@ -183,6 +183,8 @@ class WorkerConfiguration(LoggingMixin):
             namespace=namespace,
             name=pod_id,
             image=kube_executor_config.image or self.kube_config.kube_image,
+            image_pull_policy=(kube_executor_config.image_pull_policy or
+                               self.kube_config.kube_image_pull_policy),
             cmds=['bash', '-cx', '--'],
             args=[airflow_command],
             labels={

http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/2fa155fe/scripts/ci/kubernetes/kube/configmaps.yaml
----------------------------------------------------------------------
diff --git a/scripts/ci/kubernetes/kube/configmaps.yaml 
b/scripts/ci/kubernetes/kube/configmaps.yaml
index c341f1a..7b91aa2 100644
--- a/scripts/ci/kubernetes/kube/configmaps.yaml
+++ b/scripts/ci/kubernetes/kube/configmaps.yaml
@@ -178,6 +178,7 @@ data:
     airflow_configmap = airflow-configmap
     worker_container_repository = airflow
     worker_container_tag = latest
+    worker_container_image_pull_policy = IfNotPresent
     delete_worker_pods = True
     git_repo = https://github.com/apache/incubator-airflow.git
     git_branch = master

Reply via email to