From cb3caaa55cc0e7c2d65878ab3e6d438b6939f1d1 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 22 Aug 2025 08:11:19 -0300 Subject: [PATCH 01/10] SIENTIAPDE-1193 chore: update sientia-dataops-library version and modify environment variables - Updated the sientia-dataops-library dependency version from 1.4.1 to 1.4.2 in requirements.txt. - Changed the GITHUB_BRANCH value in values.yaml to reflect the new task SIENTIAPDE-1193. - Enhanced the process_notifications workflow to include timestamp conversion metadata for improved data handling. --- .../workflows/subworkflows/process_notifications.py | 7 ++++++- requirements.txt | 2 +- values.yaml | 2 +- 3 files changed, 8 insertions(+), 3 deletions(-) diff --git a/orchestrator/workflows/subworkflows/process_notifications.py b/orchestrator/workflows/subworkflows/process_notifications.py index b4f6d41..1c2a0bd 100644 --- a/orchestrator/workflows/subworkflows/process_notifications.py +++ b/orchestrator/workflows/subworkflows/process_notifications.py @@ -5,6 +5,7 @@ with workflow.unsafe.imports_passed_through(): from typing import Any from datetime import timedelta from sientia_do.temporal.policies import retry_policy + from sientia_do.temporal.constants import DATETIME_FORMAT @workflow.defn(name="process_notifications") @@ -80,7 +81,11 @@ class ProcessNotifications: **metadata, "schema": input_data["schema"], "table_name": input_data["table_name"], - "data": log_report + "data": log_report, + 'timestamp_conversion': { + 'column': 'timestamp', + 'format': DATETIME_FORMAT + } }, schedule_to_close_timeout=timedelta(seconds=60), retry_policy=retry_policy diff --git a/requirements.txt b/requirements.txt index 50fa9aa..5d8b5e9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,5 +5,5 @@ redis couchbase pymongo jinja2 -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.1 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.2 prometheus-client diff --git a/values.yaml b/values.yaml index 3396de3..c2c69e9 100644 --- a/values.yaml +++ b/values.yaml @@ -151,7 +151,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-1199-revisar-e-testar-observabilidade" + value: "SIENTIAPDE-1193-conferir-como-a-escrita-de-datetime-ocorre-no-temporal" - name: PYTHON_APP value: "orchestrator.worker.worker" From 9672a9cb7e2e4a0806418af8deb48ea3b538605b Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 22 Aug 2025 11:45:11 -0300 Subject: [PATCH 02/10] SIENTIAPDE-1193 refactor: update date handling and remove deprecated constants - Replaced the usage of datetime.now() with a new now() function from sientia_do.temporal.constants for consistent timestamp handling across multiple files. - Updated the date format from DEFAULT_DATE_FORMAT to DATETIME_FORMAT_MS_WITH_TZ in various activities to ensure uniformity in timestamp formatting. - Removed the obsolete patterns.py file that contained the DEFAULT_DATE_FORMAT constant. --- orchestrator/activities/formatters.py | 11 +++++------ orchestrator/activities/mongo_db.py | 12 ++++++------ orchestrator/activities/slot_manager.py | 8 ++++---- orchestrator/utils/patterns.py | 1 - .../workflows/subworkflows/process_notifications.py | 4 ++-- requirements.txt | 2 +- 6 files changed, 18 insertions(+), 20 deletions(-) delete mode 100644 orchestrator/utils/patterns.py diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 34732be..f3212ab 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -9,14 +9,13 @@ with workflow.unsafe.imports_passed_through(): import json from typing import Any from logging import Logger - from datetime import datetime from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.temporal.activities.base import BaseActivity from sientia_do.notifications.models import NotificationLevel from orchestrator.utils.orchestrator_functions import ( scouter, predictions_batch, gather_read_tags, build_tag_config ) - from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT + from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now from math import ceil topic_separator = "\n ========== \n" @@ -63,21 +62,21 @@ class Formatters(BaseActivity): schedule_config[self.scouter_namespace][pipeline['schedule_name']] = { **scouter(pipeline), "updated_at": pipeline.get( - "updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT)) + "updated_at", now().strftime(DATETIME_FORMAT_MS_WITH_TZ)) } elif pipeline['workflow_type'] == 'predictions_batch': schedule_config[self.laborious_namespace][pipeline['schedule_name'] ] = { **predictions_batch(pipeline), "updated_at": pipeline.get( - "updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT)) + "updated_at", now().strftime(DATETIME_FORMAT_MS_WITH_TZ)) } elif pipeline['workflow_type'] == 'minimal_retrain': schedule_config[self.laborious_namespace][pipeline['schedule_name'] ] = { **minimal_retrain(pipeline), "updated_at": pipeline.get( - "updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT)) + "updated_at", now().strftime(DATETIME_FORMAT_MS_WITH_TZ)) } self.info("Processed schedules", metadata=metadata) @@ -219,7 +218,7 @@ class Formatters(BaseActivity): if schedule_name in current_schedules: update_timestamp = schedule.get( - 'updated_at', datetime.now()) + 'updated_at', now()) old_timestamp = current_schedules[schedule_name] diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index e7a3ac1..3e094b4 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -2,7 +2,6 @@ from temporalio import workflow, activity with workflow.unsafe.imports_passed_through(): - from datetime import datetime from typing import Any import traceback from logging import Logger @@ -10,7 +9,8 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.temporal.activities.base import BaseActivity - from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT + from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now + from datetime import datetime def clear_mongo_id(docs: list) -> list: @@ -216,7 +216,7 @@ class MongoDB(BaseActivity): """ updated_pipelines = input_data.get("updated_pipelines", []) metadata = input_data.get("metadata", {}) - now = datetime.now() + now = now() collection = self.database["orchestrated_schedules"] self.info("Updating pipelines timestamps...", metadata=metadata) @@ -267,7 +267,7 @@ class MongoDB(BaseActivity): success_count = 0 - now = datetime.now() + now = now() argument = [ {"schedule_name": pipeline["schedule_name"], @@ -442,7 +442,7 @@ class MongoDB(BaseActivity): data_filter = { **base_data_filter, "timestamp": { - "$gt": datetime.strptime(last_data_timestamp, DEFAULT_DATE_FORMAT) + "$gt": datetime.strptime(last_data_timestamp, DATETIME_FORMAT_MS_WITH_TZ) } } @@ -460,7 +460,7 @@ class MongoDB(BaseActivity): for item in data: item['timestamp'] = item['timestamp'].strftime( - DEFAULT_DATE_FORMAT) + DATETIME_FORMAT_MS_WITH_TZ) self.info( f"Loaded {len(data)} documents from MongoDB", diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index a7dab02..bc68d23 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -9,7 +9,7 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.temporal.activities.redis_base import Redis from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel - from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT + from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now from pandas import DataFrame from datetime import datetime, timedelta @@ -344,10 +344,10 @@ class SlotManager(Redis): else: last_sent = datetime.strptime( - last_sent, DEFAULT_DATE_FORMAT) + last_sent, DATETIME_FORMAT_MS_WITH_TZ) # Check if "notification_ttl" seconds has passed since last sent - if (datetime.now() - last_sent) > timedelta(seconds=notification_ttl): + if (now() - last_sent) > timedelta(seconds=notification_ttl): alert_type = "persistent_alerts" # Check if this group must be notified @@ -382,7 +382,7 @@ class SlotManager(Redis): self.info("Storing notification cache...", metadata=metadata) - now = datetime.now().strftime(DEFAULT_DATE_FORMAT) + now = now().strftime(DATETIME_FORMAT_MS_WITH_TZ) for index, row in log_report.iterrows(): status = row['status'] diff --git a/orchestrator/utils/patterns.py b/orchestrator/utils/patterns.py deleted file mode 100644 index fe00beb..0000000 --- a/orchestrator/utils/patterns.py +++ /dev/null @@ -1 +0,0 @@ -DEFAULT_DATE_FORMAT = '%Y-%m-%d %H:%M:%S.%f' diff --git a/orchestrator/workflows/subworkflows/process_notifications.py b/orchestrator/workflows/subworkflows/process_notifications.py index 1c2a0bd..e02fc12 100644 --- a/orchestrator/workflows/subworkflows/process_notifications.py +++ b/orchestrator/workflows/subworkflows/process_notifications.py @@ -5,7 +5,7 @@ with workflow.unsafe.imports_passed_through(): from typing import Any from datetime import timedelta from sientia_do.temporal.policies import retry_policy - from sientia_do.temporal.constants import DATETIME_FORMAT + from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ @workflow.defn(name="process_notifications") @@ -84,7 +84,7 @@ class ProcessNotifications: "data": log_report, 'timestamp_conversion': { 'column': 'timestamp', - 'format': DATETIME_FORMAT + 'format': DATETIME_FORMAT_MS_WITH_TZ } }, schedule_to_close_timeout=timedelta(seconds=60), diff --git a/requirements.txt b/requirements.txt index 5d8b5e9..f27fcce 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,5 +5,5 @@ redis couchbase pymongo jinja2 -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.2 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.3 prometheus-client From 5692fa69e1f4f81a3d74e91585710cbbc7bc661e Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 25 Aug 2025 08:30:42 -0300 Subject: [PATCH 03/10] SIENTIAPDE-1193 refactor: standardize timestamp handling across MongoDB and SlotManager - Replaced instances of `now()` with `date_now` for consistent timestamp usage in MongoDB and SlotManager activities. - Updated test cases to mock the new `now()` function, ensuring accurate timestamp handling in unit tests. - Adjusted timestamp formatting in various tests to align with the DATETIME_FORMAT_MS_WITH_TZ standard. --- orchestrator/activities/mongo_db.py | 8 ++-- orchestrator/activities/slot_manager.py | 4 +- .../orchestrator/activities/test_mongo_db.py | 40 +++++++++---------- .../activities/test_slot_manager.py | 14 +++---- .../test_process_notifications.py | 7 +++- 5 files changed, 39 insertions(+), 34 deletions(-) diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 3e094b4..e53f51a 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -216,7 +216,7 @@ class MongoDB(BaseActivity): """ updated_pipelines = input_data.get("updated_pipelines", []) metadata = input_data.get("metadata", {}) - now = now() + date_now = now() collection = self.database["orchestrated_schedules"] self.info("Updating pipelines timestamps...", metadata=metadata) @@ -233,7 +233,7 @@ class MongoDB(BaseActivity): try: collection.update_many( data_filter, - {"$set": {"updated_at": now}} + {"$set": {"updated_at": date_now}} ) success_count += 1 except Exception as e: @@ -267,12 +267,12 @@ class MongoDB(BaseActivity): success_count = 0 - now = now() + date_now = now() argument = [ {"schedule_name": pipeline["schedule_name"], "namespace": pipeline["namespace"], - "updated_at": now} + "updated_at": date_now} for pipeline in created_pipelines if pipeline["success"] ] data_filter = argument if argument else {} diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index bc68d23..862e258 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -382,12 +382,12 @@ class SlotManager(Redis): self.info("Storing notification cache...", metadata=metadata) - now = now().strftime(DATETIME_FORMAT_MS_WITH_TZ) + date_now = now().strftime(DATETIME_FORMAT_MS_WITH_TZ) for index, row in log_report.iterrows(): status = row['status'] if status == 'sent': key = f"{row['schedule']}:{row['notification_id']}" - self.set(key, now, ttl=sent_ttl) + self.set(key, date_now, ttl=sent_ttl) self.info("Notification cache stored...", metadata=metadata) diff --git a/tests/orchestrator/activities/test_mongo_db.py b/tests/orchestrator/activities/test_mongo_db.py index 742d42f..2ae912d 100644 --- a/tests/orchestrator/activities/test_mongo_db.py +++ b/tests/orchestrator/activities/test_mongo_db.py @@ -266,8 +266,8 @@ async def test_aggregate_documents_in_mongodb_missing_aggregation(mongo_db): @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_update_pipelines_timestamps_success(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_update_pipelines_timestamps_success(now_mock, mongo_db): input_data = {"updated_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test2", "namespace": "test2", "success": True} @@ -280,13 +280,13 @@ async def test_update_pipelines_timestamps_success(datetime_mock, mongo_db): {"schedule_name": "test2", "namespace": "test2"} ]}, {"$set": { - "updated_at": datetime_mock.now.return_value}} + "updated_at": now_mock.return_value}} ) @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_update_pipelines_timestamps_failure(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_update_pipelines_timestamps_failure(now_mock, mongo_db): input_data = { "updated_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "success": True}, @@ -314,8 +314,8 @@ async def test_update_pipelines_timestamps_failure(datetime_mock, mongo_db): @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_create_pipelines_timestamps_success(now_mock, mongo_db): input_data = {"created_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test2", "namespace": "test2", "success": True} @@ -325,16 +325,16 @@ async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db): mongo_db.database["pipelines"].insert_many.assert_called_once_with( [ {"schedule_name": "test1", "namespace": "test1", - "updated_at": datetime_mock.now.return_value}, + "updated_at": now_mock.return_value}, {"schedule_name": "test2", "namespace": "test2", - "updated_at": datetime_mock.now.return_value} + "updated_at": now_mock.return_value} ] ) @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_create_pipelines_timestamps_failure(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_create_pipelines_timestamps_failure(now_mock, mongo_db): input_data = {"created_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test2", "namespace": "test2", "success": True} @@ -360,8 +360,8 @@ async def test_create_pipelines_timestamps_failure(datetime_mock, mongo_db): @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_delete_pipelines_timestamps_success(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_delete_pipelines_timestamps_success(now_mock, mongo_db): input_data = {"deleted_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test2", "namespace": "test2", "success": True} @@ -377,8 +377,8 @@ async def test_delete_pipelines_timestamps_success(datetime_mock, mongo_db): @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_delete_pipelines_timestamps_failure(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_delete_pipelines_timestamps_failure(now_mock, mongo_db): input_data = {"deleted_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test2", "namespace": "test2", "success": True} @@ -573,14 +573,14 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongo_db): 'name': 'test1', 'value': 1, 'timestamp': datetime.strptime( - '2023-01-01 12:00:00.000000', '%Y-%m-%d %H:%M:%S.%f') + '2023-01-01 12:00:00.000000+0000', '%Y-%m-%d %H:%M:%S.%f%z') } ] result = await mongo_db.load_latest_data({ 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, 'collection_name': 'test_collection', - 'last_data_timestamp': '2023-01-01 12:00:00.000000', + 'last_data_timestamp': '2023-01-01 12:00:00.000000+0000', 'base_data_filter': { 'level': 'ERROR' } @@ -594,7 +594,7 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongo_db): 'level': 'ERROR', 'timestamp': { '$gt': datetime.strptime( - '2023-01-01 12:00:00.000000', '%Y-%m-%d %H:%M:%S.%f') + '2023-01-01 12:00:00.000000+0000', '%Y-%m-%d %H:%M:%S.%f%z') } }, {"_id": 0} @@ -603,7 +603,7 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongo_db): assert result == [{ 'name': 'test1', 'value': 1, - 'timestamp': '2023-01-01 12:00:00.000000' + 'timestamp': '2023-01-01 12:00:00.000000+0000' }] @@ -620,7 +620,7 @@ async def test_load_latest_data_error(mongo_db): await mongo_db.load_latest_data({ 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, 'collection_name': 'test_collection', - 'last_data_timestamp': '2023-01-01 12:00:00.000000', + 'last_data_timestamp': '2023-01-01 12:00:00.000000+0000', 'base_data_filter': { 'level': 'ERROR' } diff --git a/tests/orchestrator/activities/test_slot_manager.py b/tests/orchestrator/activities/test_slot_manager.py index debf0a7..80cd384 100644 --- a/tests/orchestrator/activities/test_slot_manager.py +++ b/tests/orchestrator/activities/test_slot_manager.py @@ -4,7 +4,7 @@ from pandas import DataFrame from pytest import mark, fixture from orchestrator.activities.slot_manager import SlotManager from sientia_do.notifications.models import NotificationLevel -from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT +from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now metadata = { "metadata": { @@ -373,13 +373,13 @@ async def test_put_last_data_timestamp_error(slot_manager): async def test_filter_notification_alerts(slot_manager): slot_manager.get = MagicMock(side_effect=[ None, - (datetime.now() - timedelta(seconds=600) - ).strftime(DEFAULT_DATE_FORMAT), - datetime.now().strftime(DEFAULT_DATE_FORMAT), + (now() - timedelta(seconds=600) + ).strftime(DATETIME_FORMAT_MS_WITH_TZ), + now().strftime(DATETIME_FORMAT_MS_WITH_TZ), None, - (datetime.now() - timedelta(seconds=600) - ).strftime(DEFAULT_DATE_FORMAT), - datetime.now().strftime(DEFAULT_DATE_FORMAT)]) + (now() - timedelta(seconds=600) + ).strftime(DATETIME_FORMAT_MS_WITH_TZ), + now().strftime(DATETIME_FORMAT_MS_WITH_TZ)]) input_data = { **metadata, diff --git a/tests/orchestrator/workflows/subworkflows/test_process_notifications.py b/tests/orchestrator/workflows/subworkflows/test_process_notifications.py index df25c84..8a41654 100644 --- a/tests/orchestrator/workflows/subworkflows/test_process_notifications.py +++ b/tests/orchestrator/workflows/subworkflows/test_process_notifications.py @@ -2,6 +2,7 @@ from unittest.mock import AsyncMock, patch, ANY, call from pytest import fixture, mark from orchestrator.workflows.subworkflows.process_notifications import ProcessNotifications from orchestrator.activities.activities import Activities +from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ @fixture @@ -76,7 +77,11 @@ async def test_run(workflow_mock, process_notifications): **metadata, 'schema': input_data['schema'], 'table_name': input_data['table_name'], - 'data': workflow_mock.execute_local_activity_method.return_value + 'data': workflow_mock.execute_local_activity_method.return_value, + 'timestamp_conversion': { + 'column': 'timestamp', + 'format': DATETIME_FORMAT_MS_WITH_TZ + } }, schedule_to_close_timeout=ANY, retry_policy=ANY From 92e5ccdcdee772303f6a1676ce5b9476bfa6f6f6 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 25 Aug 2025 08:48:02 -0300 Subject: [PATCH 04/10] SIENTIAPDE-1193 chore: update image tag and enhance datetime column handling in orchestrator functions - Updated the image tag in values.yaml from 0.4.2 to 0.4.4 for the latest version. - Added support for datetime_columns in minimal_retrain and predictions_batch functions to improve timestamp handling in orchestrator functions. --- orchestrator/utils/orchestrator_functions.py | 2 ++ values.yaml | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 38bde51..b49a771 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -48,6 +48,7 @@ def minimal_retrain(config: dict[str, Any]): "query": config['query'], "schema": "sientia_data", "table_name": "log_retrain", + "datetime_columns": config.get('datetime_columns', []), } @@ -187,6 +188,7 @@ def predictions_batch(config: dict[str, Any]): **common_config(config), "query": config['query'], + "datetime_columns": config.get('datetime_columns', []), "schema": "sientia_data", "table_name": "predictions", "retention_time": config.get('model_retention_minutes', 60) * 60, diff --git a/values.yaml b/values.yaml index c2c69e9..300ea80 100644 --- a/values.yaml +++ b/values.yaml @@ -11,7 +11,7 @@ image: # This sets the pull policy for images. pullPolicy: Always # Overrides the image tag whose default is the chart appVersion. - tag: "0.4.2" + tag: "0.4.4" # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ imagePullSecrets: From 33017ffc7ee3c0372084a38a3217b56288af06bc Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 25 Aug 2025 09:16:47 -0300 Subject: [PATCH 05/10] SIENTIAPDE-1193 Enhance test cases to include datetime_columns for improved timestamp handling in minimal_retrain and predictions_batch functions --- coverage.sh | 1 + tests/orchestrator/utils/test_orchestrator_functions.py | 6 +++++- 2 files changed, 6 insertions(+), 1 deletion(-) create mode 100755 coverage.sh diff --git a/coverage.sh b/coverage.sh new file mode 100755 index 0000000..c3235b5 --- /dev/null +++ b/coverage.sh @@ -0,0 +1 @@ +pytest --cov=orchestrator --cov-report=html && xdg-open htmlcov/index.html \ No newline at end of file diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index eb4b056..09fae10 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -41,6 +41,7 @@ def test_minimal_retrain(): "name": "test_model_name" }, "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;", + "datetime_columns": ["timestamp"] } result = minimal_retrain(config) expected = { @@ -53,6 +54,7 @@ def test_minimal_retrain(): "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;", "schema": "sientia_data", "table_name": "log_retrain", + "datetime_columns": ["timestamp"] } assert result == expected @@ -195,6 +197,7 @@ def test_predictions_batch(mock_process_path_priority, } ], "path_priority": ["STOP", "CONTINUE", "REPEAT"], + "datetime_columns": ["timestamp"] } result = predictions_batch(config) @@ -268,7 +271,8 @@ def test_predictions_batch(mock_process_path_priority, "config": {} } }, - "path_priority": ["STOP", "CONTINUE", "REPEAT"] + "path_priority": ["STOP", "CONTINUE", "REPEAT"], + "datetime_columns": ["timestamp"] } assert result == expected From d4efe1d15939339ac74ec4b500b57455f828dbb4 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 22 Aug 2025 08:11:19 -0300 Subject: [PATCH 06/10] SIENTIAPDE-1193 chore: update sientia-dataops-library version and modify environment variables - Updated the sientia-dataops-library dependency version from 1.4.1 to 1.4.2 in requirements.txt. - Changed the GITHUB_BRANCH value in values.yaml to reflect the new task SIENTIAPDE-1193. - Enhanced the process_notifications workflow to include timestamp conversion metadata for improved data handling. --- .../workflows/subworkflows/process_notifications.py | 7 ++++++- requirements.txt | 2 +- values.yaml | 2 +- 3 files changed, 8 insertions(+), 3 deletions(-) diff --git a/orchestrator/workflows/subworkflows/process_notifications.py b/orchestrator/workflows/subworkflows/process_notifications.py index b4f6d41..1c2a0bd 100644 --- a/orchestrator/workflows/subworkflows/process_notifications.py +++ b/orchestrator/workflows/subworkflows/process_notifications.py @@ -5,6 +5,7 @@ with workflow.unsafe.imports_passed_through(): from typing import Any from datetime import timedelta from sientia_do.temporal.policies import retry_policy + from sientia_do.temporal.constants import DATETIME_FORMAT @workflow.defn(name="process_notifications") @@ -80,7 +81,11 @@ class ProcessNotifications: **metadata, "schema": input_data["schema"], "table_name": input_data["table_name"], - "data": log_report + "data": log_report, + 'timestamp_conversion': { + 'column': 'timestamp', + 'format': DATETIME_FORMAT + } }, schedule_to_close_timeout=timedelta(seconds=60), retry_policy=retry_policy diff --git a/requirements.txt b/requirements.txt index 50fa9aa..5d8b5e9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,5 +5,5 @@ redis couchbase pymongo jinja2 -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.1 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.2 prometheus-client diff --git a/values.yaml b/values.yaml index 3396de3..c2c69e9 100644 --- a/values.yaml +++ b/values.yaml @@ -151,7 +151,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-1199-revisar-e-testar-observabilidade" + value: "SIENTIAPDE-1193-conferir-como-a-escrita-de-datetime-ocorre-no-temporal" - name: PYTHON_APP value: "orchestrator.worker.worker" From d7effd44e54efb73f33a2a39a9fe7b24a5d57318 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 22 Aug 2025 11:45:11 -0300 Subject: [PATCH 07/10] SIENTIAPDE-1193 refactor: update date handling and remove deprecated constants - Replaced the usage of datetime.now() with a new now() function from sientia_do.temporal.constants for consistent timestamp handling across multiple files. - Updated the date format from DEFAULT_DATE_FORMAT to DATETIME_FORMAT_MS_WITH_TZ in various activities to ensure uniformity in timestamp formatting. - Removed the obsolete patterns.py file that contained the DEFAULT_DATE_FORMAT constant. --- orchestrator/activities/formatters.py | 11 +++++------ orchestrator/activities/mongo_db.py | 12 ++++++------ orchestrator/activities/slot_manager.py | 8 ++++---- orchestrator/utils/patterns.py | 1 - .../workflows/subworkflows/process_notifications.py | 4 ++-- requirements.txt | 2 +- 6 files changed, 18 insertions(+), 20 deletions(-) delete mode 100644 orchestrator/utils/patterns.py diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 34732be..f3212ab 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -9,14 +9,13 @@ with workflow.unsafe.imports_passed_through(): import json from typing import Any from logging import Logger - from datetime import datetime from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.temporal.activities.base import BaseActivity from sientia_do.notifications.models import NotificationLevel from orchestrator.utils.orchestrator_functions import ( scouter, predictions_batch, gather_read_tags, build_tag_config ) - from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT + from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now from math import ceil topic_separator = "\n ========== \n" @@ -63,21 +62,21 @@ class Formatters(BaseActivity): schedule_config[self.scouter_namespace][pipeline['schedule_name']] = { **scouter(pipeline), "updated_at": pipeline.get( - "updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT)) + "updated_at", now().strftime(DATETIME_FORMAT_MS_WITH_TZ)) } elif pipeline['workflow_type'] == 'predictions_batch': schedule_config[self.laborious_namespace][pipeline['schedule_name'] ] = { **predictions_batch(pipeline), "updated_at": pipeline.get( - "updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT)) + "updated_at", now().strftime(DATETIME_FORMAT_MS_WITH_TZ)) } elif pipeline['workflow_type'] == 'minimal_retrain': schedule_config[self.laborious_namespace][pipeline['schedule_name'] ] = { **minimal_retrain(pipeline), "updated_at": pipeline.get( - "updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT)) + "updated_at", now().strftime(DATETIME_FORMAT_MS_WITH_TZ)) } self.info("Processed schedules", metadata=metadata) @@ -219,7 +218,7 @@ class Formatters(BaseActivity): if schedule_name in current_schedules: update_timestamp = schedule.get( - 'updated_at', datetime.now()) + 'updated_at', now()) old_timestamp = current_schedules[schedule_name] diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index e7a3ac1..3e094b4 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -2,7 +2,6 @@ from temporalio import workflow, activity with workflow.unsafe.imports_passed_through(): - from datetime import datetime from typing import Any import traceback from logging import Logger @@ -10,7 +9,8 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.temporal.activities.base import BaseActivity - from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT + from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now + from datetime import datetime def clear_mongo_id(docs: list) -> list: @@ -216,7 +216,7 @@ class MongoDB(BaseActivity): """ updated_pipelines = input_data.get("updated_pipelines", []) metadata = input_data.get("metadata", {}) - now = datetime.now() + now = now() collection = self.database["orchestrated_schedules"] self.info("Updating pipelines timestamps...", metadata=metadata) @@ -267,7 +267,7 @@ class MongoDB(BaseActivity): success_count = 0 - now = datetime.now() + now = now() argument = [ {"schedule_name": pipeline["schedule_name"], @@ -442,7 +442,7 @@ class MongoDB(BaseActivity): data_filter = { **base_data_filter, "timestamp": { - "$gt": datetime.strptime(last_data_timestamp, DEFAULT_DATE_FORMAT) + "$gt": datetime.strptime(last_data_timestamp, DATETIME_FORMAT_MS_WITH_TZ) } } @@ -460,7 +460,7 @@ class MongoDB(BaseActivity): for item in data: item['timestamp'] = item['timestamp'].strftime( - DEFAULT_DATE_FORMAT) + DATETIME_FORMAT_MS_WITH_TZ) self.info( f"Loaded {len(data)} documents from MongoDB", diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index a7dab02..bc68d23 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -9,7 +9,7 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.temporal.activities.redis_base import Redis from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel - from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT + from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now from pandas import DataFrame from datetime import datetime, timedelta @@ -344,10 +344,10 @@ class SlotManager(Redis): else: last_sent = datetime.strptime( - last_sent, DEFAULT_DATE_FORMAT) + last_sent, DATETIME_FORMAT_MS_WITH_TZ) # Check if "notification_ttl" seconds has passed since last sent - if (datetime.now() - last_sent) > timedelta(seconds=notification_ttl): + if (now() - last_sent) > timedelta(seconds=notification_ttl): alert_type = "persistent_alerts" # Check if this group must be notified @@ -382,7 +382,7 @@ class SlotManager(Redis): self.info("Storing notification cache...", metadata=metadata) - now = datetime.now().strftime(DEFAULT_DATE_FORMAT) + now = now().strftime(DATETIME_FORMAT_MS_WITH_TZ) for index, row in log_report.iterrows(): status = row['status'] diff --git a/orchestrator/utils/patterns.py b/orchestrator/utils/patterns.py deleted file mode 100644 index fe00beb..0000000 --- a/orchestrator/utils/patterns.py +++ /dev/null @@ -1 +0,0 @@ -DEFAULT_DATE_FORMAT = '%Y-%m-%d %H:%M:%S.%f' diff --git a/orchestrator/workflows/subworkflows/process_notifications.py b/orchestrator/workflows/subworkflows/process_notifications.py index 1c2a0bd..e02fc12 100644 --- a/orchestrator/workflows/subworkflows/process_notifications.py +++ b/orchestrator/workflows/subworkflows/process_notifications.py @@ -5,7 +5,7 @@ with workflow.unsafe.imports_passed_through(): from typing import Any from datetime import timedelta from sientia_do.temporal.policies import retry_policy - from sientia_do.temporal.constants import DATETIME_FORMAT + from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ @workflow.defn(name="process_notifications") @@ -84,7 +84,7 @@ class ProcessNotifications: "data": log_report, 'timestamp_conversion': { 'column': 'timestamp', - 'format': DATETIME_FORMAT + 'format': DATETIME_FORMAT_MS_WITH_TZ } }, schedule_to_close_timeout=timedelta(seconds=60), diff --git a/requirements.txt b/requirements.txt index 5d8b5e9..f27fcce 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,5 +5,5 @@ redis couchbase pymongo jinja2 -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.2 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.3 prometheus-client From 4aa7ea70fa66ec8f56e384a9a37b8879e1a8bd11 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 25 Aug 2025 08:30:42 -0300 Subject: [PATCH 08/10] SIENTIAPDE-1193 refactor: standardize timestamp handling across MongoDB and SlotManager - Replaced instances of `now()` with `date_now` for consistent timestamp usage in MongoDB and SlotManager activities. - Updated test cases to mock the new `now()` function, ensuring accurate timestamp handling in unit tests. - Adjusted timestamp formatting in various tests to align with the DATETIME_FORMAT_MS_WITH_TZ standard. --- orchestrator/activities/mongo_db.py | 8 ++-- orchestrator/activities/slot_manager.py | 4 +- .../orchestrator/activities/test_mongo_db.py | 40 +++++++++---------- .../activities/test_slot_manager.py | 14 +++---- .../test_process_notifications.py | 7 +++- 5 files changed, 39 insertions(+), 34 deletions(-) diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 3e094b4..e53f51a 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -216,7 +216,7 @@ class MongoDB(BaseActivity): """ updated_pipelines = input_data.get("updated_pipelines", []) metadata = input_data.get("metadata", {}) - now = now() + date_now = now() collection = self.database["orchestrated_schedules"] self.info("Updating pipelines timestamps...", metadata=metadata) @@ -233,7 +233,7 @@ class MongoDB(BaseActivity): try: collection.update_many( data_filter, - {"$set": {"updated_at": now}} + {"$set": {"updated_at": date_now}} ) success_count += 1 except Exception as e: @@ -267,12 +267,12 @@ class MongoDB(BaseActivity): success_count = 0 - now = now() + date_now = now() argument = [ {"schedule_name": pipeline["schedule_name"], "namespace": pipeline["namespace"], - "updated_at": now} + "updated_at": date_now} for pipeline in created_pipelines if pipeline["success"] ] data_filter = argument if argument else {} diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index bc68d23..862e258 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -382,12 +382,12 @@ class SlotManager(Redis): self.info("Storing notification cache...", metadata=metadata) - now = now().strftime(DATETIME_FORMAT_MS_WITH_TZ) + date_now = now().strftime(DATETIME_FORMAT_MS_WITH_TZ) for index, row in log_report.iterrows(): status = row['status'] if status == 'sent': key = f"{row['schedule']}:{row['notification_id']}" - self.set(key, now, ttl=sent_ttl) + self.set(key, date_now, ttl=sent_ttl) self.info("Notification cache stored...", metadata=metadata) diff --git a/tests/orchestrator/activities/test_mongo_db.py b/tests/orchestrator/activities/test_mongo_db.py index 742d42f..2ae912d 100644 --- a/tests/orchestrator/activities/test_mongo_db.py +++ b/tests/orchestrator/activities/test_mongo_db.py @@ -266,8 +266,8 @@ async def test_aggregate_documents_in_mongodb_missing_aggregation(mongo_db): @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_update_pipelines_timestamps_success(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_update_pipelines_timestamps_success(now_mock, mongo_db): input_data = {"updated_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test2", "namespace": "test2", "success": True} @@ -280,13 +280,13 @@ async def test_update_pipelines_timestamps_success(datetime_mock, mongo_db): {"schedule_name": "test2", "namespace": "test2"} ]}, {"$set": { - "updated_at": datetime_mock.now.return_value}} + "updated_at": now_mock.return_value}} ) @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_update_pipelines_timestamps_failure(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_update_pipelines_timestamps_failure(now_mock, mongo_db): input_data = { "updated_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "success": True}, @@ -314,8 +314,8 @@ async def test_update_pipelines_timestamps_failure(datetime_mock, mongo_db): @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_create_pipelines_timestamps_success(now_mock, mongo_db): input_data = {"created_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test2", "namespace": "test2", "success": True} @@ -325,16 +325,16 @@ async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db): mongo_db.database["pipelines"].insert_many.assert_called_once_with( [ {"schedule_name": "test1", "namespace": "test1", - "updated_at": datetime_mock.now.return_value}, + "updated_at": now_mock.return_value}, {"schedule_name": "test2", "namespace": "test2", - "updated_at": datetime_mock.now.return_value} + "updated_at": now_mock.return_value} ] ) @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_create_pipelines_timestamps_failure(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_create_pipelines_timestamps_failure(now_mock, mongo_db): input_data = {"created_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test2", "namespace": "test2", "success": True} @@ -360,8 +360,8 @@ async def test_create_pipelines_timestamps_failure(datetime_mock, mongo_db): @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_delete_pipelines_timestamps_success(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_delete_pipelines_timestamps_success(now_mock, mongo_db): input_data = {"deleted_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test2", "namespace": "test2", "success": True} @@ -377,8 +377,8 @@ async def test_delete_pipelines_timestamps_success(datetime_mock, mongo_db): @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_delete_pipelines_timestamps_failure(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_delete_pipelines_timestamps_failure(now_mock, mongo_db): input_data = {"deleted_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test2", "namespace": "test2", "success": True} @@ -573,14 +573,14 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongo_db): 'name': 'test1', 'value': 1, 'timestamp': datetime.strptime( - '2023-01-01 12:00:00.000000', '%Y-%m-%d %H:%M:%S.%f') + '2023-01-01 12:00:00.000000+0000', '%Y-%m-%d %H:%M:%S.%f%z') } ] result = await mongo_db.load_latest_data({ 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, 'collection_name': 'test_collection', - 'last_data_timestamp': '2023-01-01 12:00:00.000000', + 'last_data_timestamp': '2023-01-01 12:00:00.000000+0000', 'base_data_filter': { 'level': 'ERROR' } @@ -594,7 +594,7 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongo_db): 'level': 'ERROR', 'timestamp': { '$gt': datetime.strptime( - '2023-01-01 12:00:00.000000', '%Y-%m-%d %H:%M:%S.%f') + '2023-01-01 12:00:00.000000+0000', '%Y-%m-%d %H:%M:%S.%f%z') } }, {"_id": 0} @@ -603,7 +603,7 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongo_db): assert result == [{ 'name': 'test1', 'value': 1, - 'timestamp': '2023-01-01 12:00:00.000000' + 'timestamp': '2023-01-01 12:00:00.000000+0000' }] @@ -620,7 +620,7 @@ async def test_load_latest_data_error(mongo_db): await mongo_db.load_latest_data({ 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, 'collection_name': 'test_collection', - 'last_data_timestamp': '2023-01-01 12:00:00.000000', + 'last_data_timestamp': '2023-01-01 12:00:00.000000+0000', 'base_data_filter': { 'level': 'ERROR' } diff --git a/tests/orchestrator/activities/test_slot_manager.py b/tests/orchestrator/activities/test_slot_manager.py index debf0a7..80cd384 100644 --- a/tests/orchestrator/activities/test_slot_manager.py +++ b/tests/orchestrator/activities/test_slot_manager.py @@ -4,7 +4,7 @@ from pandas import DataFrame from pytest import mark, fixture from orchestrator.activities.slot_manager import SlotManager from sientia_do.notifications.models import NotificationLevel -from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT +from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now metadata = { "metadata": { @@ -373,13 +373,13 @@ async def test_put_last_data_timestamp_error(slot_manager): async def test_filter_notification_alerts(slot_manager): slot_manager.get = MagicMock(side_effect=[ None, - (datetime.now() - timedelta(seconds=600) - ).strftime(DEFAULT_DATE_FORMAT), - datetime.now().strftime(DEFAULT_DATE_FORMAT), + (now() - timedelta(seconds=600) + ).strftime(DATETIME_FORMAT_MS_WITH_TZ), + now().strftime(DATETIME_FORMAT_MS_WITH_TZ), None, - (datetime.now() - timedelta(seconds=600) - ).strftime(DEFAULT_DATE_FORMAT), - datetime.now().strftime(DEFAULT_DATE_FORMAT)]) + (now() - timedelta(seconds=600) + ).strftime(DATETIME_FORMAT_MS_WITH_TZ), + now().strftime(DATETIME_FORMAT_MS_WITH_TZ)]) input_data = { **metadata, diff --git a/tests/orchestrator/workflows/subworkflows/test_process_notifications.py b/tests/orchestrator/workflows/subworkflows/test_process_notifications.py index df25c84..8a41654 100644 --- a/tests/orchestrator/workflows/subworkflows/test_process_notifications.py +++ b/tests/orchestrator/workflows/subworkflows/test_process_notifications.py @@ -2,6 +2,7 @@ from unittest.mock import AsyncMock, patch, ANY, call from pytest import fixture, mark from orchestrator.workflows.subworkflows.process_notifications import ProcessNotifications from orchestrator.activities.activities import Activities +from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ @fixture @@ -76,7 +77,11 @@ async def test_run(workflow_mock, process_notifications): **metadata, 'schema': input_data['schema'], 'table_name': input_data['table_name'], - 'data': workflow_mock.execute_local_activity_method.return_value + 'data': workflow_mock.execute_local_activity_method.return_value, + 'timestamp_conversion': { + 'column': 'timestamp', + 'format': DATETIME_FORMAT_MS_WITH_TZ + } }, schedule_to_close_timeout=ANY, retry_policy=ANY From 575a99fd318c0e44e4ba88f9c1932c8aff487478 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 25 Aug 2025 08:48:02 -0300 Subject: [PATCH 09/10] SIENTIAPDE-1193 chore: update image tag and enhance datetime column handling in orchestrator functions - Updated the image tag in values.yaml from 0.4.2 to 0.4.4 for the latest version. - Added support for datetime_columns in minimal_retrain and predictions_batch functions to improve timestamp handling in orchestrator functions. --- orchestrator/utils/orchestrator_functions.py | 2 ++ values.yaml | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 38bde51..b49a771 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -48,6 +48,7 @@ def minimal_retrain(config: dict[str, Any]): "query": config['query'], "schema": "sientia_data", "table_name": "log_retrain", + "datetime_columns": config.get('datetime_columns', []), } @@ -187,6 +188,7 @@ def predictions_batch(config: dict[str, Any]): **common_config(config), "query": config['query'], + "datetime_columns": config.get('datetime_columns', []), "schema": "sientia_data", "table_name": "predictions", "retention_time": config.get('model_retention_minutes', 60) * 60, diff --git a/values.yaml b/values.yaml index c2c69e9..300ea80 100644 --- a/values.yaml +++ b/values.yaml @@ -11,7 +11,7 @@ image: # This sets the pull policy for images. pullPolicy: Always # Overrides the image tag whose default is the chart appVersion. - tag: "0.4.2" + tag: "0.4.4" # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ imagePullSecrets: From 2aae718e5af8111e31aafcecd31d2480796bcba8 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 25 Aug 2025 09:16:47 -0300 Subject: [PATCH 10/10] SIENTIAPDE-1193 Enhance test cases to include datetime_columns for improved timestamp handling in minimal_retrain and predictions_batch functions --- coverage.sh | 1 + tests/orchestrator/utils/test_orchestrator_functions.py | 6 +++++- 2 files changed, 6 insertions(+), 1 deletion(-) create mode 100755 coverage.sh diff --git a/coverage.sh b/coverage.sh new file mode 100755 index 0000000..c3235b5 --- /dev/null +++ b/coverage.sh @@ -0,0 +1 @@ +pytest --cov=orchestrator --cov-report=html && xdg-open htmlcov/index.html \ No newline at end of file diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index eb4b056..09fae10 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -41,6 +41,7 @@ def test_minimal_retrain(): "name": "test_model_name" }, "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;", + "datetime_columns": ["timestamp"] } result = minimal_retrain(config) expected = { @@ -53,6 +54,7 @@ def test_minimal_retrain(): "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;", "schema": "sientia_data", "table_name": "log_retrain", + "datetime_columns": ["timestamp"] } assert result == expected @@ -195,6 +197,7 @@ def test_predictions_batch(mock_process_path_priority, } ], "path_priority": ["STOP", "CONTINUE", "REPEAT"], + "datetime_columns": ["timestamp"] } result = predictions_batch(config) @@ -268,7 +271,8 @@ def test_predictions_batch(mock_process_path_priority, "config": {} } }, - "path_priority": ["STOP", "CONTINUE", "REPEAT"] + "path_priority": ["STOP", "CONTINUE", "REPEAT"], + "datetime_columns": ["timestamp"] } assert result == expected