feat: add node agent system with heartbeat collection, installer generation, and traffic flow telemetry
Add NodeAgent and TrafficFlow models to track agent status and network flows, implement /agents/heartbeat endpoint to receive interface counters, conntrack flows, firewall status, and nftables ruleset hash from agents, add nexafabric-agent.py Python script to collect host telemetry including VM/LXC interface hints via tap/fwbr regex matching, conntrack flow parsing with protocol/state/byte counters, and pve-firewall status checks,
This commit is contained in:
@@ -0,0 +1,210 @@
|
||||
#!/usr/bin/env python3
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import platform
|
||||
import re
|
||||
import socket
|
||||
import ssl
|
||||
import subprocess
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
VERSION = "0.1.0"
|
||||
VM_INTERFACE_RE = re.compile(r"(?:tap|fwbr|fwln|fwpr)(\d+)")
|
||||
|
||||
|
||||
def read_text(path: str) -> str | None:
|
||||
try:
|
||||
return Path(path).read_text(encoding="utf-8").strip()
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
|
||||
def run_command(args: list[str], timeout: int = 5) -> tuple[int, str]:
|
||||
try:
|
||||
result = subprocess.run(args, capture_output=True, text=True, timeout=timeout, check=False)
|
||||
return result.returncode, result.stdout.strip()
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
return 127, ""
|
||||
|
||||
|
||||
def collect_interfaces() -> list[dict[str, Any]]:
|
||||
interfaces = []
|
||||
for item in Path("/sys/class/net").iterdir() if Path("/sys/class/net").exists() else []:
|
||||
name = item.name
|
||||
if name == "lo":
|
||||
continue
|
||||
vm_match = VM_INTERFACE_RE.search(name)
|
||||
interfaces.append(
|
||||
{
|
||||
"name": name,
|
||||
"vmid": vm_match.group(1) if vm_match else None,
|
||||
"operstate": read_text(f"/sys/class/net/{name}/operstate") or "unknown",
|
||||
"rx_bytes": int(read_text(f"/sys/class/net/{name}/statistics/rx_bytes") or 0),
|
||||
"tx_bytes": int(read_text(f"/sys/class/net/{name}/statistics/tx_bytes") or 0),
|
||||
"rx_packets": int(read_text(f"/sys/class/net/{name}/statistics/rx_packets") or 0),
|
||||
"tx_packets": int(read_text(f"/sys/class/net/{name}/statistics/tx_packets") or 0),
|
||||
}
|
||||
)
|
||||
return interfaces
|
||||
|
||||
|
||||
def parse_conntrack_line(line: str) -> dict[str, Any] | None:
|
||||
parts = line.split()
|
||||
if len(parts) < 5 or parts[0] not in {"tcp", "udp", "icmp"}:
|
||||
return None
|
||||
protocol = parts[0]
|
||||
state = None
|
||||
if protocol == "tcp" and len(parts) > 3 and "=" not in parts[3]:
|
||||
state = parts[3]
|
||||
|
||||
values: dict[str, list[str]] = {}
|
||||
for part in parts:
|
||||
if "=" not in part:
|
||||
continue
|
||||
key, value = part.split("=", 1)
|
||||
values.setdefault(key, []).append(value)
|
||||
|
||||
src_values = values.get("src", [])
|
||||
dst_values = values.get("dst", [])
|
||||
if not src_values or not dst_values:
|
||||
return None
|
||||
|
||||
packet_values = [int(value) for value in values.get("packets", []) if value.isdigit()]
|
||||
byte_values = [int(value) for value in values.get("bytes", []) if value.isdigit()]
|
||||
sport = values.get("sport", [None])[0]
|
||||
dport = values.get("dport", [None])[0]
|
||||
return {
|
||||
"source_ip": src_values[0],
|
||||
"destination_ip": dst_values[0],
|
||||
"protocol": protocol,
|
||||
"source_port": int(sport) if sport and sport.isdigit() else None,
|
||||
"destination_port": int(dport) if dport and dport.isdigit() else None,
|
||||
"packets": sum(packet_values),
|
||||
"bytes": sum(byte_values),
|
||||
"state": state,
|
||||
}
|
||||
|
||||
|
||||
def collect_flows(limit: int = 500) -> list[dict[str, Any]]:
|
||||
code, output = run_command(["conntrack", "-L", "-o", "extended"], timeout=10)
|
||||
if code != 0 or not output:
|
||||
return []
|
||||
flows = []
|
||||
seen = set()
|
||||
for line in output.splitlines():
|
||||
flow = parse_conntrack_line(line)
|
||||
if not flow:
|
||||
continue
|
||||
key = (
|
||||
flow["source_ip"],
|
||||
flow["destination_ip"],
|
||||
flow["protocol"],
|
||||
flow.get("source_port"),
|
||||
flow.get("destination_port"),
|
||||
)
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
flows.append(flow)
|
||||
if len(flows) >= limit:
|
||||
break
|
||||
return flows
|
||||
|
||||
|
||||
def collect_conntrack() -> dict[str, Any]:
|
||||
code, output = run_command(["conntrack", "-C"])
|
||||
if code == 0 and output.isdigit():
|
||||
return {"count": int(output), "source": "conntrack"}
|
||||
for path in ("/proc/net/nf_conntrack", "/proc/net/ip_conntrack"):
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8", errors="ignore") as handle:
|
||||
return {"count": sum(1 for _ in handle), "source": path}
|
||||
except OSError:
|
||||
continue
|
||||
return {"count": None, "source": "unavailable"}
|
||||
|
||||
|
||||
def collect_firewall() -> dict[str, Any]:
|
||||
status = {}
|
||||
code, output = run_command(["systemctl", "is-active", "pve-firewall"])
|
||||
status["pve_firewall"] = output if code == 0 else "unknown"
|
||||
code, output = run_command(["nft", "-j", "list", "ruleset"], timeout=10)
|
||||
if code == 0 and output:
|
||||
status["nft_ruleset_sha256"] = hashlib.sha256(output.encode("utf-8")).hexdigest()
|
||||
else:
|
||||
status["nft_ruleset_sha256"] = None
|
||||
return status
|
||||
|
||||
|
||||
def collect_payload(config: dict[str, Any]) -> dict[str, Any]:
|
||||
uptime = read_text("/proc/uptime")
|
||||
return {
|
||||
"version": VERSION,
|
||||
"node_name": config.get("node_name"),
|
||||
"collected_at": datetime.now(timezone.utc).isoformat(),
|
||||
"hostname": socket.gethostname(),
|
||||
"kernel": platform.release(),
|
||||
"uptime_seconds": float(uptime.split()[0]) if uptime else None,
|
||||
"loadavg": list(os.getloadavg()) if hasattr(os, "getloadavg") else [],
|
||||
"interfaces": collect_interfaces(),
|
||||
"flows": collect_flows(int(config.get("flow_limit", 500))),
|
||||
"conntrack": collect_conntrack(),
|
||||
"firewall": collect_firewall(),
|
||||
"extra": {"platform": platform.platform()},
|
||||
}
|
||||
|
||||
|
||||
def post_heartbeat(config: dict[str, Any], payload: dict[str, Any]) -> None:
|
||||
api_url = str(config["api_url"]).rstrip("/")
|
||||
data = json.dumps(payload).encode("utf-8")
|
||||
request = urllib.request.Request(
|
||||
f"{api_url}/agents/heartbeat",
|
||||
data=data,
|
||||
headers={
|
||||
"Authorization": f"Bearer {config['token']}",
|
||||
"Content-Type": "application/json",
|
||||
"User-Agent": f"nexafabric-agent/{VERSION}",
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
context = None
|
||||
if not bool(config.get("verify_tls", True)):
|
||||
context = ssl._create_unverified_context()
|
||||
with urllib.request.urlopen(request, timeout=15, context=context) as response:
|
||||
response.read()
|
||||
|
||||
|
||||
def load_config(path: str) -> dict[str, Any]:
|
||||
return json.loads(Path(path).read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="NexaFabric Proxmox node telemetry agent")
|
||||
parser.add_argument("--config", default="/etc/nexafabric-agent/config.json")
|
||||
parser.add_argument("--once", action="store_true")
|
||||
args = parser.parse_args()
|
||||
config = load_config(args.config)
|
||||
interval = int(config.get("interval_seconds", 30))
|
||||
|
||||
while True:
|
||||
payload = collect_payload(config)
|
||||
try:
|
||||
post_heartbeat(config, payload)
|
||||
print(f"heartbeat ok: {payload['collected_at']}", flush=True)
|
||||
except (OSError, urllib.error.URLError, urllib.error.HTTPError) as exc:
|
||||
print(f"heartbeat failed: {exc}", flush=True)
|
||||
if args.once:
|
||||
return 0
|
||||
time.sleep(interval)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user