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

Merge pull request #7 from Aignosi/SIENTIAPDE-1150-ajustar-orquestrador-para-manter-uma-store-de-detalhes-dos-schedules

Browse Source 
Sientiapde 1150 ajustar orquestrador para manter uma store de detalhes dos schedules
This commit is contained in:
vitor-aignosi
2025-07-15 13:53:49 -03:00
committed by GitHub
parent e3a88eb1d5 2533ea882f
commit 63fa276e89
19 changed files with 940 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
README.md
View File

@@ -0,0 +1,6 @@
#PR shortcut
```
git log origin/main..HEAD --no-merges > git_log
```
Prompt:
Write a summary of PR changes in markdown. Be objective and direct. Write to file

4
orchestrator/activities/couchbase.py
View File

@@ -47,7 +47,7 @@ class Couchbase(BaseActivity):
try:
self.cluster.close()
except Exception as e:
self.logger.error("Failed to close Couchbase connection: %s", e)
self.logger.error(f"Failed to close Couchbase connection: {e}")
def __del__(self):
self.shutdown()
@@ -65,7 +65,7 @@ class Couchbase(BaseActivity):
"""
query = input_data['query']
self.logger.info("Executing couchbase query: %s", query)
self.logger.info(f"Executing couchbase query: {query}")
try:
result = self.cluster.query(query)

88
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
@@ -49,11 +53,18 @@ class Formatters(BaseActivity):
for pipeline in pipelines:
if pipeline['workflow_type'] == 'scouter':
schedule_config[self.scouter_namespace][pipeline['schedule_name']] = scouter(
pipeline)
schedule_config[self.scouter_namespace][pipeline['schedule_name']] = {
**scouter(pipeline),
"updated_at": pipeline.get(
"updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT))
}
elif pipeline['workflow_type'] == 'predictions_batch':
schedule_config[self.laborious_namespace][pipeline['schedule_name']
] = predictions_batch(pipeline)
] = {
**predictions_batch(pipeline),
"updated_at": pipeline.get(
"updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT))
}
self.logger.info("Processed schedules")
@@ -119,29 +130,52 @@ class Formatters(BaseActivity):
return slot_config
def compare_configs(self,
@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 schedules and current schedules, and updates the
to_update and to_create dictionaries.
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']
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))
update_timestamp = schedule.get(
'updated_at', datetime.now().strftime(DEFAULT_DATE_FORMAT))
if schedule != old_config:
old_timestamp = current_schedules[schedule_name]
self.logger.debug(
f"Comparing schedule {schedule_name}:{update_timestamp} vs {old_timestamp}")
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 +217,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 +298,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 +338,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 +356,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 +373,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(

65
orchestrator/activities/mongo_db.py
View File

@@ -1,6 +1,8 @@
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
@@ -8,6 +10,7 @@ with workflow.unsafe.imports_passed_through():
from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.activities.base import BaseActivity
from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT
def clear_mongo_id(docs: list) -> list:
@@ -184,3 +187,65 @@ 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)
collection = self.database["orchestrated_schedules"]
argument = [
{"schedule_name": pipeline["schedule_name"],
"namespace": pipeline["namespace"]}
for pipeline in updated_pipelines if pipeline["success"]
]
data_filter = {"$or": argument} if argument else {}
collection.update_many(
data_filter,
{"$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", [])
collection = self.database["orchestrated_schedules"]
now = datetime.now().strftime(DEFAULT_DATE_FORMAT)
argument = [
{"schedule_name": pipeline["schedule_name"],
"namespace": pipeline["namespace"],
"updated_at": now}
for pipeline in created_pipelines if pipeline["success"]
]
data_filter = argument if argument else {}
collection.insert_many(data_filter)
@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", [])
collection = self.database["orchestrated_schedules"]
argument = [
{"schedule_name": pipeline["schedule_name"],
"namespace": pipeline["namespace"]}
for pipeline in deleted_pipelines if pipeline["success"]
]
data_filter = {"$or": argument} if argument else {}
collection.delete_many(data_filter)

22
orchestrator/activities/slot_manager.py
View File

@@ -32,7 +32,7 @@ class SlotManager(Redis):
slot_keys = self.redis_client.keys("slot:opc_tags:*")
self.logger.debug("Slot keys: %s", slot_keys)
self.logger.debug(f"Slot keys: {slot_keys}")
if slot_keys:
if isinstance(slot_keys[0], bytes):
@@ -45,8 +45,8 @@ class SlotManager(Redis):
self.logger.info(f"Loaded {len(opc_slots)} OPC slots")
self.logger.debug("Loaded: \n %s",
json.dumps(opc_slots, indent=4, sort_keys=True))
self.logger.debug(
f"Loaded: \n {json.dumps(opc_slots, indent=4, sort_keys=True)}")
return opc_slots
@@ -65,7 +65,7 @@ class SlotManager(Redis):
self.logger.info(f"Loaded {len(active_ingestors)} active ingestors")
self.logger.debug("Active ingestors: \n %s", active_ingestors)
self.logger.debug(f"Active ingestors: \n {active_ingestors}")
ingestors = []
@@ -105,8 +105,7 @@ class SlotManager(Redis):
"message": "Slot updated successfully"
}
except Exception as e:
self.logger.error("Failed to update slot %s: %s",
slot, str(e))
self.logger.error(f"Failed to update slot {slot}: {str(e)}")
report[slot] = {
"success": False,
"message": str(e)
@@ -114,8 +113,8 @@ class SlotManager(Redis):
self.logger.info(f"Updated {len(to_insert)} OPC slots")
self.logger.debug("Report: \n %s",
json.dumps(report, indent=4, sort_keys=True))
self.logger.debug(
f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}")
return report
@@ -146,8 +145,7 @@ class SlotManager(Redis):
"message": "Slot deleted successfully"
}
except Exception as e:
self.logger.error("Failed to delete slot %s: %s",
slot, str(e))
self.logger.error(f"Failed to delete slot {slot}: {str(e)}")
report[slot] = {
"success": False,
"message": str(e)
@@ -155,7 +153,7 @@ class SlotManager(Redis):
self.logger.info(f"Deleted {len(to_delete)} OPC slots")
self.logger.debug("Report: \n %s",
json.dumps(report, indent=4, sort_keys=True))
self.logger.debug(
f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}")
return report

146
orchestrator/activities/temporal_manager.py
View File

@@ -80,14 +80,14 @@ class TemporalManager(BaseActivity):
orchestrated_schedules[namespace] = {}
self.logger.info(
"Getting orchestrated schedules for %s", namespace)
f"Getting orchestrated schedules for {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
self.logger.debug("Schedule id: %s", schedule_id)
self.logger.debug(f"Schedule id: {schedule_id}")
handle = client.get_schedule_handle(
schedule_id)
@@ -117,20 +117,50 @@ class TemporalManager(BaseActivity):
await sleep(0.1)
self.logger.info("Found %d orchestrated schedules",
len(orchestrated_schedules[self.scouter_namespace]) +
len(orchestrated_schedules[self.laborious_namespace]))
self.logger.info(
f"Found {len(orchestrated_schedules[self.scouter_namespace]) + len(orchestrated_schedules[self.laborious_namespace])} orchestrated schedules")
self.logger.debug("Orchestrated schedules: %s",
orchestrated_schedules)
self.logger.debug(
f"Orchestrated schedules: {orchestrated_schedules}")
self.logger.debug("Schedule handles: %s",
self.schedule_handles)
self.logger.debug(
f"Schedule handles: {self.schedule_handles}")
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(
f"Getting orchestrated schedules for {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(
f"Schedule {schedule_id} not found in mongo db, cleaning up")
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]:
async def create_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
"""
Create schedules in Temporal
@@ -145,7 +175,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)
@@ -173,8 +203,8 @@ class TemporalManager(BaseActivity):
try:
self.logger.debug(f"Creating schedule {schedule_name}:")
self.logger.debug(json.dumps(
schedule, indent=4, sort_keys=True))
self.logger.debug(
f"{json.dumps(schedule, indent=4, sort_keys=True)}")
await client.create_schedule(
schedule_name,
@@ -198,27 +228,31 @@ 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] = {
self.logger.error(
f"Failed to create schedule {schedule_name}: {str(e)}")
report.append({
"namespace": namespace,
"schedule_name": schedule_name,
"success": False,
"message": str(e)
}
})
self.logger.info(f"Processed {len(schedules_to_create)} schedules")
self.logger.debug("\n %s",
json.dumps(report, indent=4, sort_keys=True))
self.logger.debug(
f"\n {json.dumps(report, indent=4, sort_keys=True)}")
return report
@activity.defn(name="update_schedules")
async def update_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]:
async def update_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
"""
Update schedules in Temporal
@@ -233,18 +267,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")
@@ -257,8 +292,8 @@ class TemporalManager(BaseActivity):
if hasattr(schedule_action, "args"):
self.logger.debug("New schedule:")
self.logger.debug(json.dumps(
schedule, indent=4, sort_keys=True)) # NOSONAR
self.logger.debug(
f"{json.dumps(schedule, indent=4, sort_keys=True)}") # NOSONAR
schedule_action.args = [schedule]
@@ -276,27 +311,31 @@ 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] = {
self.logger.error(
f"Failed to update schedule {schedule_name}: {str(e)}")
report.append({
"namespace": namespace,
"schedule_name": schedule_name,
"success": False,
"message": str(e)
}
})
self.logger.info(f"Processed {len(schedules_to_update)} schedules")
self.logger.debug("\n %s",
json.dumps(report, indent=4, sort_keys=True))
self.logger.debug(
f"\n {json.dumps(report, indent=4, sort_keys=True)}")
return report
@activity.defn(name="delete_schedules")
async def delete_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]:
async def delete_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
"""
Delete schedules in Temporal
@@ -311,18 +350,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,21 +371,25 @@ 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] = {
self.logger.error(
f"Failed to delete schedule {schedule_name}: {str(e)}")
report.append({
"namespace": namespace,
"schedule_name": schedule_name,
"success": False,
"message": str(e)
}
})
self.logger.info(f"Processed {len(schedules_to_delete)} schedules")
self.logger.debug("\n %s",
json.dumps(report, indent=4, sort_keys=True))
self.logger.debug(
f"\n {json.dumps(report, indent=4, sort_keys=True)}")
return report

10
orchestrator/utils/connectors_config.py
View File

@@ -5,15 +5,15 @@ def build_redis_config():
return {
'host': getenv('REDIS_HOST', 'localhost'),
'port': int(getenv('REDIS_PORT', '6379')),
'username': getenv('REDIS_USERNAME', None),
'password': getenv('REDIS_PASSWORD', None)
'username': getenv('REDIS_USERNAME', 'default'),
'password': getenv('REDIS_PASSWORD', 'bdnZOpcyiL')
}
def build_mongodb_config():
username = getenv('MONGODB_USERNAME', 'sientia')
password = getenv('MONGODB_PASSWORD', 'sientia')
uri = getenv('MONGODB_URL', 'localhost:27017')
username = getenv('MONGODB_USERNAME', 'root')
password = getenv('MONGODB_PASSWORD', 'wKZDbMNU1c')
uri = getenv('MONGODB_URL', 'localhost:27018')
connection_string = f'mongodb://{username}:{password}@{uri}'
return {

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

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

7
orchestrator/worker/worker.py
View File

@@ -69,11 +69,15 @@ async def main():
# MongoDB
activities.aggregate_documents_in_mongodb,
activities.find_documents_in_mongodb,
activities.update_pipelines_timestamps,
activities.create_pipelines_timestamps,
activities.delete_pipelines_timestamps,
# Temporal
activities.load_schedule,
activities.create_schedules,
activities.update_schedules,
activities.delete_schedules,
activities.normalize_schedules,
# Formatters
activities.process_schedules,
activities.process_slots,
@@ -81,6 +85,7 @@ async def main():
activities.create_slot_config,
activities.report_schedule_orchestration,
activities.report_slot_orchestration,
activities.format_schedule_config,
]
)
]
@@ -96,7 +101,7 @@ async def main():
# If an exception occurs in any of the worker handlers, it will be propagated here.
await asyncio.gather(*handlers)
except BaseException as e:
logger.error("An unhandled exception occurred: %s", e, exc_info=True)
logger.error(f"An unhandled exception occurred: {e}", exc_info=True)
finally:
if notification_handler:
notification_handler.shutdown()

112
orchestrator/workflows/orchestrator.py
View File

@@ -14,7 +14,7 @@ class Orchestrator:
input_data['workflow_name'] = 'orchestrator'
pipeline_config_handler = workflow.execute_local_activity_method(
pipeline_config_handler = workflow.start_local_activity_method(
Activities.aggregate_documents_in_mongodb,
{
'query': input_data['pipelines_query']
@@ -23,7 +23,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60)
)
opc_servers_handler = workflow.execute_local_activity_method(
opc_servers_handler = workflow.start_local_activity_method(
Activities.find_documents_in_mongodb,
{
'query': input_data['opc_servers_query']
@@ -32,19 +32,24 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60)
)
orchestrated_schedules_handler = workflow.execute_local_activity_method(
Activities.load_schedule,
orchestrated_schedules_handler = workflow.start_local_activity_method(
Activities.find_documents_in_mongodb,
{
'query': {
'collection': 'orchestrated_schedules'
}
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=600)
start_to_close_timeout=timedelta(seconds=60)
)
current_slot_config_handler = workflow.execute_local_activity_method(
current_slot_config_handler = workflow.start_local_activity_method(
Activities.load_opc_slots,
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
)
active_ingestors_handler = workflow.execute_local_activity_method(
active_ingestors_handler = workflow.start_local_activity_method(
Activities.load_active_ingestors,
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
@@ -56,7 +61,16 @@ class Orchestrator:
opc_servers = await opc_servers_handler
active_ingestors = await active_ingestors_handler
schedules_config_handler = workflow.execute_local_activity_method(
formatted_orchestrated_schedules_handler = workflow.start_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.start_local_activity_method(
Activities.process_schedules,
{
'pipelines': pipeline_config
@@ -65,7 +79,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60)
)
slot_config_handler = workflow.execute_local_activity_method(
slot_config_handler = workflow.start_local_activity_method(
Activities.process_slots,
{
'opc_servers': opc_servers,
@@ -78,18 +92,19 @@ 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(
schedule_actions_handler = workflow.start_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,
start_to_close_timeout=timedelta(seconds=60)
)
slot_actions_handler = workflow.execute_local_activity_method(
slot_actions_handler = workflow.start_local_activity_method(
Activities.create_slot_config,
{
'current_slot_config': current_slot_config,
@@ -99,10 +114,20 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60)
)
normalize_schedules_handler = workflow.start_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(
slot_deletion_report_handler = workflow.start_activity_method(
Activities.delete_slots,
{
'to_delete': slot_actions['to_delete']
@@ -111,7 +136,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60)
)
slot_insertion_report_handler = workflow.execute_activity_method(
slot_insertion_report_handler = workflow.start_activity_method(
Activities.update_slots,
{
'to_insert': slot_actions['to_insert']
@@ -120,7 +145,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60)
)
schedule_deletion_report_handler = workflow.execute_activity_method(
schedule_deletion_report_handler = workflow.start_activity_method(
Activities.delete_schedules,
{
'schedules': schedule_actions['to_delete']
@@ -129,7 +154,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60)
)
schedule_insertion_report_handler = workflow.execute_activity_method(
schedule_insertion_report_handler = workflow.start_activity_method(
Activities.create_schedules,
{
'schedules': schedule_actions['to_create']
@@ -138,7 +163,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60)
)
schedule_update_report_handler = workflow.execute_activity_method(
schedule_update_report_handler = workflow.start_activity_method(
Activities.update_schedules,
{
'schedules': schedule_actions['to_update']
@@ -153,7 +178,9 @@ class Orchestrator:
schedule_insertion_report = await schedule_insertion_report_handler
schedule_update_report = await schedule_update_report_handler
schedule_report_handler = workflow.execute_activity_method(
if schedule_insertion_report or schedule_update_report or schedule_deletion_report:
schedule_report_handler = workflow.start_activity_method(
Activities.report_schedule_orchestration,
{
'created_schedules': schedule_insertion_report,
@@ -164,7 +191,9 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60)
)
slot_report_handler = workflow.execute_activity_method(
if slot_insertion_report or slot_deletion_report:
slot_report_handler = workflow.start_activity_method(
Activities.report_slot_orchestration,
{
'inserted_slots': slot_insertion_report,
@@ -174,5 +203,50 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60)
)
if schedule_update_report:
update_pipelines_timestamps_handler = workflow.start_activity_method(
Activities.update_pipelines_timestamps,
{
'updated_pipelines': schedule_update_report
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
)
if schedule_insertion_report:
create_pipelines_timestamps_handler = workflow.start_activity_method(
Activities.create_pipelines_timestamps,
{
'created_pipelines': schedule_insertion_report
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
)
if schedule_deletion_report:
delete_pipelines_timestamps_handler = workflow.start_activity_method(
Activities.delete_pipelines_timestamps,
{
'deleted_pipelines': schedule_deletion_report
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
)
if schedule_insertion_report or schedule_update_report or schedule_deletion_report:
await schedule_report_handler
if slot_insertion_report or slot_deletion_report:
await slot_report_handler
if schedule_update_report:
await update_pipelines_timestamps_handler
if schedule_insertion_report:
await create_pipelines_timestamps_handler
if schedule_deletion_report:
await delete_pipelines_timestamps_handler

14
samples.json
View File

@@ -14,7 +14,7 @@
"read_tags": [
{
"tag_name": "Counter",
"server_id": "default_server",
"server_id": "1",
"aggr_func": "avg",
"tag_address": "ns=2;i=2",
"frequency": "15000",
@@ -25,7 +25,7 @@
},
{
"tag_name": "Rollout",
"server_id": "default_server",
"server_id": "1",
"aggr_func": "mdn",
"tag_address": "ns=2;i=3",
"frequency": "15000",
@@ -36,7 +36,7 @@
},
{
"tag_name": "Square",
"server_id": "default_server",
"server_id": "1",
"aggr_func": "lts",
"tag_address": "ns=2;i=4",
"frequency": "15000",
@@ -56,7 +56,9 @@
"policy": "DISCARD"
}
],
"tag_retention_minutes": 60
"tag_retention_minutes": 60,
"active": true,
"updated_at": "2025-07-14 10:00:00.000000"
},
"2": {
"schedule_name": "laborious-orchestrated-pipeline",
@@ -115,7 +117,9 @@
"STOP",
"CONTINUE",
"REPEAT"
]
],
"active": true,
"updated_at": "2025-07-14 10:00:00.000000"
}
},
"opc-servers": {

109
test.ipynb
View File

@@ -110,7 +110,7 @@
},
{
"cell_type": "code",
"execution_count": 14,
"execution_count": 3,
"id": "7d01f160",
"metadata": {},
"outputs": [],
@@ -128,9 +128,6 @@
" namespace=os.getenv('TEMPORAL_NAMESPACE', 'default')\n",
")\n",
"\n",
"manager = TemporalManager(temporal_client=temporal_client,\n",
" logger=logger,\n",
" notification_handler=MagicMock())\n",
" "
]
},
@@ -382,7 +379,7 @@
},
{
"cell_type": "code",
"execution_count": 17,
"execution_count": 4,
"id": "f81b3728",
"metadata": {},
"outputs": [
@@ -390,12 +387,112 @@
"name": "stdout",
"output_type": "stream",
"text": [
"{'_client': <temporalio.client.Client object at 0x770086544f90>, 'id': 'orchestrator'}\n"
"{'id': 'orchestrator', 'schedule': ScheduleListSchedule(action=ScheduleListActionStartWorkflow(workflow='orchestrator'), spec=ScheduleSpec(calendars=[], intervals=[ScheduleIntervalSpec(every=datetime.timedelta(seconds=3600), offset=None)], cron_expressions=[], skip=[], start_at=None, end_at=None, jitter=None, time_zone_name=None), state=ScheduleListState(note=None, paused=False)), 'info': ScheduleListInfo(recent_actions=[ScheduleActionResult(scheduled_at=datetime.datetime(2025, 7, 14, 9, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 7, 14, 9, 0, 0, 165083, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='orchestrator-2025-07-14T09:00:00Z', first_execution_run_id='01980829-9700-771a-9c9d-2450eecc9af4')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 7, 14, 10, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 7, 14, 10, 0, 0, 125309, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='orchestrator-2025-07-14T10:00:00Z', first_execution_run_id='01980860-8578-759d-97fa-dbdfc784f88a')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 7, 14, 11, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 7, 14, 11, 0, 0, 124356, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='orchestrator-2025-07-14T11:00:00Z', first_execution_run_id='01980897-73f7-7e83-a7a6-20eaac38e7d4')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 7, 14, 12, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 7, 14, 12, 0, 0, 167329, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='orchestrator-2025-07-14T12:00:00Z', first_execution_run_id='019808ce-62a2-75d6-b11d-c5387cfa258f')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 7, 14, 13, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 7, 14, 13, 0, 0, 167575, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='orchestrator-2025-07-14T13:00:00Z', first_execution_run_id='01980905-5122-7807-90ae-3782d0332e79'))], next_action_times=[datetime.datetime(2025, 7, 14, 14, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 7, 14, 15, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 7, 14, 16, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 7, 14, 17, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 7, 14, 18, 0, tzinfo=datetime.timezone.utc)]), 'typed_search_attributes': TypedSearchAttributes(search_attributes=[]), 'search_attributes': {}, 'data_converter': DataConverter(payload_converter_class=<class 'temporalio.converter.DefaultPayloadConverter'>, payload_codec=None, failure_converter_class=<class 'temporalio.converter.DefaultFailureConverter'>, payload_converter=<temporalio.converter.DefaultPayloadConverter object at 0x787d51390750>, failure_converter=<temporalio.converter.DefaultFailureConverter object at 0x787d5277b590>), 'raw_entry': schedule_id: \"orchestrator\"\n",
"info {\n",
" spec {\n",
" interval {\n",
" interval {\n",
" seconds: 3600\n",
" }\n",
" }\n",
" }\n",
" workflow_type {\n",
" name: \"orchestrator\"\n",
" }\n",
" recent_actions {\n",
" schedule_time {\n",
" seconds: 1752483600\n",
" }\n",
" actual_time {\n",
" seconds: 1752483600\n",
" nanos: 165083498\n",
" }\n",
" start_workflow_result {\n",
" workflow_id: \"orchestrator-2025-07-14T09:00:00Z\"\n",
" run_id: \"01980829-9700-771a-9c9d-2450eecc9af4\"\n",
" }\n",
" start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n",
" }\n",
" recent_actions {\n",
" schedule_time {\n",
" seconds: 1752487200\n",
" }\n",
" actual_time {\n",
" seconds: 1752487200\n",
" nanos: 125309119\n",
" }\n",
" start_workflow_result {\n",
" workflow_id: \"orchestrator-2025-07-14T10:00:00Z\"\n",
" run_id: \"01980860-8578-759d-97fa-dbdfc784f88a\"\n",
" }\n",
" start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n",
" }\n",
" recent_actions {\n",
" schedule_time {\n",
" seconds: 1752490800\n",
" }\n",
" actual_time {\n",
" seconds: 1752490800\n",
" nanos: 124356906\n",
" }\n",
" start_workflow_result {\n",
" workflow_id: \"orchestrator-2025-07-14T11:00:00Z\"\n",
" run_id: \"01980897-73f7-7e83-a7a6-20eaac38e7d4\"\n",
" }\n",
" start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n",
" }\n",
" recent_actions {\n",
" schedule_time {\n",
" seconds: 1752494400\n",
" }\n",
" actual_time {\n",
" seconds: 1752494400\n",
" nanos: 167329351\n",
" }\n",
" start_workflow_result {\n",
" workflow_id: \"orchestrator-2025-07-14T12:00:00Z\"\n",
" run_id: \"019808ce-62a2-75d6-b11d-c5387cfa258f\"\n",
" }\n",
" start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n",
" }\n",
" recent_actions {\n",
" schedule_time {\n",
" seconds: 1752498000\n",
" }\n",
" actual_time {\n",
" seconds: 1752498000\n",
" nanos: 167575982\n",
" }\n",
" start_workflow_result {\n",
" workflow_id: \"orchestrator-2025-07-14T13:00:00Z\"\n",
" run_id: \"01980905-5122-7807-90ae-3782d0332e79\"\n",
" }\n",
" start_workflow_status: WORKFLOW_EXECUTION_STATUS_RUNNING\n",
" }\n",
" future_action_times {\n",
" seconds: 1752501600\n",
" }\n",
" future_action_times {\n",
" seconds: 1752505200\n",
" }\n",
" future_action_times {\n",
" seconds: 1752508800\n",
" }\n",
" future_action_times {\n",
" seconds: 1752512400\n",
" }\n",
" future_action_times {\n",
" seconds: 1752516000\n",
" }\n",
"}\n",
"}\n",
"{'_client': <temporalio.client.Client object at 0x787d53dd3490>, 'id': 'orchestrator'}\n"
]
}
],
"source": [
"async for schedule in await temporal_client.list_schedules():\n",
" print(vars(schedule))\n",
" id = schedule.id\n",
"\n",
" handle = temporal_client.get_schedule_handle(id)\n",

2
tests/orchestrator/activities/test_couchbase.py
View File

@@ -25,7 +25,7 @@ def test_shutdown_failure(couchbase):
couchbase.cluster.close.side_effect = Exception("Test error")
couchbase.shutdown()
couchbase.logger.error.assert_called_once_with(
"Failed to close Couchbase connection: %s", couchbase.cluster.close.side_effect)
f"Failed to close Couchbase connection: {couchbase.cluster.close.side_effect}")
@mark.asyncio

157
tests/orchestrator/activities/test_formatters.py
View File

@@ -18,9 +18,9 @@ def formatters():
@mark.asyncio
@patch("orchestrator.activities.formatters.scouter",
return_value="test_scouter")
return_value={"test_scouter": "test_scouter"})
@patch("orchestrator.activities.formatters.predictions_batch",
return_value="test_predictions_batch")
return_value={"test_predictions_batch": "test_predictions_batch"})
async def test_process_schedules(mock_predictions_batch, mock_scouter, formatters):
input_data = {
"pipelines": [
@@ -28,13 +28,15 @@ async def test_process_schedules(mock_predictions_batch, mock_scouter, formatter
"schedule_name": "test_schedule_name",
"workflow_type": "scouter",
"model_name": "test_model_name",
"model_id": "test_model_id"
"model_id": "test_model_id",
"updated_at": "2021-01-01"
},
{
"schedule_name": "test_schedule_name2",
"workflow_type": "predictions_batch",
"model_name": "test_model_name",
"model_id": "test_model_id"
"model_id": "test_model_id",
"updated_at": "2021-01-02"
}
]
}
@@ -43,10 +45,16 @@ async def test_process_schedules(mock_predictions_batch, mock_scouter, formatter
assert result == {
"scouter": {
"test_schedule_name": "test_scouter"
"test_schedule_name": {
"test_scouter": "test_scouter",
"updated_at": "2021-01-01"
}
},
"laborious": {
"test_schedule_name2": "test_predictions_batch"
"test_schedule_name2": {
"test_predictions_batch": "test_predictions_batch",
"updated_at": "2021-01-02"
}
}
}
@@ -247,32 +255,51 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma
}
@mark.asyncio
async def test_format_schedule_config(formatters):
input_data = {
"schedule_config": [
{"namespace": "test_namespace1", "schedule_name": "test1",
"updated_at": "2021-01-01"},
{"namespace": "test_namespace2", "schedule_name": "test2",
"updated_at": "2021-01-02"}
]
}
result = await formatters.format_schedule_config(input_data)
assert result == {
"test_namespace1": {
"test1": "2021-01-01"
},
"test_namespace2": {
"test2": "2021-01-02"
}
}
@mark.asyncio
async def test_create_schedule_config(formatters):
input_data = {
"current_schedule_config": {
"scouter": {
"test_schedule_name_to_delete": {
"frequency": 60,
"data": {"test": "test"}
},
"test_schedule_name_to_update": {
"frequency": 60,
"data": {"test": "test"}
}
"test_schedule_name_to_delete": '2021-01-01',
"test_schedule_name_to_update": '2021-01-02'
}
},
"schedule_config": {
"laborious": {
"test_schedule_name_to_create": {
"frequency": 60,
"data": {"test": "test"}
"data": {"test": "test"},
"updated_at": "2021-01-03"
}
},
"scouter": {
"test_schedule_name_to_update": {
"frequency": 60,
"data": {"test": "test2"}
"data": {"test": "test2"},
"updated_at": "2021-01-04"
}
}
}
@@ -285,7 +312,8 @@ async def test_create_schedule_config(formatters):
"laborious": {
"test_schedule_name_to_create": {
"frequency": 60,
"data": {"test": "test"}
"data": {"test": "test"},
"updated_at": "2021-01-03"
}
},
"scouter": {}
@@ -294,7 +322,8 @@ async def test_create_schedule_config(formatters):
"scouter": {
"test_schedule_name_to_update": {
"frequency": 60,
"data": {"test": "test2"}
"data": {"test": "test2"},
"updated_at": "2021-01-04"
}
},
"laborious": {}
@@ -369,95 +398,129 @@ def test_send_error_report(formatters):
def test_parse_report(formatters):
input_data = {
"test_schedule_name_to_create": {
"test_key": {
"success": True
},
"test_schedule_name_to_create_error": {
"test_key2": {
"success": False,
"error": "test_error"
"message": "test_error"
}
}
result = formatters.parse_report(input_data)
assert result == (
["test_schedule_name_to_create"],
["test_schedule_name_to_create_error"]
["test_key"],
["test_key2"]
)
def test_parse_report_schedule(formatters):
input_data = [
{
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_create",
"success": True
},
{
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_create_error",
"success": False,
"message": "test_error"
}
]
result = formatters.parse_report_schedule(input_data)
assert result == (
["test_namespace/test_schedule_name_to_create"],
["test_namespace/test_schedule_name_to_create_error: test_error"]
)
@mark.asyncio
async def test_report_schedule_orchestration(formatters):
formatters.parse_report = MagicMock(
side_effect=formatters.parse_report
formatters.parse_report_schedule = MagicMock(
side_effect=formatters.parse_report_schedule
)
formatters.send_success_report = MagicMock()
formatters.send_error_report = MagicMock()
input_data = {
"created_schedules": {
"test_schedule_name_to_create": {
"created_schedules": [
{
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_create",
"success": True
},
"test_schedule_name_to_create_error": {
{
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_create_error",
"success": False,
"error": "test_error"
"message": "test_error"
}
},
"updated_schedules": {
"test_schedule_name_to_update": {
],
"updated_schedules": [
{
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_update",
"success": True
},
"test_schedule_name_to_update_error": {
{
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_update_error",
"success": False,
"error": "test_error"
"message": "test_error"
}
},
"deleted_schedules": {
"test_schedule_name_to_delete": {
],
"deleted_schedules": [
{
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_delete",
"success": True
},
"test_schedule_name_to_delete_error": {
{
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_delete_error",
"success": False,
"error": "test_error"
}
"message": "test_error"
}
]
}
await formatters.report_schedule_orchestration(input_data)
formatters.parse_report.assert_has_calls([
formatters.parse_report_schedule.assert_has_calls([
call(input_data['created_schedules']),
call(input_data['updated_schedules']),
call(input_data['deleted_schedules'])
])
formatters.send_success_report.assert_has_calls([
call(
"Created schedules: \n test_schedule_name_to_create",
"Created schedules: \n test_namespace/test_schedule_name_to_create",
"REPORT_ORCHESTRATION_CREATED_SCHEDULES"
),
call(
"Updated schedules: \n test_schedule_name_to_update",
"Updated schedules: \n test_namespace/test_schedule_name_to_update",
"REPORT_ORCHESTRATION_UPDATED_SCHEDULES"
),
call(
"Deleted schedules: \n test_schedule_name_to_delete",
"Deleted schedules: \n test_namespace/test_schedule_name_to_delete",
"REPORT_ORCHESTRATION_DELETED_SCHEDULES"
)
])
formatters.send_error_report.assert_has_calls([
call(
"Failed to create schedules: \n test_schedule_name_to_create_error",
"Failed to create schedules: \n test_namespace/test_schedule_name_to_create_error: test_error",
"REPORT_ORCHESTRATION_CREATED_SCHEDULES",
input_data['created_schedules']
),
call(
"Failed to update schedules: \n test_schedule_name_to_update_error",
"Failed to update schedules: \n test_namespace/test_schedule_name_to_update_error: test_error",
"REPORT_ORCHESTRATION_UPDATED_SCHEDULES",
input_data['updated_schedules']
),
call(
"Failed to delete schedules: \n test_schedule_name_to_delete_error",
"Failed to delete schedules: \n test_namespace/test_schedule_name_to_delete_error: test_error",
"REPORT_ORCHESTRATION_DELETED_SCHEDULES",
input_data['deleted_schedules']
)

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

@@ -244,3 +244,58 @@ async def test_aggregate_documents_in_mongodb_missing_aggregation(mongo_db):
else:
assert False, "Expected a ValueError to be raised"
@mark.asyncio
@patch("orchestrator.activities.mongo_db.datetime")
async def test_update_pipelines_timestamps_success(datetime_mock, mongo_db):
input_data = {"updated_pipelines": [
{"schedule_name": "test1", "namespace": "test1", "success": True},
{"schedule_name": "test2", "namespace": "test2", "success": True}
]}
mongo_db.database["pipelines"].update_many.return_value = MagicMock()
await mongo_db.update_pipelines_timestamps(input_data)
mongo_db.database["pipelines"].update_many.assert_called_once_with(
{"$or": [
{"schedule_name": "test1", "namespace": "test1"},
{"schedule_name": "test2", "namespace": "test2"}
]},
{"$set": {
"updated_at": datetime_mock.now.return_value.strftime.return_value}}
)
@mark.asyncio
@patch("orchestrator.activities.mongo_db.datetime")
async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db):
input_data = {"created_pipelines": [
{"schedule_name": "test1", "namespace": "test1", "success": True},
{"schedule_name": "test2", "namespace": "test2", "success": True}
]}
mongo_db.database["pipelines"].insert_many.return_value = MagicMock()
await mongo_db.create_pipelines_timestamps(input_data)
mongo_db.database["pipelines"].insert_many.assert_called_once_with(
[
{"schedule_name": "test1", "namespace": "test1",
"updated_at": datetime_mock.now.return_value.strftime.return_value},
{"schedule_name": "test2", "namespace": "test2",
"updated_at": datetime_mock.now.return_value.strftime.return_value}
]
)
@mark.asyncio
@patch("orchestrator.activities.mongo_db.datetime")
async def test_delete_pipelines_timestamps_success(datetime_mock, mongo_db):
input_data = {"deleted_pipelines": [
{"schedule_name": "test1", "namespace": "test1", "success": True},
{"schedule_name": "test2", "namespace": "test2", "success": True}
]}
mongo_db.database["pipelines"].delete_many.return_value = MagicMock()
await mongo_db.delete_pipelines_timestamps(input_data)
mongo_db.database["pipelines"].delete_many.assert_called_once_with(
{"$or": [
{"schedule_name": "test1", "namespace": "test1"},
{"schedule_name": "test2", "namespace": "test2"}
]}
)

173
tests/orchestrator/activities/test_temporal_manager.py
View File

@@ -41,12 +41,7 @@ async def test_connect_to_temporal(connect_mock, temporal_manager):
])
@mark.asyncio
@patch("orchestrator.activities.temporal_manager.MessageToDict",
return_value={"data": base64.b64encode(json.dumps({"test": "test"}).encode('utf-8'))})
async def test_load_schedule(_mock_message_to_dict, temporal_manager):
# Create async iterator mock
async def async_iter():
async def async_iter():
yield MagicMock(
id="test-schedule-id",
search_attributes={
@@ -66,6 +61,13 @@ async def test_load_schedule(_mock_message_to_dict, temporal_manager):
}
)
@mark.asyncio
@patch("orchestrator.activities.temporal_manager.MessageToDict",
return_value={"data": base64.b64encode(json.dumps({"test": "test"}).encode('utf-8'))})
async def test_load_schedule(_mock_message_to_dict, temporal_manager):
# Create async iterator mock
handle = MagicMock(
describe=AsyncMock(
return_value=MagicMock(
@@ -134,6 +136,43 @@ async def test_load_schedule(_mock_message_to_dict, temporal_manager):
}
@mark.asyncio
async def test_normalize_schedules(temporal_manager):
input_data = {
"orchestrated_schedules": {
"scouter": {"test-scouter": "2021-01-01"},
"laborious": {"test-schedule-id1": "2021-01-01"}
}
}
temporal_manager.temporal_clients['scouter'].list_schedules = AsyncMock(
return_value=async_iter()
)
temporal_manager.temporal_clients['laborious'].list_schedules = AsyncMock(
return_value=async_iter()
)
temporal_manager.temporal_clients['scouter'].get_schedule_handle = MagicMock(
return_value=MagicMock(
delete=AsyncMock()
)
)
temporal_manager.temporal_clients['laborious'].get_schedule_handle = MagicMock(
return_value=MagicMock(
delete=AsyncMock()
)
)
await temporal_manager.normalize_schedules(input_data)
temporal_manager.temporal_clients['scouter'].get_schedule_handle.assert_has_calls([
call("test-schedule-id"),
])
temporal_manager.temporal_clients['scouter'].get_schedule_handle.return_value.delete.assert_awaited_once(
)
@mark.asyncio
@patch("orchestrator.activities.temporal_manager.parse_frequency",
side_effect=parse_frequency)
@@ -308,20 +347,26 @@ async def test_create_schedule(
)
])
assert report == {
"test-schedule": {
assert report == [
{
"schedule_name": "test-schedule",
"namespace": "scouter",
"success": True,
"message": "Schedule created successfully"
},
"test-schedule-invalid-frequency": {
{
"schedule_name": "test-schedule-invalid-frequency",
"namespace": "scouter",
"success": False,
"message": "Invalid frequency"
},
"test-schedule-laborious": {
{
"schedule_name": "test-schedule-laborious",
"namespace": "laborious",
"success": True,
"message": "Schedule created successfully"
}
}
]
@mark.asyncio
@@ -397,29 +442,72 @@ async def test_update_schedules(
}
}
handler_scouter = MagicMock(
update=AsyncMock(
update=AsyncMock(
side_effect=lambda f: f(input_mock)
)
)
)
handler_laborious = MagicMock(
update=AsyncMock(
update=AsyncMock(
side_effect=lambda f: f(input_mock)
)
)
)
temporal_manager.temporal_clients['scouter'].get_schedule_handle = MagicMock(
side_effect=[
handler_scouter,
None
]
)
temporal_manager.temporal_clients['laborious'].get_schedule_handle = MagicMock(
side_effect=[
handler_laborious
]
)
report = await temporal_manager.update_schedules(input_data)
temporal_manager.schedule_handles['scouter']['test-schedule'].update.assert_called_once()
temporal_manager.schedule_handles['laborious']['test-schedule-laborious'].update.assert_called_once()
temporal_manager.temporal_clients['scouter'].get_schedule_handle.assert_has_calls([
call("test-schedule"),
call("test-schedule_no_handler")
])
temporal_manager.temporal_clients['laborious'].get_schedule_handle.assert_has_calls([
call("test-schedule-laborious")
])
assert report == {
"test-schedule": {
handler_scouter.update.assert_called_once()
handler_laborious.update.assert_called_once()
assert report == [
{
"schedule_name": "test-schedule",
"namespace": "scouter",
"success": True,
"message": "Schedule updated successfully"
},
"test-schedule_no_handler": {
{
"schedule_name": "test-schedule_no_handler",
"namespace": "scouter",
"success": False,
"message": "Schedule test-schedule_no_handler not found"
},
"test-schedule-laborious": {
{
"schedule_name": "test-schedule-laborious",
"namespace": "laborious",
"success": True,
"message": "Schedule updated successfully"
}
}
]
@mark.asyncio
async def test_update_schedules_with_no_handle(temporal_manager):
async def test_update_schedules_with_no_client(temporal_manager):
temporal_manager.temporal_clients = {}
input_data = {
"schedules": {
@@ -436,7 +524,7 @@ async def test_update_schedules_with_no_handle(temporal_manager):
await temporal_manager.update_schedules(input_data)
except Exception as e:
assert str(
e) == f"Schedule handles for abc not found, handles: {temporal_manager.schedule_handles}"
e) == f"Temporal client for abc not found, clients: {temporal_manager.temporal_clients}"
@mark.asyncio
@@ -464,28 +552,57 @@ async def test_delete_schedules(temporal_manager):
]
}
}
handler_scouter = MagicMock(
delete=AsyncMock()
)
handler_laborious = MagicMock(
delete=AsyncMock()
)
temporal_manager.temporal_clients['scouter'].get_schedule_handle = MagicMock(
side_effect=[
handler_scouter,
None
]
)
temporal_manager.temporal_clients['laborious'].get_schedule_handle = MagicMock(
side_effect=[
handler_laborious
]
)
report = await temporal_manager.delete_schedules(input_data)
assert report == {
"test-schedule": {
handler_scouter.delete.assert_called_once()
handler_laborious.delete.assert_called_once()
assert report == [
{
"schedule_name": "test-schedule",
"namespace": "scouter",
"success": True,
"message": "Schedule deleted successfully"
},
"test-schedule_no_handler": {
{
"schedule_name": "test-schedule_no_handler",
"namespace": "scouter",
"success": False,
"message": "Schedule test-schedule_no_handler not found"
},
"test-schedule-laborious": {
{
"schedule_name": "test-schedule-laborious",
"namespace": "laborious",
"success": True,
"message": "Schedule deleted successfully"
}
}
]
@mark.asyncio
async def test_delete_schedules_with_no_handle(temporal_manager):
temporal_manager.schedule_handles = {}
async def test_delete_schedules_with_no_client(temporal_manager):
temporal_manager.temporal_clients = {}
input_data = {
"schedules": {
"abc": [
@@ -498,4 +615,4 @@ async def test_delete_schedules_with_no_handle(temporal_manager):
await temporal_manager.delete_schedules(input_data)
except Exception as e:
assert str(
e) == f"Schedule handles for abc not found, handles: {temporal_manager.schedule_handles}"
e) == f"Temporal client for abc not found, clients: {temporal_manager.temporal_clients}"

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

@@ -36,8 +36,8 @@ def test_build_redis_config_with_defaults():
assert build_redis_config() == {
'host': 'localhost',
'port': 6379,
'username': None,
'password': None
'username': 'default',
'password': 'bdnZOpcyiL'
}
@@ -71,7 +71,7 @@ def test_build_mongo_db_config_with_defaults():
environ.pop('MONGODB_URL', None)
assert build_mongodb_config() == {
'connection_string': 'mongodb://sientia:sientia@localhost:27017',
'connection_string': 'mongodb://root:wKZDbMNU1c@localhost:27018',
'database_name': 'sientia'
}

141
tests/orchestrator/workflows/test_orchestrator.py
View File

@@ -20,7 +20,7 @@ async def test_run(workflow_mock, orchestrator):
await orchestrator.run(input_data)
workflow_mock.execute_local_activity_method.assert_has_calls([
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.aggregate_documents_in_mongodb,
{
@@ -31,7 +31,7 @@ async def test_run(workflow_mock, orchestrator):
)
])
workflow_mock.execute_local_activity_method.assert_has_calls([
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.find_documents_in_mongodb,
{
@@ -42,15 +42,20 @@ async def test_run(workflow_mock, orchestrator):
)
])
workflow_mock.execute_local_activity_method.assert_has_calls([
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.load_schedule,
Activities.find_documents_in_mongodb,
{
"query": {
"collection": "orchestrated_schedules"
}
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.execute_local_activity_method.assert_has_calls([
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.load_opc_slots,
retry_policy=ANY,
@@ -58,7 +63,7 @@ async def test_run(workflow_mock, orchestrator):
)
])
workflow_mock.execute_local_activity_method.assert_has_calls([
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.load_active_ingestors,
retry_policy=ANY,
@@ -66,108 +71,188 @@ async def test_run(workflow_mock, orchestrator):
)
])
workflow_mock.execute_local_activity_method.assert_has_calls([
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.format_schedule_config,
{
'schedule_config': workflow_mock.start_local_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.process_schedules,
{
'pipelines': workflow_mock.execute_local_activity_method.return_value
'pipelines': workflow_mock.start_local_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.execute_local_activity_method.assert_has_calls([
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.process_slots,
{
'opc_servers': workflow_mock.execute_local_activity_method.return_value,
'active_ingestors': workflow_mock.execute_local_activity_method.return_value,
'pipelines': workflow_mock.execute_local_activity_method.return_value,
'opc_servers': workflow_mock.start_local_activity_method.return_value,
'active_ingestors': workflow_mock.start_local_activity_method.return_value,
'pipelines': workflow_mock.start_local_activity_method.return_value,
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.execute_local_activity_method.assert_has_calls([
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.create_schedule_config,
{
'current_schedule_config': workflow_mock.execute_local_activity_method.return_value,
'schedule_config': workflow_mock.execute_local_activity_method.return_value
'current_schedule_config': workflow_mock.start_local_activity_method.return_value,
'schedule_config': workflow_mock.start_local_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.execute_local_activity_method.assert_has_calls([
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.create_slot_config,
{
'current_slot_config': workflow_mock.execute_local_activity_method.return_value,
'slot_config': workflow_mock.execute_local_activity_method.return_value
'current_slot_config': workflow_mock.start_local_activity_method.return_value,
'slot_config': workflow_mock.start_local_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.execute_activity_method.assert_has_calls([
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.normalize_schedules,
{
'orchestrated_schedules': workflow_mock.start_local_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.delete_slots,
{
'to_delete':
workflow_mock.execute_local_activity_method.return_value['to_delete']
workflow_mock.start_local_activity_method.return_value['to_delete']
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.execute_activity_method.assert_has_calls([
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.update_slots,
{
'to_insert':
workflow_mock.execute_local_activity_method.return_value['to_insert']
workflow_mock.start_local_activity_method.return_value['to_insert']
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.execute_activity_method.assert_has_calls([
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.delete_schedules,
{
'schedules':
workflow_mock.execute_local_activity_method.return_value['to_delete']
workflow_mock.start_local_activity_method.return_value['to_delete']
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.execute_activity_method.assert_has_calls([
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.create_schedules,
{
'schedules':
workflow_mock.execute_local_activity_method.return_value['to_create']
workflow_mock.start_local_activity_method.return_value['to_create']
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.execute_activity_method.assert_has_calls([
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.update_schedules,
{
'schedules':
workflow_mock.execute_local_activity_method.return_value['to_update']
workflow_mock.start_local_activity_method.return_value['to_update']
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.report_schedule_orchestration,
{
'created_schedules': workflow_mock.start_activity_method.return_value,
'updated_schedules': workflow_mock.start_activity_method.return_value,
'deleted_schedules': workflow_mock.start_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.report_slot_orchestration,
{
'inserted_slots': workflow_mock.start_activity_method.return_value,
'deleted_slots': workflow_mock.start_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.update_pipelines_timestamps,
{
'updated_pipelines': workflow_mock.start_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.delete_pipelines_timestamps,
{
'deleted_pipelines': workflow_mock.start_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.create_pipelines_timestamps,
{
'created_pipelines': workflow_mock.start_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY

15
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.2.0"
tag: "0.2.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:
@@ -111,11 +111,20 @@ tolerations: []
affinity: {}
service:
services:
api:
enabled: false
type: ClusterIP
port: 4841
targetPort: 4841
name: api
opc:
enabled: false
type: ClusterIP
port: 4840
targetPort: 4840
name: server
env:
@@ -123,7 +132,7 @@ env:
- name: GITHUB_REPO_URL
value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
- name: GITHUB_BRANCH
value: "SIENTIAPDE-1148-separar-scouter-laborious-e-orchestrator-por-namespaces"
value: "SIENTIAPDE-1150-ajustar-orquestrador-para-manter-uma-store-de-detalhes-dos-schedules"
- name: PYTHON_APP
value: "orchestrator.worker.worker"