Repository: apache/airflow
Pull request: Allow xcom sidecar container image to be configurable
Closes #10605.

Adding a configurable *image* value to the `KubernetesHook` which is used by `PodDefaults.SIDECAR_CONTAINER.image` will allow both:
1. Images hosted on private repositories. 
2. Overriding the default `alpine:latest` image with a specific tag.

Problem statement:
Use private docker repository with K8S operator and XCOM sidecar container
**Use private docker repository with K8S operator and XCOM sidecar container**

An extra parameter to KubernetesPodOperator: docker_repository, this allows to specify the repository where the sidecar container is located

**My company force docker proxy usage for K8S**

I need to use my company docker repository, images that are not proxifed by the company docker repository are not allowed

Hints from issue discussion:
Thanks for opening your first issue here! Be sure to follow the issue template!

would you like to add this feature @mickaelgervais ? I am happy to review it 
Hi @potiuk ,

I've submitted a PR, do nopt hesitate to add comment or if I've implemented this badly. 
The sidecar container declaration is low level in K8S operator so I've tried to not add complexity.

Regards
Hi @mickaelgervais 
To understand your issue better, are you having trouble turning on xcom because 'alpine' give you 403 error? Because in your description, you are not able to use private registry, but in my case, I can pass the whole string to the image in my KubernetesPodOperator like below. I just have problem with the xcom-sidecar, because class PodDefaults hard coded image name to be 'alpine'.
`image="<private-registry>/<dir>/<dir>/<image>:<tag>"`

I have left some comments in your pr for adding a explicit image tag, we have something like private-registry/alpine:3.10, and would like to add a variable for image tags.

Hi @potiuk 

Is this feature going live in 1.10.13 or later versions?
A work around would be to add this code block in the /home/airflow/airflow_local_settings.py
PodDefaults.SIDECAR_CONTAINER = k8s.V1Container(
    name=PodDefaults.SIDECAR_CONTAINER_NAME,
    command=['sh', '-c', PodDefaults.XCOM_CMD],
    image='<local private repository image>',
    volume_mounts=[PodDefaults.VOLUME_MOUNT],
    resources=k8s.V1ResourceRequirements(
        requests={
            "cpu": "10m",
            "memory":"32Mi"
        },
        limits={
            "cpu": "0.25",
            "memory":"32Mi"
        }
    ),
)
Docker Inc started image pull rate limiting around October 30th 2020 (https://www.docker.com/blog/checking-your-current-docker-pull-rate-limits-and-status/). As a result, all KubernetesPodPerator tasks with XCOM started to fail in large deployment. Can we have the fix in Airflow 1.10 branch as well?

Based on the severity of the failure, the bug priority should be raised higher.

```
  - image: alpine
    imageID: ""
    lastState: {}
    name: airflow-xcom-sidecar
    ready: false
    restartCount: 0
    started: false
    state:
      waiting:
        message: 'rpc error: code = Unknown desc = Error response from daemon: toomanyrequests:
          You have reached your pull rate limit. You may increase the limit by authenticating
          and upgrading: https://www.docker.com/increase-rate-limit'
        reason: ErrImagePull
```
Is there a reason why this has not been implemented yet?
> Is there a reason why this has not been implemented yet?

It was waiting for someone like you to take that over and complete it possibly? Airflow has > 1700 contributors and you can become one if you want some features to be implemented faster :)
Feel free to take it over and contribute! we even have Contributors workshops  (3 of them till the end of the year). On Airflow Slack there is a #contributors-workshop dedicated channel for it! Anyone is welcome!
sure, I think it's covered in a different feature request, let me find out
and update the comments.

Can you let me know when are the workshops and how do I attend them?

Thanks,
Zhicong

On Wed, Sep 22, 2021 at 2:22 PM Jarek Potiuk ***@***.***>
wrote:

> Feel free to take it over and contribute! we even have Contributors
> workshops (3 of them till the end of the year). On Airflow Slack there is a
> #contributors-workshop dedicated channel for it! Anyone is welcome!
>
> —
> You are receiving this because you commented.
> Reply to this email directly, view it on GitHub
> <https://github.com/apache/airflow/issues/10605#issuecomment-925246273>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AB52K2YS5FVVOH7QE55OSTTUDIUHXANCNFSM4QNED7HQ>
> .
> Triage notifications on the go with GitHub Mobile for iOS
> <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675>
> or Android
> <https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub>.
>
>

Commit message: added unit tests

Patch to review:
diff --git a/airflow/config_templates/config.yml b/airflow/config_templates/config.yml
index b20fea7ff2641..e5bf8dece18bd 100644
--- a/airflow/config_templates/config.yml
+++ b/airflow/config_templates/config.yml
@@ -2425,6 +2425,14 @@
       type: string
       example: '{"grace_period_seconds": 10}'
       default: ""
+    - name: sidecar_container_image
+      description: |
+        Define the ``image`` used by the ``PodDefaults.SIDECAR_CONTAINER`` to allow private repositories,
+        as well as custom image overrides.
+      version_added: 2.4.1
+      type: string
+      example: ~
+      default: "alpine"
     - name: enable_tcp_keepalive
       description: |
         Enables TCP keepalive mechanism. This prevents Kubernetes API requests to hang indefinitely
diff --git a/airflow/config_templates/default_airflow.cfg b/airflow/config_templates/default_airflow.cfg
index 174f33dbc8bd1..df3c25feff70f 100644
--- a/airflow/config_templates/default_airflow.cfg
+++ b/airflow/config_templates/default_airflow.cfg
@@ -1213,6 +1213,10 @@ kube_client_request_args =
 # Example: delete_option_kwargs = {{"grace_period_seconds": 10}}
 delete_option_kwargs =
 
+# Define the ``image`` used by the ``PodDefaults.SIDECAR_CONTAINER`` to allow private repositories,
+# as well as custom image overrides.
+sidecar_container_image = alpine
+
 # Enables TCP keepalive mechanism. This prevents Kubernetes API requests to hang indefinitely
 # when idle connection is time-outed on services like cloud load balancers or firewalls.
 enable_tcp_keepalive = True
diff --git a/airflow/kubernetes/pod_generator.py b/airflow/kubernetes/pod_generator.py
index c3a5f91bd65cb..23f52b2cfa74b 100644
--- a/airflow/kubernetes/pod_generator.py
+++ b/airflow/kubernetes/pod_generator.py
@@ -36,6 +36,7 @@
 from kubernetes.client import models as k8s
 from kubernetes.client.api_client import ApiClient
 
+from airflow.configuration import conf
 from airflow.exceptions import AirflowConfigException, PodReconciliationError, RemovedInAirflow3Warning
 from airflow.kubernetes.pod_generator_deprecated import PodDefaults, PodGenerator as PodGeneratorDeprecated
 from airflow.utils import yaml
@@ -146,7 +147,10 @@ def add_xcom_sidecar(pod: k8s.V1Pod) -> k8s.V1Pod:
         pod_cp.spec.volumes.insert(0, PodDefaults.VOLUME)
         pod_cp.spec.containers[0].volume_mounts = pod_cp.spec.containers[0].volume_mounts or []
         pod_cp.spec.containers[0].volume_mounts.insert(0, PodDefaults.VOLUME_MOUNT)
-        pod_cp.spec.containers.append(PodDefaults.SIDECAR_CONTAINER)
+
+        sidecar = copy.deepcopy(PodDefaults.SIDECAR_CONTAINER)
+        sidecar.image = conf.get('kubernetes', 'sidecar_container_image', fallback=sidecar.image)
+        pod_cp.spec.containers.append(sidecar)
 
         return pod_cp
 
diff --git a/airflow/kubernetes/pod_generator_deprecated.py b/airflow/kubernetes/pod_generator_deprecated.py
index f9fa2785060e1..d77f3485b3d16 100644
--- a/airflow/kubernetes/pod_generator_deprecated.py
+++ b/airflow/kubernetes/pod_generator_deprecated.py
@@ -29,6 +29,8 @@
 
 from kubernetes.client import models as k8s
 
+from airflow.configuration import conf
+
 MAX_POD_ID_LEN = 253
 
 MAX_LABEL_LEN = 63
@@ -236,7 +238,10 @@ def add_sidecar(pod: k8s.V1Pod) -> k8s.V1Pod:
         pod_cp.spec.volumes.insert(0, PodDefaults.VOLUME)
         pod_cp.spec.containers[0].volume_mounts = pod_cp.spec.containers[0].volume_mounts or []
         pod_cp.spec.containers[0].volume_mounts.insert(0, PodDefaults.VOLUME_MOUNT)
-        pod_cp.spec.containers.append(PodDefaults.SIDECAR_CONTAINER)
+
+        sidecar = copy.deepcopy(PodDefaults.SIDECAR_CONTAINER)
+        sidecar.image = conf.get('kubernetes', 'sidecar_container_image', fallback=sidecar.image)
+        pod_cp.spec.containers.append(sidecar)
 
         return pod_cp
 
diff --git a/airflow/providers/cncf/kubernetes/utils/xcom_sidecar.py b/airflow/providers/cncf/kubernetes/utils/xcom_sidecar.py
index 921ea28a768a1..7cacf78295f81 100644
--- a/airflow/providers/cncf/kubernetes/utils/xcom_sidecar.py
+++ b/airflow/providers/cncf/kubernetes/utils/xcom_sidecar.py
@@ -25,6 +25,8 @@
 
 from kubernetes.client import models as k8s
 
+from airflow.configuration import conf
+
 
 class PodDefaults:
     """Static defaults for Pods"""
@@ -54,6 +56,9 @@ def add_xcom_sidecar(pod: k8s.V1Pod) -> k8s.V1Pod:
     pod_cp.spec.volumes.insert(0, PodDefaults.VOLUME)
     pod_cp.spec.containers[0].volume_mounts = pod_cp.spec.containers[0].volume_mounts or []
     pod_cp.spec.containers[0].volume_mounts.insert(0, PodDefaults.VOLUME_MOUNT)
-    pod_cp.spec.containers.append(PodDefaults.SIDECAR_CONTAINER)
+
+    sidecar = copy.deepcopy(PodDefaults.SIDECAR_CONTAINER)
+    sidecar.image = conf.get('kubernetes', 'sidecar_container_image', fallback=sidecar.image)
+    pod_cp.spec.containers.append(sidecar)
 
     return pod_cp
diff --git a/tests/kubernetes/test_pod_generator.py b/tests/kubernetes/test_pod_generator.py
index 57039c04e8600..901219ba19f98 100644
--- a/tests/kubernetes/test_pod_generator.py
+++ b/tests/kubernetes/test_pod_generator.py
@@ -38,6 +38,7 @@
     merge_objects,
 )
 from airflow.kubernetes.secret import Secret
+from tests.test_utils.config import conf_vars
 
 
 class TestPodGenerator:
@@ -777,3 +778,11 @@ def test_validate_pod_generator(self):
             PodGenerator()
         PodGenerator(pod_template_file='tests/kubernetes/pod.yaml')
         PodGenerator(pod=k8s.V1Pod())
+
+    @conf_vars({('kubernetes', 'sidecar_container_image'): 'private.repo.com/alpine:3.16.2'})
+    def test_customize_sidecar_container_image(self):
+        pod = PodGenerator.add_xcom_sidecar(
+            PodGenerator(pod_template_file='tests/kubernetes/pod.yaml').gen_pod()
+        )
+
+        assert pod.spec.containers[-1].image == 'private.repo.com/alpine:3.16.2'


Code review comments:
diff_hunk: @@ -54,6 +56,9 @@ def add_xcom_sidecar(pod: k8s.V1Pod) -> k8s.V1Pod:
     pod_cp.spec.volumes.insert(0, PodDefaults.VOLUME)
     pod_cp.spec.containers[0].volume_mounts = pod_cp.spec.containers[0].volume_mounts or []
     pod_cp.spec.containers[0].volume_mounts.insert(0, PodDefaults.VOLUME_MOUNT)
-    pod_cp.spec.containers.append(PodDefaults.SIDECAR_CONTAINER)
+
+    sidecar = copy.deepcopy(PodDefaults.SIDECAR_CONTAINER)
+    sidecar.image = conf.get('kubernetes', 'sidecar_container_image', fallback=sidecar.image)
path: airflow/providers/cncf/kubernetes/utils/xcom_sidecar.py
text: @user1:
stuff in the k8s provider should not refer to core settings.  we're trying to keep these separate.  core should only be referred to for k8s executor.  if you want to be able to modify the xcom container, should do so via airflow connection instead... or... possibly.... through an operator param... but i think airflow connection is better.  look in KPO how you can reference attrs on the hook through `self.hook`.  this can get you access to params defined in the connection.

@author:
I will get moving on the changes.

@author:
@user1 I moved the configuration to the connection.