Samsara: Towards a Multimodal Stream Processing System

Background

Samsara is a proof of concept for a novel query optimizer aimed at enabling Multimodal Stream Processing Systems with the help of Large Language Models(LLMs). LLMs are already being employed to support multimodal query processing over databases, but their inherent latency makes them unsuitable for the time-sensitive scenarios of Stream Processing. Samsara tackles the challenge by introducing a three-step optimization process aimed at reducing the amount of information sent to the expensive AI-based operators.

Architecture Overview

flowchart TB
    subgraph CF["Cloudflare (public)"]
        FE["React app (hosted)<br/>guests + admin"]
        QV["Queue UI<br/>shows queue state + generated labels"]
    end

    subgraph AWS["AWS (behind Cloudflare Tunnel, TLS/wss)"]
        S["server.py<br/>metrics / logs / job-reset"]
        MQ["manual_query_generation.py<br/>OpenAI label + prompt generation"]
        Q["Query Queue Service<br/>FIFO + 30s scheduler<br/>stores queryId / label / description / prompt / kind"]
        ING["frame-ingest endpoint"]
        K[("Kafka - internal")]
        FL["Flink naive + opt"]
        LLM["vLLM"]
    end

    subgraph LAP["Admin laptop (local, outbound only)"]
        CC["camera-capture.py<br/>live camera<br/>receives active query + prompt"]
        VR["video_receiver.py<br/>OpenCV display"]
    end

    FE -- "submit preset query" --> Q
    FE -- "submit manual description" --> S
    S -- "generate label + vLLM prompt" --> MQ
    MQ -- "validated manual query payload" --> Q

    FE -- "ws-queue / queue state" --> Q
    Q --> QV

    FE -- "metrics plots" --> S
    FE -- "admin: resubmit / logs / send-video" --> S

    CC -- "outbound: ws-active-query<br/>receive active queryId + label + description + prompt" --> Q
    CC -- "outbound: send frames with active prompt" --> ING

    ING --> K
    K --> FL
    FL --> LLM
    FL --> K
    K --> S

    VR -- "consume images / results" --- K

Loading

A Mermaid source for this diagram is also kept in project_architecture_mermaid. It is intended as a reusable diagram file: open it in a text editor, copy the Mermaid block, and paste it into https://mermaid.live to render and edit it visually.

Pre-requisites

The setup was tested on both MacOS and Ubuntu.

• Install the latest version of miniconda (or use Anaconda).
• Clone the repository
• cd into it and run setup for the role this machine plays:
  • Client only (camera worker / video receiver): ./setup.sh. Creates the demo conda env from conda_env.yml and installs Node for building the frontend.
  • Server: ./server/setup.sh. Calls ./setup.sh first, then downloads the JDK, Flink and Kafka into server/environment/ (~1 GB, which a client machine has no use for).
• run conda activate demo to activate the python environment

Infrastructure

The source-kafka.sh and run-flink.sh scripts deploy Kafka and Flink respectively. Kafka handles both the input topic (camera frames) and output topics (query results). Flink executes continuous queries and writes results back to Kafka.

Frames are produced by camera-capture.py and sent to Kafka input. Query changes flow through server/query_queue.py: the React app submits a query to server.py, the queue activates queued queries on a timer, and the camera worker receives active query updates through its ACTIVE_QUERY_WS_URL websocket client connection. Camera start/stop is also routed through server.py so the hosted UI does not need inbound access to the camera machine.

video_receiver.py consumes input and output Kafka topics and renders the streams using OpenCV.

The hosted frontend is the React app in frontend. Guests can view plots, submit query changes, and see the query queue. Admin-only controls (connection status, camera controls, logs, and job resubmission) are hidden until login.

Admin queue management is exposed through a token-protected websocket endpoint: /ws-queue-admin.

Manual query submission is available through the queue panel in the React app. The frontend sends the user description to server.py; backend validation and OpenAI-based generation produce a queue label and strict vLLM prompt. If validation or generation fails, the backend returns typed submission errors to the frontend. The OpenAI integration for this flow is isolated in server/manual_query_generation.py, including few-shot examples based on existing preset queries.

Environment Variables

Copy .env.example to .env and adjust values for your machine. The frontend reads the same root .env through frontend/vite.config.ts, so there is one runtime config file.

Core ports:

• SERVER_PORT: FastAPI port for server/server.py. Default 6981.

Frontend URL overrides:

• SERVER_WS_URL: websocket base URL for metrics, logs, job reset, and camera control. Leave empty to use the current browser hostname with SERVER_PORT.
• SERVER_HTTP_URL: HTTP base URL for admin login. Leave empty to use the current browser hostname with SERVER_PORT.
• QUEUE_WS_URL: websocket base URL for queue submit/state streams. Leave empty to reuse SERVER_WS_URL.

Kafka:

• KAFKA_BOOTSTRAP_SERVERS: Kafka bootstrap endpoint. Locally this is usually localhost:9092; for the hosted demo use the authenticated Kafka tunnel endpoint.
• KAFKA_SECURITY_PROTOCOL, KAFKA_SASL_MECHANISM, KAFKA_SASL_USERNAME, KAFKA_SASL_PASSWORD, KAFKA_SSL_CAFILE: optional secure Kafka settings. Leave empty for plaintext local Kafka.

Flink:

• FLINK_REST_URL: Flink REST base URL used by job_redeployment.py. Default http://localhost:8085.
• FLINK_BIN: path to the Flink CLI binary used for job submission and cancellation.
• FLINK_JOB_DIR: path to the directory containing the Python Flink jobs.

LLM and manual-query generation:

• LLM_BASE_URL: OpenAI-compatible base URL used by vLLM-backed models in llm_call.py. Default http://localhost:8000/v1.
• OPENAI_API_KEY: OpenAI key used server-side to transform manual descriptions into queue label + vLLM prompt.
• OPENAI_GEN_ENABLED: enable/disable manual query generation (true by default).
• OPENAI_GEN_MODEL: model used for manual query generation (default gpt-4o-mini).
• OPENAI_GEN_TIMEOUT_SECONDS: timeout for generation requests.
• OPENAI_GEN_MIN_DESCRIPTION_CHARS: minimum manual-description length.
• OPENAI_GEN_MAX_DESCRIPTION_CHARS: maximum manual-description length.

Backend:

• ADMIN_PASSWORD: shared password for the hosted admin view.
• ADMIN_TOKEN_SECRET: secret used to sign short-lived admin tokens. Use a long random value outside local demos.
• ADMIN_TOKEN_TTL_SECONDS: admin token lifetime in seconds. Defaults to 43200.
• QUERY_DELAY_SECONDS: FIFO query queue delay/slot length. Defaults to 30.
• CORS_ORIGINS: comma-separated allowed frontend origins.
• DEFAULT_QUERY: initial query before anything is queued. Defaults to color-detect.
• ACTIVE_QUERY_WS_URL: websocket URL the camera worker connects to for active query updates. Defaults to ${SERVER_WS_URL}/ws-active-query.

Queries

Flink query jobs are in server/queries/: naive.py and optimized.py. Shared processing functions are in helper_functions.py.

How To Run

• Run setup.sh to create environment dependencies.
• Activate with conda activate demo.
• Copy .env.example to .env and tune config if needed.
• Update Kafka config at environment/kafka_2.13-4.1.0/config/server.properties (log.dirs=./kraft-combined-logs).
• In server/, run ./source-kafka.sh to start Kafka.
• Update Flink config at server/environment/flink-1.20.3/conf/config.yaml:
  • set numberOfTaskSlots to 2.
  • set jobmanager.rpc.port to 6200 and rest.port to 8085.
• Still in server/, run ./run-flink.sh to start Flink.
• Run python client/camera-capture.py to start the local camera worker. It connects to ACTIVE_QUERY_WS_URL, receives active query updates, and sends frames to Kafka. It does not bind to a local HTTP or websocket port.
• In server/, run uvicorn server:app --reload --port "$SERVER_PORT". This server is used for plotting, query queueing, admin login, camera control, logs, and resubmitting jobs.
• Run python client/video_receiver.py to start consuming the output of the queries from kafka. This also shows the input and output videos on three separate panes.