dpone-airflow-pack 0.66.8

Lightweight Airflow scheduler-side pack provider for dpone GitOps workloads

dpone-airflow-pack

dpone-airflow-pack is the lightweight Airflow scheduler/webserver provider for dpone GitOps packs.

It only reads a static airflow-pack.json and builds visible Airflow/Kubernetes tasks. It does not import the full dpone runtime and intentionally contains no source/sink/native transfer dependencies such as ClickHouse, MSSQL, pyodbc, pandas, polars, or ConnectorX.

Recommended Airflow DAG import:

from dpone_airflow_pack import build_dpone_gitops_task_group_from_pack

The full dpone[full,accel] package belongs in the KPO runtime image, not in the scheduler image.

Remote pack cache

Remote airflow-pack.json artifacts can be published to object storage and mirrored into a bounded local scheduler cache.

Bootstrap one generation:

dpone-airflow-pack-sync \
  --once \
  --index-uri s3://do-dwh/dpone-artifacts/prod/airflow-dags-dev/latest/pack-index.json \
  --reader-connection-id s3_dpone_artifacts_reader \
  --cache-dir /opt/airflow/dags/.dpone-cache/airflow \
  --status-path /opt/airflow/dags/.dpone-cache/airflow/status/last-sync-status.json \
  --airflow-variable-key dpone_airflow_pack_cache_status

Refresh without restarting Airflow pods:

dpone-airflow-pack-sync \
  --watch \
  --interval-seconds 60 \
  --jitter-seconds 15 \
  --index-uri s3://do-dwh/dpone-artifacts/prod/airflow-dags-dev/latest/pack-index.json \
  --reader-connection-id s3_dpone_artifacts_reader \
  --cache-dir /opt/airflow/dags/.dpone-cache/airflow \
  --status-path /opt/airflow/dags/.dpone-cache/airflow/status/last-sync-status.json \
  --airflow-variable-key dpone_airflow_pack_cache_status

The sync loop is fail-open. On S3, connection, index, or hash errors it writes a warning status and preserves the previous current generation. DAG parse stays local and cache-only.

When --airflow-variable-key is set, the sync process also publishes the same redacted status payload into Airflow metadata. This is the recommended diagnostic channel for KubernetesExecutor deployments because runtime worker pods do not need to mount the scheduler or dagProcessor cache volume. Variable publish tries the Airflow model first and then a direct metadata DB upsert for sidecar/init contexts where Airflow 3 task SDK variables are not available. Publish failures are warnings only and never fail the Airflow pod.

Connection resolution is scheduler-side safe for Airflow 2.x and 3.x: environment connection URI first, Airflow hook second, and metadata database fallback third. The metadata fallback matters for Airflow 3 sidecar/init containers, where provider hooks may not resolve DB-backed connections outside task execution context even though the metadata row exists.

GitOps domain groups

For DAGs that should run a whole GitOps domain or a named workflow group, keep the grouping in the domain catalog and resolve it through the lightweight provider:

workflow_groups:
  daily:
    workload_ids:
      - dim_customer
      - fact_order

from dpone_airflow_pack import workload_ids_from_gitops_domain

workload_ids = workload_ids_from_gitops_domain(
    repo_root,
    "sales",
    group="daily",
)

The resolver only reads YAML files and validates that every group member exists in the same domain catalog.

Inspect the current cache status:

dpone-airflow-pack-cache-status \
  --cache-dir /opt/airflow/dags/.dpone-cache/airflow

Airflow execution policy

airflow-pack.json can carry a scheduler-side execution policy:

{
  "airflow": {
    "execution": {
      "task_executor": "KubernetesExecutor",
      "deferrable": true,
      "on_finish_action": "delete_succeeded_pod",
      "get_logs": true,
      "logging_interval_seconds": 60
    }
  }
}

The lightweight provider maps it to KubernetesPodOperator kwargs (executor, deferrable, on_finish_action, get_logs, and logging_interval) and treats those fields as pack-owned. In deferrable mode, logging_interval_seconds controls how often the task resumes to fetch fresh pod logs. DAG-local overrides may still set DAG-owned values such as retries, pool, or in_cluster.