# orca.ec2 > A module of morca's API. Runs on EC2: a head instance and worker instances per run, started and ended by the scheduler (DESIGN.md, phase 5, PR 5). With `ORCA_RUNNER=ec2` the scheduler starts a run's start `s` as: - **a head instance** (`--head-type` at registration, else `ORCA_EC2_HEAD_TYPE`, on-demand): the dista head, for a pipeline with tasks, and the driver, ` resume `, beside it on localhost; - **`workers` worker instances** (`--worker-type`, else `ORCA_EC2_WORKER_TYPE`; Spot with `--spot`), for a pipeline with tasks, started straight after with `DISTA_HEAD` at the head's private address. Each is tagged `orca.run`, `orca.start`, `orca.pipeline` and `orca.role`, told to terminate when it shuts down, and started from user data that runs the pipeline's image (`ORCA_EC2_MODE=docker`, the default: `docker run --network host`, after `ORCA_EC2_PREAMBLE`, such as an ECR login) or the binary itself (`direct`, for the tests' stand-in EC2), then shuts down (`ORCA_EC2_SHUTDOWN`). The head keeps the driver's output in the run's `driver.log` (` keep-log`) before it does, and if the driver crashed, fails the attempts it left running with the crash's first frames (`orca.crash`). Beside its work each instance runs ` sample` (`orca.sample`; in docker mode as a container with `--pid host`), whose lines its script uploads to the run's `metrics/-.jsonl` with its log (` put-log`), every `ORCA_CLUSTER_LOG_SECONDS` and once when it ends, after telling the sampler to write a last line and stop (`--until`). With `--worker-procs N` (`auto`: one per vCPU of the worker type) each worker instance's script starts `N` dista worker processes against the same head, rather than one, since a process runs one task at a time: each takes a port of its own, so they share the host. The script notes one that exits, forwards a SIGTERM to all of them, and shuts the instance down once the last has gone. Counts and costs stay in instances (`workers`, the budget); a worker's log is its processes' logs joined. A pipeline registered with `--workers MIN..MAX` scales: its run starts `MIN` workers, which may be none, and every tick `scale` reads the head's demand (`dista status --json`, from port 6390 on the head's private address, through the pipeline's binary: ` dista-status`, since dista's sockets can't share orca's binary with flare's) and asks `dista.scale.decide` what to do. It starts the workers asked for, up to `MAX`, and drains the ones named with `dista drain`; a drained worker exits once its work and results have moved, and its instance's script shuts it down. `ORCA_SCALE_UP_AFTER` (5 s), `ORCA_SCALE_IDLE` (60 s) and `ORCA_SCALE_STEP` (4) tune the policy. The budget counts `MAX`. A run is scaled, and its workers are started, as its pipeline was registered when the run started (`orca.runs.spec`, kept at each start), so registering the pipeline again meanwhile doesn't strand a live run on settings it never had; its next start takes the new registration. A run that needs workers (`MAX` above 0) whose head has tasks waiting and no worker for `ORCA_NO_WORKERS_SECONDS` (600; 0 turns it off) has the attempt that waits fail with `no workers joined in 10 min (tasks waiting: N)`, and its instances terminated. Its driver stops beating with them, so the scheduler starts it again as it does a run whose process died, until its retries are spent. Since when it has waited is `orca.runs.waiting_since`, which the run's page shows. `timeout_seconds` on a step bounds an attempt whatever it waits for. A pipeline's secrets (`--secret NAME=ID`) are read by each instance's script at start, after the preamble, with `aws secretsmanager get-secret-value` (`ORCA_SECRET_FETCH` replaces the command, which is given the ID), exported as NAME, given to the containers, and named in `ORCA_SECRETS` so that every log orca writes masks them. Every tick, `sweep` terminates the instances of runs that aren't running at that start, and those older than `ORCA_RUN_DEADLINE` (a day). A head whose run has just ended is left `ORCA_EC2_HEAD_GRACE` seconds (60) to finish its script: keep the driver's log, upload its own and its samples, and shut itself down. Instances get `ORCA_EC2_DATABASE_URL` as `ORCA_DATABASE_URL` (the control instance's private address), the warehouse, the logs, the region and the scheduler's variables named in `ORCA_RUN_ENV`; `ORCA_EC2_AMI`, `ORCA_EC2_SUBNET`, `ORCA_EC2_SECURITY_GROUP` and `ORCA_EC2_PROFILE` say where and as what they run. `cpu_millis` is a run's instances' vCPUs, which the scheduler checks against `ORCA_VCPU_BUDGET` first. **The warm pool** (`ORCA_EC2_POOL_SECONDS`, 150; 0 turns it off). An instance's script is then a loop over runs, and the next run of the same pipeline, started as the same instances, takes its head and workers rather than launching new ones: - **The key.** Each instance's row has a digest of what it was started as (`pool_key`): role, type, market, AMI, subnet, security group, profile, the pipeline's registry hash, and its script but for the first run's line, which holds the image, binary, secrets' IDs, the environment it was given and its worker processes. A run takes only instances of its own keys, so a new registration, image or setting never reuses old ones. - **Back to the pool.** A head's driver exits; a worker's ` pool-wait` (`orca.pool`) sees its run end and the script kills its dista processes, which don't end with their head. Each uploads that run's logs and samples and runs ` pool-next`, which marks the run's row `idle` and beats `polled` every second until it's handed a run. Its run's row stays live meanwhile, so idle instances count against the budget (a run of the same keys counts them as its own), and `/ops/pools` shows them as `idle (pooled)`. A start of a pipeline waits while instances of its keys are still on their way there (at most the head's grace). - **Hand-over** (`_take`). In one transaction, the idle rows of the key whose run has ended, that weren't drained and that beat within `ORCA_EC2_POOL_LIVE_SECONDS` (5) are locked `FOR UPDATE SKIP LOCKED`, checked with `DescribeInstances`, ended (`pooled: on to run N`) and replaced by rows of the new run (`pooled`); `pool-next` sees its new row, records that it `joined`, and the next pass runs it: a new dista head process for the head, new worker processes pointed at the run's head, secrets read again, and logs and samples under the new run. A run takes pooled workers up to what it starts (`MIN`, and what `scale` adds), then launches the rest; the others stay pooled. - **Failure.** One handed a run that hasn't joined within `ORCA_EC2_POOL_JOIN_SECONDS` (30) is terminated (`pooled: never took the run up`) and replaced: a worker by a new worker, a head by new instances for the same start. The sweep ends idle instances after `ORCA_EC2_POOL_SECONDS` (`pooled: idle N s, shut down`), those that stop beating, and those that don't reach the pool within the head's grace; their rows end before they're terminated, so one handed on meanwhile isn't. An instance's tags keep its first run; the sweep goes by its live row. A run that's restarted, or a worker that was drained, ends as before. ## Structs ### `Ec2Spec` ```mojo struct Ec2Spec ``` One start of a run, as the instances it needs. Implements: `Copyable`, `Movable` Fields: - `run: Int` - `start: Int` - `pipeline: String` - `image: String` - `binary: String` - `tasks: Bool`: Whether the pipeline calls dista tasks: a head and workers. - `head_type: String` - `worker_type: String` - `workers: Int` - `spot: Bool` - `allow_new_code: Bool` - `min_workers: Int`: The workers it starts with; fewer than `workers`, it scales. - `secrets: List[Tuple[String, String]]`: Environment variable and Secrets Manager ID of each secret. - `worker_procs: Int`: The dista worker processes on each worker instance; 0 is one per vCPU of the worker type. - `registry_hash: String`: The pipeline's code, which an instance from the pool must have been started with. - `env: List[Tuple[String, String]]`: The pipeline's own variables (orca.toml's `env`), exported on every instance over those `ORCA_RUN_ENV` names, and so part of the pool's key. ### `Ec2Runner` ```mojo struct Ec2Runner ``` Implements: `Movable` Fields: - `ec2: Ec2` - `mode: String` - `head_type: String` - `worker_type: String` - `deadline: Int` - `grace: Float64`: How long a head may go on after its run ended, to shut itself down, before the sweep terminates it. - `env: List[String]` - `up_after: Float64` - `idle: Float64` - `step: Int` - `short_since: Dict[String, Float64]`: For each scaling run, `run-start`, since when it's been short of workers: `decide`'s memory between ticks. - `no_workers: Int`: `ORCA_NO_WORKERS_SECONDS` (600): how long a run whose head has tasks waiting and no worker may go on before its attempt is failed; 0 turns it off. - `pool_seconds: Float64`: `ORCA_EC2_POOL_SECONDS` (150): how long an instance whose run ended waits, idle, to be handed another; 0 turns the pool off, and every instance ends with its run. - `pool_live: Float64`: `ORCA_EC2_POOL_LIVE_SECONDS` (5): how recently an idle instance must have asked for a run (`pool-next`) to be handed one. - `pool_join: Float64`: `ORCA_EC2_POOL_JOIN_SECONDS` (30): how long an instance handed a run has to take it up before it's terminated and replaced by a new one. #### `__init__` ```mojo def __init__(out self) ``` #### `head_of` ```mojo def head_of(self, s: String) -> String ``` #### `worker_of` ```mojo def worker_of(self, s: String) -> String ``` #### `procs_of` ```mojo def procs_of(self, r: Ec2Spec) -> Int ``` The dista worker processes on each of the run's worker instances: `worker_procs`, or one per vCPU for `auto` (0). #### `cpu_of` ```mojo def cpu_of(self, tasks: Bool, head_type: String, worker_type: String, workers: Int) -> Int ``` A run's instances' vCPUs, in thousandths. #### `pool_key` ```mojo def pool_key(self, r: Ec2Spec, role: StringSpan[ImmStaticOrigin], itype: String, spot: Bool) -> String ``` What an instance of `r`'s in `role` is started as, as a digest: its type, market, image, network and profile, the pipeline's code, and its script but for the first run's line (the image, binary, secrets' IDs, environment, worker processes and the rest). An idle instance is handed only to a run whose key is its own, so a new registration, image or setting never reuses one; empty with the pool off. #### `pool_keys` ```mojo def pool_keys(self, r: Ec2Spec) -> List[String] ``` The keys of the head and, for a pipeline with tasks, the workers of `r`'s runs; none with the pool off. #### `start` ```mojo def start(self, mut db: State, r: Ec2Spec, fresh: Bool = False) ``` Starts the run's head instance, then its workers, and records them in `orca.instances`: from the pool first, where an idle instance of the same key is (unless `fresh`), else new ones. What `r` is, the pipeline's registration now, is kept with the run, so its workers are started as its head was however the pipeline is registered later. #### `settling` ```mojo def settling(self, mut db: State, keys: List[String]) -> Bool ``` Whether instances of `keys` whose run just ended are still on their way to the pool (finishing the run's logs, stopping its processes): a start waits for them, a few seconds, rather than launching new ones beside them. At most the head's grace: after that the sweep terminates them. #### `scale` ```mojo def scale(mut self, mut db: State) -> Tuple[Int, Int] ``` Grows and shrinks the workers of every running run that scales, from its head's demand; returns the workers started and drained. #### `policy` ```mojo def policy(self, lo: Int, hi: Int, per: Int = Int(1)) -> Policy ``` The scaling policy over worker processes: `lo` to `hi` of them, and `ORCA_SCALE_STEP` instances of `per` at most at once. #### `sweep` ```mojo def sweep(self, mut db: State) -> Int ``` Terminates the instances of runs that aren't running at their start, and those older than the deadline, but for the pool's: an instance whose run ended goes to the pool, and is terminated once it has been idle `ORCA_EC2_POOL_SECONDS`, or has stopped asking for a run. Returns how many it terminated. ## Functions ### `vcpus` ```mojo def vcpus(instance_type: String) -> Int ``` An instance type's vCPUs: from `ORCA_EC2_VCPUS` (`type=n,...`), else by its size, as Graviton families are sized. ## Aliases ### `HEAD_PORT` ```mojo comptime HEAD_PORT = 6390 ``` ### `HEAD_SECONDS` ```mojo comptime HEAD_SECONDS = 10 ``` How long ` dista-status` or `dista-drain` may take, asking a run's head, before it's killed and the run skipped this tick.