initial: Ethica Agent v0.1.0
Lightweight telemetry agent (stdlib-only, Python 3.8+) that collects
metrics via pre-defined widget types and pushes to POST /api/company/metrics.
Widget types: system_cpu, system_memory, system_disk, http_check,
port_check, process_check, custom_cmd
Platform: Linux (fully). Windows collectors stubbed (NOT_IMPL markers)
ready for porting without changing call sites. system_disk / http_check /
port_check / custom_cmd already cross-platform.
Delivery: run directly (python3 ethica-agent.py --config ethica-agent.yml),
as a systemd service (ethica-agent.service), or as a Docker container.
This commit is contained in:
+352
@@ -0,0 +1,352 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Ethica Agent — lightweight telemetry client for Ethica.no "home".
|
||||
|
||||
Collects metrics from pre-defined widget types and pushes them to
|
||||
POST /api/company/metrics on the configured Ethica home.
|
||||
|
||||
Requirements : Python 3.8+, no external packages (stdlib only).
|
||||
Platforms : Linux (fully supported).
|
||||
Windows: collectors are stubbed — add _*_windows()
|
||||
implementations when ready (see NOT_IMPL markers).
|
||||
macOS: http_check / port_check / system_disk work today;
|
||||
cpu/memory/process need _*_darwin() stubs.
|
||||
|
||||
Widget types : system_cpu, system_memory, system_disk,
|
||||
http_check, port_check, process_check, custom_cmd
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import platform
|
||||
import shutil
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import urllib.request
|
||||
from urllib.error import URLError
|
||||
|
||||
VERSION = "0.1.0"
|
||||
SYSTEM = platform.system() # "Linux" | "Windows" | "Darwin"
|
||||
|
||||
|
||||
# ── Config (minimal YAML subset) ───────────────────────────────────────────────
|
||||
|
||||
def _cast(v: str):
|
||||
"""Cast a YAML scalar string to int, float, bool, or str."""
|
||||
if not isinstance(v, str):
|
||||
return v
|
||||
if v.lower() in ("true", "yes", "on"):
|
||||
return True
|
||||
if v.lower() in ("false", "no", "off"):
|
||||
return False
|
||||
try:
|
||||
return int(v)
|
||||
except ValueError:
|
||||
pass
|
||||
try:
|
||||
return float(v)
|
||||
except ValueError:
|
||||
pass
|
||||
return v
|
||||
|
||||
|
||||
def load_config(path: str) -> dict:
|
||||
"""Load ethica-agent.yml.
|
||||
|
||||
Handles the specific subset used by this agent:
|
||||
root-level key: value
|
||||
widgets list - type: ...
|
||||
key: value
|
||||
Comments (#) and blank lines are ignored.
|
||||
"""
|
||||
with open(path, encoding="utf-8") as f:
|
||||
lines = f.readlines()
|
||||
|
||||
root: dict = {}
|
||||
widgets: list = []
|
||||
current: dict | None = None
|
||||
|
||||
for raw in lines:
|
||||
line = raw.rstrip()
|
||||
stripped = line.lstrip()
|
||||
if not stripped or stripped.startswith("#"):
|
||||
continue
|
||||
indent = len(line) - len(stripped)
|
||||
|
||||
if stripped.startswith("- "):
|
||||
current = {}
|
||||
widgets.append(current)
|
||||
stripped = stripped[2:]
|
||||
|
||||
if ":" in stripped:
|
||||
k, _, v = stripped.partition(":")
|
||||
k, v = k.strip(), v.strip()
|
||||
if indent == 0:
|
||||
if k != "widgets":
|
||||
root[k] = _cast(v)
|
||||
elif current is not None:
|
||||
current[k] = _cast(v)
|
||||
|
||||
if widgets:
|
||||
root["widgets"] = widgets
|
||||
return root
|
||||
|
||||
|
||||
# ── Platform-specific collectors ───────────────────────────────────────────────
|
||||
# Pattern: one _*_linux() implementation per metric, a _*_windows() stub
|
||||
# (NOT_IMPL), and a dispatcher that selects by SYSTEM. Add the Windows/Darwin
|
||||
# body when you're ready — the call site never changes.
|
||||
|
||||
def _cpu_linux() -> float:
|
||||
"""Two-sample /proc/stat read (0.5 s window) → CPU usage %."""
|
||||
def _read():
|
||||
with open("/proc/stat") as f:
|
||||
parts = f.readline().split()
|
||||
vals = list(map(int, parts[1:]))
|
||||
return vals[3], sum(vals) # idle, total
|
||||
i1, t1 = _read()
|
||||
time.sleep(0.5)
|
||||
i2, t2 = _read()
|
||||
dt = t2 - t1
|
||||
return round(100.0 * (1 - (i2 - i1) / max(dt, 1)), 1)
|
||||
|
||||
|
||||
def _cpu_windows() -> float:
|
||||
# NOT_IMPL: use ctypes + kernel32.GetSystemTimes or wmi.WMI().Win32_Processor()
|
||||
raise NotImplementedError("system_cpu on Windows — not yet implemented")
|
||||
|
||||
|
||||
def cpu_percent() -> float:
|
||||
if SYSTEM == "Linux":
|
||||
return _cpu_linux()
|
||||
if SYSTEM == "Windows":
|
||||
return _cpu_windows()
|
||||
raise NotImplementedError(f"system_cpu on {SYSTEM} — not yet implemented")
|
||||
|
||||
|
||||
def _mem_linux() -> float:
|
||||
info: dict = {}
|
||||
with open("/proc/meminfo") as f:
|
||||
for line in f:
|
||||
k, _, v = line.partition(":")
|
||||
info[k.strip()] = int(v.split()[0])
|
||||
total = info.get("MemTotal", 1)
|
||||
avail = info.get("MemAvailable", 0)
|
||||
return round(100.0 * (total - avail) / total, 1)
|
||||
|
||||
|
||||
def _mem_windows() -> float:
|
||||
# NOT_IMPL: use ctypes + GlobalMemoryStatusEx from kernel32
|
||||
raise NotImplementedError("system_memory on Windows — not yet implemented")
|
||||
|
||||
|
||||
def mem_percent() -> float:
|
||||
if SYSTEM == "Linux":
|
||||
return _mem_linux()
|
||||
if SYSTEM == "Windows":
|
||||
return _mem_windows()
|
||||
raise NotImplementedError(f"system_memory on {SYSTEM} — not yet implemented")
|
||||
|
||||
|
||||
def disk_percent(path: str = "/") -> float:
|
||||
"""shutil.disk_usage is cross-platform (Linux, Windows, macOS)."""
|
||||
u = shutil.disk_usage(path)
|
||||
return round(100.0 * u.used / u.total, 1)
|
||||
|
||||
|
||||
def _process_alive_linux(name: str) -> bool:
|
||||
try:
|
||||
r = subprocess.run(["pgrep", "-x", name], capture_output=True, timeout=5)
|
||||
return r.returncode == 0
|
||||
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||
pass
|
||||
for pid in os.listdir("/proc"):
|
||||
if not pid.isdigit():
|
||||
continue
|
||||
try:
|
||||
with open(f"/proc/{pid}/comm") as f:
|
||||
if f.read().strip() == name:
|
||||
return True
|
||||
except OSError:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
def _process_alive_windows(name: str) -> bool:
|
||||
# NOT_IMPL: subprocess.run(['tasklist', '/FI', f'IMAGENAME eq {name}'])
|
||||
raise NotImplementedError("process_check on Windows — not yet implemented")
|
||||
|
||||
|
||||
def process_alive(name: str) -> bool:
|
||||
if SYSTEM == "Linux":
|
||||
return _process_alive_linux(name)
|
||||
if SYSTEM == "Windows":
|
||||
return _process_alive_windows(name)
|
||||
raise NotImplementedError(f"process_check on {SYSTEM} — not yet implemented")
|
||||
|
||||
|
||||
# ── Widget runners ─────────────────────────────────────────────────────────────
|
||||
|
||||
def _pct_status(pct: float, degrade: int, offline: int) -> str:
|
||||
if pct >= offline:
|
||||
return "offline"
|
||||
if pct >= degrade:
|
||||
return "degraded"
|
||||
return "online"
|
||||
|
||||
|
||||
def run_widget(w: dict) -> tuple:
|
||||
"""Run one widget config dict.
|
||||
|
||||
Returns (name, kind, status, latency_ms, version, note).
|
||||
Raises NotImplementedError for unimplemented platform collectors.
|
||||
Raises ValueError for unknown widget type.
|
||||
"""
|
||||
wtype = w.get("type", "")
|
||||
name = w.get("name", wtype)
|
||||
kind = w.get("kind", "system")
|
||||
|
||||
if wtype == "system_cpu":
|
||||
pct = cpu_percent()
|
||||
status = _pct_status(pct, w.get("degrade_threshold", 80), w.get("offline_threshold", 95))
|
||||
return name, kind, status, None, "", f"cpu {pct}%"
|
||||
|
||||
if wtype == "system_memory":
|
||||
pct = mem_percent()
|
||||
status = _pct_status(pct, w.get("degrade_threshold", 85), w.get("offline_threshold", 97))
|
||||
return name, kind, status, None, "", f"mem {pct}%"
|
||||
|
||||
if wtype == "system_disk":
|
||||
path = w.get("path", "/")
|
||||
pct = disk_percent(path)
|
||||
status = _pct_status(pct, w.get("degrade_threshold", 80), w.get("offline_threshold", 95))
|
||||
return name, "system", status, None, "", f"disk {pct}% ({path})"
|
||||
|
||||
if wtype == "http_check":
|
||||
url = w["url"]
|
||||
timeout = w.get("timeout", 10)
|
||||
t0 = time.monotonic()
|
||||
try:
|
||||
req = urllib.request.Request(url, headers={"User-Agent": f"ethica-agent/{VERSION}"})
|
||||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||||
ok = r.status < 400
|
||||
except Exception:
|
||||
ok = False
|
||||
lat = int((time.monotonic() - t0) * 1000)
|
||||
return name, "endpoint", "online" if ok else "offline", lat, "", url
|
||||
|
||||
if wtype == "port_check":
|
||||
host = w["host"]
|
||||
port = int(w["port"])
|
||||
timeout = w.get("timeout", 5)
|
||||
t0 = time.monotonic()
|
||||
try:
|
||||
with socket.create_connection((host, port), timeout=timeout):
|
||||
ok = True
|
||||
except Exception:
|
||||
ok = False
|
||||
lat = int((time.monotonic() - t0) * 1000)
|
||||
return name, "endpoint", "online" if ok else "offline", lat, "", f"{host}:{port}"
|
||||
|
||||
if wtype == "process_check":
|
||||
proc = w["process"]
|
||||
alive = process_alive(proc)
|
||||
return name, "system", "online" if alive else "offline", None, "", proc
|
||||
|
||||
if wtype == "custom_cmd":
|
||||
cmd = w["cmd"]
|
||||
timeout = w.get("timeout", 30)
|
||||
try:
|
||||
r = subprocess.run(cmd, shell=True, timeout=timeout, capture_output=True)
|
||||
code = r.returncode
|
||||
except subprocess.TimeoutExpired:
|
||||
code = 2
|
||||
except Exception:
|
||||
code = 3
|
||||
# Convention: exit 0 = online, exit 1 = degraded, anything else = offline
|
||||
status = {0: "online", 1: "degraded"}.get(code, "offline")
|
||||
return name, kind, status, None, str(w.get("version", "")), str(w.get("note", ""))
|
||||
|
||||
raise ValueError(f"unknown widget type: {wtype!r}")
|
||||
|
||||
|
||||
# ── Push ───────────────────────────────────────────────────────────────────────
|
||||
|
||||
def push(home: str, token: str, name: str, kind: str, status: str,
|
||||
latency_ms, version: str, note: str) -> int:
|
||||
url = home.rstrip("/") + "/api/company/metrics"
|
||||
payload: dict = {"name": name, "kind": kind, "status": status,
|
||||
"version": version, "note": note}
|
||||
if latency_ms is not None:
|
||||
payload["latency_ms"] = latency_ms
|
||||
data = json.dumps(payload).encode()
|
||||
req = urllib.request.Request(
|
||||
url, data=data, method="POST",
|
||||
headers={"X-Company-Token": token,
|
||||
"Content-Type": "application/json",
|
||||
"User-Agent": f"ethica-agent/{VERSION}"})
|
||||
with urllib.request.urlopen(req, timeout=15) as r:
|
||||
return r.status
|
||||
|
||||
|
||||
# ── Main ───────────────────────────────────────────────────────────────────────
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Ethica Agent — push telemetry to Ethica.no home.")
|
||||
parser.add_argument("--config", default="ethica-agent.yml",
|
||||
help="path to config file (default: ethica-agent.yml)")
|
||||
parser.add_argument("--once", action="store_true",
|
||||
help="collect once and exit (useful for cron / testing)")
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
cfg = load_config(args.config)
|
||||
except FileNotFoundError:
|
||||
print(f"Config not found: {args.config}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
token = cfg.get("token", "")
|
||||
home = cfg.get("home", "https://ethica.no")
|
||||
interval = int(cfg.get("interval", 60))
|
||||
widgets = cfg.get("widgets", [])
|
||||
|
||||
if not token or token == "REPLACE_ME":
|
||||
print("Set 'token' in your config (Company → API in the admin panel).",
|
||||
file=sys.stderr)
|
||||
sys.exit(1)
|
||||
if not widgets:
|
||||
print("No widgets configured — nothing to do.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
print(f"Ethica Agent {VERSION} platform={SYSTEM} home={home}"
|
||||
f" widgets={len(widgets)} interval={interval}s")
|
||||
|
||||
def _run_once():
|
||||
for w in widgets:
|
||||
label = w.get("name", w.get("type", "?"))
|
||||
try:
|
||||
name, kind, status, lat, ver, note = run_widget(w)
|
||||
code = push(home, token, name, kind, status, lat, ver, note)
|
||||
print(f" [{status:<9}] {name} → {code}")
|
||||
except NotImplementedError as e:
|
||||
print(f" [SKIP ] {label}: {e}", file=sys.stderr)
|
||||
except (URLError, OSError) as e:
|
||||
print(f" [NET-ERR ] {label}: {e}", file=sys.stderr)
|
||||
except Exception as e:
|
||||
print(f" [ERROR ] {label}: {type(e).__name__}: {e}", file=sys.stderr)
|
||||
|
||||
if args.once:
|
||||
_run_once()
|
||||
return
|
||||
|
||||
while True:
|
||||
_run_once()
|
||||
time.sleep(interval)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user