Cleanup Kibana Parameter Source

With this commit we remove some dead code from the Kibana parameter source. We
also turn `dashboard` into a mandatory parameter and add tests.

Relates elastic#53

Author: danielmitterdorfer

README.md

@@ -312,7 +312,7 @@ The generator allows data to be generated in real-time or against a set date/tin
 
 ### elasticlogs\_kibana\_source
 
-This parameter source supports simulating two different types of dashboards.
+This parameter source supports simulating three different types of dashboards. One of the following needs to be selected by specifying the mandatory parameter `dashboard`:
 
 **traffic** - This dashboard contains 7 visualisations and presents different types of traffic statistics. In structure it is similar to the `Nginx Overview` dashboard that comes with the Filebeat Nginx Module. It does aggregate across all records in the index and is therefore a quite 'heavy' dashboard.
 

eventdata/parameter_sources/elasticlogs_kibana_source.py

@@ -55,7 +55,7 @@ class ElasticlogsKibanaSource:
     Simulates a set of sample Kibana dashboards for the elasticlogs data set.
 
     It expects the parameter hash to contain the following keys:
-        "dashboard"             -   String indicating which dashboard to simulate. Options are 'traffic', 'content_issues' and 'discover'. Defaults to 'traffic'.
+        "dashboard"             -   String indicating which dashboard to simulate. Options are 'traffic', 'content_issues' and 'discover'.
         "query_string"          -   String indicating file to load or list of strings indicating actual query parameters to randomize during benchmarking. Defaults 
                                     to ["*"], If a list has been specified, a random value will be selected.
         "index_pattern"         -   String or list of strings representing the index pattern to query. Defaults to 'elasticlogs-*'. If a list has 
@@ -85,13 +85,14 @@ def __init__(self, track, params, **kwargs):
         self._indices = track.indices
         self._index_pattern = params.get("index_pattern", "elasticlogs-*")
         self._query_string_list = ["*"]
-        self._dashboard = "traffic"
+        self._dashboard = params["dashboard"]
         self._discover_size = params.get("discover_size", 500)
         self._ignore_throttled = params.get("ignore_throttled", True)
         self._debug = params.get("debug", False)
         self._pre_filter_shard_size = params.get("pre_filter_shard_size", 1)
         self._window_length = params.get("window_length", "1d")
         self.infinite = True
+        self.utcnow = kwargs.get("utcnow", datetime.datetime.utcnow)
 
         random.seed()
 
@@ -106,11 +107,8 @@ def __init__(self, track, params, **kwargs):
             else:
                 self._query_string_list = params["query_string"]
 
-        if "dashboard" in params.keys():
-            if params["dashboard"] in available_dashboards:
-                self._dashboard = params["dashboard"]
-            else:
-                logger.info("[kibana] Illegal dashboard configured (%s). Using default dashboard instead.", params["dashboard"])
+        if self._dashboard not in available_dashboards:
+            raise ConfigurationError("Unknown dashboard [{}]. Must be one of {}.".format(self._dashboard, available_dashboards))
 
         key = "{}_@timestamp".format(self._index_pattern)
         if key in gs.global_fieldstats.keys():
@@ -121,8 +119,8 @@ def __init__(self, track, params, **kwargs):
         else:
             self._fieldstats_provided = False
 
-        re1 = re.compile("^(\d+\.*\d*)([dhm])$")
-        re2 = re.compile("^(\d+\.*\d*)%$")
+        re1 = re.compile(r"^(\d+\.*\d*)([dhm])$")
+        re2 = re.compile(r"^(\d+\.*\d*)%$")
 
         m1 = re1.match(self._window_length)
         m2 = re2.match(self._window_length)
@@ -218,7 +216,7 @@ def __window_boundary_to_ms(self, wb):
             return wb["ts_ms"]
 
     def __current_epoch_ms(self):
-        dt = datetime.datetime.utcnow()
+        dt = self.utcnow()
         return (dt-epoch).total_seconds() *1000
 
     def __parse_window_parameters(self, window_spec):
@@ -250,7 +248,7 @@ def __parse_window_parameters(self, window_spec):
 
                 window_end_spec.append({"type": "relative", "offset_ms": offset_ms})
             elif m2:
-                c = re.split("\D+", i)
+                c = re.split(r"\D+", i)
 
                 dt = datetime.datetime.strptime("{} {} {} {} {} {} UTC".format(c[0], c[1], c[2], c[3], c[4], c[5]), "%Y %m %d %H %M %S %Z")
                 epoch_ms = (int)((dt-epoch).total_seconds() * 1000)
@@ -279,21 +277,6 @@ def __parse_window_parameters(self, window_spec):
 
         return window_end_spec
 
-    def __unit_string_to_milliseconds(self, string):
-        m = re.match(r"^(\d+\.*\d*)([dhm])$", string)
-        if m:
-            val = float(m.group(1))
-            unit = m.group(2)
-
-            if unit == "m":
-                return False, int(60*val*1000)
-            elif unit == "h":
-                return False, int(3600*val*1000)
-            else:
-                return False, int(86400*val*1000)
-        else:
-            return True, None
-
     def __determine_interval(self, window_size_seconds, target_bars, max_bars):
         available_intervals = [
             {"unit": "1s",      "length_sec": 1},
@@ -341,11 +324,6 @@ def __determine_interval(self, window_size_seconds, target_bars, max_bars):
     def __get_preference(self):
         return int(round(time.time() * 1000))
 
-    def __print_ts(self, ts):
-        ts_s = int(ts / 1000)
-        dt = datetime.datetime.utcfromtimestamp(ts_s)
-        return dt.isoformat()
-
     def __content_issues_dashboard(self, index_pattern, query_string, interval, ts_min_ms, ts_max_ms, ignore_throttled):
         preference = self.__get_preference()
         header = {

tests/parameter_sources/__init__.py

@@ -14,3 +14,8 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+
+
+class StaticTrack:
+    def __init__(self):
+        self.indices = []

tests/parameter_sources/elasticlogs_bulk_source_test.py

@@ -16,6 +16,7 @@
 # under the License.
 
 from eventdata.parameter_sources.elasticlogs_bulk_source import ElasticlogsBulkSource
+from tests.parameter_sources import StaticTrack
 
 
 class StaticEventGenerator:
@@ -32,11 +33,6 @@ def generate_event(self):
         return self.doc, self.index, self.type
 
 
-class StaticTrack:
-    def __init__(self):
-        self.indices = []
-
-
 def test_generates_a_complete_bulk():
     expected_bulk_size = 10
 

tests/parameter_sources/elasticlogs_kibana_source_test.py

@@ -0,0 +1,121 @@
+# Licensed to Elasticsearch B.V. under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Elasticsearch B.V. licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#	http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from datetime import datetime
+from unittest import mock
+
+import pytest
+
+from eventdata.parameter_sources.elasticlogs_kibana_source import ElasticlogsKibanaSource, ConfigurationError
+from tests.parameter_sources import StaticTrack
+
+
+@mock.patch("time.time")
+def test_create_discover(time):
+    time.return_value = 5000
+
+    param_source = ElasticlogsKibanaSource(track=StaticTrack(), params={
+        "dashboard": "discover"
+    }, utcnow=lambda: datetime(year=2019, month=11, day=11))
+    response = param_source.params()
+
+    assert response == {
+        "body": [
+            {
+                "index": "elasticlogs-*",
+                "ignore_unavailable": True,
+                "preference": 5000000,
+                "ignore_throttled": True
+            },
+            {
+                "version": True,
+                "size": 500,
+                "sort": [
+                    {"@timestamp": {"order": "desc", "unmapped_type": "boolean"}}
+                ],
+                "_source": {"excludes": []},
+                "aggs": {
+                    "2": {
+                        "date_histogram": {
+                            "field": "@timestamp",
+                            "interval": "30m",
+                            "time_zone": "Europe/London",
+                            "min_doc_count": 1}
+                    }
+                },
+                "stored_fields": ["*"],
+                "script_fields": {},
+                "docvalue_fields": ["@timestamp"],
+                "query": {
+                    "bool": {
+                        "must": [
+                            {
+                                "query_string": {
+                                    "query": "*",
+                                    "analyze_wildcard": True,
+                                    "default_field": "*"
+                                }
+                            },
+                            {
+                                "range": {
+                                    "@timestamp": {
+                                        "gte": 1573344000000,
+                                        "lte": 1573430400000,
+                                        "format": "epoch_millis"
+                                    }
+                                }
+                            }
+                        ],
+                        "filter": [],
+                        "should": [],
+                        "must_not": []
+                    }
+                },
+                "highlight": {
+                    "pre_tags": ["@kibana-highlighted-field@"],
+                    "post_tags": ["@/kibana-highlighted-field@"],
+                    "fields": {"*": {}},
+                    "fragment_size": 2147483647
+                }
+            }
+        ],
+        "meta_data": {
+            "interval": "30m",
+            "index_pattern": "elasticlogs-*",
+            "query_string": "*",
+            "dashboard": "discover",
+            "window_length": "1d",
+            "ignore_throttled": True,
+            "pre_filter_shard_size": 1,
+            "debug": False
+        }
+    }
+
+
+def test_dashboard_is_mandatory():
+    with pytest.raises(KeyError) as ex:
+        ElasticlogsKibanaSource(track=StaticTrack(), params={})
+
+    assert "'dashboard'" == str(ex.value)
+
+
+def test_invalid_dashboard_raises_error():
+    with pytest.raises(ConfigurationError) as ex:
+        ElasticlogsKibanaSource(track=StaticTrack(), params={"dashboard": "unknown"})
+
+    assert "Unknown dashboard [unknown]. Must be one of ['traffic', 'content_issues', 'discover']." == str(ex.value)
+