#!/usr/bin/env python3 — coding: utf-8 — “”” 서버 상태 모니터링 데이터 수집 모듈 파일명: /opt/server-monitor/scripts/monitor.py “”” import psutil import os import socket import datetime import logging from dotenv import loaddotenv 환경 변수 로드 loaddotenv(‘/opt/server-monitor/config/.env’) 로깅 설정 logging.basicConfig( filename=’/opt/server-monitor/logs/monitor.log’, level=getattr(logging, os.getenv(‘LOGLEVEL’, ‘INFO’)), format=’%(asctime)s – %(levelname)s – %(message)s’, datefmt=’%Y-%m-%d %H:%M:%S’ ) logger = logging.getLogger(name) 임계값 설정 CPUTHRESHOLD = int(os.getenv(‘CPUTHRESHOLD’, 80)) MEMORYTHRESHOLD = int(os.getenv(‘MEMORYTHRESHOLD’, 85)) DISKTHRESHOLD = int(os.getenv(‘DISKTHRESHOLD’, 90)) SERVERNAME = os.getenv(‘SERVERNAME’, socket.gethostname()) def getcpuinfo(): “””CPU 사용률 및 관련 정보 수집””” cpupercent = psutil.cpupercent(interval=2) cpucount = psutil.cpucount(logical=True) cpufreq = psutil.cpufreq() loadavg = os.getloadavg() return { ‘usagepercent’: cpupercent, ‘corecount’: cpucount, ‘frequencymhz’: round(cpufreq.current, 2) if cpufreq else None, ‘loadavg1min’: round(loadavg[0], 2), ‘loadavg5min’: round(loadavg[1], 2), ‘loadavg15min’: round(loadavg[2], 2), ‘iscritical’: cpupercent >= CPUTHRESHOLD } def getmemoryinfo(): “””메모리 사용량 정보 수집””” mem = psutil.virtualmemory() swap = psutil.swapmemory() return { ‘totalgb’: round(mem.total / (10243), 2), ‘usedgb’: round(mem.used / (10243), 2), ‘availablegb’: round(mem.available / (10243), 2), ‘usagepercent’: mem.percent, ‘swaptotalgb’: round(swap.total / (10243), 2), ‘swapusedgb’: round(swap.used / (10243), 2), ‘swappercent’: swap.percent, ‘iscritical’: mem.percent >= MEMORYTHRESHOLD } def getdiskinfo(): “””디스크 사용량 정보 수집 (마운트된 모든 파티션)””” disklist = [] partitions = psutil.diskpartitions() for partition in partitions: if ‘loop’ in partition.device: continue try: usage = psutil.diskusage(partition.mountpoint) disklist.append({ ‘device’: partition.device, ‘mountpoint’: partition.mountpoint, ‘totalgb’: round(usage.total / (10243), 2), ‘usedgb’: round(usage.used / (10243), 2), ‘freegb’: round(usage.free / (10243), 2), ‘usagepercent’: usage.percent, ‘iscritical’: usage.percent >= DISKTHRESHOLD }) except PermissionError: continue return disklist def getnetworkinfo(): “””네트워크 인터페이스 및 연결 정보 수집””” netio = psutil.netiocounters() connections = psutil.netconnections() established = sum(1 for c in connections if c.status == ‘ESTABLISHED’) return { ‘bytessentmb’: round(netio.bytessent / (10242), 2), ‘bytesrecvmb’: round(netio.bytesrecv / (10242), 2), ‘packetssent’: netio.packetssent, ‘packetsrecv’: netio.packetsrecv, ‘establishedconnections’: established } def gettopprocesses(limit=5): “””CPU 사용률 상위 프로세스 목록 반환””” processes = [] for proc in psutil.processiter([‘pid’, ‘name’, ‘cpupercent’, ‘memorypercent’]): try: processes.append(proc.info) except (psutil.NoSuchProcess, psutil.AccessDenied): pass sortedprocs = sorted(processes, key=lambda x: x[‘cpupercent’], reverse=True) return sortedprocs[:limit] def collectserverstatus(): “””전체 서버 상태 데이터 수집 및 반환””” logger.info(f”서버 상태 수집 시작: {SERVERNAME}”) status = { ‘servername’: SERVERNAME, ‘hostname’: socket.gethostname(), ‘ipaddress’: socket.gethostbyname(socket.gethostname()), ‘collectedat’: datetime.datetime.now().strftime(‘%Y-%m-%d %H:%M:%S’), ‘cpu’: getcpuinfo(), ‘memory’: getmemoryinfo(), ‘disks’: getdiskinfo(), ‘network’: getnetworkinfo(), ‘topprocesses’: gettopprocesses(), ‘uptimehours’: round((datetime.datetime.now().timestamp() – psutil.boottime()) / 3600, 1) } # 전체 위험 상태 여부 판단 status[‘hascritical’] = ( status[‘cpu’][‘iscritical’] or status[‘memory’][‘iscritical’] or any(d[‘iscritical’] for d in status[‘disks’]) ) logger.info(f”서버 상태 수집 완료. 위험 상태: {status[‘hascritical’]}”) return status if name == ‘main’: import json result = collectserverstatus() print(json.dumps(result, ensureascii=False, indent=2))

#!/usr/bin/env python3 — coding: utf-8 — “”” Slack 알림 전송 모듈 파일명: /opt/server-monitor/scripts/slacknotifier.py “”” import requests import json import os import logging from dotenv import loaddotenv loaddotenv(‘/opt/server-monitor/config/.env’) logger = logging.getLogger(name) SLACKWEBHOOKURL = os.getenv(‘SLACKWEBHOOKURL’) def sendslackmessage(payload: dict) -> bool: “””슬랙으로 메시지 페이로드 전송””” if not SLACKWEBHOOKURL: logger.error(“SLACKWEBHOOKURL이 설정되지 않았습니다.”) return False try: response = requests.post( SLACKWEBHOOKURL, data=json.dumps(payload), headers={‘Content-Type’: ‘application/json’}, timeout=10 ) if response.statuscode == 200 and response.text == ‘ok’: logger.info(“슬랙 메시지 전송 성공”) return True else: logger.error(f”슬랙 전송 실패: {response.statuscode} – {response.text}”) return False except requests.exceptions.RequestException as e: logger.error(f”슬랙 요청 오류: {e}”) return False def buildstatusreport(status: dict) -> dict: “””서버 상태 데이터를 슬랙 Block Kit 메시지로 변환””” servername = status[‘servername’] collectedat = status[‘collectedat’] cpu = status[‘cpu’] memory = status[‘memory’] disks = status[‘disks’] network = status[‘network’] uptime = status[‘uptimehours’] # 상태에 따른 색상 및 이콘 설정 overallstatus = “위험” if status[‘hascritical’] else “정상” color = “#FF0000” if status[‘hascritical’] else “#36A64F” # 디스크 정보 문자열 생성 disktext = “” for disk in disks: criticalmark = ” [경고]” if disk[‘iscritical’] else “” disktext += f”• {disk[‘mountpoint’]}: {disk[‘usagepercent’]}% 사용 ({disk[‘usedgb’]}GB / {disk[‘totalgb’]}GB){criticalmark}\n” # 상위 프로세스 정보 proctext = “” for proc in status.get(‘topprocesses’, [])[:3]: proctext += f”• {proc[‘name’]} (PID:{proc[‘pid’]}) – CPU: {proc[‘cpupercent’]}%\n” blocks = [ { “type”: “header”, “text”: { “type”: “plaintext”, “text”: f”[{overallstatus}] {servername} 서버 상태 리포트” } }, { “type”: “section”, “fields”: [ {“type”: “mrkdwn”, “text”: f”서버명\n{servername}”}, {“type”: “mrkdwn”, “text”: f”수집 시각\n{collectedat}”}, {“type”: “mrkdwn”, “text”: f”IP 주소\n{status[‘ipaddress’]}”}, {“type”: “mrkdwn”, “text”: f”가동 시간\n{uptime}시간”} ] }, {“type”: “divider”}, { “type”: “section”, “text”: { “type”: “mrkdwn”, “text”: f”CPU 상태 {‘[경고]’ if cpu[‘iscritical’] else ‘[정상]’}\n” f”사용률: {cpu[‘usagepercent’]}% (임계값: {os.getenv(‘CPUTHRESHOLD’, 80)}%)\n” f”코어 수: {cpu[‘corecount’]} | ” f”Load Avg: {cpu[‘loadavg1min’]} / {cpu[‘loadavg5min’]} / {cpu[‘loadavg15min’]}” } }, { “type”: “section”, “text”: { “type”: “mrkdwn”, “text”: f”메모리 상태 {‘[경고]’ if memory[‘iscritical’] else ‘[정상]’}\n” f”사용률: {memory[‘usagepercent’]}% | ” f”사용: {memory[‘usedgb’]}GB / 전체: {memory[‘totalgb’]}GB\n” f”SWAP: {memory[‘swappercent’]}% ({memory[‘swapusedgb’]}GB / {memory[‘swaptotalgb’]}GB)” } }, { “type”: “section”, “text”: { “type”: “mrkdwn”, “text”: f”디스크 상태\n{disktext}” } }, { “type”: “section”, “text”: { “type”: “mrkdwn”, “text”: f”네트워크 상태\n” f”송신: {network[‘bytessentmb’]}MB | 수신: {network[‘bytesrecvmb’]}MB\n” f”활성 연결: {network[‘establishedconnections’]}개” } }, { “type”: “section”, “text”: { “type”: “mrkdwn”, “text”: f”CPU 상위 프로세스\n{proctext}” } }, { “type”: “context”, “elements”: [ { “type”: “mrkdwn”, “text”: f”자동 생성된 서버 모니터링 리포트 | {servername} | {collectedat}” } ] } ] return {“attachments”: [{“color”: color, “blocks”: blocks}]} def sendcriticalalert(status: dict) -> bool: “””위험 상태 발생 시 긴급 알림 전송””” criticalitems = [] if status[‘cpu’][‘iscritical’]: criticalitems.append(f”CPU 사용률 {status[‘cpu’][‘usagepercent’]}%”) if status[‘memory’][‘iscritical’]: criticalitems.append(f”메모리 사용률 {status[‘memory’][‘usagepercent’]}%”) for disk in status[‘disks’]: if disk[‘iscritical’]: criticalitems.append(f”디스크 {disk[‘mountpoint’]} {disk[‘usagepercent’]}%”) alerttext = “, “.join(criticalitems) payload = { “text”: f”[긴급 경고] {status[‘servername’]} 서버 임계값 초과: {alerttext}” } return sendslackmessage(payload)

#!/usr/bin/env python3 — coding: utf-8 — “”” 서버 모니터링 메인 실행 스크립트 파일명: /opt/server-monitor/runmonitor.py “”” import sys import os import json import logging from pathlib import Path 경로 설정 BASEDIR = Path(‘/opt/server-monitor’) sys.path.insert(0, str(BASEDIR)) from scripts.monitor import collectserverstatus from scripts.slacknotifier import ( sendslackmessage, buildstatusreport, sendcriticalalert ) 로깅 설정 logging.basicConfig( filename=str(BASEDIR / ‘logs’ / ‘runmonitor.log’), level=logging.INFO, format=’%(asctime)s – %(levelname)s – %(message)s’ ) logger = logging.getLogger(name) STATEFILE = BASEDIR / ‘logs’ / ‘.laststate.json’ def loadlaststate(): “””이전 실행 상태 로드””” if STATEFILE.exists(): with open(STATEFILE, ‘r’) as f: return json.load(f) return {‘wascritical’: False} def savecurrentstate(status: dict): “””현재 실행 상태 저장””” state = {‘wascritical’: status[‘hascritical’]} with open(STATEFILE, ‘w’) as f: json.dump(state, f) def main(reportmode: str = ‘auto’): “”” 메인 실행 함수 reportmode: ‘auto’ (임계값 초과 시만), ‘full’ (항상 전체 리포트), ‘alert’ (경고만) “”” logger.info(f”모니터링 실행 시작 (모드: {reportmode})”) # 서버 상태 수집 status = collectserverstatus() laststate = loadlaststate() if reportmode == ‘full’: # 전체 리포트 전송 (정기 리포트용) payload = buildstatusreport(status) sendslackmessage(payload) logger.info(“전체 리포트 전송 완료”) elif reportmode == ‘auto’ or reportmode == ‘alert’: # 위험 상태 진입 시 긴급 알림 if status[‘hascritical’] and not laststate[‘wascritical’]: sendcriticalalert(status) payload = buildstatusreport(status) sendslackmessage(payload) logger.warning(“위험 상태 감지 – 긴급 알림 전송”) # 위험 상태에서 정상으로 복구 시 복구 알림 elif not status[‘hascritical’] and laststate[‘wascritical’]: recoverypayload = { “attachments”: [{ “color”: “#0000FF”, “text”: f”[복구] {status[‘servername’]} 서버가 정상 상태로 복구되었습니다. ({status[‘collectedat’]})” }] } sendslackmessage(recoverypayload) logger.info(“복구 알림 전송 완료”) # 상태 저장 savecurrentstate(status) logger.info(“모니터링 실행 완료”) if name == ‘main’: mode = sys.argv[1] if len(sys.argv) > 1 else ‘auto’ main(mode)

# 서버 모니터링 cron 설정 가상 환경의 Python 경로를 직접 지정하여 환경 문제 방지 PYTHON=/opt/server-monitor/venv/bin/python3 MONITORSCRIPT=/opt/server-monitor/runmonitor.py 5분마다 임계값 감시 (위험 상태 감지 시 즉시 슬랙 알림) /5 $PYTHON $MONITORSCRIPT auto >> /opt/server-monitor/logs/cron.log 2>&1 매일 오전 9시에 전체 서버 상태 리포트 전송 (일일 리포트) 0 9 $PYTHON $MONITORSCRIPT full >> /opt/server-monitor/logs/cron.log 2>&1 매주 월요일 오전 8시 50분에 주간 요약 리포트 전송 50 8 1 $PYTHON $MONITORSCRIPT full >> /opt/server-monitor/logs/cron.log 2>&1

/opt/server-monitor/logs/.log { daily rotate 30 compress delaycompress missingok notifempty create 0644 ubuntu ubuntu dateext dateformat -%Y%m%d }

def checkhttpendpoints(endpoints: list) -> list: “””HTTP 엔드포인트 가용성 및 응답 시간 체크””” import requests import time results = [] for url in endpoints: starttime = time.time() try: response = requests.get(url, timeout=10, allowredirects=True) elapsedms = round((time.time() – starttime) 1000, 2) results.append({ ‘url’: url, ‘statuscode’: response.statuscode, ‘responsetimems’: elapsedms, ‘ishealthy’: response.statuscode < 400, ‘isslow’: elapsedms > 2000 }) except requests.exceptions.RequestException as e: results.append({ ‘url’: url, ‘statuscode’: None, ‘responsetimems’: None, ‘ishealthy’: False, ‘error’: str(e) }) return results

#!/bin/bash 서버 모니터링 시스템 자동 설치 스크립트 setup.sh set -e echo “=== 서버 모니터링 시스템 설치 시작 ===” 디렉토리 생성 sudo mkdir -p /opt/server-monitor/{scripts,logs,config} sudo chown -R $USER:$USER /opt/server-monitor Python

가상 환경 생성 및 패키지 설치 cd /opt/server-monitor python3 -m venv venv source venv/bin/activate pip install psutil requests python-dotenv 스크립트 파일 권한 설정 chmod +x /opt/server-monitor/runmonitor.py 환경 변수 파일 생성 (값은 수동으로 입력 필요) if [ ! -f /opt/server-monitor/config/.env ]; then cat > /opt/server-monitor/config/.env << ‘EOF’ SLACKWEBHOOKURL=https://hooks.slack.com/services/YOUR/WEBHOOK/URL SERVERNAME=my-server-01 CPUTHRESHOLD=80 MEMORYTHRESHOLD=85 DISKTHRESHOLD=90 LOGLEVEL=INFO EOF echo “config/.env 파일이 생성되었습니다. SLACKWEBHOOKURL을 반드시 수정하세요.” fi logrotate 설정 sudo tee /etc/logrotate.d/server-monitor > /dev/null << ‘EOF’ /opt/server-monitor/logs/.log { daily rotate 30 compress delaycompress missingok notifempty } EOF crontab 설정 PYTHONPATH=/opt/server-monitor/venv/bin/python3 SCRIPTPATH=/opt/server-monitor/runmonitor.py (crontab -l 2>/dev/null; echo “/5 $PYTHONPATH $SCRIPTPATH auto >> /opt/server-monitor/logs/cron.log 2>&1”) | crontab – (crontab -l 2>/dev/null; echo “0 9 * $PYTHONPATH $SCRIPTPATH full >> /opt/server-monitor/logs/cron.log 2>&1”) | crontab – echo “=== 설치 완료 ===” echo “config/.env 파일에서 SLACKWEBHOOKURL을 설정하세요.” echo “설치 확인: crontab -l”