openclaw/workspace/james/agents/watcher/watcher.py

#!/usr/bin/env python3
import json
import os
import shutil
import subprocess
import time
import uuid
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple

from james.lib.log import get_logger
from james.lib.nodes import node_name, node_role, register_node, registry_payload
from james.lib.schema import validate_event

log = get_logger("watcher")


@dataclass
class CpuSample:
    total: int
    idle: int


def read_cpu_sample() -> CpuSample:
    with open("/proc/stat", "r", encoding="utf-8") as handle:
        parts = handle.readline().split()
    values = list(map(int, parts[1:]))
    total = sum(values)
    idle = values[3] + values[4] if len(values) > 4 else values[3]
    return CpuSample(total=total, idle=idle)


def cpu_percent(prev: CpuSample, cur: CpuSample) -> float:
    total_delta = cur.total - prev.total
    idle_delta = cur.idle - prev.idle
    if total_delta <= 0:
        return 0.0
    return max(0.0, min(100.0, 100.0 * (total_delta - idle_delta) / total_delta))


def read_mem() -> Tuple[int, int, float]:
    mem_total = 0
    mem_available = 0
    with open("/proc/meminfo", "r", encoding="utf-8") as handle:
        for line in handle:
            if line.startswith("MemTotal:"):
                mem_total = int(line.split()[1])
            elif line.startswith("MemAvailable:"):
                mem_available = int(line.split()[1])
    mem_used = max(0, mem_total - mem_available)
    percent = (mem_used / mem_total * 100.0) if mem_total else 0.0
    return mem_total, mem_used, percent


def read_disk(path: str = "/") -> Tuple[int, int, float]:
    usage = shutil.disk_usage(path)
    used = usage.total - usage.free
    percent = (used / usage.total * 100.0) if usage.total else 0.0
    return usage.total, used, percent


def read_loadavg() -> Tuple[float, float, float]:
    return os.getloadavg()


def docker_containers() -> Dict[str, str]:
    try:
        result = subprocess.run(
            ["docker", "ps", "-a", "--format", "{{.ID}} {{.Names}} {{.State}}"],
            check=True,
            capture_output=True,
            text=True,
        )
    except Exception as exc:
        log.warning("docker ps failed", error=str(exc))
        return {}

    containers = {}
    for line in result.stdout.splitlines():
        parts = line.strip().split(maxsplit=2)
        if len(parts) != 3:
            continue
        _, name, state = parts
        containers[name] = state
    return containers


def gpu_stats() -> Optional[List[dict]]:
    if shutil.which("nvidia-smi") is None:
        return None
    try:
        result = subprocess.run(
            [
                "nvidia-smi",
                "--query-gpu=index,name,utilization.gpu,utilization.memory,memory.total,memory.used,temperature.gpu",
                "--format=csv,noheader,nounits",
            ],
            check=True,
            capture_output=True,
            text=True,
        )
    except Exception as exc:
        log.warning("nvidia-smi failed", error=str(exc))
        return None

    stats = []
    for line in result.stdout.splitlines():
        parts = [part.strip() for part in line.split(",")]
        if len(parts) < 7:
            continue
        index, name, util_gpu, util_mem, mem_total, mem_used, temp = parts[:7]
        try:
            total = float(mem_total)
            used = float(mem_used)
            vram_percent = round((used / total * 100.0) if total else 0.0, 2)
        except ValueError as exc:
            log.warning("gpu vram parse error", gpu=index, error=str(exc))
            vram_percent = 0.0
        stats.append(
            {
                "index": index,
                "name": name,
                "util_gpu_percent": float(util_gpu) if util_gpu else 0.0,
                "util_mem_percent": float(util_mem) if util_mem else 0.0,
                "vram_total_mb": float(mem_total) if mem_total else 0.0,
                "vram_used_mb": float(mem_used) if mem_used else 0.0,
                "vram_used_percent": vram_percent,
                "temp_c": float(temp) if temp else 0.0,
            }
        )
    return stats or None


def redis_url() -> str:
    url = os.environ.get("REDIS_URL")
    if url:
        return url
    host = os.environ.get("REDIS_HOST", "127.0.0.1")
    port = os.environ.get("REDIS_PORT", "6379")
    password = os.environ.get("REDIS_PASSWORD")
    auth = f":{password}@" if password else ""
    return f"redis://{auth}{host}:{port}/0"


def publish(
    r,
    channel: str,
    event: dict,
    rate_limit_window: float,
    rate_limit_severities: set,
    low_ttl_seconds: int,
) -> None:
    validate_event(event)

    if rate_limit_window > 0 and event.get("severity") in rate_limit_severities:
        key = f"james:rate:{channel}:{event['source']}:{event['type']}:{event['severity']}"
        if not r.set(key, "1", nx=True, ex=int(rate_limit_window)):
            log.debug("rate-limited", event_id=event.get("id"), channel=channel)
            return

    r.publish(channel, json.dumps(event))

    if low_ttl_seconds > 0 and event.get("severity") == "low":
        cache_key = f"james:low:{channel}:{event['source']}:{event['type']}"
        r.setex(cache_key, int(low_ttl_seconds), json.dumps(event))


def base_event(event_type: str, severity: str, payload: dict) -> dict:
    return {
        "id": f"evt-{time.strftime('%Y%m%d')}-{uuid.uuid4().hex[:8]}",
        "source": node_name(),
        "type": event_type,
        "timestamp": int(time.time()),
        "severity": severity,
        "payload": payload,
    }


def main() -> int:
    interval = float(os.environ.get("JAMES_INTERVAL", "30"))
    disk_threshold = float(os.environ.get("JAMES_DISK_WARN", "80"))
    mem_threshold = float(os.environ.get("JAMES_MEM_WARN", "85"))
    cpu_threshold = float(os.environ.get("JAMES_CPU_WARN", "90"))
    gpu_vram_threshold = float(os.environ.get("JAMES_GPU_VRAM_WARN", "90"))
    channel = os.environ.get("JAMES_CHANNEL", "james.events.system")
    role = node_role()
    rate_limit_window = float(os.environ.get("JAMES_RATE_LIMIT_WINDOW", "30"))
    rate_limit_severities = {
        item.strip()
        for item in os.environ.get("JAMES_RATE_LIMIT_SEVERITIES", "low,medium").split(",")
        if item.strip()
    }
    low_ttl_seconds = int(os.environ.get("JAMES_LOW_EVENT_TTL", "300"))
    registry_enabled = os.environ.get("JAMES_NODE_REGISTRY", "1").strip().lower() not in {
        "0",
        "false",
        "no",
        "off",
    }
    registry_ttl = int(os.environ.get("JAMES_NODE_REGISTRY_TTL", "180"))

    try:
        import redis
    except Exception as exc:  # pragma: no cover - runtime guard
        log.error("redis is required", exc_info=True)
        return 2

    url = redis_url()
    try:
        r = redis.Redis.from_url(url)
        r.ping()
    except Exception:
        log.error("redis connection failed", exc_info=True)
        return 1

    log.info("started", channel=channel, interval=interval, role=role)

    prev_cpu = read_cpu_sample()
    prev_containers = docker_containers()
    docker_available = shutil.which("docker") is not None
    gpu_available = shutil.which("nvidia-smi") is not None

    if registry_enabled:
        try:
            register_node(r, registry_payload(interval, docker_available, gpu_available), registry_ttl)
        except Exception:
            log.warning("node registry update failed", exc_info=True)

    while True:
        time.sleep(interval)

        try:
            if registry_enabled:
                try:
                    register_node(r, registry_payload(interval, docker_available, gpu_available), registry_ttl)
                except Exception:
                    log.warning("node registry update failed", exc_info=True)

            cur_cpu = read_cpu_sample()
            cpu = cpu_percent(prev_cpu, cur_cpu)
            prev_cpu = cur_cpu

            mem_total, mem_used, mem_percent = read_mem()
            disk_total, disk_used, disk_percent = read_disk("/")
            load1, load5, load15 = read_loadavg()

            containers = docker_containers()
            running = sum(1 for state in containers.values() if state == "running")
            stopped = sum(1 for state in containers.values() if state != "running")
            gpus = gpu_stats()

            payload = {
                "role": role,
                "cpu_percent": round(cpu, 2),
                "loadavg": [load1, load5, load15],
                "mem_total_kb": mem_total,
                "mem_used_kb": mem_used,
                "mem_used_percent": round(mem_percent, 2),
                "disk_total_bytes": disk_total,
                "disk_used_bytes": disk_used,
                "disk_used_percent": round(disk_percent, 2),
                "docker": {
                    "running": running,
                    "stopped": stopped,
                },
            }
            if gpus:
                payload["gpu"] = gpus

            publish(
                r,
                channel,
                base_event("system.metrics", "low", payload),
                rate_limit_window,
                rate_limit_severities,
                low_ttl_seconds,
            )

            if disk_percent >= disk_threshold:
                publish(
                    r,
                    channel,
                    base_event(
                        "disk.high_usage",
                        "high" if disk_percent >= 90 else "medium",
                        {"disk_used_percent": round(disk_percent, 2)},
                    ),
                    rate_limit_window,
                    rate_limit_severities,
                    low_ttl_seconds,
                )

            if mem_percent >= mem_threshold:
                publish(
                    r,
                    channel,
                    base_event(
                        "memory.high_usage",
                        "high" if mem_percent >= 92 else "medium",
                        {"mem_used_percent": round(mem_percent, 2)},
                    ),
                    rate_limit_window,
                    rate_limit_severities,
                    low_ttl_seconds,
                )

            if cpu >= cpu_threshold:
                publish(
                    r,
                    channel,
                    base_event(
                        "cpu.high_usage",
                        "high" if cpu >= 95 else "medium",
                        {"cpu_percent": round(cpu, 2)},
                    ),
                    rate_limit_window,
                    rate_limit_severities,
                    low_ttl_seconds,
                )

            if gpus:
                for gpu in gpus:
                    if gpu.get("vram_used_percent", 0.0) >= gpu_vram_threshold:
                        publish(
                            r,
                            channel,
                            base_event(
                                "gpu.vram_high",
                                "high" if gpu["vram_used_percent"] >= 95 else "medium",
                                {
                                    "index": gpu.get("index"),
                                    "name": gpu.get("name"),
                                    "vram_used_percent": gpu.get("vram_used_percent"),
                                    "vram_used_mb": gpu.get("vram_used_mb"),
                                    "vram_total_mb": gpu.get("vram_total_mb"),
                                },
                            ),
                            rate_limit_window,
                            rate_limit_severities,
                            low_ttl_seconds,
                        )

            for name, state in containers.items():
                prev = prev_containers.get(name)
                if prev and prev != state:
                    event_type = "container.stopped" if state != "running" else "container.running"
                    log.info("container state changed", container=name, prev=prev, state=state)
                    publish(
                        r,
                        channel,
                        base_event(
                            event_type,
                            "medium",
                            {"container": name, "state": state},
                        ),
                        rate_limit_window,
                        rate_limit_severities,
                        low_ttl_seconds,
                    )
            prev_containers = containers

        except Exception:
            log.error("collection loop error", exc_info=True)


if __name__ == "__main__":
    raise SystemExit(main())