se_health

Instructions

Implementation Reference

• vmware_avi/ops/se_mgmt.py:40-88 (handler)

  Core handler: check_se_health() queries virtualservice-inventory and serviceengine-inventory, builds SE→VS placement map, prints name/status/VS count for each SE.

  def check_se_health() -> None:
      """Check SE health and resource usage.
  
      VS count is reconstructed by inverting the VS→SE placement map from
      ``/virtualservice-inventory`` (``runtime.vip_summary[].service_engine[]``).
      On 22.x Controllers the per-SE ``runtime`` object does not expose any
      VS list field, so the previous attempts at ``se_vs_list`` / ``vs_ref`` /
      ``virtualservice_refs`` always produced 0 and gave false "idle SE"
      signals.
      """
      cfg = load_config()
      mgr = AviConnectionManager(cfg)
      session = mgr.connect()
  
      # Build SE-UUID → set of VS-UUIDs it hosts, then collapse to counts.
      # De-duping per VS (not per (VS, VIP) pair) ensures a VS with multiple
      # VIPs landing on the same SE still only counts once.
      vs_resp = session.get("virtualservice-inventory")
      se_vs_map: dict[str, set[str]] = {}
      for vs in (vs_resp.json() if hasattr(vs_resp, "json") else vs_resp).get("results", []):
          vs_uuid = vs.get("uuid") or (vs.get("config") or {}).get("uuid", "")
          if not vs_uuid:
              continue
          runtime = vs.get("runtime") or {}
          for vip in runtime.get("vip_summary") or []:
              for se in vip.get("service_engine") or []:
                  se_uuid = se.get("uuid")
                  if se_uuid:
                      se_vs_map.setdefault(se_uuid, set()).add(vs_uuid)
  
      resp = session.get("serviceengine-inventory")
      ses = (resp.json() if hasattr(resp, "json") else resp).get("results", [])
  
      console.print("\n[bold]Service Engine Health[/bold]")
      for se in ses:
          cfg_data = se.get("config", {}) or {}
          runtime = se.get("runtime", {}) or {}
          name = cfg_data.get("name", "N/A")
          uuid = se.get("uuid") or cfg_data.get("uuid", "")
          oper = (runtime.get("oper_status", {}) or {}).get("state", "N/A")
          vs_count = len(se_vs_map.get(uuid, set()))
  
          status_color = "green" if oper == "OPER_UP" else "red"
          console.print(
              f"  [{status_color}]{name}[/{status_color}]: {oper}, "
              f"VS count: {vs_count}"
          )
  
      console.print()

• mcp_server/server.py:241-249 (registration)

  MCP tool registration: se_health() decorated with @mcp.tool and @vmware_tool, calls check_se_health via _capture_output.

  @mcp.tool(annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True})
  @vmware_tool(risk_level="low")
  def se_health() -> str:
      """[READ] Check health of all Service Engines — CPU, memory, disk usage, and connectivity.
  
      Use when VS health degrades to check if the issue is at the SE level.
      """
      from vmware_avi.ops.se_mgmt import check_se_health
      return _capture_output(check_se_health)

• vmware_avi/cli.py:225-230 (registration)

  CLI registration: se_health command under 'se health' subcommand in the Typer CLI.

  @se_app.command("health")
  def se_health() -> None:
      """Check Service Engine health."""
      from vmware_avi.ops.se_mgmt import check_se_health
  
      check_se_health()

• tests/test_mcp_tools.py:28-29 (registration)

  Tool listing test: 'se_health' listed as an expected MCP tool name.

  "se_health",
  # AKO mode

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

  Helper: _capture_output captures Rich console output as a string for MCP tool responses.

  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()