[AIRFLOW-3703] Add dnsPolicy option for KubernetesPodOperator (#4520)

airflow/contrib/operators/kubernetes_pod_operator.py

@@ -96,6 +96,8 @@ class KubernetesPodOperator(BaseOperator):
     :param pod_runtime_info_envs: environment variables about
                                   pod runtime information (ip, namespace, nodeName, podName)
     :type pod_runtime_info_envs: list[PodRuntimeEnv]
+    :param dnspolicy: Specify a dnspolicy for the pod
+    :type dnspolicy: str
     """
     template_fields = ('cmds', 'arguments', 'env_vars', 'config_file')
 
@@ -136,6 +138,7 @@ class KubernetesPodOperator(BaseOperator):
             pod.configmaps = self.configmaps
             pod.security_context = self.security_context
             pod.pod_runtime_info_envs = self.pod_runtime_info_envs
+            pod.dnspolicy = self.dnspolicy
 
             launcher = pod_launcher.PodLauncher(kube_client=client,
                                                 extract_xcom=self.do_xcom_push)
@@ -196,6 +199,7 @@ class KubernetesPodOperator(BaseOperator):
                  configmaps=None,
                  security_context=None,
                  pod_runtime_info_envs=None,
+                 dnspolicy=None,
                  *args,
                  **kwargs):
         super().__init__(*args, **kwargs)
@@ -231,3 +235,4 @@ class KubernetesPodOperator(BaseOperator):
         self.configmaps = configmaps or []
         self.security_context = security_context or {}
         self.pod_runtime_info_envs = pod_runtime_info_envs or []
+        self.dnspolicy = dnspolicy

airflow/kubernetes/kubernetes_request_factory/kubernetes_request_factory.py

@@ -204,6 +204,11 @@ class KubernetesRequestFactory(metaclass=ABCMeta):
         if pod.hostnetwork:
             req['spec']['hostNetwork'] = pod.hostnetwork
 
+    @staticmethod
+    def extract_dnspolicy(pod, req):
+        if pod.dnspolicy:
+            req['spec']['dnsPolicy'] = pod.dnspolicy
+
     @staticmethod
     def extract_image_pull_secrets(pod, req):
         if pod.image_pull_secrets:

airflow/kubernetes/kubernetes_request_factory/pod_request_factory.py

@@ -67,6 +67,7 @@ spec:
         self.extract_hostnetwork(pod, req)
         self.extract_tolerations(pod, req)
         self.extract_security_context(pod, req)
+        self.extract_dnspolicy(pod, req)
         return req
 
 
@@ -137,4 +138,5 @@ spec:
         self.extract_hostnetwork(pod, req)
         self.extract_tolerations(pod, req)
         self.extract_security_context(pod, req)
+        self.extract_dnspolicy(pod, req)
         return req

airflow/kubernetes/pod.py

@@ -81,6 +81,8 @@ class Pod:
     :param pod_runtime_info_envs: environment variables about
                                   pod runtime information (ip, namespace, nodeName, podName)
     :type pod_runtime_info_envs: list[PodRuntimeEnv]
+    :param dnspolicy: Specify a dnspolicy for the pod
+    :type dnspolicy: str
     """
     def __init__(
             self,
@@ -108,7 +110,8 @@ class Pod:
             tolerations=None,
             security_context=None,
             configmaps=None,
-            pod_runtime_info_envs=None
+            pod_runtime_info_envs=None,
+            dnspolicy=None
     ):
         self.image = image
         self.envs = envs or {}
@@ -135,3 +138,4 @@ class Pod:
         self.security_context = security_context
         self.configmaps = configmaps or []
         self.pod_runtime_info_envs = pod_runtime_info_envs or []
+        self.dnspolicy = dnspolicy

tests/minikube/test_kubernetes_pod_operator.py

@@ -171,6 +171,21 @@ class KubernetesPodOperatorTest(unittest.TestCase):
         )
         k.execute(None)
 
+    @staticmethod
+    def test_pod_dnspolicy():
+        k = KubernetesPodOperator(
+            namespace='default',
+            image="ubuntu:16.04",
+            cmds=["bash", "-cx"],
+            arguments=["echo 10"],
+            labels={"foo": "bar"},
+            name="test",
+            task_id="task",
+            hostnetwork=True,
+            dnspolicy="ClusterFirstWithHostNet"
+        )
+        k.execute(None)
+
     @staticmethod
     def test_pod_node_selectors():
         node_selectors = {