Skip to content

RahulGIT24/Scalable-Log-Processor

BranchesTags

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

15 Commits
.github/workflows
.github/workflows
 
 
client
client
 
 
log_processor
log_processor
 
 
.gitignore
.gitignore
 
 
Readme.md
Readme.md
 
 
docker-compose.yml
docker-compose.yml
 
 
thumbnail.png
thumbnail.png
 
 

Repository files navigation

LogStream — High-Throughput Log Processing Pipeline

An end-to-end, fault-tolerant log ingestion and observability pipeline designed to handle massive concurrent traffic.

Built using an event-driven architecture, LogStream decouples ingestion, processing, storage, and visualization layers to ensure scalability, resilience, and near real-time observability under heavy load.

Features

  • High-throughput asynchronous log ingestion
  • Redis Streams-based event pipeline
  • Consumer Group architecture for concurrent workers
  • Real-time observability dashboard using WebSockets
  • Infinite scrolling searchable log explorer
  • SHA-256 based deduplication
  • Batch database insertion using executemany
  • Fault-tolerant replay handling using Redis Pending Entries List (PEL)
  • PostgreSQL-backed persistent storage
  • Full-text log searching
  • Live ingestion velocity monitoring
  • Horizontally scalable worker architecture
  • API-key authenticated & rate-limited ingestion endpoint
  • Structured, request-correlated logging
  • Liveness/readiness health probes and Prometheus metrics

Architecture Overview

                ┌────────────────────┐
                │    Client Apps     │
                └─────────┬──────────┘
                          │
                          ▼
                ┌────────────────────┐
                │  FastAPI Ingestion │
                │      Layer         │
                └─────────┬──────────┘
                          │
                          ▼
                ┌────────────────────┐
                │   Redis Streams    │
                │  Consumer Groups   │
                └─────────┬──────────┘
                          │
              ┌───────────┴───────────┐
              ▼                       ▼
      ┌──────────────┐       ┌──────────────┐
      │ Worker #1    │       │ Worker #2    │
      │ asyncio      │       │ asyncio      │
      └──────┬───────┘       └──────┬───────┘
             │                      │
             └──────────┬───────────┘
                        ▼
              ┌────────────────────┐
              │    PostgreSQL      │
              │   Persistent DB    │
              └────────────────────┘

                        │
                        ▼
              ┌────────────────────┐
              │ Redis Pub/Sub      │
              │ Live Metrics Bus   │
              └─────────┬──────────┘
                        ▼
              ┌────────────────────┐
              │ React Dashboard    │
              │ WebSocket Client   │
              └────────────────────┘

Tech Stack

Backend

  • FastAPI
  • Python 3.11+
  • SQLAlchemy
  • asyncpg
  • asyncio
  • Redis Streams
  • Redis Pub/Sub
  • PostgreSQL
  • Alembic
  • pydantic-settings
  • prometheus-fastapi-instrumentator
  • uv

Frontend

  • React
  • TypeScript
  • Tailwind CSS
  • Recharts
  • Lucide Icons

Key Engineering Decisions

Event-Driven Architecture

Instead of writing directly to PostgreSQL during ingestion, logs are first pushed into Redis Streams.

This decouples:

  • API responsiveness
  • database writes
  • processing workload

Result:

  • sub-millisecond ingestion latency
  • higher throughput
  • resilient backpressure handling

Idempotent Processing

Every log message is hashed using SHA-256 before insertion.

Combined with PostgreSQL:

ON CONFLICT DO NOTHING

workers can safely replay failed batches without corrupting the database.

This makes recovery from worker crashes deterministic and safe.

Batch Inserts Instead of Row Inserts

Workers collect logs in batches of 100 and use:

executemany()

instead of individual insert queries.

This reduces:

  • database round trips
  • transaction overhead
  • disk sync pressure

Result:

  • significantly higher throughput
  • lower database CPU utilization

Consumer Groups for Horizontal Scaling

Redis Consumer Groups allow multiple workers to process logs concurrently without duplication.

Scaling becomes simple:

uv run processor.py

Run multiple instances across machines or containers.

Real-Time Observability Without DB Polling

Instead of continuously polling PostgreSQL for updates:

  • workers publish lightweight metrics to Redis Pub/Sub
  • FastAPI listens to Pub/Sub
  • WebSocket broadcasts updates to connected clients

Result:

  • near-zero database load during live monitoring
  • efficient real-time dashboard streaming

Production Readiness

Beyond the core pipeline, the backend includes the operational scaffolding a real deployment needs:

  • Centralized configuration - one Settings object (app/core/config.py) validated at startup, instead of os.getenv sprinkled across modules.
  • Structured logging with request correlation - every HTTP request gets an X-Request-ID (generated or forwarded), bound to a contextvars context so all log lines emitted while handling it share the same id (app/core/logging.py, app/core/middleware.py).
  • Global error handling - a single exception handler returns a consistent JSON error body and logs the full traceback server-side, instead of duplicating try/except per endpoint.
  • Liveness vs. readiness - GET /health is a cheap liveness probe; GET /health/ready actually pings Postgres and Redis and returns 503 if either is down, the distinction Kubernetes (and any real load balancer) expects.
  • Authenticated, rate-limited ingestion - POST /api/ingest requires an X-API-Key header and is rate-limited per key/IP via a Redis fixed-window counter, so a public log-ingestion endpoint can't be used to spam the pipeline or run up database writes. The read-only analytics/dashboard endpoints stay open.
  • Metrics - GET /metrics exposes Prometheus-formatted request latency/count metrics plus a custom logs_ingested_total counter.
  • Security headers - baseline headers (X-Content-Type-Options, X-Frame-Options, Referrer-Policy, Permissions-Policy) applied to every response.
  • Real horizontal scaling - each worker process now derives a unique Redis consumer name (hostname + random suffix, or CONSUMER_NAME if pinned) instead of a hardcoded consumer1. Previously every scaled-out worker replica would have collided under the same consumer identity, silently defeating the consumer-group fan-out the architecture is built around. Workers also shut down gracefully on SIGTERM/SIGINT after finishing their current batch.

Environment Variables

Copy log_processor/.env.example to log_processor/.env and fill in real values:

REDIS_HOST=localhost
REDIS_PORT=6379

DB_HOST=localhost
DB_PORT=5432
DB_USER=postgres
DB_PASS=your_password
DB_NAME=logger

DATABASE_URL=postgresql://postgres:root@localhost:5432/logger
TABLE_NAME=logs

# Comma-separated list of origins allowed to call the API
CORS_ORIGINS=http://localhost:5173

# Required header (X-API-Key) for POST /api/ingest - generate a real secret
# before deploying, e.g. `python -c "import secrets; print(secrets.token_hex(32))"`
INGESTION_API_KEY=dev-only-change-me

# Fixed-window rate limit applied to POST /api/ingest, keyed by API key/IP
RATE_LIMIT_MAX_REQUESTS=100
RATE_LIMIT_WINDOW_SECONDS=60

All configuration is loaded once at startup through a single Settings object (app/core/config.py, backed by pydantic-settings) instead of os.getenv calls scattered across the codebase - a missing/misnamed variable fails fast at boot rather than silently at request time.

Backend Setup

Install uv

pip install uv

Create Virtual Environment

uv venv

Activate environment:

Linux/macOS

source .venv/bin/activate

Windows

.venv\Scripts\activate

Install Dependencies

If using pyproject.toml:

uv sync

Database Migrations (Alembic)

Initialize Alembic:

uv run alembic init alembic

Create migration:

uv run alembic revision --autogenerate -m "create logs table"

Run migrations:

uv run alembic upgrade head

Running the Backend

Start FastAPI Server

uv run uvicorn app.main:app --reload --port 8000

Start Worker

uv run processor.py

Run multiple workers in separate terminals for horizontal scaling.

Frontend Setup

Navigate into frontend directory:

npm install -g pnpm
cd client

Install dependencies:

pnpm install

Run development server:

pnpm run dev

Docker Setup (Recommended)

The easiest way to run LogStream is using Docker Compose. This starts all required services automatically, including PostgreSQL, Redis, the backend API, frontend, worker, and database migrations.

Prerequisites

  • Docker
  • Docker Compose

Verify your installation:

docker --version
docker compose version

Clone the Repository

git clone https://github.com/RahulGIT24/Scalable-Log-Processor
cd Scalable-Log-Processor

Create .docker.env

Inside the log_processor directory, create a file named:

log_processor/.docker.env

Add the following configuration:

REDIS_HOST=redis
REDIS_PORT=6379

DB_HOST=postgres
DB_PORT=5432
DB_USER=postgres
DB_PASS=root
DB_NAME=logger

DATABASE_URL=postgresql://postgres:root@postgres:5432/logger

TABLE_NAME=logs

Note: Inside Docker containers, services communicate using the Docker Compose service names (postgres, redis) rather than localhost.

Build the Images

docker compose build

Start the Application

docker compose up

Run in detached mode if preferred:

docker compose up -d

Services Started

Docker Compose automatically starts:

  • PostgreSQL
  • Redis
  • FastAPI Backend
  • React Frontend (served using Nginx)
  • Background Log Processing Worker
  • Alembic Migration Service

No additional setup is required.

Access the Application

Service URL
Frontend http://localhost:5173
Backend API http://localhost:8000

Stop the Application

Stop all containers:

docker compose down

To also remove the PostgreSQL volume:

docker compose down -v

Rebuild After Dependency Changes

If Dockerfiles or dependencies change:

docker compose build --no-cache
docker compose up

Docker Architecture

Docker Compose runs the following services:

  • frontend — React application served via Nginx
  • backend — FastAPI REST API
  • worker — Redis Streams consumer responsible for processing logs
  • migrate — Alembic migration runner
  • postgres — PostgreSQL database
  • redis — Redis Streams and Pub/Sub server

Running Tests

Backend

cd log_processor
uv sync
uv run pytest -v

Tests for ingestion, the pydantic Log model, and the worker's process_batch/make_group logic are fully hermetic (Redis and the DB are mocked) and always run.

Tests for the /api/analytics endpoints need a real Postgres database (SQLAlchemy's ARRAY column type isn't supported by SQLite). They only run when TEST_DATABASE_URL is set, and are skipped otherwise:

docker run -d --rm -p 5432:5432 \
  -e POSTGRES_USER=postgres -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=logstream_test \
  postgres:15

TEST_DATABASE_URL=postgresql://postgres:postgres@localhost:5432/logstream_test uv run pytest -v

TEST_DATABASE_URL is intentionally separate from DATABASE_URL so tests never accidentally run against whatever database your local .env points at.

Frontend

cd client
pnpm install
pnpm run test          # single run
pnpm run test:watch    # watch mode
pnpm run test:coverage # with coverage report

Continuous Integration

Every push and pull request to master/main runs .github/workflows/ci.yml, which:

  • spins up Postgres and Redis service containers, runs Alembic migrations, and runs the backend pytest suite
  • installs frontend dependencies, lints, runs the Vitest suite, and builds the client

Stress Testing

A stress testing script is included to simulate high concurrent traffic.

Run:

uv run stress_test.py

Default configuration:

  • 500 concurrent connections
  • 5000 log messages

Expected behavior:

  • API accepts requests immediately
  • Redis buffers ingestion spikes
  • workers batch-process logs asynchronously
  • dashboard displays live ingestion spikes in real time

API Endpoints

Interactive OpenAPI docs are available at /docs (Swagger UI) and /redoc while the backend is running.

Ingestion

POST /api/ingest/

Push a log into the Redis Stream. Requires an X-API-Key header matching INGESTION_API_KEY, and is rate-limited to RATE_LIMIT_MAX_REQUESTS requests per RATE_LIMIT_WINDOW_SECONDS per key (401 if the key is missing/wrong, 429 if the limit is exceeded).

Example request:

curl -X POST http://localhost:8000/api/ingest/ \
  -H "Content-Type: application/json" \
  -H "X-API-Key: dev-only-change-me" \
  -d '{"type": "ERROR", "message": "Database connection FAILED"}'

Tags shown in the dashboard aren't submitted by the client - the worker derives them from any ALL-CAPS words in message (e.g. FAILED above) before storing the log.

Analytics

GET /api/analytics/overview

Returns aggregated counts grouped by log level.

GET /api/analytics/search

Search historical logs with pagination.

Example:

/api/analytics/search?search=database&limit=50&offset=0

Real-Time Monitoring

WebSocket /api/analytics/ws

Streams:

  • ingestion metrics
  • live logs
  • processing statistics

Operations

GET /health

Liveness probe - always returns 200 if the process is up.

GET /health/ready

Readiness probe - pings Postgres and Redis, returns 200 ({"status": "ok"}) only if both respond, otherwise 503 with per-component detail.

GET /metrics

Prometheus-formatted metrics: HTTP request count/latency plus logs_ingested_total.

Frontend Dashboard Features

  • Live WebSocket updates
  • Infinite scroll log explorer
  • Debounced full-text search
  • Real-time ingestion velocity graph
  • Log level aggregation cards
  • Responsive observability UI
  • Search without database polling

Demo Video

Enjoy Stress Testing LogStream in action:

Demo Video

Performance Notes

Under stress testing:

  • ingestion remains non-blocking
  • Redis absorbs traffic spikes
  • workers process asynchronously
  • PostgreSQL write amplification remains low due to batching

The architecture is designed around:

  • backpressure tolerance
  • fault recovery
  • horizontal scalability
  • observability-first design

About

A log processor created by redis streams using Fastapi with an intutive UI dashboard

Resources

Readme
Activity

Stars

1 star

Watchers

0 watching

Forks

0 forks
Report repository

Releases

Packages

Contributors

Languages