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

SIENTIAPDE-1150

Browse Source 
fix: implement schedule normalization and timestamp management in MongoDB activities, enhance orchestration workflow with new formatting and normalization methods
This commit is contained in:
vitor-aignosi
2025-07-14 12:59:52 -03:00
parent e3a88eb1d5
commit 43206b578d
10 changed files with 648 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
orchestrator/activities/formatters.py
View File

@@ -1,15 +1,19 @@
from sientia_do.notifications.models import NotificationLevel
from temporalio import activity, workflow
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 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 math import ceil
@@ -119,29 +123,48 @@ class Formatters(BaseActivity):
return slot_config
def compare_configs(self,
schedules: dict[str, Any], current_schedules: dict[str, Any],
to_update: dict[str, Any], to_create: dict[str, Any],
namespace: str):
"""
Compares the schedules and current schedules, and updates the
to_update and to_create dictionaries.
@activity.defn(name="format_schedule_config")
async def format_schedule_config(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Formats the schedule config to a dictionary with the schedule name as the key.
input_data:
- schedule_config (list[dict[str, Any]]): The schedule config to format.
Returns:
- dict[str, Any]: The formatted schedule config.
"""
schedule_config = input_data['schedule_config']
config = {}
for schedule in schedule_config:
namespace = schedule['namespace']
schedule_name = schedule['schedule_name']
updated_at = schedule['updated_at']
if namespace not in config:
config[namespace] = {}
config[namespace][schedule_name] = updated_at
return config
def compare_config_timestamps(self,
schedules: dict[str, Any], current_schedules: dict[str, Any],
to_update: dict[str, Any], to_create: dict[str, Any],
namespace: str):
"""
Compares the timestamps of the schedule and the current schedule.
"""
for schedule_name, schedule in schedules.items():
if schedule_name in current_schedules:
old_config = current_schedules[schedule_name]['data']
update_timestamp = schedule.get(
'updated_at', datetime.now().strftime(DEFAULT_DATE_FORMAT))
self.logger.debug(f"Comparing {schedule_name}:")
self.logger.debug(json.dumps(
old_config, indent=4, sort_keys=True))
self.logger.debug(json.dumps(
schedule, indent=4, sort_keys=True))
old_timestamp = current_schedules[schedule_name]
if schedule != old_config:
if update_timestamp > old_timestamp:
to_update[namespace][schedule_name] = schedule
elif schedule_name not in current_schedules:
else:
to_create[namespace][schedule_name] = schedule
@activity.defn(name="create_schedule_config")
@@ -183,7 +206,7 @@ class Formatters(BaseActivity):
for namespace, schedules in schedule_config.items():
current_schedules = current_schedule_config.get(namespace, {})
self.compare_configs(
self.compare_config_timestamps(
schedules, current_schedules, to_update, to_create, namespace)
for namespace, schedules in current_schedule_config.items():
@@ -264,6 +287,15 @@ class Formatters(BaseActivity):
attachment_content=json.dumps(attachment, indent=4, sort_keys=True)
)
def parse_report_schedule(self, input_data: dict[str, Any]) -> tuple[list[str], list[str]]:
success_keys = [f"{value['namespace']}/{value['schedule_name']}"
for value in input_data if value['success']]
error_keys = [f"{value['namespace']}/{value['schedule_name']}: {value['message']}"
for value in input_data if not value['success']]
return success_keys, error_keys
def parse_report(self, input_data: dict[str, Any]) -> tuple[list[str], list[str]]:
success_keys = [key for key, value
in input_data.items() if value['success']]
@@ -295,7 +327,8 @@ class Formatters(BaseActivity):
# Send report for created schedules
if len(created_schedules) > 0:
success_keys, error_keys = self.parse_report(created_schedules)
success_keys, error_keys = self.parse_report_schedule(
created_schedules)
if len(success_keys) > 0:
self.send_success_report(
@@ -312,7 +345,8 @@ class Formatters(BaseActivity):
# Send report for updated schedules
if len(updated_schedules) > 0:
success_keys, error_keys = self.parse_report(updated_schedules)
success_keys, error_keys = self.parse_report_schedule(
updated_schedules)
if len(success_keys) > 0:
self.send_success_report(
@@ -328,7 +362,8 @@ class Formatters(BaseActivity):
)
if len(deleted_schedules) > 0:
success_keys, error_keys = self.parse_report(deleted_schedules)
success_keys, error_keys = self.parse_report_schedule(
deleted_schedules)
if len(success_keys) > 0:
self.send_success_report(

56
orchestrator/activities/mongo_db.py
View File

@@ -1,5 +1,8 @@
import datetime
from temporalio import workflow, activity
from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT
with workflow.unsafe.imports_passed_through():
from typing import Any
import traceback
@@ -184,3 +187,56 @@ class MongoDB(BaseActivity):
self.logger.error(trace)
raise e
@activity.defn(name="update_pipelines_timestamps")
async def update_pipelines_timestamps(self, input_data: dict[str, Any]) -> None:
"""
Update the timestamps of the pipelines in the MongoDB collection.
input_data:
- updated_pipelines (list): List of updated pipelines.
"""
updated_pipelines = input_data.get("updated_pipelines", [])
now = datetime.now().strftime(DEFAULT_DATE_FORMAT)
self.database["pipelines"].update_many(
{"$or": [
{"schedule_name": pipeline["schedule_name"],
"namespace": pipeline["namespace"]}
for pipeline in updated_pipelines
]},
{"$set": {"updated_at": now}}
)
@activity.defn(name="create_pipelines_timestamps")
async def create_pipelines_timestamps(self, input_data: dict[str, Any]) -> None:
"""
Create the timestamps of the pipelines in the MongoDB collection.
input_data:
- created_pipelines (list): List of created pipelines.
"""
created_pipelines = input_data.get("created_pipelines", [])
now = datetime.now().strftime(DEFAULT_DATE_FORMAT)
self.database["pipelines"].insert_many([{
"schedule_name": pipeline["schedule_name"],
"namespace": pipeline["namespace"],
"updated_at": now
} for pipeline in created_pipelines])
@activity.defn(name="delete_pipelines_timestamps")
async def delete_pipelines_timestamps(self, input_data: dict[str, Any]) -> None:
"""
Delete the timestamps of the pipelines in the MongoDB collection.
input_data:
- deleted_pipelines (list): List of deleted pipelines.
"""
deleted_pipelines = input_data.get("deleted_pipelines", [])
self.database["pipelines"].delete_many({
"$or": [
{"schedule_name": pipeline["schedule_name"],
"namespace": pipeline["namespace"]}
for pipeline in deleted_pipelines
]
})

91
orchestrator/activities/temporal_manager.py
View File

@@ -129,6 +129,37 @@ class TemporalManager(BaseActivity):
return orchestrated_schedules
@activity.defn(name="normalize_schedules")
async def normalize_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Normalize schedules. Removes schedules with no update time in mongo db collection "orchestrated_schedules".
input_data:
- orchestrated_schedules (dict[str, Any]): The orchestrated schedules to compare.
"""
self.logger.info("Getting orchestrated schedules...")
orchestrated_schedules = input_data.get('orchestrated_schedules', {})
for namespace, client in self.temporal_clients.items():
schedules = orchestrated_schedules.get(namespace, {})
self.logger.info(
"Getting orchestrated schedules for %s", namespace)
async for schedule in await client.list_schedules():
search_attrs = getattr(schedule, "search_attributes", {})
if search_attrs.get("orchestrated", ["false"]) == ["true"]:
schedule_id = schedule.id
if schedule_id not in schedules:
self.logger.info(
"Schedule %s not found in mongo db, cleaning up", schedule_id)
handle = client.get_schedule_handle(
schedule_id)
await handle.delete()
@activity.defn(name="create_schedules")
async def create_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
@@ -145,7 +176,7 @@ class TemporalManager(BaseActivity):
schedules_to_create = input_data['schedules']
report = {}
report = []
for namespace, schedules in schedules_to_create.items():
client = self.temporal_clients.get(namespace)
@@ -198,17 +229,21 @@ class TemporalManager(BaseActivity):
search_attributes=search_attributes
)
report[schedule_name] = {
report.append({
"namespace": namespace,
"schedule_name": schedule_name,
"success": True,
"message": "Schedule created successfully"
}
})
except Exception as e:
self.logger.error("Failed to create schedule %s: %s",
schedule_name, str(e))
report[schedule_name] = {
report.append({
"namespace": namespace,
"schedule_name": schedule_name,
"success": False,
"message": str(e)
}
})
self.logger.info(f"Processed {len(schedules_to_create)} schedules")
@@ -233,18 +268,19 @@ class TemporalManager(BaseActivity):
schedules_to_update = input_data['schedules']
report = {}
report = []
for namespace, schedules in schedules_to_update.items():
schedule_handles = self.schedule_handles.get(namespace, None)
client = self.temporal_clients.get(namespace)
if schedule_handles is None:
if not client:
raise ValueError(
f"Schedule handles for {namespace} not found, handles: {self.schedule_handles}")
f"Temporal client for {namespace} not found, clients: {self.temporal_clients}")
for schedule_name, schedule in schedules.items():
try:
handler = schedule_handles.get(schedule_name)
handler = client.get_schedule_handle(
schedule_name)
if not handler:
raise ValueError(f"Schedule {schedule_name} not found")
@@ -276,17 +312,21 @@ class TemporalManager(BaseActivity):
del update_schedule
report[schedule_name] = {
report.append({
"namespace": namespace,
"schedule_name": schedule_name,
"success": True,
"message": "Schedule updated successfully"
}
})
except Exception as e:
self.logger.error("Failed to update schedule %s: %s",
schedule_name, str(e))
report[schedule_name] = {
report.append({
"namespace": namespace,
"schedule_name": schedule_name,
"success": False,
"message": str(e)
}
})
self.logger.info(f"Processed {len(schedules_to_update)} schedules")
@@ -311,18 +351,19 @@ class TemporalManager(BaseActivity):
schedules_to_delete = input_data['schedules']
report = {}
report = []
for namespace, schedules in schedules_to_delete.items():
schedule_handles = self.schedule_handles.get(namespace, None)
client = self.temporal_clients.get(namespace)
if schedule_handles is None:
if not client:
raise ValueError(
f"Schedule handles for {namespace} not found, handles: {self.schedule_handles}")
f"Temporal client for {namespace} not found, clients: {self.temporal_clients}")
for schedule_name in schedules:
try:
handler = schedule_handles.get(schedule_name)
handler = client.get_schedule_handle(
schedule_name)
if not handler:
raise ValueError(f"Schedule {schedule_name} not found")
@@ -331,17 +372,21 @@ class TemporalManager(BaseActivity):
del self.schedule_handles[namespace][schedule_name]
report[schedule_name] = {
report.append({
"namespace": namespace,
"schedule_name": schedule_name,
"success": True,
"message": "Schedule deleted successfully"
}
})
except Exception as e:
self.logger.error("Failed to delete schedule %s: %s",
schedule_name, str(e))
report[schedule_name] = {
report.append({
"namespace": namespace,
"schedule_name": schedule_name,
"success": False,
"message": str(e)
}
})
self.logger.info(f"Processed {len(schedules_to_delete)} schedules")

1
orchestrator/utils/patterns.py Normal file
View File

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

29
orchestrator/workflows/orchestrator.py
View File

@@ -33,7 +33,12 @@ class Orchestrator:
)
orchestrated_schedules_handler = workflow.execute_local_activity_method(
Activities.load_schedule,
Activities.find_documents_in_mongodb,
{
'query': {
'collection': 'orchestrated_schedules'
}
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=600)
)
@@ -56,6 +61,15 @@ class Orchestrator:
opc_servers = await opc_servers_handler
active_ingestors = await active_ingestors_handler
formatted_orchestrated_schedules_handler = workflow.execute_local_activity_method(
Activities.format_schedule_config,
{
'schedule_config': orchestrated_schedules
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
)
schedules_config_handler = workflow.execute_local_activity_method(
Activities.process_schedules,
{
@@ -78,11 +92,12 @@ class Orchestrator:
schedules_config = await schedules_config_handler
slot_config = await slot_config_handler
formatted_orchestrated_schedules = await formatted_orchestrated_schedules_handler
schedule_actions_handler = workflow.execute_local_activity_method(
Activities.create_schedule_config,
{
'current_schedule_config': orchestrated_schedules,
'current_schedule_config': formatted_orchestrated_schedules,
'schedule_config': schedules_config
},
retry_policy=retry_policy,
@@ -99,8 +114,18 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60)
)
normalize_schedules_handler = workflow.execute_local_activity_method(
Activities.normalize_schedules,
{
'orchestrated_schedules': formatted_orchestrated_schedules
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
)
schedule_actions = await schedule_actions_handler
slot_actions = await slot_actions_handler
await normalize_schedules_handler
slot_deletion_report_handler = workflow.execute_activity_method(
Activities.delete_slots,