mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 11:03:59 +00:00
318 lines
10 KiB
Python
318 lines
10 KiB
Python
"""Monitor subcommand — read port-forwards.cfg and manage kubectl port-forward processes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from knoe import knoe_conf
|
|
from knoe.config import _expand_path, _collect_cfg_vars, _expand_cfg_value
|
|
from knoe.core.env import (
|
|
_build_required_port_forwards,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Port-forwards.cfg I/O
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _parse_pf_cfg(path: Path) -> list[dict]:
|
|
"""Parse a port-forwards.cfg file and return a list of mapping dicts."""
|
|
mappings: list[dict] = []
|
|
if not path.exists():
|
|
return mappings
|
|
for raw_line in path.read_text().splitlines():
|
|
line = raw_line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
# Format: id: local=... remote=... ns=... svc=... address=...
|
|
if ":" not in line:
|
|
continue
|
|
m_id, _, rest = line.partition(":")
|
|
m_id = m_id.strip()
|
|
parts: dict[str, str] = {"id": m_id}
|
|
for token in rest.split():
|
|
if "=" in token:
|
|
k, v = token.split("=", 1)
|
|
parts[k.strip()] = v.strip()
|
|
mappings.append(parts)
|
|
return mappings
|
|
|
|
|
|
def write_port_forwards_cfg(path: Path, mappings: list[str]) -> None:
|
|
"""Write a port-forwards.cfg from internal mapping strings."""
|
|
lines = [
|
|
"# Port forward configuration for Knoe services (generated).",
|
|
"# Format: id: local=<hostPort> remote=<servicePort> ns=<namespace> svc=<service> address=<bind>",
|
|
"",
|
|
]
|
|
for m in mappings:
|
|
parts: dict[str, str] = {}
|
|
for token in m.split(";"):
|
|
if "=" in token:
|
|
k, v = token.split("=", 1)
|
|
parts[k.strip()] = v.strip()
|
|
m_id = parts.get("id", "")
|
|
if not m_id:
|
|
continue
|
|
local = parts.get("hostPort", "")
|
|
remote = parts.get("servicePort", "")
|
|
ns = parts.get("namespace", "")
|
|
target = parts.get("target", "")
|
|
addr = parts.get("address", "0.0.0.0")
|
|
svc = target[4:] if target.startswith("svc/") else target
|
|
lines.append(
|
|
f"{m_id}: local={local} remote={remote} ns={ns} svc={svc} address={addr}"
|
|
)
|
|
path.write_text("\n".join(lines) + "\n")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Conflict detection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def check_port_conflicts(mappings: list[dict]) -> list[str]:
|
|
"""Return list of conflict descriptions (empty == no conflicts)."""
|
|
seen: dict[str, str] = {}
|
|
conflicts: list[str] = []
|
|
for m in mappings:
|
|
port = m.get("local", "")
|
|
m_id = m.get("id", "?")
|
|
if port in seen:
|
|
conflicts.append(f"Port {port} conflict: {seen[port]} and {m_id}")
|
|
else:
|
|
seen[port] = m_id
|
|
return conflicts
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# kubectl port-forward launcher
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _build_kubectl_cmd(m: dict) -> list[str]:
|
|
"""Build a kubectl port-forward command from a mapping dict."""
|
|
ns = m.get("ns", "default")
|
|
svc = m.get("svc", "")
|
|
local = m.get("local", "")
|
|
remote = m.get("remote", "")
|
|
addr = m.get("address", "0.0.0.0")
|
|
|
|
if not svc.startswith("svc/"):
|
|
svc = f"svc/{svc}"
|
|
|
|
cmd = ["kubectl", "port-forward", "-n", ns, svc, f"{local}:{remote}"]
|
|
if addr and addr != "127.0.0.1":
|
|
cmd.extend(["--address", addr])
|
|
return cmd
|
|
|
|
|
|
def _start_port_forwards(
|
|
mappings: list[dict], verbose: bool = False
|
|
) -> list[tuple[dict, subprocess.Popen | None]]:
|
|
"""Start kubectl port-forward for each mapping; return (mapping, proc) pairs."""
|
|
procs: list[tuple[dict, subprocess.Popen | None]] = []
|
|
for m in mappings:
|
|
cmd = _build_kubectl_cmd(m)
|
|
m_id = m.get("id", "?")
|
|
if verbose:
|
|
print(f"[PORT-FWD] Starting: {' '.join(cmd)}", flush=True)
|
|
try:
|
|
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
procs.append((m, proc))
|
|
print(
|
|
f" ✓ {m_id}: localhost:{m.get('local','')} → {m.get('ns','')}/{m.get('svc','')}:{m.get('remote','')}",
|
|
flush=True,
|
|
)
|
|
except Exception as e:
|
|
print(f" ✗ {m_id}: failed to start — {e}", flush=True)
|
|
procs.append((m, None))
|
|
return procs
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Monitor loop
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def run_monitor(
|
|
controller, cfg_path: str | None, pf_cfg_path: str | None, verbose: bool = False
|
|
) -> int:
|
|
"""Main entry point for the 'monitor' subcommand."""
|
|
project_root = controller.project_root
|
|
|
|
# Resolve knoe.cfg for context
|
|
if cfg_path:
|
|
knoe_cfg = Path(_expand_path(cfg_path))
|
|
else:
|
|
conf_dir = knoe_conf.resolve_knoe_conf_dir(project_root)
|
|
try:
|
|
knoe_cfg = knoe_conf.ensure_entrypoint(conf_dir)
|
|
except Exception:
|
|
knoe_cfg = knoe_conf.entrypoint_path(conf_dir)
|
|
|
|
# Read knoe.cfg (layered) to extract deployment parameters
|
|
cfg = knoe_conf.load_layered_config(knoe_cfg)
|
|
# Build vars: start with User section (concrete values), then overlay Global/System
|
|
cfg_vars: dict[str, str] = {}
|
|
if cfg.has_section("User"):
|
|
cfg_vars.update(dict(cfg.items("User")))
|
|
overlay = _collect_cfg_vars(cfg)
|
|
# Only overlay non-self-referencing values from Global/System
|
|
for k, v in overlay.items():
|
|
if "${" not in v:
|
|
cfg_vars[k] = v
|
|
# Resolve remaining references iteratively
|
|
for _ in range(5):
|
|
changed = False
|
|
for k, v in list(cfg_vars.items()):
|
|
expanded = _expand_cfg_value(v, cfg_vars)
|
|
if expanded != v:
|
|
cfg_vars[k] = expanded
|
|
changed = True
|
|
if not changed:
|
|
break
|
|
|
|
mode = (
|
|
_expand_cfg_value(
|
|
cfg.get("Global", "DEPLOYMENT_MODE", fallback="k3d"), cfg_vars
|
|
).strip()
|
|
or "k3d"
|
|
)
|
|
service_ns = (
|
|
_expand_cfg_value(
|
|
cfg.get("Global", "SERVICE_NAMESPACE", fallback="default"), cfg_vars
|
|
).strip()
|
|
or "default"
|
|
)
|
|
db_ns = (
|
|
_expand_cfg_value(
|
|
cfg.get("Global", "NAMESPACE", fallback="default"), cfg_vars
|
|
).strip()
|
|
or "default"
|
|
)
|
|
db_host_port = (
|
|
_expand_cfg_value(
|
|
cfg.get("Global", "DB_HOST_PORT", fallback="5432"), cfg_vars
|
|
).strip()
|
|
or "5432"
|
|
)
|
|
supabase_enabled = _expand_cfg_value(
|
|
cfg.get("Inputs", "init_cluster.supabase_enabled", fallback="false"), cfg_vars
|
|
).strip().lower() in ("true", "1", "yes")
|
|
supabase_ns = "supabase"
|
|
gitops_enabled = _expand_cfg_value(
|
|
cfg.get("Inputs", "init_cluster.gitops_enabled", fallback="false"), cfg_vars
|
|
).strip().lower() in ("true", "1", "yes")
|
|
gitops_ns = _expand_cfg_value(
|
|
cfg.get("GitOps", "GITOPS_NAMESPACE", fallback="gitea"), cfg_vars
|
|
).strip() or "gitea"
|
|
|
|
# Resolve port-forwards.cfg path
|
|
if pf_cfg_path:
|
|
pf_path = Path(pf_cfg_path)
|
|
if not pf_path.is_absolute():
|
|
pf_path = project_root / pf_path
|
|
else:
|
|
pf_path = project_root / "conf" / "port-forwards.cfg"
|
|
|
|
# Generate canonical mappings
|
|
raw_mappings = _build_required_port_forwards(
|
|
mode=mode,
|
|
service_ns=service_ns,
|
|
argocd_ns="argocd",
|
|
db_ns=db_ns,
|
|
db_host_port=db_host_port,
|
|
supabase_enabled=supabase_enabled,
|
|
supabase_namespace=supabase_ns,
|
|
gitops_enabled=gitops_enabled,
|
|
gitops_namespace=gitops_ns,
|
|
)
|
|
|
|
# Write / refresh the port-forwards.cfg
|
|
write_port_forwards_cfg(pf_path, raw_mappings)
|
|
print(f"[CONFIG] Port-forwards config written to {pf_path}", flush=True)
|
|
|
|
# Parse the written file back for monitoring
|
|
mappings = _parse_pf_cfg(pf_path)
|
|
|
|
# Check for conflicts
|
|
conflicts = check_port_conflicts(mappings)
|
|
if conflicts:
|
|
print("\n[CONFLICT] Port conflicts detected:", flush=True)
|
|
for c in conflicts:
|
|
print(f" ⚠ {c}", flush=True)
|
|
return 1
|
|
|
|
print(f"\n[OK] {len(mappings)} port-forward mappings — no conflicts.\n", flush=True)
|
|
|
|
# Print summary table
|
|
print(
|
|
f"{'Service':<25} {'Local':>6} → {'Remote':>6} {'Namespace':<25} {'Target':<40} {'Address'}",
|
|
flush=True,
|
|
)
|
|
print("-" * 130, flush=True)
|
|
for m in mappings:
|
|
print(
|
|
f"{m.get('id',''):<25} {m.get('local',''):>6} → {m.get('remote',''):>6} {m.get('ns',''):<25} svc/{m.get('svc',''):<36} {m.get('address','')}",
|
|
flush=True,
|
|
)
|
|
print(flush=True)
|
|
|
|
# Start port-forwards
|
|
procs = _start_port_forwards(mappings, verbose=verbose)
|
|
|
|
failed = [m_id for (m, p) in procs if p is None for m_id in [m.get("id", "?")]]
|
|
started = [(m, p) for (m, p) in procs if p is not None]
|
|
|
|
if failed:
|
|
print(f"\n[WARN] Failed to start: {', '.join(failed)}", flush=True)
|
|
|
|
if not started:
|
|
print("[ERROR] No port-forwards could be started.", flush=True)
|
|
return 1
|
|
|
|
# Brief monitoring: wait a few seconds and check which processes are still alive
|
|
print(f"\n[MONITOR] Checking port-forward health (3s)...", flush=True)
|
|
time.sleep(3)
|
|
|
|
alive = 0
|
|
dead_services: list[str] = []
|
|
for m, p in started:
|
|
m_id = m.get("id", "?")
|
|
if p.poll() is None:
|
|
alive += 1
|
|
else:
|
|
stderr_out = ""
|
|
try:
|
|
stderr_out = p.stderr.read().decode(errors="replace").strip()
|
|
except Exception:
|
|
pass
|
|
dead_services.append(m_id)
|
|
print(
|
|
f" ✗ {m_id}: exited (rc={p.returncode}) {stderr_out[:200]}", flush=True
|
|
)
|
|
|
|
print(f"\n[RESULT] {alive}/{len(started)} port-forwards running.", flush=True)
|
|
if dead_services:
|
|
print(
|
|
f"[WARN] Services that did not come up: {', '.join(dead_services)}",
|
|
flush=True,
|
|
)
|
|
|
|
# Cleanup
|
|
for m, p in started:
|
|
if p.poll() is None:
|
|
try:
|
|
p.terminate()
|
|
except Exception:
|
|
pass
|
|
|
|
return 0 if alive > 0 else 1
|