Add new setting to configure batch sizes for Beats transport

Closes #93.
eht16 · Apr 13, 2024 · be6e30e · be6e30e
1 parent 6c60c27
commit be6e30e
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 3 deletions.
diff --git a/docs/config.rst b/docs/config.rst
@@ -329,6 +329,17 @@ for easy modification.
     *Default*: ``50``
 
 
+``constants.QUEUED_EVENTS_BEATS_BATCH_SIZE``
+
+    Maximum number of events to be sent to Logstash in one batch when using the Beats transport,
+    each batch of events is sent using the same connection and can be considered as a kind
+    of transaction. Should be smaller than `QUEUED_EVENTS_BATCH_SIZE`.
+
+    *Type*: ``integer``
+
+    *Default*: ``25``
+
+
 ``constants.DATABASE_EVENT_CHUNK_SIZE``
 
     Maximum number of events to be updated within one SQLite statement

diff --git a/logstash_async/constants.py b/logstash_async/constants.py
@@ -25,6 +25,10 @@ class Constants:
     QUEUED_EVENTS_FLUSH_COUNT = 50
     # maximum number of events to be sent to Logstash in one batch (i.e. using a single connection)
     QUEUED_EVENTS_BATCH_SIZE = 50
+    # maximum number of events to be sent to Logstash in one batch when using the Beats transport,
+    # each batch of events is sent using the same connection and can be considered as a kind
+    # of transaction. Should be smaller than QUEUED_EVENTS_BATCH_SIZE.
+    QUEUED_EVENTS_BEATS_BATCH_SIZE = 25
     # maximum number of events to be updated within one SQLite statement
     DATABASE_EVENT_CHUNK_SIZE = 750
     # timeout in seconds to "connect" (i.e. open) the SQLite database

diff --git a/logstash_async/transport.py b/logstash_async/transport.py
@@ -230,8 +230,6 @@ def _send_via_socket(self, data):
 
 class BeatsTransport:
 
-    _batch_size = 10
-
     # ----------------------------------------------------------------------
     def __init__(  # pylint: disable=too-many-arguments
             self,
@@ -264,7 +262,7 @@ def close(self):
     def send(self, events, use_logging=False):
         client = pylogbeat.PyLogBeatClient(use_logging=use_logging, **self._client_arguments)
         with client:
-            for events_subset in ichunked(events, self._batch_size):
+            for events_subset in ichunked(events, constants.QUEUED_EVENTS_BEATS_BATCH_SIZE):
                 client.send(events_subset)