Merge pull request #123 from MITLibraries/timx-137-update-timeout-config

Timx 137 update timeout config

Author: hakbailey

Pipfile.lock

@@ -18,6 +18,7 @@
 
 logger = logging.getLogger(__name__)
 
+REQUEST_CONFIG = configure_opensearch_bulk_settings()
 
 # Cluster functions
 
@@ -27,11 +28,15 @@ def configure_opensearch_client(url: str) -> OpenSearch:
 
     Includes the appropriate AWS credentials configuration if the URL is not localhost.
     """
+    logger.info("OpenSearch request configurations: %s", REQUEST_CONFIG)
     if url == "localhost":
         return OpenSearch(
             hosts=[{"host": url, "port": "9200"}],
             http_auth=("admin", "admin"),
             connection_class=RequestsHttpConnection,
+            max_retries=REQUEST_CONFIG["OPENSEARCH_BULK_MAX_RETRIES"],
+            retry_on_timeout=True,
+            timeout=REQUEST_CONFIG["OPENSEARCH_REQUEST_TIMEOUT"],
         )
 
     credentials = boto3.Session().get_credentials()
@@ -43,6 +48,9 @@ def configure_opensearch_client(url: str) -> OpenSearch:
         use_ssl=True,
         verify_certs=True,
         connection_class=RequestsHttpConnection,
+        max_retries=REQUEST_CONFIG["OPENSEARCH_BULK_MAX_RETRIES"],
+        retry_on_timeout=True,
+        timeout=REQUEST_CONFIG["OPENSEARCH_REQUEST_TIMEOUT"],
     )
 
 
@@ -315,16 +323,13 @@ def bulk_index(
     Returns total sums of: records created, records updated, errors, and total records
     processed.
     """
-    bulk_config = configure_opensearch_bulk_settings()
     result = {"created": 0, "updated": 0, "errors": 0, "total": 0}
     actions = helpers.generate_bulk_actions(index, records, "index")
     responses = streaming_bulk(
         client,
         actions,
-        max_chunk_bytes=bulk_config["OPENSEARCH_BULK_MAX_CHUNK_BYTES"],
-        max_retries=bulk_config["OPENSEARCH_BULK_MAX_RETRIES"],
+        max_chunk_bytes=REQUEST_CONFIG["OPENSEARCH_BULK_MAX_CHUNK_BYTES"],
         raise_on_error=False,
-        request_timeout=bulk_config["OPENSEARCH_REQUEST_TIMEOUT"],
     )
     for response in responses:
         if response[0] is False:
@@ -350,7 +355,6 @@ def bulk_index(
     logger.info("All records ingested, refreshing index.")
     response = client.indices.refresh(
         index=index,
-        request_timeout=bulk_config["OPENSEARCH_REQUEST_TIMEOUT"],
     )
     logger.debug(response)
     return result