resource_monitor.py

from __future__ import annotations

import hashlib
import json
import math
import multiprocessing as mp
import os
import platform
import queue
import threading
import time
import urllib.request
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

GB = 1024 ** 3
MB = 1024 ** 2


def _read_text(path: str) -> str | None:
    try:
        candidate = Path(path)
        if candidate.exists():
            return candidate.read_text(encoding="utf-8", errors="replace").strip()
    except OSError:
        return None
    return None


def _safe_float(value: Any, default: float = 0.0) -> float:
    try:
        return float(value)
    except (TypeError, ValueError):
        return default


def _cpu_model_name() -> str:
    cpuinfo = _read_text("/proc/cpuinfo") or ""
    for line in cpuinfo.splitlines():
        key, separator, value = line.partition(":")
        if separator and key.strip().lower() in {"model name", "hardware", "processor"}:
            clean = value.strip()
            if clean and not clean.isdigit():
                return clean
    return platform.processor() or platform.machine() or "Unknown CPU"


def _cgroup_cpu_quota() -> tuple[float | None, str, str]:
    cpu_max = _read_text("/sys/fs/cgroup/cpu.max")
    if cpu_max:
        parts = cpu_max.split()
        if len(parts) >= 2:
            raw = f"{parts[0]} {parts[1]}"
            if parts[0] == "max":
                return None, "cgroup_v2_cpu.max", raw
            try:
                quota = int(parts[0])
                period = int(parts[1])
                if quota > 0 and period > 0:
                    return quota / period, "cgroup_v2_cpu.max", raw
            except (ValueError, ZeroDivisionError):
                pass

    quota_text = _read_text("/sys/fs/cgroup/cpu/cpu.cfs_quota_us")
    period_text = _read_text("/sys/fs/cgroup/cpu/cpu.cfs_period_us")
    if quota_text and period_text:
        raw = f"{quota_text} {period_text}"
        try:
            quota = int(quota_text)
            period = int(period_text)
            if quota < 0:
                return None, "cgroup_v1_cpu", raw
            if quota > 0 and period > 0:
                return quota / period, "cgroup_v1_cpu", raw
        except (ValueError, ZeroDivisionError):
            pass
    return None, "unavailable", "unavailable"


def _cgroup_memory_limit() -> tuple[int | None, str, str]:
    for path, source in (
        ("/sys/fs/cgroup/memory.max", "cgroup_v2_memory.max"),
        ("/sys/fs/cgroup/memory/memory.limit_in_bytes", "cgroup_v1_memory.limit"),
    ):
        text = _read_text(path)
        if not text:
            continue
        if text == "max":
            return None, source, text
        try:
            value = int(text)
        except ValueError:
            continue
        # Some cgroup-v1 hosts expose an enormous sentinel instead of "max".
        if value >= (1 << 60):
            return None, source, text
        if value > 0:
            return value, source, text
    return None, "unavailable", "unavailable"


def _visible_total_ram() -> int:
    try:
        import psutil  # type: ignore

        return int(psutil.virtual_memory().total)
    except Exception:
        meminfo = _read_text("/proc/meminfo") or ""
        for line in meminfo.splitlines():
            if line.startswith("MemTotal:"):
                try:
                    return int(line.split()[1]) * 1024
                except (IndexError, ValueError):
                    break
    return 0


def _memory_current() -> tuple[int, str]:
    for path, source in (
        ("/sys/fs/cgroup/memory.current", "cgroup_v2_memory.current"),
        ("/sys/fs/cgroup/memory/memory.usage_in_bytes", "cgroup_v1_memory.usage_in_bytes"),
    ):
        text = _read_text(path)
        if text:
            try:
                return max(0, int(text)), source
            except ValueError:
                pass
    try:
        import psutil  # type: ignore

        return int(psutil.Process(os.getpid()).memory_info().rss), "process_rss_fallback"
    except Exception:
        return 0, "unavailable"


@dataclass(frozen=True)
class HardwareSnapshot:
    cpu_model: str
    logical_cpu_visible: int
    affinity_cpu_count: int
    cpu_quota_vcpu: float | None
    effective_vcpu: float
    cpu_limit_source: str
    cpu_limit_raw: str
    visible_ram_bytes: int
    cgroup_ram_limit_bytes: int | None
    effective_ram_bytes: int
    ram_limit_source: str
    ram_limit_raw: str
    platform: str
    python_version: str


@dataclass(frozen=True)
class ResourceSnapshot:
    """Latest real CloudVaultHub container resource reading.

    cpu_used_vcpu is average CPU seconds divided by wall seconds. For example,
    2.50 means that the container used the equivalent of 2.5 CPU cores during
    the sample. It is not a simulated percentage.
    """

    cpu_used_vcpu: float
    ram_used_bytes: int
    ram_used_gb: float
    effective_vcpu: float
    logical_cpu_visible: int
    effective_ram_bytes: int
    effective_ram_gb: float
    sample_seconds: float
    updated_at_unix: float
    cpu_source: str
    ram_source: str


class SystemResourceMonitor:
    def __init__(self, refresh_seconds: float = 60.0) -> None:
        self.refresh_seconds = max(5.0, float(refresh_seconds))
        self._lock = threading.RLock()
        self._stop = threading.Event()
        self._hardware = self._detect_hardware()
        self._snapshot = ResourceSnapshot(
            cpu_used_vcpu=0.0,
            ram_used_bytes=0,
            ram_used_gb=0.0,
            effective_vcpu=self._hardware.effective_vcpu,
            logical_cpu_visible=self._hardware.logical_cpu_visible,
            effective_ram_bytes=self._hardware.effective_ram_bytes,
            effective_ram_gb=round(self._hardware.effective_ram_bytes / GB, 3),
            sample_seconds=0.0,
            updated_at_unix=time.time(),
            cpu_source="waiting_for_first_sample",
            ram_source="waiting_for_first_sample",
        )
        self._print_startup_report()
        self._thread = threading.Thread(
            target=self._run,
            daemon=True,
            name="cloudvault-resource-monitor",
        )
        self._thread.start()

    def _detect_hardware(self) -> HardwareSnapshot:
        visible = max(1, os.cpu_count() or 1)
        try:
            affinity = max(1, len(os.sched_getaffinity(0)))
        except (AttributeError, OSError):
            affinity = visible
        quota, cpu_source, cpu_raw = _cgroup_cpu_quota()
        candidates = [float(visible), float(affinity)]
        if quota is not None:
            candidates.append(float(quota))
        effective_vcpu = max(0.01, min(candidates))

        visible_ram = _visible_total_ram()
        ram_limit, ram_source, ram_raw = _cgroup_memory_limit()
        ram_candidates = [value for value in (visible_ram, ram_limit) if value and value > 0]
        effective_ram = min(ram_candidates) if ram_candidates else max(0, visible_ram)

        return HardwareSnapshot(
            cpu_model=_cpu_model_name(),
            logical_cpu_visible=visible,
            affinity_cpu_count=affinity,
            cpu_quota_vcpu=round(quota, 3) if quota is not None else None,
            effective_vcpu=round(effective_vcpu, 3),
            cpu_limit_source=cpu_source,
            cpu_limit_raw=cpu_raw,
            visible_ram_bytes=visible_ram,
            cgroup_ram_limit_bytes=ram_limit,
            effective_ram_bytes=effective_ram,
            ram_limit_source=ram_source,
            ram_limit_raw=ram_raw,
            platform=platform.platform(),
            python_version=platform.python_version(),
        )

    def _print_startup_report(self) -> None:
        h = self._hardware
        cpu_quota = "unlimited/not exposed" if h.cpu_quota_vcpu is None else f"{h.cpu_quota_vcpu:.2f} vCPU"
        ram_limit = "unlimited/not exposed" if h.cgroup_ram_limit_bytes is None else f"{h.cgroup_ram_limit_bytes / GB:.2f} GB"
        print("\n========== CLOUDVAULTHUB v172 HARDWARE ==========")
        print(f"CPU model              : {h.cpu_model}")
        print(f"Logical CPUs visible   : {h.logical_cpu_visible}")
        print(f"CPU affinity count     : {h.affinity_cpu_count}")
        print(f"cgroup CPU quota       : {cpu_quota} ({h.cpu_limit_source}: {h.cpu_limit_raw})")
        print(f"Effective CPU capacity : {h.effective_vcpu:.2f} vCPU")
        print(f"Visible RAM            : {h.visible_ram_bytes / GB:.2f} GB")
        print(f"cgroup RAM limit       : {ram_limit} ({h.ram_limit_source}: {h.ram_limit_raw})")
        print(f"Effective RAM capacity : {h.effective_ram_bytes / GB:.2f} GB")
        print(f"Platform               : {h.platform}")
        print(f"Python                 : {h.python_version}")
        print("=================================================\n", flush=True)

    def hardware(self) -> HardwareSnapshot:
        return self._hardware

    def _cpu_counter_seconds(self) -> tuple[float, str]:
        cpu_stat = _read_text("/sys/fs/cgroup/cpu.stat")
        if cpu_stat:
            for line in cpu_stat.splitlines():
                key, _, value = line.partition(" ")
                if key == "usage_usec":
                    try:
                        return int(value.strip()) / 1_000_000.0, "cgroup_v2_cpu.stat"
                    except ValueError:
                        break
        cpuacct = _read_text("/sys/fs/cgroup/cpuacct/cpuacct.usage")
        if cpuacct:
            try:
                return int(cpuacct) / 1_000_000_000.0, "cgroup_v1_cpuacct"
            except ValueError:
                pass
        try:
            process_times = os.times()
            return float(process_times.user + process_times.system), "process_times_fallback"
        except (AttributeError, OSError):
            return 0.0, "unavailable"

    def _store_sample(
        self,
        previous_cpu: float,
        previous_wall: float,
        current_cpu: float,
        current_wall: float,
        cpu_source: str,
    ) -> None:
        elapsed = max(0.001, current_wall - previous_wall)
        cpu_delta = max(0.0, current_cpu - previous_cpu)
        used_vcpu = cpu_delta / elapsed
        used_vcpu = min(max(0.0, used_vcpu), max(self._hardware.effective_vcpu, 0.01))
        ram_bytes, ram_source = _memory_current()
        snapshot = ResourceSnapshot(
            cpu_used_vcpu=round(used_vcpu, 3),
            ram_used_bytes=ram_bytes,
            ram_used_gb=round(ram_bytes / GB, 3),
            effective_vcpu=self._hardware.effective_vcpu,
            logical_cpu_visible=self._hardware.logical_cpu_visible,
            effective_ram_bytes=self._hardware.effective_ram_bytes,
            effective_ram_gb=round(self._hardware.effective_ram_bytes / GB, 3),
            sample_seconds=round(elapsed, 3),
            updated_at_unix=time.time(),
            cpu_source=cpu_source,
            ram_source=ram_source,
        )
        with self._lock:
            self._snapshot = snapshot

    def _run(self) -> None:
        previous_cpu, source = self._cpu_counter_seconds()
        previous_wall = time.monotonic()
        if self._stop.wait(1.0):
            return
        current_cpu, source = self._cpu_counter_seconds()
        current_wall = time.monotonic()
        self._store_sample(previous_cpu, previous_wall, current_cpu, current_wall, source)
        previous_cpu, previous_wall = current_cpu, current_wall
        while not self._stop.wait(self.refresh_seconds):
            current_cpu, source = self._cpu_counter_seconds()
            current_wall = time.monotonic()
            self._store_sample(previous_cpu, previous_wall, current_cpu, current_wall, source)
            previous_cpu, previous_wall = current_cpu, current_wall

    def snapshot(self) -> ResourceSnapshot:
        with self._lock:
            return self._snapshot

    def as_dict(self) -> dict[str, Any]:
        return asdict(self.snapshot())

    def stop(self) -> None:
        self._stop.set()


def _cpu_benchmark_worker(
    duration_seconds: float,
    stop_event: Any,
    output_queue: Any,
) -> None:
    started_wall = time.perf_counter()
    started_cpu = time.process_time()
    deadline = started_wall + max(0.5, duration_seconds)
    payload = b"CloudVaultHub-v172-CPU-benchmark"
    iterations = 0
    try:
        while time.perf_counter() < deadline and not stop_event.is_set():
            # 128 hashes per cancellation check keeps the worker responsive.
            for _ in range(128):
                payload = hashlib.sha256(payload).digest()
            iterations += 128
        output_queue.put(
            {
                "iterations": iterations,
                "cpu_seconds": max(0.0, time.process_time() - started_cpu),
                "wall_seconds": max(0.001, time.perf_counter() - started_wall),
            }
        )
    except BaseException as exc:  # child process must always report completion
        try:
            output_queue.put({"error": str(exc), "iterations": iterations})
        except Exception:
            pass


@dataclass
class BenchmarkState:
    status: str = "idle"
    stage: str = "Ready"
    progress: int = 0
    started_at: str | None = None
    finished_at: str | None = None
    cancelled: bool = False
    error: str = ""
    result: dict[str, Any] = field(default_factory=dict)


class SystemBenchmarkManager:
    """One safe, cancellable Admin benchmark for CPU, RAM and HTTP latency."""

    def __init__(self, monitor: SystemResourceMonitor) -> None:
        self.monitor = monitor
        self._lock = threading.RLock()
        self._stop = threading.Event()
        self._mp_stop: Any = None
        self._thread: threading.Thread | None = None
        self._children: list[Any] = []
        self._state = BenchmarkState()

    def _set(self, **updates: Any) -> None:
        with self._lock:
            for key, value in updates.items():
                setattr(self._state, key, value)

    def state(self) -> dict[str, Any]:
        with self._lock:
            payload = asdict(self._state)
        payload["hardware"] = asdict(self.monitor.hardware())
        payload["live"] = asdict(self.monitor.snapshot())
        return payload

    def start(self, ping_url: str) -> tuple[bool, str]:
        with self._lock:
            if self._thread and self._thread.is_alive():
                return False, "A hardware measurement is already running."
            self._stop = threading.Event()
            self._mp_stop = None
            self._children = []
            self._state = BenchmarkState(
                status="running",
                stage="Preparing real hardware measurement",
                progress=2,
                started_at=datetime.now(timezone.utc).isoformat(),
            )
            self._thread = threading.Thread(
                target=self._run,
                args=(str(ping_url or "").strip(),),
                daemon=True,
                name="cloudvault-admin-hardware-benchmark",
            )
            self._thread.start()
        return True, "Hardware measurement started."

    def stop(self) -> bool:
        with self._lock:
            running = bool(self._thread and self._thread.is_alive())
            if not running:
                return False
            self._state.stage = "Stopping safely"
            self._state.cancelled = True
            self._stop.set()
            if self._mp_stop is not None:
                try:
                    self._mp_stop.set()
                except Exception:
                    pass
            children = list(self._children)
        for child in children:
            try:
                if child.is_alive():
                    child.join(timeout=0.25)
                if child.is_alive():
                    child.terminate()
            except Exception:
                pass
        return True

    def _measure_ping(self, url: str) -> dict[str, Any]:
        if not url:
            return {"available": False, "error": "PUBLIC_BASE_URL is empty."}
        samples: list[float] = []
        errors: list[str] = []
        for _ in range(3):
            if self._stop.is_set():
                break
            started = time.perf_counter()
            try:
                request = urllib.request.Request(
                    url,
                    method="GET",
                    headers={"User-Agent": "CloudVaultHub-v172-hardware-test"},
                )
                with urllib.request.urlopen(request, timeout=6) as response:
                    response.read(64)
                samples.append((time.perf_counter() - started) * 1000.0)
            except Exception as exc:
                errors.append(str(exc)[:240])
            if self._stop.wait(0.15):
                break
        if not samples:
            return {"available": False, "error": errors[-1] if errors else "Measurement stopped."}
        return {
            "available": True,
            "samples_ms": [round(value, 2) for value in samples],
            "average_ms": round(sum(samples) / len(samples), 2),
            "minimum_ms": round(min(samples), 2),
            "maximum_ms": round(max(samples), 2),
            "errors": errors,
        }

    def _measure_memory(self) -> dict[str, Any]:
        # 128 MiB keeps the test safe even on CPU Basic while still measuring
        # real copy bandwidth. It never tries to fill the advertised RAM.
        size = 128 * MB
        source = bytearray(os.urandom(1024)) * (size // 1024)
        target = bytearray(size)
        passes = 6
        started = time.perf_counter()
        completed = 0
        for _ in range(passes):
            if self._stop.is_set():
                break
            target[:] = source
            completed += 1
        elapsed = max(0.001, time.perf_counter() - started)
        bytes_copied = size * completed
        checksum = hashlib.sha256(target[:4096]).hexdigest()[:12] if completed else ""
        del source, target
        return {
            "buffer_mb": size / MB,
            "passes": completed,
            "elapsed_seconds": round(elapsed, 3),
            "copy_mbps": round((bytes_copied / MB) / elapsed, 2) if completed else 0.0,
            "checksum": checksum,
        }

    def _measure_cpu(self) -> dict[str, Any]:
        hardware = self.monitor.hardware()
        desired = max(1, min(16, hardware.logical_cpu_visible, int(math.ceil(hardware.effective_vcpu))))
        duration = 5.0
        try:
            context = mp.get_context("fork")
        except ValueError:
            context = mp.get_context()
        stop_event = context.Event()
        output_queue = context.Queue()
        with self._lock:
            self._mp_stop = stop_event
        processes = [
            context.Process(
                target=_cpu_benchmark_worker,
                args=(duration, stop_event, output_queue),
                daemon=True,
                name=f"cvh-cpu-benchmark-{index + 1}",
            )
            for index in range(desired)
        ]
        with self._lock:
            self._children = processes
        wall_started = time.perf_counter()
        for process in processes:
            process.start()
        while any(process.is_alive() for process in processes):
            if self._stop.wait(0.1):
                stop_event.set()
                break
        for process in processes:
            process.join(timeout=1.0)
            if process.is_alive():
                process.terminate()
                process.join(timeout=0.5)
        wall_elapsed = max(0.001, time.perf_counter() - wall_started)
        reports: list[dict[str, Any]] = []
        while True:
            try:
                reports.append(output_queue.get_nowait())
            except queue.Empty:
                break
            except Exception:
                break
        total_iterations = sum(int(item.get("iterations", 0)) for item in reports)
        total_cpu_seconds = sum(_safe_float(item.get("cpu_seconds")) for item in reports)
        estimated_parallel = min(
            float(desired),
            max(0.0, total_cpu_seconds / wall_elapsed),
        )
        return {
            "workers_requested": desired,
            "workers_reported": len(reports),
            "duration_seconds": round(wall_elapsed, 3),
            "hashes": total_iterations,
            "hashes_per_second": round(total_iterations / wall_elapsed, 2),
            "aggregate_cpu_seconds": round(total_cpu_seconds, 3),
            "mathematical_parallel_vcpu": round(estimated_parallel, 2),
            "effective_cgroup_vcpu": hardware.effective_vcpu,
            "cancelled": self._stop.is_set(),
        }

    def _run(self, ping_url: str) -> None:
        result: dict[str, Any] = {}
        try:
            self._set(stage="Measuring server ping", progress=10)
            result["ping"] = self._measure_ping(ping_url)
            if self._stop.is_set():
                raise InterruptedError("Measurement stopped by the developer.")

            self._set(stage="Measuring RAM copy speed", progress=32)
            result["memory"] = self._measure_memory()
            if self._stop.is_set():
                raise InterruptedError("Measurement stopped by the developer.")

            self._set(stage="Measuring parallel CPU capacity", progress=55)
            result["cpu"] = self._measure_cpu()
            if self._stop.is_set():
                raise InterruptedError("Measurement stopped by the developer.")

            result["hardware"] = asdict(self.monitor.hardware())
            result["live_after_test"] = asdict(self.monitor.snapshot())
            self._set(
                status="complete",
                stage="Measurement completed",
                progress=100,
                finished_at=datetime.now(timezone.utc).isoformat(),
                result=result,
            )
            print("[CloudVaultHub v172 benchmark] " + json.dumps(result, ensure_ascii=False), flush=True)
        except InterruptedError as exc:
            self._set(
                status="cancelled",
                stage=str(exc),
                progress=min(99, max(0, self._state.progress)),
                finished_at=datetime.now(timezone.utc).isoformat(),
                cancelled=True,
                result=result,
            )
        except Exception as exc:
            self._set(
                status="error",
                stage="Measurement failed",
                finished_at=datetime.now(timezone.utc).isoformat(),
                error=str(exc),
                result=result,
            )
        finally:
            with self._lock:
                self._children = []
                self._mp_stop = None


RESOURCE_MONITOR = SystemResourceMonitor(refresh_seconds=60.0)
RESOURCE_BENCHMARK = SystemBenchmarkManager(RESOURCE_MONITOR)