videoflow.runtime package

Submodules

videoflow.runtime.health module

A tiny stdlib HTTP server exposing Kubernetes health probes and Prometheus metrics for a running worker, plus an InstrumentedMessenger that feeds it.

Endpoints (default port 8080):

/readyz

200 once the node has started processing (see readiness note below), else 503

/healthz

200 while the run loop is beating, 503 if it has stalled

/metrics

Prometheus text exposition of per-node processing metrics

Kept dependency-free (no prometheus_client) so the base image stays lean; the metrics text format is simple enough to emit by hand.

class videoflow.runtime.health.HealthServer(state: HealthState, port: int = 8080)

Bases: object

start() → None

stop() → None

class videoflow.runtime.health.HealthState(node_name: str)

Bases: object

Thread-safe holder for readiness/liveness/metrics, shared between the run loop (via the messenger) and the HTTP handler.

beat() → None

incr(counter: str, amount: int = 1) → None

is_live() → bool

is_ready() → bool

mark_ready() → None

observe(metric: str, value: float | None) → None

render_metrics() → str

class videoflow.runtime.health.InstrumentedMessenger(inner: Messenger, state: HealthState)

Bases: Messenger

Wraps a real Messenger and updates a HealthState as messages flow, without any change to the Task classes. Readiness note: the node is marked ready on its first messenger activity (first send or receive), which happens only after node.open() has returned inside NodeTask.run() — so a slow model-loading open() correctly keeps the pod un-ready until it finishes.

ack_inputs() → None

Acknowledge the input group last returned by receive_message — called by the task only after the node has processed it (and, for a processor, published its output). This ack-after-process ordering is what makes a crash mid-processing safe: the un-acked message is redelivered instead of lost.

check_for_termination() → bool

Returns true if a flow-wide termination signal has been received on the control channel. Used by videoflow.core.task.ProducerTask to stop pulling new input even before it naturally reaches StopIteration.

close() → None

Release any broker resources held by the messenger. Default: no-op.

fail_inputs(exc: BaseException) → None

Report that the node raised while processing the last input group. The messenger decides whether to redeliver (BATCH, up to a retry limit, then dead-letter) or drop (REALTIME).

last_input_info() → dict | None

Per-parent envelope info (event_ts, metadata, trace_id, seq) for the input group last returned by receive_message (exposed to nodes as ctx.input_info). Default: None.

last_input_key() → str | None

A stable identity for the input group last returned by receive_message, used as a sink idempotency key. Default: None (no idempotency).

publish_message(message: Any, metadata: dict | None = None) → None

Publishes this node’s own output message. Depending on the flow’s configured retention policy (REALTIME vs BATCH), this may drop the message if downstream consumers are behind.

publish_stop_signal() → None

Publishes a termination marker on this node’s own subject. Unlike publish_message, this is never dropped regardless of retention policy — every downstream consumer must observe it exactly once.

receive_message() → dict

Blocks until this node has received a complete input: one message from every real parent, all derived from the same upstream originating event (see the trace_id propagation scheme in the concrete implementation) — or until every parent has signaled termination.

• Returns:

  • a dict {parent_name: {"message": ..., "metadata": ..., "is_stop_signal": bool}} with exactly one entry per real parent of this node.

set_output_event_timestamp(value: float) → None

Set the event time (epoch seconds) stamped on this node’s next published output (via RuntimeContext.set_event_timestamp) — when the underlying real-world event was captured. Producers of time-sensitive data (cameras, sensors) should set this; downstream nodes inherit their input group’s event time automatically. Default: no-op.

set_output_partition_key(value: Any) → None

Set the partition key attached to this node’s next published output (via RuntimeContext.set_partition_key), so a downstream partitioned node can route by a business key. Default: no-op.

videoflow.runtime.idempotency module

Optional sink-side idempotency: dedupe the effects of a consumer across message redelivery/restart, giving “exactly-once-ish” side effects on top of the broker’s at-least-once delivery. A consumer opts in with ConsumerNode(idempotent=True) and the flow must be given a Redis URL (reuses the blob-store Redis).

Consumers are single sinks (not replicated), so a plain check-then-mark against a shared store is race-free.

class videoflow.runtime.idempotency.IdempotencyStore

Bases: object

mark(key: str) → None

seen(key: str) → bool

class videoflow.runtime.idempotency.RedisIdempotencyStore(url: str, ttl_seconds: int = 86400)

Bases: IdempotencyStore

mark(key: str) → None

seen(key: str) → bool

videoflow.runtime.idempotency.idempotency_key(flow_id: str, node_name: str, message_id: str) → str

videoflow.runtime.logging_config module

Logging configuration for workers. Opt into JSON structured logs (one object per line, easy to ship to a log aggregator) by setting VF_STRUCTURED_LOGS=1; otherwise a plain human-readable format is used. Node-scoped fields (flow/run/node/ replica/trace ids) are included when a log record carries them as extra=....

class videoflow.runtime.logging_config.JsonFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)

Bases: Formatter

format(record: LogRecord) → str

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

videoflow.runtime.logging_config.configure_logging() → None

videoflow.runtime.provision module

Provisioning entrypoint (python -m videoflow.provision): creates every stream and durable consumer a flow needs, before its workers start. Run as a one-shot Kubernetes init Job so that BATCH interest-retention streams have their consumers registered before any message is published (otherwise early messages are dropped).

Driven by environment variables:

VF_NATS_URL nats://host:port VF_FLOW_ID stable flow id VF_RUN_ID per-run id VF_FLOW_TYPE realtime | batch VF_MAX_RETRIES optional; BATCH redelivery attempts (default 3) VF_FLOW_SPECS_JSON the compiled NodeSpecs as a JSON list, OR VF_FLOW_SPECS_PATH path to a file with that JSON (default /etc/videoflow/specs.json)

videoflow.runtime.provision.main() → None

videoflow.runtime.worker module

The single process entrypoint that runs exactly one graph node — used identically whether the node is launched as a local subprocess (LocalProcessEngine) or as a Kubernetes pod. It’s fully driven by environment variables so it needs no access to the original graph-building script.

VF_NODE_CLASS fully-qualified class, e.g. videoflow.processors.basic.IdentityProcessor VF_NODE_PARAMS_JSON JSON dict of constructor kwargs (from NodeSpec.params) VF_NODE_KIND producer | processor | consumer. Must match the family of

VF_NODE_CLASS; the compiler writes both together, and the worker rejects a disagreement up front (see require_node_kind).

VF_NODE_NAME this node’s stable name VF_PARENT_NAMES comma-separated parent node names (’’ if none) VF_HAS_CHILDREN ‘1’ or ‘0’ VF_NATS_URL nats://host:port VF_FLOW_ID shared flow identifier (stable across runs) VF_FLOW_TYPE realtime | batch VF_RUN_ID per-run identifier that scopes this run’s broker streams VF_REPLICA_ID index of this replica (0 for single-task nodes) VF_ACK_WAIT_SECONDS optional; per-message ack deadline (default 60) VF_MAX_RETRIES optional; BATCH redelivery attempts before dead-letter (default 3) VF_EOS_QUIESCENCE_MS optional; drain quiescence window before honoring EOS (default 500) VF_NB_TASKS optional; replica count of this node (for partition ownership) VF_PARTITION_BY optional; partition key (‘trace_id’ or a metadata field) VF_JOIN_POLICY_JSON optional; JSON JoinPolicy for a multi-parent node VF_BLOB_REDIS_URL optional; enables the external blob store for large payloads.

The store is chosen by the URL’s scheme (redis:// and rediss:// built in; others via register_blob_store), so the name is historical rather than a restriction to Redis.

VF_BLOB_READERS optional; how many downstream reads each message this node

publishes receives — enables refcounted blob reclamation (PROTOCOL.md BLOB-5). Unset ⇒ blobs are TTL-only.

VF_BLOB_TTL_SECONDS optional; TTL for offloaded payloads (PROTOCOL.md BLOB-7).

Unset ⇒ flow-type default (3600 realtime / 86400 batch).

VF_ENVELOPE_VERSION optional; wire envelope version to emit (only 4, protobuf)

videoflow.runtime.worker.build_node_from_env() → Node

videoflow.runtime.worker.main() → None

videoflow.runtime.worker.require_node_kind(node: Node, expected: Type[_N], kind: str) → _N

Check that the node built from VF_NODE_CLASS matches the node family declared by VF_NODE_KIND, and return it narrowed to that family.

The two env vars are written together by the compiler, so a mismatch means the ConfigMap was hand-edited or the image is stale. Without this check the wrong Task type is constructed and the disagreement only surfaces later, deep in the run loop, as an opaque AttributeError (e.g. calling next() on a consumer).

• Raises:

  • ValueError: if node is not an instance of expected.

videoflow.runtime.worker.run_from_env() → None