#!/usr/bin/env python3 """ 통합 서버 정보 수집 스크립트 프로파일 기반으로 BIOS/iDRAC/RAID 설정 수집 """ import sys import yaml from pathlib import Path from concurrent.futures import ThreadPoolExecutor import logging sys.path.insert(0, str(Path(__file__).parent.parent)) from core.idrac_client import IDRACClient from core.parsers import InfoParser logging.basicConfig( level=logging.INFO, format='%(asctime)s [INFO] root: %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) def load_profile(profile_name: str) -> dict: """프로파일 로드 및 상속 처리""" profile_file = Path(__file__).parent.parent / "profiles" / "server_info_profiles.yaml" if not profile_file.exists(): logging.error(f"프로파일 파일 없음: {profile_file}") sys.exit(1) with open(profile_file, 'r', encoding='utf-8') as f: profiles = yaml.safe_load(f)['profiles'] profile = profiles.get(profile_name) if not profile: logging.error(f"프로파일 '{profile_name}' 없음. 사용 가능: {list(profiles.keys())}") sys.exit(1) # extends 처리 if 'extends' in profile: base = profiles[profile['extends']].copy() # 리스트 병합 for key in ['firmware', 'bios_settings', 'idrac_settings', 'raid_settings', 'custom_items']: if key in profile: base_items = base.get(key, []) profile_items = profile[key] # 중복 제거하면서 병합 base[key] = base_items + profile_items base.update({k: v for k, v in profile.items() if k not in ['firmware', 'bios_settings', 'idrac_settings', 'raid_settings', 'custom_items']}) profile = base return profile def collect_server_info_for_ip(ip: str, profile: dict, output_dir: Path) -> tuple: """단일 IP에 대한 서버 정보 수집""" try: client = IDRACClient(ip) parser = InfoParser() # 서비스 태그 추출 sysinfo = client.get_sysinfo() svc_tag = parser.extract_service_tag(sysinfo) if not svc_tag: return ip, False, "서비스 태그 추출 실패" # 결과 파일 생성 output_file = output_dir / f"{svc_tag}.txt" with output_file.open('w', encoding='utf-8', newline='\n') as f: f.write(f"Dell EMC Server Bios,iDRAC,R/C Setting (SVC Tag: {svc_tag})\n\n") # 펌웨어 버전 정보 if 'firmware' in profile: f.write("-" * 42 + " Firmware Version 정보 " + "-" * 42 + "\n") for idx, item in enumerate(profile['firmware'], 1): value = _get_value_from_command(client, item['command'], item['pattern']) f.write(f"{idx}. {item['name']} : {value}\n") f.write("\n") # BIOS 설정 정보 if 'bios_settings' in profile: f.write("-" * 45 + " Bios 설정 정보 " + "-" * 46 + "\n") for idx, item in enumerate(profile['bios_settings'], 1): value = _get_value_from_command(client, item['command'], item['pattern']) f.write(f"{idx:02d}. {item['name']} : {value}\n") f.write("\n") # iDRAC 설정 정보 if 'idrac_settings' in profile: f.write("-" * 45 + " iDRAC 설정 정보 " + "-" * 45 + "\n") for idx, item in enumerate(profile['idrac_settings'], 1): value = _get_value_from_command(client, item['command'], item['pattern']) f.write(f"{idx:02d}. {item['name']} : {value}\n") f.write("\n") # RAID 설정 정보 if 'raid_settings' in profile: f.write("-" * 45 + " Raid 설정 정보 " + "-" * 46 + "\n") for idx, item in enumerate(profile['raid_settings'], 1): value = _get_value_from_command(client, item['command'], item['pattern']) f.write(f"{idx:02d}. {item['name']} : {value}\n") f.write("\n") # 커스텀 항목 (있을 경우) if 'custom_items' in profile and profile['custom_items']: f.write("-" * 45 + " 커스텀 항목 " + "-" * 46 + "\n") for idx, item in enumerate(profile['custom_items'], 1): value = _get_value_from_command(client, item['command'], item['pattern']) f.write(f"{idx:02d}. {item['name']} : {value}\n") f.write("\n") return ip, True, f"저장: {output_file.name}" except Exception as e: logging.error(f"[{ip}] 오류: {e}") return ip, False, f"오류: {e}" def _get_value_from_command(client: IDRACClient, command: str, pattern: str) -> str: """racadm 명령 실행 및 값 추출""" # 캐싱된 데이터 사용 if command == "getsysinfo": output = client.get_sysinfo() elif command == "hwinventory": output = client.get_hwinventory() else: _, output = client.run_command(command) # 패턴 매칭 value = InfoParser.extract_value_by_pattern(output, pattern) return value if value else "N/A" def main(): if len(sys.argv) < 2: print("Usage: python collect_server_info.py [profile_name]") print("Available profiles: default, intel_server, amd_server, minimal") sys.exit(1) ip_file = Path(sys.argv[1]) profile_name = sys.argv[2] if len(sys.argv) > 2 else "default" if not ip_file.exists(): logging.error(f"IP 파일 없음: {ip_file}") sys.exit(1) # 프로파일 로드 profile = load_profile(profile_name) logging.info(f"프로파일 사용: {profile_name} - {profile.get('description', '')}") # 출력 디렉토리 (data/temp/staging) script_dir = Path(__file__).resolve().parent # unified 디렉토리에 있으므로, data 디렉토리는 3단계 상위 (unified -> scripts -> data) data_dir = script_dir.parent.parent # 만약 구조가 다를 경우를 대비한 안전장치 (scripts가 아닐 경우 등) if data_dir.name != "data": # fallback: 현재 작업 디렉토리 기준 data 찾기 시도 data_dir = Path("data").resolve() output_dir = data_dir / "temp" / "staging" output_dir.mkdir(parents=True, exist_ok=True) # IP 목록 읽기 ips = [line.strip() for line in ip_file.read_text(encoding='utf-8').splitlines() if line.strip()] if not ips: logging.error("IP 목록이 비어 있습니다.") sys.exit(1) logging.info(f"총 {len(ips)}대 처리 시작") # 병렬 처리 (서버 정보 수집은 racadm 호출이 많아 worker 수 줄임) ok, fail = 0, 0 with ThreadPoolExecutor(max_workers=10) as executor: futures = {executor.submit(collect_server_info_for_ip, ip, profile, output_dir): ip for ip in ips} for future in futures: ip, success, msg = future.result() if success: logging.info(f"[OK] {ip} - {msg}") ok += 1 else: logging.error(f"[FAIL] {ip} - {msg}") fail += 1 logging.info(f"완료: 성공 {ok}대 / 실패 {fail}대") if __name__ == "__main__": main()