Add a filter to rate limit logs

* Add configuration options to enable rate limiting:

  - rate_limit_interval
  - rate_limit_burst
  - rate_limit_except_level

* Add oslo_log.rate_limit submodule
* Add public functins:

  - install_filter(burst, interval, except_level)
  - uninstall_filter()

* Add unit tests
* Add a new dependency, monotonic, to get a monotonic clock

Default: rate limiting is disabled and logs at CRITICAL level are not
rate limited.

DocImpact
Change-Id: Ic58dafceefde1b109721a58631c223522bf4cc9c

5 files changed, 289 insertions, 0 deletions

diff --git a/oslo_log/_options.py b/oslo_log/_options.py
index 24fc2c9..8b10384 100644
--- a/oslo_log/_options.py
+++ b/oslo_log/_options.py
@@ -146,6 +146,21 @@ log_opts = [
                default='[instance: %(uuid)s] ',
                help='The format for an instance UUID that is passed with the '
                     'log message.'),
+
+    cfg.IntOpt('rate_limit_interval',
+               default=0,
+               help='Interval, number of seconds, of log rate limiting.'),
+    cfg.IntOpt('rate_limit_burst',
+               default=0,
+               help='Maximum number of logged messages per '
+                    'rate_limit_interval.'),
+    cfg.StrOpt('rate_limit_except_level',
+               default='CRITICAL',
+               help='Log level name used by rate limiting: CRITICAL, ERROR, '
+                    'INFO, WARNING, DEBUG or empty string. Logs with level '
+                    'greater or equal to rate_limit_except_level are not '
+                    'filtered. An empty string means that all levels are '
+                    'filtered.'),
 ]
 
 
diff --git a/oslo_log/log.py b/oslo_log/log.py
index 1d7f3e2..80e86f5 100644
--- a/oslo_log/log.py
+++ b/oslo_log/log.py
@@ -415,6 +415,12 @@ def _setup_logging_from_conf(conf, project, version):
         else:
             logger.setLevel(level_name)
 
+    if conf.rate_limit_burst >= 1 and conf.rate_limit_interval >= 1:
+        from oslo_log import rate_limit
+        rate_limit.install_filter(conf.rate_limit_burst,
+                                  conf.rate_limit_interval,
+                                  conf.rate_limit_except)
+
 _loggers = {}
 
 
diff --git a/oslo_log/rate_limit.py b/oslo_log/rate_limit.py
new file mode 100644
index 0000000..f1da909
--- /dev/null
+++ b/oslo_log/rate_limit.py
@@ -0,0 +1,157 @@
+# Copyright 2016 Red Hat, Inc. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+
+import logging
+
+try:
+    from time import monotonic as monotonic_clock   # noqa
+except ImportError:
+    from monotonic import monotonic as monotonic_clock   # noqa
+
+
+class _LogRateLimit(logging.Filter):
+    def __init__(self, burst, interval, except_level=None):
+        logging.Filter.__init__(self)
+        self.burst = burst
+        self.interval = interval
+        self.except_level = except_level
+        self.logger = logging.getLogger()
+        self._reset()
+
+    def _reset(self, now=None):
+        if now is None:
+            now = monotonic_clock()
+        self.counter = 0
+        self.end_time = now + self.interval
+        self.emit_warn = False
+
+    def filter(self, record):
+        if (self.except_level is not None
+           and record.levelno >= self.except_level):
+            # don't limit levels >= except_level
+            return True
+
+        timestamp = monotonic_clock()
+        if timestamp >= self.end_time:
+            self._reset(timestamp)
+            self.counter += 1
+            return True
+
+        self.counter += 1
+        if self.counter <= self.burst:
+            return True
+        if self.emit_warn:
+            # Allow to log our own warning: self.logger is also filtered by
+            # rate limiting
+            return True
+
+        if self.counter == self.burst + 1:
+            self.emit_warn = True
+            self.logger.error("Logging rate limit: "
+                              "drop after %s records/%s sec",
+                              self.burst, self.interval)
+            self.emit_warn = False
+
+        # Drop the log
+        return False
+
+
+def _iter_loggers():
+    """Iterate on existing loggers."""
+
+    # Sadly, Logger.manager and Manager.loggerDict are not documented,
+    # but there is no logging public function to iterate on all loggers.
+
+    # The root logger is not part of loggerDict.
+    yield logging.getLogger()
+
+    manager = logging.Logger.manager
+    for logger in manager.loggerDict.values():
+        if isinstance(logger, logging.PlaceHolder):
+            continue
+        yield logger
+
+
+_LOG_LEVELS = {
+    'CRITICAL': logging.CRITICAL,
+    'ERROR': logging.ERROR,
+    'INFO': logging.INFO,
+    'WARNING': logging.WARNING,
+    'DEBUG': logging.DEBUG,
+}
+
+
+def install_filter(burst, interval, except_level='CRITICAL'):
+    """Install a rate limit filter on existing and future loggers.
+
+    Limit logs to *burst* messages every *interval* seconds, except of levels
+    >= *except_level*. *except_level* is a log level name like 'CRITICAL'. If
+    *except_level* is an empty string, all levels are filtered.
+
+    The filter uses a monotonic clock, the timestamp of log records is not
+    used.
+
+    Raise an exception if a rate limit filter is already installed.
+    """
+
+    if install_filter.log_filter is not None:
+        raise RuntimeError("rate limit filter already installed")
+
+    try:
+        except_levelno = _LOG_LEVELS[except_level]
+    except KeyError:
+        raise ValueError("invalid log level name: %r" % except_level)
+
+    log_filter = _LogRateLimit(burst, interval, except_levelno)
+
+    install_filter.log_filter = log_filter
+    install_filter.logger_class = logging.getLoggerClass()
+
+    class RateLimitLogger(install_filter.logger_class):
+        def __init__(self, *args, **kw):
+            logging.Logger.__init__(self, *args, **kw)
+            self.addFilter(log_filter)
+
+    # Setup our own logger class to automatically add the filter
+    # to new loggers.
+    logging.setLoggerClass(RateLimitLogger)
+
+    # Add the filter to all existing loggers
+    for logger in _iter_loggers():
+        logger.addFilter(log_filter)
+
+install_filter.log_filter = None
+install_filter.logger_class = None
+
+
+def uninstall_filter():
+    """Uninstall the rate filter installed by install_filter().
+
+    Do nothing if the filter was already uninstalled.
+    """
+
+    if install_filter.log_filter is None:
+        # not installed (or already uninstalled)
+        return
+
+    # Restore the old logger class
+    logging.setLoggerClass(install_filter.logger_class)
+
+    # Remove the filter from all existing loggers
+    for logger in _iter_loggers():
+        logger.removeFilter(install_filter.log_filter)
+
+    install_filter.logger_class = None
+    install_filter.log_filter = None
diff --git a/oslo_log/tests/unit/test_rate_limit.py b/oslo_log/tests/unit/test_rate_limit.py
new file mode 100644
index 0000000..a81baa2
--- /dev/null
+++ b/oslo_log/tests/unit/test_rate_limit.py
@@ -0,0 +1,110 @@
+# Copyright 2016 Red Hat, Inc. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import logging
+
+import mock
+from oslotest import base as test_base
+import six
+
+from oslo_log import rate_limit
+
+
+class LogRateLimitTestCase(test_base.BaseTestCase):
+    def tearDown(self):
+        super(LogRateLimitTestCase, self).tearDown()
+        rate_limit.uninstall_filter()
+
+    def install_filter(self, *args):
+        rate_limit.install_filter(*args)
+
+        logger = logging.getLogger()
+
+        # remove handlers to not pollute stdout
+        def restore_handlers(logger, handlers):
+            for handler in handlers:
+                logger.addHandler(handler)
+
+        self.addCleanup(restore_handlers, logger, list(logger.handlers))
+        for handler in list(logger.handlers):
+            logger.removeHandler(handler)
+
+        # install our handler writing logs into a StringIO
+        stream = six.StringIO()
+        handler = logging.StreamHandler(stream)
+        logger.addHandler(handler)
+
+        return (logger, stream)
+
+    @mock.patch('oslo_log.rate_limit.monotonic_clock')
+    def test_rate_limit(self, mock_clock):
+        mock_clock.return_value = 1
+        logger, stream = self.install_filter(2, 1)
+
+        # first burst
+        logger.error("message 1")
+        logger.error("message 2")
+        logger.error("message 3")
+        self.assertEqual(stream.getvalue(),
+                         'message 1\n'
+                         'message 2\n'
+                         'Logging rate limit: drop after 2 records/1 sec\n')
+
+        # second burst (clock changed)
+        stream.seek(0)
+        stream.truncate()
+        mock_clock.return_value = 2
+
+        logger.error("message 4")
+        logger.error("message 5")
+        logger.error("message 6")
+        self.assertEqual(stream.getvalue(),
+                         'message 4\n'
+                         'message 5\n'
+                         'Logging rate limit: drop after 2 records/1 sec\n')
+
+    @mock.patch('oslo_log.rate_limit.monotonic_clock')
+    def test_rate_limit_except_level(self, mock_clock):
+        mock_clock.return_value = 1
+        logger, stream = self.install_filter(1, 1, 'CRITICAL')
+
+        # first burst
+        logger.error("error 1")
+        logger.error("error 2")
+        logger.critical("critical 3")
+        logger.critical("critical 4")
+        self.assertEqual(stream.getvalue(),
+                         'error 1\n'
+                         'Logging rate limit: drop after 1 records/1 sec\n'
+                         'critical 3\n'
+                         'critical 4\n')
+
+    def test_install_twice(self):
+        rate_limit.install_filter(100, 1)
+        self.assertRaises(RuntimeError, rate_limit.install_filter, 100, 1)
+
+    @mock.patch('oslo_log.rate_limit.monotonic_clock')
+    def test_uninstall(self, mock_clock):
+        mock_clock.return_value = 1
+        logger, stream = self.install_filter(1, 1)
+        rate_limit.uninstall_filter()
+
+        # not limited
+        logger.error("message 1")
+        logger.error("message 2")
+        logger.error("message 3")
+        self.assertEqual(stream.getvalue(),
+                         'message 1\n'
+                         'message 2\n'
+                         'message 3\n')
diff --git a/requirements.txt b/requirements.txt
index 5df9d38..3dc12c4 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -12,3 +12,4 @@ oslo.serialization>=1.10.0 # Apache-2.0
 debtcollector>=1.2.0 # Apache-2.0
 pyinotify>=0.9.6;sys_platform!='win32' and sys_platform!='darwin' and sys_platform!='sunos5' # MIT
 python-dateutil>=2.4.2 # BSD
+monotonic>=0.6 # Apache-2.0