ako_ingress_diagnose

Instructions

Implementation Reference

• vmware_avi/ops/ako_ingress.py:104-183 (handler)

  The core handler function `diagnose_ingress()` that executes the ingress diagnosis logic: fetches the Ingress resource, checks annotations, IngressClass, TLS secrets, and backend service endpoints, then prints issues and suggestions.

  def diagnose_ingress(
      name: str, namespace: str = "default", context: str | None = None
  ) -> None:
      """Deep diagnosis of a specific Ingress."""
      cfg = load_config()
      k8s = K8sConnectionManager(cfg)
  
      from kubernetes.client import NetworkingV1Api
  
      net_v1 = NetworkingV1Api(k8s.get_client(context))
  
      try:
          ing = net_v1.read_namespaced_ingress(name, namespace)
      except Exception:
          console.print(f"[red]Ingress '{name}' not found in namespace '{namespace}'.[/red]")
          raise SystemExit(1)
  
      console.print(f"\n[bold]Diagnosing Ingress: {namespace}/{name}[/bold]\n")
  
      annotations = ing.metadata.annotations or {}
      console.print("[bold]Annotations:[/bold]")
      for k, v in annotations.items():
          console.print(f"  {k}: {v}")
  
      issues: list[str] = []
      suggestions: list[str] = []
  
      # Check IngressClass
      ingress_class = ing.spec.ingress_class_name or annotations.get(
          "kubernetes.io/ingress.class", ""
      )
      if not ingress_class:
          issues.append("No IngressClass specified")
          suggestions.append("Add spec.ingressClassName: 'avi-lb' or annotation kubernetes.io/ingress.class: 'avi'")
      elif ingress_class not in ("avi", "avi-lb"):
          issues.append(f"IngressClass '{ingress_class}' is not AKO")
          suggestions.append(f"Change to 'avi' or 'avi-lb'")
  
      # Check TLS secrets
      if ing.spec.tls:
          for tls in ing.spec.tls:
              if tls.secret_name:
                  try:
                      k8s.core_v1(context).read_namespaced_secret(tls.secret_name, namespace)
                  except Exception as exc:
                      err_msg = str(exc)
                      if "404" in err_msg or "Not Found" in err_msg:
                          issues.append(f"TLS secret '{tls.secret_name}' missing")
                          suggestions.append(f"Create secret: kubectl create secret tls {tls.secret_name} ...")
                      else:
                          issues.append(f"TLS secret '{tls.secret_name}' check failed: {err_msg[:100]}")
  
      # Check backend services exist
      if ing.spec.rules:
          for rule in ing.spec.rules:
              if rule.http and rule.http.paths:
                  for path in rule.http.paths:
                      if not path.backend or not path.backend.service or not path.backend.service.name:
                          issues.append("Path has no backend service configured")
                          continue
                      svc_name = path.backend.service.name
                      try:
                          k8s.core_v1(context).read_namespaced_service(svc_name, namespace)
                      except Exception:
                          issues.append(f"Backend service '{svc_name}' not found")
                          suggestions.append(f"Verify service '{svc_name}' exists in namespace '{namespace}'")
  
      if issues:
          console.print(f"\n[red]Issues Found ({len(issues)}):[/red]")
          for i, issue in enumerate(issues, 1):
              console.print(f"  {i}. {issue}")
          console.print(f"\n[yellow]Suggestions:[/yellow]")
          for i, sug in enumerate(suggestions, 1):
              console.print(f"  {i}. {sug}")
      else:
          console.print("\n[green]No issues found with Ingress configuration.[/green]")
          console.print("If VS is still not created, check AKO logs and sync status:")
          console.print("  vmware-avi ako logs --since 5m")
          console.print("  vmware-avi ako sync status")
      console.print()

• mcp_server/server.py:385-398 (schema)

  MCP tool definition with input schema: parameters `name` (str, required), `namespace` (str, default 'default'), `context` (str, optional). Decorated with @mcp.tool and @vmware_tool(risk_level='low').

  @mcp.tool(annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True})
  @vmware_tool(risk_level="low")
  def ako_ingress_diagnose(name: str, namespace: str = "default", context: str | None = None) -> str:
      """[READ] Diagnose why a specific Ingress has no corresponding Virtual Service.
  
      Checks annotations, TLS config, service endpoints, and AKO logs for errors.
  
      Args:
          name: Ingress resource name.
          namespace: K8s namespace (default 'default').
          context: K8s context name (optional).
      """
      from vmware_avi.ops.ako_ingress import diagnose_ingress
      return _capture_output(diagnose_ingress, name, namespace, context)

• mcp_server/server.py:385-398 (registration)

  Tool registered via @mcp.tool decorator on the `ako_ingress_diagnose` function in the MCP server.

  @mcp.tool(annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True})
  @vmware_tool(risk_level="low")
  def ako_ingress_diagnose(name: str, namespace: str = "default", context: str | None = None) -> str:
      """[READ] Diagnose why a specific Ingress has no corresponding Virtual Service.
  
      Checks annotations, TLS config, service endpoints, and AKO logs for errors.
  
      Args:
          name: Ingress resource name.
          namespace: K8s namespace (default 'default').
          context: K8s context name (optional).
      """
      from vmware_avi.ops.ako_ingress import diagnose_ingress
      return _capture_output(diagnose_ingress, name, namespace, context)

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

  The `_capture_output()` helper captures Rich console output from the handler function and returns it as 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()