每日午夜无异常时无法创建新空日志文件,寻求技术解决方案
问题分析与解决方案
你的核心问题是自定义的RosCanTimedRotatingFileHandler原本为5分钟轮转设计,但当前配置为每日午夜轮转时,逻辑冲突导致无法自动生成新日志文件。以下是具体问题点和修复方案:
关键问题点
- 轮转逻辑冲突:自定义Handler的
__init__和doRollover方法都是针对5分钟间隔编写的,当使用when='midnight'时,时间计算逻辑完全错误,破坏了午夜轮转的时间点。 - 文件名处理混乱:初始化时日志文件名已包含5分钟时间戳,后续轮转时又通过
split('-')[0]截断文件名,导致原文件被重命名为无后缀的基础名,新文件名生成逻辑混乱。 - 异常静默吞噬:类定义内的
try-except会覆盖__init__和doRollover的异常,导致问题无法被及时排查。
解决方案一:使用原生Handler实现每日轮转
如果仅需要每日午夜自动生成新日志,直接使用Python原生的TimedRotatingFileHandler即可,无需自定义子类:
import logging import logging.handlers as handlers import os from datetime import datetime LOGGER = logging.getLogger('tractor_init_config_manager') LOGGER.setLevel(logging.INFO) # 定义日志格式 FORMATTER = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) LOG_PATH = "/opt/monarch/debug/" # 使用不带时间戳的基础文件名,让Handler自动生成后缀 BASE_FILE_NAME = "ros_can_logs/integration.log" FULL_LOG_PATH = os.path.join(LOG_PATH, BASE_FILE_NAME) def folder_exists(file_path): """检查目录是否存在,不存在则创建""" try: path = os.path.dirname(file_path) os.makedirs(path, exist_ok=True) # 初始化空日志文件(如果不存在) if not os.path.exists(file_path): with open(file_path, 'w') as f: f.write("Initiating the logging...\n") except Exception as err: print(err) folder_exists(FULL_LOG_PATH) # 配置每日午夜轮转 LOG_HANDLER = handlers.TimedRotatingFileHandler( FULL_LOG_PATH, when='midnight', interval=1, backupCount=10, encoding='utf-8', delay=False ) # 设置日志文件后缀格式 LOG_HANDLER.suffix = "%Y-%m-%d.log" LOG_HANDLER.setLevel(logging.INFO) LOG_HANDLER.setFormatter(FORMATTER) LOGGER.addHandler(LOG_HANDLER) def info(msg): LOGGER.info(str(msg)) def error(msg): LOGGER.error(str(msg))
解决方案二:修复自定义Handler兼容两种轮转模式
如果需要同时保留5分钟轮转和每日午夜轮转的能力,修改自定义Handler的分支逻辑:
import codecs import glob import logging import logging.handlers as handlers import os import re import time from datetime import datetime import syslog class RosCanTimedRotatingFileHandler(handlers.TimedRotatingFileHandler): """支持5分钟轮转和每日午夜轮转的日志Handler""" def __init__(self, *args, **kwargs): self.target_when = kwargs.get('when', 'M') super().__init__(*args, **kwargs) if self.target_when == 'M' and self.interval == 5: # 5分钟轮转专属配置 self.suffix = '%Y-%m-%d-%H-%M-%S.log' self.extMatch = re.compile(r"^\d{4}-\d{2}-\d{2}-\d{2}-\d{2}-\d{2}\.log$") # 计算下一个5分钟整点的轮转时间 now = datetime.now() next_rollover = now.replace(minute=(now.minute - now.minute %5)+5, second=0, microsecond=0) self.rolloverAt = time.mktime(next_rollover.timetuple()) elif self.target_when == 'midnight' and self.interval ==1: # 每日午夜轮转专属配置 self.suffix = '%Y-%m-%d.log' self.extMatch = re.compile(r"^\d{4}-\d{2}-\d{2}\.log$") # 使用原生午夜轮转时间计算逻辑 pass def doRollover(self): try: if self.target_when == 'M' and self.interval ==5: # 5分钟轮转处理逻辑 time_t = self.rolloverAt - self.interval*60 time_tuple = time.localtime(time_t) base_name = self.baseFilename.rsplit('-', 1)[0] if '-' in self.baseFilename else self.baseFilename dfn = f"{base_name}-{time.strftime(self.suffix, time_tuple)}" if os.path.exists(dfn): os.remove(dfn) if os.path.exists(self.baseFilename): os.rename(self.baseFilename, dfn) # 生成新的5分钟对齐文件名 cur_time = datetime.now() current_timestamp = cur_time.replace( minute=(cur_time.minute - cur_time.minute %5), second=0, microsecond=0 ).strftime("%Y-%m-%d-%H-%M-%S") self.baseFilename = f"{base_name}-{current_timestamp}.log" # 清理超过备份数量的旧日志 if self.backupCount >0: files_list = glob.glob(f"{base_name}-20*") if len(files_list) > self.backupCount: files_list.sort() os.remove(files_list[0]) elif self.target_when == 'midnight' and self.interval ==1: # 每日午夜轮转处理逻辑,复用原生逻辑并确保空文件生成 super().doRollover() # 确保新日志文件被创建(即使无日志写入) if not os.path.exists(self.baseFilename): self.stream = codecs.open(self.baseFilename, 'w', self.encoding) if self.encoding else open(self.baseFilename, 'w') # 重新打开日志流 self.stream = codecs.open(self.baseFilename, 'w', self.encoding) if self.encoding else open(self.baseFilename, 'w') # 计算下一次轮转时间 current_time = time.time() new_rollover_at = self.computeRollover(current_time) while new_rollover_at <= current_time: new_rollover_at += self.interval if self.target_when != 'M' else self.interval*60 self.rolloverAt = new_rollover_at except Exception as err: syslog.syslog(f"error in giles_log file of sync service: {err}") # 初始化日志 LOGGER = logging.getLogger('tractor_init_config_manager') LOGGER.setLevel(logging.INFO) FORMATTER = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) LOG_PATH = "/opt/monarch/debug/" BASE_FILE_NAME = "ros_can_logs/integration.log" FULL_LOG_PATH = os.path.join(LOG_PATH, BASE_FILE_NAME) def folder_exists(file_path): try: path = os.path.dirname(file_path) os.makedirs(path, exist_ok=True) if not os.path.exists(file_path): with open(file_path, 'w') as f: f.write("Initiating the logging...\n") except Exception as err: print(err) folder_exists(FULL_LOG_PATH) LOG_HANDLER = RosCanTimedRotatingFileHandler( FULL_LOG_PATH, when='midnight', interval=1, backupCount=10, delay=False ) LOG_HANDLER.setLevel(logging.INFO) LOG_HANDLER.setFormatter(FORMATTER) LOGGER.addHandler(LOG_HANDLER) def info(msg): LOGGER.info(str(msg)) def error(msg): LOGGER.error(str(msg))
内容的提问来源于stack exchange,提问作者tushar aloorkar
相关产品推荐
相关产品推荐

