Correct bulk size in bulk param source

With this commit we correct the bulk size to *not* take the
action-and-metadata line into account for the bulk size calculation.
This problem has been introduced in
863d532.

Relates elastic#45

Author: danielmitterdorfer

eventdata/parameter_sources/elasticlogs_bulk_source.py

@@ -81,7 +81,13 @@ def __init__(self, track, params, **kwargs):
         self.orig_args = [track, params, kwargs]
         self._indices = track.indices
         self._params = params
-        self._randomevent = RandomEvent(params)
+        # we could also do `kwargs.get("random_event", RandomEvent(params))` but that would call the constructor eagerly
+        # which we want to avoid because this can cause significant overhead.
+        if "random_event" in kwargs:
+            self._randomevent = kwargs["random_event"]
+        else:
+            self._randomevent = RandomEvent(params)
+
 
         self._bulk_size = 1000
         if 'bulk-size' in params.keys():
@@ -192,7 +198,8 @@ def params(self):
         response = {
             "body": "\n".join(bulk_array),
             "action-metadata-present": True,
-            "bulk-size": len(bulk_array)
+            # the bulk array contains the action-and-metadata line and the actual document
+            "bulk-size": len(bulk_array) // 2
         }
 
         if "pipeline" in self._params.keys():

tests/parameter_sources/elasticlogs_bulk_source_test.py

@@ -0,0 +1,51 @@
+from eventdata.parameter_sources.elasticlogs_bulk_source import ElasticlogsBulkSource
+
+
+class StaticEventGenerator:
+    def __init__(self, index, type, doc, at_most=-1):
+        self.index = index
+        self.type = type
+        self.doc = doc
+        self.at_most = at_most
+
+    def generate_event(self):
+        if self.at_most == 0:
+            raise StopIteration()
+        self.at_most -= 1
+        return self.doc, self.index, self.type
+
+
+class StaticTrack:
+    def __init__(self):
+        self.indices = []
+
+
+def test_generates_a_complete_bulk():
+    expected_bulk_size = 10
+
+    generator = StaticEventGenerator(index="elasticlogs", type="_doc", doc='{"location": [-0.1485188,51.5250666]}')
+    param_source = ElasticlogsBulkSource(track=StaticTrack(), params={
+        "index": "elasticlogs",
+        "bulk-size": expected_bulk_size
+    }, random_event=generator)
+    client_param_source = param_source.partition(partition_index=0, total_partitions=1)
+
+    generated_params = client_param_source.params()
+    assert len(generated_params["body"].split("\n")) == 2 * expected_bulk_size
+    assert generated_params["action-metadata-present"] is True
+    assert generated_params["bulk-size"] == expected_bulk_size
+
+
+def test_generates_a_bulk_that_ends_prematurely():
+    generator = StaticEventGenerator(index="elasticlogs", type="_doc", doc='{"loc": [-0.14851,51.5250]}', at_most=5)
+    param_source = ElasticlogsBulkSource(track=StaticTrack(), params={
+        "index": "elasticlogs",
+        "bulk-size": 10
+    }, random_event=generator)
+    client_param_source = param_source.partition(partition_index=0, total_partitions=1)
+
+    generated_params = client_param_source.params()
+    # the actual bulk size does not matter but instead that the generator stopped prematurely after 5 items
+    assert len(generated_params["body"].split("\n")) == 10
+    assert generated_params["action-metadata-present"] is True
+    assert generated_params["bulk-size"] == 5