# Repo Airflow Map — ground truth

How Apache Airflow is actually set up and used in **this monorepo's data-pipeline stack**.
This is the factual backbone every lesson is anchored to. All paths are repo-relative;
`file:line` refs are from this branch.

> Verified by a code scan on 2026-07-11. ⚠️ marks corrections to common assumptions (and to
> the `airflow-job` skill). If a lesson and the code disagree, trust the code.
> **Airflow is NOT in the learner's three services** (conversationmgmt/notification/spike) —
> it's the data pipeline (codename **hephaestus** + Spark), sitting downstream of them.

---

## Corrections up front

1. ⚠️ **It's Airflow 3.x**, self-hosted on GKE via the official Helm chart — **not** the
   stale `appVersion: 1.16.0` in a wrapper `Chart.yaml`. Prod/stag run `3.2.1`, local `3.1.2`.
2. ⚠️ **DAGs are generated from `airflow/dags/rules.yaml`** by `generate.py` + Jinja2
   templates — **not** hand-written `.py` per DAG.
3. ⚠️ **No sensors anywhere** in `airflow/`. The skill's `SparkKubernetesSensor` is
   illustrative only. (Sensors are still on the certification syllabus — learn the concept.)
4. ⚠️ **`./local/run.bash cron` is a separate Docker-Compose cron runner, unrelated to
   Airflow.** The Airflow local path is the `make` targets below.

---

## Distribution, version & architecture

- **Self-hosted Apache Airflow on Kubernetes (GKE)** via the official `apache/airflow` Helm
  chart (v1.18.0), wrapped in `deployments/helm/data-warehouse/custom-airflow/`.
- **Airflow 3.x**: `airflowVersion: "3.2.1"` (stag/dorp/prod values), `3.1.2` local. Custom
  image `.../manaverse/airflow:3.2.1-rev2`.
- **Executor**: `KubernetesExecutor,LocalExecutor` (`stag-manabie-values.yaml:17`).
- **Airflow-3 components present**: `apiServer` (the AF3 replacement for the webserver),
  `scheduler`, `dagProcessor`, `triggerer`, `workers`.
- **Metadata DB**: bundled PostgreSQL 16. **Auth**: GitHub OAuth (FAB security manager),
  GitHub-team → role mapping. **Ingress**: Istio VirtualService.

## Where DAGs live — dynamic generation

DAGs are generated, not authored:
```
airflow/dags/rules.yaml   ── registry (685 lines): global + a `dags:` list
        │  generate.py (+ generate_lmsdwh.py) loads it, renders Jinja2 templates
        ▼  airflow/dags/templates/*.py  (~28 templates)
airflow/dags/{org}/{env}/{env}-{org}-{name}-{tenant}.py   +  one .yaml K8s manifest per job
```
- Registry entry (`rules.yaml:19`): `name`, `template`, `schedule`, `image_tag`, `tags`,
  `failed_mentions`, `jobs:`, `deployments:`.
- `generate.py:152-260` iterates dags → deployments → tenants → jobs; shells out to
  `generate_<job_type>.sh` to render the manifest, then renders the DAG `.py`.
- Validations in the generator: `failed_mentions` required (`:162`, raises);
  SparkApplication name ≤ 40 chars (`:18-65`).
- Output is committed and must match `rules.yaml`/templates (drift check — see Testing).

## Operators & task types (counted across templates)

| Operator | Uses | Job type in `rules.yaml` |
|---|---|---|
| `SparkKubernetesOperator` | 60 | `spark_app` (submits a `SparkApplication` CRD) |
| `KubernetesPodOperator` | 10 | `python_pod`, `scheduling_job`, `camel_job` |
| `KubernetesJobOperator` | — | `python_job` (+ `KubernetesDeleteJobOperator`) |
| `PythonOperator` | 8 | in-DAG Python (e.g. read Variables → XCom) |
| `BashOperator` | 3 | small shell steps |
| sensors | **0** | — (⚠️ none used here) |

Representative task graph (`templates/sync-learnosity-data-with-val.py:102`):
`camel_job >> lrn_questions >> spark_app >> spark_val`. Spark tasks:
`application_file="{{env}}-{{org}}-{{spark_app}}-{{tenant}}.yaml"`, `do_xcom_push=False`,
`delete_on_termination=False`.

## Scheduling

- Cron strings live in `rules.yaml` → template var `{{schedule}}` → `schedule="..."` (AF3).
  Examples: `"00 18 * * *"`, `"0 * * * *"`; `schedule: None` = manual-only.
- Default when omitted: `"0 0 * * *"` (`generate.py:154`).
- `catchup=False` in **every** template's `default_args` (`common-spark-job.py:24`) → no
  backfill by default; `start_date = datetime(2024, 1, 1)` hard-coded (`:15`);
  `dagrun_timeout=90m`; `max_active_runs`/`retries` from `rules.yaml`.
- Per-tenant schedule override: `generate.py:222` (`<tenant>_schedule`).

## The data-pipeline flow (hephaestus + Spark)

```
app services' Postgres (conversationmgmt/notification/spike/…)
   │  Debezium / logical replication  (migrations/lmsdwh/1001: CREATE PUBLICATION lmsdwh_publication)
   ▼
Kafka Connect  (hephaestus upserts connectors — cmd/server/hephaestus/upsert_kafka_connector_lmsdwh.go)
   ▼
lmsdwh warehouse (Postgres)  (hephaestus_migrate_lmsdwh; migrations/lmsdwh/ = 77 files)
   ▼
Airflow-orchestrated Spark transforms  (dim_*/fact_* tables: airflow/spark/src/libs/lmsdwh/tables/)
```
**hephaestus** (`internal/hephaestus`, `cmd/server/hephaestus/`) is the CDC/connector
plumbing — a set of bootstrap jobs, not a gRPC app. The three app services are decoupled
*upstream* sources (via Kafka/CDC); they have no Airflow dependency.

## Spark integration

- Spark jobs run as **`SparkApplication` CRDs on the Kubeflow Spark Operator** (not
  `spark-submit` from Airflow). Template: `deployments/helm/platforms/spark/templates/sparkapp.yaml`
  (`kind: SparkApplication`, `apiVersion: sparkoperator.k8s.io/v1beta2`).
- Per-app sidecar flags: `enable_xcom` (adds the `/airflow/xcom` sidecar), `enable_sql_proxy`
  (cloud-sql-proxy), `enable_airflow_params` (mounts params configmap at `/airflow-params/`).
- **PySpark**; source `airflow/spark/src/` (shared code packaged as `libs.zip`); rendered
  `sparkVersion: 3.5.1`. App configs in `deployments/helm/platforms/spark/*-values.yaml`.

## XCom & dynamic params

- **Spark → Airflow**: PySpark writes `{"metadata": …}` to `/airflow/xcom/return.json`
  (`airflow/spark/src/libs/xcom.py:3-18`); the `enable_xcom` sidecar surfaces it;
  `do_xcom_push=True` on the operator picks it up.
- **Airflow Variables → XCom → downstream**: a `PythonOperator` does `Variable.get(...)` +
  `xcom_push(...)`, then `get_vars >> spark_app` (`spark-job-with-airflow-variables.py:44-76`).
- **XCom-pull into a downstream manifest**: `{{ ti.xcom_pull(...) }}` rendered as a later
  task's `application_file` (`create-configmap_v2.py`).
- **Dynamic params**: `params={{ airflow_params }}` on the DAG; passed to pods via
  `env_vars={"AIRFLOW_PARAMS": "{{ params | tojson }}"}`; declared in `rules.yaml` as
  `airflow_params: '{"key": Param(default, type="integer")}'`.

## Config, access-control & alerting

- **Connections**: `connections.yaml` (k8s Secret `airflow-connections`) mounted at
  `/opt/airflow/connections.yaml`; referenced by id (e.g. `slack-webhook-staging`). K8s
  cluster conn per env in `rules.yaml:2-6` → `{{k8s_conn_id}}`.
- **Variables**: JSON configmap `env.json` mounted; read via `Variable.get(...)`.
- **Secrets**: SOPS-encrypted; GCP Workload Identity for pods.
- **Access-control (FAB RBAC)**: `access_control:` in `rules.yaml` (`:57`, role →
  `[can_edit, can_read, …]`) → template `access_control={{ access_control }}`. Roles seeded
  from `roles.json`.
- **Slack alerting**: `templates/utils/alert_utils.py` — `slack_alert()`/`slack_success()`
  via `SlackWebhookNotifier`; DAGs set `on_failure_callback=[slack_alert({{failed_mentions}})]`.
  `failed_mentions` is **required** per DAG.

## Deployment & local run

- Generated DAGs are **synced to a GCS bucket** that Airflow mounts read-only via **gcsfuse**
  (gitSync disabled). `AIRFLOW__CORE__DAGS_FOLDER=/opt/airflow/dags_gcs`.
- **Generate**: `make generate-dags` (`AIRFLOW_ENV=stag|dorp|prod`). **CI**:
  `.github/workflows/deployment.airflow.dags.yaml` regenerates + rsyncs to
  `gs://<env>-…/dags/` on push to `develop` touching `airflow/dags/**`.
- **Local**: `make start-lmsdwh-dev` → `make sync-dags` (kubectl cp into pods) →
  `make forward-airflow` (port-forward `svc/airflow-api-server` to `localhost:8080`).

## Testing / validation

- ⚠️ **DAG generation drift check is the main "lint"**:
  `.github/scripts/ensure-make-generate-dags.bash` runs `make generate-dags` and fails if the
  tree changed — committed DAGs must match `rules.yaml` + templates (run in the pre-merge
  pipeline).
- **Semantic validation** inside the generator (`failed_mentions` required, SparkApp name
  length) and the Helm render (`sparkapp.yaml` `fail`).
- **hephaestus** has Go + BDD tests for the CDC/connector layer. **No PySpark pytest suite**;
  data validation is done as in-pipeline Spark `*-validation` tasks + Great Expectations.