`orca.ec2`

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, <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.

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 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). 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.

Structs

`Ec2Spec`

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.

`Ec2Runner`

struct Ec2Runner

Implements Movable

Fields: ec2: Ec2; mode: String; head_type: String; worker_type: String; deadline: Int; 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.

`init`

def __init__(out self)

`head_of`

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

`worker_of`

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

`cpu_of`

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

A run's instances' vCPUs, in thousandths.

`start`

def start(self, mut db: State, r: Ec2Spec)

Starts the run's head instance, then its workers, and records them in orca.instances.

`scale`

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`

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

`sweep`

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; returns how many.

Functions

`vcpus`

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

comptime HEAD_PORT = 6390