From 79027c33d3969c79a357d09cc08034c8aaf626e5 Mon Sep 17 00:00:00 2001 From: Fernando Moraes Silveira Date: Thu, 17 Sep 2026 14:46:46 +0200 Subject: [PATCH] FIX: added op_type create and mandatory @timestamp to Elastic metrics --- .github/workflows/tests.yml | 2 +- requirements.txt | 2 +- tests/test_elasticsearch.py | 7 ++++++- tests/test_hooks.py | 2 +- time_execution/backends/elasticsearch.py | 24 +++++++++++++++++------- 5 files changed, 26 insertions(+), 11 deletions(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 965d5c7..5cf4a4c 100644 --- a/.github/workflows/tests.yml +++ b/.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 diff --git a/requirements.txt b/requirements.txt index 33072ee..4394738 100644 --- a/requirements.txt +++ b/requirements.txt @@ -11,7 +11,7 @@ mock isort black flake8 -mypy +mypy<2.0.0 # Distribution setuptools; python_version >= "3.12" diff --git a/tests/test_elasticsearch.py b/tests/test_elasticsearch.py index b02003f..f9a57e8 100644 --- a/tests/test_elasticsearch.py +++ b/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) diff --git a/tests/test_hooks.py b/tests/test_hooks.py index 7504e7d..50b9dd9 100644 --- a/tests/test_hooks.py +++ b/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"} diff --git a/time_execution/backends/elasticsearch.py b/time_execution/backends/elasticsearch.py index 2632cd0..6dfe4dc 100644 --- a/time_execution/backends/elasticsearch.py +++ b/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)