Merge pull request #7 from searxng/metrics

Metrics

5 files changed, 145 insertions, 131 deletions

diff --git a/searx/search/__init__.py b/searx/search/__init__.py
index f777e8595..9b26f38de 100644
--- a/searx/search/__init__.py
+++ b/searx/search/__init__.py
@@ -18,7 +18,7 @@ along with searx. If not, see < http://www.gnu.org/licenses/ >.
 import typing
 import gc
 import threading
-from time import time
+from timeit import default_timer
 from uuid import uuid4
 from _thread import start_new_thread
 
@@ -31,6 +31,7 @@ from searx.plugins import plugins
 from searx.search.models import EngineRef, SearchQuery
 from searx.search.processors import processors, initialize as initialize_processors
 from searx.search.checker import initialize as initialize_checker
+from searx.metrics import initialize as initialize_metrics, counter_inc, histogram_observe_time
 
 
 logger = logger.getChild('search')
@@ -50,6 +51,7 @@ else:
 def initialize(settings_engines=None, enable_checker=False):
     settings_engines = settings_engines or settings['engines']
     initialize_processors(settings_engines)
+    initialize_metrics([engine['name'] for engine in settings_engines])
     if enable_checker:
         initialize_checker()
 
@@ -106,13 +108,16 @@ class Search:
         for engineref in self.search_query.engineref_list:
             processor = processors[engineref.name]
 
+            # stop the request now if the engine is suspend
+            if processor.extend_container_if_suspended(self.result_container):
+                continue
+
             # set default request parameters
             request_params = processor.get_params(self.search_query, engineref.category)
             if request_params is None:
                 continue
 
-            with threading.RLock():
-                processor.engine.stats['sent_search_count'] += 1
+            counter_inc('engine', engineref.name, 'search', 'count', 'sent')
 
             # append request to list
             requests.append((engineref.name, self.search_query.query, request_params))
@@ -157,7 +162,7 @@ class Search:
 
         for th in threading.enumerate():
             if th.name == search_id:
-                remaining_time = max(0.0, self.actual_timeout - (time() - self.start_time))
+                remaining_time = max(0.0, self.actual_timeout - (default_timer() - self.start_time))
                 th.join(remaining_time)
                 if th.is_alive():
                     th._timeout = True
@@ -180,12 +185,10 @@ class Search:
 
     # do search-request
     def search(self):
-        self.start_time = time()
-
+        self.start_time = default_timer()
         if not self.search_external_bang():
             if not self.search_answerers():
                 self.search_standard()
-
         return self.result_container
 
 
diff --git a/searx/search/checker/impl.py b/searx/search/checker/impl.py
index e54b3f68d..dd090c513 100644
--- a/searx/search/checker/impl.py
+++ b/searx/search/checker/impl.py
@@ -4,8 +4,8 @@ import typing
 import types
 import functools
 import itertools
-import threading
 from time import time
+from timeit import default_timer
 from urllib.parse import urlparse
 
 import re
@@ -17,6 +17,7 @@ from searx import network, logger
 from searx.results import ResultContainer
 from searx.search.models import SearchQuery, EngineRef
 from searx.search.processors import EngineProcessor
+from searx.metrics import counter_inc
 
 
 logger = logger.getChild('searx.search.checker')
@@ -385,9 +386,8 @@ class Checker:
         engineref_category = search_query.engineref_list[0].category
         params = self.processor.get_params(search_query, engineref_category)
         if params is not None:
-            with threading.RLock():
-                self.processor.engine.stats['sent_search_count'] += 1
-            self.processor.search(search_query.query, params, result_container, time(), 5)
+            counter_inc('engine', search_query.engineref_list[0].name, 'search', 'count', 'sent')
+            self.processor.search(search_query.query, params, result_container, default_timer(), 5)
         return result_container
 
     def get_result_container_tests(self, test_name: str, search_query: SearchQuery) -> ResultContainerTests:
diff --git a/searx/search/processors/abstract.py b/searx/search/processors/abstract.py
index 26dab069f..854f6df6a 100644
--- a/searx/search/processors/abstract.py
+++ b/searx/search/processors/abstract.py
@@ -1,17 +1,110 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
+import threading
 from abc import abstractmethod, ABC
+from timeit import default_timer
+
 from searx import logger
+from searx.engines import settings
+from searx.network import get_time_for_thread, get_network
+from searx.metrics import histogram_observe, counter_inc, count_exception, count_error
+from searx.exceptions import SearxEngineAccessDeniedException
 
 
 logger = logger.getChild('searx.search.processor')
+SUSPENDED_STATUS = {}
+
+
+class SuspendedStatus:
+
+    __slots__ = 'suspend_end_time', 'suspend_reason', 'continuous_errors', 'lock'
+
+    def __init__(self):
+        self.lock = threading.Lock()
+        self.continuous_errors = 0
+        self.suspend_end_time = 0
+        self.suspend_reason = None
+
+    @property
+    def is_suspended(self):
+        return self.suspend_end_time >= default_timer()
+
+    def suspend(self, suspended_time, suspend_reason):
+        with self.lock:
+            # update continuous_errors / suspend_end_time
+            self.continuous_errors += 1
+            if suspended_time is None:
+                suspended_time = min(settings['search']['max_ban_time_on_fail'],
+                                     self.continuous_errors * settings['search']['ban_time_on_fail'])
+            self.suspend_end_time = default_timer() + suspended_time
+            self.suspend_reason = suspend_reason
+        logger.debug('Suspend engine for %i seconds', suspended_time)
+
+    def resume(self):
+        with self.lock:
+            # reset the suspend variables
+            self.continuous_errors = 0
+            self.suspend_end_time = 0
+            self.suspend_reason = None
 
 
 class EngineProcessor(ABC):
 
+    __slots__ = 'engine', 'engine_name', 'lock', 'suspended_status'
+
     def __init__(self, engine, engine_name):
         self.engine = engine
         self.engine_name = engine_name
+        key = get_network(self.engine_name)
+        key = id(key) if key else self.engine_name
+        self.suspended_status = SUSPENDED_STATUS.setdefault(key, SuspendedStatus())
+
+    def handle_exception(self, result_container, reason, exception, suspend=False, display_exception=True):
+        # update result_container
+        error_message = str(exception) if display_exception and exception else None
+        result_container.add_unresponsive_engine(self.engine_name, reason, error_message)
+        # metrics
+        counter_inc('engine', self.engine_name, 'search', 'count', 'error')
+        if exception:
+            count_exception(self.engine_name, exception)
+        else:
+            count_error(self.engine_name, reason)
+        # suspend the engine ?
+        if suspend:
+            suspended_time = None
+            if isinstance(exception, SearxEngineAccessDeniedException):
+                suspended_time = exception.suspended_time
+            self.suspended_status.suspend(suspended_time, reason)  # pylint: disable=no-member
+
+    def _extend_container_basic(self, result_container, start_time, search_results):
+        # update result_container
+        result_container.extend(self.engine_name, search_results)
+        engine_time = default_timer() - start_time
+        page_load_time = get_time_for_thread()
+        result_container.add_timing(self.engine_name, engine_time, page_load_time)
+        # metrics
+        counter_inc('engine', self.engine_name, 'search', 'count', 'successful')
+        histogram_observe(engine_time, 'engine', self.engine_name, 'time', 'total')
+        if page_load_time is not None:
+            histogram_observe(page_load_time, 'engine', self.engine_name, 'time', 'http')
+
+    def extend_container(self, result_container, start_time, search_results):
+        if getattr(threading.current_thread(), '_timeout', False):
+            # the main thread is not waiting anymore
+            self.handle_exception(result_container, 'Timeout', None)
+        else:
+            # check if the engine accepted the request
+            if search_results is not None:
+                self._extend_container_basic(result_container, start_time, search_results)
+            self.suspended_status.resume()
+
+    def extend_container_if_suspended(self, result_container):
+        if self.suspended_status.is_suspended:
+            result_container.add_unresponsive_engine(self.engine_name,
+                                                     self.suspended_status.suspend_reason,
+                                                     suspended=True)
+            return True
+        return False
 
     def get_params(self, search_query, engine_category):
         # if paging is not supported, skip
diff --git a/searx/search/processors/offline.py b/searx/search/processors/offline.py
index ede8eb5e1..5186b346a 100644
--- a/searx/search/processors/offline.py
+++ b/searx/search/processors/offline.py
@@ -1,51 +1,26 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
-import threading
-from time import time
 from searx import logger
-from searx.metrology.error_recorder import record_exception, record_error
 from searx.search.processors.abstract import EngineProcessor
 
 
-logger = logger.getChild('search.processor.offline')
+logger = logger.getChild('searx.search.processor.offline')
 
 
 class OfflineProcessor(EngineProcessor):
 
     engine_type = 'offline'
 
-    def _record_stats_on_error(self, result_container, start_time):
-        engine_time = time() - start_time
-        result_container.add_timing(self.engine_name, engine_time, engine_time)
-
-        with threading.RLock():
-            self.engine.stats['errors'] += 1
-
     def _search_basic(self, query, params):
         return self.engine.search(query, params)
 
     def search(self, query, params, result_container, start_time, timeout_limit):
         try:
             search_results = self._search_basic(query, params)
-
-            if search_results:
-                result_container.extend(self.engine_name, search_results)
-
-                engine_time = time() - start_time
-                result_container.add_timing(self.engine_name, engine_time, engine_time)
-                with threading.RLock():
-                    self.engine.stats['engine_time'] += engine_time
-                    self.engine.stats['engine_time_count'] += 1
-
+            self.extend_container(result_container, start_time, search_results)
         except ValueError as e:
-            record_exception(self.engine_name, e)
-            self._record_stats_on_error(result_container, start_time)
+            # do not record the error
             logger.exception('engine {0} : invalid input : {1}'.format(self.engine_name, e))
         except Exception as e:
-            record_exception(self.engine_name, e)
-            self._record_stats_on_error(result_container, start_time)
-            result_container.add_unresponsive_engine(self.engine_name, 'unexpected crash', str(e))
+            self.handle_exception(result_container, 'unexpected crash', e)
             logger.exception('engine {0} : exception : {1}'.format(self.engine_name, e))
-        else:
-            if getattr(threading.current_thread(), '_timeout', False):
-                record_error(self.engine_name, 'Timeout')
diff --git a/searx/search/processors/online.py b/searx/search/processors/online.py
index 66719ea9b..c39937023 100644
--- a/searx/search/processors/online.py
+++ b/searx/search/processors/online.py
@@ -1,23 +1,21 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
 from time import time
-import threading
 import asyncio
 
 import httpx
 
 import searx.network
-from searx.engines import settings
 from searx import logger
 from searx.utils import gen_useragent
 from searx.exceptions import (SearxEngineAccessDeniedException, SearxEngineCaptchaException,
                               SearxEngineTooManyRequestsException,)
-from searx.metrology.error_recorder import record_exception, record_error
+from searx.metrics.error_recorder import count_error
 
 from searx.search.processors.abstract import EngineProcessor
 
 
-logger = logger.getChild('search.processor.online')
+logger = logger.getChild('searx.search.processor.online')
 
 
 def default_request_params():
@@ -41,11 +39,6 @@ class OnlineProcessor(EngineProcessor):
         if params is None:
             return None
 
-        # skip suspended engines
-        if self.engine.suspend_end_time >= time():
-            logger.debug('Engine currently suspended: %s', self.engine_name)
-            return None
-
         # add default params
         params.update(default_request_params())
 
@@ -97,9 +90,10 @@ class OnlineProcessor(EngineProcessor):
             status_code = str(response.status_code or '')
             reason = response.reason_phrase or ''
             hostname = response.url.host
-            record_error(self.engine_name,
-                         '{} redirects, maximum: {}'.format(len(response.history), soft_max_redirects),
-                         (status_code, reason, hostname))
+            count_error(self.engine_name,
+                        '{} redirects, maximum: {}'.format(len(response.history), soft_max_redirects),
+                        (status_code, reason, hostname),
+                        secondary=True)
 
         return response
 
@@ -130,89 +124,38 @@ class OnlineProcessor(EngineProcessor):
         # set the network
         searx.network.set_context_network_name(self.engine_name)
 
-        # suppose everything will be alright
-        http_exception = False
-        suspended_time = None
-
         try:
             # send requests and parse the results
             search_results = self._search_basic(query, params)
-
-            # check if the engine accepted the request
-            if search_results is not None:
-                # yes, so add results
-                result_container.extend(self.engine_name, search_results)
-
-                # update engine time when there is no exception
-                engine_time = time() - start_time
-                page_load_time = searx.network.get_time_for_thread()
-                result_container.add_timing(self.engine_name, engine_time, page_load_time)
-                with threading.RLock():
-                    self.engine.stats['engine_time'] += engine_time
-                    self.engine.stats['engine_time_count'] += 1
-                    # update stats with the total HTTP time
-                    self.engine.stats['page_load_time'] += page_load_time
-                    self.engine.stats['page_load_count'] += 1
-        except Exception as e:
-            record_exception(self.engine_name, e)
-
-            # Timing
-            engine_time = time() - start_time
-            page_load_time = searx.network.get_time_for_thread()
-            result_container.add_timing(self.engine_name, engine_time, page_load_time)
-
-            # Record the errors
-            with threading.RLock():
-                self.engine.stats['errors'] += 1
-
-            if (issubclass(e.__class__, (httpx.TimeoutException, asyncio.TimeoutError))):
-                result_container.add_unresponsive_engine(self.engine_name, 'HTTP timeout')
-                # requests timeout (connect or read)
-                logger.error("engine {0} : HTTP requests timeout"
+            self.extend_container(result_container, start_time, search_results)
+        except (httpx.TimeoutException, asyncio.TimeoutError) as e:
+            # requests timeout (connect or read)
+            self.handle_exception(result_container, 'HTTP timeout', e, suspend=True, display_exception=False)
+            logger.error("engine {0} : HTTP requests timeout"
+                         "(search duration : {1} s, timeout: {2} s) : {3}"
+                         .format(self.engine_name, time() - start_time,
+                                 timeout_limit,
+                                 e.__class__.__name__))
+        except (httpx.HTTPError, httpx.StreamError) as e:
+            # other requests exception
+            self.handle_exception(result_container, 'HTTP error', e, suspend=True, display_exception=False)
+            logger.exception("engine {0} : requests exception"
                              "(search duration : {1} s, timeout: {2} s) : {3}"
-                             .format(self.engine_name, engine_time, timeout_limit, e.__class__.__name__))
-                http_exception = True
-            elif (issubclass(e.__class__, (httpx.HTTPError, httpx.StreamError))):
-                result_container.add_unresponsive_engine(self.engine_name, 'HTTP error')
-                # other requests exception
-                logger.exception("engine {0} : requests exception"
-                                 "(search duration : {1} s, timeout: {2} s) : {3}"
-                                 .format(self.engine_name, engine_time, timeout_limit, e))
-                http_exception = True
-            elif (issubclass(e.__class__, SearxEngineCaptchaException)):
-                result_container.add_unresponsive_engine(self.engine_name, 'CAPTCHA required')
-                logger.exception('engine {0} : CAPTCHA'.format(self.engine_name))
-                suspended_time = e.suspended_time  # pylint: disable=no-member
-            elif (issubclass(e.__class__, SearxEngineTooManyRequestsException)):
-                result_container.add_unresponsive_engine(self.engine_name, 'too many requests')
-                logger.exception('engine {0} : Too many requests'.format(self.engine_name))
-                suspended_time = e.suspended_time  # pylint: disable=no-member
-            elif (issubclass(e.__class__, SearxEngineAccessDeniedException)):
-                result_container.add_unresponsive_engine(self.engine_name, 'blocked')
-                logger.exception('engine {0} : Searx is blocked'.format(self.engine_name))
-                suspended_time = e.suspended_time  # pylint: disable=no-member
-            else:
-                result_container.add_unresponsive_engine(self.engine_name, 'unexpected crash')
-                # others errors
-                logger.exception('engine {0} : exception : {1}'.format(self.engine_name, e))
-        else:
-            if getattr(threading.current_thread(), '_timeout', False):
-                record_error(self.engine_name, 'Timeout')
-
-        # suspend the engine if there is an HTTP error
-        # or suspended_time is defined
-        with threading.RLock():
-            if http_exception or suspended_time:
-                # update continuous_errors / suspend_end_time
-                self.engine.continuous_errors += 1
-                if suspended_time is None:
-                    suspended_time = min(settings['search']['max_ban_time_on_fail'],
-                                         self.engine.continuous_errors * settings['search']['ban_time_on_fail'])
-                self.engine.suspend_end_time = time() + suspended_time
-            else:
-                # reset the suspend variables
-                self.engine.continuous_errors = 0
-                self.engine.suspend_end_time = 0
+                             .format(self.engine_name, time() - start_time,
+                                     timeout_limit,
+                                     e))
+        except SearxEngineCaptchaException as e:
+            self.handle_exception(result_container, 'CAPTCHA required', e, suspend=True, display_exception=False)
+            logger.exception('engine {0} : CAPTCHA'.format(self.engine_name))
+        except SearxEngineTooManyRequestsException as e:
+            self.handle_exception(result_container, 'too many requests', e, suspend=True, display_exception=False)
+            logger.exception('engine {0} : Too many requests'.format(self.engine_name))
+        except SearxEngineAccessDeniedException as e:
+            self.handle_exception(result_container, 'blocked', e, suspend=True, display_exception=False)
+            logger.exception('engine {0} : Searx is blocked'.format(self.engine_name))
+        except Exception as e:
+            self.handle_exception(result_container, 'unexpected crash', e, display_exception=False)
+            logger.exception('engine {0} : exception : {1}'.format(self.engine_name, e))
 
     def get_default_tests(self):
         tests = {}