추석 연휴 리눅스 서버 점검 스크립트
연휴 출근을 막기 위한 서버 사전 점검
연휴가 길어질수록 서버 엔지니어에게는 불안한 시간이기도 하다. 평소라면 바로 대응했을 장애가, 하필 연휴 중에 터지면 결국 출근길에 오르게 되는 경우가 종종 있다.
이 글은 그런 상황을 조금이라도 줄여보기 위해, 연휴 시작 전 미리 서버 상태를 점검해두는 방법과 그 과정에서 만든 스크립트를 정리한 기록이다.
"""
서버 점검 스크립트 (Windows PC에서 실행)
- SSH ID/PW로 접속 후 sudo su root로 권한 상승
- CPU / 메모리 / 디스크 사용량 조회 (설정 변경 없음, 조회 전용 명령만 사용)
- 사용률이 임계치를 넘으면 "이상" 으로 자동 표시
- 결과를 텍스트 파일(메모장으로 열림)로 저장
사전 준비 (Windows PC, 최초 1회):
pip install paramiko
"""
import paramiko
import time
import re
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor
# ===================== 설정 =====================
# SSH 접속 정보 (하드코딩)
SSH_USER = "여기에_ID_입력"
SSH_PASS = "여기에_PW_입력"
SSH_PORT = 22
# 점검할 서버 목록 (IP 또는 호스트명)
SERVER_LIST = [
"10.110.1.11",
"10.110.1.12",
"10.110.1.13",
# 필요한 만큼 추가
]
BATCH_SIZE = 10 # 동시 접속 대수 (서버 부하 방지용)
DELAY_SEC = 2 # 배치 사이 대기 시간(초)
# 임계치 (%) - 이 값을 넘으면 "이상"으로 표시
CPU_THRESHOLD = 80
MEM_THRESHOLD = 80
DISK_THRESHOLD = 80
MARK_START = "___CHK_START___"
MARK_END = "___CHK_END___"
# ===================== 원격 셸(root) 유틸 =====================
def open_root_shell(host):
"""SSH 접속 후 sudo su - root 로 전환된 인터랙티브 셸(channel)을 반환"""
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(
hostname=host,
port=SSH_PORT,
username=SSH_USER,
password=SSH_PASS,
timeout=5,
banner_timeout=5,
auth_timeout=5,
)
chan = ssh.invoke_shell()
time.sleep(1)
_drain(chan) # 최초 로그인 배너 비우기
# 입력 에코를 꺼서, 보낸 명령이 그대로 되돌아와 마커 파싱을 깨는 문제 방지
chan.send("stty -echo\n")
time.sleep(0.5)
_drain(chan)
chan.send("sudo su -\n")
time.sleep(1)
_drain(chan)
# 프롬프트 개행 등 필터링을 쉽게 하기 위해 색상/이스케이프 최소화
chan.send("unset PROMPT_COMMAND; PS1='#-#-# '\n")
time.sleep(0.5)
_drain(chan)
return ssh, chan
def _drain(chan, wait=0.3):
"""채널 버퍼에 쌓인 내용을 읽어서 반환"""
time.sleep(wait)
data = ""
while chan.recv_ready():
data += chan.recv(65536).decode(errors="ignore")
time.sleep(0.1)
return data
def run_cmd(chan, cmd, timeout=10):
"""마커로 감싸서 명령을 실행하고, 그 명령의 순수 출력만 파싱해서 반환"""
full_cmd = f'echo {MARK_START}; {cmd}; echo {MARK_END}\n'
chan.send(full_cmd)
start_t = time.time()
buf = ""
while time.time() - start_t < timeout:
buf += _drain(chan, wait=0.2)
if MARK_END in buf:
break
# MARK_START 다음 ~ MARK_END 이전 구간만 추출
m = re.search(re.escape(MARK_START) + r"(.*?)" + re.escape(MARK_END), buf, re.DOTALL)
if not m:
return buf.strip()
output = m.group(1)
# 맨 앞의 개행 및 에코된 명령줄 찌꺼기 제거
lines = output.splitlines()
lines = [l for l in lines if l.strip() and MARK_START not in l]
return "\n".join(lines).strip()
# ===================== 판정 로직 =====================
def check_cpu(vmstat_output):
"""vmstat 마지막 줄에서 idle(15번째 컬럼)을 읽어 사용률 계산"""
lines = [l for l in vmstat_output.splitlines() if l.strip()]
if not lines:
return None, "vmstat 결과 없음"
last = lines[-1].split()
try:
idle = float(last[14]) # 0-based index 14 = 15번째 컬럼
usage = round(100 - idle, 1)
return usage, None
except (IndexError, ValueError):
return None, "CPU 파싱 실패"
def check_mem(mem_pct_output):
try:
usage = round(float(mem_pct_output.strip()), 1)
return usage, None
except ValueError:
return None, "메모리 파싱 실패"
def check_disk(df_output):
"""마운트포인트별 사용률(%) 목록 반환: [(mount, pct), ...]"""
results = []
for line in df_output.splitlines():
parts = line.split()
if len(parts) < 2:
continue
mount, pct = parts[0], parts[1]
pct_num = pct.replace("%", "")
if pct_num.isdigit():
results.append((mount, int(pct_num)))
return results
# ===================== 서버별 점검 함수 =====================
def check_server(host):
lines = []
lines.append(f"\n{'='*60}")
lines.append(f"서버: {host}")
lines.append(f"{'='*60}")
try:
ssh, chan = open_root_shell(host)
# --- 사람이 보기 좋은 원본 출력 ---
top_raw = run_cmd(chan, "top -bn1 | head -15")
free_raw = run_cmd(chan, "free -h")
df_raw = run_cmd(chan, "df -hT")
lines.append("\n--- CPU (top) ---")
lines.append(top_raw if top_raw else "(결과 없음)")
lines.append("\n--- 메모리 (free -h) ---")
lines.append(free_raw if free_raw else "(결과 없음)")
lines.append("\n--- 디스크 (df -hT) ---")
lines.append(df_raw if df_raw else "(결과 없음)")
# --- 임계치 판정용 수치 데이터 ---
vmstat_raw = run_cmd(chan, "vmstat 1 2")
mem_pct_raw = run_cmd(chan, "free | awk '/Mem:/ {printf \"%.1f\", $3/$2*100}'")
df_pct_raw = run_cmd(chan, "df -hP | awk 'NR>1 {print $6, $5}'")
cpu_usage, cpu_err = check_cpu(vmstat_raw)
mem_usage, mem_err = check_mem(mem_pct_raw)
disk_usages = check_disk(df_pct_raw)
lines.append("\n--- 이상 여부 판정 ---")
issues = []
if cpu_usage is not None:
status = "이상(임계치 초과)" if cpu_usage >= CPU_THRESHOLD else "정상"
lines.append(f"CPU 사용률: {cpu_usage}% - {status} (기준 {CPU_THRESHOLD}%)")
if cpu_usage >= CPU_THRESHOLD:
issues.append(f"CPU {cpu_usage}%")
else:
lines.append(f"CPU 사용률: 확인 불가 ({cpu_err})")
if mem_usage is not None:
status = "이상(임계치 초과)" if mem_usage >= MEM_THRESHOLD else "정상"
lines.append(f"메모리 사용률: {mem_usage}% - {status} (기준 {MEM_THRESHOLD}%)")
if mem_usage >= MEM_THRESHOLD:
issues.append(f"메모리 {mem_usage}%")
else:
lines.append(f"메모리 사용률: 확인 불가 ({mem_err})")
if disk_usages:
for mount, pct in disk_usages:
status = "이상(임계치 초과)" if pct >= DISK_THRESHOLD else "정상"
lines.append(f"디스크 [{mount}]: {pct}% - {status} (기준 {DISK_THRESHOLD}%)")
if pct >= DISK_THRESHOLD:
issues.append(f"디스크 {mount} {pct}%")
else:
lines.append("디스크 사용률: 확인 불가")
if issues:
lines.append(f"\n>>> [경고] 임계치 초과 항목: {', '.join(issues)}")
else:
lines.append("\n>>> 전체 정상")
ssh.close()
except Exception as e:
lines.append(f"[접속/실행 실패] {e}")
return host, "\n".join(lines)
def batch_check(hosts):
all_text = []
for i in range(0, len(hosts), BATCH_SIZE):
batch = hosts[i:i + BATCH_SIZE]
print(f"[진행중] {i+1} ~ {min(i+BATCH_SIZE, len(hosts))} / {len(hosts)} 대 점검 중...")
with ThreadPoolExecutor(max_workers=len(batch)) as executor:
futures = [executor.submit(check_server, h) for h in batch]
for f in futures:
host, text = f.result()
all_text.append(text)
if i + BATCH_SIZE < len(hosts):
time.sleep(DELAY_SEC)
return all_text
# ===================== 실행 =====================
if __name__ == "__main__":
start_time = datetime.now()
print(f"\n점검 시작: {start_time.strftime('%Y-%m-%d %H:%M:%S')}")
print(f"대상 서버 수: {len(SERVER_LIST)}대\n")
results = batch_check(SERVER_LIST)
end_time = datetime.now()
filename = f"server_check_result_{start_time.strftime('%Y%m%d_%H%M%S')}.txt"
with open(filename, "w", encoding="utf-8") as f:
f.write("서버 점검 결과\n")
f.write(f"점검 시작: {start_time.strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write(f"점검 종료: {end_time.strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write(f"대상 서버 수: {len(SERVER_LIST)}대\n")
f.write(f"임계치: CPU {CPU_THRESHOLD}% / 메모리 {MEM_THRESHOLD}% / 디스크 {DISK_THRESHOLD}%\n")
f.write("\n".join(results))
f.write("\n\n" + "="*60 + "\n점검 완료\n")
print(f"\n점검 완료! 결과 파일: {filename}")
print("스크립트와 같은 폴더에 저장되었습니다. 메모장으로 열어 확인하세요.")
1. import 부분 (12~17번 줄)
import paramiko
import time
import re
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor- paramiko: 파이썬에서 SSH 접속을 가능하게 해주는 외부 라이브러리 (MobaXterm 같은 걸 코드로 하는 거라 생각하면 됩니다)
- time: sleep()으로 몇 초 기다리는 용도
- re: 정규표현식(regex) — 긴 텍스트에서 원하는 패턴만 뽑아낼 때 사용
- datetime: 현재 시각을 가져와서 파일명이나 로그에 시간 찍는 용도
- ThreadPoolExecutor: 여러 서버를 동시에 점검하기 위한 멀티스레드 도구
2. 설정값 (19~43번 줄)
SSH_USER = "여기에_ID_입력"
SSH_PASS = "여기에_PW_입력"
SERVER_LIST = ["10.110.1.11", ...]
BATCH_SIZE = 10
CPU_THRESHOLD = 80
MARK_START = "___CHK_START___"이건 그냥 **"변수 선언"**입니다. 나중에 코드에서 SSH_USER라고 쓰면 여기 적힌 값이 그대로 들어갑니다. 프로그램 앞부분에 이렇게 설정값을 모아두면, 나중에 값만 바꾸고 싶을 때 코드 로직은 안 건드리고 여기만 고치면 됩니다. (실력 늘리는 팁: 하드코딩할 값은 위에 몰아두는 습관이 좋은 코드의 시작입니다.)
MARK_START/MARK_END는 왜 필요했는지는 4번에서 설명합니다.
3. open_root_shell() — 접속해서 root까지 들어가기 (48~79번 줄)
def open_root_shell(host):
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(hostname=host, port=SSH_PORT, username=SSH_USER, password=SSH_PASS, ...)- def 함수이름(매개변수): → 함수 정의. host라는 값을 받아서 실행되는 코드 블록입니다.
- paramiko.SSHClient() → SSH 접속 객체를 하나 만듭니다 (빈 껍데기)
- set_missing_host_key_policy(...) → 처음 접속하는 서버라 "known_hosts"에 등록 안 돼있어도 그냥 접속 허용 (MobaXterm에서 "이 서버 처음 접속하는데 계속하시겠습니까?" 물어보는 걸 자동으로 Yes 처리하는 것과 같음)
- ssh.connect(...) → 실제로 접속 시도. ID/PW/타임아웃 넣어서 로그인
chan = ssh.invoke_shell()
time.sleep(1)
_drain(chan)- invoke_shell() → 여기가 핵심입니다. exec_command()(명령 하나 던지고 결과 받고 끝)와 다르게, 터미널 화면을 하나 띄워놓고 계속 타이핑하듯 쓸 수 있는 채널을 엽니다. sudo su처럼 "비밀번호 입력하세요" 같은 대화형 작업을 하려면 이 방식이 필요합니다.
- time.sleep(1) → 접속 직후 배너(환영 메시지 등)가 뜨는 시간을 기다려줍니다
- _drain(chan) → 지금까지 화면에 쌓인 내용을 읽어서 버립니다 (안 읽으면 버퍼에 계속 쌓여서 나중에 파싱할 때 방해됨)
chan.send("stty -echo\n")- chan.send(...) = 터미널에 타이핑하는 것과 동일. \n은 엔터키
- stty -echo는 "내가 입력한 글자를 화면에 그대로 반사하지 마라"는 명령 — 이걸 꺼야 나중에 마커 파싱이 꼬이지 않습니다 (아까 겪으신 문제의 해결책)
chan.send("sudo su -\n")- 말씀하신 환경에선 비밀번호 없이 바로 root로 넘어가니, 그냥 이 한 줄만 보내면 됩니다
chan.send("unset PROMPT_COMMAND; PS1='#-#-# '\n")- PS1은 리눅스 프롬프트 모양([root@server ~]# 같은 거)을 정하는 변수. 색깔 코드나 복잡한 프롬프트가 있으면 나중에 텍스트 파싱할 때 이상한 특수문자가 섞이니, 아주 단순한 문자열로 바꿔서 깔끔하게 만드는 겁니다
return ssh, chan- 함수가 끝나면 ssh(연결 자체)와 chan(대화형 터미널)을 둘 다 돌려줍니다. 나중에 다른 함수에서 이 둘을 계속 재사용합니다.
4. _drain()과 run_cmd() — 명령 실행하고 결과만 쏙 빼오기 (82~112번 줄)
이 부분이 이 스크립트에서 가장 어려운 핵심 트릭입니다.
def _drain(chan, wait=0.3):
time.sleep(wait)
data = ""
while chan.recv_ready():
data += chan.recv(65536).decode(errors="ignore")
time.sleep(0.1)
return data- recv_ready(): "지금 읽을 데이터가 도착해있냐?"를 물어보는 것
- while 루프: 데이터가 남아있는 동안 계속 읽음 (한 번에 다 안 올 수도 있어서)
- recv(65536): 최대 65536바이트만큼 읽기
- .decode(errors="ignore"): 컴퓨터는 데이터를 바이트(숫자)로 주고받는데, 이걸 사람이 읽는 문자(한글/영어)로 변환. 혹시 깨지는 바이트가 있어도 에러 없이 무시하고 넘어감
왜 이런 함수가 필요한가? SSH 통신은 "물어보면 바로 답이 오는" 방식이 아니라, 네트워크로 데이터가 뚝뚝 끊어져서 옵니다. 그래서 "지금 도착한 만큼만" 여러 번 나눠 읽어야 합니다.
def run_cmd(chan, cmd, timeout=10):
full_cmd = f'echo {MARK_START}; {cmd}; echo {MARK_END}\n'
chan.send(full_cmd)- 여기가 아까 문제됐던 부분입니다. 왜 명령어 앞뒤에 echo로 마커를 붙일까요?
- 터미널 화면에는 우리가 보낸 명령, 그 명령의 실제 결과, 그다음 프롬프트가 뒤섞여서 줄줄이 나옵니다. 그 중에서 "진짜 결과"만 골라내려면 눈에 띄는 표식(마커)을 결과 앞뒤에 심어두고, 그 사이만 잘라내는 방식을 씁니다.
- 예를 들어 top -bn1을 실행하면 화면에는:
___CHK_START___
(top 명령의 실제 출력...)
___CHK_END___이렇게 나오고, 우리는 이 두 마커 사이만 가져오면 됩니다.
start_t = time.time()
buf = ""
while time.time() - start_t < timeout:
buf += _drain(chan, wait=0.2)
if MARK_END in buf:
break- timeout(10초) 안에 종료마커(MARK_END)가 나올 때까지 계속 조금씩 읽어서 buf(버퍼, 임시 저장공간)에 쌓습니다
- 마커가 보이면 명령이 끝났다고 판단하고 루프 탈출
m = re.search(re.escape(MARK_START) + r"(.*?)" + re.escape(MARK_END), buf, re.DOTALL)- 이게 정규표현식입니다. 풀어서 설명하면:
- re.escape(MARK_START) : 마커 문자열을 정규식에서 그대로 문자 취급하도록 이스케이프 처리
- (.*?) : "아무 문자나, 최소한으로(non-greedy)" — 시작 마커와 끝 마커 사이의 내용을 잡아옴
- re.DOTALL : .이 줄바꿈 문자까지 포함하도록 (여러 줄짜리 결과도 다 잡게)
- 즉 "MARK_START 다음부터 MARK_END 전까지의 텍스트"를 통째로 뽑아내는 코드입니다
lines = [l for l in lines if l.strip() and MARK_START not in l]- 이건 리스트 컴프리헨션이라는 파이썬 문법입니다. 풀어 쓰면:
lines2 = []
for l in lines:
if l.strip() and MARK_START not in l:
lines2.append(l)- "빈 줄이 아니고, 마커 글자가 섞여있지 않은 줄만 남겨라" — 찌꺼기 제거용
5. 판정 함수들 (117~150번 줄)
def check_cpu(vmstat_output):
lines = [l for l in vmstat_output.splitlines() if l.strip()]
last = lines[-1].split()
idle = float(last[14])- splitlines() : 텍스트를 줄바꿈 기준으로 잘라서 리스트로 만듦
- lines[-1] : 리스트의 맨 마지막 항목 (vmstat은 첫 실행 결과 다음에 "실측값"이 두 번째 줄로 나오는데, 그게 진짜 데이터라 마지막 줄을 씀)
- .split() : 공백 기준으로 잘라서 단어별 리스트로 만듦 (vmstat 출력은 여러 숫자가 띄어쓰기로 나열돼있음)
- last[14] : 그 중 15번째 값(파이썬은 0부터 세니까 인덱스 14)이 idle(유휴)% 값
- 100 - idle : CPU 사용률 = 100에서 유휴율을 뺀 값
check_mem, check_disk도 같은 원리로 "문자열 → 숫자 변환 → 계산"만 다르게 한 겁니다.
6. check_server() — 한 대의 서버를 처음부터 끝까지 처리 (155~223번 줄)
def check_server(host):
lines = []
...
try:
ssh, chan = open_root_shell(host)
top_raw = run_cmd(chan, "top -bn1 | head -15")
...- 지금까지 만든 함수들을 순서대로 조립하는 함수입니다: 접속 → 명령 실행 → 결과 정리 → 판정 → 텍스트로 합치기
- try / except: "혹시 에러 나면 프로그램 전체가 죽지 말고, 그 서버만 실패 처리하고 넘어가라"는 안전장치. 100대 중 1대가 접속 안 되도 나머지 99대는 계속 점검되게 해줍니다
7. batch_check() — 여러 서버를 동시에 (226~238번 줄)
with ThreadPoolExecutor(max_workers=len(batch)) as executor:
futures = [executor.submit(check_server, h) for h in batch]
for f in futures:
host, text = f.result()- 서버를 한 대씩 순서대로 하면 100대면 엄청 오래 걸립니다. 그래서 10대씩 묶어서 동시에 접속하도록 스레드(작업 실행 흐름)를 여러 개 띄웁니다
- executor.submit(check_server, h) : "check_server(h)를 백그라운드에서 실행해줘" 라고 예약
- f.result() : 그 작업이 끝날 때까지 기다렸다가 결과값을 받아옴
8. 메인 실행부 (243~264번 줄)
if __name__ == "__main__":- "이 파일을 직접 실행했을 때만" 아래 코드를 돌려라 (다른 파일에서 이 파일을 불러다 쓸 때는 자동 실행 안 되게 하는 파이썬 관례)
with open(filename, "w", encoding="utf-8") as f:
f.write(...)- open(파일명, "w") : 쓰기 모드로 파일 생성. with를 쓰면 다 쓰고 나서 자동으로 파일을 닫아줌 (안 닫으면 파일이 잠기거나 내용이 덜 써질 수 있음)
'infra > Linux & Windows' 카테고리의 다른 글
| Rocky Linux 서버, 관리자 명령어 실행 이력은 어떻게 남기고 있을까 (bash trap + rsyslog 감사 구조 분석) (0) | 2026.09.10 |
|---|---|
| HCI VMware 환경에서 Ansible 운영 가이드 — 300대 VM을 효율적으로 관리하는 법 (0) | 2026.04.29 |
| 리눅스 서버 관리자가 알아야 할 기본 명령어 & 로그 확인 가이드 (0) | 2026.04.21 |
| nginx 포트 설정, 어디서 오는 걸까? — 실전 트러블슈팅 (0) | 2026.04.10 |
| journalctl 완벽 정리 — Rocky Linux 9 systemd 로그 관리 (0) | 2026.04.07 |