Airflow Mesos Provider

The Airflow Mesos Provider integrates Apache Airflow with Apache Mesos.

It provides two execution models:

• MesosExecutor: distributes regular Airflow task workloads through a Mesos framework.
• MesosOperator: starts a single DAG task as a Mesos container and waits for it to finish.

Quick start

1. Install the provider and configure Airflow: Installation
2. Understand the responsibilities and data flow: Architecture
3. Use the MesosOperator for individual container tasks.
4. Read the DAG examples for complete examples.
5. Verify the setup with Testing and development.

Requirements

• Apache Airflow 3.x recommended; the provider declares Airflow >=2.0.
• Apache Mesos 1.6 or newer.
• Python 3.x.
• For Docker containers: a Mesos agent with the Docker containerizer enabled and access to the required image.

SSL and Mesos authentication are optional, but recommended for production clusters.

Further documentation

• Configuration
• Executor API
• Troubleshooting

Installation

Package installation

pip install avmesos_airflow_provider

The provider also requires a compatible Airflow, avmesos, and HTTP environment. For local development, shell.nix provides a reproducible environment.

Configure Airflow

To run regular Airflow tasks through Mesos:

[core]
executor = avmesos_airflow_provider.executors.mesos_executor.MesosExecutor

The complete example configuration is available in Configuration.

Local development

nix-shell

The Nix shell installs Airflow, avmesos, the provider, and PostgreSQL. It also initializes the local Airflow database and DAG environment.

Use MesosOperator

The operator does not require a second executor. The Airflow scheduler with MesosExecutor exposes the internal API on port 11000. A minimal example:

from datetime import datetime

from airflow import DAG
from avmesos_airflow_provider.operators.mesos import MesosOperator

with DAG("hello_mesos", schedule=None, start_date=datetime(2024, 1, 1), catchup=False) as dag:
    MesosOperator(
        task_id="hello",
        image="alpine:3.20",
        command="echo hello",
        cpus=0.1,
        mem_limit="128m",
    )

See the Operator reference for details.

Architecture

The provider offers two distinct execution paths:

MesosExecutor

MesosExecutor replaces the Airflow executor. Airflow submits scheduled task workloads to a Mesos framework. The framework accepts Mesos offers and starts Airflow tasks as container tasks.

This path is suitable when an entire Airflow DAG, or many regular Airflow tasks, should be distributed through Mesos.

MesosOperator

MesosOperator runs inside a regular Airflow DAG and starts exactly one task as a Mesos container. It is modeled after DockerOperator, but uses the local API provided by MesosExecutor.

The flow is:

1. The operator sends the container request to POST /v0/queue_command.
2. The MesosExecutor queues the request and accepts a matching Mesos offer.
3. The operator polls GET /v0/task/<task_id>.
4. The operator waits for TASK_FINISHED or reports a terminal failure to Airflow.

The API runs at http://localhost:11000 by default. It is provided by the scheduler process and is not the Mesos master API on port 5050.

Data flow

Airflow Scheduler
      |
      | MesosExecutor API :11000
      v
MesosExecutor Framework
      |
      | Mesos scheduler protocol
      v
Mesos Master :5050
      |
      v
Mesos Agent -> Docker/Mesos Container

The operator does not create its own Mesos framework. Offer distribution, resource checking, and framework authentication therefore remain centralized in the existing executor.

Configuration

Airflow executor

Enable the executor in airflow.cfg:

[core]
executor = avmesos_airflow_provider.executors.mesos_executor.MesosExecutor

Mesos configuration

The values below are examples. Adapt hosts, credentials, and images to your environment.

[mesos]
mesos_ssl = True
master = mesos-master.example.invalid:5050
framework_name = Airflow
checkpoint = True
failover_timeout = 604800
command_shell = True

task_cpu = 0.1
task_memory = 512
task_disk = 1000

authenticate = True
default_principal = <MESOS_PRINCIPAL>
default_secret = <MESOS_SECRET>

docker_image_slave = <AIRFLOW_RUNTIME_IMAGE>
docker_volume_driver = local
docker_volume_dag_name = airflowdags
docker_volume_dag_container_path = /airflow/dags/
docker_volume_logs_name = airflowlogs
docker_volume_logs_container_path = /airflow/logs/
docker_sock = /var/run/docker.sock
docker_user_group_id = <DOCKER_GROUP_ID>
docker_network_mode = bridge
docker_environment = []

api_username = <API_USERNAME>
api_password = <API_PASSWORD>
operator_api_url = http://localhost:11000

operator_api_url is the address used by MesosOperator when no airflow_scheduler_url is set on the task.

Attributes

Global attributes apply to executor tasks. Task-specific attributes are added by the operator or through executor_config:

mesos_attributes = ["airflow:true", "gpu:true?:cpu:true"]

An operator can specify additional attributes:

MesosOperator(
    task_id="cpu_task",
    image="alpine:3.20",
    command="echo hello",
    attributes=["cpu:true"],
)

Do not commit secrets or production infrastructure addresses in DAG files. Use Airflow Connections, Variables, or external secret backends for environment-specific values.

MesosOperator

MesosOperator runs one container task under Apache Mesos. Its behavior follows the Airflow DockerOperator model, but execution takes place through the MesosExecutor API.

Example

from datetime import datetime

from airflow import DAG
from avmesos_airflow_provider.operators.mesos import MesosOperator

with DAG(
    dag_id="mesos_operator_example",
    schedule=None,
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:
    MesosOperator(
        task_id="hello_mesos",
        image="alpine:3.20",
        command="echo hello from Mesos",
        cpus=0.1,
        mem_limit="128m",
        attributes=["airflow:true"],
    )

Parameters

Airflow standard parameters such as task_id, retries, pool, and queue are supported through BaseOperator.

Status behavior

The operator succeeds when the status is:

TASK_FINISHED

The following states raise AirflowException:

TASK_FAILED
TASK_ERROR
TASK_KILLED
TASK_LOST
TASK_UNREACHABLE

HTTP errors, invalid JSON responses, and exceeding startup_timeout are also reported as task failures.

Cancellation limitation

The current executor API has no separate kill endpoint for directly queued operator tasks. on_kill() logs this limitation. For long-running tasks, use Airflow timeouts and container commands that can be stopped in a controlled way.

DAG examples

MesosExecutor

The executor distributes regular Airflow tasks through Mesos. The DAG task itself does not need a special operator class:

from datetime import datetime
from airflow import DAG
from airflow.providers.standard.operators.bash import BashOperator

with DAG("executor_example", schedule=None, start_date=datetime(2024, 1, 1), catchup=False) as dag:
    BashOperator(
        task_id="show_date",
        bash_command="date",
        executor_config={
            "cpus": 0.2,
            "mem_limit": "256m",
            "attributes": ["airflow:true"],
        },
    )

MesosOperator

For one container task:

from datetime import datetime
from airflow import DAG
from avmesos_airflow_provider.operators.mesos import MesosOperator

with DAG("operator_example", schedule=None, start_date=datetime(2024, 1, 1), catchup=False) as dag:
    run = MesosOperator(
        task_id="run_command",
        image="alpine:3.20",
        command=["/bin/sh", "-c", "echo operator-ok && uname -a"],
        cpus=0.1,
        mem_limit="128m",
        environment={"EXAMPLE_MODE": "true"},
        attributes=["airflow:true"],
    )

A complete, intentionally short test DAG is available at docs/examples/dags/mesos_operator_test.py.

Development and testing

Nix environment

shell.nix provides Python, Airflow, PostgreSQL, and the provider dependencies:

nix-shell

The shell hook creates a virtual environment under /tmp/python-dev, initializes the local Airflow database, and installs the provider in editable mode.

Unit tests

The unit tests use synthetic HTTP responses and do not require a live Mesos or Airflow service:

make test

The test target runs:

python3 -m unittest discover -s tests -v

Build

make build

This creates a source distribution and wheel. Before committing, run at least make test, make build, and git diff --check.

Live smoke test

For a real integration test, load the test DAG into Airflow and trigger it manually. Requirements are a running Airflow scheduler with MesosExecutor, a reachable executor API on port 11000, and a Mesos cluster with matching agent attributes.

The Mesos master UI on port 5050 is for observation only. The operator does not communicate directly with the master UI.

Troubleshooting

Operator remains in polling

Check the following:

1. Is the Airflow scheduler running with MesosExecutor?
2. Is airflow_scheduler_url correct and reachable on port 11000?
3. Is the framework registered with the Mesos master?
4. Are there matching CPU, memory, and attribute offers?

The executor API returns HTTP 200 during queueing only to confirm acceptance into the queue. The operator then continues waiting for the Mesos status.

TASK_FAILED or TASK_ERROR

Check Mesos agent logs and the task details in the Mesos UI. Common causes include an unavailable image, insufficient resources, unmatched attributes, or an incorrect container/network mode.

TASK_LOST

The agent or framework connection was lost. Check whether the framework has reconnected and whether the agent is active.

API returns 401

Check the API configuration and the endpoint being used. /v0/dags is protected; the operator uses /v0/queue_command and /v0/task/<task_id>. Do not expose the API through a public reverse proxy without suitable authentication.

DAG is not loaded

First check imports in isolation:

airflow dags list-import-errors

Then make sure dags_folder points to the directory containing the DAG and that the provider is installed in the same Python environment as Airflow.

Resources do not match

cpus, mem_limit, and disk must fit the available Mesos offers. For attributes, at least one active agent must satisfy every constraint. Global attributes from mesos_attributes and task-specific attributes are used together.

Executor API

The API is provided by MesosExecutor in the Airflow scheduler. The default address is http://localhost:11000.

POST /v0/queue_command

Queues a direct container task.

Example request body:

{
  "airflow_task_id": "airflow.example.hello",
  "container_type": "DOCKER",
  "command": ["/bin/sh", "-c", "echo hello"],
  "image": "alpine:3.20",
  "cpus": 0.1,
  "mem_limit": "128m",
  "attributes": ["airflow:true"],
  "environment": {"MODE": "test"}
}

A successful acceptance returns HTTP 200. This only means that the request was accepted into the executor queue; Mesos execution has not necessarily finished yet.

GET /v0/task/<task_id>

Returns the latest known Mesos status for the task. The task_id must be URL-encoded if it contains characters outside the usual task ID format.

The operator expects an object with a status field, for example:

{
  "status": {
    "task_id": {"value": "airflow.example.hello"},
    "state": "TASK_FINISHED"
  }
}

During execution, TASK_STAGING, TASK_STARTING, and TASK_RUNNING may occur. Terminal failure states are translated into an Airflow task failure by the operator.

Security

The API should only be reachable on the internal Airflow/scheduler network. Store credentials in Airflow configuration or secret backends, not in versioned DAGs. Mesos authentication to the cluster is configured separately through [mesos].