""" iDRAC 클라이언트 - racadm 명령 실행 및 결과 캐싱 """ from typing import Optional, Dict, Tuple import subprocess import os import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s [INFO] root: %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) class IDRACClient: """ racadm 명령 실행 및 결과 캐싱을 담당하는 클라이언트 """ def __init__(self, ip: str, user: str = None, password: str = None): """ Args: ip: iDRAC IP 주소 user: iDRAC 사용자명 (기본값: 환경변수 IDRAC_USER 또는 'root') password: iDRAC 비밀번호 (기본값: 환경변수 IDRAC_PASS 또는 'calvin') """ self.ip = ip self.user = user or os.getenv("IDRAC_USER", "root") self.password = password or os.getenv("IDRAC_PASS", "calvin") self._cache: Dict[str, str] = {} # 중복 호출 방지 캐시 def run_command(self, command: str) -> Tuple[int, str]: """ racadm 명령 실행 Args: command: 실행할 racadm 명령 (예: "getsysinfo", "get bios.BiosBootSettings") Returns: (return_code, output) 튜플 """ cmd = f"racadm -r {self.ip} -u {self.user} -p {self.password} {command}" try: result = subprocess.run( cmd, shell=True, capture_output=True, text=True, check=False, timeout=30 # 30초 타임아웃 ) return result.returncode, result.stdout except subprocess.TimeoutExpired: logging.error(f"[{self.ip}] 명령 타임아웃: {command}") return 1, "" except Exception as e: logging.error(f"[{self.ip}] 명령 실행 오류: {e}") return 1, f"Error: {e}" def get_sysinfo(self) -> str: """ getsysinfo 명령 실행 (캐싱) Returns: getsysinfo 출력 결과 """ if "sysinfo" not in self._cache: _, output = self.run_command("getsysinfo") self._cache["sysinfo"] = output return self._cache["sysinfo"] def get_hwinventory(self) -> str: """ hwinventory 명령 실행 (캐싱) Returns: hwinventory 출력 결과 """ if "hwinventory" not in self._cache: _, output = self.run_command("hwinventory") self._cache["hwinventory"] = output return self._cache["hwinventory"] def get_swinventory(self) -> str: """ swinventory 명령 실행 (캐싱) Returns: swinventory 출력 결과 """ if "swinventory" not in self._cache: _, output = self.run_command("swinventory") self._cache["swinventory"] = output return self._cache["swinventory"] def get_infiniband_pages(self) -> str: """ InfiniBand 설정 페이지 목록 조회 (InfiniBand.VndrConfigPage) Returns: racadm get InfiniBand.VndrConfigPage 출력 """ if "infiniband_pages" not in self._cache: _, output = self.run_command("get InfiniBand.VndrConfigPage") self._cache["infiniband_pages"] = output return self._cache["infiniband_pages"] def get_infiniband_page_detail(self, page_number: str) -> str: """ 특정 InfiniBand 설정 페이지 상세 조회 Args: page_number: 페이지 번호 (예: "1") Returns: racadm get InfiniBand.VndrConfigPage.{page_number} 출력 """ cache_key = f"infiniband_page_{page_number}" if cache_key not in self._cache: _, output = self.run_command(f"get InfiniBand.VndrConfigPage.{page_number}") self._cache[cache_key] = output return self._cache[cache_key] def clear_cache(self): """캐시 초기화""" self._cache.clear()