Make benchmarks work with existing docker-compose setup

benchmarking/__init__.py

@@ -0,0 +1,4 @@
+# Licensed to Elasticsearch B.V under one or more agreements.
+# Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
+# See the LICENSE file in the project root for more information
+

benchmarking/benchmarks/__init__.py

@@ -0,0 +1,4 @@
+# Licensed to Elasticsearch B.V under one or more agreements.
+# Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
+# See the LICENSE file in the project root for more information
+

benchmarking/benchmarks/benchmark_001_ping.py

@@ -1,3 +1,7 @@
+# Licensed to Elasticsearch B.V under one or more agreements.
+# Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
+# See the LICENSE file in the project root for more information
+
 from benchmarking.lib import benchmarks, Operation, RunnerConfig
 
 
@@ -6,10 +10,5 @@ def run_func(_: int, config: RunnerConfig) -> None:
 
 
 benchmarks.register(
-    Operation(
-        action="ping",
-        category="core",
-        run_func=run_func,
-        num_repetitions=10000
-    )
+    Operation(action="ping", category="core", run_func=run_func, num_repetitions=10000)
 )

benchmarking/benchmarks/benchmark_002_info.py

@@ -1,3 +1,7 @@
+# Licensed to Elasticsearch B.V under one or more agreements.
+# Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
+# See the LICENSE file in the project root for more information
+
 from benchmarking.lib import benchmarks, Operation, RunnerConfig
 
 
@@ -6,10 +10,5 @@ def run_func(_: int, config: RunnerConfig) -> None:
 
 
 benchmarks.register(
-    Operation(
-        action="info",
-        category="core",
-        run_func=run_func,
-        num_repetitions=10000
-    )
+    Operation(action="info", category="core", run_func=run_func, num_repetitions=10000)
 )

benchmarking/benchmarks/benchmark_003_get.py

@@ -1,6 +1,6 @@
-# Licensed to Elasticsearch B.V. under one or more agreements.
-# Elasticsearch B.V. licenses this file to you under the Apache 2.0 License.
-# See the LICENSE file in the project root for more information.
+# Licensed to Elasticsearch B.V under one or more agreements.
+# Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
+# See the LICENSE file in the project root for more information
 
 from benchmarking.lib import benchmarks, Operation, RunnerConfig
 
@@ -22,6 +22,6 @@ def run_func(_: int, config: RunnerConfig) -> None:
         category="core",
         run_func=run_func,
         setup_func=setup_func,
-        num_repetitions=10000
+        num_repetitions=10000,
     )
 )

benchmarking/benchmarks/benchmark_004_index.py

@@ -0,0 +1,44 @@
+# Licensed to Elasticsearch B.V under one or more agreements.
+# Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
+# See the LICENSE file in the project root for more information
+
+# Licensed to Elasticsearch B.V. under one or more agreements.
+# Elasticsearch B.V. licenses this file to you under the Apache 2.0 License.
+# See the LICENSE file in the project root for more information.
+
+import random
+import functools
+from benchmarking.lib import benchmarks, Operation, RunnerConfig
+
+
+@functools.lru_cache()
+def small_doc(data_path):
+    with open(data_path / "small/document.json", mode="rb") as f:
+        return f.read()
+
+
+def setup_func(_: int, config: RunnerConfig) -> None:
+    config.target_es.indices.delete(index="test-bench-index", ignore=404)
+    config.target_es.indices.create(index="test-bench-index")
+    config.target_es.cluster.health(wait_for_status="yellow")
+
+
+def run_func(n: int, config: RunnerConfig) -> None:
+    resp = config.target_es.index(
+        index="test-bench-index",
+        id="%d-%d" % (n, random.randint(1, 10000)),
+        body=small_doc(config.data_path),
+    )
+    assert resp["result"] == "created", resp["result"]
+
+
+benchmarks.register(
+    Operation(
+        action="index",
+        category="core",
+        run_func=run_func,
+        setup_func=setup_func,
+        num_warmups=100,
+        num_repetitions=10000,
+    )
+)

benchmarking/benchmarks/benchmark_005_bulk.py

@@ -1,9 +1,16 @@
+# Licensed to Elasticsearch B.V under one or more agreements.
+# Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
+# See the LICENSE file in the project root for more information
+
 from benchmarking.lib import benchmarks, Operation, RunnerConfig
 
 
 def setup_func(_: int, config: RunnerConfig):
     config.target_es.delete(index="test-bench-bulk", ignore=404)
-    config.target_es.create(index="test-bench-bulk", body={"settings": {"number_of_shards": 3, "refresh_interval": "5s"}})
+    config.target_es.create(
+        index="test-bench-bulk",
+        body={"settings": {"number_of_shards": 3, "refresh_interval": "5s"}},
+    )
     config.target_es.cluster.health(wait_for_status="yellow")
 
 

benchmarking/lib.py

@@ -1,10 +1,18 @@
+# Licensed to Elasticsearch B.V under one or more agreements.
+# Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
+# See the LICENSE file in the project root for more information
+
 import attr
 import sys
 import time
 from pathlib import Path
 from datetime import datetime
 from typing import Callable, Any, Optional, List, Dict
-from elasticsearch import Elasticsearch, TransportError, ElasticsearchException
+from elasticsearch import (
+    Elasticsearch,
+    TransportError,
+    ElasticsearchException,
+)
 from elasticsearch.helpers import bulk
 
 
@@ -13,6 +21,7 @@ class Service:
     type: str = attr.ib()
     name: str = attr.ib()
     version: str = attr.ib()
+    os_family: str = attr.ib()
     git_branch: str = attr.ib()
     git_commit: str = attr.ib()
 
@@ -24,8 +33,8 @@ class RunnerConfig:
     build_id: str = attr.ib()
     category: str = attr.ib()
     environment: str = attr.ib()
-    os_family: str = attr.ib()
-    service: Service = attr.ib()
+    client: Service = attr.ib()
+    target_service: Service = attr.ib()
     data_path: Path = attr.ib()
 
 
@@ -62,28 +71,45 @@ def to_json(self, config: RunnerConfig) -> Dict[str, Any]:
             "tags": ["bench", "elasticsearch-py"],
             "event": {
                 "action": self.action,
-                "duration": self.duration
+                "dataset": str(config.data_path),
+                "duration": self.duration,
             },
+            "http": {"response": {"status_code": self.status_code,}},
             "benchmark": {
                 "build_id": config.build_id,
                 "environment": config.environment,
-                "category": config.category,
+                "category": "core",
                 "repetitions": self.num_repetitions,
                 "operations": self.num_operations,
                 "runner": {
+                    "os": {"family": config.client.os_family},
                     "service": {
-                        "type": "client",
-                        "name": "elasticsearch-py",
-                        "version": config.service.version,
-                        "git_branch": config.service.git_branch,
-                        "git_commit": config.service.git_commit,
+                        "type": config.client.type,
+                        "name": config.client.name,
+                        "version": config.client.version,
+                        "git": {
+                            "branch": config.client.git_branch,
+                            "commit": config.client.git_commit,
+                        },
                     },
                     "runtime": {
                         "name": "python",
-                        "version": ".".join(str(x) for x in sys.version_info[:3])
-                    }
-                }
-            }
+                        "version": ".".join(str(x) for x in sys.version_info[:3]),
+                    },
+                },
+                "target": {
+                    "os": {"family": config.target_service.os_family},
+                    "service": {
+                        "type": config.target_service.type,
+                        "name": config.target_service.name,
+                        "version": config.target_service.version,
+                        "git": {
+                            "branch": config.target_service.git_branch,
+                            "commit": config.target_service.git_commit,
+                        },
+                    },
+                },
+            },
         }
 
 
@@ -95,8 +121,8 @@ def run(self, operation: Operation):
         if operation.setup_func:
             operation.setup_func(0, self.config)
 
-        for _ in range(operation.num_warmups):
-            operation.run_func(0, self.config)
+        for i in range(operation.num_warmups):
+            operation.run_func(i, self.config)
 
         stats_to_publish = []
         for i in range(operation.num_repetitions):
@@ -114,17 +140,20 @@ def run(self, operation: Operation):
             except ElasticsearchException:
                 pass
             finally:
-                duration = time.time() - start_time
-                stats_to_publish.append(Stats(
-                    action=operation.action,
-                    start_time=start_time,
-                    duration=duration,
-                    status_code=status_code,
-                    outcome=outcome,
-                    num_warmups=operation.num_warmups,
-                    num_repetitions=operation.num_repetitions,
-                    num_operations=operation.num_operations,
-                ))
+                # duration is in nanoseconds
+                duration = int((time.time() - start_time) * (10 ** 9))
+                stats_to_publish.append(
+                    Stats(
+                        action=operation.action,
+                        start_time=start_time,
+                        duration=duration,
+                        status_code=status_code,
+                        outcome=outcome,
+                        num_warmups=operation.num_warmups,
+                        num_repetitions=operation.num_repetitions,
+                        num_operations=operation.num_operations,
+                    )
+                )
 
         self.publish_stats(stats_to_publish)
 
@@ -134,14 +163,12 @@ def publish_stats(self, stats: List[Stats]) -> None:
 
         def stream_actions():
             for stat in stats:
-                yield {
-                    "_index": index_name,
-                    "doc": stat.to_json(self.config)
-                }
+                doc = stat.to_json(self.config)
+                doc["_index"] = index_name
+                yield doc
 
         bulk(
-            client=self.config.report_es,
-            actions=stream_actions(),
+            client=self.config.report_es, actions=stream_actions(),
         )
 
 

benchmarking/main.py

@@ -1,10 +1,16 @@
+# Licensed to Elasticsearch B.V under one or more agreements.
+# Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
+# See the LICENSE file in the project root for more information
+
 import sys
 from pathlib import Path
+import platform
+
 sys.path.append(str(Path(__file__).absolute().parent.parent))
 
 import logging
 import os
-from elasticsearch import Elasticsearch
+from elasticsearch import Elasticsearch, __versionstr__
 from benchmarking.lib import benchmarks, Runner, RunnerConfig, Service
 
 
@@ -46,15 +52,23 @@ def main():
         build_id=config["BUILD_ID"],
         category=os.getenv("CLIENT_BENCHMARK_CATEGORY", ""),
         environment=config["CLIENT_BENCHMARK_ENVIRONMENT"],
-        os_family=config["TARGET_SERVICE_OS_FAMILY"],
-        service=Service(
+        client=Service(
+            type="client",
+            name="elasticsearch-py",
+            version=__versionstr__,
+            os_family=platform.system().lower(),
+            git_branch=config["CLIENT_BRANCH"],
+            git_commit=config["CLIENT_COMMIT"],
+        ),
+        target_service=Service(
             type=config["TARGET_SERVICE_TYPE"],
             name=config["TARGET_SERVICE_NAME"],
             version=config["TARGET_SERVICE_VERSION"],
+            os_family=config["TARGET_SERVICE_OS_FAMILY"],
             git_branch=config.get("TARGET_SERVICE_GIT_BRANCH"),
-            git_commit=config.get("TARGET_SERVICE_GIT_COMMIT")
+            git_commit=config.get("TARGET_SERVICE_GIT_COMMIT"),
         ),
-        data_path=Path(config["DATA_SOURCE"]).absolute()
+        data_path=Path(config["DATA_SOURCE"]).absolute(),
     )
     runner = Runner(runner_config)