98 lines
2.8 KiB
Python
98 lines
2.8 KiB
Python
"""Read-only host discovery used during registration and heartbeat."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import platform
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from . import __version__
|
|
|
|
|
|
def _ip_address() -> Optional[str]:
|
|
try:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
sock.connect(("192.0.2.1", 9))
|
|
return str(sock.getsockname()[0])
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
def _tool_version(command: str, args: List[str]) -> Optional[str]:
|
|
executable = shutil.which(command)
|
|
if not executable:
|
|
return None
|
|
try:
|
|
result = subprocess.run(
|
|
[executable] + args,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=3,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return None
|
|
output = (result.stdout or result.stderr).strip().splitlines()
|
|
return output[0][:128] if output else None
|
|
|
|
|
|
def discover_toolchain() -> Dict[str, str]:
|
|
probes = {
|
|
"fio": ("fio", ["--version"]),
|
|
"nvme-cli": ("nvme", ["version"]),
|
|
"smartctl": ("smartctl", ["--version"]),
|
|
}
|
|
versions: Dict[str, str] = {}
|
|
for name, (command, args) in probes.items():
|
|
version = _tool_version(command, args)
|
|
if version:
|
|
versions[name] = version
|
|
return versions
|
|
|
|
|
|
def system_device_paths() -> List[str]:
|
|
if platform.system().lower() != "linux":
|
|
return []
|
|
try:
|
|
for line in Path("/proc/mounts").read_text(encoding="utf-8").splitlines():
|
|
fields = line.split()
|
|
if len(fields) >= 2 and fields[1] == "/" and fields[0].startswith("/dev/"):
|
|
return [os.path.realpath(fields[0])]
|
|
except OSError:
|
|
return []
|
|
return []
|
|
|
|
|
|
def registration_payload(
|
|
*,
|
|
host_id: Optional[str] = None,
|
|
host_name: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
return {
|
|
"host_id": host_id,
|
|
"host_name": host_name or platform.node() or socket.gethostname(),
|
|
"agent_version": __version__,
|
|
"os_family": platform.system().lower() or "unknown",
|
|
"os_version": platform.platform(),
|
|
"kernel": platform.release(),
|
|
"cpu": platform.processor() or None,
|
|
"motherboard": None,
|
|
"bios_version": None,
|
|
"ip_address": _ip_address(),
|
|
"toolchain": discover_toolchain(),
|
|
"labels": {"system_device_paths": system_device_paths()},
|
|
}
|
|
|
|
|
|
def heartbeat_payload() -> Dict[str, Any]:
|
|
return {
|
|
"healthy": True,
|
|
"agent_version": __version__,
|
|
"ip_address": _ip_address(),
|
|
"toolchain": discover_toolchain(),
|
|
"device_probe": {"system_device_paths": system_device_paths()},
|
|
}
|