f92e161ec6
The aggregate containers_not_running event carried service=None, which the observer skips when building service state and incidents — stability-agent's flagship signal never opened an incident (recon D15). Emit one event per non-running container instead, tagged with the compose service name from the com.docker.compose.service label (same pattern as node-agent's _canonical_container_name fix from May), falling back to the container name with Docker's stale-state hash prefix stripped; never crashes on unlabeled containers. 'created' compose tracking artifacts are skipped — they are not running services and would open fake incidents now that the event is actionable. Adds the service's first test suite covering the label-extraction helper. Smoke-run performed with runtime paths redirected (no docker build, authoring only): main loop runs, service names resolve on live solaria containers. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
386 lines
14 KiB
Python
386 lines
14 KiB
Python
import os
|
|
import time
|
|
import json
|
|
import datetime
|
|
import uuid
|
|
import socket
|
|
import shutil
|
|
import http.client
|
|
|
|
# Configuration from environment
|
|
CHECK_INTERVAL = int(os.environ.get("STABILITY_CHECK_INTERVAL", "60"))
|
|
DISK_THRESHOLD_PCT = float(os.environ.get("DISK_THRESHOLD_PCT", "90.0"))
|
|
MQTT_HOST = os.environ.get("MQTT_HOST")
|
|
MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883"))
|
|
NODE_NAME = os.environ.get("NODE_NAME", "chelsty")
|
|
REDIS_HOST = os.environ.get("REDIS_HOST")
|
|
REDIS_PORT = int(os.environ.get("REDIS_PORT", "6379"))
|
|
REDIS_ENABLED = os.environ.get("REDIS_ENABLED", "true").lower() == "true" if REDIS_HOST else False
|
|
SOURCE = "stability-agent"
|
|
|
|
STATE_DIR = "/opt/homelab/state"
|
|
EVENTS_BASE_DIR = "/opt/homelab/events"
|
|
HEARTBEAT_FILE = os.path.join(STATE_DIR, "stability-agent.heartbeat")
|
|
STATUS_FILE = os.path.join(STATE_DIR, "stability-agent.json")
|
|
|
|
def get_timestamp():
|
|
return datetime.datetime.utcnow().isoformat() + "Z"
|
|
|
|
def get_datestamp():
|
|
return datetime.datetime.utcnow().strftime("%Y-%m-%d")
|
|
|
|
def emit_event(event_type, severity, message, service=None, details=None):
|
|
timestamp = get_timestamp()
|
|
event = {
|
|
"id": str(uuid.uuid4()),
|
|
"timestamp": timestamp,
|
|
"node": NODE_NAME,
|
|
"source": SOURCE,
|
|
"type": event_type,
|
|
"severity": severity,
|
|
"message": message,
|
|
"details": details or {}
|
|
}
|
|
if service:
|
|
event["service"] = service
|
|
|
|
date_str = get_datestamp()
|
|
event_dir = os.path.join(EVENTS_BASE_DIR, date_str, NODE_NAME)
|
|
try:
|
|
os.makedirs(event_dir, exist_ok=True)
|
|
event_file = os.path.join(event_dir, "events.jsonl")
|
|
with open(event_file, "a") as f:
|
|
f.write(json.dumps(event) + "\n")
|
|
except Exception as e:
|
|
print(f"Failed to write event to filesystem: {e}")
|
|
|
|
# Redis publishing
|
|
if REDIS_ENABLED and redis_client:
|
|
try:
|
|
redis_client.xadd("homelab:events", {
|
|
"node": NODE_NAME,
|
|
"type": event_type,
|
|
"severity": severity,
|
|
"timestamp": str(int(time.time())),
|
|
"message": message,
|
|
"details": json.dumps(details or {})
|
|
})
|
|
except Exception as e:
|
|
print(f"Failed to publish event to Redis: {e}")
|
|
# Do not crash, already logged to filesystem
|
|
|
|
print(f"[{severity}] {message}")
|
|
|
|
def check_disk():
|
|
total, used, free = shutil.disk_usage("/")
|
|
percent = (used / total) * 100
|
|
details = {
|
|
"total_gb": total // (2**30),
|
|
"used_gb": used // (2**30),
|
|
"free_gb": free // (2**30),
|
|
"percent": round(percent, 2)
|
|
}
|
|
|
|
if percent > DISK_THRESHOLD_PCT:
|
|
emit_event("disk_usage_high", "warning", f"Disk usage is high: {details['percent']}%", details=details)
|
|
|
|
return details
|
|
|
|
class DockerClient:
|
|
def __init__(self, socket_path="/var/run/docker.sock"):
|
|
self.socket_path = socket_path
|
|
|
|
def _request(self, path):
|
|
class UnixHTTPConnection(http.client.HTTPConnection):
|
|
def __init__(self, socket_path):
|
|
super().__init__("localhost")
|
|
self.socket_path = socket_path
|
|
def connect(self):
|
|
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
self.sock.settimeout(5.0)
|
|
self.sock.connect(self.socket_path)
|
|
|
|
if not os.path.exists(self.socket_path):
|
|
return None
|
|
|
|
conn = UnixHTTPConnection(self.socket_path)
|
|
try:
|
|
conn.request("GET", path)
|
|
res = conn.getresponse()
|
|
if res.status == 200:
|
|
return json.loads(res.read().decode())
|
|
return None
|
|
except Exception as e:
|
|
print(f"Docker API error: {e}")
|
|
return None
|
|
finally:
|
|
conn.close()
|
|
|
|
def get_containers(self):
|
|
return self._request("/containers/json?all=1")
|
|
|
|
def container_service_name(container):
|
|
"""Return the compose service name for a /containers/json entry.
|
|
|
|
Priority (same pattern as node-agent's _canonical_container_name):
|
|
1. com.docker.compose.service label — the clean compose-file key, immune
|
|
to the "<12-hex>_" prefix Docker uses for stale project-state entries.
|
|
2. Container name with that stale-state prefix stripped — fallback for
|
|
non-Compose containers.
|
|
Never raises: missing/None Labels or Names degrade to "unknown".
|
|
"""
|
|
labels = container.get("Labels") or {}
|
|
if isinstance(labels, dict):
|
|
compose_svc = (labels.get("com.docker.compose.service") or "").strip()
|
|
if compose_svc:
|
|
return compose_svc
|
|
names = container.get("Names") or []
|
|
name = names[0].lstrip("/") if names else ""
|
|
if (len(name) > 13
|
|
and name[12] == "_"
|
|
and all(ch in "0123456789abcdef" for ch in name[:12])):
|
|
name = name[13:]
|
|
return name or "unknown"
|
|
|
|
def check_docker():
|
|
client = DockerClient()
|
|
if not os.path.exists(client.socket_path):
|
|
return {"status": "unavailable", "message": "Docker socket not found"}
|
|
|
|
containers = client.get_containers()
|
|
if containers is None:
|
|
emit_event("docker_api_error", "warning", "Could not connect to Docker socket API")
|
|
return {"status": "error", "error": "Could not connect to Docker socket API"}
|
|
|
|
summary = []
|
|
unhealthy_containers = []
|
|
for c in containers:
|
|
state = c.get("State", "")
|
|
status = c.get("Status", "")
|
|
name = c.get("Names", ["unknown"])[0].lstrip("/")
|
|
|
|
container_info = {
|
|
"name": name,
|
|
"service": container_service_name(c),
|
|
"state": state,
|
|
"status": status
|
|
}
|
|
summary.append(container_info)
|
|
|
|
# "created" containers are Docker Compose internal tracking artifacts
|
|
# (never started) — not a running service, nothing to remediate.
|
|
if state != "running" and state != "created":
|
|
unhealthy_containers.append(container_info)
|
|
|
|
# One event per container, tagged with its compose service name. The
|
|
# observer keys service state and incidents on event["service"] and skips
|
|
# events without it — the old aggregate event (service=None, all names in
|
|
# one message) never opened an incident (recon D15).
|
|
for info in unhealthy_containers:
|
|
emit_event(
|
|
"containers_not_running", "warning",
|
|
f"Container '{info['name']}' is not running (state={info['state']})",
|
|
service=info["service"],
|
|
details={"container": info},
|
|
)
|
|
|
|
return {"status": "ok", "containers": summary}
|
|
|
|
def check_tailscale():
|
|
# Check for tailscale socket or interface
|
|
socket_path = "/var/run/tailscale/tailscaled.sock"
|
|
socket_available = os.path.exists(socket_path)
|
|
interface_available = os.path.exists("/sys/class/net/tailscale0")
|
|
|
|
return {
|
|
"available": socket_available or interface_available,
|
|
"details": {
|
|
"socket": socket_available,
|
|
"interface": interface_available
|
|
}
|
|
}
|
|
|
|
def check_mqtt():
|
|
if not MQTT_HOST:
|
|
return {"configured": False}
|
|
|
|
try:
|
|
with socket.create_connection((MQTT_HOST, MQTT_PORT), timeout=5):
|
|
return {"configured": True, "reachable": True}
|
|
except Exception as e:
|
|
emit_event("mqtt_unreachable", "error", f"MQTT broker at {MQTT_HOST}:{MQTT_PORT} is unreachable", details={"error": str(e)})
|
|
return {"configured": True, "reachable": False, "error": str(e)}
|
|
|
|
class RedisClient:
|
|
def __init__(self, host, port=6379):
|
|
self.host = host
|
|
self.port = port
|
|
self.sock = None
|
|
|
|
def _connect(self):
|
|
if self.sock:
|
|
try:
|
|
# Check if socket is still alive
|
|
self.sock.send(b"", socket.MSG_DONTWAIT)
|
|
return True
|
|
except (socket.error, AttributeError):
|
|
self.sock = None
|
|
|
|
try:
|
|
self.sock = socket.create_connection((self.host, self.port), timeout=2)
|
|
self.sock.settimeout(2.0)
|
|
return True
|
|
except Exception as e:
|
|
self.sock = None
|
|
print(f"Redis connection error: {e}")
|
|
return False
|
|
|
|
def _send_command(self, *args):
|
|
if not self._connect():
|
|
return False
|
|
|
|
# RESP array
|
|
cmd = f"*{len(args)}\r\n"
|
|
for arg in args:
|
|
s_arg = str(arg)
|
|
cmd += f"${len(s_arg.encode('utf-8'))}\r\n{s_arg}\r\n"
|
|
|
|
try:
|
|
self.sock.sendall(cmd.encode('utf-8'))
|
|
# Basic response reading
|
|
resp = self.sock.recv(4096)
|
|
if resp.startswith(b"-"):
|
|
print(f"Redis error response: {resp.decode().strip()}")
|
|
return False
|
|
return True
|
|
except Exception as e:
|
|
print(f"Redis send error: {e}")
|
|
if self.sock:
|
|
self.sock.close()
|
|
self.sock = None
|
|
return False
|
|
|
|
def hset(self, key, mapping):
|
|
args = ["HSET", key]
|
|
for k, v in mapping.items():
|
|
args.extend([k, v])
|
|
return self._send_command(*args)
|
|
|
|
def xadd(self, key, fields):
|
|
args = ["XADD", key, "*"]
|
|
for k, v in fields.items():
|
|
args.extend([k, v])
|
|
return self._send_command(*args)
|
|
|
|
redis_client = RedisClient(REDIS_HOST, REDIS_PORT) if REDIS_ENABLED else None
|
|
|
|
def main():
|
|
print(f"Starting stability-agent on {NODE_NAME}...")
|
|
|
|
# Ensure directories exist
|
|
os.makedirs(STATE_DIR, exist_ok=True)
|
|
os.makedirs(EVENTS_BASE_DIR, exist_ok=True)
|
|
|
|
while True:
|
|
try:
|
|
status = {
|
|
"timestamp": get_timestamp(),
|
|
"node": NODE_NAME,
|
|
"checks": {}
|
|
}
|
|
|
|
status["checks"]["disk"] = check_disk()
|
|
status["checks"]["docker"] = check_docker()
|
|
status["checks"]["tailscale"] = check_tailscale()
|
|
status["checks"]["mqtt"] = check_mqtt()
|
|
|
|
# Zigbee2MQTT container check
|
|
z2m_present = False
|
|
z2m_running = False
|
|
if status["checks"]["docker"]["status"] == "ok":
|
|
for c in status["checks"]["docker"]["containers"]:
|
|
if "zigbee2mqtt" in c["name"]:
|
|
z2m_present = True
|
|
if c["state"] == "running":
|
|
z2m_running = True
|
|
|
|
status["checks"]["zigbee2mqtt"] = {
|
|
"present": z2m_present,
|
|
"running": z2m_running
|
|
}
|
|
|
|
# Write heartbeat
|
|
with open(HEARTBEAT_FILE, "w") as f:
|
|
f.write(get_timestamp())
|
|
|
|
# Write status summary
|
|
with open(STATUS_FILE, "w") as f:
|
|
json.dump(status, f, indent=2)
|
|
|
|
# Redis publishing
|
|
if REDIS_ENABLED and redis_client:
|
|
try:
|
|
# Node state
|
|
node_health = "healthy"
|
|
for check in status["checks"].values():
|
|
if isinstance(check, dict) and check.get("status") == "error":
|
|
node_health = "unhealthy"
|
|
|
|
# Redis publishing for node state
|
|
redis_client.hset(f"homelab:nodes:{NODE_NAME}", {
|
|
"id": NODE_NAME,
|
|
"hostname": NODE_NAME,
|
|
"health": node_health,
|
|
"status": "online",
|
|
"last_seen": status["timestamp"],
|
|
"capabilities": json.dumps(["docker", "tailscale", "mqtt", "disk"]),
|
|
"checks": json.dumps(status["checks"])
|
|
})
|
|
|
|
# Always publish stability-agent itself as a service
|
|
redis_client.hset(f"homelab:services:{NODE_NAME}:stability-agent", {
|
|
"name": "stability-agent",
|
|
"node": NODE_NAME,
|
|
"health": "healthy",
|
|
"desired_state": "running",
|
|
"actual_state": "running",
|
|
"deployment_state": "deployed",
|
|
"updated_at": status["timestamp"],
|
|
"dependencies": json.dumps([]),
|
|
"recommendations": json.dumps([])
|
|
})
|
|
|
|
# Services discovered from Docker
|
|
if status["checks"]["docker"]["status"] == "ok":
|
|
for c in status["checks"]["docker"]["containers"]:
|
|
service_name = c["name"]
|
|
if service_name == "stability-agent":
|
|
continue # Already published above
|
|
|
|
service_health = "healthy" if c["state"] == "running" else "unhealthy"
|
|
|
|
redis_client.hset(f"homelab:services:{NODE_NAME}:{service_name}", {
|
|
"name": service_name,
|
|
"node": NODE_NAME,
|
|
"health": service_health,
|
|
"desired_state": "running",
|
|
"actual_state": c["state"],
|
|
"deployment_state": "deployed",
|
|
"updated_at": status["timestamp"],
|
|
"dependencies": json.dumps([]),
|
|
"recommendations": json.dumps([])
|
|
})
|
|
except Exception as e:
|
|
print(f"Failed to publish to Redis: {e}")
|
|
# Local event for Redis error
|
|
emit_event("redis_publish_error", "warning", f"Failed to publish to Redis: {e}", details={"error": str(e)})
|
|
|
|
except Exception as e:
|
|
print(f"Error in main loop: {e}")
|
|
emit_event("agent_error", "error", f"Internal agent error: {e}", details={"error": str(e)})
|
|
|
|
time.sleep(CHECK_INTERVAL)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|