Coverage for gco_mcp/tools/monitoring.py: 100.00%

1"""Cluster observability MCP tools (thin wrappers over `gco monitoring`). 

2 

3Mirrors :mod:`tools.analytics`: read-only tools are tagged ``safe``, cdk.json 

4toggles and user creation are ``low-risk``, and user deletion is ``destructive`` 

5and gated behind ``GCO_ENABLE_DESTRUCTIVE_OPERATIONS``. There is intentionally no 

6``monitoring open`` tool — that is an interactive, long-running port-forward, not 

7a request/response operation. 

8""" 

9 

10import asyncio 

11 

12import cli_runner 

13from audit import audit_logged 

14from server import mcp 

15 

16 

17@mcp.tool(tags={"safe", "monitoring"}) 

18@audit_logged 

19async def monitoring_status() -> str: 

20 """`gco monitoring status` — show the cluster observability toggle + config.""" 

21 return await asyncio.to_thread(cli_runner._run_cli, "monitoring", "status") 

22 

23 

24@mcp.tool(tags={"safe", "monitoring"}) 

25@audit_logged 

26async def monitoring_users_list() -> str: 

27 """`gco monitoring users list` — list Grafana users via the admin API. 

28 

29 Reaches Grafana over a `gco monitoring open` port-forward; the admin 

30 credential is read from the chart-generated Secret (or $GCO_GRAFANA_ADMIN_PASSWORD). 

31 """ 

32 return await asyncio.to_thread(cli_runner._run_cli, "monitoring", "users", "list") 

33 

34 

35# ============================================================================= 

36# Mutating tools (low-risk) 

37# ============================================================================= 

38 

39 

40@mcp.tool(tags={"low-risk", "monitoring"}) 

41@audit_logged 

42async def enable_monitoring() -> str: 

43 """`gco monitoring enable` — flip cluster observability on in cdk.json. 

44 

45 Note: this only edits the cdk.json toggle. It does not take effect until 

46 ``gco stacks deploy`` (or deploy-all) reinstalls kube-prometheus-stack. 

47 """ 

48 return await asyncio.to_thread(cli_runner._run_cli, "monitoring", "enable", "-y") 

49 

50 

51@mcp.tool(tags={"low-risk", "monitoring"}) 

52@audit_logged 

53async def disable_monitoring() -> str: 

54 """`gco monitoring disable` — flip cluster observability off in cdk.json. 

55 

56 Note: this only edits the cdk.json toggle. The in-cluster stack is removed 

57 on the next ``gco stacks deploy``. 

58 """ 

59 return await asyncio.to_thread(cli_runner._run_cli, "monitoring", "disable", "-y") 

60 

61 

62@mcp.tool(tags={"low-risk", "monitoring"}) 

63@audit_logged 

64async def monitoring_user_add(username: str, password: str, email: str | None = None) -> str: 

65 """`gco monitoring users add` — create a Grafana user via the admin API. 

66 

67 Args: 

68 username: Grafana login for the new user. 

69 password: Password to set for the new user. 

70 email: Optional email address for the new user. 

71 """ 

72 args = ["monitoring", "users", "add", "--username", username, "--password", password] 

73 if email: 

74 args += ["--email", email] 

75 return await asyncio.to_thread(cli_runner._run_cli, *args) 

76 

77 

78# ============================================================================= 

79# Destructive tools — gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS 

80# ============================================================================= 

81 

82 

83import contextlib # noqa: E402 

84 

85from feature_flags import FLAG_DESTRUCTIVE_OPERATIONS, is_enabled # noqa: E402 

86 

87 

88async def _ctx_warning(message: str) -> None: 

89 """Emit ``ctx.warning(...)`` from inside a tool body, no-op when no Context.""" 

90 try: 

91 from fastmcp.server.dependencies import get_context 

92 

93 ctx = get_context() 

94 except Exception: 

95 return 

96 with contextlib.suppress(Exception): 

97 await ctx.warning(message) 

98 

99 

100if is_enabled(FLAG_DESTRUCTIVE_OPERATIONS): 

101 

102 @mcp.tool(tags={"destructive", "monitoring"}) 

103 @audit_logged 

104 async def monitoring_user_remove(username: str) -> str: 

105 """[gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS] destructive. 

106 

107 `gco monitoring users remove` — delete a Grafana user via the admin 

108 API. Cannot be undone — the user and their Grafana-owned resources are 

109 permanently removed. 

110 

111 Args: 

112 username: Grafana login or email to remove. 

113 """ 

114 await _ctx_warning(f"Removing Grafana user {username!r} — this cannot be undone.") 

115 return await asyncio.to_thread( 

116 cli_runner._run_cli, 

117 "monitoring", 

118 "users", 

119 "remove", 

120 "--username", 

121 username, 

122 "--yes", 

123 )