yuqei
57fc8d350b
Security Quality Gate / npm audit, Bandit, pip-audit, detect-secrets (pull_request) Successful in 11m32s
457 lines
17 KiB
Python
457 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""Run read-only HTTP interface benchmarks for the V2 smoke environment."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import math
|
|
import os
|
|
import statistics
|
|
import sys
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import quote
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
DEFAULT_ARTIFACT_ROOT = REPO_ROOT / "project-docs" / "qa-artifacts" / "v2-interface-benchmark"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Scenario:
|
|
key: str
|
|
label: str
|
|
service: str
|
|
path: str
|
|
target_p95_ms: float
|
|
|
|
|
|
def now_stamp() -> str:
|
|
return datetime.now().strftime("%Y-%m-%dT%H-%M-%S")
|
|
|
|
|
|
def checked_at() -> str:
|
|
return datetime.now(timezone.utc).astimezone().isoformat()
|
|
|
|
|
|
def normalize_base_url(value: str) -> str:
|
|
return value.rstrip("/")
|
|
|
|
|
|
def build_url(base_url: str, path: str) -> str:
|
|
return f"{normalize_base_url(base_url)}{path}"
|
|
|
|
|
|
def fetch_once(
|
|
url: str,
|
|
timeout_seconds: float,
|
|
scenario_key: str,
|
|
iteration: int,
|
|
headers: dict[str, str],
|
|
) -> dict[str, Any]:
|
|
started = time.perf_counter()
|
|
request = Request(
|
|
url,
|
|
headers={
|
|
"Accept": "application/json",
|
|
"User-Agent": "yuqei-v2-interface-benchmark/1.0",
|
|
**headers,
|
|
},
|
|
method="GET",
|
|
)
|
|
status_code: int | None = None
|
|
body = b""
|
|
error = ""
|
|
try:
|
|
with urlopen(request, timeout=timeout_seconds) as response:
|
|
status_code = response.getcode()
|
|
body = response.read()
|
|
except HTTPError as exc:
|
|
status_code = exc.code
|
|
try:
|
|
body = exc.read()
|
|
except Exception:
|
|
body = b""
|
|
error = f"HTTPError: {exc.reason}"
|
|
except URLError as exc:
|
|
error = f"URLError: {exc.reason}"
|
|
except TimeoutError as exc:
|
|
error = f"TimeoutError: {exc}"
|
|
except Exception as exc: # pragma: no cover - defensive benchmark logging.
|
|
error = f"{type(exc).__name__}: {exc}"
|
|
elapsed_ms = (time.perf_counter() - started) * 1000
|
|
ok = status_code is not None and 200 <= status_code < 300
|
|
if not ok and not error:
|
|
error = f"HTTP {status_code}"
|
|
return {
|
|
"scenario": scenario_key,
|
|
"iteration": iteration,
|
|
"ok": ok,
|
|
"status_code": status_code,
|
|
"elapsed_ms": elapsed_ms,
|
|
"response_bytes": len(body),
|
|
"error": error,
|
|
}
|
|
|
|
|
|
def fetch_json(url: str, timeout_seconds: float, headers: dict[str, str]) -> Any:
|
|
request = Request(url, headers={"Accept": "application/json", **headers}, method="GET")
|
|
with urlopen(request, timeout=timeout_seconds) as response:
|
|
payload = response.read().decode("utf-8")
|
|
return json.loads(payload)
|
|
|
|
|
|
def percentile(values: list[float], percent: float) -> float | None:
|
|
if not values:
|
|
return None
|
|
ordered = sorted(values)
|
|
if len(ordered) == 1:
|
|
return ordered[0]
|
|
rank = (percent / 100) * (len(ordered) - 1)
|
|
lower = math.floor(rank)
|
|
upper = math.ceil(rank)
|
|
if lower == upper:
|
|
return ordered[int(rank)]
|
|
weight = rank - lower
|
|
return ordered[lower] * (1 - weight) + ordered[upper] * weight
|
|
|
|
|
|
def round_or_none(value: float | None, digits: int = 2) -> float | None:
|
|
if value is None:
|
|
return None
|
|
return round(value, digits)
|
|
|
|
|
|
def summarize_scenario(
|
|
scenario: Scenario,
|
|
url: str,
|
|
results: list[dict[str, Any]],
|
|
duration_seconds: float,
|
|
) -> dict[str, Any]:
|
|
ok_results = [item for item in results if item["ok"]]
|
|
ok_latencies = [float(item["elapsed_ms"]) for item in ok_results]
|
|
total = len(results)
|
|
errors = total - len(ok_results)
|
|
status_codes: dict[str, int] = {}
|
|
for item in results:
|
|
code = str(item["status_code"] if item["status_code"] is not None else "ERR")
|
|
status_codes[code] = status_codes.get(code, 0) + 1
|
|
|
|
p95_ms = percentile(ok_latencies, 95)
|
|
avg_ms = statistics.fmean(ok_latencies) if ok_latencies else None
|
|
total_bytes = sum(int(item["response_bytes"]) for item in results)
|
|
avg_bytes = total_bytes / total if total else 0
|
|
error_rate = (errors / total * 100) if total else 0
|
|
passed = errors == 0 and (p95_ms is not None and p95_ms <= scenario.target_p95_ms)
|
|
|
|
return {
|
|
"key": scenario.key,
|
|
"label": scenario.label,
|
|
"service": scenario.service,
|
|
"method": "GET",
|
|
"path": scenario.path,
|
|
"url": url,
|
|
"requests": total,
|
|
"ok": len(ok_results),
|
|
"errors": errors,
|
|
"error_rate_percent": round(error_rate, 4),
|
|
"status_codes": dict(sorted(status_codes.items())),
|
|
"duration_seconds": round(duration_seconds, 3),
|
|
"rps": round(total / duration_seconds, 2) if duration_seconds > 0 else None,
|
|
"min_ms": round_or_none(min(ok_latencies) if ok_latencies else None),
|
|
"avg_ms": round_or_none(avg_ms),
|
|
"p50_ms": round_or_none(percentile(ok_latencies, 50)),
|
|
"p95_ms": round_or_none(p95_ms),
|
|
"p99_ms": round_or_none(percentile(ok_latencies, 99)),
|
|
"max_ms": round_or_none(max(ok_latencies) if ok_latencies else None),
|
|
"avg_response_bytes": round(avg_bytes, 2),
|
|
"total_response_bytes": total_bytes,
|
|
"target_p95_ms": scenario.target_p95_ms,
|
|
"passed": passed,
|
|
}
|
|
|
|
|
|
def write_csv(path: Path, results: list[dict[str, Any]]) -> None:
|
|
fields = [
|
|
"scenario",
|
|
"iteration",
|
|
"ok",
|
|
"status_code",
|
|
"elapsed_ms",
|
|
"response_bytes",
|
|
"error",
|
|
]
|
|
with path.open("w", encoding="utf-8", newline="") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=fields)
|
|
writer.writeheader()
|
|
for item in results:
|
|
row = dict(item)
|
|
row["elapsed_ms"] = f"{float(row['elapsed_ms']):.4f}"
|
|
writer.writerow(row)
|
|
|
|
|
|
def format_number(value: Any) -> str:
|
|
if value is None:
|
|
return "-"
|
|
if isinstance(value, float):
|
|
return f"{value:.2f}"
|
|
return str(value)
|
|
|
|
|
|
def write_markdown(path: Path, summary: dict[str, Any]) -> None:
|
|
lines: list[str] = []
|
|
run = summary["run"]
|
|
lines.append("# V2 Interface Benchmark Summary")
|
|
lines.append("")
|
|
lines.append(f"- Status: {summary['status']}")
|
|
lines.append(f"- Checked at: {run['checked_at']}")
|
|
lines.append(f"- Contract API: {run['contract_api_base']}")
|
|
lines.append(f"- AI Platform API: {run['ai_api_base']}")
|
|
lines.append(f"- Requests per scenario: {run['requests_per_scenario']}")
|
|
lines.append(f"- Concurrency: {run['concurrency']}")
|
|
lines.append(f"- Warmup per scenario: {run['warmup_requests']} (excluded from stats)")
|
|
lines.append(f"- Timeout seconds: {run['timeout_seconds']}")
|
|
lines.append("")
|
|
lines.append("## Dataset Hints")
|
|
for key, value in summary["dataset"].items():
|
|
lines.append(f"- {key}: {value}")
|
|
lines.append("")
|
|
lines.append("## Results")
|
|
lines.append(
|
|
"| Scenario | Requests | Error % | RPS | p50 ms | p95 ms | p99 ms | Avg bytes | Target p95 ms | Status |"
|
|
)
|
|
lines.append("| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | --- |")
|
|
for item in summary["scenarios"]:
|
|
status = "pass" if item["passed"] else "fail"
|
|
lines.append(
|
|
"| {label} | {requests} | {error_rate_percent} | {rps} | {p50_ms} | {p95_ms} | {p99_ms} | "
|
|
"{avg_response_bytes} | {target_p95_ms} | {status} |".format(
|
|
label=item["label"],
|
|
requests=item["requests"],
|
|
error_rate_percent=format_number(item["error_rate_percent"]),
|
|
rps=format_number(item["rps"]),
|
|
p50_ms=format_number(item["p50_ms"]),
|
|
p95_ms=format_number(item["p95_ms"]),
|
|
p99_ms=format_number(item["p99_ms"]),
|
|
avg_response_bytes=format_number(item["avg_response_bytes"]),
|
|
target_p95_ms=format_number(item["target_p95_ms"]),
|
|
status=status,
|
|
)
|
|
)
|
|
lines.append("")
|
|
lines.append("## Notes")
|
|
lines.append("- All measured requests use GET and do not mutate remote data.")
|
|
lines.append("- Percentiles are calculated from successful responses only; error rate is reported separately.")
|
|
lines.append("- The benchmark uses Python stdlib HTTP clients, so numbers include client-side connection overhead.")
|
|
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Run V2 read-only interface benchmark.")
|
|
parser.add_argument("--contract-api-base", default="http://localhost:8100")
|
|
parser.add_argument("--ai-api-base", default="http://localhost:8101")
|
|
parser.add_argument("--requests-per-scenario", type=int, default=60)
|
|
parser.add_argument("--concurrency", type=int, default=6)
|
|
parser.add_argument("--warmup", type=int, default=3)
|
|
parser.add_argument("--timeout-seconds", type=float, default=15.0)
|
|
parser.add_argument("--keyword", default="\u5408\u540c")
|
|
parser.add_argument("--artifact-dir", default="")
|
|
parser.add_argument("--include-heavy-explain", action="store_true")
|
|
parser.add_argument("--tenant-id", default="default")
|
|
parser.add_argument("--user-id", default="user-current-admin")
|
|
parser.add_argument("--contract-proxy-key-env", default="CONTRACT_API_WEB_PROXY_SHARED_SECRET")
|
|
parser.add_argument("--ai-service-key-env", default="AI_PLATFORM_SERVICE_SHARED_SECRET")
|
|
return parser.parse_args()
|
|
|
|
|
|
def request_headers(tenant_id: str, user_id: str, auth_header: str, secret: str) -> dict[str, str]:
|
|
headers: dict[str, str] = {}
|
|
if tenant_id.strip():
|
|
headers["X-Tenant-Id"] = tenant_id.strip()
|
|
if user_id.strip():
|
|
headers["X-User-Id"] = user_id.strip()
|
|
if secret:
|
|
headers[auth_header] = secret
|
|
return headers
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
if args.requests_per_scenario <= 0:
|
|
raise SystemExit("--requests-per-scenario must be positive")
|
|
if args.concurrency <= 0:
|
|
raise SystemExit("--concurrency must be positive")
|
|
if args.warmup < 0:
|
|
raise SystemExit("--warmup must be zero or positive")
|
|
|
|
contract_api_base = normalize_base_url(args.contract_api_base)
|
|
ai_api_base = normalize_base_url(args.ai_api_base)
|
|
contract_proxy_key = os.getenv(args.contract_proxy_key_env, "").strip()
|
|
ai_service_key = os.getenv(args.ai_service_key_env, "").strip()
|
|
contract_headers = request_headers(
|
|
args.tenant_id,
|
|
args.user_id,
|
|
"X-Contract-Proxy-Key",
|
|
contract_proxy_key,
|
|
)
|
|
ai_headers = request_headers(
|
|
args.tenant_id,
|
|
args.user_id,
|
|
"X-AI-Platform-Service-Key",
|
|
ai_service_key,
|
|
)
|
|
artifact_dir = Path(args.artifact_dir) if args.artifact_dir else DEFAULT_ARTIFACT_ROOT / now_stamp()
|
|
if not artifact_dir.is_absolute():
|
|
artifact_dir = REPO_ROOT / artifact_dir
|
|
artifact_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
keyword = quote(args.keyword)
|
|
dataset: dict[str, Any] = {}
|
|
contract_list_probe_url = build_url(contract_api_base, "/api/v1/contracts?page=1&page_size=1")
|
|
contract_probe = fetch_json(contract_list_probe_url, args.timeout_seconds, contract_headers)
|
|
contract_items = contract_probe.get("items", []) if isinstance(contract_probe, dict) else []
|
|
if not contract_items:
|
|
raise SystemExit("No contracts found; cannot discover contract detail scenario.")
|
|
contract_id = contract_items[0]["id"]
|
|
dataset["contract_total"] = contract_probe.get("total", "unknown")
|
|
dataset["detail_contract_id"] = contract_id
|
|
|
|
search_probe_url = build_url(
|
|
contract_api_base,
|
|
f"/api/v1/search?keyword={keyword}&page=1&page_size=10",
|
|
)
|
|
try:
|
|
search_probe = fetch_json(search_probe_url, args.timeout_seconds, contract_headers)
|
|
if isinstance(search_probe, dict):
|
|
dataset["unified_search_total"] = search_probe.get("total", "unknown")
|
|
except Exception as exc:
|
|
dataset["unified_search_total"] = f"probe_failed: {exc}"
|
|
|
|
scenarios = [
|
|
Scenario(
|
|
key="contract_list",
|
|
label="Contract list",
|
|
service="contract-api",
|
|
path="/api/v1/contracts?page=1&page_size=20",
|
|
target_p95_ms=500,
|
|
),
|
|
Scenario(
|
|
key="contract_detail",
|
|
label="Contract detail",
|
|
service="contract-api",
|
|
path=f"/api/v1/contracts/{contract_id}",
|
|
target_p95_ms=500,
|
|
),
|
|
Scenario(
|
|
key="approval_todo",
|
|
label="Approval todo",
|
|
service="contract-api",
|
|
path="/api/v1/contracts?status=approval_pending&page=1&page_size=20",
|
|
target_p95_ms=500,
|
|
),
|
|
Scenario(
|
|
key="unified_search",
|
|
label="Unified search",
|
|
service="contract-api",
|
|
path=f"/api/v1/search?keyword={keyword}&page=1&page_size=10",
|
|
target_p95_ms=500,
|
|
),
|
|
Scenario(
|
|
key="ai_trace",
|
|
label="AI Trace list",
|
|
service="ai-platform-api",
|
|
path="/api/v1/audit/ai-runs?limit=5",
|
|
target_p95_ms=1500,
|
|
),
|
|
Scenario(
|
|
key="knowledge_documents_search",
|
|
label="Knowledge document search",
|
|
service="ai-platform-api",
|
|
path=f"/api/v1/knowledge-documents/search?keyword={keyword}&limit=5",
|
|
target_p95_ms=800,
|
|
),
|
|
Scenario(
|
|
key="knowledge_chunks_hybrid_search",
|
|
label="Knowledge chunk hybrid search",
|
|
service="ai-platform-api",
|
|
path=f"/api/v1/knowledge-chunks/search?keyword={keyword}&strategy=hybrid&limit=5&candidate_limit=20",
|
|
target_p95_ms=1500,
|
|
),
|
|
]
|
|
if args.include_heavy_explain:
|
|
scenarios.append(
|
|
Scenario(
|
|
key="knowledge_chunks_explain",
|
|
label="Knowledge chunk explain",
|
|
service="ai-platform-api",
|
|
path=f"/api/v1/knowledge-chunks/search/explain?keyword={keyword}&strategy=hybrid&limit=5&candidate_limit=20",
|
|
target_p95_ms=2500,
|
|
)
|
|
)
|
|
|
|
all_results: list[dict[str, Any]] = []
|
|
scenario_summaries: list[dict[str, Any]] = []
|
|
for scenario in scenarios:
|
|
base_url = contract_api_base if scenario.service == "contract-api" else ai_api_base
|
|
headers = contract_headers if scenario.service == "contract-api" else ai_headers
|
|
url = build_url(base_url, scenario.path)
|
|
for index in range(args.warmup):
|
|
fetch_once(url, args.timeout_seconds, scenario.key, -index - 1, headers)
|
|
|
|
started = time.perf_counter()
|
|
scenario_results: list[dict[str, Any]] = []
|
|
with ThreadPoolExecutor(max_workers=args.concurrency) as executor:
|
|
futures = [
|
|
executor.submit(fetch_once, url, args.timeout_seconds, scenario.key, index + 1, headers)
|
|
for index in range(args.requests_per_scenario)
|
|
]
|
|
for future in as_completed(futures):
|
|
scenario_results.append(future.result())
|
|
duration_seconds = time.perf_counter() - started
|
|
scenario_results.sort(key=lambda item: item["iteration"])
|
|
all_results.extend(scenario_results)
|
|
scenario_summaries.append(summarize_scenario(scenario, url, scenario_results, duration_seconds))
|
|
|
|
status = "pass" if all(item["passed"] for item in scenario_summaries) else "fail"
|
|
summary = {
|
|
"status": status,
|
|
"run": {
|
|
"checked_at": checked_at(),
|
|
"contract_api_base": contract_api_base,
|
|
"ai_api_base": ai_api_base,
|
|
"requests_per_scenario": args.requests_per_scenario,
|
|
"concurrency": args.concurrency,
|
|
"warmup_requests": args.warmup,
|
|
"timeout_seconds": args.timeout_seconds,
|
|
"artifact_dir": str(artifact_dir),
|
|
"tenant_id": args.tenant_id,
|
|
"user_id": args.user_id,
|
|
"contract_proxy_auth_configured": bool(contract_proxy_key),
|
|
"ai_service_auth_configured": bool(ai_service_key),
|
|
},
|
|
"dataset": dataset,
|
|
"scenarios": scenario_summaries,
|
|
}
|
|
|
|
summary_json = artifact_dir / "interface-benchmark-summary.json"
|
|
summary_csv = artifact_dir / "interface-benchmark-results.csv"
|
|
summary_md = artifact_dir / "interface-benchmark-summary.md"
|
|
summary_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
write_csv(summary_csv, all_results)
|
|
write_markdown(summary_md, summary)
|
|
|
|
print(f"V2_INTERFACE_BENCHMARK_{status.upper()} {artifact_dir}")
|
|
return 0 if status == "pass" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|