modules/resource_module.py - master - gabelchik/dpcs-diagnostics

modules/resource_module.py
74 строки3 KB
Leonid Kalintsev
messages
21 июл 2026, 15:09
21 июл 2026, 15:0955529c2
from typing import Dict, Any
from core.interfaces import DiagnosticModule


class ResourceModule(DiagnosticModule):
    def __init__(self, name: str, mod_cfg: Dict[str, Any]):
        super().__init__(name)
        self._limits = mod_cfg.get("limits") or {}

    def execute(self, actual_state: dict = None) -> Dict[str, Any]:
        if not self._limits:
            return {
                "status": "FAILURE",
                "message": "Эталонные данные отсуствуют",
                "metrics": {},
                "components": {}
            }

        if not actual_state:
            return {
                "status": "FAILURE",
                "message": "Данные от узла отсутствуют",
                "metrics": {},
                "limits": self._limits
            }

        actual_resources = actual_state.get(self.name) or {}
        actual_values = actual_resources.get("limits") or {}

        cpu = actual_values.get("max_cpu", 0)
        ram = actual_values.get("max_ram", 0)
        disk = actual_values.get("max_disk", 0)

        max_cpu_limit = self._limits.get("max_cpu")
        max_ram_limit = self._limits.get("max_ram")
        max_disk_limit = self._limits.get("max_disk")

        if max_cpu_limit is None or max_ram_limit is None or max_disk_limit is None:
            return {
                "status": "FAILURE",
                "message": "В конфигурации заданы не все ресурсные лимиты",
                "metrics": {
                    "cpu_percent": cpu,
                    "ram_percent": ram,
                    "disk_percent": disk
                },
                "limits": self._limits
            }

        problems = []
        if cpu > max_cpu_limit:
            problems.append(f"Превышение по cpu: {cpu}% > {max_cpu_limit}%")
        if ram > max_ram_limit:
            problems.append(f"Превышение по ram: {ram}% > {max_ram_limit}%")
        if disk > max_disk_limit:
            problems.append(f"Превышение по disk: {disk}% > {max_disk_limit}%")

        is_failed = len(problems) > 0
        message = "; ".join(problems) if is_failed else "Нагрузка на ресурсы в пределах нормы"

        return {
            "status": "FAILURE" if is_failed else "SUCCESS",
            "message": message,
            "metrics": {
                "cpu_percent": cpu,
                "ram_percent": ram,
                "disk_percent": disk
            },
            "limits": {
                "max_cpu": max_cpu_limit,
                "max_ram": max_ram_limit,
                "max_disk": max_disk_limit
            }
        }