修正windows多次调用进程间lock产生的死锁问题

Author: ShichaoMa

log_to_kafka/__init__.py

@@ -8,4 +8,8 @@
 
 HOME_PAGE = "https://www.github.com/ShichaoMa/log_to_kafka"
 
-from .logger import LogFactory, Logger, LogObject, KafkaHandler
+from .logger import LogFactory, Logger, LogObject, KafkaHandler, FixedConcurrentRotatingFileHandler, ConcurrentRotatingFileHandler
+
+__all__ = ["LogFactory", "Logger", "LogObject", "KafkaHandler",
+           "FixedConcurrentRotatingFileHandler", "ConcurrentRotatingFileHandler",
+           "HOME_PAGE", "AUTHOR_EMAIL", "VERSION", "AUTHOR"]

log_to_kafka/logger.py

@@ -7,16 +7,47 @@
 import sys
 import time
 import traceback
+
 from functools import wraps
+from portalocker import lock, unlock, LOCK_EX
 
-from cloghandler import ConcurrentRotatingFileHandler
+from cloghandler import ConcurrentRotatingFileHandler, NullLogRecord
 from kafka import KafkaClient, SimpleProducer
 from kafka.common import FailedPayloadsError
 from pythonjsonlogger import jsonlogger
 
 import default_settings
 from settings_wrapper import SettingsWrapper
 
+class FixedConcurrentRotatingFileHandler(ConcurrentRotatingFileHandler):
+    """
+    修正windows多次调用进程间lock产生的死锁问题
+    """
+    def acquire(self):
+        """ Acquire thread and file locks.  Re-opening log for 'degraded' mode.
+        """
+        # handle thread lock
+        logging.Handler.acquire(self)
+        # Issue a file lock.  (This is inefficient for multiple active threads
+        # within a single process. But if you're worried about high-performance,
+        # you probably aren't using this log handler.)
+        if self.stream_lock:
+            # If stream_lock=None, then assume close() was called or something
+            # else weird and ignore all file-level locks.
+            if self.stream_lock.closed:
+                # Daemonization can close all open file descriptors, see
+                # https://bugzilla.redhat.com/show_bug.cgi?id=952929
+                # Try opening the lock file again.  Should we warn() here?!?
+                try:
+                    self._open_lockfile()
+                except Exception:
+                    self.handleError(NullLogRecord())
+                    # Don't try to open the stream lock again
+                    self.stream_lock = None
+                    return
+            unlock(self.stream_lock)
+            lock(self.stream_lock, LOCK_EX)
+
 
 def failedpayloads_wrapper(max_iter_times, _raise=False):
 
@@ -188,8 +219,12 @@ def set_logger(self, logger=None):
             except OSError as exception:
                 if exception.errno != errno.EEXIST:
                     raise
+            if os.name == "nt":
+                handler = FixedConcurrentRotatingFileHandler
+            else:
+                handler = ConcurrentRotatingFileHandler
             self.logger.set_handler(
-                ConcurrentRotatingFileHandler(
+                handler(
                     os.path.join(my_dir, my_file),
                     backupCount=my_backups,
                     maxBytes=my_bytes))

setup.py

@@ -6,7 +6,7 @@
 except:
     from distutils.core import setup
 
-VERSION = '1.0.2'
+VERSION = '1.0.3'
 
 AUTHOR = "cn"
 

test/default_settings.py

@@ -0,0 +1,21 @@
+# -*- coding:utf-8 -*-
+
+
+# 日志级别
+LOG_LEVEL = 'DEBUG'
+# 是否标准到控制台
+LOG_STDOUT = False
+# 是否输出json格式
+LOG_JSON = False
+# log文件目录
+LOG_DIR = "logs"
+# 每个log最大大小
+LOG_MAX_BYTES = '10MB'
+# log备份数量
+LOG_BACKUPS = 5
+# 是否发送到kafka
+TO_KAFKA = False
+# kafka地址
+KAFKA_HOSTS = "192.168.200.90:9092"
+# kafka topic
+TOPIC = "jay-cluster-logs"

test/test_concurrent.py

@@ -0,0 +1,20 @@
+# -*- coding:utf-8 -*-
+import os
+import sys
+sys.path.insert(0, "..")
+import errno
+import logging
+from log_to_kafka.logger import FixedConcurrentRotatingFileHandler
+my_dir = "logs"
+try:
+    os.makedirs(my_dir)
+except OSError as exception:
+    if exception.errno != errno.EEXIST:
+        raise
+logger = logging.getLogger()
+logger.setLevel(logging.DEBUG)
+logger.addHandler(
+FixedConcurrentRotatingFileHandler(
+    os.path.join(my_dir, "test.log"),
+    backupCount=5))
+logger.info("this is a log. ")

test/test_lock.py

@@ -0,0 +1,9 @@
+from portalocker import lock, unlock, LOCK_EX, LOCK_NB, LockException
+
+a = open("logs/test.lock", "w")
+unlock(a)
+unlock(a)
+print 11111111
+lock(a, LOCK_EX)
+print 222222
+unlock(a)

test/test_logger.py

@@ -0,0 +1,11 @@
+from log_to_kafka import Logger
+
+class MyClass(Logger):
+    name = "log_name"
+def __init__(self, settings_file):
+    super(MyClass, self).__init__(settings_file)
+
+
+MC = MyClass("default_settings.py")
+MC.set_logger()
+MC.logger.debug("....")

test/test_stream.py

@@ -0,0 +1,6 @@
+import logging
+import sys
+logger = logging.getLogger()
+logger.setLevel(logging.DEBUG)
+logger.addHandler(logging.StreamHandler(sys.stdout))
+logger.info('aaaaa')