Fix OpenSearch connection exhaustion under load

Author: femalves

scan_explorer_service/open_search.py

@@ -3,6 +3,19 @@
 from flask import current_app
 from scan_explorer_service.utils.search_utils import EsFields, OrderOptions
 
+
+def _get_os_client():
+    if not hasattr(current_app, '_os_client') or current_app._os_client is None:
+        url = current_app.config.get('OPEN_SEARCH_URL')
+        current_app._os_client = opensearchpy.OpenSearch(
+            url,
+            timeout=30,
+            max_retries=2,
+            retry_on_timeout=True,
+            pool_maxsize=20,
+        )
+    return current_app._os_client
+
 def create_query_string_query(query_string: str):
     query =  {
         "query": {
@@ -70,7 +83,7 @@ def append_highlight(query: dict):
 
 
 def es_search(query: dict) -> Iterator[str]:
-    es = opensearchpy.OpenSearch(current_app.config.get('OPEN_SEARCH_URL'))
+    es = _get_os_client()
     resp = es.search(index=current_app.config.get(
         'OPEN_SEARCH_INDEX'), body=query)
     return resp

scan_explorer_service/utils/search_utils.py

@@ -53,6 +53,8 @@ class OrderOptions(str, enum.Enum):
 
 def parse_query_args(args):
     qs = re.sub(':\s*', ':', args.get('q', '', str))
+    if not qs or not qs.strip():
+        raise ValueError('Query string is required')
 
     qs, qs_dict = parse_query_string(qs)
 

scan_explorer_service/views/metadata.py

@@ -9,6 +9,7 @@
 from scan_explorer_service.views.view_utils import ApiErrors
 from scan_explorer_service.views.manifest import _cache_delete
 from scan_explorer_service.open_search import EsFields, page_os_search, aggregate_search, page_ocr_os_search
+import opensearchpy
 import requests
 
 bp_metadata = Blueprint('metadata', __name__, url_prefix='/metadata')
@@ -183,6 +184,9 @@ def article_search():
             page_count = page_os_search(qs, page, limit, sort)['hits']['total']['value']
         agg_limit = current_app.config.get("OPEN_SEARCH_AGG_BUCKET_LIMIT", 10000)
         return jsonify(serialize_os_article_result(result, page, limit, text_query, collection_count, page_count, agg_limit))
+    except (opensearchpy.exceptions.ConnectionError, opensearchpy.exceptions.ConnectionTimeout, opensearchpy.exceptions.TransportError) as e:
+        current_app.logger.exception(f"OpenSearch error: {e}")
+        return jsonify(message='Search service temporarily unavailable', type=ApiErrors.SearchError.value), 503
     except Exception as e:
         current_app.logger.exception(f"An exception has occurred: {e}")
         return jsonify(message=str(e), type=ApiErrors.SearchError.value), 400
@@ -200,6 +204,9 @@ def collection_search():
             text_query = qs_dict[SearchOptions.FullText.value]
         agg_limit = current_app.config.get("OPEN_SEARCH_AGG_BUCKET_LIMIT", 10000)
         return jsonify(serialize_os_collection_result(result, page, limit, text_query, agg_limit))
+    except (opensearchpy.exceptions.ConnectionError, opensearchpy.exceptions.ConnectionTimeout, opensearchpy.exceptions.TransportError) as e:
+        current_app.logger.exception(f"OpenSearch error: {e}")
+        return jsonify(message='Search service temporarily unavailable', type=ApiErrors.SearchError.value), 503
     except Exception as e:
         return jsonify(message=str(e), type=ApiErrors.SearchError.value), 400
 
@@ -214,6 +221,9 @@ def page_search():
         if SearchOptions.FullText.value in qs_dict.keys():
             text_query = qs_dict[SearchOptions.FullText.value]
         return jsonify(serialize_os_page_result(result, page, limit, text_query))
+    except (opensearchpy.exceptions.ConnectionError, opensearchpy.exceptions.ConnectionTimeout, opensearchpy.exceptions.TransportError) as e:
+        current_app.logger.exception(f"OpenSearch error: {e}")
+        return jsonify(message='Search service temporarily unavailable', type=ApiErrors.SearchError.value), 503
     except Exception as e:
         return jsonify(message=str(e), type=ApiErrors.SearchError.value), 400
 
@@ -244,5 +254,8 @@ def get_page_ocr():
             result = page_ocr_os_search(collection_id, page_number)
             return serialize_os_page_ocr_result(result)
 
+    except (opensearchpy.exceptions.ConnectionError, opensearchpy.exceptions.ConnectionTimeout, opensearchpy.exceptions.TransportError) as e:
+        current_app.logger.exception(f"OpenSearch error: {e}")
+        return jsonify(message='Search service temporarily unavailable', type=ApiErrors.SearchError.value), 503
     except Exception as e:
         return jsonify(message=str(e), type=ApiErrors.SearchError.value), 400