vs_error_logs

Instructions

Implementation Reference

• vmware_avi/ops/analytics.py:127-172 (handler)

  Core handler function that queries the AVI analytics/logs endpoint for error logs (non-200 responses) for a given Virtual Service, filtering by duration.

  def show_error_logs(vs_name: str, since: str = "1h") -> None:
      """Show request error logs for a VS.
  
      ``since`` accepts an integer number of seconds or a suffixed shorthand
      such as ``'1h'``, ``'30m'``, ``'24h'``, ``'7d'``.
      """
      cfg = load_config()
      mgr = AviConnectionManager(cfg)
      session = mgr.connect()
  
      vs = session.get_object_by_name("virtualservice", vs_name)
      if not vs:
          console.print(f"[red]Virtual Service '{vs_name}' not found.[/red]")
          raise SystemExit(1)
  
      try:
          duration_seconds = _parse_duration_seconds(since)
      except ValueError as e:
          console.print(f"[red]Invalid duration: {e}[/red]")
          raise SystemExit(1) from None
  
      uuid = vs["uuid"]
      # AVI 22.x requires the VS UUID as an explicit ``virtualservice`` URL
      # parameter on /analytics/logs; passing it only inside ``filter`` as
      # ``co(vs_uuid,<uuid>)`` yields HTTP 400 "VirtualService ID required".
      resp = session.get("analytics/logs", params={
          "type": "1",
          "virtualservice": uuid,
          "filter": "ne(response_code,200)",
          "page_size": "50",
          "duration": str(duration_seconds),
      })
      logs = (resp.json() if hasattr(resp, "json") else resp).get("results", [])
  
      console.print(f"\n[bold]Error Logs: {vs_name} (last {since} = {duration_seconds}s)[/bold]")
      for log in logs:
          ts = log.get("report_timestamp", "")
          code = log.get("response_code", "")
          uri = log.get("uri_path", "")[:80]
          client = log.get("client_ip", "")
          console.print(f"  [{ts}] {code} {uri} from {client}")
  
      if not logs:
          console.print("  [green]No errors found.[/green]")
      console.print()

• mcp_server/server.py:218-230 (registration)

  Registration of the vs_error_logs tool as an MCP tool via @mcp.tool decorator with read-only annotations. Delegates to show_error_logs in analytics module.

  @mcp.tool(annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True})
  @vmware_tool(risk_level="low")
  def vs_error_logs(vs_name: str, since: str = "1h") -> str:
      """[READ] Show recent request error logs for a Virtual Service — HTTP status codes, client IPs, URIs, and response times.
  
      Use to diagnose 5xx errors or latency spikes.
  
      Args:
          vs_name: Virtual Service name.
          since: Time window, e.g. '1h', '30m', '2d' (default '1h').
      """
      from vmware_avi.ops.analytics import show_error_logs
      return _capture_output(show_error_logs, vs_name, since)

• mcp_server/server.py:24-48 (helper)

  Output capture helper used by vs_error_logs to redirect Rich console output from show_error_logs into a string for MCP response.

  def _capture_output(func, *args, **kwargs) -> str:
      """Run a function and capture its Rich console output as plain text."""
      import importlib  # noqa: F401 — used via sys.modules lookup
      import sys
  
      buf = StringIO()
      from rich.console import Console
      capture_console = Console(file=buf, force_terminal=False, width=120)
  
      mod_name = func.__module__
      mod = sys.modules.get(mod_name)
      original_console = getattr(mod, "console", None) if mod else None
  
      if mod and original_console is not None:
          mod.console = capture_console
  
      try:
          func(*args, **kwargs)
      except SystemExit:
          pass
      finally:
          if mod and original_console is not None:
              mod.console = original_console
  
      return buf.getvalue()

• vmware_avi/ops/analytics.py:16-39 (helper)

  Duration parsing helper used by show_error_logs to convert user-friendly time strings (e.g. '1h', '30m') to seconds for the AVI API.

  def _parse_duration_seconds(value: str | int) -> int:
      """Parse a duration string (e.g. '1h', '30m', '24h', '7d') to seconds.
  
      The AVI analytics API requires duration as a non-negative integer number
      of seconds. Accept shorthand suffixes so users can pass '1h' naturally.
      """
      if isinstance(value, int):
          if value < 0:
              raise ValueError("duration must be non-negative")
          return value
  
      s = str(value).strip().lower()
      if not s:
          raise ValueError("duration must be non-empty")
  
      m = re.fullmatch(r"(\d+)\s*([smhd]?)", s)
      if not m:
          raise ValueError(
              f"invalid duration {value!r}: expected integer seconds or suffix "
              "s/m/h/d (e.g. '1h', '30m', '3600')"
          )
      n = int(m.group(1))
      unit = m.group(2) or "s"
      return n * {"s": 1, "m": 60, "h": 3600, "d": 86400}[unit]

• tests/test_mcp_tools.py:15-47 (schema)

  Test expectation list confirming vs_error_logs is one of the 29 expected tool names.

  EXPECTED_TOOL_NAMES = {
      # Traditional mode
      "vs_list",
      "vs_status",
      "vs_toggle",
      "pool_members",
      "pool_member_enable",
      "pool_member_disable",
      "ssl_list",
      "ssl_expiry_check",
      "vs_analytics",
      "vs_error_logs",
      "se_list",
      "se_health",
      # AKO mode
      "ako_status",
      "ako_logs",
      "ako_restart",
      "ako_version",
      "ako_config_show",
      "ako_config_diff",
      "ako_config_upgrade",
      "ako_ingress_check",
      "ako_ingress_map",
      "ako_ingress_diagnose",
      "ako_ingress_fix_suggest",
      "ako_sync_status",
      "ako_sync_diff",
      "ako_sync_force",
      "ako_clusters",
      "ako_cluster_overview",
      "ako_amko_status",
  }