Merge "proxy: refactor error limiter to a class"

Author: Zuul

swift/common/error_limiter.py

@@ -0,0 +1,91 @@
+# Copyright (c) 2021 NVIDIA
+#
+# 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 collections
+from time import time
+
+from swift.common.utils import node_to_string
+
+
+class ErrorLimiter(object):
+    """
+    Tracks the number of errors that have occurred for nodes. A node will be
+    considered to be error-limited for a given interval of time after it has
+    accumulated more errors than a given limit.
+
+    :param suppression_interval: The number of seconds for which a node is
+        error-limited once it has accumulated more than ``suppression_limit``
+        errors. Should be a float value.
+    :param suppression_limit: The number of errors that a node must accumulate
+        before it is considered to be error-limited. Should be an int value.
+    """
+    def __init__(self, suppression_interval, suppression_limit):
+        self.suppression_interval = float(suppression_interval)
+        self.suppression_limit = int(suppression_limit)
+        self.stats = collections.defaultdict(dict)
+
+    def node_key(self, node):
+        """
+        Get the key under which a node's error stats will be stored.
+
+        :param node: dictionary describing a node.
+        :return: string key.
+        """
+        return node_to_string(node)
+
+    def is_limited(self, node):
+        """
+        Check if the node is currently error limited.
+
+        :param node: dictionary of node to check
+        :returns: True if error limited, False otherwise
+        """
+        now = time()
+        node_key = self.node_key(node)
+        error_stats = self.stats.get(node_key)
+
+        if error_stats is None or 'errors' not in error_stats:
+            return False
+
+        if 'last_error' in error_stats and error_stats['last_error'] < \
+                now - self.suppression_interval:
+            self.stats.pop(node_key)
+            return False
+        return error_stats['errors'] > self.suppression_limit
+
+    def limit(self, node):
+        """
+        Mark a node as error limited. This immediately pretends the
+        node received enough errors to trigger error suppression. Use
+        this for errors like Insufficient Storage. For other errors
+        use :func:`increment`.
+
+        :param node: dictionary of node to error limit
+        """
+        node_key = self.node_key(node)
+        error_stats = self.stats[node_key]
+        error_stats['errors'] = self.suppression_limit + 1
+        error_stats['last_error'] = time()
+
+    def increment(self, node):
+        """
+        Increment the error count and update the time of the last error for
+        the given ``node``.
+
+        :param node: dictionary describing a node.
+        """
+        node_key = self.node_key(node)
+        error_stats = self.stats[node_key]
+        error_stats['errors'] = error_stats.get('errors', 0) + 1
+        error_stats['last_error'] = time()

swift/proxy/server.py

@@ -31,6 +31,7 @@
 from swift.common.http import is_server_error
 from swift.common.storage_policy import POLICIES
 from swift.common.ring import Ring
+from swift.common.error_limiter import ErrorLimiter
 from swift.common.utils import Watchdog, get_logger, \
     get_remote_client, split_path, config_true_value, generate_trans_id, \
     affinity_key_function, affinity_locality_predicate, list_from_csv, \
@@ -204,7 +205,6 @@ def __init__(self, conf, logger=None, account_ring=None,
                                      statsd_tail_prefix='proxy-server')
         else:
             self.logger = logger
-        self._error_limiting = {}
         self.backend_user_agent = 'proxy-server %s' % os.getpid()
 
         swift_dir = conf.get('swift_dir', '/etc/swift')
@@ -218,10 +218,12 @@ def __init__(self, conf, logger=None, account_ring=None,
         self.client_chunk_size = int(conf.get('client_chunk_size', 65536))
         self.trans_id_suffix = conf.get('trans_id_suffix', '')
         self.post_quorum_timeout = float(conf.get('post_quorum_timeout', 0.5))
-        self.error_suppression_interval = \
+        error_suppression_interval = \
             float(conf.get('error_suppression_interval', 60))
-        self.error_suppression_limit = \
+        error_suppression_limit = \
             int(conf.get('error_suppression_limit', 10))
+        self.error_limiter = ErrorLimiter(error_suppression_interval,
+                                          error_suppression_limit)
         self.recheck_container_existence = \
             int(conf.get('recheck_container_existence',
                          DEFAULT_RECHECK_CONTAINER_EXISTENCE))
@@ -646,27 +648,14 @@ def set_node_timing(self, node, timing):
         timing = round(timing, 3)  # sort timings to the millisecond
         self.node_timings[node['ip']] = (timing, now + self.timing_expiry)
 
-    def _error_limit_node_key(self, node):
-        return node_to_string(node)
-
     def error_limited(self, node):
         """
         Check if the node is currently error limited.
 
         :param node: dictionary of node to check
         :returns: True if error limited, False otherwise
         """
-        now = time()
-        node_key = self._error_limit_node_key(node)
-        error_stats = self._error_limiting.get(node_key)
-
-        if error_stats is None or 'errors' not in error_stats:
-            return False
-        if 'last_error' in error_stats and error_stats['last_error'] < \
-                now - self.error_suppression_interval:
-            self._error_limiting.pop(node_key, None)
-            return False
-        limited = error_stats['errors'] > self.error_suppression_limit
+        limited = self.error_limiter.is_limited(node)
         if limited:
             self.logger.debug(
                 'Node error limited: %s', node_to_string(node))
@@ -677,32 +666,23 @@ def error_limit(self, node, msg):
         Mark a node as error limited. This immediately pretends the
         node received enough errors to trigger error suppression. Use
         this for errors like Insufficient Storage. For other errors
-        use :func:`error_occurred`.
+        use :func:`increment`.
 
         :param node: dictionary of node to error limit
         :param msg: error message
         """
-        node_key = self._error_limit_node_key(node)
-        error_stats = self._error_limiting.setdefault(node_key, {})
-        error_stats['errors'] = self.error_suppression_limit + 1
-        error_stats['last_error'] = time()
+        self.error_limiter.limit(node)
         self.logger.error('%(msg)s %(node)s',
                           {'msg': msg, 'node': node_to_string(node)})
 
-    def _incr_node_errors(self, node):
-        node_key = self._error_limit_node_key(node)
-        error_stats = self._error_limiting.setdefault(node_key, {})
-        error_stats['errors'] = error_stats.get('errors', 0) + 1
-        error_stats['last_error'] = time()
-
     def error_occurred(self, node, msg):
         """
         Handle logging, and handling of errors.
 
         :param node: dictionary of node to handle errors for
         :param msg: error message
         """
-        self._incr_node_errors(node)
+        self.error_limiter.increment(node)
         if isinstance(msg, bytes):
             msg = msg.decode('utf-8')
         self.logger.error('%(msg)s %(node)s',
@@ -721,7 +701,7 @@ def exception_occurred(self, node, typ, additional_info,
         :param typ: server type
         :param additional_info: additional information to log
         """
-        self._incr_node_errors(node)
+        self.error_limiter.increment(node)
         if 'level' in kwargs:
             log = functools.partial(self.logger.log, kwargs.pop('level'))
             if 'exc_info' not in kwargs:

test/unit/common/test_error_limiter.py

@@ -0,0 +1,102 @@
+# Copyright (c) 2021 NVIDIA
+#
+# 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 unittest
+import mock
+from time import time
+
+from swift.common.error_limiter import ErrorLimiter
+from test.unit import FakeRing
+
+
+class TestErrorLimiter(unittest.TestCase):
+    def setUp(self):
+        self.ring = FakeRing()
+
+    def test_init_config(self):
+        config = {'suppression_interval': 100.9,
+                  'suppression_limit': 5}
+        limiter = ErrorLimiter(**config)
+        self.assertEqual(limiter.suppression_interval, 100.9)
+        self.assertEqual(limiter.suppression_limit, 5)
+
+        config = {'suppression_interval': '100.9',
+                  'suppression_limit': '5'}
+        limiter = ErrorLimiter(**config)
+        self.assertEqual(limiter.suppression_interval, 100.9)
+        self.assertEqual(limiter.suppression_limit, 5)
+
+    def test_init_bad_config(self):
+        with self.assertRaises(ValueError):
+            ErrorLimiter(suppression_interval='bad',
+                         suppression_limit=1)
+
+        with self.assertRaises(TypeError):
+            ErrorLimiter(suppression_interval=None,
+                         suppression_limit=1)
+
+        with self.assertRaises(ValueError):
+            ErrorLimiter(suppression_interval=0,
+                         suppression_limit='bad')
+
+        with self.assertRaises(TypeError):
+            ErrorLimiter(suppression_interval=0,
+                         suppression_limit=None)
+
+    def test_is_limited(self):
+        node = self.ring.devs[-1]
+        limiter = ErrorLimiter(suppression_interval=60, suppression_limit=10)
+
+        now = time()
+        with mock.patch('swift.common.error_limiter.time', return_value=now):
+            self.assertFalse(limiter.is_limited(node))
+            limiter.limit(node)
+            self.assertTrue(limiter.is_limited(node))
+            node_key = limiter.node_key(node)
+            self.assertEqual(limiter.stats.get(node_key),
+                             {'errors': limiter.suppression_limit + 1,
+                              'last_error': now})
+
+    def test_increment(self):
+        node = self.ring.devs[-1]
+        limiter = ErrorLimiter(suppression_interval=60, suppression_limit=10)
+        last_errors = 0
+        node_key = limiter.node_key(node)
+        for _ in range(limiter.suppression_limit):
+            limiter.increment(node)
+            node_errors = limiter.stats.get(node_key)
+            self.assertGreater(node_errors['errors'], last_errors)
+            self.assertFalse(limiter.is_limited(node))
+            last_errors = node_errors['errors']
+
+        # One more to make sure it is > suppression_limit
+        limiter.increment(node)
+        node_errors = limiter.stats.get(node_key)
+        self.assertEqual(limiter.suppression_limit + 1,
+                         node_errors['errors'])
+        self.assertTrue(limiter.is_limited(node))
+        last_time = node_errors['last_error']
+
+        # Simulate time with no errors have gone by.
+        now = last_time + limiter.suppression_interval + 1
+        with mock.patch('swift.common.error_limiter.time',
+                        return_value=now):
+            self.assertFalse(limiter.is_limited(node))
+            self.assertFalse(limiter.stats.get(node_key))
+
+    def test_node_key(self):
+        limiter = ErrorLimiter(suppression_interval=60, suppression_limit=10)
+        node = self.ring.devs[0]
+        expected = '%s:%s/%s' % (node['ip'], node['port'], node['device'])
+        self.assertEqual(expected, limiter.node_key(node))

test/unit/proxy/controllers/test_container.py

@@ -318,7 +318,7 @@ def test_node_errors(self):
 
         for method in ('PUT', 'DELETE', 'POST'):
             def test_status_map(statuses, expected):
-                self.app._error_limiting = {}
+                self.app.error_limiter.stats.clear()
                 req = Request.blank('/v1/a/c', method=method)
                 with mocked_http_conn(*statuses) as fake_conn:
                     resp = req.get_response(self.app)
@@ -355,7 +355,7 @@ def test_status_map(statuses, expected):
             test_status_map(base_status[:2] + [507] + base_status[2:], 201)
             self.assertEqual(node_error_count(
                 self.app, self.container_ring.devs[2]),
-                self.app.error_suppression_limit + 1)
+                self.app.error_limiter.suppression_limit + 1)
 
     def test_response_codes_for_GET(self):
         nodes = self.app.container_ring.replicas

test/unit/proxy/controllers/test_obj.py

@@ -1214,7 +1214,7 @@ def test_PUT_connect_exceptions(self):
         self.app.sort_nodes = lambda n, *args, **kwargs: n  # disable shuffle
 
         def test_status_map(statuses, expected):
-            self.app._error_limiting = {}
+            self.app.error_limiter.stats.clear()
             req = swob.Request.blank('/v1/a/c/o.jpg', method='PUT',
                                      body=b'test body')
             with set_http_connect(*statuses):
@@ -1249,7 +1249,7 @@ def test_status_map(statuses, expected):
         test_status_map(((507, None), 201, 201, 201), 201)
         self.assertEqual(
             node_error_count(self.app, object_ring.devs[0]),
-            self.app.error_suppression_limit + 1)
+            self.app.error_limiter.suppression_limit + 1)
         # response errors
         test_status_map(((100, Timeout()), 201, 201), 201)
         self.assertEqual(
@@ -1260,7 +1260,7 @@ def test_status_map(statuses, expected):
         test_status_map((201, (100, 507), 201), 201)
         self.assertEqual(
             node_error_count(self.app, object_ring.devs[1]),
-            self.app.error_suppression_limit + 1)
+            self.app.error_limiter.suppression_limit + 1)
 
     def test_PUT_connect_exception_with_unicode_path(self):
         expected = 201
@@ -2385,7 +2385,13 @@ def make_key(r):
             }
             for r in log.requests[:4]
         }
-        self.assertEqual(self.app._error_limiting, expected_error_limiting)
+        actual = {}
+        for n in self.app.get_object_ring(int(self.policy)).devs:
+            node_key = self.app.error_limiter.node_key(n)
+            stats = self.app.error_limiter.stats.get(node_key) or {}
+            if stats:
+                actual[self.app.error_limiter.node_key(n)] = stats
+        self.assertEqual(actual, expected_error_limiting)
 
     def test_GET_not_found_when_404_newer(self):
         # if proxy receives a 404, it keeps waiting for other connections until
@@ -2410,14 +2416,14 @@ def test_GET_primaries_error_during_rebalance(self):
             with mocked_http_conn(*codes):
                 resp = req.get_response(self.app)
             self.assertEqual(resp.status_int, 404)
-            self.app._error_limiting = {}  # Reset error limiting
+            self.app.error_limiter.stats.clear()  # Reset error limiting
 
         # one more timeout is past the tipping point
         codes[self.policy.object_ring.replica_count - 2] = Timeout()
         with mocked_http_conn(*codes):
             resp = req.get_response(self.app)
         self.assertEqual(resp.status_int, 503)
-        self.app._error_limiting = {}  # Reset error limiting
+        self.app.error_limiter.stats.clear()  # Reset error limiting
 
         # unless we have tombstones
         with mocked_http_conn(*codes, headers={'X-Backend-Timestamp': '1'}):