chore: Major codebase cleanup - remove deprecated files and organize structure
- Docker: Delete 23 deprecated Dockerfiles, fix CI/CD to use Dockerfile.foxhunt-build - Config: Remove 36 .env files, keep 4 essential, delete config/environments/ - Docs: Archive 614 Wave D files to docs/archive/wave_d/, 95% reduction in root - Scripts: Delete 56 deprecated scripts, keep 58 production-critical (49% reduction) - Python: Organize 37 scripts into scripts/python/ subdirectories, delete ml/python/ - Build: Remove 1GB artifacts, delete old venvs, clean Python cache from git - Migrations: Delete deprecated directory (4,432 lines), remove duplicate database/migrations/ - Infrastructure: Delete deployment/ (61 files), docs/scripts/ (8 files) Total impact: ~2,500 files cleaned, 750MB+ space freed, zero production impact All deleted scripts backed up to archives. runpod/ and tests/runpod/ preserved. data_acquisition_service retained per user request.
This commit is contained in:
jgrusewski
228
scripts/python/testing/concurrent_connection_test.py
Executable file
228
scripts/python/testing/concurrent_connection_test.py
Executable file
@@ -0,0 +1,228 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Concurrent Connection Load Test for Foxhunt HFT Trading System
|
||||
Tests system behavior under 10, 50, 100, and 200+ concurrent gRPC connections
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
import statistics
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Optional
|
||||
import grpc
|
||||
import sys
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
# JWT token for authentication (same as used in previous tests)
|
||||
JWT_TOKEN = "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0X3VzZXIiLCJleHAiOjk5OTk5OTk5OTksImp0aSI6InRlc3RfdXNlciIsInJvbGVzIjpbInRyYWRlciJdLCJwZXJtaXNzaW9ucyI6WyJ0cmFkZSJdfQ.kHXiGFA0JjGmW66SiFRTUvzd2S6mOGN8pDfr9VJ8mAM"
|
||||
|
||||
@dataclass
|
||||
class ConnectionMetrics:
|
||||
"""Metrics for a single connection attempt"""
|
||||
connection_time_ms: float
|
||||
request_latency_ms: float
|
||||
success: bool
|
||||
error_message: Optional[str] = None
|
||||
|
||||
@dataclass
|
||||
class LoadTestResult:
|
||||
"""Aggregated results for a load test run"""
|
||||
concurrent_connections: int
|
||||
total_requests: int
|
||||
successful_requests: int
|
||||
failed_requests: int
|
||||
connection_time_p50: float
|
||||
connection_time_p95: float
|
||||
connection_time_p99: float
|
||||
latency_p50: float
|
||||
latency_p95: float
|
||||
latency_p99: float
|
||||
error_rate: float
|
||||
total_duration_secs: float
|
||||
throughput_rps: float
|
||||
|
||||
async def execute_single_connection(endpoint: str, connection_id: int) -> ConnectionMetrics:
|
||||
"""Execute a single connection and request to test connection behavior"""
|
||||
connection_start = time.time()
|
||||
|
||||
try:
|
||||
# Create gRPC channel with connection settings
|
||||
channel = grpc.aio.insecure_channel(
|
||||
endpoint,
|
||||
options=[
|
||||
('grpc.max_receive_message_length', 100 * 1024 * 1024),
|
||||
('grpc.max_send_message_length', 100 * 1024 * 1024),
|
||||
('grpc.keepalive_time_ms', 30000),
|
||||
('grpc.keepalive_timeout_ms', 20000),
|
||||
('grpc.http2.max_pings_without_data', 0),
|
||||
('grpc.keepalive_permit_without_calls', 1),
|
||||
]
|
||||
)
|
||||
|
||||
connection_time = (time.time() - connection_start) * 1000 # ms
|
||||
|
||||
# Execute a simple unary RPC call to test the connection
|
||||
request_start = time.time()
|
||||
|
||||
# Simple channel state check
|
||||
await channel.channel_ready()
|
||||
|
||||
request_latency = (time.time() - request_start) * 1000 # ms
|
||||
|
||||
await channel.close()
|
||||
|
||||
return ConnectionMetrics(
|
||||
connection_time_ms=connection_time,
|
||||
request_latency_ms=request_latency,
|
||||
success=True,
|
||||
error_message=None
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
connection_time = (time.time() - connection_start) * 1000
|
||||
return ConnectionMetrics(
|
||||
connection_time_ms=connection_time,
|
||||
request_latency_ms=0,
|
||||
success=False,
|
||||
error_message=str(e)
|
||||
)
|
||||
|
||||
async def run_concurrent_load_test(endpoint: str, concurrent_connections: int) -> LoadTestResult:
|
||||
"""Run load test with specified number of concurrent connections"""
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Testing with {concurrent_connections} concurrent connections")
|
||||
print(f"{'='*60}")
|
||||
|
||||
test_start = time.time()
|
||||
|
||||
# Create concurrent tasks
|
||||
tasks = [
|
||||
execute_single_connection(endpoint, i)
|
||||
for i in range(concurrent_connections)
|
||||
]
|
||||
|
||||
# Execute all tasks concurrently
|
||||
metrics = await asyncio.gather(*tasks)
|
||||
|
||||
total_duration = time.time() - test_start
|
||||
|
||||
# Calculate statistics
|
||||
connection_times = [m.connection_time_ms for m in metrics]
|
||||
latencies = [m.request_latency_ms for m in metrics if m.success]
|
||||
|
||||
successful = sum(1 for m in metrics if m.success)
|
||||
failed = len(metrics) - successful
|
||||
error_rate = (failed / len(metrics)) * 100.0
|
||||
|
||||
def percentile(values: List[float], p: float) -> float:
|
||||
if not values:
|
||||
return 0.0
|
||||
sorted_values = sorted(values)
|
||||
idx = int((len(sorted_values) - 1) * p)
|
||||
return sorted_values[idx]
|
||||
|
||||
conn_p50 = percentile(connection_times, 0.50)
|
||||
conn_p95 = percentile(connection_times, 0.95)
|
||||
conn_p99 = percentile(connection_times, 0.99)
|
||||
|
||||
lat_p50 = percentile(latencies, 0.50) if latencies else 0
|
||||
lat_p95 = percentile(latencies, 0.95) if latencies else 0
|
||||
lat_p99 = percentile(latencies, 0.99) if latencies else 0
|
||||
|
||||
throughput = successful / total_duration if total_duration > 0 else 0
|
||||
|
||||
return LoadTestResult(
|
||||
concurrent_connections=concurrent_connections,
|
||||
total_requests=len(metrics),
|
||||
successful_requests=successful,
|
||||
failed_requests=failed,
|
||||
connection_time_p50=conn_p50,
|
||||
connection_time_p95=conn_p95,
|
||||
connection_time_p99=conn_p99,
|
||||
latency_p50=lat_p50,
|
||||
latency_p95=lat_p95,
|
||||
latency_p99=lat_p99,
|
||||
error_rate=error_rate,
|
||||
total_duration_secs=total_duration,
|
||||
throughput_rps=throughput
|
||||
)
|
||||
|
||||
def print_result(result: LoadTestResult):
|
||||
"""Print detailed results for a test run"""
|
||||
success_pct = (result.successful_requests / result.total_requests) * 100.0
|
||||
|
||||
print(f"\nResults for {result.concurrent_connections} concurrent connections:")
|
||||
print(f" Total Requests: {result.total_requests}")
|
||||
print(f" Successful: {result.successful_requests} ({success_pct:.1f}%)")
|
||||
print(f" Failed: {result.failed_requests} ({result.error_rate:.1f}%)")
|
||||
print(f" Duration: {result.total_duration_secs:.2f}s")
|
||||
print(f" Throughput: {result.throughput_rps:.2f} conn/s")
|
||||
print(f"\n Connection Times:")
|
||||
print(f" P50: {result.connection_time_p50:.2f}ms")
|
||||
print(f" P95: {result.connection_time_p95:.2f}ms")
|
||||
print(f" P99: {result.connection_time_p99:.2f}ms")
|
||||
print(f"\n Request Latencies:")
|
||||
print(f" P50: {result.latency_p50:.2f}ms")
|
||||
print(f" P95: {result.latency_p95:.2f}ms")
|
||||
print(f" P99: {result.latency_p99:.2f}ms")
|
||||
|
||||
async def main():
|
||||
"""Main test execution function"""
|
||||
print("\n┌" + "─"*70 + "┐")
|
||||
print("│ Foxhunt HFT Trading System - Concurrent Connection Load Test │")
|
||||
print("└" + "─"*70 + "┘")
|
||||
|
||||
# Test against Trading Service directly (port 50052)
|
||||
endpoint = "localhost:50052"
|
||||
test_levels = [10, 50, 100, 200]
|
||||
|
||||
all_results = []
|
||||
|
||||
for connections in test_levels:
|
||||
result = await run_concurrent_load_test(endpoint, connections)
|
||||
print_result(result)
|
||||
all_results.append(result)
|
||||
|
||||
# Add delay between tests
|
||||
if connections < 200:
|
||||
print("\nWaiting 10 seconds before next test...")
|
||||
await asyncio.sleep(10)
|
||||
|
||||
# Print summary table
|
||||
print("\n\n┌" + "─"*100 + "┐")
|
||||
print("│ SUMMARY TABLE - ALL TEST LEVELS" + " "*50 + "│")
|
||||
print("├" + "─"*100 + "┤")
|
||||
print("│ Connections │ Success │ Error % │ Conn P99 │ Lat P50 │ Lat P95 │ Lat P99 │ Throughput │")
|
||||
print("├" + "─"*100 + "┤")
|
||||
|
||||
for result in all_results:
|
||||
success_pct = (result.successful_requests / result.total_requests) * 100.0
|
||||
print(f"│ {result.concurrent_connections:11d} │ {success_pct:6.1f}% │ {result.error_rate:6.1f}% │ "
|
||||
f"{result.connection_time_p99:7.1f}ms │ {result.latency_p50:6.1f}ms │ "
|
||||
f"{result.latency_p95:6.1f}ms │ {result.latency_p99:6.1f}ms │ "
|
||||
f"{result.throughput_rps:8.1f} c/s │")
|
||||
|
||||
print("└" + "─"*100 + "┘")
|
||||
|
||||
# Determine PASS/FAIL based on success criteria
|
||||
result_100 = next((r for r in all_results if r.concurrent_connections == 100), None)
|
||||
|
||||
if result_100:
|
||||
pass_criteria = (
|
||||
result_100.error_rate < 1.0 and
|
||||
result_100.latency_p99 < 100
|
||||
)
|
||||
|
||||
if pass_criteria:
|
||||
print("\n✅ TEST PASSED - System successfully handled 100+ concurrent connections")
|
||||
print(" - Error rate < 1%")
|
||||
print(" - P99 latency < 100ms")
|
||||
else:
|
||||
print("\n❌ TEST FAILED - System did not meet success criteria")
|
||||
print(f" - Error rate: {result_100.error_rate:.2f}% (required: <1%)")
|
||||
print(f" - P99 latency: {result_100.latency_p99:.2f}ms (required: <100ms)")
|
||||
|
||||
return all_results
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
342
scripts/python/testing/db_load_test.py
Executable file
342
scripts/python/testing/db_load_test.py
Executable file
@@ -0,0 +1,342 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
PostgreSQL Load Test for Foxhunt Trading System
|
||||
Tests database performance under increasing concurrent load
|
||||
"""
|
||||
|
||||
import psycopg2
|
||||
import psycopg2.pool
|
||||
import time
|
||||
import random
|
||||
import multiprocessing as mp
|
||||
from datetime import datetime
|
||||
from typing import Dict, List, Tuple
|
||||
|
||||
DB_CONFIG = {
|
||||
'host': 'localhost',
|
||||
'port': 5432,
|
||||
'database': 'foxhunt',
|
||||
'user': 'foxhunt',
|
||||
'password': 'foxhunt_dev_password'
|
||||
}
|
||||
|
||||
TEST_DURATION = 30 # seconds per scenario
|
||||
SYMBOLS = ['BTC/USD', 'ETH/USD', 'SOL/USD']
|
||||
|
||||
def get_connection():
|
||||
"""Get database connection"""
|
||||
return psycopg2.connect(**DB_CONFIG)
|
||||
|
||||
def worker_process(worker_id: int, duration: int, result_queue: mp.Queue):
|
||||
"""Worker process that executes database operations"""
|
||||
conn = get_connection()
|
||||
conn.autocommit = True
|
||||
cursor = conn.cursor()
|
||||
|
||||
start_time = time.time()
|
||||
end_time = start_time + duration
|
||||
|
||||
inserts = 0
|
||||
selects = 0
|
||||
updates = 0
|
||||
complex_queries = 0
|
||||
errors = 0
|
||||
|
||||
while time.time() < end_time:
|
||||
try:
|
||||
operation = random.randint(1, 10)
|
||||
symbol = random.choice(SYMBOLS)
|
||||
|
||||
# 40% INSERT
|
||||
if operation <= 4:
|
||||
cursor.execute("""
|
||||
INSERT INTO orders (account_id, symbol, side, order_type, quantity,
|
||||
limit_price, venue, created_at, updated_at)
|
||||
VALUES (%s, %s, 'buy', 'limit', 100000000, 5000000000000, 'load_test',
|
||||
EXTRACT(EPOCH FROM NOW())*1000000000,
|
||||
EXTRACT(EPOCH FROM NOW())*1000000000)
|
||||
""", (f'worker_{worker_id}', symbol))
|
||||
inserts += 1
|
||||
|
||||
# 30% SELECT
|
||||
elif operation <= 7:
|
||||
cursor.execute("""
|
||||
SELECT id, status, quantity, filled_quantity
|
||||
FROM orders
|
||||
WHERE symbol = %s
|
||||
ORDER BY created_at DESC
|
||||
LIMIT 10
|
||||
""", (symbol,))
|
||||
cursor.fetchall()
|
||||
selects += 1
|
||||
|
||||
# 20% UPDATE
|
||||
elif operation <= 9:
|
||||
cursor.execute("""
|
||||
UPDATE orders
|
||||
SET status = 'partially_filled',
|
||||
filled_quantity = filled_quantity + 10000000,
|
||||
updated_at = EXTRACT(EPOCH FROM NOW())*1000000000
|
||||
WHERE id = (
|
||||
SELECT id FROM orders
|
||||
WHERE status = 'pending' AND symbol = %s
|
||||
LIMIT 1
|
||||
)
|
||||
""", (symbol,))
|
||||
updates += 1
|
||||
|
||||
# 10% Complex query
|
||||
else:
|
||||
cursor.execute("""
|
||||
SELECT symbol,
|
||||
COUNT(*) as order_count,
|
||||
SUM(quantity) as total_quantity,
|
||||
AVG(limit_price) as avg_price
|
||||
FROM orders
|
||||
WHERE created_at > EXTRACT(EPOCH FROM (NOW() - INTERVAL '1 hour'))*1000000000
|
||||
GROUP BY symbol
|
||||
ORDER BY order_count DESC
|
||||
""")
|
||||
cursor.fetchall()
|
||||
complex_queries += 1
|
||||
|
||||
except Exception as e:
|
||||
errors += 1
|
||||
print(f"Worker {worker_id} error: {e}")
|
||||
|
||||
cursor.close()
|
||||
conn.close()
|
||||
|
||||
result_queue.put({
|
||||
'worker_id': worker_id,
|
||||
'inserts': inserts,
|
||||
'selects': selects,
|
||||
'updates': updates,
|
||||
'complex': complex_queries,
|
||||
'errors': errors
|
||||
})
|
||||
|
||||
def run_load_test(num_workers: int, test_name: str, duration: int) -> Dict:
|
||||
"""Run load test with specified number of workers"""
|
||||
print(f"\n{'='*60}")
|
||||
print(f"TEST: {test_name} ({num_workers} concurrent workers)")
|
||||
print(f"{'='*60}")
|
||||
|
||||
result_queue = mp.Queue()
|
||||
processes = []
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
# Launch workers
|
||||
for i in range(num_workers):
|
||||
p = mp.Process(target=worker_process, args=(i, duration, result_queue))
|
||||
p.start()
|
||||
processes.append(p)
|
||||
|
||||
# Wait for all workers
|
||||
for p in processes:
|
||||
p.join()
|
||||
|
||||
end_time = time.time()
|
||||
actual_duration = end_time - start_time
|
||||
|
||||
# Collect results
|
||||
total_inserts = 0
|
||||
total_selects = 0
|
||||
total_updates = 0
|
||||
total_complex = 0
|
||||
total_errors = 0
|
||||
|
||||
while not result_queue.empty():
|
||||
result = result_queue.get()
|
||||
total_inserts += result['inserts']
|
||||
total_selects += result['selects']
|
||||
total_updates += result['updates']
|
||||
total_complex += result['complex']
|
||||
total_errors += result['errors']
|
||||
|
||||
total_ops = total_inserts + total_selects + total_updates + total_complex
|
||||
tps = total_ops / actual_duration if actual_duration > 0 else 0
|
||||
|
||||
results = {
|
||||
'test_name': test_name,
|
||||
'num_workers': num_workers,
|
||||
'duration': actual_duration,
|
||||
'total_ops': total_ops,
|
||||
'inserts': total_inserts,
|
||||
'selects': total_selects,
|
||||
'updates': total_updates,
|
||||
'complex': total_complex,
|
||||
'errors': total_errors,
|
||||
'tps': tps
|
||||
}
|
||||
|
||||
# Print results
|
||||
print(f"Duration: {actual_duration:.2f} seconds")
|
||||
print(f"Total Operations: {total_ops}")
|
||||
print(f" - INSERTs: {total_inserts}")
|
||||
print(f" - SELECTs: {total_selects}")
|
||||
print(f" - UPDATEs: {total_updates}")
|
||||
print(f" - Complex: {total_complex}")
|
||||
print(f" - Errors: {total_errors}")
|
||||
print(f"Transactions per Second (TPS): {tps:.2f}")
|
||||
|
||||
# Check connection pool and locks
|
||||
try:
|
||||
conn = get_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
print("\n=== Connection Pool Status ===")
|
||||
cursor.execute("SELECT state, COUNT(*) FROM pg_stat_activity WHERE datname='foxhunt' GROUP BY state")
|
||||
for row in cursor.fetchall():
|
||||
print(f" {row[0]}: {row[1]}")
|
||||
|
||||
print("\n=== Lock Contention ===")
|
||||
cursor.execute("SELECT COUNT(*) FROM pg_locks WHERE NOT granted")
|
||||
locks_waiting = cursor.fetchone()[0]
|
||||
print(f" Locks Waiting: {locks_waiting}")
|
||||
|
||||
print("\n=== Deadlocks ===")
|
||||
cursor.execute("SELECT deadlocks FROM pg_stat_database WHERE datname='foxhunt'")
|
||||
deadlocks = cursor.fetchone()[0]
|
||||
print(f" Total Deadlocks: {deadlocks}")
|
||||
|
||||
cursor.close()
|
||||
conn.close()
|
||||
except Exception as e:
|
||||
print(f"Error checking pool status: {e}")
|
||||
|
||||
return results
|
||||
|
||||
def print_database_health():
|
||||
"""Print database health metrics"""
|
||||
conn = get_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
print("\n=== Database Configuration ===")
|
||||
cursor.execute("SHOW max_connections")
|
||||
print(f" Max Connections: {cursor.fetchone()[0]}")
|
||||
|
||||
cursor.execute("SHOW shared_buffers")
|
||||
print(f" Shared Buffers: {cursor.fetchone()[0]}")
|
||||
|
||||
cursor.execute("SHOW synchronous_commit")
|
||||
print(f" Synchronous Commit: {cursor.fetchone()[0]}")
|
||||
|
||||
print("\n=== Cache Hit Ratio ===")
|
||||
cursor.execute("""
|
||||
SELECT (sum(blks_hit)::float / NULLIF(sum(blks_hit) + sum(blks_read), 0) * 100)::numeric(5,2)
|
||||
FROM pg_stat_database WHERE datname='foxhunt'
|
||||
""")
|
||||
print(f" {cursor.fetchone()[0]}%")
|
||||
|
||||
print("\n=== Baseline Table Counts ===")
|
||||
cursor.execute("SELECT COUNT(*) FROM orders")
|
||||
print(f" Orders: {cursor.fetchone()[0]}")
|
||||
|
||||
cursor.close()
|
||||
conn.close()
|
||||
|
||||
def print_final_analysis():
|
||||
"""Print final analysis"""
|
||||
conn = get_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
print("\n" + "="*60)
|
||||
print("FINAL ANALYSIS")
|
||||
print("="*60)
|
||||
|
||||
print("\n=== Table Statistics ===")
|
||||
cursor.execute("""
|
||||
SELECT relname, n_tup_ins, n_tup_upd, n_tup_del,
|
||||
seq_scan, idx_scan,
|
||||
CASE WHEN seq_scan + idx_scan > 0
|
||||
THEN (idx_scan::float / (seq_scan + idx_scan) * 100)::numeric(5,2)
|
||||
ELSE 0 END as idx_scan_pct
|
||||
FROM pg_stat_user_tables
|
||||
WHERE relname IN ('orders', 'executions', 'fills', 'positions')
|
||||
ORDER BY n_tup_ins DESC
|
||||
""")
|
||||
print(f"{'Table':<12} {'Inserts':<10} {'Updates':<10} {'Deletes':<10} {'Seq Scan':<10} {'Idx Scan':<10} {'Idx %':<8}")
|
||||
print("-" * 80)
|
||||
for row in cursor.fetchall():
|
||||
print(f"{row[0]:<12} {row[1]:<10} {row[2]:<10} {row[3]:<10} {row[4]:<10} {row[5]:<10} {row[6]:<8}")
|
||||
|
||||
print("\n=== Index Usage (Top 5) ===")
|
||||
cursor.execute("""
|
||||
SELECT tablename, indexname, idx_scan, idx_tup_read
|
||||
FROM pg_stat_user_indexes
|
||||
WHERE tablename IN ('orders', 'executions', 'fills', 'positions')
|
||||
AND idx_scan > 0
|
||||
ORDER BY idx_scan DESC
|
||||
LIMIT 5
|
||||
""")
|
||||
print(f"{'Table':<12} {'Index':<30} {'Scans':<10} {'Rows Read':<12}")
|
||||
print("-" * 70)
|
||||
for row in cursor.fetchall():
|
||||
print(f"{row[0]:<12} {row[1]:<30} {row[2]:<10} {row[3]:<12}")
|
||||
|
||||
print("\n=== Database Size ===")
|
||||
cursor.execute("""
|
||||
SELECT pg_size_pretty(pg_database_size('foxhunt'))
|
||||
""")
|
||||
print(f" Total Size: {cursor.fetchone()[0]}")
|
||||
|
||||
print("\n=== Table Sizes ===")
|
||||
cursor.execute("""
|
||||
SELECT relname,
|
||||
pg_size_pretty(pg_total_relation_size(relid)) AS total_size
|
||||
FROM pg_stat_user_tables
|
||||
WHERE relname IN ('orders', 'executions', 'fills', 'positions')
|
||||
ORDER BY pg_total_relation_size(relid) DESC
|
||||
""")
|
||||
for row in cursor.fetchall():
|
||||
print(f" {row[0]}: {row[1]}")
|
||||
|
||||
print("\n=== Final Cache Hit Ratio ===")
|
||||
cursor.execute("""
|
||||
SELECT (sum(blks_hit)::float / NULLIF(sum(blks_hit) + sum(blks_read), 0) * 100)::numeric(5,2)
|
||||
FROM pg_stat_database WHERE datname='foxhunt'
|
||||
""")
|
||||
print(f" {cursor.fetchone()[0]}%")
|
||||
|
||||
cursor.close()
|
||||
conn.close()
|
||||
|
||||
def main():
|
||||
"""Main test execution"""
|
||||
print("="*60)
|
||||
print("PostgreSQL Load Test - Foxhunt Trading System")
|
||||
print(f"Test Duration: {TEST_DURATION} seconds per scenario")
|
||||
print(f"Start Time: {datetime.now()}")
|
||||
print("="*60)
|
||||
|
||||
# Initial health check
|
||||
print_database_health()
|
||||
|
||||
# Run tests with increasing concurrency
|
||||
results = []
|
||||
results.append(run_load_test(1, "BASELINE (Single Worker)", TEST_DURATION))
|
||||
results.append(run_load_test(10, "MODERATE LOAD", TEST_DURATION))
|
||||
results.append(run_load_test(50, "HIGH LOAD", TEST_DURATION))
|
||||
results.append(run_load_test(100, "STRESS TEST", TEST_DURATION))
|
||||
|
||||
# Final analysis
|
||||
print_final_analysis()
|
||||
|
||||
# Summary
|
||||
print("\n" + "="*60)
|
||||
print("TEST SUMMARY")
|
||||
print("="*60)
|
||||
print(f"{'Test':<25} {'Workers':<10} {'TPS':<12} {'Errors':<10}")
|
||||
print("-" * 60)
|
||||
for r in results:
|
||||
print(f"{r['test_name']:<25} {r['num_workers']:<10} {r['tps']:<12.2f} {r['errors']:<10}")
|
||||
|
||||
print("\n" + "="*60)
|
||||
print(f"Test Completed: {datetime.now()}")
|
||||
print("="*60)
|
||||
|
||||
if __name__ == '__main__':
|
||||
mp.set_start_method('fork')
|
||||
main()
|
||||
451
scripts/python/testing/load_test.py
Executable file
451
scripts/python/testing/load_test.py
Executable file
@@ -0,0 +1,451 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Comprehensive Load Test for Trading Service
|
||||
|
||||
Tests:
|
||||
1. Baseline latency (single client)
|
||||
2. Concurrent connections (100 clients)
|
||||
3. Sustained load (5+ minutes)
|
||||
4. Database performance
|
||||
5. Resource monitoring
|
||||
6. Production readiness assessment
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
import statistics
|
||||
import sys
|
||||
import uuid
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from typing import List, Tuple
|
||||
import requests
|
||||
|
||||
# Metrics storage
|
||||
class PerformanceMetrics:
|
||||
def __init__(self):
|
||||
self.latencies_ms = []
|
||||
self.successful = 0
|
||||
self.failed = 0
|
||||
self.start_time = None
|
||||
self.end_time = None
|
||||
|
||||
def record_success(self, latency_ms: float):
|
||||
self.latencies_ms.append(latency_ms)
|
||||
self.successful += 1
|
||||
|
||||
def record_failure(self):
|
||||
self.failed += 1
|
||||
|
||||
def get_percentiles(self) -> Tuple[float, float, float, float, float]:
|
||||
if not self.latencies_ms:
|
||||
return (0, 0, 0, 0, 0)
|
||||
|
||||
sorted_lat = sorted(self.latencies_ms)
|
||||
n = len(sorted_lat)
|
||||
|
||||
return (
|
||||
sorted_lat[0], # min
|
||||
sorted_lat[n // 2], # p50
|
||||
sorted_lat[int(n * 0.95)], # p95
|
||||
sorted_lat[int(n * 0.99)], # p99
|
||||
sorted_lat[-1] # max
|
||||
)
|
||||
|
||||
def print_summary(self):
|
||||
duration = (self.end_time - self.start_time) if self.end_time and self.start_time else 0
|
||||
total = self.successful + self.failed
|
||||
success_rate = (self.successful / total * 100) if total > 0 else 0
|
||||
throughput = self.successful / duration if duration > 0 else 0
|
||||
|
||||
min_lat, p50, p95, p99, max_lat = self.get_percentiles()
|
||||
|
||||
print("\n" + "="*70)
|
||||
print(" TRADING SERVICE LOAD TEST RESULTS")
|
||||
print("="*70)
|
||||
print(f"Test Duration: {duration:.2f}s")
|
||||
print(f"Total Orders: {total}")
|
||||
print(f"Successful Orders: {self.successful} ({success_rate:.2f}%)")
|
||||
print(f"Failed Orders: {self.failed}")
|
||||
print(f"Throughput: {throughput:.0f} orders/sec")
|
||||
print("-"*70)
|
||||
print(" LATENCY METRICS")
|
||||
print("-"*70)
|
||||
print(f"Min Latency: {min_lat:.2f}ms")
|
||||
print(f"P50 Latency: {p50:.2f}ms")
|
||||
print(f"P95 Latency: {p95:.2f}ms")
|
||||
print(f"P99 Latency: {p99:.2f}ms")
|
||||
print(f"Max Latency: {max_lat:.2f}ms")
|
||||
print("="*70)
|
||||
|
||||
print("\n📊 PERFORMANCE ASSESSMENT:")
|
||||
|
||||
if throughput >= 10000:
|
||||
print(f"✅ Throughput target ACHIEVED: {throughput:.0f} orders/sec (target: 10K orders/sec)")
|
||||
elif throughput >= 5000:
|
||||
print(f"⚠️ Throughput ACCEPTABLE: {throughput:.0f} orders/sec (target: 10K orders/sec)")
|
||||
else:
|
||||
print(f"❌ Throughput BELOW target: {throughput:.0f} orders/sec (target: 10K orders/sec)")
|
||||
|
||||
if p99 < 100:
|
||||
print(f"✅ P99 latency EXCELLENT: {p99:.2f}ms (< 100ms)")
|
||||
elif p99 < 500:
|
||||
print(f"⚠️ P99 latency ACCEPTABLE: {p99:.2f}ms (< 500ms)")
|
||||
else:
|
||||
print(f"❌ P99 latency HIGH: {p99:.2f}ms (> 500ms)")
|
||||
|
||||
if success_rate >= 99.0:
|
||||
print(f"✅ Success rate EXCELLENT: {success_rate:.2f}%")
|
||||
elif success_rate >= 95.0:
|
||||
print(f"⚠️ Success rate ACCEPTABLE: {success_rate:.2f}%")
|
||||
else:
|
||||
print(f"❌ Success rate POOR: {success_rate:.2f}%")
|
||||
|
||||
|
||||
def submit_order_http(order_id: str, symbol: str) -> Tuple[bool, float]:
|
||||
"""Submit order via HTTP (fallback if gRPC unavailable)"""
|
||||
url = "http://localhost:8081/api/v1/orders"
|
||||
|
||||
payload = {
|
||||
"order_id": order_id,
|
||||
"symbol": symbol,
|
||||
"side": "buy",
|
||||
"order_type": "limit",
|
||||
"quantity": "1.0",
|
||||
"price": "50000.0",
|
||||
"time_in_force": "gtc"
|
||||
}
|
||||
|
||||
start = time.time()
|
||||
try:
|
||||
response = requests.post(url, json=payload, timeout=5)
|
||||
latency_ms = (time.time() - start) * 1000
|
||||
return response.status_code == 200, latency_ms
|
||||
except Exception as e:
|
||||
latency_ms = (time.time() - start) * 1000
|
||||
return False, latency_ms
|
||||
|
||||
|
||||
async def test_1_baseline_latency():
|
||||
"""Test 1: Baseline latency with single client"""
|
||||
print("\n" + "="*70)
|
||||
print(" TEST 1: BASELINE LATENCY (Single Client)")
|
||||
print("="*70)
|
||||
|
||||
metrics = PerformanceMetrics()
|
||||
num_requests = 1000
|
||||
symbols = ["BTC/USD", "ETH/USD", "SOL/USD", "AVAX/USD", "MATIC/USD"]
|
||||
|
||||
print(f"📊 Sending {num_requests} orders sequentially...")
|
||||
|
||||
metrics.start_time = time.time()
|
||||
|
||||
for i in range(num_requests):
|
||||
order_id = str(uuid.uuid4())
|
||||
symbol = symbols[i % len(symbols)]
|
||||
|
||||
success, latency_ms = submit_order_http(order_id, symbol)
|
||||
|
||||
if success:
|
||||
metrics.record_success(latency_ms)
|
||||
else:
|
||||
metrics.record_failure()
|
||||
if metrics.failed <= 5:
|
||||
print(f"❌ Order {i} failed")
|
||||
|
||||
metrics.end_time = time.time()
|
||||
metrics.print_summary()
|
||||
|
||||
|
||||
async def test_2_concurrent_connections():
|
||||
"""Test 2: Concurrent connections (100 clients)"""
|
||||
print("\n" + "="*70)
|
||||
print(" TEST 2: CONCURRENT CONNECTIONS (100 Clients)")
|
||||
print("="*70)
|
||||
|
||||
num_clients = 100
|
||||
orders_per_client = 100
|
||||
metrics = PerformanceMetrics()
|
||||
symbols = ["BTC/USD", "ETH/USD", "SOL/USD", "AVAX/USD", "MATIC/USD"]
|
||||
|
||||
print(f"🚀 Spawning {num_clients} concurrent clients ({orders_per_client} orders each)...")
|
||||
|
||||
metrics.start_time = time.time()
|
||||
|
||||
def client_worker(client_id: int):
|
||||
for order_idx in range(orders_per_client):
|
||||
order_id = str(uuid.uuid4())
|
||||
symbol = symbols[(client_id * orders_per_client + order_idx) % len(symbols)]
|
||||
|
||||
success, latency_ms = submit_order_http(order_id, symbol)
|
||||
|
||||
if success:
|
||||
metrics.record_success(latency_ms)
|
||||
else:
|
||||
metrics.record_failure()
|
||||
|
||||
with ThreadPoolExecutor(max_workers=num_clients) as executor:
|
||||
futures = [executor.submit(client_worker, i) for i in range(num_clients)]
|
||||
for future in futures:
|
||||
future.result()
|
||||
|
||||
metrics.end_time = time.time()
|
||||
metrics.print_summary()
|
||||
|
||||
|
||||
async def test_3_sustained_load():
|
||||
"""Test 3: Sustained load (5 minutes)"""
|
||||
print("\n" + "="*70)
|
||||
print(" TEST 3: SUSTAINED LOAD (5 Minutes)")
|
||||
print("="*70)
|
||||
|
||||
test_duration_secs = 300
|
||||
num_clients = 50
|
||||
target_rate_per_client = 200 # 10K total / 50 clients = 200 per client
|
||||
|
||||
metrics = PerformanceMetrics()
|
||||
symbols = ["BTC/USD", "ETH/USD", "SOL/USD", "AVAX/USD", "MATIC/USD"]
|
||||
|
||||
print(f"🚀 Starting {num_clients} clients for {test_duration_secs} seconds...")
|
||||
print(f"🎯 Target: {num_clients * target_rate_per_client} orders/sec total")
|
||||
|
||||
shutdown_flag = [False]
|
||||
metrics.start_time = time.time()
|
||||
|
||||
def client_worker(client_id: int):
|
||||
order_count = 0
|
||||
delay_secs = 1.0 / target_rate_per_client
|
||||
|
||||
while not shutdown_flag[0]:
|
||||
order_id = str(uuid.uuid4())
|
||||
symbol = symbols[order_count % len(symbols)]
|
||||
|
||||
success, latency_ms = submit_order_http(order_id, symbol)
|
||||
|
||||
if success:
|
||||
metrics.record_success(latency_ms)
|
||||
else:
|
||||
metrics.record_failure()
|
||||
|
||||
order_count += 1
|
||||
time.sleep(delay_secs)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=num_clients) as executor:
|
||||
futures = [executor.submit(client_worker, i) for i in range(num_clients)]
|
||||
|
||||
# Run for specified duration
|
||||
time.sleep(test_duration_secs)
|
||||
shutdown_flag[0] = True
|
||||
|
||||
# Wait for all clients to finish
|
||||
for future in futures:
|
||||
try:
|
||||
future.result(timeout=5)
|
||||
except:
|
||||
pass
|
||||
|
||||
metrics.end_time = time.time()
|
||||
metrics.print_summary()
|
||||
|
||||
|
||||
async def test_4_database_performance():
|
||||
"""Test 4: Database performance"""
|
||||
print("\n" + "="*70)
|
||||
print(" TEST 4: DATABASE PERFORMANCE")
|
||||
print("="*70)
|
||||
|
||||
num_orders = 5000
|
||||
symbols = ["BTC/USD", "ETH/USD", "SOL/USD"]
|
||||
|
||||
print(f"📊 Submitting {num_orders} orders to measure database performance...")
|
||||
|
||||
start_time = time.time()
|
||||
success_count = 0
|
||||
failure_count = 0
|
||||
|
||||
for i in range(num_orders):
|
||||
order_id = str(uuid.uuid4())
|
||||
symbol = symbols[i % len(symbols)]
|
||||
|
||||
success, _ = submit_order_http(order_id, symbol)
|
||||
|
||||
if success:
|
||||
success_count += 1
|
||||
else:
|
||||
failure_count += 1
|
||||
if failure_count <= 5:
|
||||
print(f"❌ Order {i} failed")
|
||||
|
||||
duration = time.time() - start_time
|
||||
throughput = success_count / duration
|
||||
|
||||
print("\n📈 DATABASE PERFORMANCE:")
|
||||
print(f" Duration: {duration:.2f}s")
|
||||
print(f" Successful: {success_count}")
|
||||
print(f" Failed: {failure_count}")
|
||||
print(f" DB Writes/sec: {throughput:.0f}")
|
||||
|
||||
if throughput >= 2000:
|
||||
print(f"✅ Database performance EXCELLENT: {throughput:.0f} writes/sec")
|
||||
elif throughput >= 1000:
|
||||
print(f"⚠️ Database performance ACCEPTABLE: {throughput:.0f} writes/sec")
|
||||
else:
|
||||
print(f"❌ Database performance LOW: {throughput:.0f} writes/sec (expected >2000)")
|
||||
|
||||
|
||||
async def test_5_resource_monitoring():
|
||||
"""Test 5: Resource monitoring"""
|
||||
print("\n" + "="*70)
|
||||
print(" TEST 5: RESOURCE MONITORING")
|
||||
print("="*70)
|
||||
|
||||
# Check service health
|
||||
health_url = "http://localhost:8081/health"
|
||||
print(f"🏥 Checking service health at {health_url}...")
|
||||
|
||||
try:
|
||||
response = requests.get(health_url, timeout=5)
|
||||
print(f"✅ Health check response: {response.status_code}")
|
||||
print(f" Body: {response.text[:200]}")
|
||||
except Exception as e:
|
||||
print(f"⚠️ Health check failed: {e}")
|
||||
|
||||
# Check Prometheus metrics
|
||||
metrics_url = "http://localhost:9092/metrics"
|
||||
print(f"\n📊 Checking Prometheus metrics at {metrics_url}...")
|
||||
|
||||
try:
|
||||
response = requests.get(metrics_url, timeout=5)
|
||||
lines = [l for l in response.text.split('\n') if l and not l.startswith('#')]
|
||||
print(f"✅ Found {len(lines)} metric entries")
|
||||
|
||||
# Show key metrics
|
||||
for line in lines[:20]:
|
||||
if any(keyword in line for keyword in ['orders', 'latency', 'cpu', 'memory']):
|
||||
print(f" {line[:100]}")
|
||||
except Exception as e:
|
||||
print(f"⚠️ Metrics check failed: {e}")
|
||||
|
||||
|
||||
async def test_6_production_readiness():
|
||||
"""Test 6: Production readiness assessment"""
|
||||
print("\n" + "="*70)
|
||||
print(" TEST 6: PRODUCTION READINESS ASSESSMENT")
|
||||
print("="*70)
|
||||
|
||||
num_clients = 50
|
||||
orders_per_client = 200
|
||||
metrics = PerformanceMetrics()
|
||||
symbols = ["BTC/USD", "ETH/USD", "SOL/USD", "AVAX/USD", "MATIC/USD"]
|
||||
|
||||
print(f"🎯 Production simulation: {num_clients} clients, {orders_per_client} orders each")
|
||||
|
||||
metrics.start_time = time.time()
|
||||
|
||||
def client_worker(client_id: int):
|
||||
for order_idx in range(orders_per_client):
|
||||
order_id = str(uuid.uuid4())
|
||||
symbol = symbols[(client_id * orders_per_client + order_idx) % len(symbols)]
|
||||
|
||||
success, latency_ms = submit_order_http(order_id, symbol)
|
||||
|
||||
if success:
|
||||
metrics.record_success(latency_ms)
|
||||
else:
|
||||
metrics.record_failure()
|
||||
|
||||
with ThreadPoolExecutor(max_workers=num_clients) as executor:
|
||||
futures = [executor.submit(client_worker, i) for i in range(num_clients)]
|
||||
for future in futures:
|
||||
future.result()
|
||||
|
||||
metrics.end_time = time.time()
|
||||
metrics.print_summary()
|
||||
|
||||
# Production readiness criteria
|
||||
total = metrics.successful + metrics.failed
|
||||
success_rate = (metrics.successful / total * 100) if total > 0 else 0
|
||||
duration = metrics.end_time - metrics.start_time
|
||||
throughput = metrics.successful / duration if duration > 0 else 0
|
||||
_, _, _, p99, _ = metrics.get_percentiles()
|
||||
|
||||
print("\n🎯 PRODUCTION READINESS:")
|
||||
|
||||
passed = 0
|
||||
total_checks = 3
|
||||
|
||||
# Check 1: Success rate
|
||||
if success_rate >= 99.0:
|
||||
print(f"✅ Success rate: {success_rate:.2f}% (>= 99%)")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"❌ Success rate: {success_rate:.2f}% (< 99%)")
|
||||
|
||||
# Check 2: Throughput
|
||||
if throughput >= 5000.0:
|
||||
print(f"✅ Throughput: {throughput:.0f} orders/sec (>= 5000)")
|
||||
passed += 1
|
||||
elif throughput >= 3000.0:
|
||||
print(f"⚠️ Throughput: {throughput:.0f} orders/sec (>= 3000)")
|
||||
passed += 0.5
|
||||
else:
|
||||
print(f"❌ Throughput: {throughput:.0f} orders/sec (< 3000)")
|
||||
|
||||
# Check 3: P99 latency
|
||||
if p99 < 100.0:
|
||||
print(f"✅ P99 latency: {p99:.2f}ms (< 100ms)")
|
||||
passed += 1
|
||||
elif p99 < 500.0:
|
||||
print(f"⚠️ P99 latency: {p99:.2f}ms (< 500ms)")
|
||||
passed += 0.5
|
||||
else:
|
||||
print(f"❌ P99 latency: {p99:.2f}ms (>= 500ms)")
|
||||
|
||||
print(f"\n📊 OVERALL: {passed}/{total_checks} checks passed")
|
||||
|
||||
if passed >= 2.5:
|
||||
print("🎉 PRODUCTION READY!")
|
||||
elif passed >= 2.0:
|
||||
print("⚠️ Acceptable for production with monitoring")
|
||||
else:
|
||||
print("❌ Not ready for production deployment")
|
||||
|
||||
|
||||
async def main():
|
||||
"""Run all load tests"""
|
||||
print("\n" + "="*70)
|
||||
print(" FOXHUNT TRADING SERVICE - COMPREHENSIVE LOAD TEST")
|
||||
print("="*70)
|
||||
print("\nTarget: Trading Service at http://localhost:8081")
|
||||
print("gRPC Port: 50052 | Health Port: 8081 | Metrics Port: 9092")
|
||||
|
||||
tests = [
|
||||
("Test 1: Baseline Latency", test_1_baseline_latency),
|
||||
("Test 2: Concurrent Connections", test_2_concurrent_connections),
|
||||
("Test 4: Database Performance", test_4_database_performance),
|
||||
("Test 5: Resource Monitoring", test_5_resource_monitoring),
|
||||
("Test 6: Production Readiness", test_6_production_readiness),
|
||||
# Test 3 (sustained load) commented out for quick runs
|
||||
# ("Test 3: Sustained Load", test_3_sustained_load),
|
||||
]
|
||||
|
||||
for test_name, test_func in tests:
|
||||
try:
|
||||
print(f"\n\n{'='*70}")
|
||||
print(f"Starting: {test_name}")
|
||||
print(f"{'='*70}")
|
||||
await test_func()
|
||||
except KeyboardInterrupt:
|
||||
print("\n\n⚠️ Test interrupted by user")
|
||||
break
|
||||
except Exception as e:
|
||||
print(f"\n❌ Test failed with error: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
print("\n" + "="*70)
|
||||
print(" LOAD TEST SUITE COMPLETED")
|
||||
print("="*70)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
483
scripts/python/testing/sustained_load_test.py
Executable file
483
scripts/python/testing/sustained_load_test.py
Executable file
@@ -0,0 +1,483 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Sustained Load Test for Trading Service
|
||||
Wave 141 Phase 5 - Agent 262
|
||||
|
||||
Test Configuration:
|
||||
- Duration: 5 minutes (300 seconds)
|
||||
- Target: 1,000+ orders/minute (~16.67 orders/sec)
|
||||
- Monitoring: Time-series performance, resource usage, memory leaks
|
||||
- Symbols: BTC/USD, ETH/USD rotation
|
||||
|
||||
Metrics Tracked:
|
||||
- Throughput over time
|
||||
- Latency degradation
|
||||
- Memory usage trend
|
||||
- CPU usage trend
|
||||
- Error rate
|
||||
- Service health
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
import statistics
|
||||
import sys
|
||||
import uuid
|
||||
import json
|
||||
from datetime import datetime
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from typing import List, Dict, Tuple
|
||||
import requests
|
||||
import threading
|
||||
|
||||
# Time-series metrics storage
|
||||
class TimeSeriesMetrics:
|
||||
def __init__(self):
|
||||
self.data_points = [] # List of (timestamp, metrics_dict)
|
||||
self.lock = threading.Lock()
|
||||
|
||||
def record_data_point(self, metrics: Dict):
|
||||
with self.lock:
|
||||
self.data_points.append({
|
||||
'timestamp': time.time(),
|
||||
'datetime': datetime.now().isoformat(),
|
||||
**metrics
|
||||
})
|
||||
|
||||
def get_time_series(self, metric_name: str) -> List[Tuple[float, float]]:
|
||||
"""Get time series for a specific metric"""
|
||||
with self.lock:
|
||||
return [(dp['timestamp'], dp.get(metric_name, 0))
|
||||
for dp in self.data_points if metric_name in dp]
|
||||
|
||||
def analyze_trend(self, metric_name: str) -> Dict:
|
||||
"""Analyze trend for degradation detection"""
|
||||
series = self.get_time_series(metric_name)
|
||||
if len(series) < 2:
|
||||
return {'trend': 'insufficient_data'}
|
||||
|
||||
# Split into first half and second half
|
||||
mid = len(series) // 2
|
||||
first_half = [val for _, val in series[:mid]]
|
||||
second_half = [val for _, val in series[mid:]]
|
||||
|
||||
first_avg = statistics.mean(first_half) if first_half else 0
|
||||
second_avg = statistics.mean(second_half) if second_half else 0
|
||||
|
||||
if first_avg == 0:
|
||||
degradation_pct = 0
|
||||
else:
|
||||
degradation_pct = ((second_avg - first_avg) / first_avg) * 100
|
||||
|
||||
return {
|
||||
'first_half_avg': first_avg,
|
||||
'second_half_avg': second_avg,
|
||||
'degradation_pct': degradation_pct,
|
||||
'trend': 'stable' if abs(degradation_pct) < 10 else 'degraded'
|
||||
}
|
||||
|
||||
class PerformanceMetrics:
|
||||
def __init__(self):
|
||||
self.latencies_ms = []
|
||||
self.successful = 0
|
||||
self.failed = 0
|
||||
self.start_time = None
|
||||
self.end_time = None
|
||||
self.lock = threading.Lock()
|
||||
|
||||
def record_success(self, latency_ms: float):
|
||||
with self.lock:
|
||||
self.latencies_ms.append(latency_ms)
|
||||
self.successful += 1
|
||||
|
||||
def record_failure(self):
|
||||
with self.lock:
|
||||
self.failed += 1
|
||||
|
||||
def get_current_stats(self) -> Dict:
|
||||
"""Get current statistics snapshot"""
|
||||
with self.lock:
|
||||
if not self.latencies_ms:
|
||||
return {
|
||||
'total': self.successful + self.failed,
|
||||
'successful': self.successful,
|
||||
'failed': self.failed,
|
||||
'min_lat': 0, 'p50': 0, 'p95': 0, 'p99': 0, 'max_lat': 0
|
||||
}
|
||||
|
||||
sorted_lat = sorted(self.latencies_ms)
|
||||
n = len(sorted_lat)
|
||||
|
||||
return {
|
||||
'total': self.successful + self.failed,
|
||||
'successful': self.successful,
|
||||
'failed': self.failed,
|
||||
'min_lat': sorted_lat[0],
|
||||
'p50': sorted_lat[n // 2],
|
||||
'p95': sorted_lat[int(n * 0.95)] if n > 20 else sorted_lat[-1],
|
||||
'p99': sorted_lat[int(n * 0.99)] if n > 100 else sorted_lat[-1],
|
||||
'max_lat': sorted_lat[-1]
|
||||
}
|
||||
|
||||
def submit_order_http(order_id: str, symbol: str) -> Tuple[bool, float]:
|
||||
"""Submit order via HTTP to Trading Service"""
|
||||
url = "http://localhost:8081/api/v1/orders"
|
||||
|
||||
payload = {
|
||||
"order_id": order_id,
|
||||
"symbol": symbol,
|
||||
"side": "buy",
|
||||
"order_type": "limit",
|
||||
"quantity": "1.0",
|
||||
"price": "50000.0" if "BTC" in symbol else "3000.0",
|
||||
"time_in_force": "gtc"
|
||||
}
|
||||
|
||||
start = time.time()
|
||||
try:
|
||||
response = requests.post(url, json=payload, timeout=10)
|
||||
latency_ms = (time.time() - start) * 1000
|
||||
return response.status_code == 200, latency_ms
|
||||
except Exception as e:
|
||||
latency_ms = (time.time() - start) * 1000
|
||||
return False, latency_ms
|
||||
|
||||
def get_service_metrics() -> Dict:
|
||||
"""Fetch current service metrics from Prometheus endpoint"""
|
||||
try:
|
||||
response = requests.get("http://localhost:9092/metrics", timeout=2)
|
||||
metrics = {}
|
||||
|
||||
for line in response.text.split('\n'):
|
||||
if line.startswith('#') or not line.strip():
|
||||
continue
|
||||
|
||||
if 'trading_orders_total' in line:
|
||||
metrics['orders_total'] = float(line.split()[-1])
|
||||
elif 'process_resident_memory_bytes' in line:
|
||||
metrics['memory_bytes'] = float(line.split()[-1])
|
||||
elif 'process_cpu_seconds_total' in line:
|
||||
metrics['cpu_seconds'] = float(line.split()[-1])
|
||||
|
||||
return metrics
|
||||
except:
|
||||
return {}
|
||||
|
||||
def get_docker_stats() -> Dict:
|
||||
"""Get Docker container resource usage"""
|
||||
try:
|
||||
import subprocess
|
||||
result = subprocess.run(
|
||||
['docker', 'stats', '--no-stream', '--format',
|
||||
'json', 'foxhunt-trading-service'],
|
||||
capture_output=True, text=True, timeout=3
|
||||
)
|
||||
|
||||
if result.returncode == 0 and result.stdout:
|
||||
stats = json.loads(result.stdout)
|
||||
return {
|
||||
'cpu_pct': stats.get('CPUPerc', '0%').rstrip('%'),
|
||||
'memory_usage': stats.get('MemUsage', '0B'),
|
||||
'memory_pct': stats.get('MemPerc', '0%').rstrip('%')
|
||||
}
|
||||
except:
|
||||
pass
|
||||
|
||||
return {}
|
||||
|
||||
def monitor_resources(time_series: TimeSeriesMetrics, metrics: PerformanceMetrics,
|
||||
shutdown_flag: List[bool]):
|
||||
"""Background thread to monitor system resources"""
|
||||
interval = 5 # Sample every 5 seconds
|
||||
|
||||
while not shutdown_flag[0]:
|
||||
# Get current performance stats
|
||||
stats = metrics.get_current_stats()
|
||||
|
||||
# Get service metrics
|
||||
service_metrics = get_service_metrics()
|
||||
|
||||
# Get Docker stats
|
||||
docker_stats = get_docker_stats()
|
||||
|
||||
# Calculate current throughput
|
||||
elapsed = time.time() - metrics.start_time if metrics.start_time else 1
|
||||
throughput = stats['successful'] / elapsed if elapsed > 0 else 0
|
||||
|
||||
# Record data point
|
||||
time_series.record_data_point({
|
||||
'throughput': throughput,
|
||||
'latency_p50': stats['p50'],
|
||||
'latency_p99': stats['p99'],
|
||||
'success_count': stats['successful'],
|
||||
'failure_count': stats['failed'],
|
||||
'memory_bytes': service_metrics.get('memory_bytes', 0),
|
||||
'cpu_pct': float(docker_stats.get('cpu_pct', 0)),
|
||||
'memory_pct': float(docker_stats.get('memory_pct', 0))
|
||||
})
|
||||
|
||||
time.sleep(interval)
|
||||
|
||||
async def run_sustained_load_test():
|
||||
"""Run 5-minute sustained load test"""
|
||||
print("\n" + "="*70)
|
||||
print(" SUSTAINED LOAD TEST (5 Minutes)")
|
||||
print("="*70)
|
||||
print(f"Target: 1,000+ orders/minute (~16.67 orders/sec)")
|
||||
print(f"Duration: 300 seconds")
|
||||
print(f"Symbols: BTC/USD, ETH/USD")
|
||||
print("="*70 + "\n")
|
||||
|
||||
test_duration_secs = 300
|
||||
num_clients = 8
|
||||
target_orders_per_sec = 17 # Slightly above 16.67 target
|
||||
orders_per_client_per_sec = target_orders_per_sec / num_clients
|
||||
|
||||
metrics = PerformanceMetrics()
|
||||
time_series = TimeSeriesMetrics()
|
||||
shutdown_flag = [False]
|
||||
|
||||
symbols = ["BTC/USD", "ETH/USD"]
|
||||
|
||||
print(f"🚀 Starting {num_clients} clients for {test_duration_secs} seconds...")
|
||||
print(f"🎯 Target: {target_orders_per_sec:.1f} orders/sec total\n")
|
||||
|
||||
# Start resource monitoring thread
|
||||
monitor_thread = threading.Thread(
|
||||
target=monitor_resources,
|
||||
args=(time_series, metrics, shutdown_flag),
|
||||
daemon=True
|
||||
)
|
||||
|
||||
metrics.start_time = time.time()
|
||||
monitor_thread.start()
|
||||
|
||||
# Print progress updates
|
||||
def print_progress():
|
||||
last_update = time.time()
|
||||
while not shutdown_flag[0]:
|
||||
current = time.time()
|
||||
if current - last_update >= 30: # Update every 30 seconds
|
||||
elapsed = current - metrics.start_time
|
||||
stats = metrics.get_current_stats()
|
||||
throughput = stats['successful'] / elapsed if elapsed > 0 else 0
|
||||
|
||||
print(f"⏱️ {elapsed:.0f}s elapsed | "
|
||||
f"Orders: {stats['successful']} | "
|
||||
f"Throughput: {throughput:.1f}/sec | "
|
||||
f"P99: {stats['p99']:.1f}ms | "
|
||||
f"Errors: {stats['failed']}")
|
||||
|
||||
last_update = current
|
||||
time.sleep(1)
|
||||
|
||||
progress_thread = threading.Thread(target=print_progress, daemon=True)
|
||||
progress_thread.start()
|
||||
|
||||
# Client workers
|
||||
def client_worker(client_id: int):
|
||||
order_count = 0
|
||||
delay_secs = 1.0 / orders_per_client_per_sec
|
||||
|
||||
while not shutdown_flag[0]:
|
||||
order_id = str(uuid.uuid4())
|
||||
symbol = symbols[order_count % len(symbols)]
|
||||
|
||||
success, latency_ms = submit_order_http(order_id, symbol)
|
||||
|
||||
if success:
|
||||
metrics.record_success(latency_ms)
|
||||
else:
|
||||
metrics.record_failure()
|
||||
|
||||
order_count += 1
|
||||
time.sleep(delay_secs)
|
||||
|
||||
# Start client threads
|
||||
with ThreadPoolExecutor(max_workers=num_clients) as executor:
|
||||
futures = [executor.submit(client_worker, i) for i in range(num_clients)]
|
||||
|
||||
# Run for specified duration
|
||||
time.sleep(test_duration_secs)
|
||||
shutdown_flag[0] = True
|
||||
|
||||
# Wait for all clients to finish
|
||||
for future in futures:
|
||||
try:
|
||||
future.result(timeout=5)
|
||||
except:
|
||||
pass
|
||||
|
||||
metrics.end_time = time.time()
|
||||
|
||||
# Final statistics
|
||||
print("\n" + "="*70)
|
||||
print(" TEST COMPLETED - ANALYZING RESULTS")
|
||||
print("="*70 + "\n")
|
||||
|
||||
# Performance summary
|
||||
stats = metrics.get_current_stats()
|
||||
duration = metrics.end_time - metrics.start_time
|
||||
throughput = stats['successful'] / duration if duration > 0 else 0
|
||||
success_rate = (stats['successful'] / stats['total'] * 100) if stats['total'] > 0 else 0
|
||||
|
||||
print("📊 PERFORMANCE SUMMARY:")
|
||||
print(f" Duration: {duration:.2f}s")
|
||||
print(f" Total Orders: {stats['total']}")
|
||||
print(f" Successful: {stats['successful']} ({success_rate:.2f}%)")
|
||||
print(f" Failed: {stats['failed']}")
|
||||
print(f" Throughput: {throughput:.1f} orders/sec")
|
||||
print(f" Orders/Minute: {throughput * 60:.0f}")
|
||||
print()
|
||||
print("📈 LATENCY METRICS:")
|
||||
print(f" Min: {stats['min_lat']:.2f}ms")
|
||||
print(f" P50: {stats['p50']:.2f}ms")
|
||||
print(f" P95: {stats['p95']:.2f}ms")
|
||||
print(f" P99: {stats['p99']:.2f}ms")
|
||||
print(f" Max: {stats['max_lat']:.2f}ms")
|
||||
|
||||
# Trend analysis
|
||||
print("\n" + "="*70)
|
||||
print(" DEGRADATION ANALYSIS")
|
||||
print("="*70 + "\n")
|
||||
|
||||
throughput_trend = time_series.analyze_trend('throughput')
|
||||
latency_trend = time_series.analyze_trend('latency_p99')
|
||||
memory_trend = time_series.analyze_trend('memory_bytes')
|
||||
|
||||
print(f"📉 Throughput Trend:")
|
||||
print(f" First Half Avg: {throughput_trend['first_half_avg']:.1f} orders/sec")
|
||||
print(f" Second Half Avg: {throughput_trend['second_half_avg']:.1f} orders/sec")
|
||||
print(f" Change: {throughput_trend['degradation_pct']:+.1f}%")
|
||||
print(f" Status: {throughput_trend['trend'].upper()}")
|
||||
|
||||
print(f"\n⏱️ Latency Trend (P99):")
|
||||
print(f" First Half Avg: {latency_trend['first_half_avg']:.1f}ms")
|
||||
print(f" Second Half Avg: {latency_trend['second_half_avg']:.1f}ms")
|
||||
print(f" Change: {latency_trend['degradation_pct']:+.1f}%")
|
||||
print(f" Status: {latency_trend['trend'].upper()}")
|
||||
|
||||
print(f"\n💾 Memory Trend:")
|
||||
if memory_trend['first_half_avg'] > 0:
|
||||
print(f" First Half Avg: {memory_trend['first_half_avg']/1024/1024:.1f}MB")
|
||||
print(f" Second Half Avg: {memory_trend['second_half_avg']/1024/1024:.1f}MB")
|
||||
print(f" Change: {memory_trend['degradation_pct']:+.1f}%")
|
||||
print(f" Status: {memory_trend['trend'].upper()}")
|
||||
else:
|
||||
print(f" Status: No memory data available")
|
||||
|
||||
# Pass/Fail Assessment
|
||||
print("\n" + "="*70)
|
||||
print(" SUCCESS CRITERIA EVALUATION")
|
||||
print("="*70 + "\n")
|
||||
|
||||
passed_checks = 0
|
||||
total_checks = 5
|
||||
|
||||
# Check 1: Sustained throughput
|
||||
orders_per_minute = throughput * 60
|
||||
if orders_per_minute >= 1000:
|
||||
print(f"✅ Throughput: {orders_per_minute:.0f} orders/min (>= 1,000)")
|
||||
passed_checks += 1
|
||||
else:
|
||||
print(f"❌ Throughput: {orders_per_minute:.0f} orders/min (< 1,000)")
|
||||
|
||||
# Check 2: Success rate
|
||||
if success_rate >= 99.0:
|
||||
print(f"✅ Success Rate: {success_rate:.2f}% (>= 99%)")
|
||||
passed_checks += 1
|
||||
else:
|
||||
print(f"❌ Success Rate: {success_rate:.2f}% (< 99%)")
|
||||
|
||||
# Check 3: No performance degradation
|
||||
if abs(throughput_trend['degradation_pct']) < 10:
|
||||
print(f"✅ Performance Stable: {throughput_trend['degradation_pct']:+.1f}% change (< 10%)")
|
||||
passed_checks += 1
|
||||
else:
|
||||
print(f"❌ Performance Degraded: {throughput_trend['degradation_pct']:+.1f}% change (>= 10%)")
|
||||
|
||||
# Check 4: No memory leak
|
||||
if memory_trend['first_half_avg'] == 0 or abs(memory_trend['degradation_pct']) < 20:
|
||||
print(f"✅ No Memory Leak: {memory_trend['degradation_pct']:+.1f}% change (< 20%)")
|
||||
passed_checks += 1
|
||||
else:
|
||||
print(f"❌ Memory Leak Detected: {memory_trend['degradation_pct']:+.1f}% change (>= 20%)")
|
||||
|
||||
# Check 5: Service health
|
||||
try:
|
||||
health_response = requests.get("http://localhost:8081/health", timeout=5)
|
||||
if health_response.status_code == 200:
|
||||
print(f"✅ Service Health: Healthy post-test")
|
||||
passed_checks += 1
|
||||
else:
|
||||
print(f"❌ Service Health: Unhealthy post-test")
|
||||
except:
|
||||
print(f"❌ Service Health: Unreachable post-test")
|
||||
|
||||
print(f"\n📊 OVERALL: {passed_checks}/{total_checks} checks passed\n")
|
||||
|
||||
if passed_checks >= 4:
|
||||
print("🎉 TEST PASSED - PRODUCTION READY")
|
||||
verdict = "PASS"
|
||||
elif passed_checks >= 3:
|
||||
print("⚠️ TEST PASSED WITH WARNINGS - Monitor in production")
|
||||
verdict = "PASS_WITH_WARNINGS"
|
||||
else:
|
||||
print("❌ TEST FAILED - Not ready for sustained load")
|
||||
verdict = "FAIL"
|
||||
|
||||
print("="*70 + "\n")
|
||||
|
||||
# Save detailed report
|
||||
report = {
|
||||
'test_config': {
|
||||
'duration_secs': test_duration_secs,
|
||||
'num_clients': num_clients,
|
||||
'target_orders_per_sec': target_orders_per_sec,
|
||||
'symbols': symbols
|
||||
},
|
||||
'performance': {
|
||||
'duration': duration,
|
||||
'total_orders': stats['total'],
|
||||
'successful': stats['successful'],
|
||||
'failed': stats['failed'],
|
||||
'success_rate_pct': success_rate,
|
||||
'throughput_per_sec': throughput,
|
||||
'orders_per_minute': orders_per_minute
|
||||
},
|
||||
'latency': {
|
||||
'min_ms': stats['min_lat'],
|
||||
'p50_ms': stats['p50'],
|
||||
'p95_ms': stats['p95'],
|
||||
'p99_ms': stats['p99'],
|
||||
'max_ms': stats['max_lat']
|
||||
},
|
||||
'trends': {
|
||||
'throughput': throughput_trend,
|
||||
'latency_p99': latency_trend,
|
||||
'memory': memory_trend
|
||||
},
|
||||
'time_series': time_series.data_points,
|
||||
'verdict': verdict,
|
||||
'checks_passed': f"{passed_checks}/{total_checks}"
|
||||
}
|
||||
|
||||
with open('/home/jgrusewski/Work/foxhunt/sustained_load_test_results.json', 'w') as f:
|
||||
json.dump(report, f, indent=2)
|
||||
|
||||
print("💾 Detailed results saved to: sustained_load_test_results.json\n")
|
||||
|
||||
return verdict, report
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
verdict, report = asyncio.run(run_sustained_load_test())
|
||||
sys.exit(0 if verdict == "PASS" else 1)
|
||||
except KeyboardInterrupt:
|
||||
print("\n⚠️ Test interrupted by user")
|
||||
sys.exit(2)
|
||||
except Exception as e:
|
||||
print(f"\n❌ Test failed with error: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
Reference in New Issue
Block a user