StreamGuard

Real-time event pipeline with windowed metrics and anomaly detection.

StreamGuard ingests events (logs, API latency, user actions), validates and deduplicates them, aggregates windowed metrics (error rate, throughput, percentiles), detects anomalies in real time, and serves results via an API.

📖 Full documentation: See ARCHITECTURE.md for a comprehensive technical deep-dive: data models, architecture, implementation details, and module-by-module breakdown.

---

Quick Start

# Create venv and install
python -m venv venv
source venv/bin/activate  # or `venv\Scripts\activate` on Windows
pip install -e ".[dev]"

# Start server (no API key required for local dev)
streamguard run

# Open the dashboard in your browser
open http://127.0.0.1:8000/dashboard

# Or replay sample events from another terminal
streamguard replay --file sample_events.jsonl --rate 100

---

Architecture

                    +------------------+
                    |   POST /ingest   |
                    +--------+---------+
                             |
                             v
+----------+     +-----------+-----------+     +----------------+
|  Client  |---->|  FastAPI   |  Validate  |--->|  Bounded Queue |
+----------+     +-----------+-----------+     +--------+--------+
                             |                          |
                             | 422 invalid              | 503 full
                             v                          v
                    +----------------+          +----------------+
                    |  Deduplicator  |<---------|    Processor   |
                    +----------------+          +--------+-------+
                                                         |
                      +----------------+                  |
                      |   Aggregator   |<-----------------+
                      | (ring buffers) |                  |
                      +--------+-------+                  |
                               |                          v
                               |                  +----------------+
                               +----------------->| Anomaly Detect |
                                                  +--------+-------+
                                                           |
                                                           v
                                                  +----------------+
                                                  |  SQLite / API  |
                                                  +----------------+

• API layer: Validates events, enqueues to bounded queue, returns 202/422/503
• Queue: In-memory asyncio.Queue with backpressure
• Processor: Dedup → aggregate → anomaly detection
• Storage: SQLite for alerts and dedup keys; in-memory for metrics

---

Event Schema

{
  "event_id": "uuid-or-hash",
  "ts": "2026-02-11T12:34:56.123Z",
  "service": "auth-service",
  "endpoint": "/login",
  "status_code": 200,
  "latency_ms": 123.4,
  "message": "optional string",
  "tags": {"env": "prod", "region": "us-east-1"}
}

---

API Endpoints

Endpoint	Method	Description
/	GET	Redirects to dashboard
/dashboard	GET	Dashboard UI — metrics, alerts, health, send test events
/ingest	POST	Submit event (202 accepted, 422 invalid, 503 backpressure)
/metrics/current	GET	Current metrics (?window_sec=60&service=&endpoint=)
/alerts/recent	GET	Recent alerts (?limit=50&since=)
/health	GET	Health (queue depth, uptime)

---

Configuration

Copy .env.example to .env and adjust:

API_KEY=your-secret-key    # Required for auth; omit for dev
QUEUE_MAX_SIZE=10000
WINDOW_SIZES=60,300

---

CLI Commands

Command	Description
streamguard run	Start API server
streamguard replay --file F --rate N	Replay JSONL/CSV to /ingest
streamguard generate --file F --minutes M --pattern P	Generate synthetic events

Patterns: normal, spike_latency, spike_errors, throughput_drop

Note: The dashboard at /dashboard provides full feature parity—generate, replay, send events, view metrics and alerts. Only streamguard run must be done from the terminal (the server must be started before the dashboard is accessible).

---

Demo Script

# 1. Start service
streamguard run

# 2. Generate events with latency spike
streamguard generate --file spike.jsonl --minutes 2 --pattern spike_latency

# 3. Replay
streamguard replay --file spike.jsonl --rate 300

# 4. Check metrics and alerts
curl "http://127.0.0.1:8000/metrics/current?window_sec=60"
curl "http://127.0.0.1:8000/alerts/recent"

---

Testing

pytest tests/ -v
pytest tests/ --cov=streamguard

---

Design Choices

• Ring buffer over full history: O(1) update, bounded memory, predictable
• Fixed histogram for percentiles: Bounded memory, fast, easy to explain
• EWMA + deviation for anomalies: Simple, tunable, no ML dependency
• LRU dedup: O(1) lookup, evicts oldest when full

---

Universal Integration (Real Services)

StreamGuard works with any application. The event format is the contract:

1. Instrument your app — After each request (or in middleware), POST to http://your-streamguard:8000/ingest:

{
  "event_id": "unique-id-per-request",
  "ts": "2026-02-11T12:34:56.123Z",
  "service": "auth-service",
  "endpoint": "/login",
  "status_code": 200,
  "latency_ms": 123.4,
  "message": "optional",
  "tags": {"env": "prod"}
}

2. Supported frameworks — Add the HTTP call in Flask, FastAPI, Express, Rails, etc. Libraries like httpx, requests, or fetch work.

3. Sample events — sample_events.jsonl and streamguard generate are for demos/testing. Real services send live events via the API.

---

Scaling (Production)

• Replace in-memory queue with Kafka or RabbitMQ
• Use Redis for dedup keys
• Export metrics to Prometheus
• Horizontal scaling with partitioning by (service, endpoint)