Allow xcom sidecar container image to be configurable

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

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

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
 

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
 

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

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'