orchestrata_ultimate.kernel.resource_monitor

Reference › orchestrata_ultimate › kernel › resource_monitor

The system resource monitor of the kernel. ResourceMonitor samples CPU, memory and disk usage, compares each sample with configurable thresholds and reports alerts. It serves SPEC-001 story US-007 (REQ-001-US-007-001, -009, -010). It depends only on the traces decorator and nothing in the package imports it; the performance package is the related scalability toolbox.

take_snapshot reads get_cpu_percent (default sampling interval 0.1 seconds), get_memory_stats and get_disk_stats, builds a ResourceSnapshot (percentages plus used and total megabytes or gigabytes, with to_dict), and calls check_thresholds. That method adds a ResourceAlert for each of cpu, memory and disk whose percentage is above its threshold and counts the alerts. The snapshot is added to a history capped at history_size (100 by default), and every alert is passed to the registered callbacks (register_alert_callback); a callback exception is logged and does not stop the others. Thresholds default to module constants and can be changed with set_thresholds. get_average_cpu and get_peak_cpu summarise the recent history, is_cpu_healthy samples the CPU and compares it with the threshold, total_alerts returns the alert count, and get_system_summary returns a dictionary overview.

At a glance

Class diagram

Dependency diagram

API

Classes

ResourceAlert (dataclass)

Decorators: dataclass, traces(spec='001', story='US-007', reqs=['REQ-001-US-007-010']) · Line 30

Alert triggered when a resource exceeds its threshold.

ResourceSnapshot (dataclass)

Decorators: dataclass, traces(spec='001', story='US-007', reqs=['REQ-001-US-007-001']) · Line 49

Point-in-time snapshot of system resources.

ResourceMonitor

Decorators: traces(spec='001', story='US-007', reqs=['REQ-001-US-007-001', 'REQ-001-US-007-009', 'REQ-001-US-007-010', 'REQ-001-US-… · Line 96

Monitors system resource utilization with configurable thresholds.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/kernel/resource_monitor.py

# mypy: ignore-errors
"""System resource monitoring for Orchestrata.

Monitors CPU, memory, and disk utilization with configurable thresholds
and alerting support.

Spec: 002-postman / US-010 / REQ-002-US-010-237
"""

import logging
import time
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any

import psutil

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)

# Default thresholds
DEFAULT_CPU_THRESHOLD = 50.0  # percent
DEFAULT_MEMORY_THRESHOLD = 80.0  # percent
DEFAULT_DISK_THRESHOLD = 90.0  # percent


@dataclass
@traces(spec="001", story="US-007", reqs=["REQ-001-US-007-010"])
class ResourceAlert:
    """Alert triggered when a resource exceeds its threshold."""

    resource: str  # "cpu", "memory", "disk"
    current_value: float
    threshold: float
    timestamp: float = field(default_factory=time.time)
    message: str = ""

    def __post_init__(self) -> None:
        if not self.message:
            self.message = (
                f"{self.resource} usage at {self.current_value:.1f}% "
                f"exceeds threshold of {self.threshold:.1f}%"
            )


@dataclass
@traces(spec="001", story="US-007", reqs=["REQ-001-US-007-001"])
class ResourceSnapshot:
    """Point-in-time snapshot of system resources."""

    cpu_percent: float
    memory_percent: float
    memory_used_mb: float
    memory_total_mb: float
    disk_percent: float
    disk_used_gb: float
    disk_total_gb: float
    timestamp: float = field(default_factory=time.time)
    alerts: list[ResourceAlert] = field(default_factory=list)

    @traces(spec="001", story="US-007", reqs=["REQ-001-US-007-026"])
    def to_dict(self) -> dict[str, Any]:
        return {
            "cpu_percent": self.cpu_percent,
            "memory_percent": self.memory_percent,
            "memory_used_mb": round(self.memory_used_mb, 1),
            "memory_total_mb": round(self.memory_total_mb, 1),
            "disk_percent": self.disk_percent,
            "disk_used_gb": round(self.disk_used_gb, 2),
            "disk_total_gb": round(self.disk_total_gb, 2),
            "timestamp": self.timestamp,
            "alerts": [
                {
                    "resource": a.resource,
                    "current": a.current_value,
                    "threshold": a.threshold,
                    "message": a.message,
                }
                for a in self.alerts
            ],
        }


@traces(
    spec="001",
    story="US-007",
    reqs=[
        "REQ-001-US-007-001",
        "REQ-001-US-007-009",
        "REQ-001-US-007-010",
        "REQ-001-US-007-025",
        "REQ-001-US-007-027",
    ],
)
class ResourceMonitor:
    """Monitors system resource utilization with configurable thresholds.

    Provides real-time CPU, memory, and disk monitoring with alert generation
    when usage exceeds configured thresholds. Maintains a history of snapshots
    for trend analysis.
    """

    def __init__(
        self,
        cpu_threshold: float = DEFAULT_CPU_THRESHOLD,
        memory_threshold: float = DEFAULT_MEMORY_THRESHOLD,
        disk_threshold: float = DEFAULT_DISK_THRESHOLD,
        history_size: int = 100,
    ) -> None:
        self._cpu_threshold = cpu_threshold
        self._memory_threshold = memory_threshold
        self._disk_threshold = disk_threshold
        self._history_size = history_size
        self._history: list[ResourceSnapshot] = []
        self._alert_callbacks: list = []
        self._total_alerts = 0

    @property
    def cpu_threshold(self) -> float:
        return self._cpu_threshold

    @property
    def memory_threshold(self) -> float:
        return self._memory_threshold

    @property
    def disk_threshold(self) -> float:
        return self._disk_threshold

    @property
    @traces(spec="001", story="US-007", reqs=["REQ-001-US-007-027"])
    def history(self) -> list[ResourceSnapshot]:
        return list(self._history)

    @property
    @traces(spec="001", story="US-007", reqs=["REQ-001-US-007-010"])
    def total_alerts(self) -> int:
        return self._total_alerts

    @traces(spec="001", story="US-007", reqs=["REQ-001-US-007-021"])
    def get_cpu_percent(self, interval: float = 0.1) -> float:
        """Get current CPU utilization percentage."""
        return psutil.cpu_percent(interval=interval)

    @traces(spec="001", story="US-007", reqs=["REQ-001-US-007-022"])
    def get_memory_stats(self) -> dict[str, float]:
        """Get current memory utilization."""
        mem = psutil.virtual_memory()
        return {
            "percent": mem.percent,
            "used_mb": mem.used / (1024 * 1024),
            "total_mb": mem.total / (1024 * 1024),
            "available_mb": mem.available / (1024 * 1024),
        }

    @traces(spec="001", story="US-007", reqs=["REQ-001-US-007-024"])
    def get_disk_stats(self, path: str = "/") -> dict[str, float]:
        """Get current disk utilization."""
        disk = psutil.disk_usage(path)
        return {
            "percent": disk.percent,
            "used_gb": disk.used / (1024**3),
            "total_gb": disk.total / (1024**3),
            "free_gb": disk.free / (1024**3),
        }

    @traces(spec="001", story="US-007", reqs=["REQ-001-US-007-010"])
    def check_thresholds(self, snapshot: ResourceSnapshot) -> list[ResourceAlert]:
        """Check if any resource exceeds its threshold."""
        alerts = []

        if snapshot.cpu_percent > self._cpu_threshold:
            alerts.append(
                ResourceAlert(
                    resource="cpu",
                    current_value=snapshot.cpu_percent,
                    threshold=self._cpu_threshold,
                )
            )

        if snapshot.memory_percent > self._memory_threshold:
            alerts.append(
                ResourceAlert(
                    resource="memory",
                    current_value=snapshot.memory_percent,
                    threshold=self._memory_threshold,
                )
            )

        if snapshot.disk_percent > self._disk_threshold:
            alerts.append(
                ResourceAlert(
                    resource="disk",
                    current_value=snapshot.disk_percent,
                    threshold=self._disk_threshold,
                )
            )

        self._total_alerts += len(alerts)
        return alerts

    @traces(spec="001", story="US-007", reqs=["REQ-001-US-007-001"])
    def take_snapshot(self, cpu_interval: float = 0.1) -> ResourceSnapshot:
        """Take a point-in-time resource snapshot with threshold checks."""
        cpu = self.get_cpu_percent(interval=cpu_interval)
        mem = self.get_memory_stats()
        disk = self.get_disk_stats()

        snapshot = ResourceSnapshot(
            cpu_percent=cpu,
            memory_percent=mem["percent"],
            memory_used_mb=mem["used_mb"],
            memory_total_mb=mem["total_mb"],
            disk_percent=disk["percent"],
            disk_used_gb=disk["used_gb"],
            disk_total_gb=disk["total_gb"],
        )

        alerts = self.check_thresholds(snapshot)
        snapshot.alerts = alerts

        # Maintain history
        self._history.append(snapshot)
        if len(self._history) > self._history_size:
            self._history = self._history[-self._history_size :]

        # Fire callbacks
        for alert in alerts:
            for callback in self._alert_callbacks:
                try:
                    callback(alert)
                except Exception as e:
                    logger.warning("Alert callback failed: %s", e)

        return snapshot

    @traces(spec="001", story="US-007", reqs=["REQ-001-US-007-027"])
    def get_average_cpu(self, last_n: int = 10) -> float:
        """Get average CPU usage over last N snapshots."""
        if not self._history:
            return 0.0
        recent = self._history[-last_n:]
        return sum(s.cpu_percent for s in recent) / len(recent)

    @traces(spec="001", story="US-007", reqs=["REQ-001-US-007-027"])
    def get_peak_cpu(self, last_n: int = 10) -> float:
        """Get peak CPU usage over last N snapshots."""
        if not self._history:
            return 0.0
        recent = self._history[-last_n:]
        return max(s.cpu_percent for s in recent)

    def is_cpu_healthy(self) -> bool:
        """Check if CPU usage is below threshold."""
        cpu = self.get_cpu_percent(interval=0.1)
        return cpu <= self._cpu_threshold

    @traces(
        spec="001", story="US-007", reqs=["REQ-001-US-007-010", "REQ-001-US-007-019"]
    )
    def register_alert_callback(
        self, callback: Callable[[ResourceAlert], None]
    ) -> None:
        """Register a callback for resource alerts."""
        self._alert_callbacks.append(callback)

    @traces(spec="001", story="US-007", reqs=["REQ-001-US-007-020"])
    def set_thresholds(
        self,
        cpu: float | None = None,
        memory: float | None = None,
        disk: float | None = None,
    ) -> None:
        """Update resource thresholds at runtime."""
        if cpu is not None:
            self._cpu_threshold = cpu
        if memory is not None:
            self._memory_threshold = memory
        if disk is not None:
            self._disk_threshold = disk

    @traces(
        spec="001",
        story="US-007",
        reqs=["REQ-001-US-007-009", "REQ-001-US-007-025", "REQ-001-US-007-026"],
    )
    def get_system_summary(self) -> dict[str, Any]:
        """Get a complete system resource summary."""
        snapshot = self.take_snapshot()
        return {
            "snapshot": snapshot.to_dict(),
            "thresholds": {
                "cpu": self._cpu_threshold,
                "memory": self._memory_threshold,
                "disk": self._disk_threshold,
            },
            "history_size": len(self._history),
            "total_alerts": self._total_alerts,
            "cpu_healthy": snapshot.cpu_percent <= self._cpu_threshold,
        }