Async Jobs & Workflow Pipelines

Sailfish observes your data pipelines and async jobs with the same fidelity it brings to your web and API traffic. The first integration is Apache Airflow — every DAG run, task instance, warehouse query, retry, and failure becomes a session you can search, group, and link back to the customers it affected.

The capture layer is platform-neutral. Today: Airflow 2.1 – 3.x. The same adapter pattern will support Prefect, Dagster, and other orchestrators as they ship — the data model you'll learn here doesn't change when you add another one.

What Sailfish Captures

For every task instance and DAG run, Sailfish captures:

Category	What's captured
Identity	DAG ID, run ID, task ID, attempt number, map index, task group, operator type
Timing	Logical time, data interval start/end, queued/started/ended timestamps, duration
State	A normalized state (running · success · failed · retrying · skipped · upstream_failed · deferred · cancelled · rescheduled) — the same vocabulary across every orchestrator
Topology	Upstream + downstream task IDs, trigger rule, inlets and outlets (dataset/asset URIs)
Trigger	Why this run started — scheduled, backfill, manually triggered, dataset-driven, retry
Scheduling context	Schedule expression, catchup enabled / catchup-run flag, run-creation lag, prior run's logical time
Worker	Hostname, PID, pool, queue, executor name, Kubernetes pod/namespace/node, OOM-killed flag
Warehouse queries	For every Postgres / Snowflake / BigQuery / Redshift / MySQL / MSSQL query a task runs: a normalized SQL fingerprint (sha256), the warehouse-side query ID, row count, bytes scanned, target table, duration
Outputs	Dataset URI, row count, schema-delta vs the prior run, freshness delta
Retries	Previous-attempt session, attempt number, retry reason
Failures	Exception class, trace hash, scheduler-side reasons (parse failure, queue timeout, heartbeat timeout, OOM, manual cancel)
User attribution	Which of your users were active during the task — so a failed nightly rollup links back to the customers it affected

Each task instance becomes one Sailfish session, with the full universal workflow trait blob attached. Hot fields (DAG, run, task, state) are promoted to dedicated columns so filtering and grouping in the Sailfish UI is instant.

What Sailfish Does Not Capture by Default

Privacy is the default. The following are off unless you explicitly opt in:

Off by default	Opt-in env var	What it captures when on
Raw SQL text	SAILFISH_AIRFLOW_CAPTURE_SQL_VALUES	The original SQL string (otherwise only the normalized fingerprint + warehouse query ID).
XCom values	SAILFISH_AIRFLOW_CAPTURE_XCOM_VALUES	XCom payloads (otherwise only keys + sizes).
dag_run.conf / task params	SAILFISH_AIRFLOW_CAPTURE_DAG_RUN_CONF	The conf JSON (size-capped at 4 KB, credential keys still redacted).
Raw owner / alert emails	SAILFISH_AIRFLOW_CAPTURE_OWNER_EMAILS_RAW	Original email addresses (otherwise sha256-hashed to 16 hex chars so "which on-call got paged" still joins).

Regardless of opt-ins, fields whose keys match password, secret, token, api_key, credential, auth, or bearer are always replaced with [redacted] — at the SDK during capture and again in the Sailfish backend before storage.

What You See in Sailfish

• Every DAG run + task instance appears as a session. Filter by state, group by pipeline, search by task ID, or sort by duration.
• Failed runs link to the affected users via attribution intervals — answer "which customers did last night's failed rollup impact" in one click.
• Warehouse queries appear as a correlated sub-list on each task. Group by SQL fingerprint to find every job that touched a given table or wrote a given shape.
• Retries are linked to their previous-attempt session so you can replay the forensic timeline across attempts.
• Sessions are partitioned per-company and retained for 90 days.

What You See in Airflow

Sailfish ships a plugin that runs inside your Airflow webserver. After install, your Airflow UI gets:

• Sailfish Debugger page in the left nav — the control plane: global on/off, capture mode, per-DAG policy table.
• "View in Sailfish" operator extra link on every task instance page — deep-links straight to that task's captured session.
• DAG-run status badge on each DAG-run row, showing capture state at a glance.
• Grid view column showing the same state across every task instance in a run (Airflow 2.5+).
• Trigger modal extension — choose capture options directly when manually triggering a DAG (Airflow 2.6+).

If a plugin surface ever fails to load on your specific Airflow build, the rest of the plugin keeps working — every surface is failure-isolated so a single missing dependency cannot take down the page.

Controlling Capture per DAG

Capture is controlled by a two-level policy: a company-wide global policy plus optional per-DAG overrides. Both are edited from the Sailfish Debugger page inside Airflow, or from the Sailfish dashboard.

Global capture mode:

Mode	Meaning
Off	No capture at all. The integration installs but every lifecycle hook short-circuits.
On for manually triggered DAGs	Capture only when a human (or external system) triggers a run, not on the schedule.
On only for failed retries	First attempt runs uncaptured; if it fails and Airflow retries, the retry is captured.
On for selected DAGs	Capture only DAGs whose per-DAG policy explicitly opts in.

Per-DAG override:

Policy	Meaning
Use global	Inherit the global capture mode. (Default for every DAG.)
Always on	Always capture this DAG, even when the global mode is Off.
Failed retry only	Like the global mode of the same name, scoped to this DAG.
Always off	Never capture this DAG.
Blocked	Hard-deny. Even an explicit capture-arm request from the Sailfish dashboard or the REST API is refused. Use for PII-sensitive jobs that compliance has flagged as off-limits.

The Blocked policy is enforced both in the Airflow plugin and on the Sailfish backend — there is no path to capture a blocked DAG.

Attributing DAG Runs to Your Users

The most valuable signal Sailfish surfaces from a backend job is who was affected when it broke. A DAG run has no inherent user — you declare attribution from your task code:

from sf_veritas import (
    set_active_users, add_active_user, remove_active_user,
    clear_active_users, active_user, active_users_from,
)

# A task processing a single customer:
def process_customer(customer_id: str):
    with active_user(customer_id):
        ...   # everything inside is attributed to this customer

# A batch processing many customers atomically:
def nightly_rollup(customer_ids: list[str]):
    set_active_users(customer_ids)
    ...

# Derive users from function arguments:
@active_users_from(lambda batch: [row["customer_id"] for row in batch])
def reduce_batch(batch): ...

Attribution is recorded as event-sourced intervals — 14:00–14:05 → {A}, 14:05–14:10 → {A, B}, 14:10–14:15 → {B}. Multiple users can be active at once. Retroactive corrections are supported from the Sailfish dashboard if you discover an attribution error after the fact.

If you'd rather not edit DAG code, set SAILFISH_AIRFLOW_USER_EXTRACTOR=path.to.fn — Sailfish will call that function with the Airflow task context on every task start and use the returned IDs.

Installation

1. Install the SDK

Use the same sf-veritas SDK you'd install for any Python service — there is no separate Airflow extra. Install it in the Python environment your Airflow scheduler and workers run in.

pip install sf-veritas

If you use Poetry, uv, pdm, or Pipenv, see the Python integration guide for the matching invocation.

2. Configure the API key

Set these in your Airflow scheduler, webserver, and worker environments:

SAILFISH_API_KEY=<your key from app.sailfish.ai/settings/configuration>
SAILFISH_GRAPHQL_ENDPOINT=https://api-service.sailfish.ai/graphql/

3. Restart Airflow

Sailfish registers itself through Airflow's airflow.plugins entry point — there is no plugins/ file to edit and no airflow.cfg change required. Restart your scheduler and webserver and the plugin loads automatically.

On startup you'll see one log line confirming the integration tier:

[sf-veritas] Workflow-orchestrator telemetry — active: airflow

Supported Airflow Versions

Version	Integration	Notes
3.0 +	Listener API	Full feature set, including the declarative external-views manifest.
2.5 – 2.x	Listener API	Full feature set.
2.1 – 2.4	Plugin + DAG callbacks	All telemetry; some plugin UI surfaces (Grid column, Trigger modal extension) are unavailable on older webservers.
< 2.1	No-op	Sailfish logs one warning and exits silently. Your scheduler is never blocked.

The integration is built to span this entire support window — adding a newer Airflow major release does not require an SDK upgrade as long as Airflow's listener API remains compatible.

Safety Guarantees

• Hard kill switch. SAILFISH_WORKFLOW_HARD_DISABLE=1 disables the whole workflow-telemetry layer in any process — adapters register but every lifecycle hook is a no-op. Use this to roll back instantly without uninstalling the SDK.
• Plugin-only kill switch. SAILFISH_AIRFLOW_PLUGIN_HARD_DISABLE=1 turns off the in-Airflow UI surfaces while leaving telemetry capture on. Use this if a Sailfish UI surface ever conflicts with a custom Airflow plugin.
• Failure isolation. Every listener callback, plugin route, and UI surface is wrapped in exception handling — a bug in the integration cannot crash your scheduler, worker, or webserver. The worst case is a single missing surface, never a downed scheduler.
• Sampling. SAILFISH_WORKFLOW_SAMPLE_RATE (0.0 – 1.0) caps the share of pipeline runs captured. Use SAILFISH_WORKFLOW_SAMPLE_RATES_JSON for per-DAG overrides, e.g. '{"high_volume_dag": 0.1, "noisy_dag": 0.0}'. Failed and retried runs are always kept regardless of sample rate — error-biased retention is on by default.
• Per-company partitioning. Captured sessions are isolated per company and retained for 90 days; the underlying storage is partitioned so one company's data never lives next to another's.
• No raw warehouse SQL stored by default. Only the normalized SQL fingerprint and warehouse-side query ID — meaning Sailfish can group structurally identical queries across runs without ever storing the literals your queries contain.

Environment Variable Reference

Kill switches

Variable	Default	Effect
SAILFISH_WORKFLOW_HARD_DISABLE	false	Disable the universal workflow-telemetry layer and every orchestrator adapter.
SAILFISH_AIRFLOW_PLUGIN_HARD_DISABLE	false	Disable the in-Airflow plugin UI only. Telemetry still flows.

Sampling

Variable	Default	Effect
SAILFISH_WORKFLOW_SAMPLE_RATE	1.0	Per-pipeline-run sampling rate. Failed/retried runs always kept.
SAILFISH_WORKFLOW_SAMPLE_RATES_JSON	""	Per-DAG sampling overrides as JSON, e.g. '{"big_dag": 0.1}'.
SAILFISH_WORKFLOW_BURST_CAPTURE_LIMIT_PER_MIN	100	Cap on verbose burst-capture invocations per function per minute.

Opt-in capture

Variable	Default	Effect when set
SAILFISH_AIRFLOW_CAPTURE_SQL_VALUES	false	Store raw SQL text alongside the fingerprint.
SAILFISH_AIRFLOW_CAPTURE_XCOM_VALUES	false	Store XCom values, not just keys + sizes.
SAILFISH_AIRFLOW_CAPTURE_DAG_RUN_CONF	false	Store dag_run.conf / task params (size-capped, credential keys still redacted).
SAILFISH_AIRFLOW_CAPTURE_OWNER_EMAILS_RAW	false	Store original owner / alert emails (hashed otherwise).

User attribution

Variable	Default	Effect
SAILFISH_AIRFLOW_USER_EXTRACTOR	""	Dotted path to a callable extractor(context) → list[str] that derives active users from the Airflow task context on every task start.

Connectivity

Variable	Default	Effect
SAILFISH_API_KEY	—	Required. Your Sailfish API key.
SAILFISH_GRAPHQL_ENDPOINT	Sailfish cloud	Override for self-hosted or local development.
SAILFISH_BASE_URL	https://app.sailfishqa.com	Used by the Airflow plugin to construct "View in Sailfish" deep links and to call the policy REST API.

Future Orchestrators

The universal workflow-telemetry layer is platform-neutral. Prefect and Dagster integrations are built on the same data model, the same normalized state vocabulary, and surface in the same Sailfish UI. When they ship, they will appear here without changing what you see for Airflow.

Troubleshooting

I don't see my DAG runs in Sailfish.

1. Confirm sf-veritas is installed in the same Python environment as your scheduler and workers (pip show sf-veritas).
2. Confirm SAILFISH_API_KEY and SAILFISH_GRAPHQL_ENDPOINT are set in the scheduler and worker processes — not only the webserver.
3. Check that SAILFISH_WORKFLOW_HARD_DISABLE is not set.
4. Check the DAG's per-DAG policy isn't Always off or Blocked, and that the global capture mode is compatible (Off means nothing is captured).
5. Look in the scheduler log for the [sf-veritas] Workflow-orchestrator telemetry — active: airflow line. If it says skipped: airflow, the SDK didn't recognize the process as Airflow — check the install path.

The "Sailfish Debugger" page doesn't appear in Airflow's nav.

• On Airflow 2.1 – 2.4 the plugin UI is reduced; some surfaces are unavailable on older webservers.
• On Airflow 2.5+, check SAILFISH_AIRFLOW_PLUGIN_HARD_DISABLE is not set and look at the webserver log for the [sf-veritas] startup line.

A specific DAG must never be captured.

Set its per-DAG policy to Blocked. This refuses every capture path — the in-Airflow plugin, the Sailfish dashboard arm button, and direct REST API calls. The block is enforced both in your Airflow process and on the Sailfish backend.

The capture is too verbose / generating too many sessions.

Lower SAILFISH_WORKFLOW_SAMPLE_RATE globally, or set a per-DAG override in SAILFISH_WORKFLOW_SAMPLE_RATES_JSON for the noisiest pipelines. Failed and retried runs will still be captured at full fidelity.