FIX: added op_type create and mandatory @timestamp to Elastic metrics

.github/workflows/tests.yml

@@ -22,7 +22,7 @@ jobs:
       with:
         python-version: ${{ matrix.python-version }}
     - name: Install Tox and other dependencies
-      run: pip install tox tox-docker wheel twine 'setuptools; python_version >= "3.12"'
+      run: pip install tox tox-docker wheel twine 'setuptools; python_version >= "3.12"' 'vcs-versioning; python_version >= "3.12"'
     - name: Run tests with Tox
       # Run tox using the version of Python in `PATH`
       run: tox -e py

requirements.txt

@@ -11,7 +11,7 @@ mock
 isort
 black
 flake8
-mypy
+mypy<2.0.0
 
 # Distribution
 setuptools; python_version >= "3.12"

tests/test_elasticsearch.py

@@ -129,6 +129,7 @@ def test_error_warning(self, mocked_logger):
                 "writing metric %r failure %r",
                 {
                     "timestamp": datetime(2016, 7, 13),
+                    "@timestamp": datetime(2016, 7, 13),
                     "value": None,
                     "name": "test:metric",
                 },
@@ -159,7 +160,11 @@ def test_bulk_write_error(self, mocked_logger):
             "time_execution.backends.elasticsearch.Elasticsearch.bulk",
             side_effect=transport_error,
         )
-        metrics = [1, 2, 3]
+        metrics = [
+            {"name": "metric.name", "value": 1, "timestamp": 1},
+            {"name": "metric.name", "value": 2, "timestamp": 2},
+            {"name": "metric.name", "value": 3, "timestamp": 3},
+        ]
         with es_index_error_ctx:
             self.backend.bulk_write(metrics)
             mocked_logger.warning.assert_called_once_with("bulk_write metrics %r failure %r", metrics, transport_error)

tests/test_hooks.py

@@ -269,7 +269,7 @@ def generator_hook(func, func_args, func_kwargs) -> GeneratorHookReturnType:
             assert func_args == (42,)
             assert func_kwargs == {"bar": 100500}
             assert not is_started, "the decorated function should not run just yet"
-            (response, _exception, _metrics) = yield
+            response, _exception, _metrics = yield
             assert is_started
             assert response == "response"
             return {"key": "value"}

time_execution/backends/elasticsearch.py

@@ -1,8 +1,8 @@
 import logging
-from datetime import datetime
+from datetime import datetime, timezone
 
 from elasticsearch import Elasticsearch
-from elasticsearch.exceptions import TransportError
+from elasticsearch.exceptions import BadRequestError, TransportError
 
 from time_execution.backends.base import BaseMetricsBackend
 
@@ -29,7 +29,7 @@ def __init__(
         self.client = Elasticsearch(hosts=hosts, *args, **kwargs)
 
     def get_index(self):
-        return self.index_pattern.format(index=self.index, date=datetime.now())
+        return self.index_pattern.format(index=self.index, date=datetime.now(timezone.utc).replace(tzinfo=None))
 
     def write(self, name, **data):
         """
@@ -41,20 +41,24 @@ def write(self, name, **data):
         """
 
         data["name"] = name
+        now = datetime.now(timezone.utc).replace(tzinfo=None)
         if not ("timestamp" in data):
-            data["timestamp"] = datetime.utcnow()
+            data["timestamp"] = now
+        if not ("@timestamp" in data):
+            data["@timestamp"] = now
 
         try:
             index_params = {
                 "index": self.get_index(),
                 "id": None,
                 "body": data,
+                "op_type": "create",
             }
             if self.pipeline:
                 index_params["pipeline"] = self.pipeline
 
             self.client.index(**index_params)
-        except TransportError as exc:
+        except (BadRequestError, TransportError) as exc:
             logger.warning("writing metric %r failure %r", data, exc)
 
     def bulk_write(self, metrics):
@@ -66,8 +70,14 @@ def bulk_write(self, metrics):
         """
         actions = []
         index = self.get_index()
+        now = datetime.now(timezone.utc).replace(tzinfo=None)
         for metric in metrics:
-            actions.append({"index": {"_index": index}})
+            if not ("timestamp" in metric):
+                metric["timestamp"] = now
+            if not ("@timestamp" in metric):
+                metric["@timestamp"] = now
+
+            actions.append({"create": {"_index": index}})
             actions.append(metric)
 
         bulk_params = {"operations": actions}
@@ -76,5 +86,5 @@ def bulk_write(self, metrics):
 
         try:
             self.client.bulk(**bulk_params)
-        except TransportError as exc:
+        except (BadRequestError, TransportError) as exc:
             logger.warning("bulk_write metrics %r failure %r", metrics, exc)