videoflow.engines package

Submodules

videoflow.engines.kubernetes module

Execution engine that runs a distributed flow on a Kubernetes cluster: it renders one Deployment/Job (+ ConfigMap) per node and applies them with kubectl. The worker container image is the same videoflow.worker entrypoint the local engine uses — only the launch mechanism (pods vs. subprocesses) differs.

Requires kubectl on PATH, configured against the target cluster, and the per-component images already built and pushed (see docker/).

class videoflow.engines.kubernetes.KubernetesExecutionEngine(nats_url: str, namespace: str = 'default', default_image: str | None = None, image_overrides: dict | None = None, blob_redis_url: str | None = None, blob_ttl_seconds: int | None = None, specs: list | None = None, kubectl: str = 'kubectl', envelope_version: int | None = None, provision_image: str | None = None, autoscaling: bool = False, max_replicas: int = 10, nats_monitoring_endpoint: str | None = None, mounts: list[Mount] | None = None, gpu_runtime_class: str | None = None, gpu_mode: str = 'exclusive', gpu_resource_name: str | None = None, gpu_autoscaling: bool = False, image_pull_policy: str = 'IfNotPresent', supervision: SupervisionPolicy | None = None, priority_class: str | None = None, profile_requests: dict[str, str] | None = None, stream_replicas: int = 1, rollout_policy: str | None = None, gpu_nodes: list[str] | None = None, resources: dict[str, dict[str, str]] | None = None, single_run: bool = False)

Bases: ExecutionEngine

• Arguments:

  • nats_url: URL workers use to reach NATS from inside the cluster, e.g. nats://nats.videoflow.svc:4222.

  • namespace: target namespace (must already exist).

  • default_image: image ref for nodes that don’t declare their own image= (e.g. ghcr.io/acme/app:v1, built FROM videoflow-base with your code).

  • image_overrides: optional mapping of node name to image ref (wins over both).

  • blob_redis_url: optional Redis URL for the large-payload blob store.

  • blob_ttl_seconds: TTL override for offloaded payloads (PROTOCOL.md BLOB-7); None lets workers pick the flow-type default.

  • specs: optional precompiled NodeSpec list; compiled from tasks_data if omitted.

  • kubectl: kubectl binary name/path.

  • mounts: optional Mount records (see manifests.parse_mounts) — hostPath volumes added to every node workload.

  • gpu_runtime_class: runtimeClassName for GPU pods (nvidia on k3s and other distros where the NVIDIA runtime is opt-in rather than the node default). Without it a GPU pod schedules but sees no device.

  • gpu_mode: GPU strategy name ('exclusive', the default: whole-device claims via the extended resource; see deploy.gpu).

  • gpu_resource_name: deploy-level extended-resource name for GPU claims (clusters advertising whole devices under a non-default name).

  • gpu_autoscaling: include GPU nodes in KEDA autoscaling (off by default — each extra replica claims whole GPUs).

  • image_pull_policy: imagePullPolicy for every rendered container. Defaults to IfNotPresent, which is what lets a locally built image loaded into the cluster actually run (see manifests.render_manifests).

dump_failed_logs(nodes: List[str], node_label: str | None = None) → None

Prints the recent pod logs of each failed node (before teardown removes them).

join_task_processes() → None

Blocks until every node Job for this run reaches a terminal state. Keeps Flow.join()’s documented block-then-return contract: a watchdog abort (unschedulable pods — the flow can never finish) is logged, not raised, so programmatic callers are not crashed mid-join; the CLI’s deploy path calls wait_for_completion directly and does get the exception.

rollout_report(deadline_secs: int = 150, poll_secs: int = 3, unschedulable_grace_secs: int = 30) → RolloutReport

Bounded post-apply health check for a REALTIME run — the counterpart of the wait_for_completion watchdog (a REALTIME deploy otherwise returns immediately, and a broken node fails silently: producers keep publishing, frames are evicted, downstream output never appears). Success means every pod is Ready (the readiness probe passes only after node.open() completes) or Succeeded (a finite-producer Job); merely being scheduled is not enough — a pod that finds a node and then dies in open() (CUDA OOM, a bad image, the startup-probe window killing a slow model load) crash-loops with no Unschedulable condition to see.

Exits early both ways: as soon as every pod is Ready on two consecutive polls (seconds, in the healthy case), and as soon as any pod is confirmed failing — a fatal waiting reason or repeated restarts on two consecutive polls, or scheduler-Unschedulable past unschedulable_grace_secs with no autoscaler scale-up in flight. The default deadline exceeds the startup-probe window so a probe kill of a slow open() is observable at all.

• Arguments:

  • deadline_secs: give up after this long without a verdict; pods still not Ready are then reported as warnings, not failures (a slow model load is slow, not proven dead).

  • poll_secs: seconds between kubectl polls.

  • unschedulable_grace_secs: how long a pod may sit Unschedulable before it is a confirmed failure (tolerates scheduling churn).

• Returns:

  • A RolloutReport — failing non-empty means the deploy should dump logs and exit non-zero.

signal_flow_termination() → None

Stops the flow and removes its resources (used by Flow.stop).

teardown() → None

Tears down this run: publishes the control stop + deletes the run’s broker streams (best-effort — the deploying host may not be able to reach the NATS URL the in-cluster workers use), then always deletes every Kubernetes resource for this run by label. The k8s cleanup is the guarantee; the broker step is skipped with a warning if NATS is unreachable.

wait_for_completion(poll_secs: int = 3, unschedulable_grace_secs: int = 60) → List[str]

Blocks until every node Job for this run has succeeded or failed. Returns the list of failed node names (empty when the whole flow completed cleanly). Fails fast two ways: returns the instant any node Job exhausts its backoffLimit, and raises RuntimeError when any pod has sat scheduler-Unschedulable (e.g. Insufficient nvidia.com/gpu) for unschedulable_grace_secs — an unschedulable pod never runs, never consumes its backoffLimit, and would otherwise leave this loop (and the whole flow, via backpressure) hanging forever. The grace period tolerates scheduling churn, and the abort is skipped while a cluster-autoscaler scale-up is in flight.

class videoflow.engines.kubernetes.RolloutReport(failing: list[tuple[str, str]], warnings: list[str])

Bases: object

Outcome of rollout_report. failing are confirmed failures the deploy should abort on — (node label, human detail) pairs; warnings are ambiguous findings (slow startup, no pods yet, autoscaler scale-up in flight) that deserve stderr but not a non-zero exit.

failing: list[tuple[str, str]]

warnings: list[str]

videoflow.engines.local module

Execution engine that runs a distributed flow entirely on the local machine, one OS subprocess per node (per replica, for nb_tasks > 1), all talking to a local NATS server. Same videoflow.worker code path Kubernetes uses — only the way processes are started differs — so it’s the primary way to develop and test a flow without a cluster.

Prerequisite: a running NATS JetStream server, e.g. nats-server -js or docker run -p 4222:4222 nats -js.

class videoflow.engines.local.LocalProcessEngine(nats_url: str = 'nats://localhost:4222', blob_redis_url: str | None = None, specs: List[NodeSpec] | None = None, local_docker_nats_url: str | None = None, python_path: list | None = None, inherit_python_path: bool = True, default_image: str | None = None, blob_ttl_seconds: int | None = None, supervision: SupervisionPolicy | None = None, profile_requests: dict[str, str] | None = None, gpu_policy: str = 'shared')

Bases: ExecutionEngine

• Arguments:

  • nats_url: URL of the NATS server every worker connects to.

  • blob_redis_url: optional Redis URL for the large-payload blob store.

  • specs: optional precompiled list of NodeSpec. If not given, they are compiled from the flow’s tasks_data at allocate_and_run_tasks time.

  • python_path: extra directories prepended to each worker’s PYTHONPATH.

  • inherit_python_path: also re-export this process’s own sys.path additions (default True) — what makes node classes defined next to the graph importable in the workers. Set False for a hermetic child env.

  • default_image: image used for a native component that declares no image= — the solution image run-local auto-builds. A node’s own image= still wins.

  • blob_ttl_seconds: TTL override for offloaded payloads (PROTOCOL.md BLOB-7); None lets workers pick the flow-type default (3600s realtime / 86400s batch).

  • supervision: how a dead worker is restarted. Defaults to SupervisionPolicy.local() — the same restart count Kubernetes uses, with a compressed backoff. This is the point of the parameter: a crash used to recover in the cluster and hang here, which made local development the one place the recovery path was never exercised. Pass SupervisionPolicy.disabled() (--no-restart) for a tight debug loop.

events() → EventLog

Lifecycle events for this run — the same records the Kubernetes engine emits.

failures() → List[tuple]

(node_name, replica_idx, returncode) for each worker that gave up.

join_task_processes() → None

Blocking method. It is supposed to make the calling process sleep until all task processes have finished processing.

report_failures() → None

Prints one line per lifecycle event worth reading — restarts included, so a run that recovered says so. Local workers inherit stdout/stderr, so their tracebacks are already on the terminal; this is the index, not a dump.

signal_flow_termination() → None

Signals the execution environment that the flow needs to stop.

wait_for_completion() → List[str]

Blocks until every worker process exits, restarting failed ones per the supervision policy. Returns the names of nodes that ran out of restarts (empty when the flow ran cleanly) — the same contract as KubernetesExecutionEngine.wait_for_completion, and now the same behaviour: a crash that the cluster would recover from recovers here too, because the restarted worker rebinds the same durable and its un-acked messages are redelivered.

A worker killed by SIGINT/SIGTERM is not counted and not restarted: that is Ctrl-C or flow.stop() propagating, not a failure.

Every worker is watched concurrently: one waiter thread per child reports its exit on a queue, and this loop drains the queue. Waiting on the children one after another — the previous shape — meant a processor that died while the source ahead of it in the list was still running was not restarted (or the flow not failed) until that source exited, which for an unbounded source is never: a healthy producer hid a dead downstream worker indefinitely.

videoflow.engines.local.allocate_local_gpus(specs: List[NodeSpec], flow_id: str, run_id: str, backend: LocalAllocationBackend) → dict[tuple[str, int], Mapping[str, str]]

The GPU environment of every worker, decided before any worker is launched: CUDA_VISIBLE_DEVICES (UUIDs), VF_GPU_COUNT (the delivered count) and VF_GPU_GRANT_JSON. Strict policy raises ResourceUnavailable naming every reason the flow does not fit, and refuses an unobservable host; the shared policy launches on an unobservable host without a mask but marks every grant host='unobserved' so nobody reads it as a zero-GPU machine.

videoflow.engines.local.assign_local_gpus(specs: List[NodeSpec], host_gpus: list[int]) → dict[tuple[str, int], list[int]]

Deterministic device assignment for a local run: walking specs in order, each replica of each GPU node takes the next gpu_count ordinals from host_gpus — the local twin of the exclusive Kubernetes grant (RFC 0003), so a worker’s CUDA_VISIBLE_DEVICES shows exactly its granted devices. Docker-run native components are skipped: they cannot receive the mask (see _runs_via_docker), so granting them ordinals would only starve the workers that can.

When demand exceeds len(host_gpus) the walk wraps around (duplicates within one replica are collapsed, with a per-replica warning naming the short grant) and a single aggregate warning is logged: sharing devices is fine for dev, but the same flow will not schedule that way on Kubernetes. An empty host_gpus returns an empty mapping — no env gets set, so CPU-fallback GPU nodes on a GPU-less machine behave exactly as before.

videoflow.engines.local.inherited_python_path() → list

The sys.path entries this process added beyond the interpreter’s own defaults — typically the graph/solution directory (inserted by videoflow.deploy.compile.load_flow), an editable checkout, or a test support dir.

Worker subprocesses inherit the environment but not sys.path, so without re-exporting these as PYTHONPATH every node class living next to the graph fails to import in its worker.

videoflow.engines.local.local_workload_requests(specs: List[NodeSpec], flow_id: str, run_id: str) → list[WorkloadRequest]

One WorkloadRequest per replica of every GPU node the local engine can grant devices to (docker-run natives excluded, see _runs_via_docker), in launch order: whole-device requests are exclusive, a node declaring gpu_memory_gib is a cooperative sharer whose declared peak is that demand.

videoflow.engines.local.needs_container_image(spec: NodeSpec) → bool

Whether running spec locally requires a container image to exist.

Only a native component with no runtime.localCommand does: a Python node runs as a host subprocess in the current interpreter, and a native component with a localCommand runs that binary directly. This is the predicate run-local uses to decide whether to auto-build at all — most flows are pure Python, and building a (possibly CUDA) solution image to launch a few subprocesses would be a large and pointless cost.