Airflow DockerOperator unable to mount tmp directory correctly

Question

I am trying to run a simple python script within a docker run command scheduled with Airflow.

I have followed the instructions here Airflow init.

My .env file:

AIRFLOW_UID=1000
AIRFLOW_GID=0

And the docker-compose.yaml is based on the default one docker-compose.yaml. I had to add - /var/run/docker.sock:/var/run/docker.sock as an additional volume to run docker inside of docker.

My dag is configured as followed:

""" this is an example dag """
from datetime import timedelta

from airflow import DAG

from airflow.operators.docker_operator import DockerOperator
from airflow.utils.dates import days_ago
from docker.types import Mount

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['[email protected]'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 10,
    'retry_delay': timedelta(minutes=5),
}
with DAG(
    'msg_europe_etl',
    default_args=default_args,
    description='Process MSG_EUROPE ETL',
    schedule_interval=timedelta(minutes=15),
    start_date=days_ago(0),
    tags=['satellite_data'],
) as dag:

    download_and_store = DockerOperator(
        task_id='download_and_store',
        image='satellite_image:latest',
        auto_remove=True,
        api_version='1.41',
        mounts=[Mount(source='/home/archive_1/archive/satellite_data',
                      target='/app/data'),
                Mount(source='/home/dlassahn/projects/forecast-system/meteoIntelligence-satellite',
                      target='/app')],
        command="python3 src/scripts.py download_satellite_images "
                     "{{ (execution_date - macros.timedelta(hours=4)).strftime('%Y-%m-%d %H:%M') }} "
                     "'msg_europe' ",
    )

    download_and_store

The Airflow log:

[2021-08-03 17:23:58,691] {docker.py:231} INFO - Starting docker container from image satellite_image:latest
[2021-08-03 17:23:58,702] {taskinstance.py:1501} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.6/site-packages/docker/api/client.py", line 268, in _raise_for_status
    response.raise_for_status()
  File "/home/airflow/.local/lib/python3.6/site-packages/requests/models.py", line 943, in raise_for_status
    raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 400 Client Error: Bad Request for url: http+docker://localhost/v1.41/containers/create

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py", line 1157, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py", line 1331, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py", line 1361, in _execute_task
    result = task_copy.execute(context=context)
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/providers/docker/operators/docker.py", line 319, in execute
    return self._run_image()
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/providers/docker/operators/docker.py", line 258, in _run_image
    tty=self.tty,
  File "/home/airflow/.local/lib/python3.6/site-packages/docker/api/container.py", line 430, in create_container
    return self.create_container_from_config(config, name)
  File "/home/airflow/.local/lib/python3.6/site-packages/docker/api/container.py", line 441, in create_container_from_config
    return self._result(res, True)
  File "/home/airflow/.local/lib/python3.6/site-packages/docker/api/client.py", line 274, in _result
    self._raise_for_status(response)
  File "/home/airflow/.local/lib/python3.6/site-packages/docker/api/client.py", line 270, in _raise_for_status
    raise create_api_error_from_http_exception(e)
  File "/home/airflow/.local/lib/python3.6/site-packages/docker/errors.py", line 31, in create_api_error_from_http_exception
    raise cls(e, response=response, explanation=explanation)
docker.errors.APIError: 400 Client Error for http+docker://localhost/v1.41/containers/create: Bad Request ("invalid mount config for type "bind": bind source path does not exist: /tmp/airflowtmp037k87u6")

Trying to set mount_tmp_dir=False yield to an Dag ImportError because of unknown Keyword Argument mount_tmp_dir. (this might be an issue for the Documentation)

Nevertheless I do not know how to configure the tmp directory correctly.

My Airflow Version: 2.1.2

Answer 1

There was a bug in Docker Provider 2.0.0 which prevented Docker Operator to run with Docker-In-Docker solution.

You need to upgrade to the latest Docker Provider 2.1.0 https://airflow.apache.org/docs/apache-airflow-providers-docker/stable/index.html#id1

You can do it by extending the image as described in https://airflow.apache.org/docs/docker-stack/build.html#extending-the-image with - for example - this docker file:

FROM apache/airflow
RUN pip install --no-cache-dir apache-airflow-providers-docker==2.1.0

The operator will work out-of-the-box in this case with "fallback" mode (and Warning message), but you can also disable the mount that causes the problem. More explanation from the https://airflow.apache.org/docs/apache-airflow-providers-docker/stable/_api/airflow/providers/docker/operators/docker/index.html

By default, a temporary directory is created on the host and mounted into a container to allow storing files that together exceed the default disk size of 10GB in a container. In this case The path to the mounted directory can be accessed via the environment variable AIRFLOW_TMP_DIR.

If the volume cannot be mounted, warning is printed and an attempt is made to execute the docker command without the temporary folder mounted. This is to make it works by default with remote docker engine or when you run docker-in-docker solution and temporary directory is not shared with the docker engine. Warning is printed in logs in this case.

If you know you run DockerOperator with remote engine or via docker-in-docker you should set mount_tmp_dir parameter to False. In this case, you can still use mounts parameter to mount already existing named volumes in your Docker Engine to achieve similar capability where you can store files exceeding default disk size of the container,

Answer 2

I had the same issue and all "recommended" ways of solving the issue here and setting up mount_dir params as descripted here just lead to other errors. The one solution that helped me was wrapping the invocated by docker code with the VPN (actually this hack was taken from another docker-powered DAG that used VPN and worked well).

So the final solution looks like:

#!/bin/bash

connect_to_vpn.sh &
sleep 10
python3 my_func.py
sleep 10
stop_vpn.sh
wait -n
exit $?

To connect to VPN I used openconnect. The took can be installed with apt install and supports anyconnect protocol (it was my crucial requirement).