service.argo_backend¶

The backend of the service on a cluster: Argo Workflows, Kueue and BuildKit.

A run is an Argo Workflow (simrun.argo.to_argo) in the namespace of the cluster; Kueue admits its pods (plain pods with a queue label) and preempts lower priorities. The status of a workflow is read from the nodes of its DAG (workflow_status): a node retried by Argo (a Retry node) is Pending while its attempt waits for admission, Preempted while the attempt after an eviction waits (a pod deleted by Kueue or lost, or a worker stopped on SIGTERM), Retrying while the attempt after another failure waits (a crash, a pod killed after its grace period), Running while its attempt runs; the failed attempts are listed per worker with their exit codes and whether the kernel killed them for memory (OOMKilled).

The nodes are the ones labelled simrun/worker=true (the nodes of the ResourceFlavor of Kueue) without a taint which keeps pods off, a reservation of simrun aside; their cores and memory are the allocatable ones minus the wait container of the Argo executor and minus the requests of the pods on them which are no pods of runs (the DaemonSets of the cluster, e.g. of the NFS driver and of the monitoring), rounded down to whole cores and GiB, so a pod the planner sizes for a node fits it beside them.

A reservation is the taint simrun/reserved=<run id>:NoExecute, which evicts the pods of other runs from the node, and the label simrun/reserved=<run id>, which the pods of the run select.

An image is built by a Job of buildctl, which sends the build context (a ConfigMap of its files, owned by the Job) to the BuildKit daemon of the cluster and pushes the image to the registry with the credentials of the optional Secret simrun-registry-auth; whether the registry has an image is asked its API.

A thread watches the workflows of the runs and wakes the controller on every change, so the state of a run follows Argo at once; the controller still polls, which covers a watch that is restarted.

ArgoBackend ¶

ArgoBackend(config, api=None)

The backend of the service on a cluster of Argo Workflows and Kueue.

Create the backend.

Parameters:

Name	Type	Description	Default
`config`	`ClusterConfig`	The cluster.	required
`api`	`ApiClient \| None`	The client of the Kubernetes API; by default the service account of the pod, else the kubeconfig.	`None`

subscribe ¶

subscribe(callback)

Call callback on every change of a workflow of a run.

The first subscription starts the thread of the watch of the workflows, which every subscription shares.

close ¶

close()

Stop the watch of the workflows, at its next event at the latest.

join_watch ¶

join_watch(timeout_s=None)

Wait until the thread of the watch ended, after close.

Parameters:

Name	Type	Description	Default
`timeout_s`	`float \| None`	The longest wait in seconds, None for no limit.	`None`

submit ¶

submit(record, plans, types, cluster)

Create the workflow of a run; returns its name.

A workflow of the name which exists already is taken as submitted: the name of the workflow of a run is fixed, so a resumed submission creates it once.

Raises:

Type	Description
`ValueError`	The workflow cannot be built, e.g. a task has no image.

status ¶

status(name)

The status of a workflow; KeyError when it does not exist.

cancel ¶

cancel(name)

Terminate a workflow and its pods; KeyError when it does not exist.

stop ¶

stop(name)

Stop a workflow whose run is decided; KeyError when it does not exist.

logs ¶

logs(name, task_id, tail_lines=None)

The logs of the main containers of the pods of a task, oldest first.

Parameters:

Name	Type	Description	Default
`name`	`str`	The workflow.	required
`task_id`	`str`	The task.	required
`tail_lines`	`int \| None`	The last lines of every log, all without.	`None`

Returns:

Type	Description
`list[PodLog]`	A log per pod which still exists, every attempt, with whether the pod
`list[PodLog]`	terminated; a log which cannot be read says why, e.g. that its pod
`list[PodLog]`	has not started. At most
`list[PodLog]`	`LOG_LIMIT_BYTES` of a log are read (the first of its last `tail_lines`
`list[PodLog]`	lines).

nodes ¶

nodes()

The worker nodes on which pods can run, with their reservations.

Their capacity is the allocatable one less the requests of the pods bound to them which have not terminated and are no pods of runs, e.g. of DaemonSets; Kueue counts the pods of the runs itself.

placements ¶

placements()

The pods of the runs which have not terminated, with nodes and requests.

reserve ¶

reserve(node, run_id)

Taint and label a node for a run; KeyError for an unknown node.

The taint replaces a reservation the node had; the resource version of the node guards the list of taints against a concurrent change.

unreserve ¶

unreserve(node, run_id)

Remove the taint and the label of the reservation of a run.

Raises:

Type	Description
`KeyError`	The node does not exist.

build_image ¶

build_image(context, ref)

Build the image of a flat build context with BuildKit and push it.

Raises:

Type	Description
`ValueError`	The context has a directory or is too large for a ConfigMap.
`RuntimeError`	The build failed.
`TimeoutError`	The build took longer than `BUILD_TIMEOUT_S`.

has_image ¶

has_image(ref)

Whether the registry has the image; asks the API of the registry.

Raises:

Type	Description
`ValueError`	The image is not in the registry of the cluster.
`RuntimeError`	The registry answered neither found nor not found.

workflow_status ¶

workflow_status(workflow)

The status of a workflow from its Argo object.

Parameters:

Name	Type	Description	Default
`workflow`	`dict[str, Any]`	The workflow as the API returns it.	required

Returns:

Type	Description
`BackendStatus`	The phase of the workflow, of every task node of its DAG by its name relative
`BackendStatus`	to the workflow, and the failed attempts of the pods per worker id, per task
`BackendStatus`	id of an unsplit task and per `<task>-merge` of a merge.

pod_requests ¶

pod_requests(pod)

The cores and GiB a pod requests, as the scheduler counts them.

The containers together, or an init container alone when it requests more (the init containers run before the containers, one at a time; a sidecar, an init container which keeps running, counts with the containers and the init containers after it), plus the overhead of the pod.

Parameters:

Name	Type	Description	Default
`pod`	`dict[str, Any]`	The pod as the API serialises it.	required

Returns:

Type	Description
`tuple[float, float]`	The cores and the GiB.

node_from ¶

node_from(node, used=(0.0, 0.0))

The node of the planner of a Kubernetes node, None when pods cannot run on it.

Parameters:

Name	Type	Description	Default
`node`	`dict[str, Any]`	The node as the API serialises it.	required
`used`	`tuple[float, float]`	The cores and GiB the pods on the node request which are no pods of runs (e.g. of DaemonSets).	`(0.0, 0.0)`

Returns:

Type	Description
`Node \| None`	The node with its allocatable cores and memory minus one executor container
`Node \| None`	and minus `used`, rounded down, for the planner, the allocatable cores and
`Node \| None`	memory minus `used` as the scheduler counts the requests of the pods of the
`Node \| None`	runs, and its reservation; None for a node which is unschedulable or has a
`Node \| None`	taint other than a reservation which keeps pods off.

manifest_url ¶

manifest_url(ref, config)

The URL of the manifest of an image of the registry of the cluster.

Parameters:

Name	Type	Description	Default
`ref`	`str`	The image, `<registry>/<repository>[:<tag>\|@<digest>]`.	required
`config`	`ClusterConfig`	The cluster.	required

Returns:

Type	Description
`str`	The URL of the manifest in the API of the registry.

Raises:

Type	Description
`ValueError`	The image is not in the registry of the cluster.

backend ¶

backend()

The backend of the settings named by SIMRUN_CONFIG, for serve.