Pipeline

Streaming Pipeline

prohori pipeline runs Prohori continuously instead of once per file. It chains the parsers, the rules and the incident builder as consumers and stores events, alerts, incidents and rejected messages in a database. It runs in two ways:

• Lite mode: one process, no Docker, results in a DuckDB file.
• The Docker stack: Redpanda, PostgreSQL and one container per stage.

The full page is docs/PIPELINE.md.

Install it with pip install "prohori[pipeline]" (on PyPI since v0.3.0), or run it from source with uv sync --extra pipeline.

prohori simulate web-scanning --seed 1 --out logs   # write a simulated log to replay
prohori pipeline replay logs/web-scanning.log --times 2   # replay twice, then stop
prohori pipeline tail /var/log/auth.log --year 2026       # follow a live log until Ctrl+C

State lives in .prohori/pipeline/ (--state), and results go to .prohori/pipeline/prohori.duckdb (--db). Rule, parser and enrichment options work as for prohori incidents.

How it works

tail / replay -> raw-logs -> normalise -> events -> detect -> alerts -> incidents -> incidents
                                 |                                                     |
                                 +-> dead-letter            every topic -> DuckDB sink <+

• Normalise parses each line, using one parser per file.
• Detect drops events it has already seen, then runs the same rule engine as prohori detect.
• Incidents waits until a group of alerts can no longer grow, then builds the incident with the same builder as prohori incidents.
• The sink writes each batch in one transaction.

What you can rely on

• Same answers as batch. For every simulator scenario, the stream's incidents equal those of prohori incidents on the same file. This holds even when the file is replayed twice, and CI tests it on Linux and Windows.
• Nothing lost, nothing doubled. Delivery is at least once. Every id is a hash of its content, and the database ignores ids it already has, so replays, redeliveries and crashes add no rows.
• Bad input is set aside. A line no parser accepts, or a malformed message, goes to the dead_letters table with the reason, and is counted.
• Live logs. The tailer handles rotation by rename and by truncation, and resumes after a restart. On Windows it shares the file, so the writing program can still rotate it. Ctrl+C stops cleanly.

The database

Table	Holds
events	Every parsed event: time, source, action, outcome, IP, user, HTTP status, full JSON
alerts	Rule alerts: rule id, severity, time range, title, event count, full JSON
incidents	Incidents: severity, score, confidence, time range, title, alert count, full JSON
dead_letters	Rejected messages: stage, topic, key, reason, payload

SELECT id, severity, score, title FROM incidents ORDER BY score DESC;

The Docker stack

Run it from a clone of the repository, with Docker running:

uv run prohori pipeline up                    # build, start and wait until healthy
uv run prohori simulate web-scanning --seed 1 --out logs   # write a simulated log to replay
uv run prohori pipeline replay logs/web-scanning.log --bootstrap 127.0.0.1:19092
uv run prohori pipeline status                # services, health and warehouse rows
uv run prohori pipeline down                  # --volumes also deletes the data

Service	What it is	Memory limit
redpanda	Kafka-compatible broker, one node	1.5 GB
postgres	Warehouse (raw schema)	512 MB
normalise, detect, incidents, sink	One Prohori container per stage	256-384 MB each
grafana	Provisioned dashboards on 127.0.0.1:13000 (Dashboard)	512 MB
airflow (with --orchestration)	Scheduled jobs on 127.0.0.1:18080 (Orchestration)	3 GB
warehouse-init	Runs once per start: schemas and Grafana's read-only role	64 MB

• Footprint. Replaying a 50,700-line log, the stack peaked at about 700 MB of its limits (about 3.9 GB with Grafana).
• Local only. Ports are bound to 127.0.0.1: the broker on 19092, PostgreSQL on 15432 and Grafana on 13000. The PostgreSQL password defaults to a local-only value; set PROHORI_POSTGRES_PASSWORD to change it.
• Same answers as batch. A CI smoke test replays a simulated attack twice into the stack and checks that each event is stored once and that the incidents equal the batch command's.
• Incidents are written after a quiet minute, so a replay's last incident appears quickly. A very slow attack can be split into two incidents.

Query the warehouse:

docker exec -it prohori-postgres-1 psql -U prohori -d prohori -c "SELECT severity, score, title FROM raw.incidents"

Limits

• Lite mode handles about 2,800 lines a second on a laptop, which suits one host's logs. The batch command handles about 14,000.
• Topics are not trimmed. Delete the state directory to start over.
• --models (anomaly detection) stays a batch feature.
• The stack is a single-node broker on one machine, with no broker authentication.
• The stored rows feed the dbt models in Warehouse and the Dashboard; Orchestration schedules the jobs around them.