diff --git a/projects/agentic_tools/ci_base.py b/projects/agentic_tools/ci_base.py index 22247beb3..12a8c0f70 100644 --- a/projects/agentic_tools/ci_base.py +++ b/projects/agentic_tools/ci_base.py @@ -153,24 +153,34 @@ def _get_vaults_for_key(self, key: str) -> list[str]: def _resolve_notification_provider(self): """Auto-discover notification provider from config. - Reads ``notifications.slack.provider_module`` (a fully-qualified - class path like ``projects.foo.notifications.MyProvider``) and - instantiates it. Returns None if not configured. + Preference order: + 1. ``notifications.slack.provider_module`` (custom class path) + 2. ``CaliperSlackProvider`` when ``notifications.slack.channel_id`` is set + 3. None (Slack disabled) """ provider_path = config.project.get_config( "notifications.slack.provider_module", None, print=False, warn=False ) - if not provider_path: - return None - - try: - module_path, class_name = provider_path.rsplit(".", 1) - mod = importlib.import_module(module_path) - provider_cls = getattr(mod, class_name) - return provider_cls() - except Exception as e: - logger.warning("Failed to resolve notification provider '%s': %s", provider_path, e) - return None + if provider_path: + try: + module_path, class_name = provider_path.rsplit(".", 1) + mod = importlib.import_module(module_path) + provider_cls = getattr(mod, class_name) + return provider_cls() + except Exception as e: + raise RuntimeError( + f"Failed to resolve notification provider '{provider_path}': {e}" + ) from e + + channel_id = config.project.get_config( + "notifications.slack.channel_id", None, print=False, warn=False + ) + if channel_id: + from projects.core.notifications.caliper_slack import CaliperSlackProvider + + return CaliperSlackProvider() + + return None # ------------------------------------------------------------------ # Build diff --git a/projects/agentic_tools/locust/locust_runtime/locustfile_main.py b/projects/agentic_tools/locust/locust_runtime/locustfile_main.py index f5d1bfb23..f34f85a22 100644 --- a/projects/agentic_tools/locust/locust_runtime/locustfile_main.py +++ b/projects/agentic_tools/locust/locust_runtime/locustfile_main.py @@ -20,6 +20,7 @@ _user_class_modules = [ "responses_users", "mcp_session_user", + "mcp_2026_user", ] diff --git a/projects/agentic_tools/locust/locust_users/__init__.py b/projects/agentic_tools/locust/locust_users/__init__.py index 9b5fa4c66..6ec95dbcd 100644 --- a/projects/agentic_tools/locust/locust_users/__init__.py +++ b/projects/agentic_tools/locust/locust_users/__init__.py @@ -9,6 +9,7 @@ - ResponsesMCPBenchmarkUser (responses_mcp_benchmark_user.py) - ChatCompletionsUser (chat_completions_user.py) - MCPSessionUser (mcp_session_user.py) +- MCP2026User (mcp_2026_user.py) Shared utilities: - _common.py Prompt loading, round-robin counter diff --git a/projects/agentic_tools/locust/locust_users/mcp_2026_user.py b/projects/agentic_tools/locust/locust_users/mcp_2026_user.py new file mode 100644 index 000000000..3bdf0805f --- /dev/null +++ b/projects/agentic_tools/locust/locust_users/mcp_2026_user.py @@ -0,0 +1,273 @@ +""" +Locust user for MCP 2026-07-28 (stateless) through mcp-gw or direct. + +Does not initialize, discover, or tools/list. The first RPC is tools/call. + +Reusable across any FORGE project that benchmarks MCP servers or gateways. +Copied into Locust pods as a sibling of locustfile_main.py (/scripts/). + +Import (host / tests):: + + from projects.agentic_tools.locust.locust_users.mcp_2026_user import MCP2026User + +Environment variables (set by the Locust job template): + TOOL_PREFIX: prefix for tool names ("mock_" via gateway, "" direct) + HOST_HEADER: Host header for gateway routing (empty = none) + CALLS_PER_SESSION: tool calls before rebinding host (0 = never; no handshake) + NUM_SERVERS: registered servers for scale-out (0 = single server) +""" + +from __future__ import annotations + +import json +import os +import random +import time +from dataclasses import dataclass, field + +import requests +from locust import User, between, events, task + +PROTOCOL_2026 = "2026-07-28" + +TOOL_PREFIX = os.environ.get("TOOL_PREFIX", "") +HOST_HEADER = os.environ.get("HOST_HEADER", "") +CALLS_PER_SESSION = int(os.environ.get("CALLS_PER_SESSION", "0")) +NUM_SERVERS = int(os.environ.get("NUM_SERVERS", "0")) + +TOOL_SEQUENCE = [ + {"name": "alpha", "args": {"input": "test"}}, + {"name": "bravo", "args": {"input": "test"}}, + {"name": "charlie", "args": {"input": "test"}}, + {"name": "delta", "args": {"input": "test"}}, + {"name": "echo", "args": {"input": "test"}}, + {"name": "foxtrot", "args": {"input": "test"}}, + {"name": "golf", "args": {"input": "test"}}, + {"name": "hotel", "args": {"input": "test"}}, + {"name": "india", "args": {"input": "test"}}, + {"name": "juliet", "args": {"input": "test"}}, +] + + +@dataclass +class MCPResponse: + success: bool + response_time_ms: float + status_code: int + data: dict | None = None + error: str | None = None + + +@dataclass +class MCP2026Client: + """Stateless MCP client. First RPC may be tools/call.""" + + base_url: str + host_header: str | None = None + request_id: int = field(default=0, init=False) + timeout: float = 30.0 + + def _next_id(self) -> int: + self.request_id += 1 + return self.request_id + + def _headers(self, *, method: str, name: str | None = None) -> dict: + h = { + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + "MCP-Protocol-Version": PROTOCOL_2026, + "Mcp-Method": method, + } + if name: + h["Mcp-Name"] = name + if self.host_header: + h["Host"] = self.host_header + return h + + def _with_meta(self, params: dict | None) -> dict: + out = dict(params) if params else {} + meta = dict(out["_meta"]) if isinstance(out.get("_meta"), dict) else {} + meta.setdefault("io.modelcontextprotocol/protocolVersion", PROTOCOL_2026) + meta.setdefault( + "io.modelcontextprotocol/clientInfo", + {"name": "forge-mcp-2026", "version": "1.0.0"}, + ) + meta.setdefault("io.modelcontextprotocol/clientCapabilities", {}) + out["_meta"] = meta + return out + + def _parse_body(self, response, elapsed_ms: float) -> MCPResponse: + if response.status_code != 200: + return MCPResponse( + success=False, + response_time_ms=elapsed_ms, + status_code=response.status_code, + error=f"HTTP {response.status_code}: {response.text[:200]}", + ) + + try: + ct = response.headers.get("Content-Type", "") + text = response.text + + if "text/event-stream" in ct or text.lstrip().startswith("event:"): + data = None + for line in text.split("\n"): + line = line.strip() + if line.startswith("data:"): + data = json.loads(line[5:].strip()) + break + if data is None: + return MCPResponse( + success=False, + response_time_ms=elapsed_ms, + status_code=response.status_code, + error="Empty SSE response", + ) + else: + data = response.json() + + if not isinstance(data, dict): + return MCPResponse( + success=False, + response_time_ms=elapsed_ms, + status_code=response.status_code, + error=f"Expected JSON object, got {type(data).__name__}", + ) + + if "error" in data: + msg = data["error"] + if isinstance(msg, dict): + msg = msg.get("message", str(msg)) + return MCPResponse( + success=False, + response_time_ms=elapsed_ms, + status_code=response.status_code, + error=str(msg), + ) + + return MCPResponse( + success=True, + response_time_ms=elapsed_ms, + status_code=response.status_code, + data=data.get("result"), + ) + except json.JSONDecodeError as e: + return MCPResponse( + success=False, + response_time_ms=elapsed_ms, + status_code=response.status_code, + error=f"Invalid JSON: {e}", + ) + + def _send( + self, method: str, params: dict | None = None, *, name: str | None = None + ) -> MCPResponse: + payload = { + "jsonrpc": "2.0", + "method": method, + "id": self._next_id(), + "params": self._with_meta(params), + } + start = time.perf_counter() + try: + resp = requests.post( + f"{self.base_url}/mcp", + json=payload, + headers=self._headers(method=method, name=name), + timeout=self.timeout, + ) + elapsed = (time.perf_counter() - start) * 1000 + return self._parse_body(resp, elapsed) + except requests.exceptions.RequestException as e: + elapsed = (time.perf_counter() - start) * 1000 + return MCPResponse(False, elapsed, 0, error=str(e)) + + def call_tool(self, name: str, arguments: dict | None = None) -> MCPResponse: + """POST tools/call. This is a valid first request on 2026-07-28.""" + return self._send( + "tools/call", + {"name": name, "arguments": arguments or {}}, + name=name, + ) + + +def _report(name: str, resp: MCPResponse): + if resp.success: + events.request.fire( + request_type="MCP", + name=name, + response_time=resp.response_time_ms, + response_length=len(str(resp.data)) if resp.data else 0, + exception=None, + context={}, + ) + else: + events.request.fire( + request_type="MCP", + name=f"FAIL:{name}", + response_time=resp.response_time_ms, + response_length=0, + exception=Exception(resp.error), + context={}, + ) + + +class MCP2026User(User): + """ + Stateless MCP user: round-robin tools/call with no protocol session. + + Time to first tool response is the first tools/call (Locust name ``ttftr``), + not initialize or server/discover. + """ + + abstract = True + wait_time = between(0.1, 0.5) + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.mcp: MCP2026Client | None = None + self.calls_done = 0 + self.seq_index = 0 + self._current_server_idx = 0 + self._emitted_ttftr = False + + def _bind_client(self) -> None: + """Point at a backend. No RPC.""" + host_header = HOST_HEADER or None + if NUM_SERVERS > 0: + server_idx = random.randint(1, NUM_SERVERS) + host_header = f"server{server_idx}.mcp.local" + self._current_server_idx = server_idx + else: + self._current_server_idx = 0 + + self.mcp = MCP2026Client( + base_url=self.host, + host_header=host_header, + ) + self.calls_done = 0 + + @task + def do_tool_call(self): + if self.mcp is None: + self._bind_client() + + if CALLS_PER_SESSION > 0 and self.calls_done >= CALLS_PER_SESSION: + self._bind_client() + + entry = TOOL_SEQUENCE[self.seq_index % len(TOOL_SEQUENCE)] + self.seq_index += 1 + + if NUM_SERVERS > 0 and self._current_server_idx > 0: + prefix = f"server{self._current_server_idx}_" + else: + prefix = TOOL_PREFIX + tool_name = f"{prefix}{entry['name']}" + + r = self.mcp.call_tool(tool_name, dict(entry["args"])) + if not self._emitted_ttftr: + _report("ttftr", r) + self._emitted_ttftr = True + else: + _report(f"call:{entry['name']}", r) + self.calls_done += 1 diff --git a/projects/agentic_tools/mcp/toolbox/deploy_mock_servers/main.py b/projects/agentic_tools/mcp/toolbox/deploy_mock_servers/main.py index 13693ac7e..7b0c60f29 100644 --- a/projects/agentic_tools/mcp/toolbox/deploy_mock_servers/main.py +++ b/projects/agentic_tools/mcp/toolbox/deploy_mock_servers/main.py @@ -34,9 +34,11 @@ def run( image: str, name_prefix: str = "mock-server", tools_per_server: int = 10, + protocol_mode: str = "stateful", labels: dict[str, str] | None = None, node_selector: dict[str, str] | None = None, tolerations: list[dict[str, str]] | None = None, + resources: dict[str, Any] | None = None, rollout_timeout: str = "120s", ) -> int: """Deploy mock servers and wait for readiness.""" @@ -59,6 +61,7 @@ def generate_and_apply_manifests(args, ctx): """Generate YAML manifests for all servers and apply them.""" merged_labels = dict(args.labels) if args.labels else {} merged_labels["forge.openshift.io/component"] = "mock-mcp" + merged_labels["forge.openshift.io/mcp-protocol"] = args.protocol_mode ctx.names = [f"{args.name_prefix}-{i}" for i in range(1, args.count + 1)] @@ -69,9 +72,11 @@ def generate_and_apply_manifests(args, ctx): namespace=args.namespace, image=args.image, tools_per_server=args.tools_per_server, + protocol_mode=args.protocol_mode, labels=merged_labels, node_selector=args.node_selector, tolerations=args.tolerations, + resources=args.resources, ) all_manifests.append(manifest) @@ -225,34 +230,42 @@ def _generate_server_manifest( namespace: str, image: str, tools_per_server: int, + protocol_mode: str, labels: dict[str, str], node_selector: dict[str, str] | None = None, tolerations: list[dict[str, str]] | None = None, + resources: dict[str, Any] | None = None, ) -> str: """Generate a YAML manifest for a single mock server Deployment + Service.""" label_set = {"app": name} label_set.update(labels) - pod_spec: dict[str, Any] = { - "containers": [ + container: dict[str, Any] = { + "name": "server", + "image": image, + "imagePullPolicy": "Always", + "args": ["--addr", ":8080"], + "env": [ + {"name": "GOGC", "value": "off"}, + {"name": "NUM_TOOLS", "value": str(tools_per_server)}, { - "name": "server", - "image": image, - "imagePullPolicy": "Always", - "args": ["--addr", ":8080"], - "env": [ - {"name": "GOGC", "value": "off"}, - {"name": "NUM_TOOLS", "value": str(tools_per_server)}, - ], - "ports": [{"containerPort": 8080, "name": "http"}], - "readinessProbe": { - "tcpSocket": {"port": 8080}, - "initialDelaySeconds": 2, - "periodSeconds": 5, - "timeoutSeconds": 2, - }, - } + "name": "STATELESS", + "value": "true" if protocol_mode == "stateless" else "false", + }, ], + "ports": [{"containerPort": 8080, "name": "http"}], + "readinessProbe": { + "tcpSocket": {"port": 8080}, + "initialDelaySeconds": 2, + "periodSeconds": 5, + "timeoutSeconds": 2, + }, + } + if resources: + container["resources"] = resources + + pod_spec: dict[str, Any] = { + "containers": [container], } if node_selector: pod_spec["nodeSelector"] = node_selector diff --git a/projects/caliper/prometheus_metrics/queries.yaml b/projects/caliper/prometheus_metrics/queries.yaml index 67f73e525..0ae8e7651 100644 --- a/projects/caliper/prometheus_metrics/queries.yaml +++ b/projects/caliper/prometheus_metrics/queries.yaml @@ -204,6 +204,15 @@ queries: unit: req/s description: HTTP 5xx error rate by workload + http_4xx_rate: + category: http + promql: >- + sum(rate(istio_requests_total + {destination_workload_namespace=~"{ns}",response_code=~"4.."}[1m])) + by (destination_workload) + unit: req/s + description: HTTP 4xx error rate by workload + http_latency_p50: category: http promql: >- diff --git a/projects/core/notifications/caliper_slack.py b/projects/core/notifications/caliper_slack.py new file mode 100644 index 000000000..750c35229 --- /dev/null +++ b/projects/core/notifications/caliper_slack.py @@ -0,0 +1,322 @@ +"""Default Slack notification provider driven by Caliper analyze output. + +Projects only need ``notifications.slack.channel_id`` in config. Message body +is built from ``kpi_analyze.json`` plus status, metadata labels, and MLflow link. +""" + +from __future__ import annotations + +import json +import logging +import os +import re +from pathlib import Path +from typing import Any + +from projects.core.notifications.helpers import ( + extract_mlflow_url, + format_kpi_value, + get_label_value, + get_test_artifacts_root, + read_test_duration, +) +from projects.core.notifications.provider import NotificationContext, SlackNotificationProvider + +logger = logging.getLogger(__name__) + +# Relative improvement beyond threshold is reported as improvement. +# Matches Caliper max_relative_regression semantics (fraction → percent). +_DEFAULT_INTERESTING_PCT = 10.0 + +_GROUP_LABEL_KEYS = ( + "num_servers", + "users", + "target", + "preset", + "protocol_mode", + "profile", + "model", + "accelerator", +) + + +class CaliperSlackProvider(SlackNotificationProvider): + """Config-driven Slack provider: channel_id + Caliper analyze report.""" + + def get_channel_id(self) -> str: + from projects.core.library import config + + channel_id = config.project.get_config( + "notifications.slack.channel_id", None, print=False, warn=False + ) + if not channel_id: + raise ValueError("notifications.slack.channel_id must be set in config.yaml") + if not isinstance(channel_id, str): + raise ValueError( + f"notifications.slack.channel_id must be a string, got {type(channel_id).__name__}" + ) + return channel_id + + def should_notify(self, context: NotificationContext) -> bool: + from projects.core.library import config + + if context.finish_reason != "success": + return True + if config.project.get_config( + "notifications.slack.notify_always", False, print=False, warn=False + ): + return True + report = load_analyze_report(context) + if not report: + return True + return bool(interesting_results(report)) + + def format_message(self, context: NotificationContext) -> str: + return format_caliper_slack_message(context) + + +def format_caliper_slack_message(context: NotificationContext) -> str: + """Build RHAIIS-style Slack body from export context + kpi_analyze.json.""" + report = load_analyze_report(context) + interesting = interesting_results(report) if report else [] + + if context.finish_reason != "success": + icon = ":x:" + headline = f"{context.project_name} test failed" + elif interesting: + has_reg = any(r.get("verdict") == "REGRESSION" for r in interesting) + has_imp = any(_is_improvement(r) for r in interesting) + if has_reg and has_imp: + icon = ":warning:" + headline = "Performance regressions and improvements detected" + elif has_reg: + icon = ":warning:" + headline = "Performance regressions detected" + else: + icon = ":large_green_circle:" + headline = "Performance improvements detected" + else: + icon = ":done-circle-check:" + headline = f"{context.project_name} test finished" + + parts = [f"{icon} *{headline}*"] + + meta = _format_metadata(context, report) + if meta: + parts.append(meta) + + mlflow_url = extract_mlflow_url(context) + if mlflow_url: + parts.append(f"*MLflow:* <{mlflow_url}|View Run>") + + if interesting: + parts.append("*Changes:*\n" + format_interesting_changes(interesting)) + elif context.finish_reason == "success" and report: + verdict = (report.get("overall") or {}).get("verdict") or ( + report.get("analysis") or {} + ).get("status", "") + if verdict in ("NO_BASELINE",): + parts.append("_No historical baseline for regression comparison._") + elif verdict in ("PASS", "no_regression"): + tested = (report.get("overall") or {}).get("total_tested", 0) + parts.append(f"_No significant KPI changes ({tested} KPIs compared)._") + + return "\n".join(parts) + + +def load_analyze_report(context: NotificationContext) -> dict[str, Any] | None: + """Load Caliper ``kpi_analyze.json`` from artifacts or config path. + + Returns None when no report file exists yet. Raises if a report file is + present but unreadable or not a JSON object. + """ + from projects.core.library import config + + root = get_test_artifacts_root(context) + candidates: list[Path] = [] + + configured = None + if config.project is not None: + configured = config.project.get_config( + "caliper.postprocess.analyze.output", None, print=False, warn=False + ) + + if configured and root: + candidates.append(root / configured) + if root: + candidates.append(root / "regression_analyze" / "kpi_analyze.json") + candidates.append(root / "kpi_analyze.json") + candidates.extend(sorted(root.glob("**/kpi_analyze.json"))) + if context.artifact_dir: + candidates.append(context.artifact_dir / "regression_analyze" / "kpi_analyze.json") + candidates.extend(sorted(context.artifact_dir.glob("**/kpi_analyze.json"))) + + seen: set[Path] = set() + for path in candidates: + try: + resolved = path.resolve() + except OSError as exc: + raise FileNotFoundError(f"Cannot resolve analyze report path {path}") from exc + if resolved in seen or not resolved.is_file(): + continue + seen.add(resolved) + return _read_analyze_json(resolved) + + return None + + +def _read_analyze_json(path: Path) -> dict[str, Any]: + try: + with path.open(encoding="utf-8") as handle: + data = json.load(handle) + except json.JSONDecodeError as exc: + raise ValueError(f"Analyze report is not valid JSON: {path}") from exc + except OSError as exc: + raise FileNotFoundError(f"Failed to read analyze report: {path}") from exc + + if not isinstance(data, dict): + raise ValueError(f"Analyze report must be a JSON object, got {type(data).__name__}: {path}") + return data + + +def interesting_results(report: dict[str, Any]) -> list[dict[str, Any]]: + """Return regression + improvement rows from a Caliper analyze report.""" + results = report.get("results") or [] + out: list[dict[str, Any]] = [] + for row in results: + if not isinstance(row, dict): + continue + if row.get("verdict") == "REGRESSION" or _is_improvement(row): + out.append(row) + return out + + +def format_interesting_changes(rows: list[dict[str, Any]]) -> str: + """Format grouped KPI changes for Slack.""" + by_group: dict[str, list[dict[str, Any]]] = {} + for row in rows: + key = _group_key(row.get("labels") or {}) + by_group.setdefault(key, []).append(row) + + lines: list[str] = [] + for group in sorted(by_group): + lines.append(f"\n*{group}:*") + for row in by_group[group]: + lines.append(_format_change_line(row)) + return "\n".join(lines).lstrip("\n") + + +def _is_improvement(row: dict[str, Any]) -> bool: + if row.get("verdict") == "REGRESSION": + return False + pct = float(row.get("relative_change_pct") or 0.0) + higher_is_better = bool(row.get("higher_is_better", True)) + threshold = _DEFAULT_INTERESTING_PCT + if higher_is_better: + return pct > threshold + return pct < -threshold + + +def _group_key(labels: dict[str, Any]) -> str: + parts: list[str] = [] + for key in _GROUP_LABEL_KEYS: + if key in labels and labels[key] not in (None, ""): + parts.append(f"{key}={labels[key]}") + return " / ".join(parts) if parts else "default" + + +def _format_change_line(row: dict[str, Any]) -> str: + kpi_id = str(row.get("kpi_id") or "kpi") + name = kpi_id.replace("mcp_gw_", "").replace("_", " ") + pct = float(row.get("relative_change_pct") or 0.0) + baseline = row.get("baseline_mean") + current = row.get("current_value") + unit = _guess_unit(kpi_id) + + baseline_s = format_kpi_value(float(baseline), unit) if baseline is not None else "n/a" + current_s = format_kpi_value(float(current), unit) if current is not None else "n/a" + + if row.get("verdict") == "REGRESSION": + direction = "dropped" if pct < 0 else "increased" + return ( + f" :red_circle: *{name}*: {direction} {abs(pct):.1f}% " + f"({baseline_s} \u2192 {current_s})" + ) + return ( + f" :large_green_circle: *{name}*: improved {abs(pct):.1f}% " + f"({baseline_s} \u2192 {current_s})" + ) + + +def _guess_unit(kpi_id: str) -> str: + if kpi_id.endswith("_ms"): + return "ms" + if kpi_id.endswith("_rate") and "failure" in kpi_id: + return "%" + if kpi_id.endswith("_rps") or "per_second" in kpi_id: + return "req/s" + if kpi_id.endswith("_bytes"): + return "bytes" + if kpi_id.endswith("_cores"): + return "cores" + return "" + + +def _format_metadata(context: NotificationContext, report: dict[str, Any] | None) -> str: + from projects.core.library import config + + lines: list[str] = [] + job_id = os.environ.get("FJOB_NAME") or os.environ.get("JOB_NAME_SAFE") or "" + if job_id: + lines.append(f"*Job:* `{job_id}`") + + root = get_test_artifacts_root(context) + version = os.environ.get("MCP_GATEWAY_VERSION") or get_label_value(root, "mcp_gateway_version") + preset = os.environ.get("MCP_GATEWAY_PRESET") or get_label_value(root, "preset") + if not preset: + preset = get_label_value(root, "selected_preset") + + compare_keys = [] + if report: + compare_keys = (report.get("analysis") or {}).get("config", {}).get("comparison_keys") or [] + + if version: + baseline_hint = _baseline_version_hint(report, compare_keys) + if baseline_hint: + lines.append(f"*Versions:* `{version}` vs `{baseline_hint}` (baseline)") + else: + lines.append(f"*Version:* `{version}`") + if preset: + lines.append(f"*Preset:* `{preset}`") + + duration = read_test_duration(context) + if duration: + lines.append(f"*Duration:* {duration}") + + slack_user = "" + if config.project is not None: + slack_user = ( + config.project.get_config("notifications.slack.user", "", print=False, warn=False) or "" + ) + if slack_user and re.match(r"^[UW][A-Z0-9]+$", slack_user): + lines.insert(0, f"*Triggered by:* <@{slack_user}>") + elif slack_user: + lines.insert(0, f"*Triggered by:* {slack_user}") + + return "\n".join(lines) + + +def _baseline_version_hint(report: dict[str, Any] | None, compare_keys: list[str]) -> str | None: + if not report or not compare_keys: + return None + results = report.get("results") or [] + if not results: + return None + baseline_values = results[0].get("baseline_values") or {} + if not isinstance(baseline_values, dict) or not baseline_values: + return None + # Keys look like "mcp_gateway_version=0.6.2" + first_flag = next(iter(baseline_values)) + if "=" in first_flag: + return first_flag.split("=", 1)[1] + return first_flag diff --git a/projects/core/tests/test_caliper_slack.py b/projects/core/tests/test_caliper_slack.py new file mode 100644 index 000000000..ecc72576f --- /dev/null +++ b/projects/core/tests/test_caliper_slack.py @@ -0,0 +1,286 @@ +"""Tests for CaliperSlackProvider and analyze-report Slack formatting.""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path +from types import ModuleType +from unittest.mock import MagicMock + +# Provider imports slack_sdk at module load; stub only when the package is absent. +try: + import slack_sdk # noqa: F401 +except ImportError: + _slack = ModuleType("slack_sdk") + _slack.WebClient = MagicMock # type: ignore[attr-defined] + _slack.errors = ModuleType("slack_sdk.errors") + _slack.errors.SlackApiError = type("SlackApiError", (Exception,), {}) # type: ignore[attr-defined] + sys.modules["slack_sdk"] = _slack + sys.modules["slack_sdk.errors"] = _slack.errors + +import pytest + +from projects.core.notifications.caliper_slack import ( + CaliperSlackProvider, + format_caliper_slack_message, + format_interesting_changes, + interesting_results, + load_analyze_report, +) +from projects.core.notifications.provider import NotificationContext + + +def _report(*, regressions: list[dict] | None = None, improvements: list[dict] | None = None): + results = [] + for row in regressions or []: + results.append( + { + "kpi_id": row["kpi_id"], + "labels": row.get( + "labels", {"num_servers": "1", "users": "16", "target": "gateway"} + ), + "current_value": row["current"], + "baseline_mean": row["baseline"], + "relative_change_pct": row["pct"], + "higher_is_better": row.get("higher_is_better", True), + "verdict": "REGRESSION", + "baseline_values": {"mcp_gateway_version=0.6.2": row["baseline"]}, + } + ) + for row in improvements or []: + results.append( + { + "kpi_id": row["kpi_id"], + "labels": row.get( + "labels", {"num_servers": "1", "users": "16", "target": "gateway"} + ), + "current_value": row["current"], + "baseline_mean": row["baseline"], + "relative_change_pct": row["pct"], + "higher_is_better": row.get("higher_is_better", True), + "verdict": "PASS", + "baseline_values": {"mcp_gateway_version=0.6.2": row["baseline"]}, + } + ) + return { + "analysis": { + "status": "REGRESSION_DETECTED" if regressions else "PASS", + "config": { + "comparison_keys": ["mcp_gateway_version"], + "max_relative_regression": 0.1, + }, + }, + "results": results, + "overall": { + "verdict": "REGRESSION_DETECTED" if regressions else "PASS", + "regression_count": len(regressions or []), + "total_tested": len(results), + "total_skipped": 0, + }, + } + + +def _write_report(tmp_path: Path, payload: dict) -> Path: + out = tmp_path / "regression_analyze" / "kpi_analyze.json" + out.parent.mkdir(parents=True) + out.write_text(json.dumps(payload), encoding="utf-8") + return out + + +class _FakeProject: + def __init__(self, tmp_path: Path | None = None, *, notify_always: bool = False): + self._tmp = tmp_path + self._notify_always = notify_always + + def get_config(self, key, default=None, print=False, warn=False): + if key == "notifications.slack.channel_id": + return "C0BEBS6929L" + if key == "notifications.slack.notify_always": + return self._notify_always + if key == "caliper.postprocess.analyze.output": + return "regression_analyze/kpi_analyze.json" + if key == "caliper.export.from" and self._tmp is not None: + return str(self._tmp) + return default + + +@pytest.fixture +def fake_config(monkeypatch): + def _install(tmp_path: Path | None = None, *, notify_always: bool = False): + project = _FakeProject(tmp_path, notify_always=notify_always) + # Prefer the real config module (CI). Fall back only when deps like + # jsonpath_ng are missing locally — never replace a loaded real module. + try: + import projects.core.library.config as config_mod + except ModuleNotFoundError: + config_mod = ModuleType("projects.core.library.config") + monkeypatch.setitem(sys.modules, "projects.core.library.config", config_mod) + monkeypatch.setattr(config_mod, "project", project, raising=False) + return project + + return _install + + +def test_interesting_results_picks_regressions_and_improvements(): + report = _report( + regressions=[ + { + "kpi_id": "mcp_gw_tool_call_rps", + "current": 1100, + "baseline": 1200, + "pct": -8.3, + "higher_is_better": True, + } + ], + improvements=[ + { + "kpi_id": "mcp_gw_tool_call_p95_ms", + "current": 20.0, + "baseline": 24.0, + "pct": -16.7, + "higher_is_better": False, + } + ], + ) + rows = interesting_results(report) + assert len(rows) == 2 + assert {r["kpi_id"] for r in rows} == { + "mcp_gw_tool_call_rps", + "mcp_gw_tool_call_p95_ms", + } + + +def test_format_interesting_changes_rhaiis_style(): + report = _report( + regressions=[ + { + "kpi_id": "mcp_gw_tool_call_rps", + "current": 1100, + "baseline": 1200, + "pct": -8.3, + } + ] + ) + text = format_interesting_changes(interesting_results(report)) + assert ":red_circle:" in text + assert "tool call rps" in text + assert "dropped 8.3%" in text + assert "num_servers=1" in text + + +def test_load_analyze_report_from_artifact_dir(tmp_path: Path, fake_config): + fake_config(tmp_path) + _write_report( + tmp_path, + _report( + regressions=[ + { + "kpi_id": "mcp_gw_requests_per_second", + "current": 100, + "baseline": 200, + "pct": -50.0, + } + ] + ), + ) + ctx = NotificationContext( + status={}, + finish_reason="success", + project_name="mcp_gateway", + artifact_dir=tmp_path, + ) + loaded = load_analyze_report(ctx) + assert loaded is not None + assert loaded["overall"]["regression_count"] == 1 + + +def test_format_message_regression_headline(tmp_path: Path, fake_config, monkeypatch): + fake_config(tmp_path) + monkeypatch.setenv("MCP_GATEWAY_VERSION", "0.7.0") + _write_report( + tmp_path, + _report( + regressions=[ + { + "kpi_id": "mcp_gw_tool_call_rps", + "current": 100, + "baseline": 200, + "pct": -50.0, + } + ] + ), + ) + ctx = NotificationContext( + status={ + "caliper_artifacts_export": { + "backends": {"mlflow": {"run_url": "https://mlflow.example/run/1"}} + } + }, + finish_reason="success", + project_name="mcp_gateway", + artifact_dir=tmp_path, + ) + msg = format_caliper_slack_message(ctx) + assert "Performance regressions detected" in msg + assert "MLflow" in msg + assert ":red_circle:" in msg + assert "0.7.0" in msg + assert "0.6.2" in msg + + +def test_provider_channel_id(fake_config): + fake_config() + assert CaliperSlackProvider().get_channel_id() == "C0BEBS6929L" + + +def test_load_analyze_report_raises_on_corrupt_json(tmp_path: Path, fake_config): + fake_config(tmp_path) + out = tmp_path / "regression_analyze" / "kpi_analyze.json" + out.parent.mkdir(parents=True) + out.write_text("{not-json", encoding="utf-8") + ctx = NotificationContext( + status={}, + finish_reason="success", + project_name="mcp_gateway", + artifact_dir=tmp_path, + ) + with pytest.raises(ValueError, match="not valid JSON"): + load_analyze_report(ctx) + + +def test_should_notify_skips_quiet_success(tmp_path: Path, fake_config): + fake_config(tmp_path) + _write_report( + tmp_path, + { + "analysis": { + "status": "PASS", + "config": {"comparison_keys": ["mcp_gateway_version"]}, + }, + "results": [ + { + "kpi_id": "mcp_gw_tool_call_rps", + "labels": {}, + "current_value": 101, + "baseline_mean": 100, + "relative_change_pct": 1.0, + "higher_is_better": True, + "verdict": "PASS", + } + ], + "overall": { + "verdict": "PASS", + "regression_count": 0, + "total_tested": 1, + "total_skipped": 0, + }, + }, + ) + ctx = NotificationContext( + status={}, + finish_reason="success", + project_name="mcp_gateway", + artifact_dir=tmp_path, + ) + assert CaliperSlackProvider().should_notify(ctx) is False diff --git a/projects/mcp_gateway/README.md b/projects/mcp_gateway/README.md index adfd06811..7320e021a 100644 --- a/projects/mcp_gateway/README.md +++ b/projects/mcp_gateway/README.md @@ -50,7 +50,8 @@ projects/ | Summary helpers | `agentic_tools/locust/helpers/summary` | Saving metrics.json + parameters.json for caliper multi-run export | | Locust K8s template | `agentic_tools/locust/templates/locust_job.yaml` | Base Job/Service YAML for distributed Locust deployments | | Locust runtime | `agentic_tools/locust/locust_runtime/` | Shared entry point, warmup hook, load shapes | -| MCP session user | `agentic_tools/locust/locust_users/mcp_session_user.py` | Locust user class for MCP protocol load | +| MCP session user | `agentic_tools/locust/locust_users/mcp_session_user.py` | Locust user class for MCP protocol load (stateful) | +| MCP 2026 user | `agentic_tools/locust/locust_users/mcp_2026_user.py` | Locust user class for MCP 2026-07-28 (stateless) | | Mock MCP server deployer | `agentic_tools/mcp/toolbox/deploy_mock_servers` | Deploying/restarting/cleaning up 1..N mock MCP server pods | | MCP HTTP client | `agentic_tools/mcp/clients/mcp_client.py` | Streamable HTTP client implementing MCP protocol | diff --git a/projects/mcp_gateway/orchestration/cleanup_phase.py b/projects/mcp_gateway/orchestration/cleanup_phase.py index 7310b4808..07dcd122e 100644 --- a/projects/mcp_gateway/orchestration/cleanup_phase.py +++ b/projects/mcp_gateway/orchestration/cleanup_phase.py @@ -82,7 +82,11 @@ def run_platform_cleanup() -> int: "Kustomize-based cleanup steps will be skipped." ) - cleanup_platform_mod.run(platform_config=platform_cfg) + scheduling = cfg.get_scheduling_config() + cleanup_platform_mod.run( + platform_config=platform_cfg, + scheduling_node_selector=scheduling.get("node_selector"), + ) cleanup_platform_clone() else: logger.info("Platform cleanup not enabled (cleanup_platform=false)") diff --git a/projects/mcp_gateway/orchestration/config.d/mock_servers.yaml b/projects/mcp_gateway/orchestration/config.d/mock_servers.yaml index fdbf27794..e7bdc7e3c 100644 --- a/projects/mcp_gateway/orchestration/config.d/mock_servers.yaml +++ b/projects/mcp_gateway/orchestration/config.d/mock_servers.yaml @@ -1,3 +1,3 @@ perf-mock-server: - image: quay.io/rh-ee-aharush/perf-mock-server:latest + image: quay.io/rh-ee-aharush/perf-mock-server:new-protocol tools_per_server: 10 diff --git a/projects/mcp_gateway/orchestration/config.d/runtime.yaml b/projects/mcp_gateway/orchestration/config.d/runtime.yaml index 1ad5f1db6..d7ebac624 100644 --- a/projects/mcp_gateway/orchestration/config.d/runtime.yaml +++ b/projects/mcp_gateway/orchestration/config.d/runtime.yaml @@ -3,6 +3,7 @@ namespace_override: null namespace: mcp-gw-bench mock_server: perf-mock-server +protocol_mode: stateful calls_per_session: 0 spawn_rate: null users_per_worker: 32 diff --git a/projects/mcp_gateway/orchestration/config.yaml b/projects/mcp_gateway/orchestration/config.yaml index d43adc1da..ecd9bb86e 100644 --- a/projects/mcp_gateway/orchestration/config.yaml +++ b/projects/mcp_gateway/orchestration/config.yaml @@ -33,7 +33,6 @@ vaults: notifications: slack: channel_id: C0BEBS6929L - provider_module: projects.mcp_gateway.orchestration.notifications.MCPGatewaySlackProvider metrics: enabled: true @@ -43,7 +42,21 @@ metrics: - gateway-system - istio-system step_seconds: 15 - query_keys: [] # empty = all queries from queries.yaml + query_keys: + - cpu_usage + - cpu_throttling + - memory_usage + - memory_rss + - memory_oom_kills + - container_restarts + - http_request_rate + - http_error_rate + - http_4xx_rate + - http_latency_p50 + - http_latency_p95 + - http_latency_p99 + - network_rx_bytes + - network_tx_bytes caliper: postprocess: @@ -70,6 +83,7 @@ caliper: output_dir: ai_data analyze: enabled: true + current_kpis: kpis.json historical_kpis: historical_data output: regression_analyze/kpi_analyze.json fail_on_regression: false diff --git a/projects/mcp_gateway/orchestration/notifications.py b/projects/mcp_gateway/orchestration/notifications.py deleted file mode 100644 index 0c4683960..000000000 --- a/projects/mcp_gateway/orchestration/notifications.py +++ /dev/null @@ -1,381 +0,0 @@ -""" -Per-project Slack notification provider for MCP Gateway. - -Sends structured performance summaries with KPI comparison against the -previous MLflow run. Reuses shared helpers from the core notification -framework. - -Channel ID is read from the project's config.yaml at -``notifications.slack.channel_id``. -""" - -from __future__ import annotations - -import json -import logging -import os -import re - -from projects.core.library import config -from projects.core.notifications.helpers import ( - build_comparison_table, - build_current_kpis_list, - extract_mlflow_url, - get_label_value, - get_test_artifacts_root, - read_test_duration, -) -from projects.core.notifications.provider import NotificationContext, SlackNotificationProvider -from projects.core.notifications.send import get_ocpci_link - -logger = logging.getLogger(__name__) - -TARGET_KPIS = [ - ("mcp_gw_requests_per_second", "RPS", "req/s"), - ("mcp_gw_p95_ms", "P95 latency", "ms"), - ("mcp_gw_p99_ms", "P99 latency", "ms"), - ("mcp_gw_failure_rate", "Failure rate", "%"), -] - -# KPI IDs are now logged directly as MLflow metric keys via the generic -# caliper kpis-to-metrics conversion — no mapping needed. -_KPI_IDS = {kpi_id for kpi_id, _, _ in TARGET_KPIS} - -_RUN_NAME_PREFIX = "forge-mcp-gateway-" - - -def _parse_run_name(run_name: str) -> dict | None: - """Parse a forge-mcp-gateway run name into components. - - Expected formats: - forge-mcp-gateway-s150-u500-vsha--YYYYMMDD-HHMMSS - forge-mcp-gateway-s150-u500-v0.7.0-YYYYMMDD-HHMMSS - forge-mcp-gateway-0.5.1-YYYYMMDD-HHMMSS - - Returns dict with keys: config, is_sha, version — or None if unparseable. - """ - if not run_name.startswith(_RUN_NAME_PREFIX): - return None - - rest = run_name[len(_RUN_NAME_PREFIX) :] - - config = None - config_match = re.match(r"(s\d+-u\d+)-", rest) - if config_match: - config = config_match.group(1) - rest = rest[config_match.end() :] - - is_sha = False - version = None - - if rest.startswith("vsha-"): - is_sha = True - sha_match = re.match(r"vsha-([a-f0-9]+)-\d{8}-\d{6}$", rest) - if sha_match: - version = sha_match.group(1) - else: - version_match = re.match(r"v?(.+?)-\d{8}-\d{6}$", rest) - if version_match: - version = version_match.group(1) - - return {"config": config, "is_sha": is_sha, "version": version} - - -class MCPGatewaySlackProvider(SlackNotificationProvider): - """Slack notification provider for the mcp_gateway project.""" - - def get_channel_id(self) -> str: - channel_id = config.project.get_config( - "notifications.slack.channel_id", None, print=False, warn=False - ) - if not channel_id: - raise ValueError("notifications.slack.channel_id must be set in config.yaml") - return channel_id - - def format_message(self, context: NotificationContext) -> str: - header = _format_header(context) - metadata = _format_metadata(context) - kpi_table = _format_kpi_table(context) - links = _format_standard_links(context) - failure_info = _format_failure_info(context) - - parts = [header, metadata, kpi_table, links, failure_info] - return "\n\n".join(filter(None, parts)) - - def get_thread_anchor(self, context: NotificationContext) -> str: - if context.pr_number: - return f"Thread for mcp_gateway PR #{context.pr_number}" - - job_name = os.environ.get("FJOB_NAME") or os.environ.get("JOB_NAME_SAFE", "") - if job_name: - return f"Thread for mcp_gateway `{job_name}`" - - return "Thread for mcp_gateway run" - - -# --------------------------------------------------------------------------- -# Message sections (MCP Gateway specific) -# --------------------------------------------------------------------------- - - -def _format_header(context: NotificationContext) -> str: - status_icon = ":done-circle-check:" if context.finish_reason == "success" else ":no-red-circle:" - duration = read_test_duration(context) - duration_str = f" after {duration}" if duration else "" - return f"{status_icon} *mcp_gateway test finished{duration_str}* {status_icon}" - - -def _format_metadata(context: NotificationContext) -> str: - version = os.environ.get("MCP_GATEWAY_VERSION", "") - preset = os.environ.get("MCP_GATEWAY_PRESET", "") - - test_root = get_test_artifacts_root(context) - if not version: - version = get_label_value(test_root, "mcp_gateway_version") or "unknown" - if not preset: - preset = get_label_value(test_root, "preset") or "default" - - return f"*Version*: `{version}` | *Preset*: `{preset}`" - - -def _format_kpi_table(context: NotificationContext) -> str: - """Build comparison table: current KPIs vs previous MLflow run.""" - current_kpis = _load_current_kpis(context) - if not current_kpis: - return "" - - # Extract current run_id from export status to avoid race with parallel jobs - current_run_id = None - if isinstance(context.status, dict): - backends = context.status.get("caliper_artifacts_export", {}).get("backends", {}) - current_run_id = backends.get("mlflow", {}).get("run_id") - - previous_kpis, previous_run_name, skip = _load_previous_kpis_from_mlflow(current_run_id) - - if skip: - context.extra["_skip_notification"] = True - return "" - - if previous_kpis: - return build_comparison_table(current_kpis, previous_kpis, previous_run_name, TARGET_KPIS) - else: - return build_current_kpis_list(current_kpis, TARGET_KPIS) - - -def _format_standard_links(context: NotificationContext) -> str: - """Generate artifact links pointing to MLflow.""" - mlflow_url = extract_mlflow_url(context) - if not mlflow_url: - return "" - - return f"\u2022 <{mlflow_url}|MLflow run (results & logs)>" - - -def _format_failure_info(context: NotificationContext) -> str: - """Include structured failure details when test failed.""" - if context.finish_reason == "success": - return "" - if not context.artifact_dir: - return "" - - try: - from projects.core.notifications.send import _get_notification_content - - def get_link(name, path, **kwargs): - return f"<{get_ocpci_link(path, **kwargs)}|{name}>" - - def get_bold(text): - return f"*{text}*" - - return _get_notification_content(context.artifact_dir, get_link, get_bold) - except Exception as e: - logger.warning("Failed to extract failure info: %s", e) - return "" - - -# --------------------------------------------------------------------------- -# KPI loading (MCP Gateway specific) -# --------------------------------------------------------------------------- - - -def _find_kpis_json(artifact_dir): - """Find kpis.json in artifact tree.""" - direct = artifact_dir / "kpis.json" - if direct.exists(): - return direct - for f in artifact_dir.glob("**/kpis.json"): - return f - - return None - - -def _load_current_kpis(context: NotificationContext) -> dict[str, float]: - """Read KPI values from kpis.json in the artifact directory.""" - test_root = get_test_artifacts_root(context) - if not test_root: - return {} - - kpis_file = _find_kpis_json(test_root) - if not kpis_file: - return {} - - target_ids = {k[0] for k in TARGET_KPIS} - kpis: dict[str, float] = {} - - try: - with open(kpis_file) as f: - data = json.load(f) - for test in data.get("tests", []): - for kpi_record in test.get("kpis", []): - kpi_id = kpi_record.get("id", "") - if kpi_id in target_ids: - value = kpi_record.get("value") - if value is not None: - kpis[kpi_id] = float(value) - except Exception as e: - logger.warning("Failed to read KPI file %s: %s", kpis_file, e) - - return kpis - - -def _load_previous_kpis_from_mlflow( - current_run_id: str | None = None, -) -> tuple[dict[str, float], str, bool]: - """Query MLflow for the previous matching run's metrics. - - Args: - current_run_id: MLflow run ID of this notification's run. When provided - the current run is located by ID (avoids races with parallel jobs). - - Matching rules: - - Same load config (e.g. s150-u500) - - Same version type (SHA-based compared only to SHA-based, release to release) - - Same preset parameter value - - Returns (metrics_dict, run_name, skip_notification): - - skip_notification is True when the previous matching run has the same - version/SHA (duplicate run, notification already sent). - """ - try: - from projects.caliper.engine.file_export.mlflow_secrets import ( - load_mlflow_secrets_yaml, - mlflow_connection_env, - ) - from projects.core.library import vault as vault_lib - - vault_name = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.name", None, print=False, warn=False - ) - vault_secret = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", - None, - print=False, - warn=False, - ) - experiment_name = config.project.get_config( - "caliper.export.backend.mlflow.config.experiment", None, print=False, warn=False - ) - - if not all([vault_name, vault_secret, experiment_name]): - logger.info("MLflow config incomplete, skipping comparison") - return {}, "", False - - secrets_path = vault_lib.get_vault_content_path(vault_name, vault_secret) - if not secrets_path or not secrets_path.exists(): - logger.info("MLflow secrets not available, skipping comparison") - return {}, "", False - - secrets = load_mlflow_secrets_yaml(secrets_path) - - with mlflow_connection_env(secrets): - import mlflow - - client = mlflow.tracking.MlflowClient() - exp = client.get_experiment_by_name(experiment_name) - if not exp: - logger.info("MLflow experiment '%s' not found", experiment_name) - return {}, "", False - - runs = client.search_runs( - experiment_ids=[exp.experiment_id], - order_by=["start_time DESC"], - max_results=50, - ) - - if len(runs) < 2: - logger.info("No previous MLflow run found for comparison") - return {}, "", False - - # Locate the current run explicitly by ID when available - current_run = None - if current_run_id: - for run in runs: - if run.info.run_id == current_run_id: - current_run = run - break - if current_run is None: - logger.info( - "Current run_id '%s' not found in recent runs, falling back to runs[0]", - current_run_id, - ) - if current_run is None: - current_run = runs[0] - - current_name = getattr(current_run.info, "run_name", "") or current_run.info.run_id[:8] - current_parsed = _parse_run_name(current_name) - current_preset = (current_run.data.params or {}).get("preset", "") - - if not current_parsed: - logger.info("Cannot parse current run name '%s', skipping comparison", current_name) - return {}, "", False - - # Find previous run matching: same config, same type, same preset - previous_run = None - for run in runs: - if run.info.run_id == current_run.info.run_id: - continue - candidate_name = getattr(run.info, "run_name", "") or run.info.run_id[:8] - candidate_parsed = _parse_run_name(candidate_name) - if not candidate_parsed: - continue - - if candidate_parsed["config"] != current_parsed["config"]: - continue - if candidate_parsed["is_sha"] != current_parsed["is_sha"]: - continue - - candidate_preset = (run.data.params or {}).get("preset", "") - if candidate_preset != current_preset: - continue - - previous_run = run - break - - if previous_run is None: - logger.info( - "No previous run matches config=%s, is_sha=%s, preset=%s", - current_parsed["config"], - current_parsed["is_sha"], - current_preset, - ) - return {}, "", False - - prev_name = getattr(previous_run.info, "run_name", "") or previous_run.info.run_id[:8] - prev_parsed = _parse_run_name(prev_name) - - # Check for duplicate: same version/SHA means already notified - if prev_parsed and prev_parsed["version"] == current_parsed["version"]: - logger.info( - "Previous matching run has same version '%s', skipping notification", - current_parsed["version"], - ) - return {}, "", True - - raw_metrics = previous_run.data.metrics or {} - metrics = {k: v for k, v in raw_metrics.items() if k in _KPI_IDS} - - return metrics, prev_name, False - - except Exception as e: - logger.warning("MLflow comparison unavailable: %s", e) - return {}, "", False diff --git a/projects/mcp_gateway/orchestration/preflight_phase.py b/projects/mcp_gateway/orchestration/preflight_phase.py index 0ddf8ed7f..97c16f1ed 100644 --- a/projects/mcp_gateway/orchestration/preflight_phase.py +++ b/projects/mcp_gateway/orchestration/preflight_phase.py @@ -338,6 +338,6 @@ def _get_mock_server_image() -> str: """Resolve the mock server container image from config.""" try: mock_cfg = cfg.get_mock_server_config() - return mock_cfg.get("image", "quay.io/rh-ee-aharush/perf-mock-server:latest") + return mock_cfg.get("image", "quay.io/rh-ee-aharush/perf-mock-server:new-protocol") except Exception: - return "quay.io/rh-ee-aharush/perf-mock-server:latest" + return "quay.io/rh-ee-aharush/perf-mock-server:new-protocol" diff --git a/projects/mcp_gateway/orchestration/prepare_phase.py b/projects/mcp_gateway/orchestration/prepare_phase.py index e7f855dc0..0423ba7a7 100644 --- a/projects/mcp_gateway/orchestration/prepare_phase.py +++ b/projects/mcp_gateway/orchestration/prepare_phase.py @@ -80,7 +80,11 @@ def run() -> int: f"Cannot install nightly build for commit {version}." ) - install_platform_mod.run(platform_config=platform_cfg) + scheduling = cfg.get_scheduling_config() + install_platform_mod.run( + platform_config=platform_cfg, + scheduling_node_selector=scheduling.get("node_selector"), + ) ensure_namespace( namespace, diff --git a/projects/mcp_gateway/orchestration/runtime_config.py b/projects/mcp_gateway/orchestration/runtime_config.py index 4d325832c..1fe28157b 100644 --- a/projects/mcp_gateway/orchestration/runtime_config.py +++ b/projects/mcp_gateway/orchestration/runtime_config.py @@ -25,6 +25,10 @@ def __init__(self) -> None: def get_mock_server_key(self) -> str: return config.project.get_config("runtime.mock_server") + def get_protocol_mode(self) -> str: + """Return MCP protocol mode: stateful (2025) or stateless (2026).""" + return config.project.get_config("runtime.protocol_mode", "stateful") + def get_mock_server_config(self) -> dict[str, Any]: key = self.get_mock_server_key() return copy.deepcopy(config.project.get_config(f"mock_servers.{key}")) @@ -197,20 +201,34 @@ def build_locust_kwargs(self, *, users: int, target: str, num_servers: int) -> d tool_prefix = "" host_header = "" + protocol_mode = self.get_protocol_mode() + runtime_dir = Path(locust_runtime.__file__).parent + users_dir = Path(locust_users.__file__).parent + + if protocol_mode == "stateless": + user_class = "MCP2026User" + extra_files = [ + str(users_dir / "mcp_2026_user.py"), + ] + else: + user_class = "MCPSessionUser" + extra_files = [ + str(users_dir / "mcp_session_user.py"), + str(self.get_mcp_client_path()), + ] + env_vars = { - "USER_CLASS": "MCPSessionUser", + "USER_CLASS": user_class, "TARGET": target, "TOOL_PREFIX": tool_prefix, "HOST_HEADER": host_header, "CALLS_PER_SESSION": str(self.get_calls_per_session()), "WARMUP_SECONDS": str(warmup_seconds), + "PROTOCOL_MODE": protocol_mode, } if num_servers > 1: env_vars["NUM_SERVERS"] = str(num_servers) - runtime_dir = Path(locust_runtime.__file__).parent - users_dir = Path(locust_users.__file__).parent - return dict( job_name=job_name, namespace=namespace, @@ -222,7 +240,7 @@ def build_locust_kwargs(self, *, users: int, target: str, num_servers: int) -> d configmap_name=f"locust-scripts-mcp-{preset}"[:63], locustfiles_dir=str(runtime_dir), locustfile_names=["locustfile_main.py", "metrics_hook.py"], - extra_files=[str(users_dir / "mcp_session_user.py"), str(self.get_mcp_client_path())], + extra_files=extra_files, env_vars=env_vars, labels={"forge.openshift.io/project": "mcp_gateway"}, node_selector=scheduling.get("node_selector"), diff --git a/projects/mcp_gateway/orchestration/test_phase.py b/projects/mcp_gateway/orchestration/test_phase.py index d5559b766..a36fd4cfe 100644 --- a/projects/mcp_gateway/orchestration/test_phase.py +++ b/projects/mcp_gateway/orchestration/test_phase.py @@ -42,6 +42,7 @@ def do_test() -> int: preset = cfg.get_preset_name() mock_server = cfg.get_mock_server_key() mock_server_cfg = cfg.get_mock_server_config() + protocol_mode = cfg.get_protocol_mode() version = cfg.get_deployed_version() servers = cfg.get_experiment_servers() @@ -82,6 +83,7 @@ def do_test() -> int: preset=preset, mock_server=mock_server, mock_server_cfg=mock_server_cfg, + protocol_mode=protocol_mode, version=version, num_servers=num_servers, users=users, @@ -99,6 +101,7 @@ def do_test() -> int: summary: dict[str, Any] = { "preset": preset, "version": version, + "protocol_mode": protocol_mode, "servers": servers, "concurrency": concurrency, "targets": targets, @@ -126,6 +129,7 @@ def run_one_test( preset: str, mock_server: str, mock_server_cfg: dict[str, Any], + protocol_mode: str, version: str, num_servers: int, users: int, @@ -138,6 +142,8 @@ def run_one_test( job_name: str, ) -> None: """Run a single test iteration inside a NextArtifactDir context.""" + version_kind = "sha" if re.fullmatch(r"[0-9a-f]{40}", version) else "release" + version_source = "ghcr" if version_kind == "sha" else "tag" write_test_labels( env.ARTIFACT_DIR, { @@ -146,7 +152,10 @@ def run_one_test( "users": str(users), "num_servers": str(num_servers), "mock_server": mock_server, + "protocol_mode": protocol_mode, "mcp_gateway_version": version, + "version_kind": version_kind, + "version_source": version_source, "tools_per_server": str(tools_per_server), }, ) @@ -203,7 +212,7 @@ def _deploy_servers( scheduling: dict[str, Any] | None = None, ) -> None: """Deploy mock server(s) and gateway infrastructure.""" - image = mock_server_cfg.get("image", "quay.io/rh-ee-aharush/perf-mock-server:latest") + image = mock_server_cfg.get("image", "quay.io/rh-ee-aharush/perf-mock-server:new-protocol") sched = scheduling or {} deploy_mock_servers.run( @@ -211,9 +220,11 @@ def _deploy_servers( count=num_servers, image=image, tools_per_server=tools_per_server, + protocol_mode=cfg.get_protocol_mode(), labels={"forge.openshift.io/project": "mcp_gateway"}, node_selector=sched.get("node_selector"), tolerations=sched.get("tolerations"), + resources=mock_server_cfg.get("resources"), ) if "gateway" in targets: diff --git a/projects/mcp_gateway/postprocess/mcp_gateway/parsing/kpis.py b/projects/mcp_gateway/postprocess/mcp_gateway/parsing/kpis.py index 274c1b62b..79bec47dd 100644 --- a/projects/mcp_gateway/postprocess/mcp_gateway/parsing/kpis.py +++ b/projects/mcp_gateway/postprocess/mcp_gateway/parsing/kpis.py @@ -19,15 +19,24 @@ from projects.caliper.engine.model import UnifiedRunModel +def _require(unified_record, key: str) -> float: + value = unified_record.metrics.get(key) + if value is None: + raise ValueError(f"{key} metric not found") + return float(value) + + +# --------------------------------------------------------------------------- +# Locust aggregated (all operations mixed) — kept for continuity +# --------------------------------------------------------------------------- + + @HigherBetter() @Format("{:.5f}") @KPIMetadata(help="Sustained request throughput", unit="req/s") def mcp_gw_requests_per_second(unified_record) -> float: """Request Rate KPI.""" - value = unified_record.metrics.get("requests_per_second") - if value is None: - raise ValueError("requests_per_second metric not found") - return float(value) + return _require(unified_record, "requests_per_second") @LowerBetter() @@ -35,10 +44,7 @@ def mcp_gw_requests_per_second(unified_record) -> float: @KPIMetadata(help="Mean response time across all requests", unit="ms") def mcp_gw_avg_response_time_ms(unified_record) -> float: """Average Response Time KPI.""" - value = unified_record.metrics.get("avg_response_time_ms") - if value is None: - raise ValueError("avg_response_time_ms metric not found") - return float(value) + return _require(unified_record, "avg_response_time_ms") @LowerBetter() @@ -46,10 +52,7 @@ def mcp_gw_avg_response_time_ms(unified_record) -> float: @KPIMetadata(help="Median (P50) response latency", unit="ms") def mcp_gw_p50_ms(unified_record) -> float: """P50 Latency KPI.""" - value = unified_record.metrics.get("p50_ms") - if value is None: - raise ValueError("p50_ms metric not found") - return float(value) + return _require(unified_record, "p50_ms") @LowerBetter() @@ -57,10 +60,7 @@ def mcp_gw_p50_ms(unified_record) -> float: @KPIMetadata(help="95th percentile response latency", unit="ms") def mcp_gw_p95_ms(unified_record) -> float: """P95 Latency KPI.""" - value = unified_record.metrics.get("p95_ms") - if value is None: - raise ValueError("p95_ms metric not found") - return float(value) + return _require(unified_record, "p95_ms") @LowerBetter() @@ -68,10 +68,7 @@ def mcp_gw_p95_ms(unified_record) -> float: @KPIMetadata(help="99th percentile response latency", unit="ms") def mcp_gw_p99_ms(unified_record) -> float: """P99 Latency KPI.""" - value = unified_record.metrics.get("p99_ms") - if value is None: - raise ValueError("p99_ms metric not found") - return float(value) + return _require(unified_record, "p99_ms") @LowerBetter() @@ -79,10 +76,163 @@ def mcp_gw_p99_ms(unified_record) -> float: @KPIMetadata(help="Fraction of failed requests", unit="%") def mcp_gw_failure_rate(unified_record) -> float: """Failure Rate KPI.""" - value = unified_record.metrics.get("failure_rate") - if value is None: - raise ValueError("failure_rate metric not found") - return float(value) + return _require(unified_record, "failure_rate") + + +# --------------------------------------------------------------------------- +# Locust per-operation (call:* / handshake / tools/list) +# --------------------------------------------------------------------------- + + +@HigherBetter() +@Format("{:.5f}") +@KPIMetadata(help="Sustained tools/call throughput", unit="req/s") +def mcp_gw_tool_call_rps(unified_record) -> float: + """Tool Call Rate KPI.""" + return _require(unified_record, "tool_call_rps") + + +@LowerBetter() +@Format("{:.5f}") +@KPIMetadata(help="Median (P50) tools/call latency", unit="ms") +def mcp_gw_tool_call_p50_ms(unified_record) -> float: + """Tool Call P50 Latency KPI.""" + return _require(unified_record, "tool_call_p50_ms") + + +@LowerBetter() +@Format("{:.5f}") +@KPIMetadata(help="95th percentile tools/call latency", unit="ms") +def mcp_gw_tool_call_p95_ms(unified_record) -> float: + """Tool Call P95 Latency KPI.""" + return _require(unified_record, "tool_call_p95_ms") + + +@LowerBetter() +@Format("{:.5f}") +@KPIMetadata(help="99th percentile tools/call latency", unit="ms") +def mcp_gw_tool_call_p99_ms(unified_record) -> float: + """Tool Call P99 Latency KPI.""" + return _require(unified_record, "tool_call_p99_ms") + + +@LowerBetter() +@Format("{:.5f}") +@KPIMetadata(help="Fraction of failed tools/call requests", unit="%") +def mcp_gw_tool_call_failure_rate(unified_record) -> float: + """Tool Call Failure Rate KPI.""" + return _require(unified_record, "tool_call_failure_rate") + + +@LowerBetter() +@Format("{:.5f}") +@KPIMetadata(help="95th percentile handshake latency (initialize or server/discover)", unit="ms") +def mcp_gw_handshake_p95_ms(unified_record) -> float: + """Handshake P95 Latency KPI.""" + return _require(unified_record, "handshake_p95_ms") + + +@LowerBetter() +@Format("{:.5f}") +@KPIMetadata( + help="95th percentile time to first tool response (2026 hot path; no handshake)", unit="ms" +) +def mcp_gw_ttftr_p95_ms(unified_record) -> float: + """Time to first tool response P95 KPI.""" + return _require(unified_record, "ttftr_p95_ms") + + +@LowerBetter() +@Format("{:.5f}") +@KPIMetadata(help="95th percentile tools/list latency", unit="ms") +def mcp_gw_tools_list_p95_ms(unified_record) -> float: + """Tools List P95 Latency KPI.""" + return _require(unified_record, "tools_list_p95_ms") + + +@HigherBetter() +@Format("{:.5f}") +@KPIMetadata(help="tools/list throughput", unit="req/s") +def mcp_gw_tools_list_rps(unified_record) -> float: + """Tools List Rate KPI.""" + return _require(unified_record, "tools_list_rps") + + +# --------------------------------------------------------------------------- +# Prometheus: broker (mcp-system) and Envoy gateway (gateway-system) +# --------------------------------------------------------------------------- + + +@LowerBetter() +@Format("{:.5f}") +@KPIMetadata(help="Average broker pod CPU over the test window", unit="cores") +def mcp_gw_broker_cpu_avg_cores(unified_record) -> float: + """Broker CPU Average KPI.""" + return _require(unified_record, "broker_cpu_avg_cores") + + +@LowerBetter() +@Format("{:.5f}") +@KPIMetadata(help="Peak broker pod CPU over the test window", unit="cores") +def mcp_gw_broker_cpu_max_cores(unified_record) -> float: + """Broker CPU Peak KPI.""" + return _require(unified_record, "broker_cpu_max_cores") + + +@LowerBetter() +@Format("{:.0f}") +@KPIMetadata(help="Average broker pod memory working set", unit="bytes") +def mcp_gw_broker_memory_avg_bytes(unified_record) -> float: + """Broker Memory Average KPI.""" + return _require(unified_record, "broker_memory_avg_bytes") + + +@LowerBetter() +@Format("{:.0f}") +@KPIMetadata(help="Peak broker pod memory working set", unit="bytes") +def mcp_gw_broker_memory_max_bytes(unified_record) -> float: + """Broker Memory Peak KPI.""" + return _require(unified_record, "broker_memory_max_bytes") + + +@LowerBetter() +@Format("{:.5f}") +@KPIMetadata(help="Average Envoy/gateway pod CPU over the test window", unit="cores") +def mcp_gw_envoy_cpu_avg_cores(unified_record) -> float: + """Envoy CPU Average KPI.""" + return _require(unified_record, "envoy_cpu_avg_cores") + + +@LowerBetter() +@Format("{:.5f}") +@KPIMetadata(help="Peak Envoy/gateway pod CPU over the test window", unit="cores") +def mcp_gw_envoy_cpu_max_cores(unified_record) -> float: + """Envoy CPU Peak KPI.""" + return _require(unified_record, "envoy_cpu_max_cores") + + +@LowerBetter() +@Format("{:.0f}") +@KPIMetadata(help="Average Envoy/gateway pod memory working set", unit="bytes") +def mcp_gw_envoy_memory_avg_bytes(unified_record) -> float: + """Envoy Memory Average KPI.""" + return _require(unified_record, "envoy_memory_avg_bytes") + + +@LowerBetter() +@Format("{:.0f}") +@KPIMetadata(help="Peak Envoy/gateway pod memory working set", unit="bytes") +def mcp_gw_envoy_memory_max_bytes(unified_record) -> float: + """Envoy Memory Peak KPI.""" + return _require(unified_record, "envoy_memory_max_bytes") + + +@LowerBetter() +@Format("{:.5f}") +@KPIMetadata(help="Istio HTTP 4xx rate (protocol validation / client errors)", unit="req/s") +def mcp_gw_http_4xx_rate(unified_record) -> float: + """HTTP 4xx Rate KPI.""" + return _require(unified_record, "http_4xx_rate") class MCPGatewayKpiHandler: @@ -93,6 +243,10 @@ class MCPGatewayKpiHandler: "preset": "distinguishing_labels.preset", "num_servers": "distinguishing_labels.num_servers", "users": "distinguishing_labels.users", + "target": "distinguishing_labels.target", + "protocol_mode": "distinguishing_labels.protocol_mode", + "mcp_gateway_version": "distinguishing_labels.mcp_gateway_version", + "version_kind": "distinguishing_labels.version_kind", } ) diff --git a/projects/mcp_gateway/postprocess/mcp_gateway/parsing/parsers.py b/projects/mcp_gateway/postprocess/mcp_gateway/parsing/parsers.py index 680d01e06..eb40c23fe 100644 --- a/projects/mcp_gateway/postprocess/mcp_gateway/parsing/parsers.py +++ b/projects/mcp_gateway/postprocess/mcp_gateway/parsing/parsers.py @@ -1,4 +1,4 @@ -"""MCP Gateway Caliper parser: reads Locust stats.csv artifacts.""" +"""MCP Gateway Caliper parser: Locust stats.csv plus Prometheus capture JSON.""" from __future__ import annotations @@ -12,10 +12,16 @@ UnifiedResultRecord, ) +from .prom_summary import summarize_prom_artifacts + logger = logging.getLogger(__name__) STATS_CSV = "stats.csv" +_HANDSHAKE_NAMES = ("initialize", "server/discover", "discover") +_TOOLS_LIST_NAME = "tools/list" +_TTFTR_NAME = "ttftr" + def _labels_from_node(node: TestBaseNode) -> dict[str, Any]: """Extract distinguishing labels from a test node.""" @@ -30,7 +36,7 @@ def _labels_from_node(node: TestBaseNode) -> dict[str, Any]: def _run_metrics_to_dict(metrics: RunMetrics) -> dict[str, Any]: """Convert RunMetrics to a flat dictionary for unified result records.""" - return { + out: dict[str, Any] = { "total_requests": metrics.total_requests, "total_failures": metrics.total_failures, "failure_rate": round(metrics.failure_rate, 6), @@ -42,6 +48,78 @@ def _run_metrics_to_dict(metrics: RunMetrics) -> dict[str, Any]: "max_ms": round(metrics.max_ms, 3), "requests_per_second": round(metrics.requests_per_second, 3), } + out.update(_operation_metrics(metrics.per_request_metrics)) + return out + + +def _row_name(key: str) -> str: + """Strip the Locust Type prefix (``MCP:call:alpha`` → ``call:alpha``).""" + if ":" not in key: + return key + return key.split(":", 1)[1] + + +def _operation_metrics(per_request: dict[str, dict[str, float]]) -> dict[str, float]: + """Flatten Locust per-name rows into handshake / tools/list / tool-call scalars.""" + out: dict[str, float] = {} + + call_ok: list[dict[str, float]] = [] + call_fail: list[dict[str, float]] = [] + for key, row in per_request.items(): + name = _row_name(key) + if name.startswith("FAIL:call:"): + call_fail.append(row) + elif name.startswith("call:"): + call_ok.append(row) + + if call_ok or call_fail: + ok_count = sum(row.get("count", 0) for row in call_ok) + fail_from_fail_rows = sum(row.get("count", 0) for row in call_fail) + fail_from_ok_rows = sum(row.get("failures", 0) for row in call_ok) + fail_count = fail_from_fail_rows + fail_from_ok_rows + total = ok_count + fail_from_fail_rows + out["tool_call_rps"] = round(sum(row.get("rps", 0.0) for row in call_ok + call_fail), 3) + if total > 0: + out["tool_call_failure_rate"] = round(fail_count / total, 6) + if call_ok: + for percentile in ("p50_ms", "p95_ms", "p99_ms"): + weighted = _weighted_avg(call_ok, percentile) + if weighted is not None: + out[f"tool_call_{percentile}"] = round(weighted, 3) + + handshake = _first_named_row(per_request, _HANDSHAKE_NAMES) + if handshake is not None: + out["handshake_p95_ms"] = round(handshake.get("p95_ms", 0.0), 3) + + ttftr = _first_named_row(per_request, (_TTFTR_NAME,)) + if ttftr is not None: + out["ttftr_p95_ms"] = round(ttftr.get("p95_ms", 0.0), 3) + + tools_list = _first_named_row(per_request, (_TOOLS_LIST_NAME,)) + if tools_list is not None: + out["tools_list_p95_ms"] = round(tools_list.get("p95_ms", 0.0), 3) + out["tools_list_rps"] = round(tools_list.get("rps", 0.0), 3) + + return out + + +def _first_named_row( + per_request: dict[str, dict[str, float]], + names: tuple[str, ...], +) -> dict[str, float] | None: + by_name = {_row_name(key): row for key, row in per_request.items()} + for name in names: + row = by_name.get(name) + if row is not None: + return row + return None + + +def _weighted_avg(rows: list[dict[str, float]], field: str) -> float | None: + total = sum(row.get("count", 0) for row in rows) + if total <= 0: + return None + return sum(row.get(field, 0.0) * row.get("count", 0) for row in rows) / total class MCPGatewayParser: @@ -77,6 +155,7 @@ def parse(self, nodes: list[TestBaseNode]) -> ParseResult: labels = _labels_from_node(node) metrics_dict = _run_metrics_to_dict(run_metrics) + metrics_dict.update(summarize_prom_artifacts(node.artifact_paths)) records.append( UnifiedResultRecord( diff --git a/projects/mcp_gateway/postprocess/mcp_gateway/parsing/prom_summary.py b/projects/mcp_gateway/postprocess/mcp_gateway/parsing/prom_summary.py new file mode 100644 index 000000000..a2e6fc757 --- /dev/null +++ b/projects/mcp_gateway/postprocess/mcp_gateway/parsing/prom_summary.py @@ -0,0 +1,101 @@ +"""Summarize Prometheus capture JSON into scalar KPI inputs. + +Capture files are written by ``caliper.prometheus_metrics.capture`` as +``{query_key}.json`` with a Prometheus ``query_range`` response inside. +""" + +from __future__ import annotations + +import json +import logging +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + +BROKER_NAMESPACE = "mcp-system" +ENVOY_NAMESPACE = "gateway-system" + +_PROM_FILES = ("cpu_usage.json", "memory_usage.json", "http_4xx_rate.json") + + +def summarize_prom_artifacts(artifact_paths: list[Path]) -> dict[str, float]: + """Return flattened broker/envoy CPU+memory and Istio 4xx scalars.""" + by_name = {path.name: path for path in artifact_paths if path.name in _PROM_FILES} + out: dict[str, float] = {} + + cpu_series = _load_matrix(by_name.get("cpu_usage.json")) + mem_series = _load_matrix(by_name.get("memory_usage.json")) + http_4xx_series = _load_matrix(by_name.get("http_4xx_rate.json")) + + _merge_ns_stats(out, "broker_cpu", cpu_series, BROKER_NAMESPACE, unit_suffix="cores") + _merge_ns_stats(out, "broker_memory", mem_series, BROKER_NAMESPACE, unit_suffix="bytes") + _merge_ns_stats(out, "envoy_cpu", cpu_series, ENVOY_NAMESPACE, unit_suffix="cores") + _merge_ns_stats(out, "envoy_memory", mem_series, ENVOY_NAMESPACE, unit_suffix="bytes") + + fourxx_values = _summed_values(http_4xx_series) + if fourxx_values: + out["http_4xx_rate"] = sum(fourxx_values) / len(fourxx_values) + elif "http_4xx_rate.json" in by_name: + out["http_4xx_rate"] = 0.0 + + return out + + +def _load_matrix(path: Path | None) -> list[dict[str, Any]]: + if path is None or not path.is_file(): + return [] + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + logger.warning("Failed to read Prometheus capture %s: %s", path, exc) + return [] + + response = payload.get("response") or {} + if response.get("status") != "success": + logger.warning( + "Prometheus capture %s status=%s", + path.name, + response.get("status", "missing"), + ) + return [] + data = response.get("data") or {} + result = data.get("result") + return result if isinstance(result, list) else [] + + +def _merge_ns_stats( + out: dict[str, float], + prefix: str, + series: list[dict[str, Any]], + namespace: str, + *, + unit_suffix: str, +) -> None: + matched = [item for item in series if _series_namespace(item) == namespace] + values = _summed_values(matched) + if not values: + return + out[f"{prefix}_avg_{unit_suffix}"] = sum(values) / len(values) + out[f"{prefix}_max_{unit_suffix}"] = max(values) + + +def _series_namespace(series: dict[str, Any]) -> str: + metric = series.get("metric") or {} + return str(metric.get("namespace") or metric.get("destination_workload_namespace") or "") + + +def _summed_values(series_list: list[dict[str, Any]]) -> list[float]: + """Sum matching series at each timestamp, then return the time series.""" + by_ts: dict[float, float] = {} + for series in series_list: + for pair in series.get("values") or []: + if not isinstance(pair, (list, tuple)) or len(pair) < 2: + continue + try: + ts = float(pair[0]) + value = float(pair[1]) + except (TypeError, ValueError): + continue + by_ts[ts] = by_ts.get(ts, 0.0) + value + return [by_ts[ts] for ts in sorted(by_ts)] diff --git a/projects/mcp_gateway/postprocess/mcp_gateway/plugin.py b/projects/mcp_gateway/postprocess/mcp_gateway/plugin.py index 78b5cdd50..3112b4d5b 100644 --- a/projects/mcp_gateway/postprocess/mcp_gateway/plugin.py +++ b/projects/mcp_gateway/postprocess/mcp_gateway/plugin.py @@ -5,6 +5,7 @@ import logging from typing import Any +from projects.caliper.engine.kpi.analyze import AnalysisConfig from projects.caliper.engine.model import ( ParseResult, PostProcessingPlugin, @@ -16,6 +17,15 @@ logger = logging.getLogger(__name__) +# Compare versions while matching on load shape / target / protocol. +analysis_config = AnalysisConfig( + comparison_keys=["mcp_gateway_version"], + ignored_keys=[], + sorting_keys=["num_servers", "users", "target"], + max_relative_regression=0.10, + min_baseline_points=1, +) + class MCPGatewayPlugin(PostProcessingPlugin): """Parses Locust stats.csv artifacts from MCP Gateway performance tests.""" diff --git a/projects/mcp_gateway/postprocess/tests/test_mcp_gateway_plugin.py b/projects/mcp_gateway/postprocess/tests/test_mcp_gateway_plugin.py index c71df8209..80728f944 100644 --- a/projects/mcp_gateway/postprocess/tests/test_mcp_gateway_plugin.py +++ b/projects/mcp_gateway/postprocess/tests/test_mcp_gateway_plugin.py @@ -2,12 +2,14 @@ from __future__ import annotations +import json from pathlib import Path import pytest import yaml from projects.caliper.engine.model import TestBaseNode, UnifiedRunModel +from projects.caliper.prometheus_metrics.queries import load_queries from projects.mcp_gateway.postprocess.mcp_gateway.parsing.kpis import MCPGatewayKpiHandler from projects.mcp_gateway.postprocess.mcp_gateway.parsing.parsers import MCPGatewayParser from projects.mcp_gateway.postprocess.mcp_gateway.plugin import MCPGatewayPlugin, get_plugin @@ -20,10 +22,66 @@ ",Aggregated,1000,2,33.5,30,60,80,120,300,31.5\n" ) -TEST_LABELS = {"preset": "smoke", "target": "gateway", "users": "16", "num_servers": "1"} +MCP_STATS_CSV = ( + "Type,Name,Request Count,Failure Count,Average Response Time," + "50%,90%,95%,99%,Max Response Time,Requests/s\n" + "MCP,initialize,16,0,40.0,38,50,55,70,90,0.5\n" + "MCP,tools/list,16,0,20.0,18,28,30,40,50,0.5\n" + "MCP,call:alpha,400,0,12.0,10,18,20,28,40,12.5\n" + "MCP,call:bravo,400,0,14.0,12,20,24,32,45,12.5\n" + "MCP,FAIL:call:alpha,10,10,80.0,70,90,100,120,150,0.3\n" + ",Aggregated,842,10,14.5,12,22,26,40,150,26.3\n" +) +STATELESS_STATS_CSV = ( + "Type,Name,Request Count,Failure Count,Average Response Time," + "50%,90%,95%,99%,Max Response Time,Requests/s\n" + "MCP,ttftr,16,0,11.0,9,14,16,20,28,0.5\n" + "MCP,call:alpha,800,0,9.0,8,12,14,18,25,25.0\n" + ",Aggregated,816,0,9.1,8,13,15,20,40,25.5\n" +) -def _make_test_node(base_dir: Path, name: str, stats_csv: str, labels: dict) -> TestBaseNode: +TEST_LABELS = { + "preset": "smoke", + "target": "gateway", + "users": "16", + "num_servers": "1", + "protocol_mode": "stateful", + "mcp_gateway_version": "0.7.0", + "version_kind": "release", +} + + +def _prom_capture(query_key: str, series: list[dict]) -> dict: + return { + "query_key": query_key, + "response": { + "status": "success", + "data": {"resultType": "matrix", "result": series}, + }, + } + + +def _series( + namespace: str, pod: str, values: list[float], extra_metric: dict | None = None +) -> dict: + metric = {"namespace": namespace, "pod": pod} + if extra_metric: + metric.update(extra_metric) + return { + "metric": metric, + "values": [[1000.0 + i, str(v)] for i, v in enumerate(values)], + } + + +def _make_test_node( + base_dir: Path, + name: str, + stats_csv: str, + labels: dict, + *, + prom_files: dict[str, dict] | None = None, +) -> TestBaseNode: """Create a test base directory with stats.csv and __test_labels__.yaml.""" node_dir = base_dir / name node_dir.mkdir(parents=True, exist_ok=True) @@ -35,16 +93,52 @@ def _make_test_node(base_dir: Path, name: str, stats_csv: str, labels: dict) -> encoding="utf-8", ) + if prom_files: + raw_dir = node_dir / "metrics" / "raw" + raw_dir.mkdir(parents=True, exist_ok=True) + for filename, payload in prom_files.items(): + (raw_dir / filename).write_text(json.dumps(payload), encoding="utf-8") + artifact_paths = sorted( p for p in node_dir.rglob("*") if p.is_file() and p.name != "__test_labels__.yaml" ) return TestBaseNode( directory=node_dir, + test_path=Path(name), test_labels={"version": "1", "labels": labels}, artifact_paths=artifact_paths, ) +def _prom_files() -> dict[str, dict]: + return { + "cpu_usage.json": _prom_capture( + "cpu_usage", + [ + _series("mcp-system", "mcp-gateway-abc", [0.2, 0.4, 0.6]), + _series("gateway-system", "mcp-gateway-istio-xyz", [0.1, 0.1, 0.4]), + _series("mcp-gw-bench", "locust-master", [0.9, 0.9, 0.9]), + ], + ), + "memory_usage.json": _prom_capture( + "memory_usage", + [ + _series("mcp-system", "mcp-gateway-abc", [100.0, 200.0, 300.0]), + _series("gateway-system", "mcp-gateway-istio-xyz", [50.0, 50.0, 80.0]), + ], + ), + "http_4xx_rate.json": _prom_capture( + "http_4xx_rate", + [ + { + "metric": {"destination_workload": "mcp-gateway-istio"}, + "values": [[1000.0, "0.2"], [1001.0, "0.4"]], + } + ], + ), + } + + # --------------------------------------------------------------------------- # Parser tests # --------------------------------------------------------------------------- @@ -55,7 +149,7 @@ def test_parse_creates_records(self, tmp_path: Path): node = _make_test_node(tmp_path, "run-a", SAMPLE_STATS_CSV, TEST_LABELS) parser = MCPGatewayParser() - result = parser.parse(tmp_path, [node]) + result = parser.parse([node]) assert len(result.records) == 1 assert result.warnings == [] @@ -70,7 +164,7 @@ def test_parse_does_not_write_metrics_json(self, tmp_path: Path): node = _make_test_node(tmp_path, "run-a", SAMPLE_STATS_CSV, TEST_LABELS) parser = MCPGatewayParser() - parser.parse(tmp_path, [node]) + parser.parse([node]) assert not (tmp_path / "run-a" / "metrics.json").exists() @@ -80,25 +174,26 @@ def test_parse_no_stats_csv(self, tmp_path: Path): (node_dir / "master.log").write_text("log") node = TestBaseNode( directory=node_dir, + test_path=Path("run-empty"), test_labels={"version": "1", "labels": TEST_LABELS}, artifact_paths=[node_dir / "master.log"], ) parser = MCPGatewayParser() - result = parser.parse(tmp_path, [node]) + result = parser.parse([node]) assert len(result.records) == 1 assert result.records[0].metrics.get("no_stats_csv_found") is True assert result.records[0].parse_notes == ["No stats.csv file found"] def test_parse_multiple_nodes(self, tmp_path: Path): - labels_a = {"preset": "smoke", "target": "gateway", "users": "16", "num_servers": "1"} - labels_b = {"preset": "smoke", "target": "gateway", "users": "64", "num_servers": "2"} + labels_a = {**TEST_LABELS, "users": "16"} + labels_b = {**TEST_LABELS, "users": "64", "num_servers": "2"} node_a = _make_test_node(tmp_path, "run-a", SAMPLE_STATS_CSV, labels_a) node_b = _make_test_node(tmp_path, "run-b", SAMPLE_STATS_CSV, labels_b) parser = MCPGatewayParser() - result = parser.parse(tmp_path, [node_a, node_b]) + result = parser.parse([node_a, node_b]) assert len(result.records) == 2 paths = {r.test_base_path for r in result.records} @@ -109,6 +204,49 @@ def test_parse_multiple_nodes(self, tmp_path: Path): assert not (tmp_path / node_name / "metrics.json").exists() assert not (tmp_path / node_name / "parameters.json").exists() + def test_parse_promotes_locust_operations(self, tmp_path: Path): + node = _make_test_node(tmp_path, "run-mcp", MCP_STATS_CSV, TEST_LABELS) + record = MCPGatewayParser().parse([node]).records[0] + + assert record.metrics["handshake_p95_ms"] == 55.0 + assert record.metrics["tools_list_p95_ms"] == 30.0 + assert record.metrics["tools_list_rps"] == 0.5 + assert record.metrics["tool_call_rps"] == pytest.approx(25.3) + assert record.metrics["tool_call_p95_ms"] == pytest.approx(22.0) + assert record.metrics["tool_call_failure_rate"] == pytest.approx(10 / 810, abs=1e-6) + + def test_parse_stateless_ttftr(self, tmp_path: Path): + labels = {**TEST_LABELS, "protocol_mode": "stateless"} + node = _make_test_node(tmp_path, "run-sl", STATELESS_STATS_CSV, labels) + record = MCPGatewayParser().parse([node]).records[0] + + assert "handshake_p95_ms" not in record.metrics + assert "tools_list_p95_ms" not in record.metrics + assert record.metrics["ttftr_p95_ms"] == 16.0 + assert record.distinguishing_labels["protocol_mode"] == "stateless" + + def test_parse_prometheus_resource_kpis(self, tmp_path: Path): + node = _make_test_node( + tmp_path, "run-prom", MCP_STATS_CSV, TEST_LABELS, prom_files=_prom_files() + ) + record = MCPGatewayParser().parse([node]).records[0] + + assert record.metrics["broker_cpu_avg_cores"] == pytest.approx(0.4) + assert record.metrics["broker_cpu_max_cores"] == pytest.approx(0.6) + assert record.metrics["envoy_cpu_avg_cores"] == pytest.approx(0.2) + assert record.metrics["envoy_cpu_max_cores"] == pytest.approx(0.4) + assert record.metrics["broker_memory_avg_bytes"] == pytest.approx(200.0) + assert record.metrics["envoy_memory_max_bytes"] == pytest.approx(80.0) + assert record.metrics["http_4xx_rate"] == pytest.approx(0.3) + + def test_parse_http_4xx_zero_when_empty(self, tmp_path: Path): + prom = { + "http_4xx_rate.json": _prom_capture("http_4xx_rate", []), + } + node = _make_test_node(tmp_path, "run-4xx", MCP_STATS_CSV, TEST_LABELS, prom_files=prom) + record = MCPGatewayParser().parse([node]).records[0] + assert record.metrics["http_4xx_rate"] == 0.0 + # --------------------------------------------------------------------------- # KPI tests @@ -119,7 +257,7 @@ class TestMCPGatewayKpis: def test_compute_kpis(self, tmp_path: Path): node = _make_test_node(tmp_path, "run-a", SAMPLE_STATS_CSV, TEST_LABELS) parser = MCPGatewayParser() - parse_result = parser.parse(tmp_path, [node]) + parse_result = parser.parse([node]) model = UnifiedRunModel( plugin_module="projects.mcp_gateway.postprocess.mcp_gateway.plugin", @@ -136,6 +274,34 @@ def test_compute_kpis(self, tmp_path: Path): assert kpi_dict["mcp_gw_requests_per_second"]["value"] == 31.5 assert kpi_dict["mcp_gw_p95_ms"]["value"] == 80.0 assert kpi_dict["mcp_gw_failure_rate"]["value"] == pytest.approx(0.002, abs=1e-4) + assert "mcp_gw_tool_call_rps" not in kpi_dict + + def test_compute_operation_and_prom_kpis(self, tmp_path: Path): + node = _make_test_node( + tmp_path, "run-full", MCP_STATS_CSV, TEST_LABELS, prom_files=_prom_files() + ) + parse_result = MCPGatewayParser().parse([node]) + model = UnifiedRunModel( + plugin_module="projects.mcp_gateway.postprocess.mcp_gateway.plugin", + base_directory=str(tmp_path), + test_nodes=[node], + unified_result_records=parse_result.records, + ) + + kpis = MCPGatewayKpiHandler.compute_kpis(model) + kpi_dict = {k["kpi_id"]: k for k in kpis} + + assert kpi_dict["mcp_gw_tool_call_rps"]["value"] == pytest.approx(25.3) + assert kpi_dict["mcp_gw_tool_call_p95_ms"]["value"] == pytest.approx(22.0) + assert kpi_dict["mcp_gw_handshake_p95_ms"]["value"] == 55.0 + assert kpi_dict["mcp_gw_tools_list_p95_ms"]["value"] == 30.0 + assert kpi_dict["mcp_gw_broker_cpu_avg_cores"]["value"] == pytest.approx(0.4) + assert kpi_dict["mcp_gw_envoy_memory_max_bytes"]["value"] == pytest.approx(80.0) + assert kpi_dict["mcp_gw_http_4xx_rate"]["value"] == pytest.approx(0.3) + assert kpi_dict["mcp_gw_tool_call_p95_ms"]["labels"]["protocol_mode"] == "stateful" + assert kpi_dict["mcp_gw_tool_call_p95_ms"]["labels"]["target"] == "gateway" + assert kpi_dict["mcp_gw_tool_call_p95_ms"]["labels"]["mcp_gateway_version"] == "0.7.0" + assert kpi_dict["mcp_gw_tool_call_p95_ms"]["labels"]["version_kind"] == "release" def test_compute_kpis_skips_missing_records(self): from projects.caliper.engine.model import UnifiedResultRecord @@ -167,11 +333,17 @@ def test_get_plugin_returns_instance(self): plugin = get_plugin() assert isinstance(plugin, MCPGatewayPlugin) + def test_analysis_config_present(self): + from projects.mcp_gateway.postprocess.mcp_gateway import plugin as plugin_mod + + assert plugin_mod.analysis_config.comparison_keys == ["mcp_gateway_version"] + assert plugin_mod.analysis_config.max_relative_regression == 0.10 + def test_plugin_parse_and_kpis(self, tmp_path: Path): node = _make_test_node(tmp_path, "run-a", SAMPLE_STATS_CSV, TEST_LABELS) plugin = get_plugin() - parse_result = plugin.parse(tmp_path, [node]) + parse_result = plugin.parse([node]) assert len(parse_result.records) == 1 model = UnifiedRunModel( @@ -194,3 +366,12 @@ def test_visualize_returns_empty(self, tmp_path: Path): ) result = plugin.visualize(model, tmp_path, None, None, None) assert result == [] + + +class TestQueryCatalog: + def test_http_4xx_rate_is_loadable(self): + specs = load_queries(namespaces=["mcp-system", "gateway-system"], keys=["http_4xx_rate"]) + assert len(specs) == 1 + assert specs[0].key == "http_4xx_rate" + assert "4.." in specs[0].promql + assert "mcp-system|gateway-system" in specs[0].promql diff --git a/projects/mcp_gateway/toolbox/cleanup_platform/main.py b/projects/mcp_gateway/toolbox/cleanup_platform/main.py index f28860a0a..e69ba6487 100644 --- a/projects/mcp_gateway/toolbox/cleanup_platform/main.py +++ b/projects/mcp_gateway/toolbox/cleanup_platform/main.py @@ -26,12 +26,14 @@ def run( *, platform_config: dict[str, Any], + scheduling_node_selector: dict[str, str] | None = None, ) -> int: """ Remove the full MCP Gateway platform stack in reverse order. Args: platform_config: Platform configuration dict from infrastructure.yaml + scheduling_node_selector: Labels to remove from worker nodes after cleanup """ execute_tasks(locals()) return 0 @@ -49,6 +51,7 @@ def resolve_config(args, ctx): ctx.gateway_namespace = config.get("gateway_namespace", "gateway-system") ctx.ctrl = config.get("mcp_gateway_controller", {}) ctx.steps = config.get("steps", []) + ctx.node_selector = args.scheduling_node_selector or {} return f"Cleanup config resolved: {len(ctx.steps)} steps" @@ -193,6 +196,23 @@ def delete_namespaces(args, ctx): return f"Deleted namespaces: {', '.join(to_delete)}" +@always +@task +def unlabel_worker_nodes(args, ctx): + """Remove scheduling node_selector labels from worker nodes.""" + selector = ctx.node_selector + if not selector: + return "No node_selector configured, skipping" + + label_sel = ",".join(f"{k}={v}" for k, v in selector.items()) + result = oc("get", "nodes", "-l", label_sel, "-o", "name", check=False, log_stdout=False) + nodes = [line.rsplit("/", 1)[-1] for line in result.stdout.splitlines() if line.strip()] + for node in nodes: + oc("label", "node", node, *[f"{k}-" for k in selector], check=False) + logger.info("Removed scheduling labels from node %s", node) + return f"Removed scheduling labels from {len(nodes)} node(s)" + + @always @task def capture_cleanup_state(args, ctx): diff --git a/projects/mcp_gateway/toolbox/install_platform/main.py b/projects/mcp_gateway/toolbox/install_platform/main.py index 40179036a..97ff43583 100644 --- a/projects/mcp_gateway/toolbox/install_platform/main.py +++ b/projects/mcp_gateway/toolbox/install_platform/main.py @@ -38,12 +38,14 @@ def run( *, platform_config: dict[str, Any], + scheduling_node_selector: dict[str, str] | None = None, ) -> int: """ Install the full MCP Gateway platform stack. Args: platform_config: Platform configuration dict from infrastructure.yaml + scheduling_node_selector: Labels to apply to worker nodes before install """ execute_tasks(locals()) return 0 @@ -75,9 +77,68 @@ def validate_config(args, ctx): logger.info("MCP Gateway namespace: %s", ctx.mcp_gateway_namespace) logger.info("Gateway namespace: %s", ctx.gateway_namespace) + ctx.node_selector = args.scheduling_node_selector or {} + return f"Config validated: {len(ctx.steps)} steps, kustomize_base={ctx.kustomize_base}" +@task +def label_worker_nodes(args, ctx): + """Select the worker node with the most allocatable resources and label it.""" + selector = ctx.node_selector + if not selector: + return "No node_selector configured, skipping" + + import json + + result = oc( + "get", + "nodes", + "-l", + "node-role.kubernetes.io/worker", + "-o", + "json", + check=False, + log_stdout=False, + ) + if result.returncode != 0 or not result.stdout.strip(): + return "No worker nodes found" + + nodes_data = json.loads(result.stdout) + best_node = None + best_score = -1 + + for node in nodes_data.get("items", []): + conditions = node.get("status", {}).get("conditions", []) + ready = any(c.get("type") == "Ready" and c.get("status") == "True" for c in conditions) + if not ready: + continue + + alloc = node.get("status", {}).get("allocatable", {}) + cpu_str = alloc.get("cpu", "0") + cpu_milli = _parse_cpu_to_milli(cpu_str) + mem_str = alloc.get("memory", "0") + mem_bytes = _parse_mem_to_bytes(mem_str) + + score = cpu_milli * 1_000_000 + mem_bytes + if score > best_score: + best_score = score + best_node = node["metadata"]["name"] + + if not best_node: + return "No Ready worker nodes found" + + oc( + "label", + "node", + best_node, + "--overwrite", + *[f"{k}={v}" for k, v in selector.items()], + log_stdout=False, + ) + return f"Labeled strongest worker node with {selector}" + + @task def install_service_mesh_operator(args, ctx): """Apply Service Mesh operator via kustomize and wait for CRDs""" @@ -547,6 +608,31 @@ def _version_gte(version: str, minimum: str) -> bool: return False +def _parse_cpu_to_milli(cpu: str) -> int: + """Convert a Kubernetes CPU string to millicores.""" + if cpu.endswith("m"): + return int(cpu[:-1]) + return int(float(cpu) * 1000) + + +def _parse_mem_to_bytes(mem: str) -> int: + """Convert a Kubernetes memory string to bytes.""" + suffixes = { + "Ki": 1024, + "Mi": 1024**2, + "Gi": 1024**3, + "Ti": 1024**4, + "K": 1000, + "M": 1000**2, + "G": 1000**3, + "T": 1000**4, + } + for suffix, multiplier in suffixes.items(): + if mem.endswith(suffix): + return int(mem[: -len(suffix)]) * multiplier + return int(mem) + + def _version_spec(version: str) -> dict[str, Any]: """Return version-specific resource parameters.