Skip to content

Video Streaming Pipeline — Technical Documentation

Status: Coming Soon

The video streaming card in the Data Pipelines UI is marked as "Coming Soon" (disabled, non-clickable). The route (/blueprints/tools/data-pipelines/streaming) remains accessible for testing. The comingSoon flag is set in PIPELINE_TEMPLATES in DataPipelines.jsx.

Overview

GPU-backed video streaming pipeline. Creates a K8s Deployment on GPU nodes that processes video from RTSP cameras, MinIO files, or other sources, and exposes an HLS or WEBRTC endpoint for playback. Resources are auto-allocated based on resolution and FPS, and freed when the stream is stopped.

Architecture

Frontend (StreamingPipeline.jsx)
  ├─ Start stream → POST /streaming/sessions
  ├─ Polls GET /streaming/sessions/{id} every 3s for status
  ├─ When running → GET /streaming/sessions/{id}/stream-url
  ├─ Video player (native <video> element)
  ├─ FPS chart (recharts LineChart)
  └─ Stop stream → POST /streaming/sessions/{id}/stop

Marketplace Backend (streaming_routes.py)
  ├─ Thin proxy — authenticates user, derives PNU tenant/user UUIDs
  ├─ POST /sessions → PNU POST /api/v1/pipelines/jobs (job_type=streaming)
  ├─ GET /sessions → PNU GET /api/v1/pipelines/jobs?job_type=streaming
  ├─ GET /sessions/{id} → PNU GET /api/v1/pipelines/jobs/{id}
  ├─ POST /sessions/{id}/stop → PNU POST /api/v1/pipelines/jobs/{id}/cancel
  ├─ GET /sessions/{id}/stream-url → PNU GET /api/v1/pipelines/jobs/{id}/results
  ├─ record_streaming_usage(event="start") — called after session creation
  └─ record_streaming_usage(event="stop") — called after session stop

PNU Engine (data_processing_service.py)
  ├─ submit_job() — stores input in MinIO, creates K8s Deployment
  ├─ _create_streaming_deployment() — GPU-backed Deployment with video-streamer image
  ├ update_job_status() — polls K8s for deployment status
  ├─ cancel_job() — deletes K8s Deployment, frees GPU
  └─ data_processing_sweeper — scales idle streaming deployments to 0

Key Files

File Role
backend/routes/data_pipeline/streaming_routes.py 5 HTTP proxy endpoints (no WebSocket) + start/stop metering hooks
backend/routes/data_pipeline/_common.py Shared: derive_uuids(), proxy_error()
backend/main.py Router mounted at /api/data-pipeline/streaming
backend/middleware/auth_middleware.py:60 Protects /api/data-pipeline/streaming/sessions
frontend/src/components/marketplace/tools/data-pipelines/streaming/StreamingPipeline.jsx Full UI: video source input, format/resolution/FPS selectors, video player, GPU metrics, FPS chart, past sessions
frontend/src/api/dataFetching/dataPipelines.js createStreamingSession(), getStreamingSession(), listStreamingSessions(), stopStreamingSession(), getStreamUrl()
PNU: src/syntera_engine/api/v1/data_pipeline/jobs.py POST /pipelines/jobs — submit data processing job
PNU: src/syntera_engine/services/data_pipeline/gpu_job_service.py submit_job(), _create_batch_job(), cancel_job(), update_job_status(), finalize
PNU: src/syntera_engine/services/data_pipeline/resource_calculator.py LLM-based estimator — streaming forced to full GPU via clamp constraints
PNU: src/syntera_engine/services/data_pipeline/sweeper.py Scales idle streaming deployments to 0
PNU: src/syntera_engine/models/data_pipeline/data_processing.py DataProcessingJob model

API Endpoints

POST /api/data-pipeline/streaming/sessions

  • Auth: JWT cookie
  • Request: {"session_name": "warehouse camera", "video_source": "rtsp://...", "output_format": "hls", "resolution": "720p", "fps": 30, "processing_config": {}}
  • Process: Proxies to PNU POST /api/v1/pipelines/jobs with job_type="streaming", operation="video_stream". PNU creates a K8s Deployment with GPU resources.
  • Response: {"id": "job-uuid", "job_type": "streaming", "operation": "video_stream", "status": "queued", "needs_gpu": true, "gpu_count": 1, "gpu_profile": null, "created_at": "..."}

GET /api/data-pipeline/streaming/sessions

  • Auth: JWT cookie
  • Process: Proxies to PNU GET /api/v1/pipelines/jobs?job_type=streaming
  • Response: {"items": [...], "total": N}

GET /api/data-pipeline/streaming/sessions/{session_id}

  • Auth: JWT cookie
  • Process: Proxies to PNU GET /api/v1/pipelines/jobs/{id}
  • Response: Full job object with status, progress, GPU info, error_message

POST /api/data-pipeline/streaming/sessions/{session_id}/stop

  • Auth: JWT cookie
  • Process: Proxies to PNU POST /api/v1/pipelines/jobs/{id}/cancel. Deletes K8s Deployment, frees GPU.
  • Response: {"id": "...", "status": "cancelled", "completed_at": "..."}

GET /api/data-pipeline/streaming/sessions/{session_id}/stream-url

  • Auth: JWT cookie
  • Process: Proxies to PNU GET /api/v1/pipelines/jobs/{id}/results. Returns presigned MinIO URL for HLS playlist or WEBRTC endpoint.
  • Response: {"result_url": "https://minio.../output/job-id/result"}

GPU Resource Allocation

Operation GPU MIG Profile CPU Memory
video_stream Yes (1 GPU) Full GPU (no MIG) 8 cores 32 GB

Resources are estimated by ResourceCalculator.estimate_resources() (LLM-based) and force-clamped to full GPU for streaming. The deployment runs the syntera/video-streamer:latest container image with GPU resources.

K8s Deployment Lifecycle

  1. Create: DataProcessingService._create_batch_job() creates a K8s Deployment with 1 replica, GPU resources, and env vars (MinIO creds, input URI, output bucket, job config)
  2. Running: update_job_status() polls K8s, marks job as "running" when deployment is active
  3. Idle: sweeper.py scales idle streaming deployments to 0 after inactivity
  4. Stop: cancel_job() deletes the K8s Deployment and frees GPU resources

Frontend Flow

  1. Start Stream: Enter session name, video source URL, select format/resolution/FPS → createStreamingSession() → sets activeSession state
  2. Status updates: useQuery polls getStreamingSession() every 3s while status is queued/submitting/running (SSE is not yet available for streaming sessions — only ETL/cleaning/labeling/training jobs have SSE endpoints)
  3. Stream URL: When status becomes "running", auto-fetches stream URL via getStreamUrl()
  4. Video Player: Native <video> element displays the stream once URL is available
  5. Metrics: GPU info (count + profile), target FPS, progress percentage
  6. FPS Chart: LineChart showing FPS over time (max 60 data points)
  7. Stop: stopStreamingSession() → cancels job, clears state
  8. Past Sessions: Lists previous streaming jobs, click to view details

Container Image

The syntera/video-streamer:latest image is responsible for: - Connecting to the video source (RTSP, file, etc.) - GPU-accelerated video processing (transcoding, object detection, overlays) - Serving HLS playlist or WEBRTC endpoint - Writing output to MinIO for presigned URL access

Environment variables passed to the container: - MINIO_ENDPOINT, MINIO_ACCESS_KEY, MINIO_SECRET_KEY, MINIO_USE_PATH_STYLE - INPUT_DATA_URI — MinIO path to input data - OUTPUT_BUCKET, OUTPUT_KEY — MinIO path for output - JOB_CONFIG — JSON with session_name, video_source, output_format, resolution, fps, processing_config - JOB_TYPE, OPERATION

Migration Notes

This feature was changed from sensor-oriented WebSocket streaming to video streaming. The old model used bidirectional WebSocket connections for JSON sensor data ingestion with LLM-based transforms. The new model uses GPU-backed K8s Deployments for video processing with HLS/WEBRTC output.

What was removed:

  • WebSocket proxy endpoint (/api/data-pipeline/streaming/ws/{session_id})
  • DestinationConfig, StreamingSessionRequest (sensor-oriented models)
  • transform_instruction, generated_code, source_type: "websocket"
  • Live data stream panel, test message sender, message metrics
  • exportStreamingSession() API function

What was added:

  • VideoStreamRequest model with video_source, output_format, resolution, fps
  • GET /sessions/{id}/stream-url endpoint for retrieving playback URL
  • Video player, GPU metrics, FPS chart in frontend
  • getStreamUrl() API function
  • Proxy to PNU data processing job endpoints instead of sensor streaming endpoints

Metering

Video streaming sessions are metered via start/stop events recorded in the marketplace backend.

Recording points: - create_stream_session() — after PNU creates the job, calls record_streaming_usage(event="start") with GPU/CPU allocation - stop_stream_session() — after PNU cancels the job, calls record_streaming_usage(event="stop") with started_at, completed_at, and resource values from the PNU response

Cost formula: 

duration_hours = (completed_at - started_at) / 3600
total_cost     = (gpu_count  × STREAMING_GPU_HOURLY_RATE
                + cpu_cores  × STREAMING_CPU_HOURLY_RATE) × duration_hours

Streaming uses a higher GPU hourly rate ($2.00 vs $1.50 for data pipelines) because video encoding is GPU-intensive with sustained NVENC/NVDEC utilization. Duration is computed from started_at and completed_at on the PNU job object.

MongoDB collection: data_pipeline_usage — see Metering Overview for the full schema and data flow.