diff --git a/.gitignore b/.gitignore index fd3c51f..9a3e1d1 100644 --- a/.gitignore +++ b/.gitignore @@ -41,4 +41,6 @@ htmlcov/ .coverage # git keys -git_key* \ No newline at end of file +git_key* + +git_log \ No newline at end of file diff --git a/orchestrator/activities/activities.py b/orchestrator/activities/activities.py index 02437c2..23c0ace 100644 --- a/orchestrator/activities/activities.py +++ b/orchestrator/activities/activities.py @@ -10,7 +10,7 @@ with workflow.unsafe.imports_passed_through(): from orchestrator.activities.formatters import Formatters from typing import Any from logging import Logger - from sientia_do.notifications.handlers import NotificationHandler + from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler class Activities( # Couchbase, diff --git a/orchestrator/activities/couchbase.py b/orchestrator/activities/couchbase.py index 4e22777..7a6e1cd 100644 --- a/orchestrator/activities/couchbase.py +++ b/orchestrator/activities/couchbase.py @@ -9,7 +9,7 @@ with workflow.unsafe.imports_passed_through(): from couchbase.auth import PasswordAuthenticator from couchbase.cluster import Cluster 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.temporal.activities.base import BaseActivity diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index ca1e1f1..a5c8053 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -7,7 +7,7 @@ with workflow.unsafe.imports_passed_through(): from typing import Any from logging import Logger 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.notifications.models import NotificationLevel from orchestrator.utils.orchestrator_functions import ( @@ -42,7 +42,9 @@ class Formatters(BaseActivity): - 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'] @@ -66,10 +68,9 @@ class Formatters(BaseActivity): "updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT)) } - self.logger.info("Processed schedules") - - self.logger.debug(json.dumps( - schedule_config, indent=4, sort_keys=True)) + self.info("Processed schedules", metadata=metadata) + self.debug(json.dumps( + schedule_config, indent=4, sort_keys=True), metadata=metadata) return schedule_config @@ -91,7 +92,9 @@ class Formatters(BaseActivity): - 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'] opc_servers_list = input_data['opc_servers'] @@ -124,9 +127,9 @@ class Formatters(BaseActivity): slot_config = build_tag_config( tag, slot_config.copy(), opc_servers, number_of_slots) - self.logger.info("Processed slots") - self.logger.debug(json.dumps( - slot_config, indent=4, sort_keys=True)) + self.info("Processed slots", metadata=metadata) + self.debug(json.dumps( + slot_config, indent=4, sort_keys=True), metadata=metadata) return slot_config @@ -158,7 +161,7 @@ class Formatters(BaseActivity): 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): + namespace: str, metadata: dict[str, Any]): """ Compares the timestamps of the schedule and the current schedule. """ @@ -170,8 +173,8 @@ class Formatters(BaseActivity): old_timestamp = current_schedules[schedule_name] - self.logger.debug( - f"Comparing schedule {schedule_name}:{update_timestamp} vs {old_timestamp}") + self.debug( + f"Comparing schedule {schedule_name}:{update_timestamp} vs {old_timestamp}", metadata=metadata) if update_timestamp > old_timestamp: to_update[namespace][schedule_name] = schedule @@ -197,7 +200,9 @@ class Formatters(BaseActivity): - 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'] schedule_config = input_data['schedule_config'] @@ -218,7 +223,7 @@ class Formatters(BaseActivity): for namespace, schedules in schedule_config.items(): current_schedules = current_schedule_config.get(namespace, {}) 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 schedule_name in schedules: @@ -231,9 +236,9 @@ class Formatters(BaseActivity): "to_delete": to_delete } - self.logger.info("Created schedule config") - self.logger.debug(json.dumps( - output, indent=4, sort_keys=True)) + self.info("Created schedule config", metadata=metadata) + self.debug(json.dumps( + output, indent=4, sort_keys=True), metadata=metadata) return output @@ -256,7 +261,9 @@ class Formatters(BaseActivity): - 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'] slot_config = input_data['slot_config'] @@ -274,27 +281,29 @@ class Formatters(BaseActivity): "to_insert": slot_config } - self.logger.info("Created slot config") - self.logger.debug(json.dumps( - output, indent=4, sort_keys=True)) + self.info("Created slot config", metadata=metadata) + self.debug(json.dumps( + output, indent=4, sort_keys=True), metadata=metadata) return output - def send_success_report(self, message: str, notification_id: str) -> None: - self.notification_handler.build_and_send_notification( - notification_id, - message, - "report_orchestration", - NotificationLevel.INFO + def send_success_report(self, metadata: dict[str, Any], message: str, notification_id: str) -> None: + self.send_notification( + metadata=metadata, + notification_id=notification_id, + message=message, + 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: - self.notification_handler.build_and_send_notification( - notification_id, - message, - "report_orchestration", - NotificationLevel.ERROR, + self.send_notification( + metadata=metadata, + notification_id=notification_id, + message=message, + block="report_orchestration", + level=NotificationLevel.ERROR, 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. - 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'] updated_schedules = input_data['updated_schedules'] @@ -343,15 +353,17 @@ class Formatters(BaseActivity): if len(success_keys) > 0: self.send_success_report( - f"Created schedules: \n {', '.join(success_keys)}", - "REPORT_ORCHESTRATION_CREATED_SCHEDULES" + metadata=metadata, + message=f"Created schedules: \n {', '.join(success_keys)}", + notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES" ) if len(error_keys) > 0: self.send_error_report( - f"Failed to create schedules: \n {', '.join(error_keys)}", - "REPORT_ORCHESTRATION_CREATED_SCHEDULES", - created_schedules + metadata=metadata, + message=f"Failed to create schedules: \n {', '.join(error_keys)}", + notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES", + attachment=created_schedules ) # Send report for updated schedules @@ -361,15 +373,17 @@ class Formatters(BaseActivity): if len(success_keys) > 0: self.send_success_report( - f"Updated schedules: \n {', '.join(success_keys)}", - "REPORT_ORCHESTRATION_UPDATED_SCHEDULES" + metadata=metadata, + message=f"Updated schedules: \n {', '.join(success_keys)}", + notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES" ) if len(error_keys) > 0: self.send_error_report( - f"Failed to update schedules: \n {', '.join(error_keys)}", - "REPORT_ORCHESTRATION_UPDATED_SCHEDULES", - updated_schedules + metadata=metadata, + message=f"Failed to update schedules: \n {', '.join(error_keys)}", + notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES", + attachment=updated_schedules ) if len(deleted_schedules) > 0: @@ -378,15 +392,17 @@ class Formatters(BaseActivity): if len(success_keys) > 0: self.send_success_report( - f"Deleted schedules: \n {', '.join(success_keys)}", - "REPORT_ORCHESTRATION_DELETED_SCHEDULES" + metadata=metadata, + message=f"Deleted schedules: \n {', '.join(success_keys)}", + notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES" ) if len(error_keys) > 0: self.send_error_report( - f"Failed to delete schedules: \n {', '.join(error_keys)}", - "REPORT_ORCHESTRATION_DELETED_SCHEDULES", - deleted_schedules + metadata=metadata, + message=f"Failed to delete schedules: \n {', '.join(error_keys)}", + notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES", + attachment=deleted_schedules ) @activity.defn(name="report_slot_orchestration") @@ -402,7 +418,9 @@ class Formatters(BaseActivity): - 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'] deleted_slots = input_data['deleted_slots'] @@ -412,15 +430,17 @@ class Formatters(BaseActivity): if len(success_keys) > 0: self.send_success_report( - f"Inserted slots: \n {', '.join(success_keys)}", - "REPORT_ORCHESTRATION_INSERTED_SLOTS" + metadata=metadata, + message=f"Inserted slots: \n {', '.join(success_keys)}", + notification_id="REPORT_ORCHESTRATION_INSERTED_SLOTS" ) if len(error_keys) > 0: self.send_error_report( - f"Failed to insert slots: \n {', '.join(error_keys)}", - "REPORT_ORCHESTRATION_INSERTED_SLOTS", - inserted_slots + metadata=metadata, + message=f"Failed to insert slots: \n {', '.join(error_keys)}", + notification_id="REPORT_ORCHESTRATION_INSERTED_SLOTS", + attachment=inserted_slots ) if len(deleted_slots) > 0: @@ -428,13 +448,15 @@ class Formatters(BaseActivity): if len(success_keys) > 0: self.send_success_report( - f"Deleted slots: \n {', '.join(success_keys)}", - "REPORT_ORCHESTRATION_DELETED_SLOTS" + metadata=metadata, + message=f"Deleted slots: \n {', '.join(success_keys)}", + notification_id="REPORT_ORCHESTRATION_DELETED_SLOTS" ) if len(error_keys) > 0: self.send_error_report( - f"Failed to delete slots: \n {', '.join(error_keys)}", - "REPORT_ORCHESTRATION_DELETED_SLOTS", - deleted_slots + metadata=metadata, + message=f"Failed to delete slots: \n {', '.join(error_keys)}", + notification_id="REPORT_ORCHESTRATION_DELETED_SLOTS", + attachment=deleted_slots ) diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 2c38012..1c1510f 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -7,7 +7,7 @@ with workflow.unsafe.imports_passed_through(): import traceback from logging import Logger 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.temporal.activities.base import BaseActivity from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT @@ -92,6 +92,7 @@ class MongoDB(BaseActivity): """ query = input_data.get("query", {}) + metadata = input_data.get("metadata", {}) collection_name = query.get("collection") if not collection_name: @@ -99,8 +100,8 @@ class MongoDB(BaseActivity): filters = query.get("filters", {}) - self.logger.info( - f"Loading documents from collection '{collection_name}' with filters: {filters}") + self.info( + f"Loading documents from collection '{collection_name}' with filters: {filters}", metadata=metadata) try: collection = self.database[collection_name] @@ -109,25 +110,25 @@ class MongoDB(BaseActivity): documents = clear_mongo_id(documents) - self.logger.info( - f"Loaded {len(documents)} documents from collection '{collection_name}'") + self.info( + f"Loaded {len(documents)} documents from collection '{collection_name}'", metadata=metadata) - self.logger.debug( - f"Documents loaded: {documents}" - ) + self.debug( + f"Documents loaded: {documents}", metadata=metadata) return documents except Exception as e: trace = traceback.format_exc() - self.notification_handler.build_and_send_notification( + self.send_notification( + metadata=metadata, notification_id="MONGODB_QUERY_ERROR", message=f"Failed to execute MongoDB query: {e}", block="load_query_from_mongodb", level=NotificationLevel.ERROR, attachment_content=trace ) - self.logger.error(trace) + self.error(trace, metadata=metadata) raise e @@ -146,6 +147,7 @@ class MongoDB(BaseActivity): """ query = input_data.get("query", {}) + metadata = input_data.get("metadata", {}) collection_name = query.get("collection") if not collection_name: @@ -155,8 +157,8 @@ class MongoDB(BaseActivity): raise ValueError("Aggregation must be provided.") aggregation.append({"$project": {"_id": 0}}) - self.logger.info( - f"Aggregating documents from collection '{collection_name}' with aggregation: {aggregation}") + self.info( + f"Aggregating documents from collection '{collection_name}' with aggregation: {aggregation}", metadata=metadata) try: collection = self.database[collection_name] @@ -166,25 +168,25 @@ class MongoDB(BaseActivity): aggregated_documents = clear_mongo_id(aggregated_documents) - self.logger.info( - f"Aggregated {len(aggregated_documents)} documents from collection '{collection_name}'") + self.info( + f"Aggregated {len(aggregated_documents)} documents from collection '{collection_name}'", metadata=metadata) - self.logger.debug( - f"Aggregation result: {aggregated_documents}" - ) + self.debug( + f"Aggregation result: {aggregated_documents}", metadata=metadata) return aggregated_documents except Exception as e: trace = traceback.format_exc() - self.notification_handler.build_and_send_notification( + self.send_notification( + metadata=metadata, notification_id="MONGODB_AGGREGATION_ERROR", message=f"Failed to execute MongoDB aggregation: {e}", block="aggregate_documents_in_mongodb", level=NotificationLevel.ERROR, attachment_content=trace ) - self.logger.error(trace) + self.error(trace, metadata=metadata) raise e @@ -196,6 +198,7 @@ class MongoDB(BaseActivity): - updated_pipelines (list): List of updated pipelines. """ updated_pipelines = input_data.get("updated_pipelines", []) + metadata = input_data.get("metadata", {}) now = datetime.now().strftime(DEFAULT_DATE_FORMAT) collection = self.database["orchestrated_schedules"] @@ -205,10 +208,24 @@ class MongoDB(BaseActivity): for pipeline in updated_pipelines if pipeline["success"] ] data_filter = {"$or": argument} if argument else {} - collection.update_many( - data_filter, - {"$set": {"updated_at": now}} - ) + + try: + 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") 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 = input_data.get("created_pipelines", []) + metadata = input_data.get("metadata", {}) collection = self.database["orchestrated_schedules"] now = datetime.now().strftime(DEFAULT_DATE_FORMAT) @@ -229,7 +247,21 @@ class MongoDB(BaseActivity): for pipeline in created_pipelines if pipeline["success"] ] 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") 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 = input_data.get("deleted_pipelines", []) + metadata = input_data.get("metadata", {}) collection = self.database["orchestrated_schedules"] argument = [ @@ -248,4 +281,17 @@ class MongoDB(BaseActivity): ] 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 diff --git a/orchestrator/activities/notification.py b/orchestrator/activities/notification.py deleted file mode 100644 index e69de29..0000000 diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index cf2e49f..404c8d8 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -2,10 +2,12 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): from typing import Any + import traceback import json from logging import Logger 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): @@ -18,7 +20,7 @@ class SlotManager(Redis): password, logger, notification_handler) @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 @@ -26,32 +28,45 @@ class SlotManager(Redis): 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 = {} - 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: - if isinstance(slot_keys[0], bytes): - decoded_keys = [key.decode('utf-8') for key in slot_keys] - else: - decoded_keys = slot_keys + self.debug(f"Slot keys: {slot_keys}", metadata=metadata) - for key in decoded_keys: - opc_slots[key] = self.get(key) + if slot_keys: + 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( - f"Loaded: \n {json.dumps(opc_slots, indent=4, sort_keys=True)}") + self.info(f"Loaded {len(opc_slots)} OPC slots", metadata=metadata) return opc_slots @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 @@ -59,23 +74,41 @@ class SlotManager(Redis): 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: - if isinstance(ingestor, bytes): - ingestors.append(ingestor.decode('utf-8')) - else: - ingestors.append(ingestor) + self.debug( + f"Active ingestors: \n {active_ingestors}", metadata=metadata) - 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") 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'] - self.logger.info("Updating OPC slots...") + metadata = input_data.get("metadata", {}) + self.info("Updating OPC slots...", metadata=metadata) report = {} @@ -105,16 +139,17 @@ class SlotManager(Redis): "message": "Slot updated successfully" } 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] = { "success": False, "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( - f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}") + self.debug( + f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}", metadata=metadata) return report @@ -133,7 +168,7 @@ class SlotManager(Redis): """ to_delete = input_data['to_delete'] - self.logger.info("Deleting OPC slots...") + metadata = input_data.get("metadata", {}) report = {} @@ -145,15 +180,16 @@ class SlotManager(Redis): "message": "Slot deleted successfully" } 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] = { "success": False, "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( - f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}") + self.debug( + f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}", metadata=metadata) return report diff --git a/orchestrator/activities/temporal_manager.py b/orchestrator/activities/temporal_manager.py index 7b4b929..f8461eb 100644 --- a/orchestrator/activities/temporal_manager.py +++ b/orchestrator/activities/temporal_manager.py @@ -4,10 +4,12 @@ from temporalio.client import ( from temporalio.common import SearchAttributeKey, SearchAttributePair, TypedSearchAttributes with workflow.unsafe.imports_passed_through(): + import traceback from typing import Any 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.notifications.models import NotificationLevel from google.protobuf.json_format import MessageToDict import base64 from datetime import timedelta @@ -61,73 +63,6 @@ class TemporalManager(BaseActivity): 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") async def normalize_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]: """ @@ -135,29 +70,45 @@ class TemporalManager(BaseActivity): input_data: - 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', {}) for namespace, client in self.temporal_clients.items(): - schedules = orchestrated_schedules.get(namespace, {}) + try: + schedules = orchestrated_schedules.get(namespace, {}) - self.logger.info( - f"Getting orchestrated schedules for {namespace}") + self.info( + f"Getting orchestrated schedules for {namespace}", metadata=metadata) - 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 + 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") + if schedule_id not in schedules: + self.info( + f"Schedule {schedule_id} not found in mongo db, cleaning up", metadata=metadata) - handle = client.get_schedule_handle( - schedule_id) + handle = client.get_schedule_handle( + 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") 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'] + metadata = input_data.get("metadata", {}) report = [] @@ -202,9 +154,10 @@ class TemporalManager(BaseActivity): workflow_type = schedule['workflow_type'] try: - self.logger.debug(f"Creating schedule {schedule_name}:") - self.logger.debug( - f"{json.dumps(schedule, indent=4, sort_keys=True)}") + self.debug( + f"Creating schedule {schedule_name}:", metadata=metadata) + self.debug( + f"{json.dumps(schedule, indent=4, sort_keys=True)}", metadata=metadata) await client.create_schedule( schedule_name, @@ -235,8 +188,8 @@ class TemporalManager(BaseActivity): "message": "Schedule created successfully" }) except Exception as e: - self.logger.error( - f"Failed to create schedule {schedule_name}: {str(e)}") + self.error( + f"Failed to create schedule {schedule_name}: {str(e)}", metadata=metadata) report.append({ "namespace": namespace, "schedule_name": schedule_name, @@ -244,10 +197,11 @@ class TemporalManager(BaseActivity): "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( - f"\n {json.dumps(report, indent=4, sort_keys=True)}") + self.debug( + f"\n {json.dumps(report, indent=4, sort_keys=True)}", metadata=metadata) return report @@ -266,7 +220,7 @@ class TemporalManager(BaseActivity): """ schedules_to_update = input_data['schedules'] - + metadata = input_data.get("metadata", {}) report = [] for namespace, schedules in schedules_to_update.items(): @@ -288,12 +242,12 @@ class TemporalManager(BaseActivity): async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate: # NOSONAR schedule_action = input_data.description.schedule.action - self.logger.debug("Updating schedule:") + self.debug("Updating schedule:", metadata=metadata) if hasattr(schedule_action, "args"): - self.logger.debug("New schedule:") - self.logger.debug( - f"{json.dumps(schedule, indent=4, sort_keys=True)}") # NOSONAR + self.debug("New schedule:", metadata=metadata) + self.debug( + f"{json.dumps(schedule, indent=4, sort_keys=True)}", metadata=metadata) # NOSONAR schedule_action.args = [schedule] @@ -318,8 +272,8 @@ class TemporalManager(BaseActivity): "message": "Schedule updated successfully" }) except Exception as e: - self.logger.error( - f"Failed to update schedule {schedule_name}: {str(e)}") + self.error( + f"Failed to update schedule {schedule_name}: {str(e)}", metadata=metadata) report.append({ "namespace": namespace, "schedule_name": schedule_name, @@ -327,10 +281,11 @@ class TemporalManager(BaseActivity): "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( - f"\n {json.dumps(report, indent=4, sort_keys=True)}") + self.debug( + f"\n {json.dumps(report, indent=4, sort_keys=True)}", metadata=metadata) return report @@ -349,7 +304,7 @@ class TemporalManager(BaseActivity): """ schedules_to_delete = input_data['schedules'] - + metadata = input_data.get("metadata", {}) report = [] for namespace, schedules in schedules_to_delete.items(): @@ -378,8 +333,8 @@ class TemporalManager(BaseActivity): "message": "Schedule deleted successfully" }) except Exception as e: - self.logger.error( - f"Failed to delete schedule {schedule_name}: {str(e)}") + self.error( + f"Failed to delete schedule {schedule_name}: {str(e)}", metadata=metadata) report.append({ "namespace": namespace, "schedule_name": schedule_name, @@ -387,9 +342,10 @@ class TemporalManager(BaseActivity): "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( - f"\n {json.dumps(report, indent=4, sort_keys=True)}") + self.debug( + f"\n {json.dumps(report, indent=4, sort_keys=True)}", metadata=metadata) return report diff --git a/orchestrator/worker/worker.py b/orchestrator/worker/worker.py index a0f58bd..f7c37d4 100644 --- a/orchestrator/worker/worker.py +++ b/orchestrator/worker/worker.py @@ -14,29 +14,32 @@ with workflow.unsafe.imports_passed_through(): build_redis_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 async def main(): host = os.getenv('TEMPORAL_HOST', 'localhost:7233') + namespace = os.getenv('TEMPORAL_NAMESPACE', 'default') logger = get_logger(__name__) logger.info('Starting Worker...') logger.info('Starting Notification Handler...') + mongo_config = build_mongodb_config() 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, 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( target_host=host, - namespace=os.getenv('TEMPORAL_NAMESPACE', 'default') + namespace=namespace ) logger.info('Starting Activities...') @@ -73,7 +76,6 @@ async def main(): activities.create_pipelines_timestamps, activities.delete_pipelines_timestamps, # Temporal - activities.load_schedule, activities.create_schedules, activities.update_schedules, activities.delete_schedules, diff --git a/orchestrator/workflows/orchestrator.py b/orchestrator/workflows/orchestrator.py index 2652ac0..816919b 100644 --- a/orchestrator/workflows/orchestrator.py +++ b/orchestrator/workflows/orchestrator.py @@ -14,9 +14,19 @@ class 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( Activities.aggregate_documents_in_mongodb, { + **metadata, 'query': input_data['pipelines_query'] }, retry_policy=retry_policy, @@ -26,6 +36,7 @@ class Orchestrator: opc_servers_handler = workflow.start_local_activity_method( Activities.find_documents_in_mongodb, { + **metadata, 'query': input_data['opc_servers_query'] }, retry_policy=retry_policy, @@ -35,6 +46,7 @@ class Orchestrator: orchestrated_schedules_handler = workflow.start_local_activity_method( Activities.find_documents_in_mongodb, { + **metadata, 'query': { 'collection': 'orchestrated_schedules' } @@ -45,12 +57,18 @@ class Orchestrator: current_slot_config_handler = workflow.start_local_activity_method( Activities.load_opc_slots, + { + **metadata, + }, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60) ) active_ingestors_handler = workflow.start_local_activity_method( Activities.load_active_ingestors, + { + **metadata, + }, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60) ) @@ -64,6 +82,7 @@ class Orchestrator: formatted_orchestrated_schedules_handler = workflow.start_local_activity_method( Activities.format_schedule_config, { + **metadata, 'schedule_config': orchestrated_schedules }, retry_policy=retry_policy, @@ -73,6 +92,7 @@ class Orchestrator: schedules_config_handler = workflow.start_local_activity_method( Activities.process_schedules, { + **metadata, 'pipelines': pipeline_config }, retry_policy=retry_policy, @@ -82,6 +102,7 @@ class Orchestrator: slot_config_handler = workflow.start_local_activity_method( Activities.process_slots, { + **metadata, 'opc_servers': opc_servers, 'active_ingestors': active_ingestors, 'pipelines': pipeline_config, @@ -97,6 +118,7 @@ class Orchestrator: schedule_actions_handler = workflow.start_local_activity_method( Activities.create_schedule_config, { + **metadata, 'current_schedule_config': formatted_orchestrated_schedules, 'schedule_config': schedules_config }, @@ -107,6 +129,7 @@ class Orchestrator: slot_actions_handler = workflow.start_local_activity_method( Activities.create_slot_config, { + **metadata, 'current_slot_config': current_slot_config, 'slot_config': slot_config }, @@ -114,9 +137,10 @@ class Orchestrator: 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, { + **metadata, 'orchestrated_schedules': formatted_orchestrated_schedules }, retry_policy=retry_policy, @@ -130,6 +154,7 @@ class Orchestrator: slot_deletion_report_handler = workflow.start_activity_method( Activities.delete_slots, { + **metadata, 'to_delete': slot_actions['to_delete'] }, retry_policy=retry_policy, @@ -139,6 +164,7 @@ class Orchestrator: slot_insertion_report_handler = workflow.start_activity_method( Activities.update_slots, { + **metadata, 'to_insert': slot_actions['to_insert'] }, retry_policy=retry_policy, @@ -148,6 +174,7 @@ class Orchestrator: schedule_deletion_report_handler = workflow.start_activity_method( Activities.delete_schedules, { + **metadata, 'schedules': schedule_actions['to_delete'] }, retry_policy=retry_policy, @@ -157,6 +184,7 @@ class Orchestrator: schedule_insertion_report_handler = workflow.start_activity_method( Activities.create_schedules, { + **metadata, 'schedules': schedule_actions['to_create'] }, retry_policy=retry_policy, @@ -166,6 +194,7 @@ class Orchestrator: schedule_update_report_handler = workflow.start_activity_method( Activities.update_schedules, { + **metadata, 'schedules': schedule_actions['to_update'] }, retry_policy=retry_policy, @@ -183,6 +212,7 @@ class Orchestrator: schedule_report_handler = workflow.start_activity_method( Activities.report_schedule_orchestration, { + **metadata, 'created_schedules': schedule_insertion_report, 'updated_schedules': schedule_update_report, 'deleted_schedules': schedule_deletion_report @@ -196,6 +226,7 @@ class Orchestrator: slot_report_handler = workflow.start_activity_method( Activities.report_slot_orchestration, { + **metadata, 'inserted_slots': slot_insertion_report, 'deleted_slots': slot_deletion_report }, @@ -208,6 +239,7 @@ class Orchestrator: update_pipelines_timestamps_handler = workflow.start_activity_method( Activities.update_pipelines_timestamps, { + **metadata, 'updated_pipelines': schedule_update_report }, retry_policy=retry_policy, @@ -219,6 +251,7 @@ class Orchestrator: create_pipelines_timestamps_handler = workflow.start_activity_method( Activities.create_pipelines_timestamps, { + **metadata, 'created_pipelines': schedule_insertion_report }, retry_policy=retry_policy, @@ -230,6 +263,7 @@ class Orchestrator: delete_pipelines_timestamps_handler = workflow.start_activity_method( Activities.delete_pipelines_timestamps, { + **metadata, 'deleted_pipelines': schedule_deletion_report }, retry_policy=retry_policy, diff --git a/requirements.txt b/requirements.txt index 8c759a6..8c82e94 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,4 +4,4 @@ sqlalchemy redis couchbase 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 diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index b49dead..92ba98c 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -8,13 +8,26 @@ from orchestrator.utils.orchestrator_functions import build_tag_config @fixture def formatters(): - return Formatters( + formatters = Formatters( scouter_namespace="scouter", laborious_namespace="laborious", logger=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 @patch("orchestrator.activities.formatters.scouter", @@ -374,23 +387,33 @@ async def test_create_slot_config(formatters): def test_send_success_report(formatters): - formatters.send_success_report("test_message", "test_notification_id") - formatters.notification_handler.build_and_send_notification.assert_called_once_with( - "test_notification_id", - "test_message", - "report_orchestration", - NotificationLevel.INFO + formatters.send_success_report( + metadata=metadata, + message="test_message", + notification_id="test_notification_id" + ) + 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): formatters.send_error_report( - "test_message", "test_notification_id", {"test": "test"}) - formatters.notification_handler.build_and_send_notification.assert_called_once_with( - "test_notification_id", - "test_message", - "report_orchestration", - NotificationLevel.ERROR, + metadata=metadata, + message="test_message", + notification_id="test_notification_id", + attachment={"test": "test"} + ) + 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( {"test": "test"}, indent=4, sort_keys=True) ) @@ -446,6 +469,7 @@ async def test_report_schedule_orchestration(formatters): formatters.send_error_report = MagicMock() input_data = { + **metadata, "created_schedules": [ { "namespace": "test_namespace", @@ -496,33 +520,39 @@ async def test_report_schedule_orchestration(formatters): ]) formatters.send_success_report.assert_has_calls([ call( - "Created schedules: \n test_namespace/test_schedule_name_to_create", - "REPORT_ORCHESTRATION_CREATED_SCHEDULES" + metadata=metadata['metadata'], + message="Created schedules: \n test_namespace/test_schedule_name_to_create", + notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES" ), call( - "Updated schedules: \n test_namespace/test_schedule_name_to_update", - "REPORT_ORCHESTRATION_UPDATED_SCHEDULES" + metadata=metadata['metadata'], + message="Updated schedules: \n test_namespace/test_schedule_name_to_update", + notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES" ), call( - "Deleted schedules: \n test_namespace/test_schedule_name_to_delete", - "REPORT_ORCHESTRATION_DELETED_SCHEDULES" + metadata=metadata['metadata'], + message="Deleted schedules: \n test_namespace/test_schedule_name_to_delete", + notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES" ) ]) formatters.send_error_report.assert_has_calls([ call( - "Failed to create schedules: \n test_namespace/test_schedule_name_to_create_error: test_error", - "REPORT_ORCHESTRATION_CREATED_SCHEDULES", - input_data['created_schedules'] + metadata=metadata['metadata'], + message="Failed to create schedules: \n test_namespace/test_schedule_name_to_create_error: test_error", + notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES", + attachment=input_data['created_schedules'] ), call( - "Failed to update schedules: \n test_namespace/test_schedule_name_to_update_error: test_error", - "REPORT_ORCHESTRATION_UPDATED_SCHEDULES", - input_data['updated_schedules'] + metadata=metadata['metadata'], + message="Failed to update schedules: \n test_namespace/test_schedule_name_to_update_error: test_error", + notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES", + attachment=input_data['updated_schedules'] ), call( - "Failed to delete schedules: \n test_namespace/test_schedule_name_to_delete_error: test_error", - "REPORT_ORCHESTRATION_DELETED_SCHEDULES", - input_data['deleted_schedules'] + metadata=metadata['metadata'], + message="Failed to delete schedules: \n test_namespace/test_schedule_name_to_delete_error: test_error", + 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() input_data = { + **metadata, "inserted_slots": { "test_slot_name_to_create": { "success": True @@ -564,23 +595,27 @@ async def test_report_slot_orchestration(formatters): ]) formatters.send_success_report.assert_has_calls([ call( - "Inserted slots: \n test_slot_name_to_create", - "REPORT_ORCHESTRATION_INSERTED_SLOTS" + metadata=metadata['metadata'], + message="Inserted slots: \n test_slot_name_to_create", + notification_id="REPORT_ORCHESTRATION_INSERTED_SLOTS" ), call( - "Deleted slots: \n test_slot_name_to_delete", - "REPORT_ORCHESTRATION_DELETED_SLOTS" + metadata=metadata['metadata'], + message="Deleted slots: \n test_slot_name_to_delete", + notification_id="REPORT_ORCHESTRATION_DELETED_SLOTS" ) ]) formatters.send_error_report.assert_has_calls([ call( - "Failed to insert slots: \n test_slot_name_to_create_error", - "REPORT_ORCHESTRATION_INSERTED_SLOTS", - input_data['inserted_slots'] + metadata=metadata['metadata'], + message="Failed to insert slots: \n test_slot_name_to_create_error", + notification_id="REPORT_ORCHESTRATION_INSERTED_SLOTS", + attachment=input_data['inserted_slots'] ), call( - "Failed to delete slots: \n test_slot_name_to_delete_error", - "REPORT_ORCHESTRATION_DELETED_SLOTS", - input_data['deleted_slots'] + metadata=metadata['metadata'], + message="Failed to delete slots: \n test_slot_name_to_delete_error", + notification_id="REPORT_ORCHESTRATION_DELETED_SLOTS", + attachment=input_data['deleted_slots'] ) ]) diff --git a/tests/orchestrator/activities/test_mongo_db.py b/tests/orchestrator/activities/test_mongo_db.py index 7beb29d..355070b 100644 --- a/tests/orchestrator/activities/test_mongo_db.py +++ b/tests/orchestrator/activities/test_mongo_db.py @@ -1,3 +1,4 @@ +from curses import meta from unittest.mock import MagicMock, patch, ANY from pytest import fixture, mark from orchestrator.activities.mongo_db import clear_mongo_id @@ -50,7 +51,7 @@ def test_clear_mongo_id(): @fixture @patch("orchestrator.activities.mongo_db.MongoClient") def mongo_db(mongo_mock): - return ( + mongo = ( MongoDB( connection_string="mongodb://localhost:27017", database_name="test_db", @@ -58,6 +59,8 @@ def mongo_db(mongo_mock): notification_handler=MagicMock() ) ) + mongo.send_notification = MagicMock() + return mongo @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} ) +metadata = { + "metadata": { + "schedule_name": "test_schedule_name", + "workflow_name": "test_workflow_name", + "model_name": "test_model_name", + "model_id": "test_model_id" + } +} + @mark.asyncio 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: await mongo_db.find_documents_in_mongodb( { - "query": input_data + "query": input_data, + **metadata }) except Exception as e: 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", message="Failed to execute MongoDB query: Error", level=NotificationLevel.ERROR, @@ -202,15 +216,17 @@ async def test_aggregate_documents_in_mongodb_failure(mongo_db): try: await mongo_db.aggregate_documents_in_mongodb( { - "query": input_data + "query": input_data, + **metadata }) except Exception as e: 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", message="Failed to execute MongoDB aggregation: Error", - level=NotificationLevel.ERROR, block="aggregate_documents_in_mongodb", + level=NotificationLevel.ERROR, 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 @patch("orchestrator.activities.mongo_db.datetime") 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 @patch("orchestrator.activities.mongo_db.datetime") 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"} ]} ) + + +@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" diff --git a/tests/orchestrator/activities/test_slot_manager.py b/tests/orchestrator/activities/test_slot_manager.py index 39d65aa..9be7686 100644 --- a/tests/orchestrator/activities/test_slot_manager.py +++ b/tests/orchestrator/activities/test_slot_manager.py @@ -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 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 @@ -19,6 +29,7 @@ def slot_manager(_redis_mock): slot_manager.redis_client = MagicMock() slot_manager.logger = MagicMock() slot_manager.notification_handler = MagicMock() + slot_manager.send_notification = MagicMock() return slot_manager @@ -26,7 +37,7 @@ def slot_manager(_redis_mock): @mark.asyncio async def test_load_opc_slots_no_slot_keys(slot_manager): slot_manager.redis_client.keys.return_value = [] - assert await slot_manager.load_opc_slots() == {} + assert await slot_manager.load_opc_slots(metadata) == {} @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 == { "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 == { "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 async def test_load_active_ingestors(slot_manager): slot_manager.redis_client.keys.return_value = [ 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", "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 async def test_update_slots(slot_manager): slot_manager.set = MagicMock( diff --git a/tests/orchestrator/activities/test_temporal_manager.py b/tests/orchestrator/activities/test_temporal_manager.py index ebdd519..058aa70 100644 --- a/tests/orchestrator/activities/test_temporal_manager.py +++ b/tests/orchestrator/activities/test_temporal_manager.py @@ -1,12 +1,19 @@ from unittest.mock import MagicMock, patch, AsyncMock, call, ANY from datetime import timedelta -import base64 -import json +from sientia_do.notifications.models import NotificationLevel from pytest import fixture, mark -import pytest_asyncio from orchestrator.activities.temporal_manager import TemporalManager 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 @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['laborious'] = MagicMock() + temporal_manager.send_notification = MagicMock() 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 async def test_normalize_schedules(temporal_manager): 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 @patch("orchestrator.activities.temporal_manager.parse_frequency", side_effect=parse_frequency) diff --git a/tests/orchestrator/workflows/test_orchestrator.py b/tests/orchestrator/workflows/test_orchestrator.py index 42f5b86..a55362e 100644 --- a/tests/orchestrator/workflows/test_orchestrator.py +++ b/tests/orchestrator/workflows/test_orchestrator.py @@ -9,6 +9,16 @@ def orchestrator(): return Orchestrator() +metadata = { + 'metadata': { + 'schedule_name': 'test-schedule-name', + 'workflow_name': 'orchestrator', + 'model_name': '-', + 'model_id': '-', + } +} + + @mark.asyncio @patch("orchestrator.workflows.orchestrator.workflow", new_callable=AsyncMock) async def test_run(workflow_mock, orchestrator): @@ -24,7 +34,8 @@ async def test_run(workflow_mock, orchestrator): call( Activities.aggregate_documents_in_mongodb, { - "query": input_data["pipelines_query"] + "query": input_data["pipelines_query"], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY @@ -35,7 +46,8 @@ async def test_run(workflow_mock, orchestrator): call( Activities.find_documents_in_mongodb, { - "query": input_data["opc_servers_query"] + "query": input_data["opc_servers_query"], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY @@ -48,7 +60,8 @@ async def test_run(workflow_mock, orchestrator): { "query": { "collection": "orchestrated_schedules" - } + }, + **metadata }, retry_policy=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([ call( Activities.load_opc_slots, + { + **metadata + }, retry_policy=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([ call( Activities.load_active_ingestors, + { + **metadata + }, retry_policy=ANY, start_to_close_timeout=ANY ) @@ -75,7 +94,8 @@ async def test_run(workflow_mock, orchestrator): call( 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, start_to_close_timeout=ANY @@ -86,7 +106,8 @@ async def test_run(workflow_mock, orchestrator): call( 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, 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, 'active_ingestors': workflow_mock.start_local_activity_method.return_value, 'pipelines': workflow_mock.start_local_activity_method.return_value, + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY @@ -111,7 +133,8 @@ async def test_run(workflow_mock, orchestrator): Activities.create_schedule_config, { '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, start_to_close_timeout=ANY @@ -123,18 +146,20 @@ async def test_run(workflow_mock, orchestrator): Activities.create_slot_config, { '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, start_to_close_timeout=ANY ) ]) - workflow_mock.start_local_activity_method.assert_has_calls([ + workflow_mock.start_activity_method.assert_has_calls([ call( 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, start_to_close_timeout=ANY @@ -146,7 +171,8 @@ async def test_run(workflow_mock, orchestrator): Activities.delete_slots, { '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, start_to_close_timeout=ANY @@ -158,7 +184,8 @@ async def test_run(workflow_mock, orchestrator): Activities.update_slots, { '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, start_to_close_timeout=ANY @@ -170,7 +197,8 @@ async def test_run(workflow_mock, orchestrator): Activities.delete_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, start_to_close_timeout=ANY @@ -182,7 +210,8 @@ async def test_run(workflow_mock, orchestrator): Activities.create_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, start_to_close_timeout=ANY @@ -194,7 +223,8 @@ async def test_run(workflow_mock, orchestrator): Activities.update_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, 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, '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, start_to_close_timeout=ANY @@ -219,7 +250,8 @@ async def test_run(workflow_mock, orchestrator): Activities.report_slot_orchestration, { '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, start_to_close_timeout=ANY @@ -230,7 +262,8 @@ async def test_run(workflow_mock, orchestrator): call( 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, start_to_close_timeout=ANY @@ -241,7 +274,8 @@ async def test_run(workflow_mock, orchestrator): call( 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, start_to_close_timeout=ANY @@ -252,7 +286,8 @@ async def test_run(workflow_mock, orchestrator): call( 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, start_to_close_timeout=ANY diff --git a/values.yaml b/values.yaml index 7eecab2..c9e54cd 100644 --- a/values.yaml +++ b/values.yaml @@ -132,7 +132,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git" - 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 value: "orchestrator.worker.worker"