# orca.scheduler > A module of Orca's API. The scheduler: fires schedules, senses datasets, queues backfills and rebuilds, and starts runs. ``` orca register ./build/taxi runs `taxi describe`, records the binary orca backfill taxi 2011-01..2024-12 a run per partition that isn't done orca backfill labels @all keyed: runs of keys_per_run undone keys orca rebuild taxi.trips 2024-01 the pipelines downstream, in order orca tick one pass, then exits orca scheduler a pass every second ``` The scheduler links no pipeline. It knows one by the `describe` it printed, kept in `orca.pipelines` with the binary's path, and reads the schedule, partitions and limits from that JSON in SQL. Everything it decides is in Postgres, so it can be stopped and restarted, or run twice, at any time: - **A fire is a row.** `orca.fires` is keyed by (pipeline, fire time), and a fire's row and its `queued` run are inserted in one transaction, so a fire is never made twice. Fires are counted from the last one recorded (or from when the pipeline was registered), so the fires missed while the scheduler was down are made once each, or, with `catchup=False`, only the latest. A fire whose partition is before the pipeline's first is recorded without a run. - **Starting a run is `resume`.** A queued run has nothing done, or what a resume or rerun from `orca ui` left it, which is what ` resume ` takes (with `--allow-new-code` when the resume said so). The scheduler marks the run `running`, with no heartbeat and itself as owner, and execs the binary, which claims it. It doesn't wait: the run records its own progress. - **`max_runs`.** Under a lock on the pipeline's row, the runs that are live (a fresh heartbeat, or started by the scheduler less than the stale limit ago) are counted, and only that many fewer than `max_runs` are started. - **Backfills queue what isn't done.** A key is done when every mapped step has succeeded for it in a run of the pipeline's current code (`DONE_SQL`). A backfill drops the keys that are done or in a queued or running run, sorts the rest and queues a run per `keys_per_run` of them. - **Where runs run.** `ORCA_RUNNER` is `local` (a process here, the default), `kubernetes` (a dista cluster of pods per run, `orca.kube`) or `ec2` (a head instance and worker instances per run, `orca.ec2`). - **Sensors and rebuilds** (`orca.sensors`). A sensor pipeline's pass looks for snapshots of the datasets it reads, in `orca.commits` on every tick and from ` sense` every `ORCA_SENSE_SECONDS`, and queues a run, trigger `sensor`, for those it hasn't seen. A keyed sensor's keys dataset is sensed too: a new snapshot of it queues the keys that aren't done, in runs of `keys_per_run`. Each rebuild queues its next level once the one before has succeeded. - **Retries.** A scheduled, backfilled or sensed run that failed is started again after `ORCA_RETRY_SECONDS` (default 60) doubling with each start, plus up to a tenth more as jitter, up to `run_retries` times. One whose process died (its heartbeat went stale) is started again at once, within the same budget. A manual run is never retried by the scheduler. A run asked to cancel whose process died ends `cancelled`. The scheduler's clock is `orca.civil.now`, which `ORCA_NOW` fixes; fire times and retry times are on it. Heartbeats are on the database's clock. A run's output goes to `///driver.log` when `ORCA_LOGS` is set: written there by a local run, published there when the process is reaped if `ORCA_LOGS` is in object storage, or kept from the driver pod's log on Kubernetes (`orca.kube`). ## Structs ### `Registered` ```mojo struct Registered ``` A registered pipeline, as the scheduler reads it from `describe`. Implements: `Copyable`, `Movable` Fields: - `name: String` - `binary: String` - `registry_hash: String` - `since: Int64` - `cron: String` - `tz: String` - `fills: String` - `catchup: Bool` - `kind: String` - `start: String` - `max_runs: Int` - `run_retries: Int` - `sensor: Bool` - `keys_per_run: Int` - `keys_dataset: String`: A keyed pipeline's keys dataset; empty for time partitions. - `image: String`: The image its runs' pods run (`orca register --image`), or empty. - `workers: Int`: Its cluster's dista workers; 0 if it has no tasks. - `head_type: String`: On EC2, its runs' head instance type; empty for the default. - `worker_type: String` - `spot: Bool`: On EC2, whether its workers are Spot instances. - `min_workers: Int`: On EC2, its runs' fewest workers, scaling up to `workers`; equal to `workers` for a fixed count. - `secrets: List[Tuple[String, String]]`: Its runs' secrets: environment variable and Secrets Manager ID. ### `Scheduler` ```mojo struct Scheduler ``` Implements: `Movable` Fields: - `db: State` - `stale: Float64` - `retry_seconds: Int64` - `sense_seconds: Float64` - `start_grace: Float64`: How long a run the scheduler started may go without a heartbeat. - `budget: Int`: `ORCA_VCPU_BUDGET` in thousandths of a vCPU; 0 for none. - `runner: Optional[KubeRunner]`: Set with `ORCA_RUNNER=kubernetes`. - `ec2: Optional[Ec2Runner]`: Set with `ORCA_RUNNER=ec2`; with neither, runs are local processes. - `logs: String` #### `__init__` ```mojo def __init__(out self, url: String) ``` #### `register` ```mojo def register(mut self, binary: String, now: Int64, image: String = "", head_type: String = "", worker_type: String = "", workers: Int = Int(0), spot: Bool = False, min_workers: Int = Int(-1), secrets: List[String] = List()) -> String ``` Records the pipeline `binary` describes, with its path; its schedule counts from `now` the first time, and a sensor from now the first time it's one, with the snapshots its datasets have now seen. Queued runs move to the new code. `min_workers`, with `workers` as the most, makes an EC2 run's workers scale between them; -1 keeps a fixed count. `secrets`, `NAME=ID`s, replace the pipeline's when given. Returns the pipeline's name. #### `tick` ```mojo def tick(mut self, now: Int64) ``` One pass: reap ended runs, advance rebuilds, make due fires, sense new snapshots, schedule retries, and start what may start. #### `backfill` ```mojo def backfill(mut self, pipeline: String, keys: String, rerun: Bool = False) -> Int ``` Queues a run for each partition in `keys` that has none; see `backfill`. ## Functions ### `fill_key` ```mojo def fill_key(kind: String, fills: String, local: Civil) -> String ``` The partition a fire fills, given its time on the schedule's wall clock: the key the time falls in, or the one before it. ### `registered` ```mojo def registered(mut db: State, only: String = "") -> List[Registered] ``` Every pipeline registered with a binary, or the one named. ### `dataset_cycle` ```mojo def dataset_cycle(mut db: State, describe: String) -> String ``` How the pipeline `describe` gives would close a cycle across pipelines, through the datasets they write and read, or empty if it wouldn't. A pipeline reading a dataset it writes itself isn't one. ### `current_keys` ```mojo def current_keys(p: Registered) -> Tuple[Int64, List[String]] ``` A keyed pipeline's keys snapshot and keys, as ` keys` prints them. ### `kind_mismatch` ```mojo def kind_mismatch(mut db: State, describe: String) -> String ``` How the pipeline `describe` gives would make a sensor run keys of the other kind, a keyed sensor reading a dataset a time-partitioned pipeline writes or the other way round, or empty if it wouldn't. A keyed sensor's own keys dataset needs no mapping, whoever writes it: a new snapshot of it offers the keys it holds, which are names already. So a dated pipeline may write the keys a keyed sensor runs. ### `partition_keys` ```mojo def partition_keys(p: Registered, keys: String) -> List[String] ``` `keys` as the pipeline's partition keys, each checked: for time keys `KEY[..KEY][,...]` from its first partition on; for keyed partitions a list of names, `@all` or `@file:`, each a current key. ### `done_keys` ```mojo def done_keys(mut db: State, pipeline: String, keys: List[String]) -> Dict[String, Bool] ``` Which of `keys` are done for `pipeline` (`DONE_SQL`). ### `queue_keys` ```mojo def queue_keys(mut db: State, p: Registered, keys: List[String], trigger: String, rerun: Bool = False) -> List[Int] ``` Queues runs of `p` over those of `keys` that aren't done and aren't in a live run, sorted, `keys_per_run` to a run; returns the runs. With `rerun`, keys that are done are queued too, but never one in a live run. Under a lock on the pipeline's row, so two backfills don't both queue a key. ### `backfill` ```mojo def backfill(mut db: State, p: Registered, keys: String, rerun: Bool = False) -> Int ``` Queues runs for the keys in `keys` that aren't done and aren't in a queued or running run, `keys_per_run` to a run; returns how many runs it queued. With `rerun` (`orca backfill --rerun`) the done ones are queued again too, as for a day whose source has changed since. ## Aliases ### `DONE_SQL` ```mojo comptime DONE_SQL = "WITH k AS (SELECT DISTINCT unnest(string_to_array($2, ',')) AS key), p AS (SELECT registry_hash, describe FROM orca.pipelines WHERE name = $1), m AS (SELECT x->>'name' AS step FROM p, jsonb_array_elements(p.describe->'steps') AS x WHERE x->>'kind' = 'mapped') SELECT k.key FROM k WHERE CASE WHEN EXISTS (SELECT 1 FROM m) THEN NOT EXISTS (SELECT 1 FROM m WHERE NOT EXISTS (SELECT 1 FROM orca.steps st JOIN orca.runs r ON r.id = st.run JOIN p ON r.registry_hash = p.registry_hash WHERE r.pipeline = $1 AND st.step = m.step AND st.partition = k.key AND st.status = 'succeeded')) ELSE EXISTS (SELECT 1 FROM orca.runs r JOIN p ON r.registry_hash = p.registry_hash WHERE r.pipeline = $1 AND r.status = 'succeeded' AND k.key = ANY (string_to_array(r.partitions, ','))) END" ``` Of the comma-separated keys `$2`, those done for pipeline `$1`: every mapped step has succeeded for the key in a run of the pipeline's current code. A pipeline with no mapped step has a key done when a run of its current code over it succeeded. ### `LIVE_SQL` ```mojo comptime LIVE_SQL = "SELECT DISTINCT unnest(string_to_array(partitions, ',')) AS key FROM orca.runs WHERE pipeline = $1 AND status IN ('queued', 'running')" ``` The keys of pipeline `$1` in a queued or running run.