diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index d96ed24..0bf2649 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -2,7 +2,7 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): import traceback - from datetime import UTC + from datetime import UTC, datetime from logging import Logger from typing import Any @@ -452,7 +452,8 @@ class MongoDB(SientiaMonitoring): Required fields: - metadata (dict[str, Any]): Workflow execution metadata - collection_name (str): Name of the MongoDB collection - - last_data_timestamp (str | None): Last processed timestamp for filtering + - last_data_timestamp (str | None): Last processed timestamp + (ISO-format string) for filtering - base_data_filter (dict[str, Any]): Base query filter conditions Returns: @@ -472,19 +473,14 @@ class MongoDB(SientiaMonitoring): if last_data_timestamp is None: data_filter = base_data_filter else: - # ``notification_queue.timestamp`` is stored as a string in - # ``DATETIME_FORMAT_WITH_TZ`` (``Notification`` writes it as - # ``now().strftime(DATETIME_FORMAT_WITH_TZ)``). Coercing - # ``last_data_timestamp`` to ``datetime`` here would force a - # BSON ``String`` vs ``Date`` comparison, which always yields - # ``False`` (``String < Date`` in BSON sort order) and breaks - # incremental loading entirely. Comparing strings preserves the - # intended chronological filter because the format is - # lexicographically ordered when the timezone is fixed - # (``Notification.timestamp`` always uses UTC). + # ``notification_queue.timestamp`` is stored as a native BSON + # ``Date`` (``Notification.timestamp`` is a ``datetime``). + # ``last_data_timestamp`` arrives here as an ISO-format string + # (round-tripped through Redis), so it must be parsed back to + # ``datetime`` for the ``$gt`` comparison to be type-correct. data_filter = { **base_data_filter, - 'timestamp': {'$gt': last_data_timestamp}, + 'timestamp': {'$gt': datetime.fromisoformat(last_data_timestamp)}, } self.debug(f'Data filter: {data_filter}', metadata=metadata) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index a0a2365..6a6192b 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -335,6 +335,11 @@ class SlotManager(SientiaMonitoring): self.debug(f'Last collected timestamp to insert: {last_data_timestamp}', metadata=metadata) + # ``timestamp`` is a native ``datetime``/``Timestamp`` (BSON Date from + # Mongo), which the Redis repository's plain ``json.dumps`` cannot + # serialize. Store it as an ISO-format string instead. + last_data_timestamp = last_data_timestamp.isoformat() + try: self.redis_repository.set(key, last_data_timestamp, ttl=60 * 60 * 5) except Exception as e: diff --git a/requirements-local.txt b/requirements-local.txt index 202ddd7..8df6bac 100644 --- a/requirements-local.txt +++ b/requirements-local.txt @@ -4,5 +4,5 @@ sqlalchemy redis pymongo jinja2 -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.1 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2 prometheus-client diff --git a/requirements.txt b/requirements.txt index 0de836d..680f120 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,5 +4,5 @@ sqlalchemy redis pymongo jinja2 -sientia_do>=1.12.1 +sientia_do>=1.12.2 prometheus-client diff --git a/tests/orchestrator/activities/test_mongo_db.py b/tests/orchestrator/activities/test_mongo_db.py index 1745454..4083ce9 100644 --- a/tests/orchestrator/activities/test_mongo_db.py +++ b/tests/orchestrator/activities/test_mongo_db.py @@ -1,4 +1,4 @@ -from datetime import datetime +from datetime import UTC, datetime from unittest.mock import ANY, AsyncMock, MagicMock, patch from pytest import fixture @@ -462,7 +462,7 @@ def test_load_latest_data_none_last_data_timestamp(mongo_db): { 'name': 'test1', 'value': 1, - 'timestamp': '2023-01-01 12:00:00+0000', + 'timestamp': datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC), } ] ) @@ -482,18 +482,20 @@ def test_load_latest_data_none_last_data_timestamp(mongo_db): {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, ) - assert result == [{'name': 'test1', 'value': 1, 'timestamp': '2023-01-01 12:00:00+0000'}] + assert result == [ + {'name': 'test1', 'value': 1, 'timestamp': datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)} + ] def test_load_latest_data_not_none_last_data_timestamp(mongo_db): - """Test load_latest_data""" + """Test load_latest_data compares native datetime against native datetime in $gt""" mongo_db.mongo_db_repository.find = MagicMock( return_value=[ { 'name': 'test1', 'value': 1, - 'timestamp': '2023-01-01 12:00:00+0000', + 'timestamp': datetime(2023, 1, 1, 12, 0, 1, tzinfo=UTC), } ] ) @@ -502,7 +504,7 @@ def test_load_latest_data_not_none_last_data_timestamp(mongo_db): { 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, 'collection_name': 'test_collection', - 'last_data_timestamp': '2023-01-01 12:00:00+0000', + 'last_data_timestamp': '2023-01-01T12:00:00+00:00', 'base_data_filter': {'level': 'ERROR'}, } ) @@ -511,12 +513,56 @@ def test_load_latest_data_not_none_last_data_timestamp(mongo_db): 'test_collection', { 'level': 'ERROR', - 'timestamp': {'$gt': '2023-01-01 12:00:00+0000'}, + 'timestamp': {'$gt': datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)}, }, {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, ) - assert result == [{'name': 'test1', 'value': 1, 'timestamp': '2023-01-01 12:00:00+0000'}] + assert result == [ + {'name': 'test1', 'value': 1, 'timestamp': datetime(2023, 1, 1, 12, 0, 1, tzinfo=UTC)} + ] + + +def test_load_latest_data_run_boundary_no_skip_or_duplicate(mongo_db): + """Regression: the $gt filter built from the previous run's timestamp must not + skip the document that landed exactly on the boundary, nor re-return it.""" + + boundary_timestamp = datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC) + + def fake_find(collection_name, data_filter, metadata): + gt = data_filter.get('timestamp', {}).get('$gt') + all_docs = [ + {'name': 'before', 'value': 1, 'timestamp': datetime(2023, 1, 1, 11, 59, 59, tzinfo=UTC)}, + {'name': 'boundary', 'value': 2, 'timestamp': boundary_timestamp}, + {'name': 'after', 'value': 3, 'timestamp': datetime(2023, 1, 1, 12, 0, 1, tzinfo=UTC)}, + ] + if gt is None: + return all_docs + return [doc for doc in all_docs if doc['timestamp'] > gt] + + mongo_db.mongo_db_repository.find = MagicMock(side_effect=fake_find) + + first_run = mongo_db.load_latest_data( + { + 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, + 'collection_name': 'test_collection', + 'last_data_timestamp': None, + 'base_data_filter': {'level': 'ERROR'}, + } + ) + + assert [doc['name'] for doc in first_run] == ['before', 'boundary', 'after'] + + second_run = mongo_db.load_latest_data( + { + 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, + 'collection_name': 'test_collection', + 'last_data_timestamp': boundary_timestamp.isoformat(), + 'base_data_filter': {'level': 'ERROR'}, + } + ) + + assert [doc['name'] for doc in second_run] == ['after'] def test_load_latest_data_error(mongo_db): @@ -528,7 +574,7 @@ def test_load_latest_data_error(mongo_db): { 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, 'collection_name': 'test_collection', - 'last_data_timestamp': '2023-01-01 12:00:00+0000', + 'last_data_timestamp': '2023-01-01T12:00:00+00:00', 'base_data_filter': {'level': 'ERROR'}, } ) diff --git a/tests/orchestrator/activities/test_slot_manager.py b/tests/orchestrator/activities/test_slot_manager.py index 14e3747..b16c41f 100644 --- a/tests/orchestrator/activities/test_slot_manager.py +++ b/tests/orchestrator/activities/test_slot_manager.py @@ -1,4 +1,4 @@ -from datetime import timedelta +from datetime import UTC, datetime, timedelta from unittest.mock import ANY, AsyncMock, MagicMock, call, patch from pandas import DataFrame @@ -279,7 +279,10 @@ def test_put_last_data_timestamp_not_empty_dataframe(slot_manager): { 'name': ['sensor1', 'sensor2'], 'value': [25.5, 30.0], - 'timestamp': ['2023-01-01 12:00:00', '2023-01-01 12:00:01'], + 'timestamp': [ + datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC), + datetime(2023, 1, 1, 12, 0, 1, tzinfo=UTC), + ], } ) test_data = { @@ -294,10 +297,10 @@ def test_put_last_data_timestamp_not_empty_dataframe(slot_manager): result = slot_manager.put_last_data_timestamp(test_data) - assert result == '2023-01-01 12:00:01' + assert result == '2023-01-01T12:00:01+00:00' slot_manager.redis_repository.set.assert_called_once_with( - 'notification_last_timestamp:test_mail_type', '2023-01-01 12:00:01', ttl=18000 + 'notification_last_timestamp:test_mail_type', '2023-01-01T12:00:01+00:00', ttl=18000 ) @@ -311,7 +314,7 @@ def test_put_last_data_timestamp_error(slot_manager): { 'name': ['sensor1', 'sensor2'], 'value': [25.5, 30.0], - 'timestamp': ['2023-01-01 12:00:00'] * 2, + 'timestamp': [datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)] * 2, } ).to_dict('records'), 'mail_type': 'test_mail_type',