videoflow.deploy package

Submodules

videoflow.deploy.build module

Image auto-build for videoflow deploy: when no --image is given, find the Dockerfile next to the graph module (gpu.Dockerfile for a flow with GPU nodes), build the videoflow-base image it is FROM if missing, and build the solution image from the enclosing git root (solution Dockerfiles COPY sibling packages, so the repo root is the context).

The base images can only be auto-built from a videoflow source checkout (docker/base/Dockerfile COPYs the source tree); a wheel-only install gets a precise error with the manual commands instead.

videoflow.deploy.build.autobuild(graph_dir: str, needs_gpu: bool, context_override: str | None = None) → str | None

The whole auto-build path: find the Dockerfile, ensure its base image, build the solution image. Returns the built tag, or None when the solution ships no Dockerfile (caller falls back to explicit-image resolution).

videoflow.deploy.build.base_image_for(dockerfile_path: str) → str | None

The default of the Dockerfile’s ARG BASE_IMAGE= line, or None if it has none.

videoflow.deploy.build.build_context_for(graph_dir: str, override: str | None = None) → str

The docker build context: an explicit --build-context, else the git root enclosing the graph (solution Dockerfiles COPY sibling packages from the repo root — e.g. the offside solution copies nine of them), else the graph dir.

videoflow.deploy.build.build_image(dockerfile: str, context: str, tag: str) → None

Builds the solution image, streaming docker output (layer cache makes unchanged rebuilds fast).

videoflow.deploy.build.default_tag(graph_dir: str) → str

Deterministic human-readable tag for the auto-built solution image, e.g. videoflow-offside:latest.

videoflow.deploy.build.docker_gpus_available() → bool

Whether the local docker daemon has the NVIDIA runtime (for –gpus all).

videoflow.deploy.build.ensure_base_image(base_ref: str) → None

Makes sure base_ref (a videoflow-base:* image) exists locally, building it from the videoflow source checkout if missing.

• Raises:

  • RuntimeError when the image is missing and videoflow is not an editable/source install (the base Dockerfile COPYs the source tree, so there is nothing to build from).

videoflow.deploy.build.find_dockerfile(graph_dir: str, needs_gpu: bool) → str | None

The Dockerfile deploy builds the node image from: gpu.Dockerfile when the flow has GPU nodes and one exists, else Dockerfile. None when the solution ships neither (the caller falls back to requiring --image).

videoflow.deploy.build.image_exists(ref: str) → bool

videoflow.deploy.build.run_in_image(image: str, command: List[str], mounts: List[Mount] | None = None, workdir: str | None = None, gpus: bool = False, capture: bool = False, interactive: bool = False) → str | None

Runs a command in the solution image with the given hostPath-style mounts (Mount records from manifests.parse_mounts) — how deploy executes the prepare hook and the graph compile without the graph’s deps on the host.

• Returns:

  • the command’s stdout when capture, else None.

• Raises:

  • RuntimeError on a non-zero exit (with stderr when captured).

videoflow.deploy.cli module

Command-line entrypoint for deploying a videoflow graph to Kubernetes.

videoflow deploy path/to/graph.py

deploy is one command that does everything: it generates the solution config (asking the template’s questions when none exists), runs the solution’s prepare hook, builds and loads the node image into the detected local cluster, provisions the broker (an in-cluster dev NATS + Redis when --nats is omitted), applies the flow, and — for a BATCH flow — waits for it to run to completion and then tears down every resource (Kubernetes workloads + broker streams + owned infra). A REALTIME flow is applied and left running; stop it later with videoflow teardown. Each automatic step has an explicit override (--config, --image, --nats, --mount, --no-prepare, --no-build, …). Nothing is written to disk (beyond a generated config.yaml) unless --render-only (write manifest files + kustomization) or --dry-run (print YAML to stdout) is given.

The graph module must expose a factory (default name build_flow) that returns a built videoflow.core.flow.Flow without calling .run() on it — the CLI needs the graph, not a running flow. When the graph’s dependencies are not importable on this machine, deploy compiles it inside the solution image instead.

videoflow.deploy.cli.build_parser() → ArgumentParser

videoflow.deploy.cli.main(argv: list[str] | None = None) → None

videoflow.deploy.cluster module

Best-effort detection of what kind of Kubernetes cluster kubectl points at, and the cluster-flavor-specific mechanics that depend on it: how to load a locally built image into it, whether hostPath mounts see the local filesystem, and whether GPU pods are schedulable. Everything here is advisory — deploy still works with explicit flags when detection gets it wrong.

Each flavor is one ClusterFlavorHandler registered at import. Supporting a new one (microk8s, colima, k0s) is a class plus a register_cluster_flavor call rather than an edit to three parallel if-ladders, which is what this used to be — and what made it easy to teach detection about a flavor while forgetting to teach image loading about it.

class videoflow.deploy.cluster.ClusterFlavorHandler

Bases: object

One local-cluster flavor: how to recognize it, how to get a locally built image into it, and whether its hostPath mounts see the host filesystem.

Subclasses set name and implement matches. load_images defaults to refusing, which is the right answer for anything remote.

hostpath_warning() → str | None

Message when hostPath will not resolve against the local filesystem, else None.

load_images(images: List[str], kubectl: str = 'kubectl') → None

Side-loads locally built images so pods can run them without a registry.

• Raises:

  • RuntimeError when the images cannot be loaded (remote cluster, missing tool, failed command). The message must name the fix.

matches(context_name: str, node_labels: Callable[[], str]) → bool

Whether the current cluster is this flavor. node_labels is a callable so a handler that can decide from the context name alone costs no kubectl call; the result is shared between handlers that do need it.

name: str = ''

Flavor identifier, returned by detect_cluster and passed back to load_images/hostpath_warning.

videoflow.deploy.cluster.allocatable_gpus(kubectl: str = 'kubectl', resource: str = 'nvidia.com/gpu') → int

Total allocatable units of one GPU extended resource across all nodes (0 when no node advertises it or the cluster is unreachable).

videoflow.deploy.cluster.current_context(kubectl: str = 'kubectl') → str

videoflow.deploy.cluster.detect_cluster(kubectl: str = 'kubectl') → str

Classifies the cluster kubectl currently points at, by asking each registered flavor in order. Context-name conventions identify kind/minikube/docker-desktop; k3s installs use a generic context name (‘default’), so they are confirmed via node facts instead — fetched at most once per call, and only if some handler asks for them.

videoflow.deploy.cluster.get_cluster_flavor(cluster: str) → ClusterFlavorHandler

The handler registered under cluster.

• Raises:

  • RuntimeError when no flavor is registered under that name.

videoflow.deploy.cluster.gpu_preflight(kubectl: str = 'kubectl', gpu_runtime_class: str | None = None, demand: dict | None = None, gpu_mode: str = 'exclusive') → List[str]

Checks what a GPU node workload needs (see manifests._pod_spec): a node labeled videoflow.io/gpu-pool=true; in exclusive mode, enough allocatable units of each requested extended resource to satisfy the flow’s whole demand (an under-provisioned flow schedules partially and stalls with the rest of its pods Pending); and — where the NVIDIA container runtime is an opt-in RuntimeClass rather than the node default — a --gpu-runtime-class, without which the pod schedules and then runs with no device. Returns problem strings with copy-pasteable fixes (empty list = OK).

• Arguments:

  • demand: dict of extended-resource name -> total units the flow requests (sum over GPU nodes of nb_tasks * gpu_count), or None to skip the capacity comparison and only check that the resource exists.

  • gpu_mode: 'exclusive' or 'shared'. Shared pods carry no resource limit, so the capacity/device-plugin checks are skipped; the RuntimeClass check is escalated instead, because in shared mode the runtime class is the only thing granting device access at all.

videoflow.deploy.cluster.hostpath_warning(cluster: str) → str | None

A message when hostPath mounts will NOT resolve against the local filesystem (the cluster “node” is a VM/container with its own filesystem), else None. An unregistered flavor warns nothing rather than raising: this is advisory, and deploy should not fail over a missing warning.

videoflow.deploy.cluster.load_images(cluster: str, images: List[str], kubectl: str = 'kubectl') → None

Loads locally built docker images into the detected cluster so pods can pull them without a registry.

• Raises:

  • RuntimeError when the cluster is remote (push to a registry instead), when a required tool is missing, or when a load command fails.

videoflow.deploy.cluster.nvidia_runtimeclass(kubectl: str = 'kubectl') → str | None

The NVIDIA RuntimeClass name when the cluster registers one, else None. Prefers the conventional nvidia but also recognizes variant names (e.g. a distro registering nvidia-container-runtime) so the shared-mode escalation cannot silently no-op on them.

videoflow.deploy.cluster.register_cluster_flavor(handler: ClusterFlavorHandler, before: str | None = 'generic-remote') → None

Registers a cluster flavor. Detection tries handlers in registration order, so a new handler is inserted before before (by default the generic-remote fallback, which matches everything and must stay last).

• Arguments:

  • handler: the flavor handler instance.

  • before: name of the handler to insert ahead of, or None to append.

• Raises:

  • ValueError: before names no registered flavor.

videoflow.deploy.compile module

Compiles a graph module to a JSON specs document, for use where the operator machine cannot import the graph (its ML dependencies live only in the solution image). videoflow deploy runs this inside that image:

docker run --rm -v <graph_dir>:<graph_dir> -w <graph_dir> <image>         python -m videoflow.compile graph.py[:factory]

and rebuilds the NodeSpec list on the host with NodeSpec.from_dict — the same serialization the provision Job’s specs ConfigMap uses.

Output (stdout): {"flow_id": ..., "flow_type": ..., "specs": [...]}.

videoflow.deploy.compile.compile_to_dict(target: str, envelope_version: int | None = None) → dict

videoflow.deploy.compile.load_flow(target: str) → Flow

• Arguments:

  • target: path/to/graph.py or path/to/graph.py:factory_name (factory defaults to build_flow).

• Returns:

  • a built Flow produced by calling the factory.

videoflow.deploy.compile.main(argv: List[str] | None = None) → None

videoflow.deploy.compile.specs_from_document(document: dict | str) → tuple[str, str, List[NodeSpec]]

(flow_id, flow_type, specs) from a compile-JSON document (dict or JSON string).

videoflow.deploy.gpu module

GPU allocation strategies: how a GPU node’s pods claim devices, what to preflight before deploying them, and any cluster state a strategy needs to set up for a run and restore afterwards.

Two strategies ship. exclusive claims whole devices through an integer extended resource, so the scheduler accounts for them and a flow that outgrows the cluster stays Pending rather than thrashing. shared emits no resource limit at all: pods land on the pool by selector alone and share the physical devices through the NVIDIA runtime — dev-box semantics, no accounting, no memory isolation.

The mode used to be a bare string branched on in four places (pod resources, manifest validation, preflight, the CLI’s choices), which is why adding a third meant finding all four. A strategy is now one class registered here. The two lifecycle hooks, prepare and cleanup, are no-ops for both built-ins but exist because the strategies we expect next need them: setting device-plugin time-slicing replicas for the duration of a run and reverting after, or partitioning GPUs from measured per-component memory demand.

videoflow.deploy.gpu.DEFAULT_GPU_RESOURCE = 'nvidia.com/gpu'

Extended resource used when neither the node nor the deploy names one.

class videoflow.deploy.gpu.ExclusiveGpu

Bases: GpuStrategy

Whole-device claims through an integer extended resource (the default).

name: str = 'exclusive'

Mode name, as used by --gpu-mode.

pod_resources(spec: NodeSpec, gpu_resource_name: str | None = None) → dict

The container resources fragment for one GPU pod — {} for a strategy that requests nothing.

Stays a plain dict on purpose: it is spliced straight into a container spec, so its shape is Kubernetes’ ResourceRequirements schema and not a record we own. A strategy may legitimately emit requests, limits or neither, with arbitrary extended-resource keys.

• Arguments:

  • spec: the node’s NodeSpec (gpu_count, gpu_resource_name).

  • gpu_resource_name: deploy-level default extended-resource name.

preflight_problems(kubectl: str = 'kubectl', demand: dict[str, int] | None = None, gpu_runtime_class: str | None = None) → List[str]

Strategy-specific preflight problems, each a string naming its fix. The flavor-independent checks (cluster reachable, a labeled GPU node) are run by cluster.gpu_preflight before this is called.

• Arguments:

  • demand: extended-resource name -> units the flow requests, or None to skip capacity comparison.

  • gpu_runtime_class: the --gpu-runtime-class value, if given.

videoflow.deploy.gpu.GPU_POOL_LABEL = 'videoflow.io/gpu-pool'

Node label a GPU pod selects on, and the taint key it tolerates.

class videoflow.deploy.gpu.GpuStrategy

Bases: object

One GPU allocation mode.

A strategy owns three decisions that must agree with each other: what a GPU pod asks the scheduler for (pod_resources), what makes that request satisfiable and is therefore worth checking first (preflight_problems), and whether the cluster needs temporary reconfiguration to honour it (prepare/cleanup). Splitting them across modules is what made the old string-mode version easy to extend incorrectly.

cleanup(kubectl: str = 'kubectl') → None

Undoes prepare. Must be idempotent and tolerant: it is called after a prepare that only partly succeeded, and — for a REALTIME flow, whose lifetime outlives the deploy command — from a later videoflow teardown that passes --gpu-mode but shares no state with the deploy that ran prepare. So it cannot assume prepare completed, or ran at all.

name: str = ''

Mode name, as used by --gpu-mode.

pod_resources(spec: NodeSpec, gpu_resource_name: str | None = None) → dict

The container resources fragment for one GPU pod — {} for a strategy that requests nothing.

Stays a plain dict on purpose: it is spliced straight into a container spec, so its shape is Kubernetes’ ResourceRequirements schema and not a record we own. A strategy may legitimately emit requests, limits or neither, with arbitrary extended-resource keys.

• Arguments:

  • spec: the node’s NodeSpec (gpu_count, gpu_resource_name).

  • gpu_resource_name: deploy-level default extended-resource name.

preflight_problems(kubectl: str = 'kubectl', demand: dict[str, int] | None = None, gpu_runtime_class: str | None = None) → List[str]

Strategy-specific preflight problems, each a string naming its fix. The flavor-independent checks (cluster reachable, a labeled GPU node) are run by cluster.gpu_preflight before this is called.

• Arguments:

  • demand: extended-resource name -> units the flow requests, or None to skip capacity comparison.

  • gpu_runtime_class: the --gpu-runtime-class value, if given.

prepare(demand: dict[str, int] | None = None, kubectl: str = 'kubectl') → None

Cluster setup this strategy needs before a run’s manifests are applied. Default: nothing. A strategy that mutates cluster state here is responsible for restoring it in cleanup.

videoflow.deploy.gpu.SHARED_NEEDS_RUNTIME_CLASS = '--gpu-mode shared without --gpu-runtime-class'

The one preflight problem that is fatal in shared mode (cli hard-errors on it): shared pods carry no GPU resource limit, so the RuntimeClass is their only path to the device. A shared constant so the check never depends on message prose.

class videoflow.deploy.gpu.SharedGpu

Bases: GpuStrategy

No resource limit: pods co-schedule on the pool and share devices via the NVIDIA runtime.

name: str = 'shared'

Mode name, as used by --gpu-mode.

pod_resources(spec: NodeSpec, gpu_resource_name: str | None = None) → dict

The container resources fragment for one GPU pod — {} for a strategy that requests nothing.

Stays a plain dict on purpose: it is spliced straight into a container spec, so its shape is Kubernetes’ ResourceRequirements schema and not a record we own. A strategy may legitimately emit requests, limits or neither, with arbitrary extended-resource keys.

• Arguments:

  • spec: the node’s NodeSpec (gpu_count, gpu_resource_name).

  • gpu_resource_name: deploy-level default extended-resource name.

preflight_problems(kubectl: str = 'kubectl', demand: dict[str, int] | None = None, gpu_runtime_class: str | None = None) → List[str]

Strategy-specific preflight problems, each a string naming its fix. The flavor-independent checks (cluster reachable, a labeled GPU node) are run by cluster.gpu_preflight before this is called.

• Arguments:

  • demand: extended-resource name -> units the flow requests, or None to skip capacity comparison.

  • gpu_runtime_class: the --gpu-runtime-class value, if given.

videoflow.deploy.gpu.get_gpu_mode(name: str) → GpuStrategy

The strategy registered under name.

• Raises:

  • ValueError: no strategy is registered under that name; the message names the known modes and register_gpu_mode.

videoflow.deploy.gpu.register_gpu_mode(strategy: GpuStrategy) → None

Registers a GPU allocation strategy under its name. Registering makes the mode selectable via --gpu-mode — the CLI builds its choices from here — so a new strategy needs no CLI edit.

• Arguments:

  • strategy: the strategy instance.

• Raises:

  • ValueError: the strategy has no name.

videoflow.deploy.gpu.registered_gpu_modes() → list[str]

Registered GPU mode names, sorted. The CLI’s --gpu-mode choices.

videoflow.deploy.gpu.resolve_gpu_resource(spec: NodeSpec, default: str | None = None) → str

The extended-resource name a GPU spec requests: the node’s own, else the deploy default (--gpu-resource-name), else nvidia.com/gpu.

videoflow.deploy.images module

Resolve the container image a node’s worker runs in on Kubernetes.

There is no module-path “family” inference: a user defines their processors in their own package and builds their own image (their code + deps on top of videoflow-base), so the image must be stated explicitly. Resolution order, first match wins:

1. a deploy-time override for the node (--image-override <name>=<ref>)

2. the node’s own image= kwarg (declared in graph code)

3. the deploy-time default (--image <ref>)

If none apply, resolution raises with an actionable message instead of guessing.

This module also owns the pull policy — how the cluster obtains that image once it is named. It lives here rather than in manifests because the CLI must read the default while building its parser, and manifests imports the optional yaml extra at module scope; this module imports nothing outside the stdlib.

videoflow.deploy.images.parse_override(spec: str) → tuple

Parses a name=ref CLI override into a (name, ref) tuple.

videoflow.deploy.images.resolve_image(node_name: str, node_image: str | None, default_image: str | None = None, overrides: dict | None = None) → str

• Arguments:

  • node_name: the node’s stable name (matched against overrides).

  • node_image: the image declared on the node (Node.image), or None.

  • default_image: the deploy-time flow default (--image), or None.

  • overrides: mapping of node name to image ref (--image-override).

• Returns:

  • the resolved image ref (str).

• Raises:

  • ValueError if no image can be determined for the node.

videoflow.deploy.images.validate_image_pull_policy(policy: str) → str

• Returns:

  • policy unchanged when it is a valid Kubernetes imagePullPolicy.

• Raises:

  • ValueError naming the accepted values, so a typo fails on the operator’s machine rather than as an unschedulable pod in the cluster.

videoflow.deploy.infra module

Auto-provisioning of the dev broker infrastructure videoflow deploy needs when the user doesn’t bring their own: an in-cluster NATS JetStream server and, when the blob store is wanted, a Redis. Built as plain dicts (same convention as manifests) so they ship inside the package, parametrize by namespace, and carry an ownership label for selective teardown.

Everything here is dev/test-grade (single replica, emptyDir/no persistence) — a faithful port of k8s/nats.yaml. For production, bring your own broker (the official NATS Helm chart, a managed Redis) and pass --nats/--blob-redis-url.

Ownership rule: a pre-existing nats/redis Service in the namespace is reused as-is and never owned; only components this module applied are returned as “created” and later torn down.

videoflow.deploy.infra.ensure_infra(kubectl: str, namespace: str, need_redis: bool) → tuple

Applies the dev NATS (and, when need_redis, Redis) unless a Service of the same name already exists in the namespace (bring-your-own is reused, not owned).

• Returns:

  • (urls, created) where urls maps nats/redis to in-cluster URLs (redis is None when not needed) and created lists only the components THIS call applied (what teardown may later delete).

videoflow.deploy.infra.ensure_namespace(kubectl: str, namespace: str) → None

videoflow.deploy.infra.infra_urls(namespace: str) → dict

The in-cluster URLs workers use once the dev infra is up.

videoflow.deploy.infra.nats_manifests(namespace: str) → list

Single-replica NATS JetStream (port of k8s/nats.yaml) + infra labels.

videoflow.deploy.infra.redis_manifests(namespace: str) → list

Single-replica Redis for the large-payload blob store. Persistence off: it is transport, not storage. Memory is capped with volatile-lru eviction — every key videoflow writes carries a TTL (PROTOCOL.md BLOB-7), so under pressure Redis evicts our oldest blobs instead of growing until the node OOMs (the redis:7 default is unlimited memory with noeviction). The container limit sits above maxmemory to leave headroom for allocator fragmentation.

videoflow.deploy.infra.service_exists(kubectl: str, namespace: str, name: str) → bool

videoflow.deploy.infra.teardown_infra(kubectl: str, namespace: str, components: List[str]) → None

Deletes the given auto-provisioned components by ownership label. Best-effort (never raises).

videoflow.deploy.infra.wait_infra_ready(kubectl: str, namespace: str, created: List[str], timeout_secs: int = 120) → None

Blocks until each freshly created infra Deployment rolls out; raises on timeout.

videoflow.deploy.localinfra module

Auto-provisioning of the dev broker videoflow run-local needs when the user doesn’t bring their own: a NATS JetStream container and, for the large-payload blob store, a Redis container, both published on localhost. The local analogue of videoflow.infra, which does the same job in-cluster for videoflow deploy.

Ownership rule, identical to infra: if something is already listening on the port — docker compose up -d, a bare nats-server -js, a previous --keep-infra run — it is reused as-is and never torn down. Only containers this module started are returned as “created”, and only those are stopped later.

Dev-grade only: no persistence, no auth, host-published ports. Point --nats / --blob-redis-url at a real broker for anything else.

videoflow.deploy.localinfra.docker_available() → bool

True when a docker daemon is reachable.

videoflow.deploy.localinfra.ensure_local_infra(need_redis: bool = True, nats_url: str | None = None, redis_url: str | None = None) → Tuple[dict, List[str]]

Starts a dev NATS (and, when need_redis, Redis) container unless something is already listening on the port.

• Returns:

  • (urls, created) where urls maps nats/redis to localhost URLs (redis is None when not needed) and created lists only the components THIS call started — the only ones teardown may stop.

• Raises:

  • RuntimeError when a container must be started but docker is unavailable, or when docker run fails.

videoflow.deploy.localinfra.local_infra_urls() → dict

The localhost URLs workers use once the dev containers are up.

videoflow.deploy.localinfra.nats_running(url: str = 'nats://localhost:4222') → bool

videoflow.deploy.localinfra.port_open(host: str, port: int, timeout: float = 1.0) → bool

True when something accepts a TCP connection — the “is it already up?” probe.

videoflow.deploy.localinfra.redis_running(url: str = 'redis://localhost:6379/0') → bool

videoflow.deploy.localinfra.teardown_local_infra(components: List[str] | None) → None

Removes the given auto-started containers. Best-effort; never raises.

videoflow.deploy.localinfra.wait_local_infra_ready(created: List[str] | None, urls: dict, timeout_secs: int = 45) → None

Polls each freshly started component’s port until it accepts connections.

• Raises:

  • RuntimeError (after dumping the container’s logs) on timeout.

videoflow.deploy.manifests module

Renders Kubernetes manifests for a compiled flow (a list of NodeSpec). One workload per node, chosen by flow type:

• BATCH flow -> every node is a Job (each worker exits 0 when its upstream

  end-of-stream drains, so the whole flow runs to completion).

• REALTIME flow -> a finite producer is a Job; every other node is a Deployment

  (or a StatefulSet if partitioned) that stays up until the control-channel stop.

plus a per-node ConfigMap holding the node’s env, a shared ConfigMap for the NATS URL, and a default-deny-except-broker NetworkPolicy.

Manifests are built as plain dicts and serialized with yaml.dump rather than text-templated, so the output is always structurally valid YAML.

The dict/dataclass split here is deliberate: anything that mirrors the Kubernetes API schema (workloads, ConfigMaps, Services, pod specs, …) stays a dict, because we do not own that schema and a typed mirror of it would be a second source of truth that silently drifts from the real API — and would still have to become a dict at the YAML boundary. Records this module owns outright (Mount, ProvisionSplit) are dataclasses, since nothing external constrains their shape.

class videoflow.deploy.manifests.Mount(name: str, host_path: str, container_path: str, read_only: bool)

Bases: object

One host-path mount threaded into every node workload (and into the prepare / compile containers by deploy.build.run_in_image). Produced by parse_mounts; rendered into a hostPath volume + volumeMount by _pod_spec.

• Arguments:

  • name: the volume name, unique within the pod (vf-mount-<i>).

  • host_path: absolute path on the cluster node.

  • container_path: absolute path it appears at inside the container.

  • read_only: whether the container gets it read-only.

container_path: str

host_path: str

name: str

read_only: bool

class videoflow.deploy.manifests.ProvisionSplit(provision: List[dict], worker: List[dict])

Bases: NamedTuple

The two apply phases of a flow’s manifests, in the order they must be applied.

A NamedTuple rather than a plain @dataclass: the phases are genuinely ordered (phase 1 must be applied and awaited before phase 2), and existing callers unpack the result positionally, which keeps working unchanged.

• Arguments:

  • provision: manifests the provision Job needs before any worker publishes.

  • worker: every node’s own ConfigMap/workload/service/PDB.

provision: List[dict]

Alias for field number 0

worker: List[dict]

Alias for field number 1

videoflow.deploy.manifests.delete_resources(kubectl: str, namespace: str, flow_id: str, run_id: str | None = None) → None

Single source of truth for tearing down a flow’s Kubernetes resources: deletes every kind in DELETABLE_KINDS matching the flow-id label, scoped to one run when run_id is given (else every run of the flow). Best-effort (never raises).

videoflow.deploy.manifests.dump_manifests(manifests: List[dict]) → str

Serializes a list of manifest dicts to a single multi-document YAML string.

videoflow.deploy.manifests.flow_spec_configmap(specs: List[NodeSpec], flow_id: str, run_id: str) → dict

ConfigMap holding the compiled specs as JSON, mounted by the provisioning init Job.

videoflow.deploy.manifests.gpu_demand(specs: List[NodeSpec], default_resource: str | None = None) → dict[str, int]

Whole-flow GPU demand per extended-resource name: every replica of a GPU node claims its own gpu_count devices exclusively, so this is the allocatable capacity the cluster must have for the flow to fully schedule. The single source of truth for deploy’s preflight and videoflow explain — the two must never disagree.

Deliberately a dict[str, int] and not a dataclass: the keys are cluster-defined extended-resource names (nvidia.com/gpu, amd.com/gpu, a MIG profile, a vendor’s own), discovered from the specs at runtime. There is no fixed field set to name, and every caller iterates it against the cluster’s equally open-ended allocatable map, so this is a mapping in substance, not a record wearing a dict’s clothes.

• Arguments:

  • specs: the compiled flow. Non-GPU nodes contribute nothing.

  • default_resource: extended-resource name for GPU nodes that don’t name their own (--gpu-resource-name). Defaults to nvidia.com/gpu.

• Returns:

  • resource name -> total devices the flow needs at full scale. Empty when no node requests a GPU.

videoflow.deploy.manifests.headless_service(spec: NodeSpec, flow_id: str) → dict

Headless Service backing a partitioned node’s StatefulSet (required for stable pod network identity/ordinals).

videoflow.deploy.manifests.k8s_name(*parts: object) → str

Joins parts into a DNS-1123-safe Kubernetes resource name.

videoflow.deploy.manifests.nats_configmap(flow_id: str, nats_url: str, blob_redis_url: str | None = None, blob_ttl_seconds: int | None = None) → dict

videoflow.deploy.manifests.network_policy(flow_id: str) → dict

Allows worker pods to talk to each other and out to the broker/DNS, and denies everything else ingress by default. Kept intentionally permissive on egress since the NATS/Redis services may live in another namespace.

videoflow.deploy.manifests.node_configmap(spec: NodeSpec, flow_id: str, flow_type: str, run_id: str, envelope_version: int) → dict

videoflow.deploy.manifests.parse_mounts(values: List[str] | None) → List[Mount]

Parses repeatable --mount specs into Mount records consumed by _pod_spec.

• Arguments:

  • values: list of /host/path:/container/path[:ro] strings. The single-path shorthand /path[:ro] mounts the same absolute path on both sides (what a flow compiled against local files needs, since the paths baked into node params must resolve identically in the pods).

• Returns:

  • a list of Mount, in argument order. name is only unique within one call, so a concatenation of two parse_mounts results repeats names — fine for run_in_image (docker -v ignores them), which is the only caller that concatenates.

• Raises:

  • ValueError on a relative path or a malformed suffix.

videoflow.deploy.manifests.pod_disruption_budget(spec: NodeSpec, flow_id: str) → dict

Keeps at least one replica of a multi-replica node available during voluntary disruptions (node drains, upgrades).

videoflow.deploy.manifests.provision_init_job(flow_id: str, run_id: str, flow_type: str, image: str, nats_cm_name: str, image_pull_policy: str = 'IfNotPresent') → dict

A one-shot Job that runs videoflow.provision to create all streams/durables before workers start (required so BATCH interest-retention streams don’t drop early messages). Runs on image — any worker image has the framework + broker client installed.

Its pull policy must match the workers’: this Job runs first, so when it is the one that cannot pull, the flow fails before any worker starts and the only symptom is the provision-wait timeout.

videoflow.deploy.manifests.render_manifests(specs: List[NodeSpec], flow_id: str, flow_type: str, nats_url: str, run_id: 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, autoscaling: bool = False, max_replicas: int = 10, nats_monitoring_endpoint: str | None = None, envelope_version: int | None = None, provision_image: 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') → list

Returns a list of manifest dicts for the whole flow. The caller decides whether to yaml.dump them to files (CLI) or apply them via the API (engine).

• run_id: per-run identifier stamped into each node’s env (scopes broker streams).

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

• default_image: image ref used for any node that didn’t declare its own (--image).

• image_overrides: mapping of node name to image ref (--image-override).

• autoscaling: if True, emit a KEDA ScaledObject per processor node.

• max_replicas: upper bound for autoscaled processors.

• nats_monitoring_endpoint: NATS monitoring host:port for KEDA (defaults to nats.<namespace>.svc:8222).

• envelope_version: wire version for the whole run. The only supported version is 4 (the language-neutral protobuf wire); defaults to the ambient DEFAULT_ENVELOPE_VERSION, and an explicit incompatible pin is rejected.

• provision_image: image the one-shot provision Job runs on — must have videoflow + the broker client (i.e. a Python image), which a vendor default_image may not be. Defaults to default_image; set it explicitly (--provision-image) for flows whose default image is a non-Python vendor image.

• mounts: Mount records from parse_mounts — each becomes a hostPath volume + volumeMount on every node workload (not the provision Job, which never touches node data).

• gpu_runtime_class: runtimeClassName to put on GPU pods (--gpu-runtime-class). Needed where the NVIDIA container runtime is registered as an opt-in RuntimeClass rather than the node default — on k3s, nvidia. Without it such a pod schedules onto a GPU node and then runs with no device visible.

• gpu_mode: 'exclusive' (default — each GPU replica claims whole devices via the extended resource) or 'shared' (no resource limit; all GPU pods co-schedule onto the pool and share the physical GPUs through the NVIDIA runtime — LocalProcessEngine semantics, dev clusters only).

• gpu_resource_name: deploy-level default extended-resource name for GPU claims (--gpu-resource-name); a node’s own gpu_resource_name wins. Defaults to nvidia.com/gpu.

• gpu_autoscaling: emit KEDA ScaledObjects for GPU nodes too. Off by default: each autoscaled replica claims its own GPUs, so scaling to max_replicas can demand more devices than the cluster has and strand pods Pending.

• image_pull_policy: imagePullPolicy for every rendered container, workers and the provision Job alike (--image-pull-policy). Defaults to IfNotPresent, which is what makes a locally built-and-loaded image actually run — see DEFAULT_IMAGE_PULL_POLICY.

• Raises:

  • ValueError if a node has no resolvable image (see videoflow.deploy.images), if image_pull_policy is not a valid k8s policy, or if the wire settings are incompatible with the flow’s components.

videoflow.deploy.manifests.scaled_object(spec: NodeSpec, flow_id: str, run_id: str, nats_monitoring_endpoint: str, max_replicas: int) → dict | None

A KEDA ScaledObject that scales a processor Deployment on NATS JetStream consumer lag. nb_tasks becomes the floor (minReplicaCount); KEDA scales up toward max_replicas when the node’s input stream backs up. Returns None for node kinds that aren’t autoscaled (producers, consumers).

Requires KEDA installed in the cluster and the NATS monitoring endpoint (port 8222) reachable at nats_monitoring_endpoint.

videoflow.deploy.manifests.split_provision_manifests(manifests: List[dict], flow_id: str) → ProvisionSplit

Partitions rendered manifests into a ProvisionSplit so the caller can apply provisioning first and wait for it before starting workers.

The provision phase is everything the provision Job needs to create the broker streams/durables (including its EOS interest anchors) before any worker publishes: the broker + specs ConfigMaps, the NetworkPolicy, and the provision Job itself. The worker phase is every node’s ConfigMap/workload/service/PDB.

• Arguments:

  • manifests: the full rendered manifest list from render_manifests.

  • flow_id: the flow id the provision-phase resource names are derived from.

• Returns:

  • a ProvisionSplit(provision, worker), each preserving input order.

videoflow.deploy.manifests.workload(spec: NodeSpec, flow_id: str, flow_type: str, image: str, nats_cm_name: str, mounts: List[Mount] | None = None, gpu_runtime_class: str | None = None, gpu_mode: str = 'exclusive', gpu_resource_name: str | None = None, image_pull_policy: str = 'IfNotPresent') → dict

videoflow.deploy.solution module

Solution conventions for one-command deploys. A “solution” is a graph module shipped with optional sibling files:

• config.template.yaml — a valid config plus two extension blocks stripped on write: x-questions (inputs deploy asks for interactively when no config exists) and x-mounts (paths from the resolved config that must be hostPath-mounted into prep containers and worker pods).

• prepare.py — an idempotent prep hook deploy runs inside the solution image before compiling (its outputs get baked into the compiled specs).

x-questions entries: {key, prompt, type, default, choices, item_key, item_value}. key is a dotted path into the config (digit segments index int-keyed maps). Types: str (default), int, float, choice (with choices), path (one filesystem path, validated and absolutized), and paths (comma-separated paths expanded into a mapping via item_key, e.g. 'cam{i}', and item_value, e.g. {video: '{path}'}). That set is extensible: see register_question_type.

x-mounts entries are path templates: '{cameras.*.video}:ro' (dotted lookup into the resolved config, * fans out), '{work_dir}', '~/.videoflow:/root/.videoflow'. A single path resolves to a same-path hostPath mount (host and container see the same absolute path — required because the paths baked into node params at compile time must resolve identically in the pods); a host:container pair maps them explicitly (e.g. the operator’s home caches onto the container root’s).

videoflow.deploy.solution.ask_questions(questions: ~typing.List[dict], base_dir: str, input_fn: ~typing.Callable[[str], str] = <built-in function input>) → dict

Prompts for each x-question on the terminal; returns {dotted_key: typed_value}.

• Raises:

  • ValueError: a question names an unregistered type (checked before prompting — see validate_question_types).

videoflow.deploy.solution.ensure_config(graph_dir: str, config_arg: str | None = None, interactive: bool = True, input_fn: ~typing.Callable[[str], str] = <built-in function input>) → str | None

The config file deploy should use: an explicit --config, an existing config.yaml next to the graph, or — when the solution ships a template — one generated by asking the template’s questions. None when the solution has no config convention at all.

• Raises:

  • SystemExit listing every question when a config must be generated but the session is non-interactive.

videoflow.deploy.solution.find_prepare(graph_dir: str) → str | None

The solution’s prepare.py hook, or None when it ships none.

videoflow.deploy.solution.find_template(graph_dir: str) → str | None

videoflow.deploy.solution.load_template(template_path: str) → dict

videoflow.deploy.solution.prepare_command(config_path: str | None = None, python_exe: str = 'python') → List[str]

The argv for the prepare hook, run with the solution directory as cwd.

videoflow.deploy.solution.register_question_type(qtype: str, coercer: Callable[[dict, str, str], Any]) → None

Registers a coercer for an x-questions type, called as coercer(question, answer, base_dir) where question is the raw x-questions entry (so a coercer can read its own extra keys, as choice reads choices), answer is the operator’s raw string, and base_dir is the solution directory that relative paths resolve against.

A coercer signals a bad answer by raising ValueError; the prompt loop catches it, shows the message, and re-asks.

• Arguments:

  • qtype: the type string used in config.template.yaml.

  • coercer: callable returning the coerced config value.

videoflow.deploy.solution.registered_question_types() → list

The x-questions type names currently registered, sorted.

videoflow.deploy.solution.render_config(template: dict, answers: dict) → str

The final config YAML: template body with answers applied and x-* blocks stripped.

videoflow.deploy.solution.resolve_mounts(template: dict | None, config: dict, graph_dir: str) → List[str]

Expands the template’s x-mounts against the resolved config into single-path mount specs (/abs/path[:ro]) for manifests.parse_mounts. Relative config values resolve against graph_dir (matching a prep/compile container whose workdir is the graph dir); ~ expands.

videoflow.deploy.solution.run_prepare_local(graph_dir: str, config_path: str | None = None) → bool

Runs the solution’s prepare.py on this host, with the solution directory as the working directory so its import common resolves. Used by run-local (whose workers are local anyway) and as deploy’s fallback when there is no image to run it in.

• Returns:

  • False when the solution ships no hook (nothing was run).

• Raises:

  • subprocess.CalledProcessError when the hook exits non-zero.

videoflow.deploy.solution.validate_question_types(questions: list | None) → None

Checks every question’s type is registered, before any prompting starts.

Done up front deliberately: the prompt loop treats ValueError as a bad answer and re-asks, so an unregistered type discovered mid-loop would re-prompt forever over something the operator cannot fix by typing.

• Raises:

  • ValueError: a question names an unregistered type; the message names the offending key, the known types, and how to add one.