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

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: 'exclusive' (whole-device claims, default) or 'shared' (no resource limit — GPU pods co-schedule and share physical devices; dev clusters only, see manifests.render_manifests).

  • gpu_resource_name: deploy-level default extended-resource name for GPU claims; a node’s own gpu_resource_name wins.

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

schedulability_report(grace_secs: int = 30, poll_secs: int = 3) → List[str]

Bounded post-apply check that every pod of this run found a node — the REALTIME counterpart of the wait_for_completion watchdog (a REALTIME deploy otherwise returns immediately, and an unschedulable node fails silently: producers keep publishing, frames are evicted, downstream output never appears). Returns problem strings, empty when everything scheduled within the grace window.

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.

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)

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

failures() → List[tuple]

(node_name, replica_idx, returncode) for each worker that failed.

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 failed worker. 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. Returns the names of nodes that had at least one replica exit non-zero (empty when the flow ran cleanly) — the same contract as KubernetesExecutionEngine.wait_for_completion.

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

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