Merge pull request #497 from kytos-ng/logging/filters

Add in log filter for pacing logs + bump version to 2024.1.1

CHANGELOG.rst

@@ -6,6 +6,13 @@ All notable changes to the kytos project will be documented in this file.
 UNRELEASED - Under development
 ******************************
 
+[2024.1.1] - 2024-08-04
+***********************
+
+Added
+=====
+- Added logging filter to suppress frequently repeated messages from pacing.
+
 [2024.1.0] - 2024-07-23
 ***********************
 

docs/conf.py

@@ -64,10 +64,10 @@
 # built documents.
 #
 # The short X.Y version.
-version = u'2024.1.0'
+version = u'2024.1.1'
 show_version = False
 # The full version, including alpha/beta/rc tags.
-release = u'2024.1.0'
+release = u'2024.1.1'
 
 # The language for content autogenerated by Sphinx. Refer to documentation
 # for a list of supported languages.

kytos/core/controller.py

@@ -205,7 +205,14 @@ def _patch_core_loggers():
         reloadable_mods = [module for mod_name, module in sys.modules.items()
                            if mod_name[:str_len] == match_str]
         for module in reloadable_mods:
-            module.LOG = logging.getLogger(module.__name__)
+            new_logger = logging.getLogger(module.__name__)
+            if hasattr(module, "LOG"):
+                old_logger = module.LOG
+                for handler in old_logger.handlers:
+                    new_logger.addHandler(handler)
+                for log_filter in old_logger.filters:
+                    new_logger.addFilter(log_filter)
+            module.LOG = new_logger
 
     @staticmethod
     def loggers():

kytos/core/metadata.py

@@ -2,7 +2,7 @@
 
 The present metadata is intended to be used mainly on the setup.
 """
-__version__ = '2024.1.0'
+__version__ = '2024.1.1'
 __author__ = 'Kytos Team'
 __author_email__ = '[email protected]'
 __license__ = 'MIT'

kytos/core/pacing.py

@@ -8,8 +8,12 @@
 from limits import RateLimitItem, parse
 from limits.storage import storage_from_string
 
+from kytos.logging.filters import RepeateMessageFilter
+
 LOG = logging.getLogger(__name__)
 
+LOG.addFilter(RepeateMessageFilter(1.0, 512))
+
 
 class EmptyStrategy(limits.strategies.FixedWindowRateLimiter):
     """Rate limiter, that doesn't actually rate limit."""
@@ -139,7 +143,7 @@ async def ahit(self, action_name: str, *keys):
                 *identifiers
             )
             sleep_time = window_reset - time.time()
-            LOG.info(f'Limited reached: {identifiers}')
+            LOG.info(f'Limit reached: {identifiers}')
             await asyncio.sleep(sleep_time)
 
     def hit(self, action_name: str, *keys):

kytos/logging/filters.py

@@ -0,0 +1,45 @@
+
+from collections import OrderedDict
+from logging import LogRecord
+from threading import Lock
+
+
+class RepeateMessageFilter:
+    lockout_time: float
+    cache_size: int
+    _cache: OrderedDict[tuple, float]
+    _lock: Lock
+
+    def __init__(self, lockout_time: float, cache_size: int = 512):
+        self.lockout_time = lockout_time
+        self.cache_size = cache_size
+        self._cache = OrderedDict()
+        self._lock = Lock()
+
+    def filter(self, record: LogRecord) -> bool:
+        key = self._record_key(record)
+        current_time = record.created
+        with self._lock:
+            if key not in self._cache:
+                self._cache[key] = current_time
+                if len(self._cache) > self.cache_size:
+                    self._cache.popitem(last=False)
+                return True
+            elif current_time - self._cache[key] > self.lockout_time:
+                self._cache[key] = current_time
+                self._cache.move_to_end(key)
+                if len(self._cache) > self.cache_size:
+                    self._cache.popitem(last=False)
+                return True
+            return False
+
+    @staticmethod
+    def _record_key(record: LogRecord):
+        return (
+            record.pathname,
+            record.module,
+            record.lineno,
+            record.levelno,
+            record.msg,
+            record.args
+        )

tests/unit/test_logging/test_repeat_message_filter.py

@@ -0,0 +1,114 @@
+
+
+import unittest
+from logging import LogRecord
+
+import pytest
+
+from kytos.logging.filters import RepeateMessageFilter
+
+
+class TestRepeateMessageFilter:
+
+    @pytest.fixture(
+        params=[
+            1.0,
+            5.0,
+            10.0,
+        ]
+    )
+    def lockout_time(self, request):
+        return request.param
+
+    @pytest.fixture(
+        params=[
+            1,
+            256,
+            512,
+            1024,
+        ]
+    )
+    def cache_size(self, request):
+        return request.param
+
+    @pytest.fixture
+    def message_filter(self, lockout_time, cache_size):
+        return RepeateMessageFilter(lockout_time, cache_size)
+
+    @staticmethod
+    def make_record(msg, ct):
+        record = LogRecord(
+            "test_logger",
+            0,
+            "test_path",
+            1337,
+            msg,
+            ("arg1", "arg2"),
+            None
+        )
+        record.created = ct
+        record.relativeCreated = ct
+        return record
+
+    @pytest.fixture
+    def blockable_record(self):
+        return self.make_record("test", 0.0)
+
+    @pytest.fixture
+    def unblockable_record(self, lockout_time):
+        return self.make_record("test", lockout_time + 1)
+
+    @pytest.fixture
+    def message_filter_with_one_message(
+        self,
+        message_filter,
+        blockable_record
+    ):
+        assert message_filter.filter(blockable_record)
+        return message_filter
+
+    @pytest.fixture
+    def message_filter_with_one_message_and_junk(
+        self,
+        message_filter_with_one_message,
+        cache_size
+    ):
+        for i in range(cache_size - 1):
+            assert message_filter_with_one_message.filter(
+                self.make_record(f"junk-{i}", 0.0)
+            )
+        return message_filter_with_one_message
+
+    @pytest.fixture
+    def last_junk_record(
+        self,
+        cache_size
+    ):
+        return self.make_record(f"junk-{cache_size - 1}", 0.0)
+
+    def test_001_filter(
+        self,
+        message_filter_with_one_message,
+        blockable_record,
+        unblockable_record
+    ):
+        assert not message_filter_with_one_message.filter(blockable_record)
+        assert message_filter_with_one_message.filter(unblockable_record)
+        assert not message_filter_with_one_message.filter(blockable_record)
+        assert not message_filter_with_one_message.filter(unblockable_record)
+
+    def test_002_cache_eviction(
+        self,
+        message_filter_with_one_message_and_junk,
+        blockable_record,
+        last_junk_record
+    ):
+        assert not message_filter_with_one_message_and_junk.filter(
+            blockable_record
+        )
+        assert message_filter_with_one_message_and_junk.filter(
+            last_junk_record
+        )
+        assert message_filter_with_one_message_and_junk.filter(
+            blockable_record
+        )