Hyungi Ahn
b1f9e87d6a
mcp-infra-server를 gpu-services/infra/로 통합. core/ 순수 로직은 Agent/NanoClaude에서도 직접 import 가능. 도구: docker_status, docker_logs, service_health, disk_usage, tailscale_status, ollama_models, mlx_models. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
84 lines
2.3 KiB
Python
84 lines
2.3 KiB
Python
"""Network tools — Tailscale status."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from ..schemas import TailscaleResult, TailscalePeer
|
|
from .ssh import run_local, SSHError
|
|
|
|
TAILSCALE_BIN = "/Applications/Tailscale.app/Contents/MacOS/Tailscale"
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def _parse_tailscale(output: str) -> list[TailscalePeer]:
|
|
"""Parse `tailscale status` output into peer list.
|
|
|
|
Format: IP HOSTNAME USER@ OS STATUS_INFO
|
|
Status examples: "-" (connected/active), "idle, tx ... rx ...", "offline, last seen ..."
|
|
"""
|
|
peers = []
|
|
for line in output.strip().splitlines():
|
|
parts = line.split()
|
|
if len(parts) < 4:
|
|
continue
|
|
# Skip header-like lines
|
|
if parts[0].startswith("#") or parts[0] == "IP":
|
|
continue
|
|
|
|
ip = parts[0]
|
|
hostname = parts[1]
|
|
# parts[2] = user@ (skip)
|
|
os_name = parts[3] if len(parts) > 3 else ""
|
|
|
|
# Remaining is status info
|
|
status_text = " ".join(parts[4:]) if len(parts) > 4 else ""
|
|
|
|
if "offline" in status_text:
|
|
status = "offline"
|
|
elif "idle" in status_text:
|
|
status = "idle"
|
|
elif status_text == "-" or status_text == "":
|
|
status = "active"
|
|
else:
|
|
status = "active"
|
|
|
|
peers.append(TailscalePeer(
|
|
hostname=hostname,
|
|
ip=ip,
|
|
status=status,
|
|
os=os_name,
|
|
))
|
|
return peers
|
|
|
|
|
|
async def tailscale_status() -> TailscaleResult:
|
|
"""Get Tailscale network status (runs locally)."""
|
|
try:
|
|
stdout, _ = await run_local(f"{TAILSCALE_BIN} status")
|
|
except SSHError as e:
|
|
return TailscaleResult(
|
|
ok=False, checked_at=_now(),
|
|
error_type=e.error_type, error=str(e),
|
|
)
|
|
|
|
peers = _parse_tailscale(stdout)
|
|
|
|
warnings = []
|
|
expected_hosts = {"sub-server", "hyungi-macmini", "hyungi-macbookpro"}
|
|
found_hosts = {p.hostname for p in peers}
|
|
missing = expected_hosts - found_hosts
|
|
for h in missing:
|
|
warnings.append(f"{h} not found in tailnet")
|
|
|
|
return TailscaleResult(
|
|
ok=True,
|
|
checked_at=_now(),
|
|
peers=peers,
|
|
warnings=warnings,
|
|
raw=stdout.strip(),
|
|
)
|