File: //usr/local/aegis/PythonLoaderTemp/third_party/aegis_checker/offline/check_ipc_error.py
# -*- coding: utf-8 -*-
import sys
import re
import logging
from aegis_checker.common.aegis_client_log_parser import LogObserver, LOG_INFO, LOG_WARN
from aegis_checker.info.check_result import *
class IpcErrorLogObserver(LogObserver):
def __init__(self):
self.__ipc_error_events = []
def on_end(self, success):
if not self.__ipc_error_events:
return
for ipc_error_event in self.__ipc_error_events:
logging.warning("offline issue may be caused by ipc error when %s %s, log is %s" %
(ipc_error_event["date"], ipc_error_event["time"], ipc_error_event["content"]))
set_root_cause(ROOT_CAUSE_AEGIS_IPC_ERROR,
"offline issue may be caused by ipc error %d times, if this issue happen many times, please report to aegis supporter" % len(self.__ipc_error_events))
def on_log(self, log_date, log_time, log_type, content, line, line_num, log_file_path):
"""
2021-03-23 12:30:18 [Info] ipc srver critical error: 231
:param log_date:
:param log_time:
:param log_type:
:param content:
:param line:
:return:
"""
ipc_error_reg = r"ipc srver critical error: \d+"
if log_type == LOG_INFO and re.match(ipc_error_reg, content):
# logging.warning("ipc error on %s" % line)
self.__ipc_error_events.append({
"date": log_date,
"time": log_time,
"content": content,
})
def test():
logging.basicConfig(format='%(asctime)s [%(filename)s][%(levelname)s] %(message)s', level=logging.DEBUG)
login_observer = IpcErrorLogObserver()
log_file_path = sys.argv[1]
with open(log_file_path) as f:
regular = r"^(\d{4}-\d{2}-\d{2}) (\d{2}:\d{2}:\d{2}) \[(\w+)\] (.+)"
reg = re.compile(regular, re.I)
line_num = 0
for line in f:
match_obj = reg.match(line)
if match_obj:
log_date, log_time, log_type, content = match_obj.groups()
login_observer.on_log(log_date, log_time, log_type, content, line, line_num, log_file_path)
line_num += 1
login_observer.on_end(True)
if __name__ == '__main__':
test()