use opensearch parallel bulk (#492)

* use opensearch parallel bulk

* update changelog

CHANGELOG.md

@@ -7,6 +7,8 @@
 
 ### Improvements
 
+* use parallel_bulk api for opensearch output connector
+
 ### Bugfix
 
 

logprep/connector/opensearch/output.py

@@ -69,7 +69,7 @@ def describe(self) -> str:
 
     def _bulk(self, *args, **kwargs):
         try:
-            helpers.bulk(*args, **kwargs)
+            helpers.parallel_bulk(*args, **kwargs)
         except search.SerializationError as error:
             self._handle_serialization_error(error)
         except search.ConnectionError as error:

tests/unit/connector/test_opensearch_output.py

@@ -136,7 +136,7 @@ def test_build_failed_index_document(self):
         assert failed_document == expected
 
     @mock.patch(
-        "opensearchpy.helpers.bulk",
+        "opensearchpy.helpers.parallel_bulk",
         side_effect=search.SerializationError,
     )
     def test_write_to_search_context_calls_handle_serialization_error_if_serialization_error(
@@ -148,7 +148,7 @@ def test_write_to_search_context_calls_handle_serialization_error_if_serializati
         self.object._handle_serialization_error.assert_called()
 
     @mock.patch(
-        "opensearchpy.helpers.bulk",
+        "opensearchpy.helpers.parallel_bulk",
         side_effect=search.ConnectionError,
     )
     def test_write_to_search_context_calls_handle_connection_error_if_connection_error(self, _):
@@ -158,7 +158,7 @@ def test_write_to_search_context_calls_handle_connection_error_if_connection_err
         self.object._handle_connection_error.assert_called()
 
     @mock.patch(
-        "opensearchpy.helpers.bulk",
+        "opensearchpy.helpers.parallel_bulk",
         side_effect=helpers.BulkIndexError,
     )
     def test_write_to_search_context_calls_handle_bulk_index_error_if_bulk_index_error(self, _):
@@ -167,7 +167,7 @@ def test_write_to_search_context_calls_handle_bulk_index_error_if_bulk_index_err
         self.object._write_to_search_context({"dummy": "event"})
         self.object._handle_bulk_index_error.assert_called()
 
-    @mock.patch("opensearchpy.helpers.bulk")
+    @mock.patch("opensearchpy.helpers.parallel_bulk")
     def test__handle_bulk_index_error_calls_bulk(self, fake_bulk):
         mock_bulk_index_error = mock.MagicMock()
         mock_bulk_index_error.errors = [
@@ -181,7 +181,7 @@ def test__handle_bulk_index_error_calls_bulk(self, fake_bulk):
         self.object._handle_bulk_index_error(mock_bulk_index_error)
         fake_bulk.assert_called()
 
-    @mock.patch("opensearchpy.helpers.bulk")
+    @mock.patch("opensearchpy.helpers.parallel_bulk")
     def test_handle_bulk_index_error_calls_bulk_with_error_documents(self, fake_bulk):
         mock_bulk_index_error = mock.MagicMock()
         mock_bulk_index_error.errors = [