레거시 서비스 중 윈도우 어플리케이션에 종종 문제가 발생할 때가 있습니다.
이를 수정하는게 올바른 방법이겠지만 현실적인 문제로 관리자들이 일정 기간마다 재시작을 수행할 때가 있고, 가끔 다운되면 주말에도 접속해서 재가동 해야 하는 일이 발생하곤 합니다.
pm2와 같은 윈도우용 프로그램이 있으면 하는 생각을 해봤지만, 그건 주기적인 재실행 처리가 불가능해서 python으로 직접 만들어 보았습니다.
python은 자동 테스트 프로그램 개발을 위해 최근에 사용하기 시작했는데 생각보다 훨씬 재밌었습니다. C++/JS/Python을 넘나들면서 코딩하는게 꽤 귀찮긴 하지만...
Github(https://github.com/blackwitch/winPM)에도 같이 올려두었으니 참고하시기 바랍니다. 앞으로 편의를 위해 일부 기능을 개선/추가할 예정에 있습니다. 예를들어 지금은 이 프로그램을 콘솔 형태로 계속 띄워두어야 합니다. 이를 개선하기 위해 서비스 프로그램으로 수정할 예정입니다.
필요한 기능은 아래 두 가지였습니다.
- 정기적으로 재가동 할 것!
: 어떤 서비스는 장기적으로 가동될 때 메모리가 문제가 될 때가 있습니다. 이를 해소하는 가장 간편한 방법은 유저들이 가장 적은 시간에 몰래 재시작을 하는 거였죠. 물론 서비스가 재가동 되어도 유저들에게 전혀 문제가 없는 프로젝트이거나 유저가 0명인 시간대를 잘 골라야겠죠.
- 갑작스럽게 다운된 경우 즉시 재가동 할 것!
: 당연히 절대 다운되지 않을 완전한 서비스를 만들 수 있다면 좋겠지만 일반적으로 서비스는 출중한 한 명의 손에 의해서만 만들어지는 경우가 드물며, 그 한명도 천재이기 힘들고, 천재도 가끔 실수합니다. 게다가 그 천재가 자신이 만든 그 서비스를 영구히 수정/업데이트 하지 않죠. 종종 의도치 않게 다운되는 래거시 서비스들은 담당자들을 힘들게 합니다. 그래서 다운되는걸 못 막을 상황이라면 ... 즉시 재가동이 답이겠죠.
위 두 가지 기능을 넣고, 현재 상황을 확인할 수 있는 기능을 담기 위해 list, start, stop, restart의 명령을 처리할 수 있는 프로그램를 만들었습니다. 시스템이 다운되거나 급작스럽게 재부팅 될 경우가 있을 수 있으니 필요하다면 이 프로그램을 시작 프로그램에 등록 하는 것도 좋은 방법이라 생각되네요.
바로 코드로 들어가 보겠습니다.
import os, sys , subprocess, _thread
import win32gui, win32api, win32con, win32process
import psutil
import time
import logging
import json
import time
from datetime import datetime
# PMList에는 우리가 관리할 프로그램에 대한 정보와 현황을 담아둡니다.
# 코드에서는 관리할 프로그램들에 대한 작업을 task라는 명칭으로 사용하겠습니다.
PMList = dict() # process infomation list
# 등록된 프로그램의 full path로 키를 만들어 PMList의 키로 활용합니다.
def getPMKey(cfg_task):
try:
return hash(cfg_task["path"] + "/" + cfg_task["file"])
except Exception as e:
print("ERROR : Incorrect \"config.json\" file! ERROR => ", e)
sys.exit(0)
#############################################################
## crontab에 관련된 잘 정리된 코드가 있어 해당 코드를 활용합니다.
## crontab-like code is from https://github.com/idning/pcl/blob/master/pcl/crontab.py
class Event(object):
def __init__(self, desc, func, args=(), kwargs={}, use_thread=False):
"""
desc: min hour day month dow
day: 1 - num days
month: 1 - 12
dow: mon = 1, sun = 7
"""
self.desc = desc
self.func = func
self.args = args
self.kwargs = kwargs
self.use_thread = use_thread
#support:
# *
# 59
# 10,20,30
def _match(self, value, expr):
#print 'match', value, expr
if expr == '*':
return True
values = expr.split(',')
for v in values:
if int(v) == value:
return True
return False
def matchtime(self, t):
mins, hour, day, month, dow = self.desc.split()
return self._match(t.minute , mins) and\
self._match(t.hour , hour) and\
self._match(t.day , day) and\
self._match(t.month , month) and\
self._match(t.isoweekday() , dow)
def check(self, t):
if self.matchtime(t):
if self.use_thread:
thread.start_new_thread(self.func, self.args, self.kwargs)
else:
try:
self.func(*self.args, **self.kwargs)
except Exception as e:
logging.exception(e)
class Cron(object):
def __init__(self):
self.events = []
def add(self, desc, func, args=(), kwargs={}, use_thread=False):
self.events.append(Event(desc, func, args, kwargs, use_thread))
def run(self):
last_run = 0
while True:
#wait to a new minute start
t = time.time()
next_minute = t - t%60 + 60
while t < next_minute:
sleeptime = 60 - t%60
logging.debug('current time: %s, we will sleep %.2f seconds' %(t, sleeptime))
time.sleep(sleeptime)
t = time.time()
if last_run and next_minute - last_run != 60:
logging.warn('Cron Ignored: last_run: %s, this_time:%s' % (last_run, next_minute) )
last_run = next_minute
current = datetime(*datetime.now().timetuple()[:5])
for e in self.events:
e.check(current)
time.sleep(0.001)
## crontab-like codes
#############################################################
#############################################################
## command thread
## 유저 입력을 받아 필요한 정보를 출력하거나 지정한 명령을 수행하기 위한 코드입니다.
def command_thread():
while True:
try:
line = input('>> ')
params = line.split()
if len(params) == 0:
continue
cmd = params[0]
# 도움말을 볼 수 있는 command
if cmd == "?" or cmd =="h" or cmd =="help" :
print ("usage : <command> [options]")
print ("Command list")
print ("===============================")
print ("command | desc")
print ("list | show the process list you added")
print ("stop [title] | kill a process by the title in config")
print ("start [title] | start a process by the title in config")
print ("restart [title] | restart a process by the title in config")
print ("exit | exit this program ")
# 현재 등록된 task들의 현황을 보여줍니다.
elif cmd == "list":
print ( psutil.cpu_percent() , psutil.virtual_memory())
print ("---------------------------------------------------------------------------------------------")
print ("| %s | %s | %s | %s | %s | %s | %s |"% ("name".ljust(16)[:16],"job".ljust(12)[:12],"status".ljust(12)[:12],"restart".ljust(7)[:7],"cpu".ljust(4)[:4],"mem".ljust(8)[:8],"uptime".ljust(12)[:12]))
print ("---------------------------------------------------------------------------------------------")
try:
for key in PMList:
info = PMList[key]
print ("| %s | %s | %s | %s | %s | %s | %s |" % (info["name"].ljust(16)[:16],info["job"].ljust(12)[:12],info["status"].ljust(12)[:12],str(info["restart"]).ljust(7)[:7],str(info["cpu_usage"]).ljust(4)[:4],str(info["mem_usage"]).ljust(8)[:8],str(info["uptime"]).ljust(12)[:12]) )
except Exception:
import traceback
print (traceback.format_exc())
# winPM을 종료합니다. 가동중인 프로세스들은 그대로 둡니다.
elif cmd == "exit":
print("Press CTRL + C to exit this program.")
sys.exit(0)
# 멈춰있던 task를 실행합니다.
elif cmd == "start":
if len(params) < 2: # start 뒤에는 config.json 파일에 등록된 task 정보 중 title 값을 입력합니다.
print ("Invalid paramater(s). Type ""?"" will show you manpage")
continue
for key in PMList: # 입력된 task의 title 값은 PMList에 "name"으로 저장됩니다. 이를 비교해 해당 정보를 찾아 실행합니다.
if params[1] == PMList[key]["name"]:
startProcess(PMList[key])
break
# 실행중인 task를 중지합니다.
elif cmd == "stop":
if len(params) < 2: # stop 뒤에는 config.json 파일에 등록된 task 정보 중 title 값을 입력합니다.
print ("Invalid paramater(s). Type ""?"" will show you manpage")
continue
for key in PMList:
if params[1] == PMList[key]["name"]:
stopProcess(PMList[key])
break
# 실행중인 task를 재시작합니다.
elif cmd == "restart":
if len(params) < 2:
print ("Invalid paramater(s). Type ""?"" will show you manpage")
continue
for key in PMList:
if params[1] == PMList[key]["name"]:
stopProcess(PMList[key])
print("Wait for a sec")
time.sleep(3) # 중지에 많은 시간이 소요된다면.. 대기 시간을 늘리는 것도 좋은 방법입니다. config.json 파일 설정에까지는 넣지 않았습니다.
startProcess(PMList[key])
break
except KeyboardInterrupt:
print (" CTRL + C")
## command thread
#############################################################
# 실행 파일명으로 process id를 확인합니다.
def get_pid(process_name):
try:
processes = filter(lambda p: psutil.Process(p).name() == process_name, psutil.pids())
for pid in processes:
return pid
return 0
except Exception as e:
print ("GET_PID NOT FOUND ", process_name, e)
return 0
# pid로 window handle을 확인합니다.
def get_hwnd(pid):
def callback (hwnd, hwnds):
if win32gui.IsWindowVisible(hwnd) and win32gui.IsWindowEnabled(hwnd):
_,found_pid=win32process.GetWindowThreadProcessId(hwnd)
if found_pid==pid:
hwnds.append(hwnd)
return True
hwnds = []
win32gui.EnumWindows(callback, hwnds)
return hwnds[0] if hwnds else 0
# 실행 파일명으로 window handle을 확인합니다
def get_hwnd_byName(process_name):
pid = get_pid(process_name)
return get_hwnd(pid) if pid else 0
# 지정된 프로세스를 중단합니다.
def stopProcess(task):
HWnd = get_hwnd_byName(task["proc_name"])
if HWnd > 0:
# 프로세스 중단 시 필요하다면 중지 버튼과 종료 버튼을 눌러서 종료하게 합니다.
# 이미지 프로세싱을 넣을까 했지만 서비스 프로그램에 설마 이미지 버튼을 사용하지는 않겠죠?
try:
if task["buttons"]["STOP"]:
btnHnd= win32gui.FindWindowEx(HWnd, 0 , "Button", task["buttons"]["STOP"])
if btnHnd != 0:
win32api.SendMessage(btnHnd, win32con.BM_CLICK, 0, 0)
else:
print(task["buttons"]["STOP"], " button isn't there.")
time.sleep(5);
if task["buttons"]["QUIT"]:
btnHnd= win32gui.FindWindowEx(HWnd, 0 , "Button", task["buttons"]["QUIT"])
if btnHnd != 0:
win32api.SendMessage(btnHnd, win32con.BM_CLICK, 0, 0)
else:
print(task["buttons"]["QUIT"], " button isn't there.")
time.sleep(5);
except:
time.sleep(1)
# 종료 버튼이 지정되지 않거나 혹은 어떤 이유로 아직 종료된 상태가 아니라면 강제 종료 시킵니다.
# 종료 시 후속 작업이 많은 경우 별도 조정해서 사용하세요.
HWnd = get_hwnd_byName(task["proc_name"])
if HWnd != 0:
os.system("taskkill /f /im "+task["proc_name"])
time.sleep(1)
# 관리 정보들을 모두 초기화 합니다.
task["bStop"] = True
task["status"] = "stop"
task["pid"] = 0
task["hwnd"] = None
task["mem_usage"] = 0
task["cpu_usage"] = 0
task["uptime"] = 0
print (" STOP [" + task["name"] + "]")
# 지정된 프로세스를 가동합니다.
def startProcess(task):
# execute the app
HWnd = get_hwnd_byName(task["proc_name"])
boolAlreadyStart = True
# 현재 가동중인 상태가 아니면 프로세스를 가동합니다.
if HWnd == 0:
boolAlreadyStart = False
os.chdir(os.path.realpath(task["app_path"]))
myProcess = subprocess.Popen(task["app"])
time.sleep(5)
HWnd = get_hwnd_byName(task["proc_name"])
if HWnd == 0:
print(task["name"], " is failed to execute.")
else:
# 프로세스 가동 후 시작 버튼을 누를 필요가 있을 경우 START 버튼 라벨을 지정해 두세요.
if boolAlreadyStart == False:
try:
if task["buttons"]["START"]:
btnHnd= win32gui.FindWindowEx(HWnd, 0 , "Button", task["buttons"]["START"])
if btnHnd != 0:
win32api.SendMessage(btnHnd, win32con.BM_CLICK, 0, 0)
else:
print(task["buttons"]["START"], " button isn't there.")
except:
time.sleep(1)
task["bStop"] = False
task["status"] = "running"
task["restart"] = int(task["restart"]) + 1
print (" START [" + task["name"] + "]")
else:
print("[" + task["name"],"] is running already")
# main 함수
if __name__ == "__main__":
try:
# 분 단위로 task의 상태를 업데이트 합니다.
def task_update():
for key in PMList:
PMList[key]["pid"] = get_pid(PMList[key]["proc_name"])
if PMList[key]["hwnd"] == None and PMList[key]["pid"] != 0:
PMList[key]["hwnd"] = get_hwnd(PMList[key]["pid"])
else:
PMList[key]["hwnd"] = None
# 사용자가 stop 으로 프로세스를 중단한 상태("bStop"가 True일 경우)라면 재가동 시키지 않습니다.
if PMList[key]["pid"]> 0 and PMList[key]["bStop"] == False:
PMList[key]["status"] = psutil.Process(PMList[key]["pid"]).status()
PMList[key]["cpu_usage"] = psutil.Process(PMList[key]["pid"]).cpu_percent()
try:
PMList[key]["mem_usage"] = psutil.Process(PMList[key]["pid"]).memory_full_info().uss
except:
PMList[key]["mem_usage"] = 0
try:
pidcreated = datetime.fromtimestamp(psutil.Process(PMList[key]["pid"]).create_time())
diff = datetime.now() - pidcreated
PMList[key]["uptime"] =diff
except Exception:
import traceback
print (traceback.format_exc())
PMList[key]["uptime"] = -1
else:
PMList[key]["status"] = "stop"
PMList[key]["hwnd"] = None
PMList[key]["mem_usage"] = 0
PMList[key]["cpu_usage"] = 0
PMList[key]["uptime"] = 0
# config.json 에 정의된 task의 job을 처리합니다.
def runTask(task, schedule, cron):
myProcess = None
# 프로세스가 다운된 상태라면 재가동 합니다.
def sustenance_task():
if task["bStop"] == True:
return
pid = get_pid(task["proc_name"])
if pid == 0:
startProcess(task)
# 지정된 시간에 프로세스를 재가동합니다.
def restart_task():
if task["bStop"] == True:
return
stopProcess(task)
startProcess(task)
if task["job"] == "sustenance":
cron.add(schedule, sustenance_task)
elif task["job"] == "restart":
cron.add(schedule, restart_task)
else:
print(" Incorrect job type = ", task["job"])
# loading config file
# task에 대한 정의를 config.json이라는 파일에 별도 명시해야 합니다.
try:
with open('config.json') as config_file:
config = json.load( config_file )
if len(config['task']) <=0:
print(" There is no Task.")
except Exception as e:
print("ERROR : You need \"config.json\" file!", e)
sys.exit(0)
# task에 명시된 cron 정보로 스케쥴링에 등록합니다.
cron = Cron()
for cfg_task in config['task']:
key = getPMKey(cfg_task)
app_path = cfg_task["path"]
app =cfg_task["path"] + "/" + cfg_task["file"]
PMList[key] = {"key" : key, "app_path" : app_path, "app":app, "proc_name":cfg_task["file"],"buttons":cfg_task["buttons"], "job":cfg_task["job"], "hwnd": None, "pid": None, "name": cfg_task["title"],"bStop": False, "status": "stop", "cpu_usage": -1, "mem_usage": -1, "uptime": -1, "restart": 0}
runTask(PMList[key],cfg_task["schedule"]["cron"], cron)
# add task for updating task status
task_update()
cron.add("* * * * *", task_update)
# console interface
cmdThread = _thread.start_new_thread(command_thread, ())
cron.run()
except KeyboardInterrupt:
print(" BYE!! ")
sys.exit(0)
config.json 파일은 아래 예제를 참고하세요.
{
# task는 필요한 만큼 정의 할 수 있습니다.
"task": [
{
"name": "샘플 A 프로세스", # task에 대한 설명입니다. 작업에 영향을 미치지 않습니다.
"path": "D:/sampleA", # 실행할 파일이 있는 path를 지정합니다.
"file": "hello.exe", # 실행할 파일명을 지정합니다.
"title": "hello_1", # task 관리 시 사용하는 명칭입니다. 16자를 넘지 않게 지정하세요.
"schedule": { # cron은 체크할 주기를 지정하고 finish_datetime은 task의 종료일정을 지정하세요. (현재는 사용하지 않습니다.)
"cron": "* * * * *",
"finish_datetime": "2999-12-31"
},
"buttons": {
"START": "START" # 시작 시 어떤 버튼을 눌러야 한다면 해당 버튼의 라벨을 지정하세요.
},
"job": "sustenance" # 프로세스가 다운된 경우 재시작 합니다.
},
{
"name": "샘플 B 프로세스",
"path": "E:/sample/sampleB",
"file": "hello2.exe",
"title": "hello_2",
"schedule": {
"cron": "0 0 * * *",
"finish_datetime": "2999-12-31"
},
"buttons": {
"QUIT": "EXIT" # 종료 시 버튼을 눌러야 한다면 해당 버튼의 라벨을 지정하세요. 시작 시 눌러야할 버튼이 필요하다면 START를 같이 명시하세요.
},
"job": "restart" # 지정된 시간에 프로세스를 재시작 합니다.
}
]
}
필요한 서버에 python을 직접 설치하지 마시고 pyinstaller로 exe 파일로 만들어 사용하시길 추천 드립니다.
사실 자원이 충분하다면 문제가 되는 프로젝트를 수정하는게 좋겠지만 그렇지 않다면 이런 코드도 대안이 될거라 생각됩니다.
윈도우 서비스 형태로 개선된 버전이 필요하시면 github에 알람을 설정해 두세요.
이상입니다. 모두 즐거운 하루 되시기 바랍니다.
