gitea_admin/sientia-dataops-orchestrator_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #17 from Aignosi/SIENTIAPDE-1193-conferir-como-a-escrita-de-datetime-ocorre-no-temporal

Browse Source 
SIENTIAPDE-1193: Improve Timestamp Handling and Update Dependencies
This commit is contained in:
Bruno Domingues
2025-08-26 13:39:28 +00:00
committed by GitHub
parent 3e3926832b c581e7da47
commit e3b35bf4fa
13 changed files with 68 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
coverage.sh Executable file
View File

@@ -0,0 +1 @@
pytest --cov=orchestrator --cov-report=html && xdg-open htmlcov/index.html

11
orchestrator/activities/formatters.py
View File

@@ -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]

16
orchestrator/activities/mongo_db.py
View File

@@ -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()
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 = datetime.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 {}
@@ -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",

10
orchestrator/activities/slot_manager.py
View File

@@ -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,12 +382,12 @@ class SlotManager(Redis):
self.info("Storing notification cache...", metadata=metadata)
now = datetime.now().strftime(DEFAULT_DATE_FORMAT)
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)

2
orchestrator/utils/orchestrator_functions.py
View File

@@ -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,

1
orchestrator/utils/patterns.py
View File

@@ -1 +0,0 @@
DEFAULT_DATE_FORMAT = '%Y-%m-%d %H:%M:%S.%f'

7
orchestrator/workflows/subworkflows/process_notifications.py
View File

@@ -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_MS_WITH_TZ
@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_MS_WITH_TZ
}
},
schedule_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy

2
requirements.txt
View File

@@ -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.3
prometheus-client

40
tests/orchestrator/activities/test_mongo_db.py
View File

@@ -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'
}

14
tests/orchestrator/activities/test_slot_manager.py
View File

@@ -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,

6
tests/orchestrator/utils/test_orchestrator_functions.py
View File

@@ -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

7
tests/orchestrator/workflows/subworkflows/test_process_notifications.py
View File

@@ -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

4
values.yaml
View File

@@ -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:
@@ -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"