이 가이드는 Ubuntu 24.04.1 LTS 및 Python 에 맞게 작성되었습니다. 본 코드의 명칭은 watchcore이며, chatGPT를 통해 작성되었습니다.
2. Python 코드
24-12-11. watchcore 상태 정보가 corekeeper에 의존적인 문제를 수정하였습니다.
Python Requirement는 별도로 작성하지 않았으며, 필요한 Module은 설치하고 코드를 기동합니다.
#!/usr/bin/env python3
import subprocess
import requests
import schedule
import time
import logging
import signal
import sys
from datetime import datetime
import re
# 로깅 설정
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler()]
)
# Discord 웹훅 URL 및 관련 변수 설정
WEBHOOK_URL = "https://discord.com/api/v10/webhooks/123123123123/qgqgdqgqeg"
SERVICE_CK = "corekeeper" # CoreKeeper 서비스 이름
SERVICE_WC = "watchcore" # WatchCore 서비스 이름
TARGET_STRING = "Game ID:" # 로그에서 게임 ID를 찾기 위한 문자열
WATCHCORE_STATUS_MESSAGE = "ID:" # Watchcore 상태 메시지에서 찾을 문자열
MESSAGE_ID = None # Discord 메시지 ID 초기화
LAST_STATUS_MESSAGE = None # 마지막 서버 상태 메시지
# 초기 메시지 ID를 설정하는 함수
def initialize_message_id():
global MESSAGE_ID
if MESSAGE_ID is None:
MESSAGE_ID = get_last_message_id_from_journal()
if MESSAGE_ID:
logging.info(f"초기 메시지 ID 설정: {MESSAGE_ID}")
else:
logging.info("초기 메시지 ID를 찾지 못했습니다.")
# journalctl에서 마지막 메시지 ID를 찾는 함수
def get_last_message_id_from_journal():
try:
# 'watchcore' 서비스의 로그에서 마지막 메시지 ID를 추출
process = subprocess.run(
["journalctl", "-u", SERVICE_WC, "-n", "20", "--grep", WATCHCORE_STATUS_MESSAGE],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
if process.returncode == 0 and process.stdout:
for line in process.stdout.splitlines():
message_id = extract_message_id_from_log(line)
if message_id:
return message_id
return None
except Exception as e:
logging.error(f"journalctl에서 ID 추출 중 오류 발생: {e}", exc_info=True)
return None
# 로그에서 메시지 ID를 추출하는 함수
def extract_message_id_from_log(log_line):
match = re.search(r"ID:\s*(\d+)", log_line)
if match:
return match.group(1)
return None
# Discord 메시지를 전송하거나 업데이트하는 함수
def send_or_update_discord_message(content):
global MESSAGE_ID
data = {"content": content}
try:
# 기존 메시지 ID가 유효하지 않으면 새로운 메시지를 생성
if MESSAGE_ID:
if not is_message_id_valid(MESSAGE_ID):
logging.warning("MESSAGE_ID가 유효하지 않습니다. 새로운 메시지를 생성합니다.")
MESSAGE_ID = None
# 메시지가 이미 존재하면 PATCH로 업데이트, 없으면 새로 생성
if MESSAGE_ID:
response = requests.patch(f"{WEBHOOK_URL}/messages/{MESSAGE_ID}", json=data)
else:
response = requests.post(f"{WEBHOOK_URL}?wait=1", json=data)
if response.status_code in {200, 204}: # 성공적인 응답 처리
if not MESSAGE_ID:
message_data = response.json()
MESSAGE_ID = message_data.get("id")
logging.info(f"메시지 생성 성공, ID: {MESSAGE_ID}")
else:
logging.info("메시지 업데이트 성공")
else:
logging.warning(f"메시지 처리 실패: {response.status_code}, {response.text}")
except Exception as e:
logging.error(f"메시지 처리 중 오류 발생: {e}", exc_info=True)
# 메시지 ID 유효성 체크 함수
def is_message_id_valid(message_id):
try:
response = requests.get(f"{WEBHOOK_URL}/messages/{message_id}")
return response.status_code == 200
except Exception as e:
logging.error(f"메시지 유효성 확인 중 오류 발생: {e}", exc_info=True)
return False
# CoreKeeper 서비스 상태를 확인하고 Discord 메시지를 업데이트하는 함수
def check_service_status():
global LAST_STATUS_MESSAGE
try:
# systemctl 명령어로 CoreKeeper 서비스 상태 확인
process = subprocess.run(
["systemctl", "is-active", SERVICE_CK],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# CoreKeeper 상태에 따른 메시지 설정
if process.returncode == 0 and process.stdout.strip() == "active":
# CoreKeeper 서버가 정상일 때
last_start_time = get_last_start_time()
server_id = get_server_id() or "알 수 없음"
corekeeper_status = f"{SERVICE_CK} 서버가 정상입니다."
else:
# CoreKeeper 서버가 비정상일 때
last_start_time = "알 수 없음"
server_id = "알 수 없음"
corekeeper_status = f"{SERVICE_CK} 서버가 비정상입니다."
# Watchcore 상태는 항상 실행 중으로 설정
watchcore_status = "Watchcore가 실행 중입니다."
# 새 상태 메시지 구성
new_status_message = f"""**서버 ID**
```{server_id}```
**서버 상태**
```{corekeeper_status}```
**마지막 서버 기동 시간**
```{last_start_time}```
**Watchcore 상태**
```{watchcore_status}```"""
# 메시지가 변경된 경우에만 Discord 업데이트
if new_status_message != LAST_STATUS_MESSAGE:
send_or_update_discord_message(new_status_message)
LAST_STATUS_MESSAGE = new_status_message
except Exception as e:
logging.error(f"서버 상태 확인 중 오류 발생: {e}", exc_info=True)
# 마지막 서버 기동 시간을 확인하는 함수
def get_last_start_time():
try:
process = subprocess.run(
["journalctl", "-u", SERVICE_CK, "--reverse", "-n", "10", "--grep", TARGET_STRING],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
if process.returncode == 0 and process.stdout:
for line in process.stdout.splitlines():
if TARGET_STRING in line:
timestamp_str = " ".join(line.split()[:3])
full_timestamp = datetime.strptime(f"{datetime.now().year} {timestamp_str}", "%Y %b %d %H:%M:%S")
return full_timestamp.strftime("%Y년 %m월 %d일 %H시 %M분 %S초")
return "알 수 없음"
except Exception as e:
logging.error(f"시간 추출 중 오류 발생: {str(e)}", exc_info=True)
return "시간 추출 중 오류 발생"
# 서버 ID를 가져오는 함수
def get_server_id():
try:
process = subprocess.run(
["journalctl", "-u", SERVICE_CK, "--reverse", "-n", "10"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
for line in process.stdout.splitlines():
if TARGET_STRING in line:
return line.split(TARGET_STRING, 1)[1].strip()
return None
except Exception as e:
logging.error(f"서버 ID 추출 중 오류 발생: {str(e)}", exc_info=True)
return None
# SIGTERM 신호 처리 함수, 프로그램 종료 시 Discord에 종료 메시지를 보냄
def signal_handler(sig, frame):
logging.info("SIGTERM 신호 발생, Watchcore를 종료합니다.")
if MESSAGE_ID:
send_or_update_discord_message("**Watchcore 상태**\n```Watchcore가 종료되었습니다.```")
sys.exit(0)
# 주기적으로 서버 상태를 확인하고 메시지를 업데이트하는 watchdog 함수
def watchdog():
logging.info("Watchdog이 실행 중입니다.")
initialize_message_id() # 초기 메시지 ID 설정
signal.signal(signal.SIGTERM, signal_handler) # SIGTERM 신호 처리기 설정
# 5초마다 서버 상태를 체크하는 스케줄러 설정
schedule.every(5).seconds.do(check_service_status)
while True:
try:
schedule.run_pending()
time.sleep(5) # 5초 대기 후 반복
except Exception as e:
logging.error(f"에러 발생: {e}", exc_info=True)
time.sleep(5)
# 프로그램 실행
if __name__ == "__main__":
watchdog()
3. Systemd 서비스 설정
3.1 Systemd 서비스 파일 생성
watchcore를 systemd로 관리하기 위한 서비스 파일을 생성합니다.
3.2.1 서비스 파일 생성
다음 내용을 /etc/systemd/system/watchcore.service에 추가합니다.
