mirror of
https://github.com/Av0idD4rk/packmate.git
synced 2026-10-23 15:31:00 +00:00
162 lines
9.8 KiB
Python
162 lines
9.8 KiB
Python
"""35-minute real-network test with natural 15-minute retention, no timestamp rewrites."""
|
|
import asyncio
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
import struct
|
|
import time
|
|
from types import SimpleNamespace
|
|
|
|
import httpx
|
|
import psycopg
|
|
|
|
from capacity import API, REPORTS, call, session
|
|
from network_load import run as generate
|
|
from verify_ledger import expected_digest
|
|
|
|
|
|
async def main():
|
|
duration = int(os.getenv("MCP_SOAK_SECONDS", "2100"))
|
|
assert duration >= 2100, "soak must cross the historical 25-30 minute degradation window"
|
|
db = await psycopg.AsyncConnection.connect("host=db dbname=packmate user=packmate password=network-stress", autocommit=True)
|
|
|
|
async def rows(sql, args=()):
|
|
return await (await db.execute(sql, args)).fetchall()
|
|
|
|
async def scalar(sql, args=()):
|
|
return (await rows(sql, args))[0][0]
|
|
|
|
done = asyncio.Event()
|
|
latencies, latency_samples, samples, reasons, load_reports = [], [], [], [], []
|
|
favorites = []
|
|
report = {"configured_seconds": duration, "retention_minutes": 15, "cleanup_interval_seconds": 10,
|
|
"frames_per_stream": 8, "payload_bytes": 1024, "verified_streams": 0}
|
|
async with httpx.AsyncClient(auth=("stress", "stress"), timeout=10) as rest:
|
|
for _ in range(90):
|
|
try:
|
|
if (await rest.get(API + "/api/analysis/overview")).status_code == 200:
|
|
break
|
|
except httpx.HTTPError:
|
|
pass
|
|
await asyncio.sleep(1)
|
|
else:
|
|
raise AssertionError("backend not ready")
|
|
assert await scalar("SELECT count(*) FROM stream") == 0, "requires fresh disposable DB"
|
|
async with session() as client:
|
|
await call(client, "create_service", {"port": 18080, "settings": {"name": "soak echo", "http": False,
|
|
"urldecode_http_requests": False, "merge_adjacent_packets": False}})
|
|
pattern = await call(client, "create_pattern", {"settings": {"name": "soak marker",
|
|
"match": {"type": "substring", "value": "PMSTRESS"}, "service_port": 18080}})
|
|
pattern_id = pattern["pattern"]["id"]
|
|
started = time.monotonic()
|
|
|
|
async def traffic():
|
|
batch = 0
|
|
try:
|
|
async with session() as client:
|
|
while time.monotonic() - started < duration:
|
|
lower = await scalar("SELECT coalesce(max(id),0) FROM stream")
|
|
run_id = 9123000 + batch
|
|
result = await generate(SimpleNamespace(host="172.31.250.10", port=18080, streams=500,
|
|
frames=8, payload_bytes=1024, concurrency=32, run_id=run_id, shard=0, stream_offset=0,
|
|
report=str(REPORTS / "soak-last-load.json"), abort_file=None, minimum_duration_seconds=10))
|
|
assert result["successfulStreams"] == 500 and not result["sampleErrors"], result
|
|
load_reports.append(result)
|
|
async with asyncio.timeout(30):
|
|
while await scalar("SELECT count(*) FROM stream WHERE id>%s", (lower,)) != 500:
|
|
await asyncio.sleep(.2)
|
|
entries = await rows("""
|
|
WITH data AS (
|
|
SELECT s.id, string_agg(p.content, ''::bytea ORDER BY p.id) FILTER (WHERE p.incoming) AS inc,
|
|
string_agg(p.content, ''::bytea ORDER BY p.id) FILTER (WHERE NOT p.incoming) AS out,
|
|
count(*)=s.packets_count AND sum(octet_length(p.content))=s.size_bytes
|
|
AND s.service_id=18080 AND s.protocol=0 AS metadata,
|
|
bool_and(CASE WHEN p.incoming THEN p.destination_ip='172.31.250.10' AND p.source_ip<>'172.31.250.10'
|
|
ELSE p.source_ip='172.31.250.10' AND p.destination_ip<>'172.31.250.10' END) AS ips
|
|
FROM stream s JOIN packet p ON p.stream_id=s.id WHERE s.id>%s GROUP BY s.id)
|
|
SELECT id, substring(inc FROM 5 FOR 24), octet_length(inc), md5(inc),
|
|
octet_length(out), md5(out), metadata, ips FROM data ORDER BY id
|
|
""", (lower,))
|
|
assert len(entries) == 500
|
|
seen = set()
|
|
for sid, marker, insize, indigest, outsize, outdigest, metadata, ips in entries:
|
|
run, connection, shard, sequence = struct.unpack("!QQII", marker)
|
|
assert run == run_id and shard == sequence == 0 and 0 <= connection < 500 and connection not in seen, sid
|
|
seen.add(connection)
|
|
assert insize == outsize == 8 * 1028 and metadata and ips, sid
|
|
assert indigest == outdigest == expected_digest(run_id, connection, 0, 8, 1024), sid
|
|
assert await scalar("SELECT count(*) FROM stream_found_patterns WHERE stream_id>%s AND pattern_id=%s", (lower, pattern_id)) == 500
|
|
saved = entries[0][0]
|
|
assert (await call(client, "set_favorites", {"stream_ids": [saved]}))["updated"] == [saved]
|
|
favorites.append(saved)
|
|
report["verified_streams"] += 500
|
|
report["elapsed_seconds"] = round(time.monotonic() - started, 1)
|
|
(REPORTS / "soak-progress.json").write_text(json.dumps(report, indent=2))
|
|
batch += 1
|
|
finally:
|
|
done.set()
|
|
|
|
async def monitor():
|
|
while not done.is_set():
|
|
before = time.monotonic()
|
|
response = await rest.get(API + "/api/pcap/metrics")
|
|
response.raise_for_status()
|
|
latencies.append((time.monotonic() - before) * 1000)
|
|
latency_samples.append((time.monotonic() - started, latencies[-1]))
|
|
metrics = response.json()
|
|
assert all(metrics[k] == 0 for k in ("droppedPackets", "interfaceDroppedPackets", "applicationDroppedPackets")), metrics
|
|
sample = {"seconds": round(time.monotonic() - started, 1), "streams": await scalar("SELECT count(*) FROM stream"),
|
|
"database_bytes": await scalar("SELECT pg_database_size(current_database())"), "metrics": metrics}
|
|
samples.append(sample)
|
|
if sample["seconds"] > 1020:
|
|
assert await scalar("SELECT count(*) FROM stream WHERE NOT favorite AND start_timestamp<%s", (int(time.time()*1000)-1020000,)) == 0, sample
|
|
await asyncio.sleep(2)
|
|
|
|
async def investigations():
|
|
async with session() as client:
|
|
while not done.is_set():
|
|
job = await call(client, "search_traffic", {"query": {"last_minutes": 15, "match": {"value": "(a+)+$"}}, "limit": 1000})
|
|
async with asyncio.timeout(45):
|
|
while True:
|
|
status = await call(client, "get_job", {"job_id": job["job_id"]})
|
|
if status["status"] != "running":
|
|
assert status["status"] in ("completed", "partial"), status
|
|
reasons.append(status["reason"])
|
|
break
|
|
await asyncio.sleep(.5)
|
|
await call(client, "delete_job", {"job_id": job["job_id"]})
|
|
await asyncio.gather(traffic(), monitor(), investigations())
|
|
remaining = await scalar("SELECT count(*) FROM stream")
|
|
preserved = await scalar("SELECT count(*) FROM stream WHERE id=ANY(%s) AND favorite", (favorites,))
|
|
assert preserved == len(favorites)
|
|
assert report["verified_streams"] > remaining > 0
|
|
assert await scalar("SELECT count(*) FROM packet p LEFT JOIN stream s ON s.id=p.stream_id WHERE s.id IS NULL") == 0
|
|
assert await scalar("SELECT count(*) FROM found_pattern f LEFT JOIN packet p ON p.id=f.packet_id WHERE p.id IS NULL") == 0
|
|
p99 = sorted(latencies)[math.ceil(len(latencies)*.99)-1]
|
|
assert p99 < 500, p99
|
|
final_metrics = (await rest.get(API + "/api/pcap/metrics")).json()
|
|
assert all(final_metrics[k] == 0 for k in ("droppedPackets", "interfaceDroppedPackets", "applicationDroppedPackets")), final_metrics
|
|
first_latencies = sorted(ms for elapsed, ms in latency_samples if elapsed < 300)
|
|
last_latencies = sorted(ms for elapsed, ms in latency_samples if elapsed > duration - 300)
|
|
last_p99 = last_latencies[math.ceil(len(last_latencies)*.99)-1]
|
|
assert last_p99 < 500, last_p99
|
|
report.update(elapsed_seconds=round(time.monotonic()-started, 1), retained_streams=remaining,
|
|
naturally_deleted_streams=report["verified_streams"]-remaining, favorites_preserved=preserved,
|
|
api_p99_ms=round(p99, 3), api_samples=len(latencies), mcp_jobs=len(reasons),
|
|
job_reasons=sorted({str(x) for x in reasons}), final_metrics=final_metrics,
|
|
first_5min_api_p99_ms=round(first_latencies[math.ceil(len(first_latencies)*.99)-1], 3),
|
|
last_5min_api_p99_ms=round(last_p99, 3),
|
|
checks=["every stream payload hash/IP/metadata and live pattern verified before retention",
|
|
"natural retention removes old streams while capture and MCP continue",
|
|
"all favorites preserved, no orphan packets/matches, no capture drops"])
|
|
(REPORTS / "soak-samples.json").write_text(json.dumps(samples, indent=2))
|
|
(REPORTS / "soak-loads.json").write_text(json.dumps(load_reports, indent=2))
|
|
(REPORTS / "soak.json").write_text(json.dumps(report, indent=2))
|
|
print(json.dumps(report, indent=2), flush=True)
|
|
await db.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|