`orca.scheduler`

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 <binary> resume <run> 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 <binary> 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 <ORCA_LOGS>/<pipeline>/<run>/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`

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`

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`

def __init__(out self, url: String)

`register`

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=IDs, replace the pipeline's when given. Returns the pipeline's name.

`tick`

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`

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`

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`

def registered(mut db: State, only: String = "") -> List[Registered]

Every pipeline registered with a binary, or the one named.

`dataset_cycle`

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`

def current_keys(p: Registered) -> Tuple[Int64, List[String]]

A keyed pipeline's keys snapshot and keys, as <binary> keys prints them.

`kind_mismatch`

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`

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:<path>, each a current key.

`done_keys`

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`

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`

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

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.

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.