93 lines
3.1 KiB
Python
93 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Test di concorrenza MCP: N chiamate parallele + sonde /health e ping.
|
|
|
|
Uso (dentro il container):
|
|
docker exec loogle-mcp python /srv/scripts/test_mcp_concurrency.py [N] [tool]
|
|
|
|
Misura se una chiamata lenta blocca le altre (event loop bloccato) e la
|
|
latenza delle ricerche semantiche.
|
|
"""
|
|
|
|
import concurrent.futures as cf
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
import httpx
|
|
|
|
sys.path.insert(0, "/srv")
|
|
from app.db import get_conn # noqa: E402
|
|
from app.jwt_utils import create_access_token # noqa: E402
|
|
|
|
BASE = os.environ.get("TEST_BASE", "http://127.0.0.1:8700")
|
|
N = int(sys.argv[1]) if len(sys.argv) > 1 else 4
|
|
TOOL = sys.argv[2] if len(sys.argv) > 2 else "search_knowledge"
|
|
CLIENT_TIMEOUT = 150.0
|
|
|
|
row = get_conn().execute("SELECT id, username FROM users WHERE username='daniele'").fetchone()
|
|
token, _ = create_access_token(
|
|
{"id": row["id"], "username": row["username"]},
|
|
"context:read knowledge:read gitea:read admin",
|
|
"loadtest",
|
|
)
|
|
HEADERS = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
|
|
|
|
|
|
def rpc(method: str, params: dict, req_id: int) -> tuple[float, str]:
|
|
t0 = time.monotonic()
|
|
try:
|
|
r = httpx.post(
|
|
f"{BASE}/mcp",
|
|
headers=HEADERS,
|
|
json={"jsonrpc": "2.0", "id": req_id, "method": method, "params": params},
|
|
timeout=CLIENT_TIMEOUT,
|
|
)
|
|
body = r.json()
|
|
if "error" in body:
|
|
status = f"ERR {body['error'].get('code')}: {body['error'].get('message', '')[:90]}"
|
|
else:
|
|
text = body["result"]["content"][0]["text"] if method == "tools/call" else "{}"
|
|
payload = json.loads(text)
|
|
n = len(payload.get("results", [])) if isinstance(payload, dict) else "?"
|
|
status = f"OK results={n}"
|
|
except Exception as exc: # noqa: BLE001
|
|
status = f"EXC {type(exc).__name__}: {exc}"[:100]
|
|
return time.monotonic() - t0, status
|
|
|
|
|
|
def health() -> float:
|
|
t0 = time.monotonic()
|
|
try:
|
|
httpx.get(f"{BASE}/health", timeout=CLIENT_TIMEOUT)
|
|
except Exception: # noqa: BLE001
|
|
return -1.0
|
|
return time.monotonic() - t0
|
|
|
|
|
|
def main() -> None:
|
|
queries = ["failover keepalived", "irrigazione zone", "turni medici", "backup nas",
|
|
"runbook", "rete vlan", "home assistant", "paperless"]
|
|
t_start = time.monotonic()
|
|
with cf.ThreadPoolExecutor(max_workers=N + 2) as pool:
|
|
futs = {
|
|
pool.submit(rpc, "tools/call",
|
|
{"name": TOOL, "arguments": {"query": queries[i % len(queries)], "limit": 3}},
|
|
i + 1): f"{TOOL}#{i + 1}"
|
|
for i in range(N)
|
|
}
|
|
time.sleep(0.5)
|
|
h = pool.submit(health)
|
|
p = pool.submit(rpc, "tools/call", {"name": "ping", "arguments": {}}, 999)
|
|
print(f"/health durante il carico: {h.result():.2f}s")
|
|
pt, ps = p.result()
|
|
print(f"ping durante il carico: {pt:.2f}s {ps}")
|
|
for fut in cf.as_completed(futs):
|
|
dt, status = fut.result()
|
|
print(f"{futs[fut]:<24} {dt:6.2f}s {status}")
|
|
print(f"Totale: {time.monotonic() - t_start:.2f}s (N={N})")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|