andreas
b3aa7b585f
- When os_info arrives with no owner field, apply default_owner from server config - Stop applying default_owner unconditionally in get_host_access (now deferred to os_info handling) - os_info plugin logs debug message when injecting owner from client config Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
143 lines
5.6 KiB
Python
143 lines
5.6 KiB
Python
"""OS Information Plugin for Heartbeat.
|
|
|
|
Collects static operating system information including OS name, version,
|
|
kernel, architecture, and distribution details.
|
|
"""
|
|
|
|
import platform
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
# Import from parent package
|
|
from hbd.client.plugin import InfoPlugin
|
|
|
|
|
|
class OSInfoPlugin(InfoPlugin):
|
|
"""Collect operating system information.
|
|
|
|
This plugin gathers static OS information that rarely changes:
|
|
- OS name and version
|
|
- Kernel version
|
|
- Architecture (x86_64, arm64, etc.)
|
|
- Distribution details (for Linux)
|
|
- Python version (used by hbc)
|
|
"""
|
|
|
|
name = "os_info"
|
|
version = "1.0.0"
|
|
description = "Operating system and platform information"
|
|
interval = 0 # InfoPlugin: collect once at startup
|
|
|
|
def __init__(self, config: Optional[Dict[str, Any]] = None):
|
|
super().__init__(config)
|
|
|
|
async def initialize(self) -> bool:
|
|
"""Initialize the OS info plugin.
|
|
|
|
Returns:
|
|
True (always succeeds - platform module is stdlib)
|
|
"""
|
|
self.logger.info(f"Initializing {self.name} plugin")
|
|
return True
|
|
|
|
async def _collect_info(self) -> Dict[str, Any]:
|
|
"""Collect OS information.
|
|
|
|
Returns:
|
|
Dictionary with OS details
|
|
"""
|
|
try:
|
|
from hbd import __version__ as hbc_version
|
|
data = {
|
|
"system": platform.system(), # e.g., "Linux", "Darwin", "Windows"
|
|
"node": platform.node(), # hostname
|
|
"release": platform.release(), # kernel version
|
|
"version": platform.version(), # detailed version
|
|
"machine": platform.machine(), # e.g., "x86_64", "arm64"
|
|
"processor": platform.processor(), # processor name
|
|
"architecture": platform.architecture()[0], # e.g., "64bit"
|
|
"python_version": platform.python_version(),
|
|
"python_implementation": platform.python_implementation(),
|
|
"hbc_version": hbc_version,
|
|
"hbc_type": "full",
|
|
}
|
|
if self.config.get("owner"):
|
|
self.logger.debug(f"Adding owner from config: {self.config['owner']}")
|
|
data["owner"] = self.config["owner"]
|
|
|
|
# Add Linux-specific distribution info
|
|
if platform.system() == "Linux":
|
|
data.update(self._get_linux_distro())
|
|
|
|
# Add macOS-specific info
|
|
elif platform.system() == "Darwin":
|
|
data["macos_version"] = platform.mac_ver()[0]
|
|
|
|
# Add Windows-specific info
|
|
elif platform.system() == "Windows":
|
|
win_ver = platform.win32_ver()
|
|
data["windows_release"] = win_ver[0]
|
|
data["windows_version"] = win_ver[1]
|
|
data["windows_sp"] = win_ver[2]
|
|
data["windows_type"] = win_ver[3]
|
|
|
|
self.logger.debug(f"Collected OS info: {data['system']} {data['release']}")
|
|
return data
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error collecting OS info: {e}", exc_info=True)
|
|
return {}
|
|
|
|
def _get_linux_distro(self) -> Dict[str, str]:
|
|
"""Get Linux distribution information.
|
|
|
|
Returns:
|
|
Dictionary with distribution details
|
|
"""
|
|
distro_info = {}
|
|
|
|
# Try reading /etc/os-release (standard on modern Linux)
|
|
os_release = Path("/etc/os-release")
|
|
if os_release.exists():
|
|
try:
|
|
with open(os_release) as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if "=" in line and not line.startswith("#"):
|
|
key, value = line.split("=", 1)
|
|
# Remove quotes from value
|
|
value = value.strip('"').strip("'")
|
|
# Map common keys
|
|
if key == "NAME":
|
|
distro_info["distro_name"] = value
|
|
elif key == "VERSION":
|
|
distro_info["distro_version"] = value
|
|
elif key == "ID":
|
|
distro_info["distro_id"] = value
|
|
elif key == "VERSION_ID":
|
|
distro_info["distro_version_id"] = value
|
|
elif key == "PRETTY_NAME":
|
|
distro_info["distro_pretty_name"] = value
|
|
except Exception as e:
|
|
self.logger.warning(f"Could not read /etc/os-release: {e}")
|
|
|
|
# Fallback: try lsb_release (older systems)
|
|
elif Path("/etc/lsb-release").exists():
|
|
try:
|
|
with open("/etc/lsb-release") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if "=" in line:
|
|
key, value = line.split("=", 1)
|
|
if key == "DISTRIB_ID":
|
|
distro_info["distro_id"] = value
|
|
elif key == "DISTRIB_RELEASE":
|
|
distro_info["distro_version"] = value
|
|
elif key == "DISTRIB_DESCRIPTION":
|
|
distro_info["distro_name"] = value
|
|
except Exception as e:
|
|
self.logger.warning(f"Could not read /etc/lsb-release: {e}")
|
|
|
|
return distro_info
|