event-pipeline-monitor

Real-time monitoring for AWS event-driven pipelines. Watches SQS queues, Lambda functions, and DLQs — fires Slack alerts when things go wrong.

Built this because at my day job we had an incident where a DLQ silently accumulated 40k messages over a weekend. Nobody knew until Monday morning. This fixes that.

What it does

• polls SQS queue depths and DLQ message counts every 30s
• tracks Lambda error rates, throttles, and duration
• fires Slack alerts when queues back up or DLQs have messages
• deduplicates alerts so your phone doesn't blow up during an incident
• serves a dashboard at localhost:8000 with live metrics
• syncs alarm state from CloudWatch

Works against LocalStack locally (free, no AWS account needed) or real AWS.

Stack

• Python 3.11, FastAPI, boto3
• LocalStack for local AWS emulation
• Pydantic v2 for data models
• pytest + moto for tests

Quickstart

You need Docker and Python 3.11+.

git clone https://github.com/AdityaAudi/event-pipeline-monitor
cd event-pipeline-monitor

pip install -r requirements.txt

# start LocalStack
docker-compose up -d localstack

# wait ~10s for LocalStack to be ready, then:
python scripts/bootstrap.py

# start the monitor
uvicorn src.main:app --reload --port 8000

Dashboard at http://localhost:8000 API docs at http://localhost:8000/docs

Using Docker Compose for everything

cp .env.example .env
docker-compose up

This starts LocalStack, bootstraps the resources, and runs the monitor.

Trying it out

Once the monitor is running, use the demo script to generate traffic and trigger alerts:

# see current state
python scripts/demo.py --scenario status

# flood a queue to trigger depth alarm
python scripts/demo.py --scenario flood

# drop messages in a DLQ to trigger critical alert
python scripts/demo.py --scenario dlq

# simulate high Lambda error rate
python scripts/demo.py --scenario errors

# drain everything back to normal
python scripts/demo.py --scenario recovery

Configuration

Copy .env.example to .env and tweak as needed.

# point at LocalStack (remove this line for real AWS)
AWS_ENDPOINT_URL=http://localhost:4566

# optional — alerts log to console if not set
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/...

# thresholds
SQS_DEPTH_THRESHOLD=100
DLQ_DEPTH_THRESHOLD=1
LAMBDA_ERROR_RATE_THRESHOLD=5.0
LAMBDA_DURATION_THRESHOLD_MS=3000

POLL_INTERVAL_SECONDS=30

API

GET  /                  dashboard
GET  /health            liveness check
GET  /metrics/summary   overall health snapshot
GET  /metrics/queues    per-queue stats
GET  /metrics/lambdas   per-function stats
GET  /metrics/alarms    CloudWatch alarm states
GET  /alerts            recent alert history
POST /alerts/test       send a test Slack alert
GET  /docs              Swagger UI

Running tests

# unit tests (no AWS needed)
pytest tests/unit/ -v

# integration tests (needs LocalStack running)
pytest tests/integration/ -v

Deploying to real AWS

Remove AWS_ENDPOINT_URL from .env, configure your AWS credentials, then:

cd infrastructure/
terraform init && terraform apply
uvicorn src.main:app --host 0.0.0.0 --port 8000

Project layout

src/
  config.py     settings management
  models.py     data models (QueueMetrics, LambdaMetrics, Alert, etc.)
  collector.py  polls metrics from AWS
  alerter.py    threshold checks + Slack/SNS dispatch
  main.py       FastAPI app + background poll loop

scripts/
  bootstrap.py  set up LocalStack resources
  demo.py       generate test traffic

tests/
  unit/         fast tests, no AWS (uses moto)
  integration/  against LocalStack

dashboard/
  index.html    the frontend

infrastructure/
  main.tf       Terraform for real AWS

Known issues / TODO

• alert history is in-memory only, resets on restart (DynamoDB persistence is wired up but not fully used yet)
• LocalStack doesn't support Lambda p99 percentile metrics, so duration_p99_ms falls back to max
• the dashboard auto-refreshes every 15s which is fine for demos but could be smarter with SSE

Contributing

PRs welcome. Please add tests for any new alert types.

---

built by Aditya Ganti