service.controller¶

The controller: follows the workflows of the runs and records their end.

The state of a run is the service's, not the one of Argo (Ruling 17b): a split task is done when its <task>-merge node succeeded (when its coordinator <task>-w0 merges, TaskPlan.coordinator_merges, there is no such node and it is done when <task>-w0 succeeded and wrote out/.simrun-merged; without the marker it waits for it, running, up to MERGE_MARKER_GRACE_S from the first poll which saw the coordinator succeeded (RunRecord.merge_pending), and fails after it: the harness of its image is one which does not merge, from simrun 0.1.0) and failed when its <task>-w0 or its merge failed (after the retries of Argo), an extra worker never fails a task; an unsplit task is its node <task>. When every task is done the run succeeded, when a task failed the run failed; in both cases the controller stops the workflow, so extra workers still waiting for admission go away, and releases the reserved node. A run whose workflow ended is not failed for its phase while a task waits for the marker of its merge.

A pod is retried once with the retry memory after a kill for memory: a worker, a merge or an unsplit task killed for memory again with the retry memory fails the run (Ruling 35), whose message names the retry memory. A kill for memory needs the evidence of the out of memory killer (Ruling 39): the reason OOMKilled of the backend, or the report of an attempt which exited 137 counting kills of the out of memory killer in its cgroup; a pod killed after its grace period was evicted.

The states of the tasks and the out of memory kills are stored in the record at every poll. When the workflow is gone (deleted by its time to live before the controller saw its end) the tasks are resolved from these states and from the store: a split task whose merge wrote out/.simrun-merged is done (Ruling 23), one which waits for the marker of its merge keeps the run open until the marker is there or its wait is over; only the tasks still undetermined fail.

The reports of the workers of a task (of simrun-exec for an unsplit plugin task, by the task id) are recorded as resource observations once the state of the task is final, each marked in tasks/<task>/recorded/ right after its observation; a kill for memory of an attempt without a report is an observation of the pod shape at its limit (Ruling 24).

The logs of the pods of a task are kept in the run store once the task is final and its pods terminated, the ones which ended before the workflow is stopped or cancelled (simrun.service.logs); an ended run stays followed (pending_logs) until its logs are kept or its workflow is gone.

A terminal record is written first; a release of the node or a cancellation of the workflow which fails stays in the record (pending_release, pending_cancel) and is retried at the next polls, which do nothing else for a terminal run (Ruling 25). No call of the backend is made while the record lock is held.

The controller keeps no state in memory: every reconcile reads the records of the live runs of the index of the store (simrun.service.index), which the service rebuilds from the records at its start, so a restarted service continues where the one before stopped; a record which cannot be read is logged and skipped.

Controller ¶

Controller(store, resources, backend, types)

Maps the status of the backend onto the records of the runs.

Create the controller.

Parameters:

Name	Type	Description	Default
`store`	`IndexedRunStore`	The run store with its index.	required
`resources`	`ResourceStore`	The store of the resource observations.	required
`backend`	`Backend`	The backend the workflows run on.	required
`types`	`dict[str, TaskType]`	The task types by name.	required

wake ¶

wake()

Make the control loop reconcile now instead of at its next poll.

wait_for_wake ¶

wait_for_wake(timeout_s)

Block until woken or timeout_s passed; clears the wake.

Parameters:

Name	Type	Description	Default
`timeout_s`	`float`	The longest time to wait, in seconds.	required

start_submission ¶

start_submission(run_id, job)

Run the submission of a run in the background, see in_flight.

A job which raises fails a run which is still submitting; its workflow may exist, so it is cancelled.

Parameters:

Name	Type	Description	Default
`run_id`	`str`	The run, written `submitting`.	required
`job`	`Callable[[str], None]`	Resolves, builds and submits the run; called with its id.	required

in_flight ¶

in_flight(run_id)

Whether the submission of a run runs in this process.

close ¶

close()

Stop taking submissions, without waiting for the running ones.

A build may take an hour; the pod of the service is killed long before. A run left submitting is resumed at the next start of the service (recover), from its record.

recover ¶

recover()

Settle what a stopped service left: called at the start of the service.

The service is one process, so a run submitting since before the controller started has lost its submission: it is resumed from its record (resume, idempotent: an image which was built is found, a node reserved for the run is kept, a workflow which was created is taken as submitted). A submission created later may be in flight (the API accepts runs while this runs) and is left to reconcile, which never fails a submission running in this process. A record which is not terminal and has no workflow name was written before names were stored, and fails. The pending logs, releases and cancellations of terminal runs are retried.

reconcile ¶

reconcile()

Bring the records of the live runs up to date with the backend.

The live runs are the ones of the index which have not ended or have a release or a cancellation pending; an ended run is not read again. Runs missing from the index are indexed first (IndexedRunStore.reindex) and followed also while their index rows cannot be written.

write_if_unchanged ¶

write_if_unchanged(before, after)

Write after when the record on disk is still before.

Returns:

Type	Description
`bool`	Whether `after` was written; not when the run changed or is gone.

cancel ¶

cancel(run_id)

Cancel a run: write it cancelled, then cancel its workflow, release its node.

Raises:

Type	Description
`KeyError`	The run does not exist.
`ValueError`	The run has ended.

settle ¶

settle(run_id)

Retry the pending logs, release and cancellation of a terminal run.

The logs are kept before the workflow is cancelled; a workflow which does not exist (any more) has no logs to keep. An unknown node counts as released and an unknown workflow as cancelled; a node is released only from the reservation of this run. A run which was deleted meanwhile is ignored.

has_merge_marker ¶

has_merge_marker(task_dir)

Whether the output of a task has the marker of a merge, out/.simrun-merged.

The directories are listed instead of looking up the marker: on NFS a listing revalidates the directory, while a lookup may answer from a cached negative entry of a lookup before the pod wrote the file. The task directory is listed first, so a cached negative entry of out/ itself does not hide it either.

Parameters:

Name	Type	Description	Default
`task_dir`	`Path`	The directory of the task in the run store.	required

Returns:

Type	Description
`bool`	Whether the marker exists; not when `out/` does not exist.

task_plans ¶

task_plans(record)

The plans of the tasks of a run from its record.

A plan written before coordinator_merges existed reads it as False.

task_states ¶

task_states(workflow, types, status, plans, merge_state)

The state of every task of a workflow from the phases of its DAG nodes.

A task is running when one of its pods runs (or its coordinator is done and the merge waits), preempted when none runs and a pod evicted waits for its retry, retrying when none runs and a pod which failed otherwise waits for its retry, queued when it has pods of which none runs, waiting without pods.

Parameters:

Name	Type	Description	Default
`workflow`	`Workflow`	The workflow of the run.	required
`types`	`dict[str, TaskType]`	The task types by name.	required
`status`	`BackendStatus`	The status of the workflow in the backend.	required
`plans`	`dict[str, TaskPlan]`	The plans of the tasks, by id; the state of a split task whose plan merges in the coordinator is `merge_state` once `<task>-w0` succeeded.	required
`merge_state`	`Callable[[str], TaskState]`	The state of a task by id whose coordinator which merges succeeded: done with the marker of its merge, running while the marker is awaited, failed when the coordinator ran a harness which does not merge.	required

Returns:

Type	Description
`dict[str, TaskState]`	The state of every task by id.

attempt_of ¶

attempt_of(name)

The attempt of a report file <worker>-<attempt>.json, 0 without one.

is_oom ¶

is_oom(report, attempt, kills)

Whether a report is of a worker which was killed for memory.

Only the evidence of the out of memory killer counts (Rulings 34, 39): a peak near the limit is no kill, the kernel reclaims the page cache there, and an exit 137 may be a kill after the grace period.

Parameters:

Name	Type	Description	Default
`report`	`WorkerReport`	The report.	required
`attempt`	`int`	Its attempt.	required
`kills`	`dict[str, list[int]]`	The attempts killed for memory per worker (`RunRecord.oom_kills`).	required

Returns:

Type	Description
`bool`	True when the attempt is one killed for memory, or the report counts kills
`bool`	of the out of memory killer in its cgroup.

closed ¶

closed(record, state, tasks, message, **update)

The terminal record: tasks not final are skipped, the node is to release.

Parameters:

Name	Type	Description	Default
`record`	`RunRecord`	The record.	required
`state`	`RunState`	The terminal state.	required
`tasks`	`dict[str, TaskState]`	The states of the tasks.	required
`message`	`str \| None`	The message, e.g. why the run failed.	required
`**update`	`object`	Further fields of the record.	`{}`

Returns:

Type	Description
`RunRecord`	The record, ended now.