orca.ec2

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, <binary> resume <run>, 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 (<binary> 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 <binary> sample (orca.sample; in docker mode as a container with --pid host), whose lines its script uploads to the run's metrics/<role>-<instance>.jsonl with its log (<binary> 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: <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 <binary> 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 <binary> 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.

struct Ec2Spec

struct Ec2Runner

def __init__(out self)

def head_of(self, s: String) -> String

def worker_of(self, s: String) -> String

def procs_of(self, r: Ec2Spec) -> Int

def cpu_of(self, tasks: Bool, head_type: String, worker_type: String, workers: Int) -> Int

def pool_key(self, r: Ec2Spec, role: StringSpan[ImmStaticOrigin], itype: String, spot: Bool) -> String

def pool_keys(self, r: Ec2Spec) -> List[String]

def start(self, mut db: State, r: Ec2Spec, fresh: Bool = False)

def settling(self, mut db: State, keys: List[String]) -> Bool

def scale(mut self, mut db: State) -> Tuple[Int, Int]

def policy(self, lo: Int, hi: Int, per: Int = Int(1)) -> Policy

def sweep(self, mut db: State) -> Int

def vcpus(instance_type: String) -> Int

comptime HEAD_PORT = 6390

comptime HEAD_SECONDS = 10