"""Read-only инструменты мониторинга TrueNAS v2.0.""" import json import time import requests from typing import Any, Optional from pydantic import BaseModel, Field class Tools: """Мониторинг TrueNAS через REST API. Изменяющие операции отсутствуют.""" class Valves(BaseModel): TRUENAS_URL: str = Field(default="http://192.168.31.100", description="URL TrueNAS") TRUENAS_API_KEY: str = Field(default="", description="API-ключ TrueNAS") CACHE_TTL: int = Field(default=30, description="TTL кэша GET-запросов") MEM_CACHE_TTL: int = Field(default=60, description="TTL кэша RAM") verify_ssl: bool = Field(default=False, description="Проверять TLS") request_timeout: int = Field(default=15, description="Таймаут запроса") def __init__(self): self.valves = self.Valves() self._cache = {} self._last_error = "нет данных" @property def _base(self): return self.valves.TRUENAS_URL.rstrip("/") @property def _headers(self): return {"Authorization": f"Bearer {self.valves.TRUENAS_API_KEY}", "Content-Type": "application/json"} def clear_cache(self) -> str: """Очищает кэш запросов.""" n = len(self._cache) self._cache.clear() return f"🧹 Кэш очищен ({n} записей удалено)." def _get(self, key: str, endpoint: str, ttl: Optional[int] = None): """Выполняет кэшируемый GET-запрос.""" ttl = self.valves.CACHE_TTL if ttl is None else ttl now = time.time() if key in self._cache and ttl > 0 and now - self._cache[key][0] < ttl: return self._cache[key][1] try: r = requests.get(f"{self._base}/api/v2.0/{endpoint}", headers=self._headers, timeout=self.valves.request_timeout, verify=self.valves.verify_ssl) if r.status_code == 200: value = r.json() if ttl > 0: self._cache[key] = (now, value) return value self._last_error = f"HTTP {r.status_code} на {endpoint}: {r.text[:300]}" except Exception as e: self._last_error = f"{type(e).__name__}: {e}" return None def _first(self, name, endpoints): for ep in endpoints: value = self._get(f"{name}_{ep.replace('/', '_')}", ep) if value is not None: return value return None @staticmethod def _fmt(value: Any) -> str: if value is None: return "Н/Д" try: n = float(value) for unit in ("B", "KiB", "MiB", "GiB", "TiB"): if abs(n) < 1024: return f"{n:.1f} {unit}" n /= 1024 return f"{n:.1f} PiB" except (TypeError, ValueError): return str(value) def get_debug_info(self) -> str: """Проверяет соединение с TrueNAS.""" try: r = requests.get(f"{self._base}/api/v2.0/system/info", headers=self._headers, timeout=self.valves.request_timeout, verify=False) return f"URL: {self._base}\nКод ответа: {r.status_code}\nПоследняя ошибка: {self._last_error}" except Exception as e: return f"URL: {self._base}\nОшибка: {type(e).__name__}: {e}" def validate_api_key(self) -> str: """Проверяет read-only доступ API-ключа.""" if not self.valves.TRUENAS_API_KEY: return "❌ API-ключ не задан" checks = [("system/info", "система"), ("pool/query", "пулы"), ("disk/query", "диски"), ("app/query", "приложения"), ("alert/list", "алерты")] out, ok = ["🔐 ВАЛИДАЦИЯ API-КЛЮЧА TRUENAS", ""], 0 for ep, title in checks: value = self._get(f"validate_{ep}", ep, 0) if value is not None: out.append(f"✅ {ep:15s} — {title}") ok += 1 else: out.append(f"⚠️ {ep:15s} — недоступен") out.append(f"\nИтого: {ok}/{len(checks)}") return "\n".join(out) def get_memory_info(self) -> str: """Показывает RAM через доступные API-эндпоинты.""" data = self._first("memory", ["system/memory", "system/info", "system/get_info"]) if not isinstance(data, dict): return "💾 RAM: Н/Д (TrueNAS REST API не возвращает память в этой версии)" total = data.get("total_memory") or data.get("memory_size") or data.get("memory") or data.get("total") used = data.get("used") return f"💾 RAM\n Всего: {self._fmt(total)}\n Использовано: {self._fmt(used)}" def get_system_status(self) -> str: """Показывает системную информацию.""" d = self._get("system", "system/info") if not isinstance(d, dict): return f"⚠️ Ошибка получения системы: {self._last_error}" load = d.get("loadavg") or [] return "\n".join([f"🖥️ TrueNAS: {d.get('version', 'N/A')}", f"🔧 Процессор: {d.get('model', 'N/A')}", f"📊 Load Average: {load}", self.get_memory_info()]) def get_pools_status(self) -> str: """Показывает пулы ZFS.""" pools = self._first("pools", ["pool/query", "pool", "storage/pool"]) if not isinstance(pools, list): return f"⚠️ Пулы недоступны: {self._last_error}" rows = [] for p in pools: size, free = p.get("size", 0), p.get("free", 0) used = (size - free) / size * 100 if size else 0 rows.append(f"{'✅' if p.get('healthy') else '⚠️'} '{p.get('name', '?')}' | {p.get('status', '?')} | {used:.1f}% занято") return "\n".join(rows) or "Пулы не найдены." def get_disks_temp_and_info(self) -> str: """Показывает диски и температуры.""" disks = self._first("disks", ["disk/query", "disk", "storage/disk"]) if not isinstance(disks, list): return f"⚠️ Диски недоступны: {self._last_error}" return "\n".join(f"💾 {d.get('name', 'N/A')} | {d.get('model', '')} | {self._fmt(d.get('size'))} | {d.get('temperature', 'N/A')}°C" for d in disks) def get_apps_status(self) -> str: """Показывает установленные приложения. Установка отсутствует.""" apps = self._first("apps", ["app/query", "app", "chart/release/query"]) if not isinstance(apps, list): return f"⚠️ Приложения недоступны: {self._last_error}" icons = {"RUNNING": "🟢", "STOPPED": "⚪", "DEPLOYING": "🟡", "CRASHED": "🔴"} return "\n".join(f"{icons.get(str(a.get('state', a.get('status', 'N/A'))).upper(), '⚪')} {a.get('name', a.get('id', '?'))} | статус: {a.get('state', a.get('status', 'N/A'))} | версия: {a.get('human_version', a.get('version', 'N/A'))}" for a in apps) or "Приложения не установлены." def get_cpu_temperature(self) -> str: """Показывает температуру CPU, если она есть в API.""" d = self._get("temperature", "system/info") or {} value = d.get("cpu_temperature") or d.get("cpu_temp") or d.get("temperature") return f"🌡️ CPU: {value} °C" if value is not None else "🌡️ Температура CPU недоступна через API." def get_arc_stats(self) -> str: """Показывает ZFS ARC, если он доступен через API.""" d = self._first("arc", ["stats", "zfs/arc", "zfs/pool/stats", "pool/statistics"]) return json.dumps(d, ensure_ascii=False, indent=2, default=str) if d else "📊 ARC-статистика недоступна через API." def get_vms_status(self) -> str: """Показывает виртуальные машины.""" vms = self._first("vms", ["vm/query", "vm"]) if not isinstance(vms, list): return f"⚠️ VM недоступны: {self._last_error}" return "\n".join(f"{v.get('name', '?')} | {v.get('status', 'N/A')}" for v in vms) or "Виртуальные машины не найдены." def get_containers_status(self) -> str: """Показывает контейнеры virt/incus.""" items = self._first("containers", ["virt/instance", "virt/instance/query", "container/query"]) if not isinstance(items, list): return f"⚠️ Контейнеры недоступны: {self._last_error}" return "\n".join(f"{c.get('name', c.get('id', '?'))} | {c.get('status', c.get('state', 'N/A'))}" for c in items) or "Контейнеры не найдены." def get_active_alerts(self) -> str: """Показывает активные алерты.""" alerts = self._get("alerts", "alert/list") or [] active = [a for a in alerts if not a.get("dismissed", False)] return "✅ Все системы работают штатно." if not active else "\n".join(f"[{a.get('level', 'INFO')}] {a.get('formatted', a.get('message', '?'))}" for a in active) def get_full_report(self) -> str: """Формирует полный read-only отчёт.""" sections = [("СИСТЕМА", self.get_system_status()), ("RAM", self.get_memory_info()), ("ТЕМПЕРАТУРА", self.get_cpu_temperature()), ("ARC", self.get_arc_stats()), ("ПУЛЫ", self.get_pools_status()), ("ДИСКИ", self.get_disks_temp_and_info()), ("ПРИЛОЖЕНИЯ", self.get_apps_status()), ("VM", self.get_vms_status()), ("КОНТЕЙНЕРЫ", self.get_containers_status()), ("АЛЕРТЫ", self.get_active_alerts())] return "\n\n".join(f"--- {title} ---\n{content}" for title, content in sections)