orca.fill

[pipelines.gharchive] fills = "all" # "since_registration", the default fill_order = "oldest" # or "newest" fill_max_runs = 2 # fill runs at once, beside its other runs fill_budget_per_day = 5.0 # estimated US dollars a UTC day fill_window = "22..6" # UTC hours fill runs may start in fill_lag = "2h" # how long after its end a partition fills

With fills = "all" the scheduler queues, on each tick, runs (trigger fill) for the partitions of the pipeline's range that nothing has filled: from its first partition through the last one that has ended at least fill_lag ago, in fill_order, as long as every limit allows (fill_status).

fill_lag (0, the default, for none) is for sources that publish a partition's data some time after it ends: GH Archive puts each hour's file up minutes after the hour, so at 00:00 UTC the day that just ended isn't all there yet. A Daily day D fills from D+1 00:00 plus the lag (on the schedule's clock), a Monthly month from the 1st of the next plus the lag. Until then the partition is outside the range, for the fill and for the coverage alike (range_keys), so it isn't counted as missing; the pipeline page's Fill line says when it fills (FillStatus.lag_key). The default, since_registration, is what orca did before: a schedule fills from the pipeline's registration on, and orca backfill the rest by hand.

What a fill takes. A partition is left alone when it's done in a run of any code of the pipeline (every mapped step succeeded for it; with no mapped step, a run over it succeeded), when a live run has it, or when a fill run of it was ever queued, so a fill never runs a partition twice: one whose fill run failed past its retries, or was cancelled, is left to a person (orca backfill, or resume in the UI). Running done partitions again stays orca backfill --rerun. All of that is within the pipeline's current data version, its steps' version= (#297): raising one makes every partition fillable again, and the fill regenerates them within its limits.

Limits, each a reason the pipeline page gives for a fill being idle: the pipeline paused (by a person or by the watchdog, orca.watchdog), the hour outside fill_window, the day's budget, the pipeline's other queued runs (a scheduled or backfilled run starts first, so fills never starve them), and fill_max_runs. Fill runs don't count against max_runs.

On a worker pool (placement = "worker:<pool>") the scheduler queues the fill runs the same way, in the same order and within the same limits, and leaves them queued, held for a worker of the pool to claim (orca.workers). fill_max_runs caps the pipeline's live fill runs across the whole pool, queued ones included, so the fill never has more than that many going whatever the number of workers; each worker's --max-runs and its grants cap what one machine takes. A worker pool's runs cost nothing more than its machines, which run anyway.

The cost estimate is conservative, and fails closed. A fill run's cost is its instances' time (orca.instances) at orca.fleet's on-demand price table, times COST_MARGIN for what the table leaves out (EBS, transfer, requests). The day's spend is that over the fill runs' instances since 00:00 UTC, up to now. Before starting a run, orca also counts what the fill runs going and the new one may still cost: the pipeline's hourly price (head plus its most workers) for the longest fill run of its last _HISTORY (an hour if it has none, at least a minute, plus the warm pool's idle time on EC2), less what each has run already. A run starts only if the spend plus that stays within fill_budget_per_day. When a price isn't known (an instance type not in the table, or Kubernetes, whose cluster orca can't price), no fill run starts on a budget, and the reason says so. A pipeline whose runs cost money (placed on EC2 or Kubernetes) starts no fill run without a budget. Runs on the control host or local processes cost nothing more than the machine already running.

Keyed pipelines and sensors don't fill: orca register refuses fills = "all" for them (fill_refusal).

struct Fill

def of(fills: String, order: String, max_runs: Int, budget: Float64, window: String, lag: String = "") -> Self

def in_window(self, hour: Int) -> Bool

def window_text(self) -> String

struct FillPipeline

struct FillStatus

def __init__(out self)

def parse_fills(v: String) -> String

def parse_fill_order(v: String) -> String

def parse_fill_max_runs(v: String) -> Int

def parse_fill_budget(v: String) -> Float64

def parse_fill_window(v: String) -> Tuple[Int, Int]

def parse_fill_lag(v: String) -> Int

def lag_text(seconds: Int) -> String

def money(x: Float64) -> String

def fill_refusal(kind: String, start: String, sensor: Bool, pipeline: String) -> String

def last_ended(kind: String, tz: String, now: Int64, lag: Int = Int(0)) -> String

def lag_wait(kind: String, tz: String, now: Int64, lag: Int) -> Tuple[String, Int64]

def range_keys(kind: String, start: String, tz: String, now: Int64, lag: Int = Int(0)) -> List[String]

def run_hourly(placed: String, head_type: String, worker_type: String, workers: Int) -> Float64

def unknown_type(placed: String, head_type: String, worker_type: String, workers: Int) -> String

def fill_status(mut tx: Transaction, p: FillPipeline, now: Int64, run_retries: Int) -> FillStatus

def lag_note(s: FillStatus) -> String

def eta_text(seconds: Int) -> String

def status_json(s: FillStatus) -> String

comptime ALL = "all"

comptime SINCE_REGISTRATION = "since_registration"

comptime FILL_TRIGGER = "fill"

comptime COST_MARGIN = 1.2

comptime NO_HISTORY_SECONDS = 3600

comptime MIN_SECONDS = 60

comptime MAX_LAG_SECONDS = 2678400