Coverage for gco_mcp/resources/cluster.py: 96.08%

1"""Cluster topology resources (gco://cluster/...) for the GCO MCP server. 

2 

3Aggregates two views of regional cluster state into a single JSON 

4payload an LLM can pin: the Karpenter NodePool inventory (via the 

5``gco nodepools list`` CLI surface) and the list of pods currently 

6in ``Pending`` phase (via ``kubectl get pods``). The combination is 

7the cheapest read that answers "what shape is this cluster in right 

8now and what's stuck waiting for room to schedule". 

9""" 

10 

11from __future__ import annotations 

12 

13import json 

14from typing import Any 

15 

16import cli_runner 

17 

18from resources._eks import eks_context_for_region, is_valid_region 

19 

20_KUBECTL_TIMEOUT_SECONDS = 30 

21 

22 

23def _list_nodepools(region: str) -> dict[str, Any]: 

24 """Run ``gco nodepools list`` and return the parsed payload (or an error stub).""" 

25 raw = cli_runner._run_cli("nodepools", "list", "-r", region) 

26 try: 

27 parsed = json.loads(raw) 

28 except json.JSONDecodeError, ValueError: 

29 return {"error": "failed to parse nodepools output", "raw": raw} 

30 if isinstance(parsed, dict): 

31 return parsed 

32 return {"value": parsed} 

33 

34 

35def _pending_pods(region: str) -> dict[str, Any]: 

36 """Return pending pods from the explicitly selected regional cluster.""" 

37 try: 

38 context_arn = eks_context_for_region(region) 

39 except Exception as exc: # AWS credential/session failures become resource errors 

40 return {"error": "unable to resolve EKS context", "detail": str(exc)[:200]} 

41 try: 

42 result = cli_runner.subprocess.run( # type: ignore[attr-defined] # nosemgrep: dangerous-subprocess-use-audit - shell=False; argv built from validated region and account-qualified ARN 

43 [ 

44 "kubectl", 

45 "get", 

46 "pods", 

47 "--all-namespaces", 

48 "--field-selector", 

49 "status.phase=Pending", 

50 "-o", 

51 "json", 

52 "--context", 

53 context_arn, 

54 ], 

55 capture_output=True, 

56 text=True, 

57 timeout=_KUBECTL_TIMEOUT_SECONDS, 

58 ) 

59 except FileNotFoundError: 

60 return {"error": "kubectl not found"} 

61 except cli_runner.subprocess.TimeoutExpired: # type: ignore[attr-defined] 

62 return {"error": f"kubectl timed out after {_KUBECTL_TIMEOUT_SECONDS}s"} 

63 if result.returncode != 0: 

64 err = (result.stderr or result.stdout or "").strip() 

65 return {"error": err or "kubectl command failed", "exit_code": result.returncode} 

66 try: 

67 parsed = json.loads(result.stdout) 

68 except json.JSONDecodeError, ValueError: 

69 return {"error": "failed to parse kubectl output", "raw": result.stdout} 

70 if isinstance(parsed, dict): 

71 return parsed 

72 return {"value": parsed} 

73 

74 

75def _topology_resource(region: str) -> str: 

76 """Return a structured snapshot of nodepools plus pending pods for ``region``.""" 

77 if not is_valid_region(region): 

78 return json.dumps({"error": "invalid region", "value": region}) 

79 summary = { 

80 "region": region, 

81 "nodepools": _list_nodepools(region), 

82 "pending_pods": _pending_pods(region), 

83 } 

84 return json.dumps(summary, indent=2, default=str) 

85 

86 

87def register(mcp_instance: Any) -> None: 

88 """Register the cluster topology aggregator against the shared MCP server.""" 

89 mcp_instance.resource("gco://cluster/{region}/topology")(_topology_resource)