feat: multi-source pipeline instances — streaming + batch (PLAT-1141)

[lucasmundim]

Closes PLAT-1141, the controller half of PLAT-1071.

Summary

• Adds PipelineInstance.Spec.Sources []NamedSourceRef so a single PipelineInstance can bind N PipelineSources to N specific filter containers — the controller injects source-derived env vars into only the matching container, enabling multi-camera pipelines in one Pod.
• Keeps the singular Spec.SourceRef (now *SourceReference) as a first-class convenience form for single-source pipelines — broadcast semantics, no filter names needed, supported indefinitely (un-deprecated during review). An admission-time CEL rule enforces exactly-one-of; EffectiveSources() normalizes both shapes into one reconciler path.
• Streaming: full multi-source. Per-binding RTSP_URL (and _RTSP_USERNAME / _RTSP_PASSWORD when present) injected into the matching videoin-<filterName> container.
• Batch: full multi-source. Each binding's Bucket.Prefix is treated as the full object key (how single-file medias ship from the platform — no separate Object field; an earlier draft added one, removed before merge since nothing produces it); the reconciler emits one Pod with N init claimer containers in direct mode (skip Valkey, download exactly that key to /ws/<filterName><ext>) and per-VideoIn VIDEO_INPUT_PATH injection — ordered before the filter's config env so K8s dependent-env expansion resolves file://$(VIDEO_INPUT_PATH). Single-source batch keeps today's queue-based parallel-files-per-bucket flow bit-for-bit.
• Single-binding direct mode (batch image fix): new BucketSource.singleObject field — when a single named binding's bucket declares its prefix is a full object key (the agent sets this for single-file media kinds), the instance runs in the same direct-download path as multi-source: one claimer with S3_OBJECT_KEY, extension-preserving /ws/<filterName><ext> input path, no Valkey. Fixes single-source batch image pipelines (queue mode's fixed input.mp4 destination destroyed the extension; image-in silently found zero images and the Job hung forever — found empirically). Hand-authored CRs without the flag and fileset prefixes keep queue mode; the legacy broadcast sourceRef is never routed direct.
• Bucket-listing hygiene: zero-byte directory-placeholder objects (GCS/S3 console "folders", keys ending in /) are skipped at listing time — previously one was enqueued as a queue-mode work item, failed its download/decode attempts, hit the DLQ, and failed the whole run (observed empirically on an external_videos fileset over a console-created GCS folder).
• Operability (review hardening): PipelineSource create/update/delete now triggers reconciliation of every PipelineInstance that references it (watch + map function, cluster-wide — cross-namespace sourceRefs wake their instances too), so an instance applied before its sources self-heals; unmatched binding filterNames surface as Degraded/SourceBindingUnmatched and clear automatically once bindings validate; batch validation failures surface as Degraded/MultiSourceBatchInvalidSource / MultiSourceBatchMissingObject with operator-actionable messages, also cleared on recovery. Validation failures additionally set Progressing=False (consistent condition set for automation) and propagate status-write errors (conflict → requeue, else retry via returned error) instead of swallowing them on non-requeuing paths; the steady-state batch loop skips no-op status writes; claimer container names truncate safely at 63 chars (no trailing hyphen).

Validation

Unit tests

• make test green; make lint 0 issues.
• pipelineinstance_multisource_test.go pins both paths: streaming per-container RTSP_URL routing, singular-form broadcast preserved, EffectiveSources() normalization, multi-source batch per-binding direct-mode claimer + per-VideoIn env injection.
• CRD admission envtest (pipelineinstance_admission_test.go, 16 specs against a real API server): the sourceRef/sources CEL xor (both, neither, empty list), listMapKey uniqueness on sources[].filterName (duplicate filter rejected; two filters sharing one source accepted), and the DNS-1123 filterName pattern (9 accept/reject cases) — pinning the admission contract the agent (plainsight-deployment-agent#65) and portal (client-portal#429) build on.
• TestPerFilterInputPath pins the claimer↔filter path contract (extension preservation, .mp4 default, edge cases).

End-to-end on docker-desktop K8s

• Admission rule: rejects "neither" and "both" SourceRef/Sources configurations, accepts each individually (now also pinned by envtest, above).
• Streaming with two real RTSP streams (vod2stream-car-truck-person + vod2stream): 3/3 containers Running, 0 restarts; front-cam @ 24.0 fps, back-cam @ 23.976 fps; webvis /data returns the multi-topic JSON shape ({front_cam: {meta: {src: rtsp://A}}, back_cam: {meta: {src: rtsp://B}}}).
• Streaming with a real multi-topic filter (filter-aggregator iterating frames.items()): consumed both topics cleanly ({frame_count: 78, sources: 2}).
• Streaming with platform-exported YAML (plainsight-api#861 export, subscriber-side topic remap): pod 3/3, both remapped topics (front_cam_main / back_cam_main) at full frame rate — validates the API↔controller contract end-to-end.
• Batch single-source via Sources[]: binding resolved, reconciler proceeds to bucket listing (fails on fake bucket as designed).
• Batch multi-source: Job created with 2 init claimers named claimer-front-cam and claimer-back-cam, each carrying its specific S3_OBJECT_KEY (front.mp4 / back.mp4) + VIDEO_INPUT_PATH (/ws/front-cam.mp4 / /ws/back-cam.mp4), no Valkey env on either; front-cam and back-cam filter containers each got VIDEO_INPUT_PATH; downstream image-out container has none; Job topology parallelism=1, completions=1.

Post-change verification (2026-06-12, all four PRs live, fresh CRDs/agent/claimer): single-source image → direct mode, Job 1/1 Complete in 11s (/ws/image-in.png, extension preserved — previously hung forever); single-source video → direct mode (Added direct-mode claimer … bindings: 1 in the log), completed; multi-source video (2 files, fan-in) completed in 42s; external_videos fileset → queue mode engaged (totalFiles: 2 after the placeholder fix), 2/2 completions, completed; 2-camera stream regression: 3/3 Running, both topics live.

Cross-PR smoke (2026-06-11, all four train PRs live: API #861 → agent #65 → this controller, agent-authored CRs)

• Found and fixed a real integration bug: the agent applies the PipelineInstance before its PipelineSources (intentional — shared-source cleanup TOCTOU), so the first reconcile set Degraded/PipelineSourceNotFound; the PipelineSource watch then self-healed the deployment but the stale condition was never cleared, and the agent's provisioning monitor tore the instance down on it. Now cleared on the same pass as SourceBindingUnmatched; the two-phase race is pinned by TestReconcile_PipelineSourceNotFound_ClearsOnceSourceExists.
• Combination matrix, all green after the fix: single source (broadcast-equivalent single binding, 2/2 Running); three sources (4/4 Running, ports 5550/5552/5554, viewer consuming cam_a_main/cam_b_main/cam_c_main all live with correct per-camera meta.src); same media bound to two VideoIns (one shared ps-<media> CR, both bindings resolve to it, 3/3 Running); concurrent instances sharing a media (stopping one swept only its own sources — the survivor's shared source and pod were untouched, three separate times).

Deploy note — claimer image

Direct-download mode needs the claimer built from this PR. plainsightai/openfilter-pipelines-claimer:latest (the controller's default CLAIMER_IMAGE) only refreshes on v* release tags — cut a release after merging (or pin CLAIMER_IMAGE to a sha tag) before relying on multi-source or singleObject batch, or the claimer crashes with STREAM is required (verified empirically against the current latest).

Notes for pipeline authors (also documented in samples + DESIGN_DOCUMENT)

OpenFilter has two authoring constraints that bite every multi-source streaming pipeline. Both surfaced during the smoke test and would otherwise be silent runtime crashes:

1. Stagger output ports by 2. OpenFilter binds the configured port AND port+1 per outputs (the +1 carries ZMQ control plane). Use 5550, 5552, 5554, … — never 5550, 5551 (would crash the second VideoIn with Address already in use).
2. Disambiguate fan-in topics — two equivalent schemes (both documented in the samples): publisher-side ;topic tags on each VideoIn's sources (convention for hand-authoring), or subscriber-side remap ;main><name>_main at the fan-in filter (what the plainsight-api export emits — validated end-to-end). Without either, both VideoIns publish on the default topic main and any downstream fan-in crashes with RuntimeError: duplicate topic 'main'.

Neither is enforced by the controller — they live in the filter.config strings. The plainsight-api export (PLAT-1140, plainsight-api#861) bakes both in at export time; hand-authored CRs follow the samples.

For multi-source batch: every binding's PipelineSource must be a Bucket (RTSP makes no sense for batch) and its prefix must be a full object key. The reconciler surfaces violations as MultiSourceBatchInvalidSource / MultiSourceBatchMissingObject Degraded conditions with operator-actionable messages, and clears them once fixed.

Test plan

• make test green (including 16-spec admission envtest)
• make lint green (0 issues)
• CRD admission rule verified live (kubectl apply) and pinned by envtest
• Streaming multi-source: two real RTSP streams running concurrently in one Pod, isolated per topic
• Streaming multi-source with platform-exported YAML (API #861 contract)
• Batch single-source via Sources[]: routes through the new ResolvedSourceBinding path
• Batch multi-source: Job created with per-binding direct-mode claimers + per-container VIDEO_INPUT_PATH injection

[lucasmundim]

Empirical validation against the real multi-camera filter chain

Reviewer asked about validation against the premium multi-camera filters. Tested filter-rt-detr:0.1.2 (the upstream stage of the full reid + global-association multi-camera chain) end-to-end on docker-desktop K8s in CPU mode.

Pipeline shape:

front-cam (RTSP A) -> rt-detr-front (CPU) -> webvis (front_cam topic)
back-cam  (RTSP B) -> rt-detr-back  (CPU) -> webvis (back_cam topic)

Two distinct RTSP test streams bound via the new Spec.Sources shape, each rt-detr running independent CPU inference with its own FILTER_CAMERA_ID.

Result: 5/5 containers Running, both cameras producing distinct content-appropriate detections.

webvis /data aggregated output:

Topic	Bound RTSP source	Detections	Labels found
front_cam	vod2stream-car-truck-person/car-truck-person	35	bus, car, cell phone, person, truck
back_cam	vod2stream/stream	12	cell phone, person

The traffic video gets vehicle labels; the people video gets person labels. That's proof — not inference — that each filter-rt-detr container received only its bound camera's stream and is keeping independent per-camera detection state (each track carried its own local_track_id).

What this rules out: binding swaps, broadcast leaks, cross-camera state contamination — all of which would show up as identical detection sets on both topics or identical src URIs in the metadata.

Caveat (called out for transparency): CPU inference latency was ~100s/frame on docker-desktop. That's the price of running rt-detr without GPU; the controller's responsibility (per-container env injection, downstream wiring through openfilter's ZMQ topology) is unaffected. Real deployments on a GPU cluster get real-time inference. Adding the downstream filter-tracking-reid and filter-global-association-multicamera-tracking stages to this validated chain is pipeline authoring — same controller plumbing, more containers.

[Copilot]

Pull request overview

This PR extends the controller to support multi-source PipelineInstances (streaming + batch) by introducing per-filter source bindings while preserving the legacy single-source convenience form.

Changes:

• Adds spec.sources[] (list-map keyed by filterName) plus EffectiveSources() normalization, and updates CRD admission (CEL xor rule) to enforce exactly-one-of sourceRef/sources.
• Implements controller-side multi-source reconciliation: per-container RTSP env injection for streaming, and a new single-pod, multi-init-claimer “direct download” Job path for multi-source batch.
• Adds a new claimer “direct mode” (S3_OBJECT_KEY) and broad test/docs/sample updates to pin the contract.

Reviewed changes

Copilot reviewed 20 out of 21 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
internal/controller/pod_labels_test.go	Updates streaming deployment builder call signature for resolved source bindings.
internal/controller/pipelineinstance_streaming_update_test.go	Updates streaming update tests for SourceRef pointer + bindings slice.
internal/controller/pipelineinstance_multisource_test.go	Adds comprehensive multi-source unit tests (streaming env routing, batch job shape, binding validation, watch mapping, etc.).
internal/controller/pipelineinstance_controller.go	Introduces ResolvedSourceBinding, binding resolution, unmatched-binding validation, Degraded clearing improvements, and a PipelineSource watch+mapper.
internal/controller/pipelineinstance_controller_test.go	Updates controller integration tests for SourceRef pointer and validates VIDEO_INPUT_PATH ordering.
internal/controller/pipelineinstance_controller_streaming.go	Refactors streaming path to accept resolved bindings and inject RTSP env per-container; anchors idle-timeout to first binding.
internal/controller/pipelineinstance_controller_reconcile_span_test.go	Updates reconcile-span tests to pass bindings slice and pointer SourceRef.
internal/controller/pipelineinstance_controller_batch.go	Routes multi-source batch to new reconciler; fixes env ordering by injecting VIDEO_INPUT_PATH before FILTER_*.
internal/controller/pipelineinstance_controller_batch_multisource.go	Adds new multi-source batch reconciler + job builder (N init claimers, direct mode, per-VideoIn VIDEO_INPUT_PATH).
internal/controller/pipelineinstance_controller_batch_multisource_test.go	Adds state-machine coverage for multi-source batch reconciler and helper behavior (naming, paths).
internal/controller/pipelineinstance_admission_test.go	Adds envtest coverage pinning CRD admission validation (xor rule, listMapKey uniqueness, filterName pattern).
DESIGN_DOCUMENT.md	Documents the new multi-source contract (streaming + batch constraints and behavior).
deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelineinstances.yaml	Updates rendered CRD schema/docs: sources, sourceRef description, CEL xor validation.
config/samples/pipelines_v1alpha1_pipelineinstance_stream_multisource.yaml	Adds sample for multi-source streaming authoring + constraints.
config/samples/pipelines_v1alpha1_pipelineinstance_batch_multisource.yaml	Adds sample for multi-source batch direct-mode behavior + constraints.
config/samples/kustomization.yaml	Includes new sample manifests.
config/crd/bases/filter.plainsight.ai_pipelineinstances.yaml	Updates CRD base with sources and CEL xor validation.
cmd/claimer/main.go	Adds direct-download mode via S3_OBJECT_KEY and relaxes STREAM/GROUP requirements in that mode.
cmd/claimer/main_test.go	Adds tests covering direct-mode config loading and validation behavior.
api/v1alpha1/zz_generated.deepcopy.go	Updates deep-copies for new types and SourceRef pointer + Sources slice.
api/v1alpha1/pipelineinstance_types.go	Adds sources, makes sourceRef optional pointer, adds NamedSourceRef and EffectiveSources(), and updates validation/docs.

Files not reviewed (1)

• api/v1alpha1/zz_generated.deepcopy.go: Generated file

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[leandrobmarinho]

Reviewed the controller-side multi-source support in depth. The binding->source mapping is correct: streaming splits bindings into a per-filter map + broadcast slice and injects each container's matching RTSP source (env ordered before FILTER_* so dependent-env expands); multi-source batch emits one claimer per binding writing /ws/ and injects VIDEO_INPUT_PATH into only the matching VideoIn; an unmatched filterName is caught up-front as Degraded, not silently dropped. Single-source backward-compat is preserved (sourceRef normalizes to a broadcast binding; CRD XValidation enforces exactly-one-of sourceRef/sources; deepcopy + the config/crd vs Helm CRD copies are in sync). Reconcile is idempotent (Get-then-Create/Patch), conditions use SetStatusCondition so no LastTransitionTime churn, and RBAC already covers pipelinesources. CI green. Looks good.

[shingonoide]

Reviewed at f54935f. This is a clean, well-tested implementation of the multi-source controller half. The condition lifecycle work stands out: the self-healing Degraded clears, the PipelineSource/Pipeline watch map-functions, and the apply-order race fix (instance applied before its sources, then stale Degraded cleared on recovery) are exactly the kind of integration corners that bite in production, and they are pinned by tests. Status-update conflict handling that requeues with fresh state rather than continuing on a mutated Conditions slice is a nice touch. CI is green.

A few non-blocking notes:

1. The ticket drafted a FILTER_SOURCES + deterministic-port (5550+index) injection scheme in the controller; the implementation instead injects per-container RTSP_URL (streaming) and VIDEO_INPUT_PATH (batch) and leaves topic/port fan-in to the filter sources: config. The PR body explains and validates this pivot with the 3-camera run, so this is just a request to keep that note discoverable (a one-liner in the design doc noting the controller no longer owns the port scheme would save the next reader the cross-check).

2. filterName is constrained to DNS-1123 labels (lowercase alphanumeric + hyphen, no underscore). That is the right constraint since the name flows into container/volume/env names, but it is stricter than the original ticket text. Please confirm the deployment-agent (#65) and portal (#429) validate filterName identically so they never emit a name the CRD rejects at admission.

3. Minor: a single-entry sources[] on a batch instance takes the legacy queue path and the binding's filterName is not used for routing (broadcast-equivalent). This is the intended no-regression behavior for one source and the unmatched-name validation still runs, so no change needed; a one-line comment noting it would help future readers.

Nothing here blocks. Nice work on the operability hardening.

[jmiller-plainsight]

Code Review: Multi-Source Pipeline Instances (PLAT-1141)

Overall, this is an exceptionally high-quality, clean, and thoroughly tested PR. The API design, normalizations, and watchers are beautifully engineered. I have some minor suggestions regarding propagating status update errors on validation failures to ensure the degraded status is always persisted properly, and a small documentation note.

[lucasmundim]

Thanks for the thorough pass — all three notes addressed or answered:

1. Port-scheme ownership note: added to DESIGN_DOCUMENT.md in c7310c9 — a paragraph under the authoring-constraints section records the pivot from the ticket's controller-injected FILTER_SOURCES/port scheme to authoring-time wiring (API export or hand-authored), with the controller owning only per-binding env injection.

2. filterName validation alignment across repos: confirmed identical end-to-end. The API validates instanceName (which becomes filterName) against the same DNS-1123 regex fail-closed at instance create (validation.InstanceNameRegex, plainsight-api#861), the portal enforces it in the editor with the same pattern (INSTANCE_NAME_PATTERN, client-portal#429), and the agent passes through only API-validated names — the CRD pattern is the backstop, and the admission envtest in this PR pins it.

3. Single-entry sources[] comment: added on reconcileBatch in c7310c9 (same as jmiller's note). Also worth flagging since your review snapshot at f54935f: single named bindings with singleObject: true buckets now take the direct path (a5697a5) — the queue path and its broadcast-equivalence remain for everything else.

[Copilot]

Pull request overview

Copilot reviewed 23 out of 24 changed files in this pull request and generated 3 comments.

Files not reviewed (1)

• api/v1alpha1/zz_generated.deepcopy.go: Generated file

[shingonoide]

Reviewed the multi-source CRD plus both reconcilers against PLAT-1141. The CEL xor on sourceRef / sources correctly rejects both-set and neither-set (including the empty sources: [] case via the explicit size() > 0), EffectiveSources() normalizes both forms into one path, and the generated deepcopy deep-copies the new slice rather than aliasing it. The per-binding env injection is ordered before FILTER_* on every builder so the $(RTSP_URL) / $(VIDEO_INPUT_PATH) expansion resolves, and the batch direct-download path preserves the file extension (the fix for the image-in case). The PipelineSource to PipelineInstance watch plus the Degraded self-heal closes the PI-before-sources creation race. CI green, including CRD-in-sync and chart parity.

Two non-blocking test-coverage gaps worth a follow-up:

1. The GPU coexistence with PR #66 is wired on the streaming and batch multi-source paths via the shared applyGPUContainerSharing, but no test asserts nvidia.com/gpu / NVIDIA_VISIBLE_DEVICES placement on a Spec.Sources instance (the 3-cams + 1-GPU-filter topology the ticket calls out).
2. The streaming RTSP_URL-before-FILTER_* ordering is correct in code but only the batch VIDEO_INPUT_PATH ordering is asserted by a test (assertEnvPrecedesFilterConfig). A streaming-side assertion would guard against future reordering regressions.

Approving. Reminder for the rollout: this needs a v* release (or a pinned CLAIMER_IMAGE) post-merge since claimer:latest predates direct mode, as the ticket notes.

[lucasmundim]

Thanks — both coverage gaps closed in the latest push:

1. GPU coexistence on multi-source: TestBuildStreamingDeployment_MultiSource_GPUCoexistence pins the exact 3-cams + 1-GPU-filter topology on a Spec.Sources instance — exactly one container holds the nvidia.com/gpu limit, the GPU filter carries NVIDIA_VISIBLE_DEVICES (and no RTSP_URL), and each camera container carries its own per-binding RTSP_URL with no GPU env.

2. Streaming env ordering: TestBuildStreamingDeployment_RTSPURLPrecedesFilterConfig asserts RTSP_URL precedes every FILTER_* entry via the same assertEnvPrecedesFilterConfig helper that already guards the batch VIDEO_INPUT_PATH ordering.

And yes on the rollout reminder — the claimer release requirement is recorded in the PR's deploy note, the plan doc's merge-order section, and PLAT-1141's as-built notes, so it can't get lost between merge and deploy.