DockerOperator: use DOCKER_HOST as default for docker_url (apache#38387)

airflow/providers/docker/CHANGELOG.rst

@@ -27,6 +27,13 @@
 Changelog
 ---------
 
+.. note::
+  The standard ``DOCKER_HOST`` environment variable now overrides the default value
+  of the ``docker_url`` parameter when set. If ``DOCKER_HOST`` is set but you want to
+  use the previous default value, then you must explicitly set
+  ``docker_url="unix://var/run/docker.sock"`` in the ``DockerOperator`` constructor
+  or ``@task.docker`` decorator.
+
 3.9.2
 .....
 

airflow/providers/docker/operators/docker.py

@@ -20,6 +20,7 @@
 from __future__ import annotations
 
 import ast
+import os
 import pickle
 import tarfile
 import warnings
@@ -94,7 +95,8 @@ class DockerOperator(BaseOperator):
         This value gets multiplied with 1024. See
         https://docs.docker.com/engine/reference/run/#cpu-share-constraint
     :param docker_url: URL of the host running the docker daemon.
-        Default is unix://var/run/docker.sock
+        Default is the value of the ``DOCKER_HOST`` environment variable or unix://var/run/docker.sock
+        if it is unset.
     :param environment: Environment variables to set in the container. (templated)
     :param private_environment: Private environment variables to set in the container.
         These are not templated, and hidden from the website.
@@ -197,7 +199,7 @@ def __init__(
         command: str | list[str] | None = None,
         container_name: str | None = None,
         cpus: float = 1.0,
-        docker_url: str = "unix://var/run/docker.sock",
+        docker_url: str | None = None,
         environment: dict | None = None,
         private_environment: dict | None = None,
         env_file: str | None = None,
@@ -277,7 +279,7 @@ def __init__(
         self.cpus = cpus
         self.dns = dns
         self.dns_search = dns_search
-        self.docker_url = docker_url
+        self.docker_url = docker_url or os.environ.get("DOCKER_HOST") or "unix://var/run/docker.sock"
         self.environment = environment or {}
         self._private_environment = private_environment or {}
         self.env_file = env_file

airflow/providers/docker/operators/docker_swarm.py

@@ -59,7 +59,8 @@ class DockerSwarmOperator(DockerOperator):
         The default is False.
     :param command: Command to be run in the container. (templated)
     :param docker_url: URL of the host running the docker daemon.
-        Default is unix://var/run/docker.sock
+        Default is the value of the ``DOCKER_HOST`` environment variable or unix://var/run/docker.sock
+        if it is unset.
     :param environment: Environment variables to set in the container. (templated)
     :param force_pull: Pull the docker image on every run. Default is False.
     :param mem_limit: Maximum amount of memory the container can use.

docs/apache-airflow-providers-docker/decorators/docker.rst

@@ -48,7 +48,8 @@ cpus
     Number of CPUs to assign to the container. This value gets multiplied with 1024.
 docker_url
     URL of the host running the docker daemon.
-    Default is unix://var/run/docker.sock
+    Default is the value of the ``DOCKER_HOST`` environment variable or unix://var/run/docker.sock
+    if it is unset.
 environment
     Environment variables to set in the container. (templated)
 private_environment

tests/providers/docker/decorators/test_docker.py

@@ -246,3 +246,41 @@ def g():
         assert some_task.expect_airflow == clone_of_docker_operator.expect_airflow
         assert some_task.use_dill == clone_of_docker_operator.use_dill
         assert some_task.pickling_library is clone_of_docker_operator.pickling_library
+
+    def test_respect_docker_host_env(self, monkeypatch, dag_maker):
+        monkeypatch.setenv("DOCKER_HOST", "tcp://docker-host-from-env:2375")
+
+        @task.docker(image="python:3.9-slim", auto_remove="force")
+        def f():
+            pass
+
+        with dag_maker():
+            ret = f()
+
+        assert ret.operator.docker_url == "tcp://docker-host-from-env:2375"
+
+    def test_docker_host_env_empty(self, monkeypatch, dag_maker):
+        monkeypatch.setenv("DOCKER_HOST", "")
+
+        @task.docker(image="python:3.9-slim", auto_remove="force")
+        def f():
+            pass
+
+        with dag_maker():
+            ret = f()
+
+        # The docker CLI ignores the empty string and defaults to unix://var/run/docker.sock
+        # We want to ensure the same behavior.
+        assert ret.operator.docker_url == "unix://var/run/docker.sock"
+
+    def test_docker_host_env_unset(self, monkeypatch, dag_maker):
+        monkeypatch.delenv("DOCKER_HOST", raising=False)
+
+        @task.docker(image="python:3.9-slim", auto_remove="force")
+        def f():
+            pass
+
+        with dag_maker():
+            ret = f()
+
+        assert ret.operator.docker_url == "unix://var/run/docker.sock"

tests/providers/docker/operators/test_docker.py

@@ -790,3 +790,20 @@ def test_skip_exit_code_invalid(self, skip_exit_code, skip_on_exit_code):
                     skip_exit_code=skip_exit_code,
                     skip_on_exit_code=skip_on_exit_code,
                 )
+
+    def test_respect_docker_host_env(self, monkeypatch):
+        monkeypatch.setenv("DOCKER_HOST", "tcp://docker-host-from-env:2375")
+        operator = DockerOperator(task_id="test", image="test")
+        assert operator.docker_url == "tcp://docker-host-from-env:2375"
+
+    def test_docker_host_env_empty(self, monkeypatch):
+        monkeypatch.setenv("DOCKER_HOST", "")
+        operator = DockerOperator(task_id="test", image="test")
+        # The docker CLI ignores the empty string and defaults to unix://var/run/docker.sock
+        # We want to ensure the same behavior.
+        assert operator.docker_url == "unix://var/run/docker.sock"
+
+    def test_docker_host_env_unset(self, monkeypatch):
+        monkeypatch.delenv("DOCKER_HOST", raising=False)
+        operator = DockerOperator(task_id="test", image="test")
+        assert operator.docker_url == "unix://var/run/docker.sock"