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

DESIGN_DOCUMENT.md

@@ -145,11 +145,37 @@ spec:
     name: <pipeline-name>
     namespace: default  # optional, defaults to PipelineInstance namespace
 
-  # sourceRef references the PipelineSource for input configuration
+  # SOURCE BINDING (PLAT-1071): exactly one of `sourceRef` or `sources`
+  # must be set. Enforced by an admission-time XValidation rule on the
+  # spec; setting both, or neither, is rejected by the kube API server.
+
+  # sourceRef (legacy, single-source): references the PipelineSource for
+  # input configuration. The source URL is broadcast to every filter
+  # container as RTSP_URL (streaming) or as a single VIDEO_INPUT_PATH
+  # download target (batch). Existing single-source CRs continue to apply
+  # unchanged.
   sourceRef:
     name: <source-name>
     namespace: default  # optional, defaults to PipelineInstance namespace
 
+  # sources (multi-source): each entry binds a PipelineSource to a
+  # specific filter container by name. The controller injects the
+  # source-derived env vars into ONLY the matching container, enabling
+  # multi-camera pipelines where each VideoIn filter consumes a distinct
+  # source. Supported in both streaming (RTSP_URL per container) and
+  # batch (per-binding direct-mode init claimer + VIDEO_INPUT_PATH per
+  # container). See section 5.4 for the per-container env injection
+  # contract; section 5.5 covers the batch-specific constraints
+  # (every binding's PipelineSource Bucket.Prefix must be a FULL object
+  # key so the direct-mode claimer has a deterministic download).
+  sources:
+    - filterName: front-cam
+      sourceRef:
+        name: pipelinesource-front
+    - filterName: back-cam
+      sourceRef:
+        name: pipelinesource-back
+
   # execution defines how the pipeline should be executed
   execution:
     parallelism: 10         # max concurrent pods (default: 10)
@@ -370,6 +396,104 @@ The claimer is a standalone Go binary (`cmd/claimer/main.go`) that runs as an in
 - On success: all filters exit 0 → Pod Succeeded
 - Filters should be idempotent (at-least-once delivery semantics)
 
+### 5.4 Multi-source contract (PLAT-1071)
+
+A PipelineInstance with `Spec.Sources` bound to N filter containers runs
+all of them in the **same Pod** (streaming mode: one Deployment, one
+replica, N+M containers; batch mode: one Job with a single Pod, one init
+claimer per binding — see 5.5). Per-container source plumbing:
+
+- The reconciler normalizes `Spec.SourceRef` (legacy) and `Spec.Sources`
+  into a uniform `[]ResolvedSourceBinding` via
+  `PipelineInstanceSpec.EffectiveSources()`. Legacy `SourceRef` becomes
+  one entry with `FilterName == ""` — the broadcast sentinel — and the
+  source env vars apply to every container in the pod (today's behavior).
+- For each entry where `FilterName != ""`, the controller injects
+  `RTSP_URL` (and, when credentials are present, `_RTSP_USERNAME` /
+  `_RTSP_PASSWORD` from the referenced Secret) into ONLY the container
+  whose `pipeline.spec.filters[].name` matches `FilterName`. Downstream
+  filters (no matching binding) receive no source env vars and wire to
+  upstream siblings through their own `sources: tcp://localhost:...`
+  filter config.
+- Every non-empty `sources[].filterName` must match a
+  `pipeline.spec.filters[].name` exactly. An unmatched binding would be a
+  silent no-op (streaming: the source env never lands on any container;
+  batch: the claimer downloads a file no filter reads), so the reconciler
+  rejects it with a `Degraded` condition (reason `SourceBindingUnmatched`)
+  whose message lists the unmatched and available filter names. The
+  condition clears automatically once the bindings validate.
+- **Idle timeout anchoring (streaming, V1)**: multi-source instances apply
+  the idle-timeout policy from `sourceBindings[0]` — the first `sources[]`
+  entry — ONLY. This is deliberate: the semantic of "ANY source idle vs
+  ALL sources idle" is not yet defined for multi-camera (see the Step 3
+  comment in `reconcileStreaming`). An `idleTimeout` set on any other
+  binding's source is silently ignored; operators who need per-source idle
+  budgets should author one PipelineInstance per source today.
+- Pipeline authoring constraints the user must satisfy when emitting a
+  multi-VideoIn Pipeline:
+  - **Stagger output ports by 2.** OpenFilter binds the configured port
+    AND `port + 1` per `outputs` (the +1 carries ZMQ control-plane
+    traffic). For N VideoIn filters use 5550, 5552, 5554, … — never
+    5550, 5551.
+  - **Tag every VideoIn output topic explicitly** via the `;topic`
+    suffix in its `sources` config. Without it every VideoIn publishes
+    to the default topic `main` and any downstream that fans-in two
+    upstreams crashes with `RuntimeError: duplicate topic 'main'`. Use
+    the filter name as the topic for determinism.
+
+  Neither constraint is enforced by the controller — they live in the
+  user-authored `filter.config` strings. See the
+  `pipelines_v1alpha1_pipelineinstance_stream_multisource.yaml` sample
+  for a complete worked example.
+
+  Note on ownership (deviation from the original PLAT-1141 draft): the
+  ticket sketched the controller injecting `FILTER_SOURCES` and a
+  deterministic port scheme (5550 + index). The implementation instead
+  treats ports, topics, and downstream `sources` wiring as
+  **authoring-time** concerns — baked into the Pipeline CRD by the
+  plainsight-api export (or by hand, per the samples) — and the
+  controller only injects the per-binding env (`RTSP_URL` /
+  `VIDEO_INPUT_PATH`) into the matching container. The pivot was
+  validated end-to-end with multi-camera runs (see PR #74).
+
+### 5.5 Multi-source batch
+
+Batch mode supports both single-source (the legacy queue-based
+parallel-files-per-bucket flow, unchanged) and multi-source bindings
+(one Pod, N init claimers, N VideoIns).
+
+The reconciler branches on `len(Sources)` at the top of
+`reconcileBatch`:
+
+- **Single-source**: existing flow. List bucket, enqueue files in
+  Valkey, parallel Pods each pop one file via the init claimer, run
+  the filter chain.
+
+- **Multi-source** (`reconcileBatchMultiSource`): no Valkey work
+  queue, no bucket listing. One init claimer per binding in **direct
+  mode** (S3_OBJECT_KEY env tells the claimer to skip Valkey and
+  download exactly that object), each writing to
+  `/ws/<filterName>.<ext>`. Filter containers run as a chain in the
+  same Pod with VIDEO_INPUT_PATH injected on each matching VideoIn
+  container. Job is `parallelism=1, completions=1` and completion is
+  the Job's own Succeeded/Failed status (no per-file accounting).
+
+Constraints on multi-source batch:
+
+- Every binding's PipelineSource must be a Bucket source — RTSP makes
+  no sense for batch.
+- Every binding's `Bucket.Prefix` must name a specific S3 object key
+  (the platform's media model carries single-file keys in prefix). The
+  default queue-based flow (list a prefix and enqueue every file under
+  it) doesn't apply here: the user has named exactly which media goes
+  to which VideoIn, so each claimer needs a deterministic key.
+
+Both constraints surface as Degraded conditions
+(`MultiSourceBatchInvalidSource` / `MultiSourceBatchMissingObject`)
+with operator-actionable messages. A controller watch on PipelineSource
+re-triggers reconcile when a referenced source changes, and the
+Degraded condition clears automatically once validation passes.
+
 ## 6. Controller Responsibilities
 
 The PipelineInstance controller (`internal/controller/pipelineinstance_controller.go`) is the orchestrator for all pipeline execution logic.

api/v1alpha1/pipeline_types.go

@@ -47,6 +47,19 @@ type BucketSource struct {
 	// +optional
 	Prefix string `json:"prefix,omitempty"`
 
+	// singleObject declares that prefix is a FULL object key naming exactly
+	// one file, not a prefix to scan. Batch instances whose every binding is
+	// a single-object bucket run in direct-download mode (one init claimer
+	// per binding, no Valkey work queue) — the same path multi-source
+	// instances always use — preserving the object's file extension in the
+	// per-filter input path (queue mode downloads to a fixed input.mp4,
+	// which breaks extension-sensitive filters like image-in). The
+	// deployment agent sets this for single-file media kinds
+	// (external_video, external_image, uploads); hand-authored CRs without
+	// it keep the queue-based prefix-scan flow.
+	// +optional
+	SingleObject bool `json:"singleObject,omitempty"`
+
 	// endpoint is the S3-compatible endpoint URL (required for non-AWS S3)
 	// Examples:
 	//   - MinIO: "http://minio.example.com:9000"

api/v1alpha1/pipelineinstance_types.go

@@ -42,15 +42,54 @@ type ExecutionConfig struct {
 	PendingTimeout *metav1.Duration `json:"pendingTimeout,omitempty"`
 }
 
-// PipelineInstanceSpec defines the desired state of PipelineInstance
+// PipelineInstanceSpec defines the desired state of PipelineInstance.
+//
+// Source binding (PLAT-1071): a PipelineInstance binds one or more
+// PipelineSources to specific filter containers in the referenced Pipeline.
+// Exactly one of the following must be set (enforced by the XValidation
+// rule on this type):
+//
+//   - sourceRef (single-source convenience form): the bound source URL is
+//     broadcast to every filter container as RTSP_URL (streaming) or as a
+//     single VIDEO_INPUT_PATH download target (batch). The natural shape
+//     for one-camera pipelines — no filter names required.
+//
+//   - sources (multi-source): each entry binds a PipelineSource to the
+//     filter container whose name matches `filterName`. The controller
+//     injects the source-derived env vars (RTSP_URL / VIDEO_INPUT_PATH)
+//     into ONLY that container. Multiple bindings produce multiple
+//     per-container source URLs, enabling multi-camera pipelines where
+//     each VideoIn filter consumes a distinct source.
+//
+// Setting both fields, or neither, is a validation error rejected at
+// admission time.
+// +kubebuilder:validation:XValidation:rule="(has(self.sourceRef) && !has(self.sources)) || (!has(self.sourceRef) && has(self.sources) && size(self.sources) > 0)",message="exactly one of `sourceRef` or `sources` must be set (sources must be non-empty)"
 type PipelineInstanceSpec struct {
 	// pipelineRef references the Pipeline resource to execute
 	// +required
 	PipelineRef PipelineReference `json:"pipelineRef"`
 
-	// sourceRef references the PipelineSource resource for input configuration
-	// +required
-	SourceRef SourceReference `json:"sourceRef"`
+	// sourceRef references the PipelineSource resource for input
+	// configuration. This is the convenience form for single-source
+	// pipelines — the source URL is broadcast to every filter container,
+	// so the author doesn't need to know filter names. For multi-source
+	// pipelines, use `sources` instead. Both forms are first-class and
+	// supported indefinitely.
+	// +optional
+	SourceRef *SourceReference `json:"sourceRef,omitempty"`
+
+	// sources binds one or more PipelineSources to specific filter
+	// containers by name. Each entry targets the container whose
+	// `pipeline.spec.filters[].name` matches `filterName`; the controller
+	// injects the source-derived env vars into only that container. Use
+	// for multi-camera pipelines where each VideoIn filter consumes a
+	// distinct source.
+	//
+	// Exactly one of `sourceRef` or `sources` must be set.
+	// +optional
+	// +listType=map
+	// +listMapKey=filterName
+	Sources []NamedSourceRef `json:"sources,omitempty"`
 
 	// execution defines how the pipeline should be executed
 	// +optional
@@ -99,6 +138,50 @@ type SourceReference struct {
 	Namespace *string `json:"namespace,omitempty"`
 }
 
+// NamedSourceRef binds a PipelineSource to a specific filter container by
+// name. The controller injects the source-derived env vars (RTSP_URL for
+// streaming, VIDEO_INPUT_PATH for batch) into only the container whose
+// `pipeline.spec.filters[].name` matches `filterName`. This lets a single
+// PipelineInstance run a multi-camera pipeline where each VideoIn filter
+// pulls a distinct source.
+type NamedSourceRef struct {
+	// filterName names the filter container in the referenced Pipeline
+	// that consumes this source. Must match a `pipeline.spec.filters[].name`
+	// value exactly. Constrained to DNS-1123 label rules (lowercase
+	// alphanum + hyphens, max 63 chars) so it can safely flow into
+	// generated container / volume / env names.
+	// +required
+	// +kubebuilder:validation:MinLength=1
+	// +kubebuilder:validation:MaxLength=63
+	// +kubebuilder:validation:Pattern=`^[a-z0-9]([-a-z0-9]*[a-z0-9])?$`
+	FilterName string `json:"filterName"`
+
+	// sourceRef references the PipelineSource resource that provides the
+	// input for the filter named by `filterName`.
+	// +required
+	SourceRef SourceReference `json:"sourceRef"`
+}
+
+// EffectiveSources returns the canonical list of (filterName, source) bindings
+// for this PipelineInstance, normalizing the singular `SourceRef` field into
+// the plural shape. When `SourceRef` is set, the returned list has a single entry
+// with an empty FilterName — a sentinel that means "broadcast to every
+// container" in the reconciler's per-binding loop. When `Sources` is set, the
+// returned list is the user's bindings verbatim. Validation at admission time
+// guarantees exactly one of the two is set.
+func (s *PipelineInstanceSpec) EffectiveSources() []NamedSourceRef {
+	if len(s.Sources) > 0 {
+		return s.Sources
+	}
+	if s.SourceRef != nil {
+		return []NamedSourceRef{{
+			FilterName: "",
+			SourceRef:  *s.SourceRef,
+		}}
+	}
+	return nil
+}
+
 // PipelineInstanceStatus defines the observed state of PipelineInstance.
 type PipelineInstanceStatus struct {
 	// counts tracks the number of files in each state (Batch mode only)

cmd/claimer/main.go

@@ -56,6 +56,12 @@ const (
 	EnvS3PathStyle    = "S3_USE_PATH_STYLE"
 	EnvS3SkipTLS      = "S3_INSECURE_SKIP_TLS_VERIFY"
 	EnvVideoInput     = "VIDEO_INPUT_PATH"
+	// EnvS3ObjectKey selects "direct download" mode for the multi-source
+	// batch path (PLAT-1071). When set, the claimer skips Valkey and
+	// downloads exactly this S3 object key to VIDEO_INPUT_PATH, then
+	// exits. Used by the per-VideoIn init claimer the batch reconciler
+	// emits for multi-source bindings.
+	EnvS3ObjectKey = "S3_OBJECT_KEY"
 
 	// Volume mount paths
 	defaultInputPath = "/ws/input.mp4"
@@ -91,6 +97,24 @@ func run() error {
 		return fmt.Errorf("failed to load config: %w", err)
 	}
 
+	// Direct-download mode (PLAT-1071): when S3_OBJECT_KEY is set, the
+	// caller has named a specific object the claimer must download. This
+	// is the multi-source batch path — one init claimer per VideoIn
+	// binding, each pointing at its bound media's object. No Valkey
+	// work-queue is involved.
+	if cfg.S3ObjectKey != "" {
+		log.Printf("Direct-download mode: bucket=%s, key=%s, dest=%s", cfg.S3Bucket, cfg.S3ObjectKey, cfg.VideoInputPath)
+		minioClient, err := createMinIOClient(cfg)
+		if err != nil {
+			return fmt.Errorf("failed to create MinIO client: %w", err)
+		}
+		if err := downloadFile(ctx, minioClient, cfg.S3Bucket, cfg.S3ObjectKey, cfg.VideoInputPath); err != nil {
+			return fmt.Errorf("failed to download file: %w", err)
+		}
+		log.Printf("Downloaded %s to %s", cfg.S3ObjectKey, cfg.VideoInputPath)
+		return nil
+	}
+
 	// Create Valkey client (with retry for outages during startup)
 	var valkeyClient valkey.Client
 	clientBackoff := newExponentialBackoff(30 * time.Second)
@@ -163,6 +187,9 @@ type Config struct {
 	S3UsePathStyle bool
 	S3SkipTLS      bool
 	VideoInputPath string
+	// S3ObjectKey selects direct-download mode (PLAT-1071). Empty in the
+	// legacy queue-based path.
+	S3ObjectKey string
 }
 
 func loadConfig() (*Config, error) {
@@ -186,18 +213,22 @@ func loadConfig() (*Config, error) {
 			}
 			return defaultInputPath
 		}(),
+		S3ObjectKey: os.Getenv(EnvS3ObjectKey),
 	}
 
 	// Parse boolean flags
 	cfg.S3UsePathStyle = getEnvOrDefault(EnvS3PathStyle, "false") == "true"
 	cfg.S3SkipTLS = getEnvOrDefault(EnvS3SkipTLS, "false") == "true"
 
-	// Validate required fields
-	if cfg.Stream == "" {
-		return nil, fmt.Errorf("STREAM is required")
-	}
-	if cfg.Group == "" {
-		return nil, fmt.Errorf("GROUP is required")
+	// STREAM / GROUP are required only in queue-based mode. In direct
+	// mode (S3_OBJECT_KEY set) the claimer skips Valkey entirely.
+	if cfg.S3ObjectKey == "" {
+		if cfg.Stream == "" {
+			return nil, fmt.Errorf("STREAM is required")
+		}
+		if cfg.Group == "" {
+			return nil, fmt.Errorf("GROUP is required")
+		}
 	}
 	// POD_NAME and POD_NAMESPACE are optional; retained for diagnostics only.
 	if cfg.S3Bucket == "" {