890 lines
31 KiB
Python
Executable File
890 lines
31 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
ops-data-collector.py — Ops Portal Dashboard Data Collector
|
|
|
|
SCHEDULE: Every 5 minutes via no_agent cron
|
|
CRON: */5 * * * * (or interval 5m in Hermes cron)
|
|
|
|
Collects all infrastructure status data from:
|
|
- Hermes cron jobs
|
|
- systemd services
|
|
- Disk & memory usage
|
|
- S3 backup status (wasabi)
|
|
- API health checks (admin-ai, caddy, ports, cloudflare)
|
|
- Service versions (hermes, caddy, python, OS)
|
|
- Router status (placeholder)
|
|
- Hetzner cloud servers
|
|
|
|
Output: /var/www/ops/data/ops-status.json
|
|
Log: /var/log/ops-collector.log
|
|
|
|
Self-contained — no external Python deps beyond stdlib.
|
|
All errors caught and reported inline — never crashes.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone, timedelta
|
|
from pathlib import Path
|
|
from urllib.request import Request, urlopen
|
|
from urllib.error import URLError, HTTPError
|
|
|
|
# ─── Configuration ────────────────────────────────────────────────────────────
|
|
|
|
OUTPUT_PATH = Path("/var/www/ops/data/ops-status.json")
|
|
LOG_PATH = Path("/var/log/ops-collector.log")
|
|
HERMES_CRON_JOBS = Path("/root/.hermes/cron/jobs.json")
|
|
HERMES_CONFIG = Path("/root/.hermes/config.yaml")
|
|
HERMES_ENV = Path("/root/.hermes/.env")
|
|
AWS_VENV = Path("/opt/awscli-venv/bin/activate")
|
|
|
|
SERVICES_MONITORED = [
|
|
"hermes.service",
|
|
"hermes-assistant.service",
|
|
"hermes-browser.service",
|
|
"shark-game.service",
|
|
"caddy.service",
|
|
"ollama.service",
|
|
"mysql-tunnel.service",
|
|
"strongswan-swanctl.service",
|
|
]
|
|
|
|
PORT_CHECKS = {
|
|
"hermes-assistant": 8082,
|
|
"shark-game": 8083,
|
|
"browser-cdp": 9222,
|
|
}
|
|
|
|
ADMIN_AI_URL = "https://admin-ai.itpropartner.com/v1/models"
|
|
CADDY_CONFIG_URL = "http://localhost:2019/config/"
|
|
CLOUDFLARE_VERIFY_URL = "https://api.cloudflare.com/client/v4/user/tokens/verify"
|
|
|
|
|
|
# ─── Logging ──────────────────────────────────────────────────────────────────
|
|
|
|
def log(msg: str) -> None:
|
|
"""Append a timestamped line to the log file."""
|
|
try:
|
|
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
with open(LOG_PATH, "a") as f:
|
|
f.write(f"[{ts}] {msg}\n")
|
|
except OSError:
|
|
pass # Fail silently — last resort
|
|
|
|
|
|
def safe(func, *args, **kwargs):
|
|
"""Run a function, returning (result, None) or (None, error_dict)."""
|
|
try:
|
|
return func(*args, **kwargs), None
|
|
except Exception as e:
|
|
err = {"status": "error", "message": str(e)}
|
|
log(f"ERROR in {func.__name__}: {e}")
|
|
return None, err
|
|
|
|
|
|
# ─── Helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
def run_cmd(cmd: list[str], timeout: int = 15) -> str | None:
|
|
"""Run a shell command and return stdout stripped, or None on failure."""
|
|
try:
|
|
r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
|
|
return r.stdout.strip() if r.returncode == 0 else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def run_cmd_all(cmd: list[str], timeout: int = 15) -> tuple[str, str, int]:
|
|
"""Run a shell command and return (stdout, stderr, returncode)."""
|
|
try:
|
|
r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
|
|
return r.stdout.strip(), r.stderr.strip(), r.returncode
|
|
except FileNotFoundError:
|
|
return "", "command not found", -1
|
|
except subprocess.TimeoutExpired:
|
|
return "", "timed out", -1
|
|
except Exception as e:
|
|
return "", str(e), -1
|
|
|
|
|
|
def http_get(url: str, headers: dict | None = None, timeout: int = 10) -> tuple[int, str | None]:
|
|
"""Perform an HTTP GET. Returns (status_code, body_or_error_msg)."""
|
|
try:
|
|
req = Request(url, headers=headers or {}, method="GET")
|
|
with urlopen(req, timeout=timeout) as resp:
|
|
body = resp.read().decode("utf-8", errors="replace")
|
|
return resp.status, body
|
|
except HTTPError as e:
|
|
return e.code, str(e.reason)
|
|
except URLError as e:
|
|
return 0, str(e.reason)
|
|
except Exception as e:
|
|
return 0, str(e)
|
|
|
|
|
|
def check_port(host: str, port: int, timeout: float = 3.0) -> str:
|
|
"""Check if a TCP port is open using /dev/tcp or socket."""
|
|
try:
|
|
import socket
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.settimeout(timeout)
|
|
result = s.connect_ex((host, port))
|
|
s.close()
|
|
return "ok" if result == 0 else "closed"
|
|
except Exception as e:
|
|
return f"error: {e}"
|
|
|
|
|
|
def parse_env_val(content: str, key: str) -> str | None:
|
|
"""Extract a value from a shell env file (export VAR=val or VAR=val)."""
|
|
m = re.search(rf'^{key}=[\'"]?(.*?)[\'"]?\s*$', content, re.MULTILINE)
|
|
if m:
|
|
return m.group(1)
|
|
m = re.search(rf'^export {key}=[\'"]?(.*?)[\'"]?\s*$', content, re.MULTILINE)
|
|
return m.group(1) if m else None
|
|
|
|
|
|
def read_config_value(text: str, key_path: str) -> str | None:
|
|
"""
|
|
Naive yaml value reader for our specific config shape.
|
|
key_path like 'providers.admin-ai.api_key' or 'model.base_url'.
|
|
Returns the scalar value or None.
|
|
"""
|
|
keys = key_path.split(".")
|
|
lines = text.splitlines()
|
|
depth = 0
|
|
target_depth = len(keys)
|
|
key_idx = 0
|
|
for i, line in enumerate(lines):
|
|
stripped = line.lstrip()
|
|
indent = len(line) - len(stripped)
|
|
if indent <= depth and stripped and not stripped.startswith("#"):
|
|
depth = indent
|
|
if stripped.startswith(keys[key_idx] + ":"):
|
|
if key_idx == target_depth - 1:
|
|
# Found the final key — extract value
|
|
val = stripped.split(":", 1)[1].strip()
|
|
return val.strip('"').strip("'")
|
|
key_idx += 1
|
|
depth = indent + 2 # next depth
|
|
elif indent <= depth:
|
|
# Reset if we've gone back a level
|
|
if key_idx > 0 and indent <= depth - 2:
|
|
key_idx -= 1
|
|
depth = indent
|
|
return None
|
|
|
|
|
|
# ─── Data Collectors ──────────────────────────────────────────────────────────
|
|
|
|
def collect_cron_jobs() -> list[dict]:
|
|
"""Extract name, schedule, last_run_at, last_status, script from Hermes cron jobs."""
|
|
results = []
|
|
if not HERMES_CRON_JOBS.exists():
|
|
return results
|
|
|
|
try:
|
|
data = json.loads(HERMES_CRON_JOBS.read_text())
|
|
except (json.JSONDecodeError, OSError) as e:
|
|
log(f"Failed to read cron jobs: {e}")
|
|
return results
|
|
|
|
script_dirs = [
|
|
Path("/root/.hermes/scripts"),
|
|
Path("/root/.hermes/cron"),
|
|
Path("/root"),
|
|
]
|
|
|
|
def find_script(name: str) -> Path | None:
|
|
if not name:
|
|
return None
|
|
for d in script_dirs:
|
|
p = d / name
|
|
if p.exists() and p.is_file():
|
|
return p
|
|
basename = Path(name).name
|
|
for d in script_dirs:
|
|
p = d / basename
|
|
if p.exists() and p.is_file():
|
|
return p
|
|
return None
|
|
|
|
for job in data.get("jobs", []):
|
|
schedule = job.get("schedule", {})
|
|
schedule_display = schedule.get("display", "unknown")
|
|
script_name = job.get("script", "")
|
|
script_path = None
|
|
script_contents = None
|
|
if script_name:
|
|
sp = find_script(script_name)
|
|
if sp:
|
|
script_path = str(sp)
|
|
try:
|
|
content = sp.read_text(encoding="utf-8", errors="replace")
|
|
if len(content) > 5000:
|
|
content = content[:5000] + "\n\n# ... (truncated)"
|
|
script_contents = content
|
|
except Exception:
|
|
pass
|
|
results.append({
|
|
"name": job.get("name", "unnamed"),
|
|
"schedule": schedule_display,
|
|
"lastRun": job.get("last_run_at", ""),
|
|
"status": job.get("last_status", "unknown"),
|
|
"script": script_name,
|
|
"script_path": script_path or "",
|
|
"script_contents": script_contents or "",
|
|
})
|
|
return results
|
|
|
|
|
|
def collect_services() -> dict:
|
|
"""Check systemd service status for monitored services."""
|
|
results = {}
|
|
for svc in SERVICES_MONITORED:
|
|
out = run_cmd(["systemctl", "is-active", svc])
|
|
results[svc.removesuffix(".service")] = out if out else "inactive"
|
|
return results
|
|
|
|
|
|
def collect_disk_memory() -> dict:
|
|
"""Collect disk and memory usage percentages."""
|
|
result = {"disk_used_pct": 0, "memory_used_pct": 0}
|
|
|
|
# Disk
|
|
out = run_cmd(["df", "-h", "/"])
|
|
if out:
|
|
lines = out.splitlines()
|
|
if len(lines) >= 2:
|
|
parts = lines[1].split()
|
|
if len(parts) >= 5:
|
|
pct = parts[4].replace("%", "")
|
|
try:
|
|
result["disk_used_pct"] = int(pct)
|
|
except ValueError:
|
|
pass
|
|
|
|
# Memory
|
|
out = run_cmd(["free", "-m"])
|
|
if out:
|
|
for line in out.splitlines():
|
|
if line.startswith("Mem:"):
|
|
parts = line.split()
|
|
if len(parts) >= 3:
|
|
try:
|
|
total = int(parts[1])
|
|
used = int(parts[2])
|
|
if total > 0:
|
|
result["memory_used_pct"] = round(used / total * 100)
|
|
except (ValueError, IndexError):
|
|
pass
|
|
break
|
|
|
|
return result
|
|
|
|
|
|
def check_s3_backup(bucket_path: str, aws_args: list[str] | None = None) -> dict:
|
|
"""
|
|
Check S3 for most recent upload timestamp.
|
|
Returns dict with last_upload, status, and optionally error.
|
|
"""
|
|
cmd = ["aws", "s3", "ls", bucket_path, "--recursive", "--endpoint-url", "https://s3.us-east-1.wasabisys.com"]
|
|
if aws_args:
|
|
cmd.extend(aws_args)
|
|
|
|
# Try with awscli venv first
|
|
env = os.environ.copy()
|
|
if AWS_VENV.exists():
|
|
# Use the aws binary directly from venv
|
|
aws_bin = AWS_VENV.parent / "aws"
|
|
if aws_bin.exists():
|
|
cmd[0] = str(aws_bin)
|
|
|
|
out, err, rc = run_cmd_all(cmd, timeout=20)
|
|
if rc != 0:
|
|
return {"status": "error", "message": err or f"exit code {rc}"}
|
|
|
|
# Parse s3 ls output — lines like: 2026-07-08 20:30:15 42 some-file
|
|
latest = None
|
|
for line in out.splitlines():
|
|
parts = line.strip().split()
|
|
if len(parts) >= 4:
|
|
ts_str = f"{parts[0]} {parts[1]}"
|
|
try:
|
|
ts = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)
|
|
if latest is None or ts > latest:
|
|
latest = ts
|
|
except ValueError:
|
|
continue
|
|
|
|
if latest is None:
|
|
return {"status": "empty", "last_upload": None}
|
|
|
|
now = datetime.now(timezone.utc)
|
|
delta = now - latest
|
|
age_hours = delta.total_seconds() / 3600
|
|
|
|
# Status: ok (< 26h), stale (26-72h), critical (> 72h)
|
|
if age_hours < 26:
|
|
status = "ok"
|
|
elif age_hours < 72:
|
|
status = "stale"
|
|
else:
|
|
status = "critical"
|
|
|
|
return {
|
|
"status": status,
|
|
"last_upload": latest.strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"age_hours": round(age_hours, 1),
|
|
}
|
|
|
|
|
|
def collect_s3_backups() -> dict:
|
|
"""Check all monitored S3 backup buckets."""
|
|
results = {}
|
|
|
|
r, err = safe(check_s3_backup, "s3://hermes-vps-backups/live/")
|
|
results["hermes-vps-backups"] = r if r else (err or {"status": "error", "message": "unknown"})
|
|
|
|
r, err = safe(check_s3_backup, "s3://hermes-vps-backups/hermes-full-backup/")
|
|
results["hermes-full-backups"] = r if r else (err or {"status": "error", "message": "unknown"})
|
|
|
|
r, err = safe(check_s3_backup, "s3://mikrotik-ccr-backups/wisp-backups/configs/")
|
|
results["mikrotik-backups"] = r if r else (err or {"status": "error", "message": "unknown"})
|
|
|
|
r, err = safe(check_s3_backup, "s3://itpropartner-system-configs/")
|
|
results["itpropartner-system-configs"] = r if r else (err or {"status": "error", "message": "unknown"})
|
|
|
|
r, err = safe(check_s3_backup, "s3://itpropartner-docker-volumes/")
|
|
results["itpropartner-docker-volumes"] = r if r else (err or {"status": "error", "message": "unknown"})
|
|
|
|
r, err = safe(check_s3_backup, "s3://itpropartner-backups/")
|
|
rcf = r if r else (err or {"status": "deprecated", "message": "Legacy bucket — no longer written to"})
|
|
if isinstance(rcf, dict) and rcf.get("status") not in ("ok", "empty"):
|
|
rcf["status"] = "deprecated"
|
|
rcf["message"] = "Legacy bucket — no longer actively written to"
|
|
results["itpropartner-backups"] = rcf
|
|
|
|
return results
|
|
|
|
|
|
def collect_api_checks() -> dict:
|
|
"""Check API endpoints and port availability."""
|
|
results = {}
|
|
|
|
# Source env vars for tokens
|
|
cloudflare_token = _get_env_val("CLOUDFLARE_API_TOKEN")
|
|
|
|
# Read admin-ai api_key from config.yaml
|
|
admin_ai_key = None
|
|
try:
|
|
config_text = HERMES_CONFIG.read_text()
|
|
# Find api_key under providers.admin-ai
|
|
in_admin_ai = False
|
|
in_providers = False
|
|
for line in config_text.splitlines():
|
|
stripped = line.lstrip()
|
|
if stripped.startswith("providers:"):
|
|
in_providers = True
|
|
continue
|
|
if in_providers and stripped.startswith("admin-ai:"):
|
|
in_admin_ai = True
|
|
continue
|
|
if in_admin_ai and stripped.startswith("api_key:"):
|
|
admin_ai_key = stripped.split(":", 1)[1].strip().strip('"').strip("'")
|
|
break
|
|
# Reset if indent goes back to 0
|
|
if in_admin_ai and not line.startswith(" ") and not line.startswith("\t"):
|
|
break
|
|
if in_providers and not line.startswith(" ") and not line.startswith("\t") and stripped:
|
|
in_providers = False
|
|
except OSError:
|
|
pass
|
|
|
|
# 1. Admin-AI models endpoint
|
|
headers = {}
|
|
if admin_ai_key:
|
|
headers["Authorization"] = f"Bearer {admin_ai_key}"
|
|
status, body = http_get(ADMIN_AI_URL, headers=headers, timeout=10)
|
|
results["admin-ai"] = "ok" if status == 200 else f"HTTP {status}"
|
|
|
|
# 2. Caddy config endpoint
|
|
status, _ = http_get(CADDY_CONFIG_URL, timeout=5)
|
|
results["caddy-config"] = "ok" if status == 200 else f"HTTP {status}"
|
|
|
|
# 3. Port checks
|
|
for name, port in PORT_CHECKS.items():
|
|
results[f"port-{port}"] = check_port("127.0.0.1", port)
|
|
|
|
# 4. Cloudflare API verify + list zones
|
|
results["cloudflare_zones"] = []
|
|
if cloudflare_token:
|
|
status, body = http_get(CLOUDFLARE_VERIFY_URL, headers={
|
|
"Authorization": f"Bearer {cloudflare_token}",
|
|
"Content-Type": "application/json",
|
|
}, timeout=10)
|
|
if status == 200:
|
|
try:
|
|
j = json.loads(body)
|
|
if j.get("success"):
|
|
results["cloudflare-api"] = "ok"
|
|
# Also fetch zone list
|
|
zstatus, zbody = http_get(
|
|
"https://api.cloudflare.com/client/v4/zones",
|
|
headers={"Authorization": f"Bearer {cloudflare_token}"},
|
|
timeout=10
|
|
)
|
|
if zstatus == 200:
|
|
zj = json.loads(zbody)
|
|
results["cloudflare_zones"] = [
|
|
{"name": z["name"], "status": z["status"]}
|
|
for z in zj.get("result", [])
|
|
]
|
|
else:
|
|
results["cloudflare-api"] = "verify_failed"
|
|
except json.JSONDecodeError:
|
|
results["cloudflare-api"] = f"HTTP {status} (bad json)"
|
|
else:
|
|
results["cloudflare-api"] = f"HTTP {status}"
|
|
else:
|
|
results["cloudflare-api"] = "no_token"
|
|
|
|
return results
|
|
|
|
|
|
def collect_versions() -> dict:
|
|
"""Collect software versions."""
|
|
results = {}
|
|
|
|
# Hermes version
|
|
out, err, rc = run_cmd_all(["hermes", "--version"], timeout=10)
|
|
if rc != 0:
|
|
out, err, rc = run_cmd_all(["hermes", "version"], timeout=10)
|
|
if out:
|
|
# Extract just the version string
|
|
m = re.search(r"v?(\d+\.\d+\.\d+)", out)
|
|
results["hermes"] = m.group(0) if m else out.splitlines()[0][:60]
|
|
else:
|
|
results["hermes"] = "unknown"
|
|
|
|
# Caddy version
|
|
out = run_cmd(["caddy", "version"])
|
|
results["caddy"] = out.split()[0] if out else "unknown"
|
|
|
|
# Python version
|
|
out = run_cmd(["python3", "--version"])
|
|
results["python"] = out.replace("Python ", "") if out else "unknown"
|
|
|
|
# OS version
|
|
deb_ver = Path("/etc/debian_version")
|
|
if deb_ver.exists():
|
|
try:
|
|
results["os"] = f"Debian {deb_ver.read_text().strip()}"
|
|
except OSError:
|
|
results["os"] = "unknown"
|
|
else:
|
|
results["os"] = "unknown"
|
|
|
|
return results
|
|
|
|
|
|
def _get_env_val(key: str) -> str | None:
|
|
"""Look up an env var: first from live os.environ, then from .env file."""
|
|
val = os.environ.get(key)
|
|
if val:
|
|
return val
|
|
try:
|
|
content = HERMES_ENV.read_text()
|
|
return parse_env_val(content, key)
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
def collect_hetzner_servers() -> list[dict]:
|
|
"""Fetch Hetzner Cloud server list via API."""
|
|
token = _get_env_val("HETZNER_API_TOKEN")
|
|
# Fall back to the token file used by snapshot-hetzner.py if not in env/.env
|
|
if not token:
|
|
token_file = Path("/root/.hermes/scripts/.hetzner_token")
|
|
try:
|
|
if token_file.exists():
|
|
token = token_file.read_text().strip()
|
|
except OSError:
|
|
token = None
|
|
if not token:
|
|
return [{"status": "error", "message": "HETZNER_API_TOKEN not found (checked env, .env, and .hetzner_token)"}]
|
|
|
|
status, body = http_get(
|
|
"https://api.hetzner.cloud/v1/servers",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
timeout=15,
|
|
)
|
|
if status != 200:
|
|
return [{"status": "error", "message": f"HTTP {status}"}]
|
|
|
|
try:
|
|
data = json.loads(body)
|
|
except json.JSONDecodeError:
|
|
return [{"status": "error", "message": "invalid JSON response"}]
|
|
|
|
servers = []
|
|
for srv in data.get("servers", []):
|
|
servers.append({
|
|
"name": srv.get("name", "unknown"),
|
|
"status": srv.get("status", "unknown"),
|
|
"type": srv.get("server_type", {}).get("name", "unknown") if isinstance(srv.get("server_type"), dict) else str(srv.get("server_type", "unknown")),
|
|
"ip": (srv.get("public_net", {}) or {}).get("ipv4", {}).get("ip", ""),
|
|
"id": srv.get("id"),
|
|
})
|
|
|
|
return servers
|
|
|
|
|
|
def get_netcup_server_info() -> dict:
|
|
"""
|
|
Returns Netcup server info. Currently uses known server data.
|
|
The Netcup CCP API requires Keycloak auth flow — not trivial for
|
|
a lightweight polling script. Returns hardcoded known values.
|
|
TODO: Implement full Netcup CCP REST API auth for dynamic data.
|
|
|
|
All RS 4000 G12 — dedicated EPYC 9645 12C/32GB/1TB NVMe.
|
|
Core is the only RS 2000 G12 (8C/15GB/512GB).
|
|
"""
|
|
return {
|
|
"servers": [
|
|
{
|
|
"id": 890903,
|
|
"name": "core",
|
|
"hostname": "core",
|
|
"template": "RS 2000 G12",
|
|
"ip": "152.53.192.33",
|
|
"role": "Ops hub — Hermes, Caddy, Portal, monitoring",
|
|
"status": "running",
|
|
},
|
|
{
|
|
"name": "app1",
|
|
"hostname": "app1",
|
|
"template": "RS 4000 G12",
|
|
"ip": "152.53.36.131",
|
|
"role": "AI/ML — LiteLLM, Ollama, Open WebUI, n8n",
|
|
"status": "running",
|
|
},
|
|
{
|
|
"name": "app2",
|
|
"hostname": "app2",
|
|
"template": "RS 4000 G12",
|
|
"ip": "152.53.39.202",
|
|
"role": "Infrastructure — UISP, UNMS, UCRM",
|
|
"status": "running",
|
|
},
|
|
{
|
|
"name": "app3",
|
|
"hostname": "app3",
|
|
"template": "RS 4000 G12",
|
|
"ip": "152.53.241.111",
|
|
"role": "Web apps — WordPress migration target",
|
|
"status": "running",
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
def collect_costs() -> dict:
|
|
"""Collect API cost data from LiteLLM, OpenRouter, and Firecrawl."""
|
|
import json, os, subprocess, urllib.request
|
|
|
|
result = {
|
|
"litellm": {"status": "unknown", "total_spend": 0, "models": []},
|
|
"openrouter": {"status": "unknown", "total_usage": 0, "daily_usage": 0, "monthly_usage": 0},
|
|
"firecrawl": {"status": "unknown", "used": 0, "limit": 1000},
|
|
"total_estimated_monthly": 0,
|
|
}
|
|
|
|
# 1. LiteLLM spend from old AI server Postgres
|
|
try:
|
|
rc, out = subprocess.getstatusoutput(
|
|
'ssh -i /root/.ssh/itpp-infra -o ConnectTimeout=5 -o BatchMode=yes root@178.156.167.181 '
|
|
'"PGPASSWORD=litellm docker exec -e PGPASSWORD=litellm litellm_postgres psql -U litellm -d litellm_db -t -c '
|
|
'\\\"SELECT model, ROUND(SUM(spend)::numeric,2) as total FROM \\\\\\\"LiteLLM_SpendLogs\\\\\\\" GROUP BY model ORDER BY total DESC LIMIT 15;\\\""'
|
|
)
|
|
if rc == 0 and out.strip():
|
|
models = []
|
|
total = 0.0
|
|
for line in out.strip().split('\n'):
|
|
line = line.strip()
|
|
if not line or line.startswith('('):
|
|
continue
|
|
parts = line.split('|')
|
|
if len(parts) == 2:
|
|
model = parts[0].strip()
|
|
try:
|
|
spend = float(parts[1].strip())
|
|
except ValueError:
|
|
spend = 0.0
|
|
if spend > 0:
|
|
models.append({"model": model, "spend": spend})
|
|
total += spend
|
|
result["litellm"] = {"status": "ok", "total_spend": round(total, 2), "models": models}
|
|
except Exception as e:
|
|
result["litellm"] = {"status": "error", "message": str(e)}
|
|
|
|
# 2. OpenRouter usage
|
|
try:
|
|
key = ""
|
|
with open(os.path.expanduser("~/.hermes/config.yaml")) as f:
|
|
for line in f:
|
|
if 'api_key' in line and 'openrouter' in open('/root/.hermes/config.yaml').read():
|
|
# Simple grep approach
|
|
pass
|
|
# Read from config
|
|
import re
|
|
config_text = open(os.path.expanduser("~/.hermes/config.yaml")).read()
|
|
for m in re.finditer(r'openrouter:\s*\n\s+api_key:\s*(\S+)', config_text):
|
|
key = m.group(1)
|
|
|
|
if key:
|
|
req = urllib.request.Request(
|
|
"https://openrouter.ai/api/v1/auth/key",
|
|
headers={"Authorization": f"Bearer {key}"}
|
|
)
|
|
resp = urllib.request.urlopen(req, timeout=10)
|
|
data = json.loads(resp.read().decode())
|
|
d = data.get("data", {})
|
|
result["openrouter"] = {
|
|
"status": "ok",
|
|
"total_usage": round(d.get("usage", 0), 2),
|
|
"daily_usage": round(d.get("usage_daily", 0), 4),
|
|
"weekly_usage": round(d.get("usage_weekly", 0), 2),
|
|
"monthly_usage": round(d.get("usage_monthly", 0), 2),
|
|
}
|
|
except Exception as e:
|
|
result["openrouter"]["message"] = str(e)
|
|
|
|
# 3. Firecrawl usage from local tracker
|
|
try:
|
|
fc_path = os.path.expanduser("~/.hermes/scripts/firecrawl-usage.json")
|
|
if os.path.exists(fc_path):
|
|
fc_data = json.loads(open(fc_path).read())
|
|
result["firecrawl"] = {
|
|
"status": "ok",
|
|
"used": fc_data.get("total_used", 0),
|
|
"limit": 1000,
|
|
"remaining": 1000 - fc_data.get("total_used", 0),
|
|
}
|
|
except Exception as e:
|
|
result["firecrawl"]["message"] = str(e)
|
|
|
|
# 4. Compute total estimated monthly
|
|
total = result["litellm"].get("total_spend", 0) if isinstance(result["litellm"].get("total_spend"), (int, float)) else 0
|
|
total += result["openrouter"].get("monthly_usage", 0) if isinstance(result["openrouter"].get("monthly_usage"), (int, float)) else 0
|
|
result["total_estimated_monthly"] = round(total, 2)
|
|
|
|
# 5. Monthly breakdown by provider
|
|
by_provider = {}
|
|
for m in result.get("litellm", {}).get("models", []):
|
|
model_name = m.get("model", "unknown")
|
|
# Categorize by provider prefix
|
|
if 'openrouter/' in model_name:
|
|
provider = 'OpenRouter'
|
|
elif 'gemini' in model_name.lower():
|
|
provider = 'Google Gemini'
|
|
elif 'claude' in model_name.lower():
|
|
provider = 'Anthropic'
|
|
elif 'deepseek' in model_name.lower():
|
|
provider = 'DeepSeek'
|
|
elif 'gpt' in model_name.lower() or 'o1' in model_name.lower() or 'o3' in model_name.lower():
|
|
provider = 'OpenAI'
|
|
elif 'grok' in model_name.lower() or 'xai' in model_name.lower():
|
|
provider = 'xAI'
|
|
elif 'groq' in model_name.lower():
|
|
provider = 'Groq'
|
|
elif 'qwen' in model_name.lower():
|
|
provider = 'Groq'
|
|
else:
|
|
provider = 'Other'
|
|
by_provider[provider] = by_provider.get(provider, 0) + m.get("spend", 0)
|
|
|
|
result["by_provider"] = {k: round(v, 2) for k, v in sorted(by_provider.items(), key=lambda x: -x[1])}
|
|
|
|
return result
|
|
|
|
|
|
def compute_overall(data: dict) -> dict:
|
|
"""Compute overall summary stats from the collected data."""
|
|
cron_jobs = data.get("cron_jobs", [])
|
|
total_jobs = len(cron_jobs)
|
|
passing = sum(1 for j in cron_jobs if j.get("status") == "ok")
|
|
failing = total_jobs - passing
|
|
|
|
disk = data.get("disk_memory", {})
|
|
disk_pct = disk.get("disk_used_pct", 0)
|
|
mem_pct = disk.get("memory_used_pct", 0)
|
|
|
|
return {
|
|
"total_jobs": total_jobs,
|
|
"passing": passing,
|
|
"failing": failing,
|
|
"disk_used_pct": disk_pct,
|
|
"memory_used_pct": mem_pct,
|
|
}
|
|
|
|
|
|
# ─── Main ─────────────────────────────────────────────────────────────────────
|
|
|
|
def collect_syncromsp() -> dict:
|
|
"""Pull data from SyncroMSP API: customers, tickets, devices."""
|
|
token = "Ta9f9d1b462271a2f4-8d63a3f025eb89451edb16f2308c2e40"
|
|
base = "https://itpropartner.syncromsp.com/api/v1"
|
|
headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
|
|
result = {"status": "ok", "customers": 0, "tickets": 0, "devices": 0}
|
|
|
|
import urllib.request
|
|
|
|
# Customers
|
|
try:
|
|
req = urllib.request.Request(f"{base}/customers", headers=headers)
|
|
resp = urllib.request.urlopen(req, timeout=10)
|
|
data = json.loads(resp.read().decode())
|
|
customers = data.get("customers", [])
|
|
result["customers"] = len(customers)
|
|
result["customer_list"] = [
|
|
c.get("business_name") or f"{c.get('first_name','')} {c.get('last_name','')}".strip()
|
|
for c in customers if c.get("business_name") or c.get("first_name")
|
|
][:20]
|
|
except Exception as e:
|
|
log(f"SyncroMSP customers failed: {e}")
|
|
|
|
# Tickets
|
|
try:
|
|
req = urllib.request.Request(f"{base}/tickets", headers=headers)
|
|
resp = urllib.request.urlopen(req, timeout=10)
|
|
data = json.loads(resp.read().decode())
|
|
result["tickets"] = data.get("total", 0)
|
|
except Exception as e:
|
|
log(f"SyncroMSP tickets failed: {e}")
|
|
|
|
return result
|
|
|
|
|
|
def main():
|
|
start = time.time()
|
|
log("Starting data collection cycle...")
|
|
|
|
# 1. Cron Jobs
|
|
cron_jobs, err = safe(collect_cron_jobs)
|
|
if err:
|
|
cron_jobs = []
|
|
log(f"cron_jobs collection failed: {err}")
|
|
|
|
# 2. Systemd Services
|
|
services, err = safe(collect_services)
|
|
if err:
|
|
services = {}
|
|
log(f"services collection failed: {err}")
|
|
|
|
# 3. Disk & Memory
|
|
disk_memory, err = safe(collect_disk_memory)
|
|
if err:
|
|
disk_memory = {"disk_used_pct": 0, "memory_used_pct": 0}
|
|
log(f"disk_memory collection failed: {err}")
|
|
|
|
# 4. S3 Backups
|
|
s3_backups, err = safe(collect_s3_backups)
|
|
if err:
|
|
s3_backups = {}
|
|
log(f"s3_backups collection failed: {err}")
|
|
|
|
# 5. API Health Checks
|
|
api_checks, err = safe(collect_api_checks)
|
|
if err:
|
|
api_checks = {}
|
|
log(f"api_checks collection failed: {err}")
|
|
|
|
# 6. Service Versions
|
|
versions, err = safe(collect_versions)
|
|
if err:
|
|
versions = {}
|
|
log(f"versions collection failed: {err}")
|
|
|
|
# 7. Router Status (placeholder)
|
|
routers = [] # Placeholder for future MikroTik/UniFi API pulls
|
|
|
|
# 8. Hetzner Servers
|
|
hetzner_servers, err = safe(collect_hetzner_servers)
|
|
if err:
|
|
hetzner_servers = []
|
|
log(f"hetzner_servers collection failed: {err}")
|
|
|
|
# Netcup server info
|
|
netcup_server, err = safe(get_netcup_server_info)
|
|
if err:
|
|
netcup_server = {}
|
|
|
|
# 9. SyncroMSP
|
|
syncromsp, err = safe(collect_syncromsp)
|
|
if err:
|
|
syncromsp = {"status": "error", "message": str(err)}
|
|
|
|
# 10. Cost Tracking
|
|
costs, err = safe(collect_costs)
|
|
if err:
|
|
costs = {"status": "error", "message": str(err)}
|
|
|
|
# Assemble payload
|
|
raw_data = {
|
|
"cron_jobs": cron_jobs,
|
|
"services": services,
|
|
"disk_memory": disk_memory,
|
|
"s3_backups": s3_backups,
|
|
"api_checks": api_checks,
|
|
"versions": versions,
|
|
"routers": routers,
|
|
"hetzner_servers": hetzner_servers,
|
|
"netcup_server": netcup_server,
|
|
"syncromsp": syncromsp,
|
|
"costs": costs,
|
|
}
|
|
|
|
overall = compute_overall(raw_data)
|
|
|
|
payload = {
|
|
"timestamp": datetime.now(timezone.utc).astimezone().isoformat(),
|
|
"collection_duration_ms": round((time.time() - start) * 1000),
|
|
"overall": overall,
|
|
"cron_jobs": cron_jobs,
|
|
"services": services,
|
|
"disk_memory": disk_memory,
|
|
"s3_backups": s3_backups,
|
|
"api_checks": {k: v for k, v in api_checks.items() if k != "cloudflare_zones"},
|
|
"cloudflare_zones": api_checks.get("cloudflare_zones", []),
|
|
"versions": versions,
|
|
"routers": routers,
|
|
"netcup_server": netcup_server,
|
|
"hetzner_servers": hetzner_servers,
|
|
"syncromsp": syncromsp,
|
|
"costs": costs,
|
|
}
|
|
|
|
# Write output
|
|
try:
|
|
OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
json_str = json.dumps(payload, indent=2, default=str)
|
|
OUTPUT_PATH.write_text(json_str)
|
|
log(f"Wrote {len(json_str)} bytes to {OUTPUT_PATH}")
|
|
except OSError as e:
|
|
log(f"CRITICAL: Failed to write output file: {e}")
|
|
print(f"CRITICAL: Failed to write output: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
elapsed = time.time() - start
|
|
log(f"Collection cycle complete in {elapsed:.2f}s — overall: {overall['passing']}/{overall['total_jobs']} jobs passing")
|
|
print(f"Collection complete in {elapsed:.2f}s — {OUTPUT_PATH}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|