bede/tfm_ainventory
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(phase-6): remove scale testing plan, simplify to docker + runbook

Browse Source 
- Delete PLAN-02-SCALE-TESTING.md (scale testing deferred to v3)
- Rename PLAN-03-BACKUP-RUNBOOK to PLAN-02-OPERATIONAL-RUNBOOK
- Phase 6 now has 2 executable plans instead of 3
This commit is contained in:
Daniel Bedeleanu
2026-04-22 18:18:00 +03:00
parent 4ea9625928
commit 4e6f940b75
5 changed files with 391 additions and 801 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
.planning/phases/06-deployment-scale/PLAN-03-BACKUP-RUNBOOK.md → .planning/phases/06-deployment-scale/PLAN-02-OPERATIONAL-RUNBOOK.md
View File

800
.planning/phases/06-deployment-scale/PLAN-02-SCALE-TESTING.md
View File

@@ -1,800 +0,0 @@
# Phase 6, Plan 2: Scale Testing & Performance Optimization
---
**plan**: 06-deployment-scale/02-scale-testing
**feature**: Load testing infrastructure, performance baseline, bottleneck identification
**status**: Ready for execution
**estimated_tasks**: 6
**total_lines**: ~600 (load testing suite ~250, DB seeding ~100, metrics collection ~150, runbook ~100)
---
## Overview
This plan builds the infrastructure to validate that the system handles production load (10K items + 5 concurrent users) without degradation. It creates:
1. **Load testing suite** (Locust) — Simulates concurrent users performing realistic workflows
2. **Database seeding** — Populates 10K items with realistic categories and attributes
3. **Metrics collection** — Monitors CPU, memory, response times during load
4. **Baseline establishment** — Documents performance envelope for future comparisons
5. **Health automation** — Automated health check monitoring during load tests
**Success**: Load test runs to completion with <2s latency at 5 concurrent users; baseline metrics published.
---
## Tasks
### Task 1: Create Load Testing Framework (Locust)
**File**: `backend/tests/load_test.py` (new, ~250 lines)
**Status**: Ready
**Description**: Locust-based load testing simulating realistic field workflows
**Content** (~250 lines):
```python
"""
Phase 6, Plan 2, Task 1: Load Testing Framework
Simulates realistic field workflows: scan → check-in/out → search → export
"""
from locust import HttpUser, task, between
from locust.contrib.fasthttp import FastHttpUser
import random
import time
class InventoryUser(FastHttpUser):
"""Simulates a field operator using the inventory system."""
wait_time = between(2, 5) # 2-5 seconds between actions
def on_start(self):
"""Login before starting tasks."""
response = self.client.post("/auth/login", json={
"username": "testuser",
"password": "testpass"
}, catch_response=True)
if response.status_code == 200:
self.token = response.json().get("access_token")
self.headers = {"Authorization": f"Bearer {self.token}"}
else:
response.failure(f"Login failed: {response.status_code}")
@task(3)
def search_item(self):
"""Search for an item (most common operation)."""
query = f"item-{random.randint(1, 10000)}"
response = self.client.get(
f"/search?q={query}",
headers=self.headers,
name="/search",
catch_response=True
)
if response.status_code == 200:
response.success()
else:
response.failure(f"Search failed: {response.status_code}")
@task(2)
def check_in_item(self):
"""Check in an item (adjust quantity +1)."""
item_id = random.randint(1, 10000)
response = self.client.patch(
f"/items/{item_id}",
json={"quantity": random.randint(1, 100)},
headers=self.headers,
name="/items/{itemId} [PATCH]",
catch_response=True
)
if response.status_code in [200, 404]: # 404 expected for some items
response.success()
else:
response.failure(f"Check-in failed: {response.status_code}")
@task(1)
def export_inventory(self):
"""Export inventory snapshot (less frequent)."""
response = self.client.get(
"/admin/exports/inventory",
headers=self.headers,
name="/admin/exports/inventory",
catch_response=True
)
if response.status_code in [200, 202]:
response.success()
else:
response.failure(f"Export failed: {response.status_code}")
@task(1)
def get_health(self):
"""Health check (baseline)."""
response = self.client.get(
"/health",
name="/health",
catch_response=True
)
if response.status_code == 200:
response.success()
else:
response.failure(f"Health check failed: {response.status_code}")
class AdminUser(FastHttpUser):
"""Simulates an admin performing dashboard operations."""
wait_time = between(5, 10)
def on_start(self):
"""Login as admin."""
response = self.client.post("/auth/login", json={
"username": "admin",
"password": "adminpass"
}, catch_response=True)
if response.status_code == 200:
self.token = response.json().get("access_token")
self.headers = {"Authorization": f"Bearer {self.token}"}
else:
response.failure(f"Admin login failed: {response.status_code}")
@task(2)
def list_items(self):
"""List items with pagination."""
skip = random.randint(0, 9900)
response = self.client.get(
f"/items?skip={skip}&limit=50",
headers=self.headers,
name="/items [paginated]",
catch_response=True
)
if response.status_code == 200:
response.success()
else:
response.failure(f"List items failed: {response.status_code}")
@task(1)
def get_audit_logs(self):
"""Retrieve audit logs."""
response = self.client.get(
"/admin/audit-logs?limit=100",
headers=self.headers,
name="/admin/audit-logs",
catch_response=True
)
if response.status_code == 200:
response.success()
else:
response.failure(f"Audit logs failed: {response.status_code}")
```
**Acceptance Criteria**:
- [ ] File uses Locust FastHttpUser (efficient)
- [ ] Simulates 5 realistic workflows (search, check-in, export, health, admin)
- [ ] Includes weight distribution (3:2:1 for common:moderate:rare)
- [ ] Can spawn multiple user types concurrently
- [ ] Task names are descriptive for reporting
**Testing**:
```bash
cd backend/tests
locust -f load_test.py --host=http://localhost:8000 --users=5 --spawn-rate=1 --run-time=5m
# Monitor: Response times, failure rates, requests/sec
```
---
### Task 2: Database Seeding Script (10K Items)
**File**: `scripts/seed_load_test_db.py` (new, ~100 lines)
**Status**: Ready
**Description**: Populate database with 10K realistic items for load testing
**Content** (~100 lines):
```python
#!/usr/bin/env python3
"""
Phase 6, Plan 2, Task 2: Database Seeding for Load Testing
Creates 10K items with realistic categories, part numbers, and barcodes.
"""
import sqlite3
import sys
from pathlib import Path
from datetime import datetime
import random
import string
DB_PATH = Path(__file__).parent.parent / "data" / "inventory.db"
CATEGORIES = [
"Electronics", "Computer Hardware", "Peripherals", "Cables & Adapters",
"Power Supplies", "Storage Devices", "Memory", "Processors",
"Networking", "Tools & Accessories", "Spare Parts"
]
ITEM_TYPES = [
"Hard Drive", "SSD", "RAM", "GPU", "CPU", "Motherboard",
"Network Card", "Power Supply", "Cable", "Connector",
"Screwdriver Set", "Thermal Paste", "PCIe Card", "USB Hub"
]
def generate_barcode():
"""Generate realistic 12-digit EAN barcode."""
return ''.join(random.choices(string.digits, k=12))
def generate_part_number():
"""Generate realistic part number."""
prefix = ''.join(random.choices(string.ascii_uppercase, k=3))
number = ''.join(random.choices(string.digits, k=6))
return f"{prefix}-{number}"
def seed_items(count=10000):
"""Create test items in database."""
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
print(f"Seeding {count} items...")
for i in range(1, count + 1):
item_name = f"item-{i:05d}"
category = random.choice(CATEGORIES)
item_type = random.choice(ITEM_TYPES)
quantity = random.randint(0, 100)
barcode = generate_barcode()
part_number = generate_part_number()
created_at = datetime.utcnow().isoformat()
updated_at = created_at
try:
cursor.execute("""
INSERT INTO items
(name, category, item_type, quantity, barcode, part_number, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (item_name, category, item_type, quantity, barcode, part_number, created_at, updated_at))
if i % 1000 == 0:
print(f" Created {i}/{count} items...")
conn.commit()
except sqlite3.IntegrityError as e:
print(f" Warning: Duplicate barcode {barcode}, retrying...")
cursor.execute("""
INSERT INTO items
(name, category, item_type, quantity, barcode, part_number, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (item_name, category, item_type, quantity, generate_barcode(), part_number, created_at, updated_at))
conn.commit()
conn.close()
print(f"Seeded {count} items successfully.")
if __name__ == "__main__":
if not DB_PATH.exists():
print(f"Error: Database not found at {DB_PATH}")
sys.exit(1)
seed_items(int(sys.argv[1]) if len(sys.argv) > 1 else 10000)
```
**Acceptance Criteria**:
- [ ] Creates 10K items with realistic data
- [ ] Avoids barcode/part number collisions
- [ ] Runs in <5 minutes
- [ ] Items distributed across categories and types
- [ ] Script idempotent (safe to run multiple times)
**Testing**:
```bash
python scripts/seed_load_test_db.py 10000
# Verify in database
sqlite3 data/inventory.db "SELECT COUNT(*) FROM items" # Should show 10000
```
---
### Task 3: Metrics Collection & Monitoring
**File**: `scripts/collect_metrics.py` (new, ~150 lines)
**Status**: Ready
**Description**: Collect CPU, memory, disk, and request metrics during load test
**Content** (~150 lines):
```python
#!/usr/bin/env python3
"""
Phase 6, Plan 2, Task 3: Metrics Collection During Load Tests
Monitors system resources and API performance.
"""
import subprocess
import json
import time
import docker
import psutil
from datetime import datetime
from pathlib import Path
METRICS_DIR = Path(__file__).parent.parent / "metrics"
METRICS_DIR.mkdir(exist_ok=True)
class MetricsCollector:
"""Collects system and container metrics during load test."""
def __init__(self, output_file=None):
self.output_file = output_file or METRICS_DIR / f"metrics-{datetime.now().isoformat()}.json"
self.docker_client = docker.from_env()
self.metrics = []
def get_container_stats(self, container_name):
"""Get stats for a specific container."""
try:
container = self.docker_client.containers.get(container_name)
stats = container.stats(stream=False)
cpu_delta = stats['cpu_stats']['cpu_usage']['total_usage'] - \
stats['precpu_stats']['cpu_usage']['total_usage']
system_delta = stats['cpu_stats']['system_cpu_usage'] - \
stats['precpu_stats']['system_cpu_usage']
cpu_percent = (cpu_delta / system_delta) * 100.0
memory_usage = stats['memory_stats']['usage'] / (1024 ** 2) # MB
return {'cpu_percent': cpu_percent, 'memory_mb': memory_usage}
except Exception as e:
print(f"Error collecting stats for {container_name}: {e}")
return None
def collect(self):
"""Collect all metrics."""
timestamp = datetime.now().isoformat()
data = {'timestamp': timestamp, 'containers': {}}
# Backend stats
backend_stats = self.get_container_stats('tfm-inventory-backend-1')
if backend_stats:
data['containers']['backend'] = backend_stats
# Frontend stats
frontend_stats = self.get_container_stats('tfm-inventory-frontend-1')
if frontend_stats:
data['containers']['frontend'] = frontend_stats
# System-wide stats
data['system'] = {
'cpu_percent': psutil.cpu_percent(interval=0.1),
'memory_percent': psutil.virtual_memory().percent,
'disk_percent': psutil.disk_usage('/').percent
}
self.metrics.append(data)
return data
def run(self, duration_seconds=300, interval_seconds=5):
"""Collect metrics for specified duration."""
print(f"Collecting metrics for {duration_seconds}s at {interval_seconds}s intervals...")
end_time = time.time() + duration_seconds
while time.time() < end_time:
self.collect()
time.sleep(interval_seconds)
self.save()
def save(self):
"""Save metrics to JSON file."""
with open(self.output_file, 'w') as f:
json.dump(self.metrics, f, indent=2)
print(f"Metrics saved to {self.output_file}")
def summarize(self):
"""Print summary of metrics."""
if not self.metrics:
return
# Extract backend CPU/memory
backend_cpus = [m['containers']['backend']['cpu_percent']
for m in self.metrics if 'backend' in m['containers']]
backend_mems = [m['containers']['backend']['memory_mb']
for m in self.metrics if 'backend' in m['containers']]
print("\n=== Load Test Summary ===")
print(f"Duration: {len(self.metrics) * 5}s")
if backend_cpus:
print(f"Backend CPU: avg={sum(backend_cpus)/len(backend_cpus):.1f}%, max={max(backend_cpus):.1f}%")
if backend_mems:
print(f"Backend Memory: avg={sum(backend_mems)/len(backend_mems):.0f}MB, max={max(backend_mems):.0f}MB")
print(f"Metrics file: {self.output_file}")
if __name__ == "__main__":
collector = MetricsCollector()
collector.run(duration_seconds=300, interval_seconds=5)
collector.summarize()
```
**Acceptance Criteria**:
- [ ] Collects backend/frontend container stats
- [ ] Records CPU %, memory (MB), disk usage
- [ ] Saves to JSON with timestamps
- [ ] Runs independently of load test
- [ ] Summary shows min/max/avg metrics
**Testing**:
```bash
python scripts/collect_metrics.py &
# In another terminal, run load test
locust -f backend/tests/load_test.py --users=5 --run-time=5m
# Check metrics output
jq . metrics/metrics-*.json | head -50
```
---
### Task 4: Performance Baseline Report
**File**: `docs/PERFORMANCE_BASELINE.md` (new, ~100 lines)
**Status**: Ready
**Description**: Document system performance under load, establish target SLOs
**Content** (~100 lines):
```markdown
# Performance Baseline Report
**Test Date**: 2026-04-22
**Database Size**: 10K items
**Concurrent Users**: 5 (3 operators, 2 admins)
**Test Duration**: 10 minutes
## System Configuration
- Backend: 2 CPU cores, 2GB RAM
- Frontend: 1 CPU core, 512MB RAM
- Database: SQLite with WAL mode enabled
## Load Test Results
### Response Times (p50/p95/p99)
| Endpoint | p50 (ms) | p95 (ms) | p99 (ms) | Status |
|----------|----------|----------|----------|--------|
| GET /health | 10 | 15 | 25 | ✓ Pass |
| GET /search | 120 | 350 | 500 | ✓ Pass |
| PATCH /items/{id} | 80 | 200 | 350 | ✓ Pass |
| GET /items (paginated) | 100 | 250 | 400 | ✓ Pass |
| POST /admin/exports | 150 | 400 | 800 | ⚠ At limit |
### Resource Utilization
| Resource | Avg | Peak | Status |
|----------|-----|------|--------|
| Backend CPU | 35% | 62% | ✓ Safe |
| Backend Memory | 480MB | 620MB | ✓ Safe |
| Database Lock Contention | Low | Medium | ✓ Acceptable |
| Disk I/O | <5% | 12% | ✓ Safe |
### Throughput
- Requests/second: 25-30
- Successful requests: 98.5%
- Failed requests: 1.5% (mostly intentional 404s)
- Sync reliability: 99.7%
## Baseline SLOs (Service Level Objectives)
We commit to the following performance targets:
```
- Search <500ms p95
- Item check-in <350ms p95
- Export start <1s
- Health check <50ms p99
- Sync success rate >99%
```
## Scaling Recommendations
**Current Capacity**: 5 concurrent users, 10K items
**Headroom**: ~30% (can handle 6-7 users before degradation)
**To Support 20+ Users**:
1. Increase backend memory to 4GB
2. Implement query caching (Redis optional)
3. Add read replicas for listing/search operations
4. Monitor database lock contention
**Database Optimization Candidates**:
- Index on (category, item_type) for filtered searches
- Partial index on active items (quantity > 0)
- WAL checkpoint tuning
## Next Steps
1. [ ] Monitor production metrics vs. baseline
2. [ ] Run load test weekly to track regressions
3. [ ] Investigate any p95 >600ms (potential bottleneck)
4. [ ] Re-baseline after major feature additions
```
**Acceptance Criteria**:
- [ ] Includes actual load test results (p50/p95/p99)
- [ ] Documents resource usage
- [ ] Establishes clear SLOs
- [ ] Provides scaling recommendations
- [ ] Baseline values are realistic and achievable
---
### Task 5: Automated Health Check Monitoring
**File**: `scripts/health_monitor.py` (new, ~80 lines)
**Status**: Ready
**Description**: Monitor service health during load tests, alert on degradation
**Content** (~80 lines):
```python
#!/usr/bin/env python3
"""
Phase 6, Plan 2, Task 5: Health Check Monitoring
Continuously monitors service health and alerts if degradation detected.
"""
import requests
import time
import sys
from datetime import datetime
BACKEND_URL = "http://localhost:8000"
FRONTEND_URL = "http://localhost:3000"
CHECK_INTERVAL = 5 # seconds
ALERT_THRESHOLD = 1000 # ms
def check_backend():
"""Check backend health."""
try:
start = time.time()
response = requests.get(f"{BACKEND_URL}/health", timeout=5)
duration = (time.time() - start) * 1000
status = "✓" if response.status_code == 200 else "✗"
return {
'status': response.status_code,
'duration_ms': duration,
'healthy': response.status_code == 200 and duration < ALERT_THRESHOLD,
'display': f"{status} Backend {response.status_code} ({duration:.0f}ms)"
}
except Exception as e:
return {
'status': 0,
'duration_ms': 0,
'healthy': False,
'display': f"✗ Backend error: {e}"
}
def check_frontend():
"""Check frontend health."""
try:
start = time.time()
response = requests.get(f"{FRONTEND_URL}/", timeout=5)
duration = (time.time() - start) * 1000
status = "✓" if response.status_code == 200 else "✗"
return {
'status': response.status_code,
'duration_ms': duration,
'healthy': response.status_code == 200,
'display': f"{status} Frontend {response.status_code} ({duration:.0f}ms)"
}
except Exception as e:
return {
'status': 0,
'duration_ms': 0,
'healthy': False,
'display': f"✗ Frontend error: {e}"
}
def monitor(duration_minutes=10):
"""Monitor health for specified duration."""
print(f"Starting health monitor for {duration_minutes} minutes...")
print("(Press Ctrl+C to stop)\n")
end_time = time.time() + (duration_minutes * 60)
failures = 0
checks = 0
while time.time() < end_time:
timestamp = datetime.now().strftime("%H:%M:%S")
backend = check_backend()
frontend = check_frontend()
print(f"[{timestamp}] {backend['display']} | {frontend['display']}")
if not (backend['healthy'] and frontend['healthy']):
failures += 1
checks += 1
time.sleep(CHECK_INTERVAL)
print(f"\n=== Monitor Summary ===")
print(f"Total checks: {checks}")
print(f"Failures: {failures} ({100*failures/checks:.1f}%)")
print(f"Success rate: {100*(1-failures/checks):.1f}%")
if __name__ == "__main__":
duration = int(sys.argv[1]) if len(sys.argv) > 1 else 10
try:
monitor(duration)
except KeyboardInterrupt:
print("\nMonitor stopped.")
```
**Acceptance Criteria**:
- [ ] Polls backend and frontend health endpoints
- [ ] Displays timestamp + status + response time
- [ ] Alerts if response time exceeds threshold
- [ ] Generates summary on completion
- [ ] Runs continuously for specified duration
**Testing**:
```bash
python scripts/health_monitor.py 5 # Monitor for 5 minutes
# Expected: All checks passing, response times stable
```
---
### Task 6: Load Test Execution Guide & Metrics Analysis
**File**: `docs/LOAD_TEST_GUIDE.md` (new, ~100 lines)
**Status**: Ready
**Description**: Step-by-step guide to run load tests and interpret results
**Content** (~100 lines):
```markdown
# Load Testing Guide
## Prerequisites
- System deployed via `./deploy.sh`
- Python 3.12+ with locust, requests, docker, psutil installed
```bash
pip install locust requests docker psutil
```
- 10K item database seeded
## Setup
### 1. Seed Database
```bash
python scripts/seed_load_test_db.py 10000
```
### 2. Start Health Monitor (Terminal 1)
```bash
python scripts/health_monitor.py 10
```
### 3. Start Metrics Collector (Terminal 2)
```bash
python scripts/collect_metrics.py
```
### 4. Run Locust Load Test (Terminal 3)
```bash
cd backend/tests
locust -f load_test.py \
--host=http://localhost:8000 \
--users=5 \
--spawn-rate=1 \
--run-time=5m \
--headless
```
## Interpreting Results
### Key Metrics
- **Response Time (p95)**: 95th percentile should be <500ms
- **Failure Rate**: Should be <1% (intentional 404s acceptable)
- **CPU Usage**: Peak should be <70%
- **Memory Usage**: Peak should be <1.5GB
### Success Criteria
- All checks pass ✓
- Load test completes without timeouts
- Metrics within baseline envelope
- No emergency restarts
### Troubleshooting
| Symptom | Cause | Fix |
|---------|-------|-----|
| High failure rate | Database lock contention | Increase WAL checkpoint interval |
| CPU >70% | Query inefficiency | Check slow query logs |
| Memory leak | Connection not released | Restart backend service |
| Timeouts after 5min | Resource exhaustion | Reduce concurrent users to 3 |
## Regression Detection
Compare latest metrics to baseline:
```bash
python -c "
import json
with open('metrics/baseline.json') as f:
baseline = json.load(f)
with open('metrics/latest.json') as f:
latest = json.load(f)
# Compare p95 response times, resource usage
"
```
## Next Steps
- [ ] Run test weekly to detect regressions
- [ ] Update baseline after major optimizations
- [ ] Investigate any p95 >500ms
- [ ] Document new bottlenecks in ARCHITECTURE.md
```
**Acceptance Criteria**:
- [ ] Step-by-step instructions for non-experts
- [ ] Clear success criteria with numbers
- [ ] Troubleshooting section covers common issues
- [ ] Links to metrics files and baseline report
- [ ] Interpretation guidance for non-technical teams
---
## Dependencies
**Upstream**:
- Plan 1 (Docker/Deployment) — Must complete first to have `deploy.sh`
- Phase 5 complete (all features implemented)
**Cross-Plan**:
- Plan 3 (Backup/Restore) uses baseline metrics as sanity check
**Blocked By**: None
---
## Testing Strategy
### Local Validation
```bash
# Test load testing framework
locust -f backend/tests/load_test.py --users=1 --run-time=30s
# Verify metrics collection
python scripts/collect_metrics.py
# Verify health monitoring
python scripts/health_monitor.py 1
```
### Integration Testing
```bash
# Full load test cycle
./deploy.sh production
python scripts/seed_load_test_db.py 10000
# Run all three monitoring tools in parallel
python scripts/health_monitor.py 10 &
python scripts/collect_metrics.py &
locust -f backend/tests/load_test.py --users=5 --run-time=5m --headless
```
### Baseline Validation
```bash
# Ensure results meet documented SLOs
# p50 search <250ms, p95 <500ms, p99 <800ms
# CPU <70%, Memory <1.5GB
```
---
## Success Metrics
- [ ] Load test framework (Locust) runs without errors
- [ ] Database seeding creates 10K items in <5 minutes
- [ ] Metrics collection records CPU/memory/disk during test
- [ ] Health monitor shows 99%+ success rate
- [ ] Performance baseline established and documented
- [ ] All tests meet SLOs (p95 <500ms, CPU <70%)
- [ ] Scaling recommendations documented
---
## Notes
- Load test uses realistic field workflows (search 3x, check-in 2x, export 1x)
- Metrics collected every 5 seconds (low overhead)
- Baseline includes p50/p95/p99 to show distribution, not just average
- SLOs are achievable with single-instance SQLite (no sharding needed)
- Weekly regression testing recommended post-launch
---
**Effort Estimate**: 18 hours (2-3 days)
**Dependencies**: Plan 1 complete (deploy.sh)
**Risk**: Low (testing infrastructure, no production changes)
---
Last updated: 2026-04-22 (Planning Phase)