Coverage for cli/commands/monitoring_cmd.py: 96.34%
173 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""GCO cluster observability command group.
3Provides the ``gco monitoring`` sub-commands:
5* ``enable`` / ``disable`` / ``status`` — flip the
6 ``cluster_observability.enabled`` toggle in ``cdk.json`` (on by default).
7* ``open`` — ``kubectl port-forward`` to Grafana / Prometheus / Alertmanager
8 over the PRIVATE EKS API endpoint. Because the endpoint is private by
9 default, ``open`` detects that posture and can tunnel to the API through an
10 SSM-managed instance (``--via-ssm``) so the forward works from a laptop.
11* ``users`` — manage Grafana users via the admin HTTP API (see
12 :mod:`cli.monitoring_user_mgmt`; wired in a follow-up command module).
14The Click wiring mirrors ``analytics_cmd.py``. In-cluster access always goes
15through the private API endpoint — there is no public Grafana ingress.
16"""
18from __future__ import annotations
20import subprocess
21import sys
22from typing import Any
24import click
25import requests
27from ..config import GCOConfig
28from ..output import get_output_formatter
30pass_config = click.make_pass_decorator(GCOConfig, ensure=True)
32# kube-prometheus-stack + OpenCost service names (release name = chart key)
33# and their service ports, plus a sensible default local port for each.
34# OpenCost's UI container listens on 9090 in-cluster; its local default is
35# 9091 so `gco monitoring open --service opencost` can run alongside a
36# Prometheus forward on 9090 without a port clash.
37_SERVICES: dict[str, dict[str, Any]] = {
38 "grafana": {
39 "target": "svc/kube-prometheus-stack-grafana",
40 "remote_port": 80,
41 "default_local_port": 3000,
42 },
43 "prometheus": {
44 "target": "svc/kube-prometheus-stack-prometheus",
45 "remote_port": 9090,
46 "default_local_port": 9090,
47 },
48 "alertmanager": {
49 "target": "svc/kube-prometheus-stack-alertmanager",
50 "remote_port": 9093,
51 "default_local_port": 9093,
52 },
53 "opencost": {
54 "target": "svc/opencost",
55 "remote_port": 9090,
56 "default_local_port": 9091,
57 },
58 "opencost-api": {
59 "target": "svc/opencost",
60 "remote_port": 9003,
61 "default_local_port": 9003,
62 },
63}
65_MONITORING_NAMESPACE = "monitoring"
66_GRAFANA_SECRET = "kube-prometheus-stack-grafana"
68# Default self-terminate backstop for an `--via-ssm auto` bastion.
69# Mirrors cli.ephemeral_bastion.DEFAULT_TTL_MINUTES (kept literal to avoid an
70# import at module load; the ephemeral_bastion module validates the range).
71_DEFAULT_BASTION_TTL_MINUTES = 120
74@click.group()
75@pass_config
76def monitoring(config: Any) -> None:
77 """Manage in-cluster observability (Prometheus + Grafana + Alertmanager)."""
80# ---------------------------------------------------------------------------
81# Toggle commands — status / enable / disable
82# ---------------------------------------------------------------------------
85@monitoring.command("status")
86@pass_config
87def monitoring_status(config: Any) -> None:
88 """Show the current cluster observability toggle state from cdk.json."""
89 from ..stacks import get_cluster_observability_config
91 formatter = get_output_formatter(config)
92 try:
93 current = get_cluster_observability_config()
94 formatter.print_info("Cluster observability config:")
95 formatter.print(current)
96 except Exception as exc: # noqa: BLE001 — surface every loader error
97 formatter.print_error(f"Failed to read cluster observability config: {exc}")
98 sys.exit(1)
101@monitoring.command("enable")
102@click.option("--yes", "-y", is_flag=True, help="Skip confirmation.")
103@pass_config
104def monitoring_enable(config: Any, yes: bool) -> None:
105 """Enable cluster observability in cdk.json (installs kube-prometheus-stack).
107 Observability is on by default; use this to re-enable after a
108 ``gco monitoring disable``. Prints the follow-up deploy command — does not
109 deploy automatically.
110 """
111 from ..stacks import update_cluster_observability_config
113 formatter = get_output_formatter(config)
114 if not yes:
115 formatter.print_info(
116 "Cluster observability (kube-prometheus-stack) will be enabled on every region."
117 )
118 click.confirm("\nEnable cluster observability?", abort=True)
120 try:
121 update_cluster_observability_config({"enabled": True})
122 formatter.print_success("Cluster observability enabled in cdk.json")
123 formatter.print_info(
124 f"Run `gco stacks deploy {config.project_name}-<region>` (or deploy-all) to apply"
125 )
126 except Exception as exc: # noqa: BLE001 — user-facing error from file I/O
127 formatter.print_error(f"Failed to enable cluster observability: {exc}")
128 sys.exit(1)
131@monitoring.command("disable")
132@click.option("--yes", "-y", is_flag=True, help="Skip confirmation.")
133@pass_config
134def monitoring_disable(config: Any, yes: bool) -> None:
135 """Disable cluster observability in cdk.json.
137 Flips ``cluster_observability.enabled`` to ``false``; the grafana /
138 prometheus / alertmanager sub-blocks (sizes, retention, rotation schedule)
139 are left untouched so preferences survive a disable/enable cycle. The
140 in-cluster stack is removed on the next deploy.
141 """
142 from ..stacks import update_cluster_observability_config
144 formatter = get_output_formatter(config)
145 if not yes:
146 formatter.print_warning("This will disable cluster observability.")
147 formatter.print_warning(
148 "Prometheus/Grafana/Alertmanager and their EBS volumes are removed on next deploy."
149 )
150 click.confirm("Are you sure?", abort=True)
152 try:
153 update_cluster_observability_config({"enabled": False})
154 formatter.print_success("Cluster observability disabled in cdk.json")
155 formatter.print_info(
156 f"Run `gco stacks deploy {config.project_name}-<region>` (or deploy-all) to apply"
157 )
158 except Exception as exc: # noqa: BLE001 — user-facing error from file I/O
159 formatter.print_error(f"Failed to disable cluster observability: {exc}")
160 sys.exit(1)
163# ---------------------------------------------------------------------------
164# open — port-forward (private endpoint aware)
165# ---------------------------------------------------------------------------
168@monitoring.command("open")
169@click.option(
170 "--service",
171 type=click.Choice(sorted(_SERVICES)),
172 default="grafana",
173 show_default=True,
174 help="Which component to port-forward.",
175)
176@click.option("--region", help="Cluster region (defaults to the first cdk.json regional entry).")
177@click.option("--local-port", type=int, help="Local port to bind (defaults per-service).")
178@click.option(
179 "--via-ssm",
180 "via_ssm",
181 metavar="INSTANCE_ID|auto",
182 help=(
183 "Tunnel to the private API endpoint through an SSM-managed instance. "
184 "Pass an instance id to use an existing one, or 'auto' to provision a "
185 "self-terminating ephemeral bastion and tear it down when the forward stops."
186 ),
187)
188@click.option(
189 "--bastion-ttl-minutes",
190 type=int,
191 default=_DEFAULT_BASTION_TTL_MINUTES,
192 show_default=True,
193 help="Self-terminate backstop (minutes) for an `--via-ssm auto` bastion.",
194)
195@click.option(
196 "--yes",
197 "-y",
198 "assume_yes",
199 is_flag=True,
200 help="Skip the confirmation prompt when provisioning an `--via-ssm auto` bastion.",
201)
202@pass_config
203def monitoring_open(
204 config: Any,
205 service: str,
206 region: str | None,
207 local_port: int | None,
208 via_ssm: str | None,
209 bastion_ttl_minutes: int,
210 assume_yes: bool,
211) -> None:
212 """Port-forward to a monitoring component over the private EKS endpoint.
214 Runs in the foreground; press Ctrl-C to stop. On a private-endpoint cluster
215 (the default) pass ``--via-ssm <instance-id>`` to tunnel through an existing
216 SSM-managed instance, or ``--via-ssm auto`` to have the CLI provision a
217 minimal, self-terminating ephemeral bastion for the session and tear it down
218 on exit.
219 """
220 from ..cluster_tunnel import open_api_server_tunnel, resolve_region
221 from ..kubectl_helpers import build_port_forward_command, update_kubeconfig
223 formatter = get_output_formatter(config)
224 svc = _SERVICES[service]
225 target_region = resolve_region(config, region)
226 cluster = f"{config.project_name}-{target_region}"
227 bind_port = local_port or svc["default_local_port"]
229 try:
230 update_kubeconfig(cluster, target_region)
231 except (RuntimeError, ValueError) as exc:
232 formatter.print_error(str(exc))
233 sys.exit(1)
235 # The shared context manager detects the endpoint posture, optionally
236 # provisions an ephemeral bastion (`--via-ssm auto`), opens the SSM tunnel,
237 # and guarantees teardown of both the tunnel and any bastion on exit.
238 try:
239 with open_api_server_tunnel(
240 formatter,
241 cluster=cluster,
242 region=target_region,
243 via_ssm=via_ssm,
244 bastion_ttl_minutes=bastion_ttl_minutes,
245 assume_yes=assume_yes,
246 ) as session:
247 cmd = build_port_forward_command(
248 _MONITORING_NAMESPACE,
249 svc["target"],
250 bind_port,
251 svc["remote_port"],
252 server=session.server,
253 tls_server_name=session.tls_server_name,
254 )
255 url = f"http://localhost:{bind_port}"
256 formatter.print_success(f"Forwarding {service} → {url} (Ctrl-C to stop)")
257 if service == "grafana":
258 formatter.print_info(
259 "Log in with the Grafana admin credential from the "
260 f"{_GRAFANA_SECRET} Secret (monitoring namespace)."
261 )
262 try:
263 _exec_port_forward(cmd)
264 except KeyboardInterrupt: # pragma: no cover - interactive Ctrl-C
265 return
266 except (RuntimeError, ValueError) as exc:
267 formatter.print_error(str(exc))
268 sys.exit(1)
271def _exec_port_forward(cmd: list[str]) -> None:
272 """Run the (validated) kubectl port-forward argv in the foreground."""
273 subprocess.run(
274 cmd, check=False
275 ) # nosemgrep: dangerous-subprocess-use-audit - argv built by build_port_forward_command; list form, no shell=True
278# ---------------------------------------------------------------------------
279# users subgroup — Grafana native users over the admin HTTP API
280# ---------------------------------------------------------------------------
283def _grafana_conn_options(func: Any) -> Any:
284 """Shared --grafana-url / --admin-user / --admin-password options."""
285 from ..monitoring_user_mgmt import DEFAULT_GRAFANA_URL
287 func = click.option(
288 "--grafana-url",
289 default=DEFAULT_GRAFANA_URL,
290 show_default=True,
291 help="Grafana base URL (reachable via `gco monitoring open`).",
292 )(func)
293 func = click.option(
294 "--admin-user",
295 help="Grafana admin username (default: read from the Grafana Secret).",
296 )(func)
297 func = click.option(
298 "--admin-password",
299 envvar="GCO_GRAFANA_ADMIN_PASSWORD",
300 help=(
301 "Grafana admin password (also $GCO_GRAFANA_ADMIN_PASSWORD; "
302 "default: read from the Grafana Secret)."
303 ),
304 )(func)
305 return func
308def _resolve_grafana_auth(admin_user: str | None, admin_password: str | None) -> tuple[str, str]:
309 """Return ``(user, password)`` from the flags, else from the Grafana Secret."""
310 if admin_password:
311 return (admin_user or "admin", admin_password)
312 from ..monitoring_user_mgmt import read_grafana_admin_credentials
314 return read_grafana_admin_credentials()
317@monitoring.group("users")
318@pass_config
319def users_cmd(config: Any) -> None:
320 """Manage Grafana users via the admin API (over `gco monitoring open`)."""
323@users_cmd.command("add")
324@click.option("--username", required=True, help="Grafana login for the new user.")
325@click.option("--email", help="Email address for the new user (optional).")
326@click.option("--password", help="Set this password. Mutually exclusive with --generate-password.")
327@click.option(
328 "--generate-password",
329 is_flag=True,
330 help="Generate a strong random password and print it once.",
331)
332@_grafana_conn_options
333@pass_config
334def users_add(
335 config: Any,
336 username: str,
337 email: str | None,
338 password: str | None,
339 generate_password: bool,
340 grafana_url: str,
341 admin_user: str | None,
342 admin_password: str | None,
343) -> None:
344 """Create a Grafana user via the admin HTTP API."""
345 from ..monitoring_user_mgmt import create_user
346 from ..monitoring_user_mgmt import generate_password as _gen
348 formatter = get_output_formatter(config)
349 if password and generate_password:
350 formatter.print_error("--password and --generate-password are mutually exclusive")
351 sys.exit(1)
352 if not password and not generate_password:
353 formatter.print_error("Pass --password or --generate-password")
354 sys.exit(1)
356 final_password = password or _gen()
357 try:
358 auth = _resolve_grafana_auth(admin_user, admin_password)
359 user_id = create_user(
360 grafana_url, auth, login=username, password=final_password, email=email
361 )
362 except (requests.RequestException, RuntimeError, ValueError) as exc:
363 formatter.print_error(f"Failed to create Grafana user {username!r}: {exc}")
364 sys.exit(1)
366 formatter.print_success(f"Created Grafana user {username!r} (id={user_id})")
367 if generate_password:
368 formatter.print_info(f"Generated password (printed exactly once): {final_password}")
371@users_cmd.command("list")
372@click.option("--as-json", "as_json", is_flag=True, help="Emit JSON instead of a table.")
373@_grafana_conn_options
374@pass_config
375def users_list(
376 config: Any,
377 as_json: bool,
378 grafana_url: str,
379 admin_user: str | None,
380 admin_password: str | None,
381) -> None:
382 """List Grafana organisation users."""
383 from ..monitoring_user_mgmt import list_users
385 formatter = get_output_formatter(config)
386 try:
387 auth = _resolve_grafana_auth(admin_user, admin_password)
388 users = list_users(grafana_url, auth)
389 except (requests.RequestException, RuntimeError, ValueError) as exc:
390 formatter.print_error(f"Failed to list Grafana users: {exc}")
391 sys.exit(1)
393 if as_json: 393 ↛ 398line 393 didn't jump to line 398 because the condition on line 393 was always true
394 import json
396 print(json.dumps(users, indent=2))
397 return
398 formatter.print(users)
401@users_cmd.command("remove")
402@click.option("--username", required=True, help="Grafana login/email to remove.")
403@click.option("--yes", is_flag=True, help="Skip the confirmation prompt.")
404@_grafana_conn_options
405@pass_config
406def users_remove(
407 config: Any,
408 username: str,
409 yes: bool,
410 grafana_url: str,
411 admin_user: str | None,
412 admin_password: str | None,
413) -> None:
414 """Delete a Grafana user by login or email."""
415 from ..monitoring_user_mgmt import delete_user, lookup_user_id
417 formatter = get_output_formatter(config)
418 if not yes: 418 ↛ 419line 418 didn't jump to line 419 because the condition on line 418 was never true
419 click.confirm(f"Delete Grafana user '{username}'?", abort=True)
421 try:
422 auth = _resolve_grafana_auth(admin_user, admin_password)
423 user_id = lookup_user_id(grafana_url, auth, username)
424 delete_user(grafana_url, auth, user_id)
425 except (requests.RequestException, RuntimeError, ValueError) as exc:
426 formatter.print_error(f"Failed to remove Grafana user {username!r}: {exc}")
427 sys.exit(1)
429 formatter.print_success(f"Deleted Grafana user {username!r}")