From aae7830d14242ac1f98232f428654c5d2c9c5eb2 Mon Sep 17 00:00:00 2001 From: Alexandre Flament Date: Tue, 13 Apr 2021 15:21:53 +0200 Subject: [mod] refactoring: processors Report to the user suspended engines. searx.search.processor.abstract: * manages suspend time (per network). * reports suspended time to the ResultContainer (method extend_container_if_suspended) * adds the results to the ResultContainer (method extend_container) * handles exceptions (method handle_exception) --- searx/search/processors/abstract.py | 98 +++++++++++++++++++++++++++++++++++++ 1 file changed, 98 insertions(+) (limited to 'searx/search/processors/abstract.py') diff --git a/searx/search/processors/abstract.py b/searx/search/processors/abstract.py index 26dab069f..e32d8f067 100644 --- a/searx/search/processors/abstract.py +++ b/searx/search/processors/abstract.py @@ -1,17 +1,115 @@ # SPDX-License-Identifier: AGPL-3.0-or-later +import threading from abc import abstractmethod, ABC +from time import time + from searx import logger +from searx.engines import settings +from searx.network import get_time_for_thread, get_network +from searx.metrology.error_recorder import record_exception, record_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 >= time() + + 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 = time() + 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 + self.lock = threading.Lock() + 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 + with self.lock: + self.engine.stats['errors'] += 1 + if exception: + record_exception(self.engine_name, exception) + else: + record_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 = time() - start_time + page_load_time = get_time_for_thread() + result_container.add_timing(self.engine_name, engine_time, page_load_time) + # metrics + with self.lock: + self.engine.stats['engine_time'] += engine_time + self.engine.stats['engine_time_count'] += 1 + # update stats with the total HTTP time + if page_load_time is not None and 'page_load_time' in self.engine.stats: + self.engine.stats['page_load_time'] += page_load_time + self.engine.stats['page_load_count'] += 1 + + 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 -- cgit v1.2.3 From 7acd7ffc02d14d175ec2a99ba984e47d8cb65d7d Mon Sep 17 00:00:00 2001 From: Alexandre Flament Date: Wed, 14 Apr 2021 17:23:15 +0200 Subject: [enh] rewrite and enhance metrics --- searx/search/processors/abstract.py | 29 ++++++++++++----------------- 1 file changed, 12 insertions(+), 17 deletions(-) (limited to 'searx/search/processors/abstract.py') diff --git a/searx/search/processors/abstract.py b/searx/search/processors/abstract.py index e32d8f067..854f6df6a 100644 --- a/searx/search/processors/abstract.py +++ b/searx/search/processors/abstract.py @@ -2,12 +2,12 @@ import threading from abc import abstractmethod, ABC -from time import time +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.metrology.error_recorder import record_exception, record_error +from searx.metrics import histogram_observe, counter_inc, count_exception, count_error from searx.exceptions import SearxEngineAccessDeniedException @@ -27,7 +27,7 @@ class SuspendedStatus: @property def is_suspended(self): - return self.suspend_end_time >= time() + return self.suspend_end_time >= default_timer() def suspend(self, suspended_time, suspend_reason): with self.lock: @@ -36,7 +36,7 @@ class SuspendedStatus: 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 = time() + suspended_time + self.suspend_end_time = default_timer() + suspended_time self.suspend_reason = suspend_reason logger.debug('Suspend engine for %i seconds', suspended_time) @@ -55,7 +55,6 @@ class EngineProcessor(ABC): def __init__(self, engine, engine_name): self.engine = engine self.engine_name = engine_name - self.lock = threading.Lock() key = get_network(self.engine_name) key = id(key) if key else self.engine_name self.suspended_status = SUSPENDED_STATUS.setdefault(key, SuspendedStatus()) @@ -65,12 +64,11 @@ class EngineProcessor(ABC): error_message = str(exception) if display_exception and exception else None result_container.add_unresponsive_engine(self.engine_name, reason, error_message) # metrics - with self.lock: - self.engine.stats['errors'] += 1 + counter_inc('engine', self.engine_name, 'search', 'count', 'error') if exception: - record_exception(self.engine_name, exception) + count_exception(self.engine_name, exception) else: - record_error(self.engine_name, reason) + count_error(self.engine_name, reason) # suspend the engine ? if suspend: suspended_time = None @@ -81,17 +79,14 @@ class EngineProcessor(ABC): def _extend_container_basic(self, result_container, start_time, search_results): # update result_container result_container.extend(self.engine_name, search_results) - engine_time = time() - start_time + 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 - with self.lock: - self.engine.stats['engine_time'] += engine_time - self.engine.stats['engine_time_count'] += 1 - # update stats with the total HTTP time - if page_load_time is not None and 'page_load_time' in self.engine.stats: - self.engine.stats['page_load_time'] += page_load_time - self.engine.stats['page_load_count'] += 1 + 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): -- cgit v1.2.3