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

Merge pull request #8 from Aignosi/SIENTIAPDE-1163-alterar-dinamica-de-notificacoes-para-usar-o-mongodb-ao-inves-do-kafka

Browse Source 
Sientiapde 1163 alterar dinamica de notificacoes para usar o mongodb ao inves do kafka
This commit is contained in:
Bruno Domingues
2025-07-22 10:14:49 -03:00
committed by GitHub
parent 63fa276e89 7c92bd2584
commit d81e8ea33f
17 changed files with 670 additions and 385 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@@ -41,4 +41,6 @@ htmlcov/
.coverage .coverage
# git keys # git keys
git_key* git_key*
git_log

2
orchestrator/activities/activities.py
View File

@@ -10,7 +10,7 @@ with workflow.unsafe.imports_passed_through():
from orchestrator.activities.formatters import Formatters from orchestrator.activities.formatters import Formatters
from typing import Any from typing import Any
from logging import Logger from logging import Logger
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
class Activities( # Couchbase, class Activities( # Couchbase,

2
orchestrator/activities/couchbase.py
View File

@@ -9,7 +9,7 @@ with workflow.unsafe.imports_passed_through():
from couchbase.auth import PasswordAuthenticator from couchbase.auth import PasswordAuthenticator
from couchbase.cluster import Cluster from couchbase.cluster import Cluster
from couchbase.options import ClusterOptions from couchbase.options import ClusterOptions
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.activities.base import BaseActivity from sientia_do.temporal.activities.base import BaseActivity

144
orchestrator/activities/formatters.py
View File

@@ -7,7 +7,7 @@ with workflow.unsafe.imports_passed_through():
from typing import Any from typing import Any
from logging import Logger from logging import Logger
from datetime import datetime from datetime import datetime
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.temporal.activities.base import BaseActivity from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.notifications.models import NotificationLevel from sientia_do.notifications.models import NotificationLevel
from orchestrator.utils.orchestrator_functions import ( from orchestrator.utils.orchestrator_functions import (
@@ -42,7 +42,9 @@ class Formatters(BaseActivity):
- dict[str, Any]: The schedule config dictionary - dict[str, Any]: The schedule config dictionary
""" """
self.logger.info("Processing schedules...") metadata = input_data.get("metadata", {})
self.info("Processing schedules...", metadata=metadata)
pipelines = input_data['pipelines'] pipelines = input_data['pipelines']
@@ -66,10 +68,9 @@ class Formatters(BaseActivity):
"updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT)) "updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT))
} }
self.logger.info("Processed schedules") self.info("Processed schedules", metadata=metadata)
self.debug(json.dumps(
self.logger.debug(json.dumps( schedule_config, indent=4, sort_keys=True), metadata=metadata)
schedule_config, indent=4, sort_keys=True))
return schedule_config return schedule_config
@@ -91,7 +92,9 @@ class Formatters(BaseActivity):
- dict[str, Any]: The slot config dictionary - dict[str, Any]: The slot config dictionary
""" """
self.logger.info("Processing slots...") metadata = input_data.get("metadata", {})
self.info("Processing slots...", metadata=metadata)
pipelines = input_data['pipelines'] pipelines = input_data['pipelines']
opc_servers_list = input_data['opc_servers'] opc_servers_list = input_data['opc_servers']
@@ -124,9 +127,9 @@ class Formatters(BaseActivity):
slot_config = build_tag_config( slot_config = build_tag_config(
tag, slot_config.copy(), opc_servers, number_of_slots) tag, slot_config.copy(), opc_servers, number_of_slots)
self.logger.info("Processed slots") self.info("Processed slots", metadata=metadata)
self.logger.debug(json.dumps( self.debug(json.dumps(
slot_config, indent=4, sort_keys=True)) slot_config, indent=4, sort_keys=True), metadata=metadata)
return slot_config return slot_config
@@ -158,7 +161,7 @@ class Formatters(BaseActivity):
def compare_config_timestamps(self, def compare_config_timestamps(self,
schedules: dict[str, Any], current_schedules: dict[str, Any], schedules: dict[str, Any], current_schedules: dict[str, Any],
to_update: dict[str, Any], to_create: dict[str, Any], to_update: dict[str, Any], to_create: dict[str, Any],
namespace: str): namespace: str, metadata: dict[str, Any]):
""" """
Compares the timestamps of the schedule and the current schedule. Compares the timestamps of the schedule and the current schedule.
""" """
@@ -170,8 +173,8 @@ class Formatters(BaseActivity):
old_timestamp = current_schedules[schedule_name] old_timestamp = current_schedules[schedule_name]
self.logger.debug( self.debug(
f"Comparing schedule {schedule_name}:{update_timestamp} vs {old_timestamp}") f"Comparing schedule {schedule_name}:{update_timestamp} vs {old_timestamp}", metadata=metadata)
if update_timestamp > old_timestamp: if update_timestamp > old_timestamp:
to_update[namespace][schedule_name] = schedule to_update[namespace][schedule_name] = schedule
@@ -197,7 +200,9 @@ class Formatters(BaseActivity):
- dict[str, Any]: The schedule config dictionary - dict[str, Any]: The schedule config dictionary
""" """
self.logger.info("Creating schedule config...") metadata = input_data.get("metadata", {})
self.info("Creating schedule config...", metadata=metadata)
current_schedule_config = input_data['current_schedule_config'] current_schedule_config = input_data['current_schedule_config']
schedule_config = input_data['schedule_config'] schedule_config = input_data['schedule_config']
@@ -218,7 +223,7 @@ class Formatters(BaseActivity):
for namespace, schedules in schedule_config.items(): for namespace, schedules in schedule_config.items():
current_schedules = current_schedule_config.get(namespace, {}) current_schedules = current_schedule_config.get(namespace, {})
self.compare_config_timestamps( self.compare_config_timestamps(
schedules, current_schedules, to_update, to_create, namespace) schedules, current_schedules, to_update, to_create, namespace, metadata)
for namespace, schedules in current_schedule_config.items(): for namespace, schedules in current_schedule_config.items():
for schedule_name in schedules: for schedule_name in schedules:
@@ -231,9 +236,9 @@ class Formatters(BaseActivity):
"to_delete": to_delete "to_delete": to_delete
} }
self.logger.info("Created schedule config") self.info("Created schedule config", metadata=metadata)
self.logger.debug(json.dumps( self.debug(json.dumps(
output, indent=4, sort_keys=True)) output, indent=4, sort_keys=True), metadata=metadata)
return output return output
@@ -256,7 +261,9 @@ class Formatters(BaseActivity):
- dict[str, Any]: The slot config dictionary - dict[str, Any]: The slot config dictionary
""" """
self.logger.info("Creating slot config...") metadata = input_data.get("metadata", {})
self.info("Creating slot config...", metadata=metadata)
current_slot_config = input_data['current_slot_config'] current_slot_config = input_data['current_slot_config']
slot_config = input_data['slot_config'] slot_config = input_data['slot_config']
@@ -274,27 +281,29 @@ class Formatters(BaseActivity):
"to_insert": slot_config "to_insert": slot_config
} }
self.logger.info("Created slot config") self.info("Created slot config", metadata=metadata)
self.logger.debug(json.dumps( self.debug(json.dumps(
output, indent=4, sort_keys=True)) output, indent=4, sort_keys=True), metadata=metadata)
return output return output
def send_success_report(self, message: str, notification_id: str) -> None: def send_success_report(self, metadata: dict[str, Any], message: str, notification_id: str) -> None:
self.notification_handler.build_and_send_notification( self.send_notification(
notification_id, metadata=metadata,
message, notification_id=notification_id,
"report_orchestration", message=message,
NotificationLevel.INFO block="report_orchestration",
level=NotificationLevel.INFO
) )
def send_error_report(self, message: str, notification_id: str, def send_error_report(self, metadata: dict[str, Any], message: str, notification_id: str,
attachment: dict[str, Any]) -> None: attachment: dict[str, Any]) -> None:
self.notification_handler.build_and_send_notification( self.send_notification(
notification_id, metadata=metadata,
message, notification_id=notification_id,
"report_orchestration", message=message,
NotificationLevel.ERROR, block="report_orchestration",
level=NotificationLevel.ERROR,
attachment_content=json.dumps(attachment, indent=4, sort_keys=True) attachment_content=json.dumps(attachment, indent=4, sort_keys=True)
) )
@@ -329,8 +338,9 @@ class Formatters(BaseActivity):
- updated_schedules (dict[str, Any]): The updated schedules. - updated_schedules (dict[str, Any]): The updated schedules.
- deleted_schedules (list[str]): The deleted schedules. - deleted_schedules (list[str]): The deleted schedules.
""" """
metadata = input_data.get("metadata", {})
self.logger.info("Reporting orchestration...") self.info("Reporting orchestration...", metadata=metadata)
created_schedules = input_data['created_schedules'] created_schedules = input_data['created_schedules']
updated_schedules = input_data['updated_schedules'] updated_schedules = input_data['updated_schedules']
@@ -343,15 +353,17 @@ class Formatters(BaseActivity):
if len(success_keys) > 0: if len(success_keys) > 0:
self.send_success_report( self.send_success_report(
f"Created schedules: \n {', '.join(success_keys)}", metadata=metadata,
"REPORT_ORCHESTRATION_CREATED_SCHEDULES" message=f"Created schedules: \n {', '.join(success_keys)}",
notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES"
) )
if len(error_keys) > 0: if len(error_keys) > 0:
self.send_error_report( self.send_error_report(
f"Failed to create schedules: \n {', '.join(error_keys)}", metadata=metadata,
"REPORT_ORCHESTRATION_CREATED_SCHEDULES", message=f"Failed to create schedules: \n {', '.join(error_keys)}",
created_schedules notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES",
attachment=created_schedules
) )
# Send report for updated schedules # Send report for updated schedules
@@ -361,15 +373,17 @@ class Formatters(BaseActivity):
if len(success_keys) > 0: if len(success_keys) > 0:
self.send_success_report( self.send_success_report(
f"Updated schedules: \n {', '.join(success_keys)}", metadata=metadata,
"REPORT_ORCHESTRATION_UPDATED_SCHEDULES" message=f"Updated schedules: \n {', '.join(success_keys)}",
notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES"
) )
if len(error_keys) > 0: if len(error_keys) > 0:
self.send_error_report( self.send_error_report(
f"Failed to update schedules: \n {', '.join(error_keys)}", metadata=metadata,
"REPORT_ORCHESTRATION_UPDATED_SCHEDULES", message=f"Failed to update schedules: \n {', '.join(error_keys)}",
updated_schedules notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES",
attachment=updated_schedules
) )
if len(deleted_schedules) > 0: if len(deleted_schedules) > 0:
@@ -378,15 +392,17 @@ class Formatters(BaseActivity):
if len(success_keys) > 0: if len(success_keys) > 0:
self.send_success_report( self.send_success_report(
f"Deleted schedules: \n {', '.join(success_keys)}", metadata=metadata,
"REPORT_ORCHESTRATION_DELETED_SCHEDULES" message=f"Deleted schedules: \n {', '.join(success_keys)}",
notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES"
) )
if len(error_keys) > 0: if len(error_keys) > 0:
self.send_error_report( self.send_error_report(
f"Failed to delete schedules: \n {', '.join(error_keys)}", metadata=metadata,
"REPORT_ORCHESTRATION_DELETED_SCHEDULES", message=f"Failed to delete schedules: \n {', '.join(error_keys)}",
deleted_schedules notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES",
attachment=deleted_schedules
) )
@activity.defn(name="report_slot_orchestration") @activity.defn(name="report_slot_orchestration")
@@ -402,7 +418,9 @@ class Formatters(BaseActivity):
- deleted_slots (list[str]): The deleted slots. - deleted_slots (list[str]): The deleted slots.
""" """
self.logger.info("Reporting orchestration...") metadata = input_data.get("metadata", {})
self.info("Reporting orchestration...", metadata=metadata)
inserted_slots = input_data['inserted_slots'] inserted_slots = input_data['inserted_slots']
deleted_slots = input_data['deleted_slots'] deleted_slots = input_data['deleted_slots']
@@ -412,15 +430,17 @@ class Formatters(BaseActivity):
if len(success_keys) > 0: if len(success_keys) > 0:
self.send_success_report( self.send_success_report(
f"Inserted slots: \n {', '.join(success_keys)}", metadata=metadata,
"REPORT_ORCHESTRATION_INSERTED_SLOTS" message=f"Inserted slots: \n {', '.join(success_keys)}",
notification_id="REPORT_ORCHESTRATION_INSERTED_SLOTS"
) )
if len(error_keys) > 0: if len(error_keys) > 0:
self.send_error_report( self.send_error_report(
f"Failed to insert slots: \n {', '.join(error_keys)}", metadata=metadata,
"REPORT_ORCHESTRATION_INSERTED_SLOTS", message=f"Failed to insert slots: \n {', '.join(error_keys)}",
inserted_slots notification_id="REPORT_ORCHESTRATION_INSERTED_SLOTS",
attachment=inserted_slots
) )
if len(deleted_slots) > 0: if len(deleted_slots) > 0:
@@ -428,13 +448,15 @@ class Formatters(BaseActivity):
if len(success_keys) > 0: if len(success_keys) > 0:
self.send_success_report( self.send_success_report(
f"Deleted slots: \n {', '.join(success_keys)}", metadata=metadata,
"REPORT_ORCHESTRATION_DELETED_SLOTS" message=f"Deleted slots: \n {', '.join(success_keys)}",
notification_id="REPORT_ORCHESTRATION_DELETED_SLOTS"
) )
if len(error_keys) > 0: if len(error_keys) > 0:
self.send_error_report( self.send_error_report(
f"Failed to delete slots: \n {', '.join(error_keys)}", metadata=metadata,
"REPORT_ORCHESTRATION_DELETED_SLOTS", message=f"Failed to delete slots: \n {', '.join(error_keys)}",
deleted_slots notification_id="REPORT_ORCHESTRATION_DELETED_SLOTS",
attachment=deleted_slots
) )

96
orchestrator/activities/mongo_db.py
View File

@@ -7,7 +7,7 @@ with workflow.unsafe.imports_passed_through():
import traceback import traceback
from logging import Logger from logging import Logger
from pymongo import MongoClient from pymongo import MongoClient
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.activities.base import BaseActivity from sientia_do.temporal.activities.base import BaseActivity
from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT
@@ -92,6 +92,7 @@ class MongoDB(BaseActivity):
""" """
query = input_data.get("query", {}) query = input_data.get("query", {})
metadata = input_data.get("metadata", {})
collection_name = query.get("collection") collection_name = query.get("collection")
if not collection_name: if not collection_name:
@@ -99,8 +100,8 @@ class MongoDB(BaseActivity):
filters = query.get("filters", {}) filters = query.get("filters", {})
self.logger.info( self.info(
f"Loading documents from collection '{collection_name}' with filters: {filters}") f"Loading documents from collection '{collection_name}' with filters: {filters}", metadata=metadata)
try: try:
collection = self.database[collection_name] collection = self.database[collection_name]
@@ -109,25 +110,25 @@ class MongoDB(BaseActivity):
documents = clear_mongo_id(documents) documents = clear_mongo_id(documents)
self.logger.info( self.info(
f"Loaded {len(documents)} documents from collection '{collection_name}'") f"Loaded {len(documents)} documents from collection '{collection_name}'", metadata=metadata)
self.logger.debug( self.debug(
f"Documents loaded: {documents}" f"Documents loaded: {documents}", metadata=metadata)
)
return documents return documents
except Exception as e: except Exception as e:
trace = traceback.format_exc() trace = traceback.format_exc()
self.notification_handler.build_and_send_notification( self.send_notification(
metadata=metadata,
notification_id="MONGODB_QUERY_ERROR", notification_id="MONGODB_QUERY_ERROR",
message=f"Failed to execute MongoDB query: {e}", message=f"Failed to execute MongoDB query: {e}",
block="load_query_from_mongodb", block="load_query_from_mongodb",
level=NotificationLevel.ERROR, level=NotificationLevel.ERROR,
attachment_content=trace attachment_content=trace
) )
self.logger.error(trace) self.error(trace, metadata=metadata)
raise e raise e
@@ -146,6 +147,7 @@ class MongoDB(BaseActivity):
""" """
query = input_data.get("query", {}) query = input_data.get("query", {})
metadata = input_data.get("metadata", {})
collection_name = query.get("collection") collection_name = query.get("collection")
if not collection_name: if not collection_name:
@@ -155,8 +157,8 @@ class MongoDB(BaseActivity):
raise ValueError("Aggregation must be provided.") raise ValueError("Aggregation must be provided.")
aggregation.append({"$project": {"_id": 0}}) aggregation.append({"$project": {"_id": 0}})
self.logger.info( self.info(
f"Aggregating documents from collection '{collection_name}' with aggregation: {aggregation}") f"Aggregating documents from collection '{collection_name}' with aggregation: {aggregation}", metadata=metadata)
try: try:
collection = self.database[collection_name] collection = self.database[collection_name]
@@ -166,25 +168,25 @@ class MongoDB(BaseActivity):
aggregated_documents = clear_mongo_id(aggregated_documents) aggregated_documents = clear_mongo_id(aggregated_documents)
self.logger.info( self.info(
f"Aggregated {len(aggregated_documents)} documents from collection '{collection_name}'") f"Aggregated {len(aggregated_documents)} documents from collection '{collection_name}'", metadata=metadata)
self.logger.debug( self.debug(
f"Aggregation result: {aggregated_documents}" f"Aggregation result: {aggregated_documents}", metadata=metadata)
)
return aggregated_documents return aggregated_documents
except Exception as e: except Exception as e:
trace = traceback.format_exc() trace = traceback.format_exc()
self.notification_handler.build_and_send_notification( self.send_notification(
metadata=metadata,
notification_id="MONGODB_AGGREGATION_ERROR", notification_id="MONGODB_AGGREGATION_ERROR",
message=f"Failed to execute MongoDB aggregation: {e}", message=f"Failed to execute MongoDB aggregation: {e}",
block="aggregate_documents_in_mongodb", block="aggregate_documents_in_mongodb",
level=NotificationLevel.ERROR, level=NotificationLevel.ERROR,
attachment_content=trace attachment_content=trace
) )
self.logger.error(trace) self.error(trace, metadata=metadata)
raise e raise e
@@ -196,6 +198,7 @@ class MongoDB(BaseActivity):
- updated_pipelines (list): List of updated pipelines. - updated_pipelines (list): List of updated pipelines.
""" """
updated_pipelines = input_data.get("updated_pipelines", []) updated_pipelines = input_data.get("updated_pipelines", [])
metadata = input_data.get("metadata", {})
now = datetime.now().strftime(DEFAULT_DATE_FORMAT) now = datetime.now().strftime(DEFAULT_DATE_FORMAT)
collection = self.database["orchestrated_schedules"] collection = self.database["orchestrated_schedules"]
@@ -205,10 +208,24 @@ class MongoDB(BaseActivity):
for pipeline in updated_pipelines if pipeline["success"] for pipeline in updated_pipelines if pipeline["success"]
] ]
data_filter = {"$or": argument} if argument else {} data_filter = {"$or": argument} if argument else {}
collection.update_many(
data_filter, try:
{"$set": {"updated_at": now}} collection.update_many(
) data_filter,
{"$set": {"updated_at": now}}
)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
metadata=metadata,
notification_id="MONGODB_UPDATE_PIPELINES_ERROR",
message=f"Failed to update pipelines timestamps: {e}",
block="update_pipelines_timestamps",
level=NotificationLevel.ERROR,
attachment_content=trace
)
self.error(trace, metadata=metadata)
raise e
@activity.defn(name="create_pipelines_timestamps") @activity.defn(name="create_pipelines_timestamps")
async def create_pipelines_timestamps(self, input_data: dict[str, Any]) -> None: async def create_pipelines_timestamps(self, input_data: dict[str, Any]) -> None:
@@ -218,6 +235,7 @@ class MongoDB(BaseActivity):
- created_pipelines (list): List of created pipelines. - created_pipelines (list): List of created pipelines.
""" """
created_pipelines = input_data.get("created_pipelines", []) created_pipelines = input_data.get("created_pipelines", [])
metadata = input_data.get("metadata", {})
collection = self.database["orchestrated_schedules"] collection = self.database["orchestrated_schedules"]
now = datetime.now().strftime(DEFAULT_DATE_FORMAT) now = datetime.now().strftime(DEFAULT_DATE_FORMAT)
@@ -229,7 +247,21 @@ class MongoDB(BaseActivity):
for pipeline in created_pipelines if pipeline["success"] for pipeline in created_pipelines if pipeline["success"]
] ]
data_filter = argument if argument else {} data_filter = argument if argument else {}
collection.insert_many(data_filter)
try:
collection.insert_many(data_filter)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
metadata=metadata,
notification_id="MONGODB_CREATE_PIPELINES_ERROR",
message=f"Failed to create pipelines timestamps: {e}",
block="create_pipelines_timestamps",
level=NotificationLevel.ERROR,
attachment_content=trace
)
self.error(trace, metadata=metadata)
raise e
@activity.defn(name="delete_pipelines_timestamps") @activity.defn(name="delete_pipelines_timestamps")
async def delete_pipelines_timestamps(self, input_data: dict[str, Any]) -> None: async def delete_pipelines_timestamps(self, input_data: dict[str, Any]) -> None:
@@ -239,6 +271,7 @@ class MongoDB(BaseActivity):
- deleted_pipelines (list): List of deleted pipelines. - deleted_pipelines (list): List of deleted pipelines.
""" """
deleted_pipelines = input_data.get("deleted_pipelines", []) deleted_pipelines = input_data.get("deleted_pipelines", [])
metadata = input_data.get("metadata", {})
collection = self.database["orchestrated_schedules"] collection = self.database["orchestrated_schedules"]
argument = [ argument = [
@@ -248,4 +281,17 @@ class MongoDB(BaseActivity):
] ]
data_filter = {"$or": argument} if argument else {} data_filter = {"$or": argument} if argument else {}
collection.delete_many(data_filter) try:
collection.delete_many(data_filter)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
metadata=metadata,
notification_id="MONGODB_DELETE_PIPELINES_ERROR",
message=f"Failed to delete pipelines timestamps: {e}",
block="delete_pipelines_timestamps",
level=NotificationLevel.ERROR,
attachment_content=trace
)
self.error(trace, metadata=metadata)
raise e

0
orchestrator/activities/notification.py
View File

110
orchestrator/activities/slot_manager.py
View File

@@ -2,10 +2,12 @@ from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through(): with workflow.unsafe.imports_passed_through():
from typing import Any from typing import Any
import traceback
import json import json
from logging import Logger from logging import Logger
from sientia_do.temporal.activities.redis_base import Redis from sientia_do.temporal.activities.redis_base import Redis
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
class SlotManager(Redis): class SlotManager(Redis):
@@ -18,7 +20,7 @@ class SlotManager(Redis):
password, logger, notification_handler) password, logger, notification_handler)
@activity.defn(name="load_opc_slots") @activity.defn(name="load_opc_slots")
async def load_opc_slots(self) -> dict[str, Any]: async def load_opc_slots(self, input_data: dict[str, Any]) -> dict[str, Any]:
""" """
Load all OPC slots from Redis Load all OPC slots from Redis
@@ -26,32 +28,45 @@ class SlotManager(Redis):
dict[str, Any]: A dictionary of OPC slots dict[str, Any]: A dictionary of OPC slots
""" """
self.logger.info("Loading OPC slots...") metadata = input_data.get("metadata", {})
self.info("Loading OPC slots...", metadata=metadata)
opc_slots = {} opc_slots = {}
slot_keys = self.redis_client.keys("slot:opc_tags:*") try:
self.logger.debug(f"Slot keys: {slot_keys}") slot_keys = self.redis_client.keys("slot:opc_tags:*")
if slot_keys: self.debug(f"Slot keys: {slot_keys}", metadata=metadata)
if isinstance(slot_keys[0], bytes):
decoded_keys = [key.decode('utf-8') for key in slot_keys]
else:
decoded_keys = slot_keys
for key in decoded_keys: if slot_keys:
opc_slots[key] = self.get(key) if isinstance(slot_keys[0], bytes):
decoded_keys = [key.decode('utf-8') for key in slot_keys]
else:
decoded_keys = slot_keys
self.logger.info(f"Loaded {len(opc_slots)} OPC slots") for key in decoded_keys:
opc_slots[key] = self.get(key)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
metadata=metadata,
notification_id="REDIS_GET_ERROR",
message=f"Failed to load OPC slots: {e}",
block="load_opc_slots",
level=NotificationLevel.ERROR,
attachment_content=trace
)
self.error(trace, metadata=metadata)
raise e
self.logger.debug( self.info(f"Loaded {len(opc_slots)} OPC slots", metadata=metadata)
f"Loaded: \n {json.dumps(opc_slots, indent=4, sort_keys=True)}")
return opc_slots return opc_slots
@activity.defn(name="load_active_ingestors") @activity.defn(name="load_active_ingestors")
async def load_active_ingestors(self) -> list[str]: async def load_active_ingestors(self, input_data: dict[str, Any]) -> list[str]:
""" """
Load all active ingestors from Redis Load all active ingestors from Redis
@@ -59,23 +74,41 @@ class SlotManager(Redis):
list[str]: A list of active ingestors list[str]: A list of active ingestors
""" """
self.logger.info("Loading active ingestors...") metadata = input_data.get("metadata", {})
active_ingestors = self.redis_client.keys("heartbeat:ingestor:*") self.info("Loading active ingestors...", metadata=metadata)
self.logger.info(f"Loaded {len(active_ingestors)} active ingestors") try:
self.logger.debug(f"Active ingestors: \n {active_ingestors}") active_ingestors = self.redis_client.keys("heartbeat:ingestor:*")
ingestors = [] self.info(
f"Loaded {len(active_ingestors)} active ingestors", metadata=metadata)
for ingestor in active_ingestors: self.debug(
if isinstance(ingestor, bytes): f"Active ingestors: \n {active_ingestors}", metadata=metadata)
ingestors.append(ingestor.decode('utf-8'))
else:
ingestors.append(ingestor)
return ingestors ingestors = []
for ingestor in active_ingestors:
if isinstance(ingestor, bytes):
ingestors.append(ingestor.decode('utf-8'))
else:
ingestors.append(ingestor)
return ingestors
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
metadata=metadata,
notification_id="REDIS_GET_ERROR",
message=f"Failed to load active ingestors: {e}",
block="load_active_ingestors",
level=NotificationLevel.ERROR,
attachment_content=trace
)
self.error(trace, metadata=metadata)
raise e
@activity.defn(name="update_slots") @activity.defn(name="update_slots")
async def update_slots(self, input_data: dict[str, Any]) -> dict[str, Any]: async def update_slots(self, input_data: dict[str, Any]) -> dict[str, Any]:
@@ -92,7 +125,8 @@ class SlotManager(Redis):
""" """
to_insert = input_data['to_insert'] to_insert = input_data['to_insert']
self.logger.info("Updating OPC slots...") metadata = input_data.get("metadata", {})
self.info("Updating OPC slots...", metadata=metadata)
report = {} report = {}
@@ -105,16 +139,17 @@ class SlotManager(Redis):
"message": "Slot updated successfully" "message": "Slot updated successfully"
} }
except Exception as e: except Exception as e:
self.logger.error(f"Failed to update slot {slot}: {str(e)}") self.error(
f"Failed to update slot {slot}: {str(e)}", metadata=metadata)
report[slot] = { report[slot] = {
"success": False, "success": False,
"message": str(e) "message": str(e)
} }
self.logger.info(f"Updated {len(to_insert)} OPC slots") self.info(f"Updated {len(to_insert)} OPC slots", metadata=metadata)
self.logger.debug( self.debug(
f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}") f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}", metadata=metadata)
return report return report
@@ -133,7 +168,7 @@ class SlotManager(Redis):
""" """
to_delete = input_data['to_delete'] to_delete = input_data['to_delete']
self.logger.info("Deleting OPC slots...") metadata = input_data.get("metadata", {})
report = {} report = {}
@@ -145,15 +180,16 @@ class SlotManager(Redis):
"message": "Slot deleted successfully" "message": "Slot deleted successfully"
} }
except Exception as e: except Exception as e:
self.logger.error(f"Failed to delete slot {slot}: {str(e)}") self.error(
f"Failed to delete slot {slot}: {str(e)}", metadata=metadata)
report[slot] = { report[slot] = {
"success": False, "success": False,
"message": str(e) "message": str(e)
} }
self.logger.info(f"Deleted {len(to_delete)} OPC slots") self.info(f"Deleted {len(to_delete)} OPC slots", metadata=metadata)
self.logger.debug( self.debug(
f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}") f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}", metadata=metadata)
return report return report

168
orchestrator/activities/temporal_manager.py
View File

@@ -4,10 +4,12 @@ from temporalio.client import (
from temporalio.common import SearchAttributeKey, SearchAttributePair, TypedSearchAttributes from temporalio.common import SearchAttributeKey, SearchAttributePair, TypedSearchAttributes
with workflow.unsafe.imports_passed_through(): with workflow.unsafe.imports_passed_through():
import traceback
from typing import Any from typing import Any
from logging import Logger from logging import Logger
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.temporal.activities.base import BaseActivity from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.notifications.models import NotificationLevel
from google.protobuf.json_format import MessageToDict from google.protobuf.json_format import MessageToDict
import base64 import base64
from datetime import timedelta from datetime import timedelta
@@ -61,73 +63,6 @@ class TemporalManager(BaseActivity):
self.laborious_namespace: laborious_client self.laborious_namespace: laborious_client
} }
@activity.defn(name="load_schedule")
async def load_schedule(self) -> dict[str, Any]:
"""
Load all orchestrated schedules from Temporal. Filters by search attribute
"Orchestrated" set to "true" and returns a dictionary of schedule_id:
{frequency, data, handle}
Returns:
dict[str, Any]: A dictionary of orchestrated schedules
"""
self.logger.info("Getting orchestrated schedules...")
orchestrated_schedules = {}
for namespace, client in self.temporal_clients.items():
orchestrated_schedules[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
self.logger.debug(f"Schedule id: {schedule_id}")
handle = client.get_schedule_handle(
schedule_id)
self.logger.debug("Handle acquired")
self.schedule_handles[namespace][schedule_id] = handle
self.logger.debug("Describing schedule...")
desc = await handle.describe(
rpc_timeout=timedelta(seconds=60)
)
self.logger.debug("Parsing args...")
for arg in desc.schedule.action.args:
data = MessageToDict(arg)['data']
data = base64.b64decode(data).decode('utf-8')
frequency = desc.schedule.spec.intervals[0].every.seconds
orchestrated_schedules[namespace][schedule_id] = {
'frequency': frequency,
'data': json.loads(data),
}
await sleep(0.1)
self.logger.info(
f"Found {len(orchestrated_schedules[self.scouter_namespace]) + len(orchestrated_schedules[self.laborious_namespace])} orchestrated schedules")
self.logger.debug(
f"Orchestrated schedules: {orchestrated_schedules}")
self.logger.debug(
f"Schedule handles: {self.schedule_handles}")
return orchestrated_schedules
@activity.defn(name="normalize_schedules") @activity.defn(name="normalize_schedules")
async def normalize_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]: async def normalize_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]:
""" """
@@ -135,29 +70,45 @@ class TemporalManager(BaseActivity):
input_data: input_data:
- orchestrated_schedules (dict[str, Any]): The orchestrated schedules to compare. - orchestrated_schedules (dict[str, Any]): The orchestrated schedules to compare.
""" """
self.logger.info("Getting orchestrated schedules...") metadata = input_data.get("metadata", {})
self.info("Getting orchestrated schedules...", metadata=metadata)
orchestrated_schedules = input_data.get('orchestrated_schedules', {}) orchestrated_schedules = input_data.get('orchestrated_schedules', {})
for namespace, client in self.temporal_clients.items(): for namespace, client in self.temporal_clients.items():
schedules = orchestrated_schedules.get(namespace, {}) try:
schedules = orchestrated_schedules.get(namespace, {})
self.logger.info( self.info(
f"Getting orchestrated schedules for {namespace}") f"Getting orchestrated schedules for {namespace}", metadata=metadata)
async for schedule in await client.list_schedules(): async for schedule in await client.list_schedules():
search_attrs = getattr(schedule, "search_attributes", {}) search_attrs = getattr(schedule, "search_attributes", {})
if search_attrs.get("orchestrated", ["false"]) == ["true"]: if search_attrs.get("orchestrated", ["false"]) == ["true"]:
schedule_id = schedule.id schedule_id = schedule.id
if schedule_id not in schedules: if schedule_id not in schedules:
self.logger.info( self.info(
f"Schedule {schedule_id} not found in mongo db, cleaning up") f"Schedule {schedule_id} not found in mongo db, cleaning up", metadata=metadata)
handle = client.get_schedule_handle( handle = client.get_schedule_handle(
schedule_id) schedule_id)
await handle.delete() await handle.delete()
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
metadata=metadata,
notification_id="TEMPORAL_NORMALIZE_SCHEDULES_ERROR",
message=f"Failed to normalize schedules: {e}",
block="normalize_schedules",
level=NotificationLevel.ERROR,
attachment_content=trace
)
self.error(trace, metadata=metadata)
raise e
@activity.defn(name="create_schedules") @activity.defn(name="create_schedules")
async def create_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]: async def create_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
@@ -174,6 +125,7 @@ class TemporalManager(BaseActivity):
""" """
schedules_to_create = input_data['schedules'] schedules_to_create = input_data['schedules']
metadata = input_data.get("metadata", {})
report = [] report = []
@@ -202,9 +154,10 @@ class TemporalManager(BaseActivity):
workflow_type = schedule['workflow_type'] workflow_type = schedule['workflow_type']
try: try:
self.logger.debug(f"Creating schedule {schedule_name}:") self.debug(
self.logger.debug( f"Creating schedule {schedule_name}:", metadata=metadata)
f"{json.dumps(schedule, indent=4, sort_keys=True)}") self.debug(
f"{json.dumps(schedule, indent=4, sort_keys=True)}", metadata=metadata)
await client.create_schedule( await client.create_schedule(
schedule_name, schedule_name,
@@ -235,8 +188,8 @@ class TemporalManager(BaseActivity):
"message": "Schedule created successfully" "message": "Schedule created successfully"
}) })
except Exception as e: except Exception as e:
self.logger.error( self.error(
f"Failed to create schedule {schedule_name}: {str(e)}") f"Failed to create schedule {schedule_name}: {str(e)}", metadata=metadata)
report.append({ report.append({
"namespace": namespace, "namespace": namespace,
"schedule_name": schedule_name, "schedule_name": schedule_name,
@@ -244,10 +197,11 @@ class TemporalManager(BaseActivity):
"message": str(e) "message": str(e)
}) })
self.logger.info(f"Processed {len(schedules_to_create)} schedules") self.info(
f"Processed {len(schedules_to_create)} schedules", metadata=metadata)
self.logger.debug( self.debug(
f"\n {json.dumps(report, indent=4, sort_keys=True)}") f"\n {json.dumps(report, indent=4, sort_keys=True)}", metadata=metadata)
return report return report
@@ -266,7 +220,7 @@ class TemporalManager(BaseActivity):
""" """
schedules_to_update = input_data['schedules'] schedules_to_update = input_data['schedules']
metadata = input_data.get("metadata", {})
report = [] report = []
for namespace, schedules in schedules_to_update.items(): for namespace, schedules in schedules_to_update.items():
@@ -288,12 +242,12 @@ class TemporalManager(BaseActivity):
async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate: # NOSONAR async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate: # NOSONAR
schedule_action = input_data.description.schedule.action schedule_action = input_data.description.schedule.action
self.logger.debug("Updating schedule:") self.debug("Updating schedule:", metadata=metadata)
if hasattr(schedule_action, "args"): if hasattr(schedule_action, "args"):
self.logger.debug("New schedule:") self.debug("New schedule:", metadata=metadata)
self.logger.debug( self.debug(
f"{json.dumps(schedule, indent=4, sort_keys=True)}") # NOSONAR f"{json.dumps(schedule, indent=4, sort_keys=True)}", metadata=metadata) # NOSONAR
schedule_action.args = [schedule] schedule_action.args = [schedule]
@@ -318,8 +272,8 @@ class TemporalManager(BaseActivity):
"message": "Schedule updated successfully" "message": "Schedule updated successfully"
}) })
except Exception as e: except Exception as e:
self.logger.error( self.error(
f"Failed to update schedule {schedule_name}: {str(e)}") f"Failed to update schedule {schedule_name}: {str(e)}", metadata=metadata)
report.append({ report.append({
"namespace": namespace, "namespace": namespace,
"schedule_name": schedule_name, "schedule_name": schedule_name,
@@ -327,10 +281,11 @@ class TemporalManager(BaseActivity):
"message": str(e) "message": str(e)
}) })
self.logger.info(f"Processed {len(schedules_to_update)} schedules") self.info(
f"Processed {len(schedules_to_update)} schedules", metadata=metadata)
self.logger.debug( self.debug(
f"\n {json.dumps(report, indent=4, sort_keys=True)}") f"\n {json.dumps(report, indent=4, sort_keys=True)}", metadata=metadata)
return report return report
@@ -349,7 +304,7 @@ class TemporalManager(BaseActivity):
""" """
schedules_to_delete = input_data['schedules'] schedules_to_delete = input_data['schedules']
metadata = input_data.get("metadata", {})
report = [] report = []
for namespace, schedules in schedules_to_delete.items(): for namespace, schedules in schedules_to_delete.items():
@@ -378,8 +333,8 @@ class TemporalManager(BaseActivity):
"message": "Schedule deleted successfully" "message": "Schedule deleted successfully"
}) })
except Exception as e: except Exception as e:
self.logger.error( self.error(
f"Failed to delete schedule {schedule_name}: {str(e)}") f"Failed to delete schedule {schedule_name}: {str(e)}", metadata=metadata)
report.append({ report.append({
"namespace": namespace, "namespace": namespace,
"schedule_name": schedule_name, "schedule_name": schedule_name,
@@ -387,9 +342,10 @@ class TemporalManager(BaseActivity):
"message": str(e) "message": str(e)
}) })
self.logger.info(f"Processed {len(schedules_to_delete)} schedules") self.info(
f"Processed {len(schedules_to_delete)} schedules", metadata=metadata)
self.logger.debug( self.debug(
f"\n {json.dumps(report, indent=4, sort_keys=True)}") f"\n {json.dumps(report, indent=4, sort_keys=True)}", metadata=metadata)
return report return report

12
orchestrator/worker/worker.py
View File

@@ -14,29 +14,32 @@ with workflow.unsafe.imports_passed_through():
build_redis_config, build_redis_config,
build_mongodb_config build_mongodb_config
) )
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.temporal.utils.logger import get_logger from sientia_do.temporal.utils.logger import get_logger
async def main(): async def main():
host = os.getenv('TEMPORAL_HOST', 'localhost:7233') host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
namespace = os.getenv('TEMPORAL_NAMESPACE', 'default')
logger = get_logger(__name__) logger = get_logger(__name__)
logger.info('Starting Worker...') logger.info('Starting Worker...')
logger.info('Starting Notification Handler...') logger.info('Starting Notification Handler...')
mongo_config = build_mongodb_config()
notification_handler = NotificationHandler( notification_handler = NotificationHandler(
servers=os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'http://localhost:9092'), connection_string=mongo_config['connection_string'],
database=mongo_config['database_name'],
logger=logger, logger=logger,
project_name=os.getenv('PROJECT_NAME', 'orchestrator'), project_name=os.getenv('PROJECT_NAME', 'orchestrator'),
) )
logger.info('Starting Temporal Client...') logger.info(f'Starting Temporal Client at {host}:{namespace}')
temporal_client = await client.Client.connect( temporal_client = await client.Client.connect(
target_host=host, target_host=host,
namespace=os.getenv('TEMPORAL_NAMESPACE', 'default') namespace=namespace
) )
logger.info('Starting Activities...') logger.info('Starting Activities...')
@@ -73,7 +76,6 @@ async def main():
activities.create_pipelines_timestamps, activities.create_pipelines_timestamps,
activities.delete_pipelines_timestamps, activities.delete_pipelines_timestamps,
# Temporal # Temporal
activities.load_schedule,
activities.create_schedules, activities.create_schedules,
activities.update_schedules, activities.update_schedules,
activities.delete_schedules, activities.delete_schedules,

36
orchestrator/workflows/orchestrator.py
View File

@@ -14,9 +14,19 @@ class Orchestrator:
input_data['workflow_name'] = 'orchestrator' input_data['workflow_name'] = 'orchestrator'
metadata = {
'metadata': {
'schedule_name': input_data.get('schedule_name', 'orchestrator'),
'model_name': '-',
'model_id': '-',
'workflow_name': input_data['workflow_name']
}
}
pipeline_config_handler = workflow.start_local_activity_method( pipeline_config_handler = workflow.start_local_activity_method(
Activities.aggregate_documents_in_mongodb, Activities.aggregate_documents_in_mongodb,
{ {
**metadata,
'query': input_data['pipelines_query'] 'query': input_data['pipelines_query']
}, },
retry_policy=retry_policy, retry_policy=retry_policy,
@@ -26,6 +36,7 @@ class Orchestrator:
opc_servers_handler = workflow.start_local_activity_method( opc_servers_handler = workflow.start_local_activity_method(
Activities.find_documents_in_mongodb, Activities.find_documents_in_mongodb,
{ {
**metadata,
'query': input_data['opc_servers_query'] 'query': input_data['opc_servers_query']
}, },
retry_policy=retry_policy, retry_policy=retry_policy,
@@ -35,6 +46,7 @@ class Orchestrator:
orchestrated_schedules_handler = workflow.start_local_activity_method( orchestrated_schedules_handler = workflow.start_local_activity_method(
Activities.find_documents_in_mongodb, Activities.find_documents_in_mongodb,
{ {
**metadata,
'query': { 'query': {
'collection': 'orchestrated_schedules' 'collection': 'orchestrated_schedules'
} }
@@ -45,12 +57,18 @@ class Orchestrator:
current_slot_config_handler = workflow.start_local_activity_method( current_slot_config_handler = workflow.start_local_activity_method(
Activities.load_opc_slots, Activities.load_opc_slots,
{
**metadata,
},
retry_policy=retry_policy, retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60) start_to_close_timeout=timedelta(seconds=60)
) )
active_ingestors_handler = workflow.start_local_activity_method( active_ingestors_handler = workflow.start_local_activity_method(
Activities.load_active_ingestors, Activities.load_active_ingestors,
{
**metadata,
},
retry_policy=retry_policy, retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60) start_to_close_timeout=timedelta(seconds=60)
) )
@@ -64,6 +82,7 @@ class Orchestrator:
formatted_orchestrated_schedules_handler = workflow.start_local_activity_method( formatted_orchestrated_schedules_handler = workflow.start_local_activity_method(
Activities.format_schedule_config, Activities.format_schedule_config,
{ {
**metadata,
'schedule_config': orchestrated_schedules 'schedule_config': orchestrated_schedules
}, },
retry_policy=retry_policy, retry_policy=retry_policy,
@@ -73,6 +92,7 @@ class Orchestrator:
schedules_config_handler = workflow.start_local_activity_method( schedules_config_handler = workflow.start_local_activity_method(
Activities.process_schedules, Activities.process_schedules,
{ {
**metadata,
'pipelines': pipeline_config 'pipelines': pipeline_config
}, },
retry_policy=retry_policy, retry_policy=retry_policy,
@@ -82,6 +102,7 @@ class Orchestrator:
slot_config_handler = workflow.start_local_activity_method( slot_config_handler = workflow.start_local_activity_method(
Activities.process_slots, Activities.process_slots,
{ {
**metadata,
'opc_servers': opc_servers, 'opc_servers': opc_servers,
'active_ingestors': active_ingestors, 'active_ingestors': active_ingestors,
'pipelines': pipeline_config, 'pipelines': pipeline_config,
@@ -97,6 +118,7 @@ class Orchestrator:
schedule_actions_handler = workflow.start_local_activity_method( schedule_actions_handler = workflow.start_local_activity_method(
Activities.create_schedule_config, Activities.create_schedule_config,
{ {
**metadata,
'current_schedule_config': formatted_orchestrated_schedules, 'current_schedule_config': formatted_orchestrated_schedules,
'schedule_config': schedules_config 'schedule_config': schedules_config
}, },
@@ -107,6 +129,7 @@ class Orchestrator:
slot_actions_handler = workflow.start_local_activity_method( slot_actions_handler = workflow.start_local_activity_method(
Activities.create_slot_config, Activities.create_slot_config,
{ {
**metadata,
'current_slot_config': current_slot_config, 'current_slot_config': current_slot_config,
'slot_config': slot_config 'slot_config': slot_config
}, },
@@ -114,9 +137,10 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) start_to_close_timeout=timedelta(seconds=60)
) )
normalize_schedules_handler = workflow.start_local_activity_method( normalize_schedules_handler = workflow.start_activity_method(
Activities.normalize_schedules, Activities.normalize_schedules,
{ {
**metadata,
'orchestrated_schedules': formatted_orchestrated_schedules 'orchestrated_schedules': formatted_orchestrated_schedules
}, },
retry_policy=retry_policy, retry_policy=retry_policy,
@@ -130,6 +154,7 @@ class Orchestrator:
slot_deletion_report_handler = workflow.start_activity_method( slot_deletion_report_handler = workflow.start_activity_method(
Activities.delete_slots, Activities.delete_slots,
{ {
**metadata,
'to_delete': slot_actions['to_delete'] 'to_delete': slot_actions['to_delete']
}, },
retry_policy=retry_policy, retry_policy=retry_policy,
@@ -139,6 +164,7 @@ class Orchestrator:
slot_insertion_report_handler = workflow.start_activity_method( slot_insertion_report_handler = workflow.start_activity_method(
Activities.update_slots, Activities.update_slots,
{ {
**metadata,
'to_insert': slot_actions['to_insert'] 'to_insert': slot_actions['to_insert']
}, },
retry_policy=retry_policy, retry_policy=retry_policy,
@@ -148,6 +174,7 @@ class Orchestrator:
schedule_deletion_report_handler = workflow.start_activity_method( schedule_deletion_report_handler = workflow.start_activity_method(
Activities.delete_schedules, Activities.delete_schedules,
{ {
**metadata,
'schedules': schedule_actions['to_delete'] 'schedules': schedule_actions['to_delete']
}, },
retry_policy=retry_policy, retry_policy=retry_policy,
@@ -157,6 +184,7 @@ class Orchestrator:
schedule_insertion_report_handler = workflow.start_activity_method( schedule_insertion_report_handler = workflow.start_activity_method(
Activities.create_schedules, Activities.create_schedules,
{ {
**metadata,
'schedules': schedule_actions['to_create'] 'schedules': schedule_actions['to_create']
}, },
retry_policy=retry_policy, retry_policy=retry_policy,
@@ -166,6 +194,7 @@ class Orchestrator:
schedule_update_report_handler = workflow.start_activity_method( schedule_update_report_handler = workflow.start_activity_method(
Activities.update_schedules, Activities.update_schedules,
{ {
**metadata,
'schedules': schedule_actions['to_update'] 'schedules': schedule_actions['to_update']
}, },
retry_policy=retry_policy, retry_policy=retry_policy,
@@ -183,6 +212,7 @@ class Orchestrator:
schedule_report_handler = workflow.start_activity_method( schedule_report_handler = workflow.start_activity_method(
Activities.report_schedule_orchestration, Activities.report_schedule_orchestration,
{ {
**metadata,
'created_schedules': schedule_insertion_report, 'created_schedules': schedule_insertion_report,
'updated_schedules': schedule_update_report, 'updated_schedules': schedule_update_report,
'deleted_schedules': schedule_deletion_report 'deleted_schedules': schedule_deletion_report
@@ -196,6 +226,7 @@ class Orchestrator:
slot_report_handler = workflow.start_activity_method( slot_report_handler = workflow.start_activity_method(
Activities.report_slot_orchestration, Activities.report_slot_orchestration,
{ {
**metadata,
'inserted_slots': slot_insertion_report, 'inserted_slots': slot_insertion_report,
'deleted_slots': slot_deletion_report 'deleted_slots': slot_deletion_report
}, },
@@ -208,6 +239,7 @@ class Orchestrator:
update_pipelines_timestamps_handler = workflow.start_activity_method( update_pipelines_timestamps_handler = workflow.start_activity_method(
Activities.update_pipelines_timestamps, Activities.update_pipelines_timestamps,
{ {
**metadata,
'updated_pipelines': schedule_update_report 'updated_pipelines': schedule_update_report
}, },
retry_policy=retry_policy, retry_policy=retry_policy,
@@ -219,6 +251,7 @@ class Orchestrator:
create_pipelines_timestamps_handler = workflow.start_activity_method( create_pipelines_timestamps_handler = workflow.start_activity_method(
Activities.create_pipelines_timestamps, Activities.create_pipelines_timestamps,
{ {
**metadata,
'created_pipelines': schedule_insertion_report 'created_pipelines': schedule_insertion_report
}, },
retry_policy=retry_policy, retry_policy=retry_policy,
@@ -230,6 +263,7 @@ class Orchestrator:
delete_pipelines_timestamps_handler = workflow.start_activity_method( delete_pipelines_timestamps_handler = workflow.start_activity_method(
Activities.delete_pipelines_timestamps, Activities.delete_pipelines_timestamps,
{ {
**metadata,
'deleted_pipelines': schedule_deletion_report 'deleted_pipelines': schedule_deletion_report
}, },
retry_policy=retry_policy, retry_policy=retry_policy,

2
requirements.txt
View File

@@ -4,4 +4,4 @@ sqlalchemy
redis redis
couchbase couchbase
pymongo pymongo
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.2.1 git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.3

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

@@ -8,13 +8,26 @@ from orchestrator.utils.orchestrator_functions import build_tag_config
@fixture @fixture
def formatters(): def formatters():
return Formatters( formatters = Formatters(
scouter_namespace="scouter", scouter_namespace="scouter",
laborious_namespace="laborious", laborious_namespace="laborious",
logger=MagicMock(), logger=MagicMock(),
notification_handler=MagicMock() notification_handler=MagicMock()
) )
formatters.send_notification = MagicMock()
return formatters
metadata = {
"metadata": {
"schedule_name": "test_schedule_name",
"workflow_name": "test_workflow_name",
"model_name": "test_model_name",
"model_id": "test_model_id"
}
}
@mark.asyncio @mark.asyncio
@patch("orchestrator.activities.formatters.scouter", @patch("orchestrator.activities.formatters.scouter",
@@ -374,23 +387,33 @@ async def test_create_slot_config(formatters):
def test_send_success_report(formatters): def test_send_success_report(formatters):
formatters.send_success_report("test_message", "test_notification_id") formatters.send_success_report(
formatters.notification_handler.build_and_send_notification.assert_called_once_with( metadata=metadata,
"test_notification_id", message="test_message",
"test_message", notification_id="test_notification_id"
"report_orchestration", )
NotificationLevel.INFO formatters.send_notification.assert_called_once_with(
metadata=metadata,
notification_id="test_notification_id",
message="test_message",
block="report_orchestration",
level=NotificationLevel.INFO
) )
def test_send_error_report(formatters): def test_send_error_report(formatters):
formatters.send_error_report( formatters.send_error_report(
"test_message", "test_notification_id", {"test": "test"}) metadata=metadata,
formatters.notification_handler.build_and_send_notification.assert_called_once_with( message="test_message",
"test_notification_id", notification_id="test_notification_id",
"test_message", attachment={"test": "test"}
"report_orchestration", )
NotificationLevel.ERROR, formatters.send_notification.assert_called_once_with(
metadata=metadata,
notification_id="test_notification_id",
message="test_message",
block="report_orchestration",
level=NotificationLevel.ERROR,
attachment_content=json.dumps( attachment_content=json.dumps(
{"test": "test"}, indent=4, sort_keys=True) {"test": "test"}, indent=4, sort_keys=True)
) )
@@ -446,6 +469,7 @@ async def test_report_schedule_orchestration(formatters):
formatters.send_error_report = MagicMock() formatters.send_error_report = MagicMock()
input_data = { input_data = {
**metadata,
"created_schedules": [ "created_schedules": [
{ {
"namespace": "test_namespace", "namespace": "test_namespace",
@@ -496,33 +520,39 @@ async def test_report_schedule_orchestration(formatters):
]) ])
formatters.send_success_report.assert_has_calls([ formatters.send_success_report.assert_has_calls([
call( call(
"Created schedules: \n test_namespace/test_schedule_name_to_create", metadata=metadata['metadata'],
"REPORT_ORCHESTRATION_CREATED_SCHEDULES" message="Created schedules: \n test_namespace/test_schedule_name_to_create",
notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES"
), ),
call( call(
"Updated schedules: \n test_namespace/test_schedule_name_to_update", metadata=metadata['metadata'],
"REPORT_ORCHESTRATION_UPDATED_SCHEDULES" message="Updated schedules: \n test_namespace/test_schedule_name_to_update",
notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES"
), ),
call( call(
"Deleted schedules: \n test_namespace/test_schedule_name_to_delete", metadata=metadata['metadata'],
"REPORT_ORCHESTRATION_DELETED_SCHEDULES" message="Deleted schedules: \n test_namespace/test_schedule_name_to_delete",
notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES"
) )
]) ])
formatters.send_error_report.assert_has_calls([ formatters.send_error_report.assert_has_calls([
call( call(
"Failed to create schedules: \n test_namespace/test_schedule_name_to_create_error: test_error", metadata=metadata['metadata'],
"REPORT_ORCHESTRATION_CREATED_SCHEDULES", message="Failed to create schedules: \n test_namespace/test_schedule_name_to_create_error: test_error",
input_data['created_schedules'] notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES",
attachment=input_data['created_schedules']
), ),
call( call(
"Failed to update schedules: \n test_namespace/test_schedule_name_to_update_error: test_error", metadata=metadata['metadata'],
"REPORT_ORCHESTRATION_UPDATED_SCHEDULES", message="Failed to update schedules: \n test_namespace/test_schedule_name_to_update_error: test_error",
input_data['updated_schedules'] notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES",
attachment=input_data['updated_schedules']
), ),
call( call(
"Failed to delete schedules: \n test_namespace/test_schedule_name_to_delete_error: test_error", metadata=metadata['metadata'],
"REPORT_ORCHESTRATION_DELETED_SCHEDULES", message="Failed to delete schedules: \n test_namespace/test_schedule_name_to_delete_error: test_error",
input_data['deleted_schedules'] notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES",
attachment=input_data['deleted_schedules']
) )
]) ])
@@ -536,6 +566,7 @@ async def test_report_slot_orchestration(formatters):
formatters.send_error_report = MagicMock() formatters.send_error_report = MagicMock()
input_data = { input_data = {
**metadata,
"inserted_slots": { "inserted_slots": {
"test_slot_name_to_create": { "test_slot_name_to_create": {
"success": True "success": True
@@ -564,23 +595,27 @@ async def test_report_slot_orchestration(formatters):
]) ])
formatters.send_success_report.assert_has_calls([ formatters.send_success_report.assert_has_calls([
call( call(
"Inserted slots: \n test_slot_name_to_create", metadata=metadata['metadata'],
"REPORT_ORCHESTRATION_INSERTED_SLOTS" message="Inserted slots: \n test_slot_name_to_create",
notification_id="REPORT_ORCHESTRATION_INSERTED_SLOTS"
), ),
call( call(
"Deleted slots: \n test_slot_name_to_delete", metadata=metadata['metadata'],
"REPORT_ORCHESTRATION_DELETED_SLOTS" message="Deleted slots: \n test_slot_name_to_delete",
notification_id="REPORT_ORCHESTRATION_DELETED_SLOTS"
) )
]) ])
formatters.send_error_report.assert_has_calls([ formatters.send_error_report.assert_has_calls([
call( call(
"Failed to insert slots: \n test_slot_name_to_create_error", metadata=metadata['metadata'],
"REPORT_ORCHESTRATION_INSERTED_SLOTS", message="Failed to insert slots: \n test_slot_name_to_create_error",
input_data['inserted_slots'] notification_id="REPORT_ORCHESTRATION_INSERTED_SLOTS",
attachment=input_data['inserted_slots']
), ),
call( call(
"Failed to delete slots: \n test_slot_name_to_delete_error", metadata=metadata['metadata'],
"REPORT_ORCHESTRATION_DELETED_SLOTS", message="Failed to delete slots: \n test_slot_name_to_delete_error",
input_data['deleted_slots'] notification_id="REPORT_ORCHESTRATION_DELETED_SLOTS",
attachment=input_data['deleted_slots']
) )
]) ])

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

@@ -1,3 +1,4 @@
from curses import meta
from unittest.mock import MagicMock, patch, ANY from unittest.mock import MagicMock, patch, ANY
from pytest import fixture, mark from pytest import fixture, mark
from orchestrator.activities.mongo_db import clear_mongo_id from orchestrator.activities.mongo_db import clear_mongo_id
@@ -50,7 +51,7 @@ def test_clear_mongo_id():
@fixture @fixture
@patch("orchestrator.activities.mongo_db.MongoClient") @patch("orchestrator.activities.mongo_db.MongoClient")
def mongo_db(mongo_mock): def mongo_db(mongo_mock):
return ( mongo = (
MongoDB( MongoDB(
connection_string="mongodb://localhost:27017", connection_string="mongodb://localhost:27017",
database_name="test_db", database_name="test_db",
@@ -58,6 +59,8 @@ def mongo_db(mongo_mock):
notification_handler=MagicMock() notification_handler=MagicMock()
) )
) )
mongo.send_notification = MagicMock()
return mongo
@patch("orchestrator.activities.mongo_db.MongoClient") @patch("orchestrator.activities.mongo_db.MongoClient")
@@ -117,6 +120,15 @@ async def test_find_documents_in_mongodb_success(mongo_db):
{"name": {"$exists": True}}, {"_id": 0} {"name": {"$exists": True}}, {"_id": 0}
) )
metadata = {
"metadata": {
"schedule_name": "test_schedule_name",
"workflow_name": "test_workflow_name",
"model_name": "test_model_name",
"model_id": "test_model_id"
}
}
@mark.asyncio @mark.asyncio
async def test_find_documents_in_mongodb_failure(mongo_db): async def test_find_documents_in_mongodb_failure(mongo_db):
@@ -130,11 +142,13 @@ async def test_find_documents_in_mongodb_failure(mongo_db):
try: try:
await mongo_db.find_documents_in_mongodb( await mongo_db.find_documents_in_mongodb(
{ {
"query": input_data "query": input_data,
**metadata
}) })
except Exception as e: except Exception as e:
assert str(e) == "Error" assert str(e) == "Error"
mongo_db.notification_handler.build_and_send_notification.assert_called_once_with( mongo_db.send_notification.assert_called_once_with(
metadata=metadata['metadata'],
notification_id="MONGODB_QUERY_ERROR", notification_id="MONGODB_QUERY_ERROR",
message="Failed to execute MongoDB query: Error", message="Failed to execute MongoDB query: Error",
level=NotificationLevel.ERROR, level=NotificationLevel.ERROR,
@@ -202,15 +216,17 @@ async def test_aggregate_documents_in_mongodb_failure(mongo_db):
try: try:
await mongo_db.aggregate_documents_in_mongodb( await mongo_db.aggregate_documents_in_mongodb(
{ {
"query": input_data "query": input_data,
**metadata
}) })
except Exception as e: except Exception as e:
assert str(e) == "Error" assert str(e) == "Error"
mongo_db.notification_handler.build_and_send_notification.assert_called_once_with( mongo_db.send_notification.assert_called_once_with(
metadata=metadata['metadata'],
notification_id="MONGODB_AGGREGATION_ERROR", notification_id="MONGODB_AGGREGATION_ERROR",
message="Failed to execute MongoDB aggregation: Error", message="Failed to execute MongoDB aggregation: Error",
level=NotificationLevel.ERROR,
block="aggregate_documents_in_mongodb", block="aggregate_documents_in_mongodb",
level=NotificationLevel.ERROR,
attachment_content=ANY attachment_content=ANY
) )
@@ -265,6 +281,35 @@ async def test_update_pipelines_timestamps_success(datetime_mock, mongo_db):
) )
@mark.asyncio
@patch("orchestrator.activities.mongo_db.datetime")
async def test_update_pipelines_timestamps_failure(datetime_mock, mongo_db):
input_data = {
"updated_pipelines": [
{"schedule_name": "test1", "namespace": "test1", "success": True},
{"schedule_name": "test2", "namespace": "test2", "success": True}
],
**metadata
}
mongo_db.database["pipelines"].update_many.side_effect = Exception("Error")
try:
await mongo_db.update_pipelines_timestamps(input_data)
except Exception as e:
assert str(e) == "Error"
mongo_db.send_notification.assert_called_once_with(
metadata=metadata['metadata'],
notification_id="MONGODB_UPDATE_PIPELINES_ERROR",
message="Failed to update pipelines timestamps: Error",
block="update_pipelines_timestamps",
level=NotificationLevel.ERROR,
attachment_content=ANY
)
else:
assert False, "Expected an exception to be raised"
@mark.asyncio @mark.asyncio
@patch("orchestrator.activities.mongo_db.datetime") @patch("orchestrator.activities.mongo_db.datetime")
async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db): async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db):
@@ -284,6 +329,33 @@ async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db):
) )
@mark.asyncio
@patch("orchestrator.activities.mongo_db.datetime")
async def test_create_pipelines_timestamps_failure(datetime_mock, mongo_db):
input_data = {"created_pipelines": [
{"schedule_name": "test1", "namespace": "test1", "success": True},
{"schedule_name": "test2", "namespace": "test2", "success": True}
],
**metadata
}
mongo_db.database["pipelines"].insert_many.side_effect = Exception("Error")
try:
await mongo_db.create_pipelines_timestamps(input_data)
except Exception as e:
assert str(e) == "Error"
mongo_db.send_notification.assert_called_once_with(
metadata=metadata['metadata'],
notification_id="MONGODB_CREATE_PIPELINES_ERROR",
message="Failed to create pipelines timestamps: Error",
block="create_pipelines_timestamps",
level=NotificationLevel.ERROR,
attachment_content=ANY
)
else:
assert False, "Expected an exception to be raised"
@mark.asyncio @mark.asyncio
@patch("orchestrator.activities.mongo_db.datetime") @patch("orchestrator.activities.mongo_db.datetime")
async def test_delete_pipelines_timestamps_success(datetime_mock, mongo_db): async def test_delete_pipelines_timestamps_success(datetime_mock, mongo_db):
@@ -299,3 +371,30 @@ async def test_delete_pipelines_timestamps_success(datetime_mock, mongo_db):
{"schedule_name": "test2", "namespace": "test2"} {"schedule_name": "test2", "namespace": "test2"}
]} ]}
) )
@mark.asyncio
@patch("orchestrator.activities.mongo_db.datetime")
async def test_delete_pipelines_timestamps_failure(datetime_mock, mongo_db):
input_data = {"deleted_pipelines": [
{"schedule_name": "test1", "namespace": "test1", "success": True},
{"schedule_name": "test2", "namespace": "test2", "success": True}
],
**metadata
}
mongo_db.database["pipelines"].delete_many.side_effect = Exception("Error")
try:
await mongo_db.delete_pipelines_timestamps(input_data)
except Exception as e:
assert str(e) == "Error"
mongo_db.send_notification.assert_called_once_with(
metadata=metadata['metadata'],
notification_id="MONGODB_DELETE_PIPELINES_ERROR",
message="Failed to delete pipelines timestamps: Error",
block="delete_pipelines_timestamps",
level=NotificationLevel.ERROR,
attachment_content=ANY
)
else:
assert False, "Expected an exception to be raised"

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

@@ -1,6 +1,16 @@
from unittest.mock import MagicMock, patch, call from unittest.mock import MagicMock, patch, call, ANY
from pytest import mark, fixture from pytest import mark, fixture
from orchestrator.activities.slot_manager import SlotManager from orchestrator.activities.slot_manager import SlotManager
from sientia_do.notifications.models import NotificationLevel
metadata = {
"metadata": {
"schedule_name": "test_schedule_name",
"workflow_name": "test_workflow_name",
"model_name": "test_model_name",
"model_id": "test_model_id"
}
}
@fixture @fixture
@@ -19,6 +29,7 @@ def slot_manager(_redis_mock):
slot_manager.redis_client = MagicMock() slot_manager.redis_client = MagicMock()
slot_manager.logger = MagicMock() slot_manager.logger = MagicMock()
slot_manager.notification_handler = MagicMock() slot_manager.notification_handler = MagicMock()
slot_manager.send_notification = MagicMock()
return slot_manager return slot_manager
@@ -26,7 +37,7 @@ def slot_manager(_redis_mock):
@mark.asyncio @mark.asyncio
async def test_load_opc_slots_no_slot_keys(slot_manager): async def test_load_opc_slots_no_slot_keys(slot_manager):
slot_manager.redis_client.keys.return_value = [] slot_manager.redis_client.keys.return_value = []
assert await slot_manager.load_opc_slots() == {} assert await slot_manager.load_opc_slots(metadata) == {}
@mark.asyncio @mark.asyncio
@@ -42,7 +53,7 @@ async def test_load_opc_slots(slot_manager):
] ]
) )
response = await slot_manager.load_opc_slots() response = await slot_manager.load_opc_slots(metadata)
assert response == { assert response == {
"slot:opc_tags:1": "value1", "slot:opc_tags:1": "value1",
@@ -64,7 +75,7 @@ async def test_load_opc_slots_no_decode(slot_manager):
] ]
) )
response = await slot_manager.load_opc_slots() response = await slot_manager.load_opc_slots(metadata)
assert response == { assert response == {
"slot:opc_tags:1": "value1", "slot:opc_tags:1": "value1",
@@ -73,17 +84,67 @@ async def test_load_opc_slots_no_decode(slot_manager):
} }
@mark.asyncio
async def test_load_opc_slots_error(slot_manager):
slot_manager.redis_client.keys.return_value = [
"slot:opc_tags:1", "slot:opc_tags:2", "slot:opc_tags:3"]
slot_manager.get = MagicMock(
side_effect=Exception("Test exception")
)
try:
await slot_manager.load_opc_slots(metadata)
except Exception as e:
assert str(e) == "Test exception"
slot_manager.send_notification.assert_called_once_with(
metadata=metadata['metadata'],
notification_id="REDIS_GET_ERROR",
message="Failed to load OPC slots: Test exception",
block="load_opc_slots",
level=NotificationLevel.ERROR,
attachment_content=ANY
)
else:
assert False, "Expected an exception to be raised"
@mark.asyncio @mark.asyncio
async def test_load_active_ingestors(slot_manager): async def test_load_active_ingestors(slot_manager):
slot_manager.redis_client.keys.return_value = [ slot_manager.redis_client.keys.return_value = [
b"heartbeat:ingestor:1", b"heartbeat:ingestor:2", "heartbeat:ingestor:3"] b"heartbeat:ingestor:1", b"heartbeat:ingestor:2", "heartbeat:ingestor:3"]
response = await slot_manager.load_active_ingestors() response = await slot_manager.load_active_ingestors(metadata)
assert response == ["heartbeat:ingestor:1", assert response == ["heartbeat:ingestor:1",
"heartbeat:ingestor:2", "heartbeat:ingestor:3"] "heartbeat:ingestor:2", "heartbeat:ingestor:3"]
@mark.asyncio
async def test_load_active_ingestors_error(slot_manager):
slot_manager.redis_client.keys.return_value = [
"heartbeat:ingestor:1", "heartbeat:ingestor:2", "heartbeat:ingestor:3"]
slot_manager.redis_client.keys.side_effect = Exception("Test exception")
try:
await slot_manager.load_active_ingestors(metadata)
except Exception as e:
assert str(e) == "Test exception"
slot_manager.send_notification.assert_called_once_with(
metadata=metadata['metadata'],
notification_id="REDIS_GET_ERROR",
message="Failed to load active ingestors: Test exception",
block="load_active_ingestors",
level=NotificationLevel.ERROR,
attachment_content=ANY
)
else:
assert False, "Expected an exception to be raised"
@mark.asyncio @mark.asyncio
async def test_update_slots(slot_manager): async def test_update_slots(slot_manager):
slot_manager.set = MagicMock( slot_manager.set = MagicMock(

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

@@ -1,12 +1,19 @@
from unittest.mock import MagicMock, patch, AsyncMock, call, ANY from unittest.mock import MagicMock, patch, AsyncMock, call, ANY
from datetime import timedelta from datetime import timedelta
import base64 from sientia_do.notifications.models import NotificationLevel
import json
from pytest import fixture, mark from pytest import fixture, mark
import pytest_asyncio
from orchestrator.activities.temporal_manager import TemporalManager from orchestrator.activities.temporal_manager import TemporalManager
from orchestrator.utils.converters import parse_frequency from orchestrator.utils.converters import parse_frequency
metadata = {
"metadata": {
"schedule_name": "test_schedule_name",
"workflow_name": "test_workflow_name",
"model_name": "test_model_name",
"model_id": "test_model_id"
}
}
@fixture @fixture
@patch("orchestrator.activities.temporal_manager.Client.connect") @patch("orchestrator.activities.temporal_manager.Client.connect")
@@ -21,6 +28,7 @@ def temporal_manager(connect_mock):
temporal_manager.temporal_clients['scouter'] = MagicMock() temporal_manager.temporal_clients['scouter'] = MagicMock()
temporal_manager.temporal_clients['laborious'] = MagicMock() temporal_manager.temporal_clients['laborious'] = MagicMock()
temporal_manager.send_notification = MagicMock()
return temporal_manager return temporal_manager
@@ -62,80 +70,6 @@ async def async_iter():
) )
@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(
schedule=MagicMock(
action=MagicMock(
args=[
MagicMock(
data=base64.b64encode(json.dumps(
{"test": "test"}).encode('utf-8'))
)
]
)
)
)
)
)
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=handle
)
temporal_manager.temporal_clients['laborious'].get_schedule_handle = MagicMock(
return_value=handle
)
describe_mock = MagicMock(
intervals=[
MagicMock(
every=MagicMock(
seconds=60
)
)
]
)
temporal_manager.temporal_clients['scouter'].get_schedule_handle.return_value.describe \
.return_value.schedule.spec = describe_mock
temporal_manager.temporal_clients['laborious'].get_schedule_handle.return_value.describe \
.return_value.schedule.spec = describe_mock
response = await temporal_manager.load_schedule()
temporal_manager.temporal_clients['scouter'].list_schedules.assert_awaited_once(
)
temporal_manager.temporal_clients['laborious'].list_schedules.assert_awaited_once(
)
assert response == {
"scouter": {
"test-schedule-id": {
"frequency": 60,
"data": {"test": "test"}
}
},
"laborious": {
"test-schedule-id": {
"frequency": 60,
"data": {"test": "test"}
}
}
}
@mark.asyncio @mark.asyncio
async def test_normalize_schedules(temporal_manager): async def test_normalize_schedules(temporal_manager):
input_data = { input_data = {
@@ -173,6 +107,29 @@ async def test_normalize_schedules(temporal_manager):
) )
@mark.asyncio
async def test_normalize_schedules_error(temporal_manager):
temporal_manager.temporal_clients['scouter'].list_schedules = AsyncMock(
side_effect=Exception("Test exception")
)
try:
await temporal_manager.normalize_schedules(metadata)
except Exception as e:
assert str(e) == "Test exception"
temporal_manager.send_notification.assert_called_once_with(
metadata=metadata['metadata'],
notification_id="TEMPORAL_NORMALIZE_SCHEDULES_ERROR",
message="Failed to normalize schedules: Test exception",
block="normalize_schedules",
level=NotificationLevel.ERROR,
attachment_content=ANY
)
else:
assert False, "Expected an exception to be raised"
@mark.asyncio @mark.asyncio
@patch("orchestrator.activities.temporal_manager.parse_frequency", @patch("orchestrator.activities.temporal_manager.parse_frequency",
side_effect=parse_frequency) side_effect=parse_frequency)

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

@@ -9,6 +9,16 @@ def orchestrator():
return Orchestrator() return Orchestrator()
metadata = {
'metadata': {
'schedule_name': 'test-schedule-name',
'workflow_name': 'orchestrator',
'model_name': '-',
'model_id': '-',
}
}
@mark.asyncio @mark.asyncio
@patch("orchestrator.workflows.orchestrator.workflow", new_callable=AsyncMock) @patch("orchestrator.workflows.orchestrator.workflow", new_callable=AsyncMock)
async def test_run(workflow_mock, orchestrator): async def test_run(workflow_mock, orchestrator):
@@ -24,7 +34,8 @@ async def test_run(workflow_mock, orchestrator):
call( call(
Activities.aggregate_documents_in_mongodb, Activities.aggregate_documents_in_mongodb,
{ {
"query": input_data["pipelines_query"] "query": input_data["pipelines_query"],
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -35,7 +46,8 @@ async def test_run(workflow_mock, orchestrator):
call( call(
Activities.find_documents_in_mongodb, Activities.find_documents_in_mongodb,
{ {
"query": input_data["opc_servers_query"] "query": input_data["opc_servers_query"],
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -48,7 +60,8 @@ async def test_run(workflow_mock, orchestrator):
{ {
"query": { "query": {
"collection": "orchestrated_schedules" "collection": "orchestrated_schedules"
} },
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -58,6 +71,9 @@ async def test_run(workflow_mock, orchestrator):
workflow_mock.start_local_activity_method.assert_has_calls([ workflow_mock.start_local_activity_method.assert_has_calls([
call( call(
Activities.load_opc_slots, Activities.load_opc_slots,
{
**metadata
},
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
) )
@@ -66,6 +82,9 @@ async def test_run(workflow_mock, orchestrator):
workflow_mock.start_local_activity_method.assert_has_calls([ workflow_mock.start_local_activity_method.assert_has_calls([
call( call(
Activities.load_active_ingestors, Activities.load_active_ingestors,
{
**metadata
},
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
) )
@@ -75,7 +94,8 @@ async def test_run(workflow_mock, orchestrator):
call( call(
Activities.format_schedule_config, Activities.format_schedule_config,
{ {
'schedule_config': workflow_mock.start_local_activity_method.return_value 'schedule_config': workflow_mock.start_local_activity_method.return_value,
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -86,7 +106,8 @@ async def test_run(workflow_mock, orchestrator):
call( call(
Activities.process_schedules, Activities.process_schedules,
{ {
'pipelines': workflow_mock.start_local_activity_method.return_value 'pipelines': workflow_mock.start_local_activity_method.return_value,
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -100,6 +121,7 @@ async def test_run(workflow_mock, orchestrator):
'opc_servers': workflow_mock.start_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, 'active_ingestors': workflow_mock.start_local_activity_method.return_value,
'pipelines': workflow_mock.start_local_activity_method.return_value, 'pipelines': workflow_mock.start_local_activity_method.return_value,
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -111,7 +133,8 @@ async def test_run(workflow_mock, orchestrator):
Activities.create_schedule_config, Activities.create_schedule_config,
{ {
'current_schedule_config': workflow_mock.start_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 'schedule_config': workflow_mock.start_local_activity_method.return_value,
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -123,18 +146,20 @@ async def test_run(workflow_mock, orchestrator):
Activities.create_slot_config, Activities.create_slot_config,
{ {
'current_slot_config': workflow_mock.start_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 'slot_config': workflow_mock.start_local_activity_method.return_value,
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
) )
]) ])
workflow_mock.start_local_activity_method.assert_has_calls([ workflow_mock.start_activity_method.assert_has_calls([
call( call(
Activities.normalize_schedules, Activities.normalize_schedules,
{ {
'orchestrated_schedules': workflow_mock.start_local_activity_method.return_value 'orchestrated_schedules': workflow_mock.start_local_activity_method.return_value,
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -146,7 +171,8 @@ async def test_run(workflow_mock, orchestrator):
Activities.delete_slots, Activities.delete_slots,
{ {
'to_delete': 'to_delete':
workflow_mock.start_local_activity_method.return_value['to_delete'] workflow_mock.start_local_activity_method.return_value['to_delete'],
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -158,7 +184,8 @@ async def test_run(workflow_mock, orchestrator):
Activities.update_slots, Activities.update_slots,
{ {
'to_insert': 'to_insert':
workflow_mock.start_local_activity_method.return_value['to_insert'] workflow_mock.start_local_activity_method.return_value['to_insert'],
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -170,7 +197,8 @@ async def test_run(workflow_mock, orchestrator):
Activities.delete_schedules, Activities.delete_schedules,
{ {
'schedules': 'schedules':
workflow_mock.start_local_activity_method.return_value['to_delete'] workflow_mock.start_local_activity_method.return_value['to_delete'],
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -182,7 +210,8 @@ async def test_run(workflow_mock, orchestrator):
Activities.create_schedules, Activities.create_schedules,
{ {
'schedules': 'schedules':
workflow_mock.start_local_activity_method.return_value['to_create'] workflow_mock.start_local_activity_method.return_value['to_create'],
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -194,7 +223,8 @@ async def test_run(workflow_mock, orchestrator):
Activities.update_schedules, Activities.update_schedules,
{ {
'schedules': 'schedules':
workflow_mock.start_local_activity_method.return_value['to_update'] workflow_mock.start_local_activity_method.return_value['to_update'],
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -207,7 +237,8 @@ async def test_run(workflow_mock, orchestrator):
{ {
'created_schedules': workflow_mock.start_activity_method.return_value, 'created_schedules': workflow_mock.start_activity_method.return_value,
'updated_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 'deleted_schedules': workflow_mock.start_activity_method.return_value,
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -219,7 +250,8 @@ async def test_run(workflow_mock, orchestrator):
Activities.report_slot_orchestration, Activities.report_slot_orchestration,
{ {
'inserted_slots': workflow_mock.start_activity_method.return_value, 'inserted_slots': workflow_mock.start_activity_method.return_value,
'deleted_slots': workflow_mock.start_activity_method.return_value 'deleted_slots': workflow_mock.start_activity_method.return_value,
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -230,7 +262,8 @@ async def test_run(workflow_mock, orchestrator):
call( call(
Activities.update_pipelines_timestamps, Activities.update_pipelines_timestamps,
{ {
'updated_pipelines': workflow_mock.start_activity_method.return_value 'updated_pipelines': workflow_mock.start_activity_method.return_value,
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -241,7 +274,8 @@ async def test_run(workflow_mock, orchestrator):
call( call(
Activities.delete_pipelines_timestamps, Activities.delete_pipelines_timestamps,
{ {
'deleted_pipelines': workflow_mock.start_activity_method.return_value 'deleted_pipelines': workflow_mock.start_activity_method.return_value,
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -252,7 +286,8 @@ async def test_run(workflow_mock, orchestrator):
call( call(
Activities.create_pipelines_timestamps, Activities.create_pipelines_timestamps,
{ {
'created_pipelines': workflow_mock.start_activity_method.return_value 'created_pipelines': workflow_mock.start_activity_method.return_value,
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY

2
values.yaml
View File

@@ -132,7 +132,7 @@ env:
- name: GITHUB_REPO_URL - name: GITHUB_REPO_URL
value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git" value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
- name: GITHUB_BRANCH - name: GITHUB_BRANCH
value: "SIENTIAPDE-1150-ajustar-orquestrador-para-manter-uma-store-de-detalhes-dos-schedules" value: "SIENTIAPDE-1163-alterar-dinamica-de-notificacoes-para-usar-o-mongodb-ao-inves-do-kafka"
- name: PYTHON_APP - name: PYTHON_APP
value: "orchestrator.worker.worker" value: "orchestrator.worker.worker"