Airflow on a locked-down platform

Apache Airflow is a workflow orchestrator. You describe your pipelines as directed acyclic graphs (DAGs), which is a fancy way of saying “steps with dependencies”, and Airflow schedules them, runs them, retries the ones that fail, and shows you what happened. People sometimes call it a Jenkins for data pipelines. That undersells the scheduling and dependency model, but it puts you in the right neighborhood. It’s also a Python application. And Python applications are supposed to be the boring case: install the package, point it at a database, run the process. On most hosts that’s exactly how it goes. On Upsun, if you deploy it the boring way, your tasks die before writing a single line of log. The scheduler looks healthy, the web UI loads, and every task instance you trigger fails with a pid of None and a 0-byte log file. That is a genuinely confusing place to start debugging. The reason is a good one. And the fix is 30 lines of C, followed by a decision that’s more interesting than the C.

Your own process disappears

Upsun mounts /proc inside every container with hidepid. That option changes who’s allowed to see a process. Instead of every process on the box being world-readable through /proc/<pid>, a process is visible only to callers that would be allowed to ptrace it. This is part of how Upsun hardens its containers, and most of the time you never notice it. The kernel decides “allowed to ptrace” with a check called ptrace_may_access. One of its conditions is that the target process has to be dumpable. A dumpable process is one the kernel will write a core dump for, and by extension one that same-user tooling is allowed to inspect. Normal processes are dumpable by default. Here’s where Airflow and the platform disagree. Airflow’s Task SDK runs each task under a small supervisor process, and that supervisor calls prctl(PR_SET_DUMPABLE, 0) on the task. In plain terms, the task tells the kernel “nobody sharing my user id gets to read my memory”. Airflow does this on purpose, to stop one task from reading another task’s memory, environment, and secrets through ptrace or process_vm_readv. On a machine with an ordinary /proc, that’s a reasonable hardening step and nothing breaks. On a hidepid /proc, it backfires. The moment the task marks itself non-dumpable, it fails the ptrace_may_access check, so it vanishes from /proc, including from the supervisor that spawned it. The supervisor watches its task with psutil, psutil goes looking for /proc/<pid>, finds nothing, and raises NoSuchProcess. The task is killed off before it does any real work. So you get two hardening mechanisms that are each correct on their own, and together they make Airflow unable to see its own children.

Thirty lines of C

Airflow reaches prctl the way a lot of Python does: through ctypes.CDLL(None).prctl(...), which resolves the libc symbol at runtime with dlsym. Anything resolved that way can be intercepted with LD_PRELOAD. That’s the whole opening. The interposer swallows exactly one call, the “make me non-dumpable” one, and passes every other prctl straight through to the real implementation:

#define _GNU_SOURCE
#include <dlfcn.h>
#include <linux/prctl.h>

typedef int (*prctl_fn)(int, unsigned long, unsigned long,
                        unsigned long, unsigned long);

int prctl(int option, unsigned long arg2, unsigned long arg3,
          unsigned long arg4, unsigned long arg5) {
    /* Swallow only "set dumpable to 0"; report success so Airflow's
       error handling stays happy. Pass everything else through. */
    if (option == PR_SET_DUMPABLE && arg2 == 0) {
        return 0;
    }
    static prctl_fn real_prctl = 0;
    if (!real_prctl) {
        real_prctl = (prctl_fn)dlsym(RTLD_NEXT, "prctl");
    }
    return real_prctl(option, arg2, arg3, arg4, arg5);
}

Build it during the container build hook, and set LD_PRELOAD on the process that runs tasks:

cc -shared -fPIC -O2 -o no_nondumpable.so no_nondumpable.c -ldl

LD_PRELOAD=/app/lib/no_nondumpable.so airflow celery worker

Because it hooks the syscall wrapper and not an Airflow function, it keeps working across Airflow upgrades, even if the internal helper gets renamed or moved. It also refuses to overreach: a PR_SET_DUMPABLE with a non-zero value, or any other prctl option, goes through untouched. Now your tasks run.

The variables you set regardless

Whichever isolation level you land on, Airflow needs a handful of values set as environment variables on the environment before the first deploy. The sensitive ones should be set as sensitive variables so they stay out of logs and the config file.

Variable	Purpose
`AIRFLOW_ADMIN_USERNAME`	Admin login, created on first deploy
`AIRFLOW_ADMIN_EMAIL`	Admin account email
`AIRFLOW_ADMIN_PASSWORD`	Admin password (sensitive)
`AIRFLOW__CORE__FERNET_KEY`	Encrypts stored connections and variables at rest (sensitive)
`AIRFLOW__API__SECRET_KEY`	Signs session cookies (sensitive)

The rest, the database connection, the executor choice, the DAGs folder, and the log folder, live in the config. Airflow reads any AIRFLOW__SECTION__KEY environment variable as a config override, which is what makes it comfortable to run this way. The database connection comes from the Upsun relationship, decoded into a SQLAlchemy URL at startup and exported as AIRFLOW__DATABASE__SQL_ALCHEMY_CONN.

The interesting part: how much isolation do you want back?

Airflow turned that protection on for a reason. By neutralizing it, you’ve reopened the door it was closing: two tasks that share a container can read each other’s memory, environment variables, and secrets. Whether that matters depends entirely on who writes your DAGs. If every DAG in your deployment comes from the same team, targets the same data, and holds the same secrets, then a task reading another task’s environment is not a threat. You’ve decided the blast radius is one team’s own code. That’s a legitimate choice, and it buys you the simplest setup. If your DAGs come from different teams, or handle secrets that shouldn’t mix, that shared container is a problem you’ll want to fix. Upsun gives you a few ways up that ladder, and they cost progressively more to run.

You don’t need the isolation

The plainest deployment runs the whole of Airflow in one container: the API server, the scheduler, the triggerer, the DAG processor, and the workers, all supervised together. Tasks run as subprocesses next to everything else. One thing worth flagging from experience: give that container real headroom. Airflow’s scheduler and workers are memory-hungry, and forking task subprocesses out of a cramped container is where you meet flaky, hard-to-reproduce failures. A generous container profile with guaranteed resources pays for itself in debugging time you don’t spend. The next step out is a dedicated worker. The web container keeps the scheduler and the UI, and the task execution moves to its own container that you can size and scale on its own:

applications:
  airflow:
    type: python:3.12
    web:
      commands:
        start: bash scripts/start_airflow.sh
    workers:
      worker:
        commands:
          start: bash scripts/start_worker.sh

Same image, same code, separate resources. Since the two containers don’t share localhost, the worker reaches Airflow’s execution API over the environment’s public route instead. This still runs multiple tasks side by side in the worker container, so the isolation trade-off is unchanged. What you gain is the ability to scale scheduling and execution independently, which is often the real reason to split them.

You want the isolation, cheaply

The cheapest way to get isolation back is to stop running tasks side by side. Set your executor parallelism to 1, and a task never shares its container with another task, because there is no other task. Nothing to peek at. This is a real option for low-volume pipelines, and it costs nothing extra. It also serializes everything, so it falls apart the moment you need two things to run at once. For a nightly batch that walks a handful of steps, that can be perfectly fine. For anything with fan-out, it isn’t.

You want the isolation, properly

The version I’m happiest with uses task containers. A task container is an on-demand, run-to-completion container: something triggers it through the API, it runs a single command with full access to your environment’s services, and it’s destroyed when the command exits. Run each Airflow task in its own task container and the isolation question disappears. Nothing else shares that container, so the non-dumpable protection you removed was protecting against a neighbor that no longer exists. The LD_PRELOAD shim is still needed for psutil to see the process, but its only downside is gone. Airflow doesn’t ship with an executor that knows how to do this, so you write one. Airflow 3’s executor interface is small: the scheduler hands you ExecuteTask workloads, each carrying the task’s identity and a short-lived API token, and you decide where they run. Instead of enqueueing to Celery, the executor triggers a task container and passes the serialized workload in as an environment variable:

"""UpsunTaskExecutor — an Airflow 3.x executor that runs every task instance in
its own ephemeral, fully-isolated Upsun *task* container.

The scheduler hands the executor `ExecuteTask` workloads (each carrying the TI
identity + a short-lived Execution API JWT). Instead of enqueueing to Celery,
this executor triggers an Upsun task (`task-runner`), passing the serialized
workload as an env var. The task container deserializes it and runs the task
via the same Task SDK `supervise()` path the celery worker uses
(runtime/task_runner.py), talking to the Execution API with the token.

Because each container runs exactly one task and is then destroyed, the
hidepid/PR_SET_DUMPABLE shim that the runner still needs has no security
downside there — nothing else shares the container.

Auth: mints a scoped token from the in-container OAuth2 service
(localhost:8200); the app must declare `authorizations: [{type: task,
resource: task-runner, action: operate}]`. Falls back to $UPSUN_API_TOKEN.

CAVEATS (by design of the tasks feature): per-task container cold-start is a few
seconds, and the tasks API caps concurrency (default 3 parallel runs). This
executor is therefore best for isolation-sensitive / long-running tasks, not
high-throughput short ones.
"""
from __future__ import annotations

import base64
import json
import logging
import os
import time
import urllib.error
import urllib.parse
import urllib.request
from typing import TYPE_CHECKING, Sequence

from airflow.executors.base_executor import BaseExecutor
from airflow.utils.state import TaskInstanceState

if TYPE_CHECKING:
    from airflow.executors import workloads
    from airflow.models.taskinstance import TaskInstance
    from airflow.models.taskinstancekey import TaskInstanceKey

log = logging.getLogger(__name__)

_API_BASE = os.environ.get("UPSUN_API_BASE", "https://api.upsun.com")
_TOKEN_URL = "http://localhost:8200/oauth2/token"
_TERMINAL = {"complete", "cancelled"}


class UpsunTaskExecutor(BaseExecutor):
    """Run each task instance in a one-shot Upsun task container."""

    #: name of the Upsun task (the `tasks:` key in .upsun/config.yaml)
    task_name = os.environ.get("UPSUN_RUNNER_TASK", "task-runner")

    def __init__(self, parallelism: int = 32, team_name: str | None = None) -> None:
        super().__init__(parallelism=parallelism, team_name=team_name)
        self._activities: dict[TaskInstanceKey, str] = {}
        self._token: str | None = None
        self._token_expiry: float = 0.0
        project = os.environ.get("PLATFORM_PROJECT", "")
        env = os.environ.get("PLATFORM_BRANCH") or os.environ.get("PLATFORM_ENVIRONMENT", "")
        self._base = f"{_API_BASE}/projects/{project}/environments/{env}"
        # Map Airflow queue -> Upsun task definition, so specific DAGs/tasks can
        # run in a differently-resourced task container. Convention: the
        # "default" queue uses self.task_name; any other queue "q" uses
        # "<task_name>-<q>" (declare that task + its app authorization in
        # config). Override/extend via UPSUN_QUEUE_TASK_MAP (JSON
        # {"<queue>": "<upsun task name>"}).
        self._queue_map: dict = json.loads(os.environ.get("UPSUN_QUEUE_TASK_MAP", "{}"))
        # The task container can't reach the api-server on localhost and has no
        # PLATFORM_ROUTES, so we resolve the public Execution API URL here (the
        # web app knows it) and pass it to the task.
        self._exec_api = self._execution_api_url()

    @staticmethod
    def _execution_api_url() -> str:
        base = os.environ.get("AIRFLOW__API__BASE_URL")
        if base:
            return base.rstrip("/") + "/execution/"
        routes = json.loads(base64.b64decode(os.environ["PLATFORM_ROUTES"]).decode())
        for url, cfg in routes.items():
            if cfg.get("primary"):
                return url.rstrip("/") + "/execution/"
        raise RuntimeError("cannot determine Execution API URL (no AIRFLOW__API__BASE_URL / primary route)")

    def _task_for_queue(self, queue: str | None) -> str:
        q = queue or "default"
        if q in self._queue_map:
            return self._queue_map[q]
        if q == "default":
            return self.task_name
        return f"{self.task_name}-{q}"

    # -- auth ----------------------------------------------------------------
    def _token_value(self, force: bool = False) -> str:
        explicit = os.environ.get("UPSUN_API_TOKEN")
        if explicit:
            return explicit
        if not force and self._token and time.monotonic() < self._token_expiry - 30:
            return self._token
        data = urllib.parse.urlencode({"grant_type": "client_credentials"}).encode()
        req = urllib.request.Request(_TOKEN_URL, data=data, method="POST")
        req.add_header("Content-Type", "application/x-www-form-urlencoded")
        req.add_header("x-token-ttl", "900")  # match our polling cadence
        with urllib.request.urlopen(req, timeout=15) as resp:
            payload = json.loads(resp.read().decode())
        self._token = payload["access_token"]
        self._token_expiry = time.monotonic() + int(payload.get("expires_in", 900))
        return self._token

    def _api(self, method: str, url: str, body: dict | None = None, _retry: bool = True) -> dict:
        data = json.dumps(body).encode() if body is not None else None
        req = urllib.request.Request(url, data=data, method=method)
        req.add_header("Authorization", f"Bearer {self._token_value()}")
        if data is not None:
            req.add_header("Content-Type", "application/json")
        try:
            with urllib.request.urlopen(req, timeout=30) as resp:
                raw = resp.read().decode()
        except urllib.error.HTTPError as e:
            if e.code == 401 and _retry:
                self._token_value(force=True)
                return self._api(method, url, body, _retry=False)
            raise RuntimeError(f"{method} {url} -> HTTP {e.code}: {e.read().decode(errors='replace')}") from e
        return json.loads(raw) if raw else {}

    # -- scheduler interface -------------------------------------------------
    def _process_workloads(self, workloads_seq: Sequence["workloads.All"]) -> None:
        from airflow.executors.workloads import ExecuteTask

        for workload in workloads_seq:
            if not isinstance(workload, ExecuteTask):
                raise ValueError(f"{type(self).__name__} cannot handle {type(workload).__name__}")
            key = workload.ti.key
            task_name = self._task_for_queue(getattr(workload.ti, "queue", None))
            try:
                activity_id = self._trigger(workload, task_name)
            except Exception as e:  # noqa: BLE001 - report as a failed launch
                self.log.error("Failed to trigger Upsun task %r for %s: %s", task_name, key, e)
                self.queued_tasks.pop(key, None)
                self.event_buffer[key] = (TaskInstanceState.FAILED, None)
                continue
            self.queued_tasks.pop(key, None)
            self.running.add(key)
            self._activities[key] = activity_id
            self.event_buffer[key] = (TaskInstanceState.QUEUED, activity_id)
            self.log.info("Task %s (queue=%s) -> %s / Upsun activity %s",
                          key, getattr(workload.ti, "queue", None), task_name, activity_id)

    def _trigger(self, workload: "workloads.ExecuteTask", task_name: str) -> str:
        body = {"variables": {"env": {
            "UPSUN_WORKLOAD": workload.model_dump_json(),
            "AIRFLOW__CORE__EXECUTION_API_SERVER_URL": self._exec_api,
        }}}
        resp = self._api("POST", f"{self._base}/tasks/{task_name}/run", body)
        emb = resp.get("_embedded", {})
        acts = emb.get("activities")
        activity = acts[0] if isinstance(acts, list) and acts else (acts if isinstance(acts, dict) else resp)
        activity_id = activity.get("id")
        if not activity_id:
            raise RuntimeError(f"no activity id in run response: {resp}")
        return activity_id

    def sync(self) -> None:
        for key, activity_id in list(self._activities.items()):
            try:
                act = self._api("GET", f"{self._base}/activities/{activity_id}")
            except Exception as e:  # noqa: BLE001 - transient API error, retry next heartbeat
                self.log.warning("Polling activity %s failed: %s", activity_id, e)
                continue
            state = (act.get("state") or act.get("status") or "").lower()
            if state not in _TERMINAL:
                continue
            result = (act.get("result") or "").lower()
            self._activities.pop(key, None)
            if state == "complete" and result != "failure":
                self.log.info("Upsun activity %s complete -> success for %s", activity_id, key)
                self.success(key)
            else:
                self.log.warning("Upsun activity %s ended state=%s result=%s -> fail for %s",
                                 activity_id, state, result, key)
                self.fail(key)

    def try_adopt_task_instances(self, tis: Sequence["TaskInstance"]) -> Sequence["TaskInstance"]:
        # We don't persist activity ids across scheduler restarts; let the
        # scheduler reschedule orphaned TIs rather than adopt them.
        return list(tis)

    def end(self) -> None:
        self.heartbeat()

    def terminate(self) -> None:  # pragma: no cover - nothing long-lived to kill here
        pass

The task container’s side is a short script that deserializes the workload and runs it through the same Task SDK path a Celery worker would use. The token in the workload authorizes it, so the container needs no database or broker access at all. The app declares an authorization to operate the task, and Upsun’s in-container token service mints a scoped, credential-less token at runtime. The part I find genuinely useful is that you can define more than one class of task container. Route memory-hungry DAGs to a large one and everything else to a small one:

tasks:
  task-runner:
    type: python:3.12
    container_profile: HIGH_MEMORY
    source:
      root: /runtime
    run:
      command: LD_PRELOAD=/app/lib/no_nondumpable.so python task_runner.py
      timeout: 3600

Because a task container only exists while a task runs, you pay for compute per task instead of for a worker that sits idle between DAG runs. The queue-to-task mapping in the executor lets you tune that per workload. Two caveats to plan around. A per-task container has a cold start of a few seconds, so this suits isolation-sensitive or long-running tasks better than a flood of sub-second ones. And task containers cap concurrency at a default of 3 parallel runs, with further triggers queuing behind them. That default is adjustable, but design for it rather than against it.

Why go to the trouble

At the end of the ladder you have Airflow running the way it’s meant to, and you have it inside an Upsun environment. That second part is what makes the effort worth it. Branch an environment and Upsun clones its data, so a staging copy of your Airflow comes up with a real Postgres database instead of an empty one. Add object storage and the buckets clone alongside it, so your DAGs run against realistic data in staging without touching production. You get scoped secrets, per-environment variables, and the same container hardening this whole article started with, applied to a data platform you can branch, test, and throw away. The 30 lines of C get you in the door. The isolation model is the choice worth making deliberately, and it’s nice that the platform lets you make it at all.

Task containers are in prerelease. Read the task containers documentation to see how they work and request access on your project.

​Your own process disappears

​Thirty lines of C

​The variables you set regardless

​The interesting part: how much isolation do you want back?

​You don’t need the isolation

​You want the isolation, cheaply

​You want the isolation, properly

​Why go to the trouble

Your own process disappears

Thirty lines of C

The variables you set regardless

The interesting part: how much isolation do you want back?

You don’t need the isolation

You want the isolation, cheaply

You want the isolation, properly

Why go to the trouble