From fdd859de5a9bb57be862c31545281d900e3c23fa Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 17 Nov 2025 16:30:29 -0300 Subject: [PATCH] SIENTIAPDE-1273 Enhance orchestrator utilities and activities with improved documentation and functionality. Update README to reflect new orchestrator functions for drift and simple metrics workflows. Refactor email, formatters, mongo_db, slot_manager, and temporal_manager activities to include detailed docstrings and improve clarity on input parameters and return values. Ensure consistent metadata handling across activities for better logging and tracking. --- README.md | 2 +- orchestrator/activities/email.py | 33 ++-- orchestrator/activities/formatters.py | 170 ++++++++++++------- orchestrator/activities/mongo_db.py | 42 +++-- orchestrator/activities/slot_manager.py | 49 ++++-- orchestrator/activities/temporal_manager.py | 9 +- orchestrator/utils/email_builder.py | 28 ++- orchestrator/utils/orchestrator_functions.py | 51 +++++- orchestrator/worker/worker.py | 9 +- 9 files changed, 271 insertions(+), 122 deletions(-) diff --git a/README.md b/README.md index 134578b..645679f 100644 --- a/README.md +++ b/README.md @@ -412,7 +412,7 @@ The orchestrator includes an advanced notification filtering system that prevent #### **Utilities (`orchestrator/utils/`)** - **Connectors Configuration**: Database and service configuration management - **Email Builder**: HTML email template generation and formatting using Jinja2 templates -- **Orchestrator Functions**: Pipeline configuration transformation utilities for scouter, predictions_batch, and minimal_retrain workflows +- **Orchestrator Functions**: Pipeline configuration transformation utilities for scouter, predictions_batch, minimal_retrain, drift, and simple_metrics workflows - **Converters**: Data type conversion and validation utilities including frequency parsing - **Templates**: HTML email templates for alerts and reports diff --git a/orchestrator/activities/email.py b/orchestrator/activities/email.py index bbad1a0..ab30c78 100644 --- a/orchestrator/activities/email.py +++ b/orchestrator/activities/email.py @@ -71,6 +71,8 @@ class Email(SientiaMonitoring): def close(self): """ Close the Email connection and clean up resources. + + Closes the SMTP server connection and shuts down the SientiaMonitoring instance. """ self.server.quit() SientiaMonitoring.shutdown(self) @@ -120,15 +122,23 @@ class Email(SientiaMonitoring): def handle_attachments(self, attachments: list[dict], msg: MIMEMultipart) -> MIMEMultipart: """ - Attaches a list of attachments to an email message. + Attach a list of attachments to an email message. + + Processes notification attachments and adds them to the email message + as base64-encoded MIME parts. Each attachment contains error details + or additional context for the notification. + Args: - attachments (List[Dict]): A list of dictionaries where each dictionary contains - the keys 'filename' and 'content' representing the attachment details. - msg (MIMEMultipart): The email message object to which the attachments will be added. + attachments (list[dict]): A list of dictionaries where each dictionary contains: + - filename (str): Name of the attachment file + - attachment_content (str): Content of the attachment + msg (MIMEMultipart): The email message object to which the attachments will be added + Returns: - MIMEMultipart: The email message object with the attachments added. + MIMEMultipart: The email message object with the attachments added + Raises: - Exception: If an attachment cannot be added, an error is logged. + Exception: If an attachment cannot be added, an error is logged and raised """ for attachment in attachments: @@ -149,14 +159,17 @@ class Email(SientiaMonitoring): def try_send_email(self, msg: MIMEMultipart, receivers: str): """ - Sends an email to the receivers with automatic reconnection handling. + Send an email to the receivers with automatic reconnection handling. + + Attempts to send an email and automatically reconnects to the SMTP server + if a disconnection occurs during transmission. Args: - msg (MIMEMultipart): The email message to send. - receivers (str): Comma-separated list of email addresses to send to. + msg (MIMEMultipart): The email message to send + receivers (str): Comma-separated list of email addresses to send to Raises: - Exception: If email sending fails after reconnection attempts. + Exception: If email sending fails after reconnection attempts """ try: self.server.sendmail(self.sender_email, receivers, msg.as_string()) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 6123ad2..9b7ce92 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -69,6 +69,8 @@ class Formatters(SientiaMonitoring): def close(self): """ Close the Formatters connection and clean up resources. + + Shuts down the SientiaMonitoring instance and releases all resources. """ SientiaMonitoring.shutdown(self) @@ -229,14 +231,18 @@ class Formatters(SientiaMonitoring): @activity.defn(name='format_schedule_config') async def format_schedule_config(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Formats the schedule config to a dictionary with the schedule name as the key. + Format the schedule config to a dictionary with the schedule name as the key. + + Transforms a list of schedule configurations into a nested dictionary structure + organized by namespace and schedule name for efficient lookup and comparison. Args: - input_data (dict[str, Any]): The input data containing the schedule config to format. - - schedule_config (list[dict[str, Any]]): The schedule config to format. + input_data (dict[str, Any]): The input data containing: + - schedule_config (list[dict[str, Any]]): The schedule config to format + - metadata (dict): Metadata for logging purposes Returns: - dict[str, Any]: The formatted schedule config. + dict[str, Any]: The formatted schedule config organized by namespace and schedule name """ metadata = input_data['metadata'] @@ -272,13 +278,16 @@ class Formatters(SientiaMonitoring): """ Compare new and current schedules to determine which should be updated or created. + Compares schedule timestamps to identify schedules that need updating (newer timestamp) + or creating (schedule doesn't exist). Results are accumulated in the provided dictionaries. + Args: - schedules (dict[str, Any]): New schedules to compare. - current_schedules (dict[str, Any]): Existing schedules to compare against. - to_update (dict[str, Any]): Output accumulator for schedules that need updating. - to_create (dict[str, Any]): Output accumulator for schedules that need creating. - namespace (str): Namespace for the schedules being compared. - metadata (dict[str, Any]): Metadata for logging purposes. + schedules (dict[str, Any]): New schedules to compare + current_schedules (dict[str, Any]): Existing schedules to compare against + to_update (dict[str, Any]): Output accumulator for schedules that need updating + to_create (dict[str, Any]): Output accumulator for schedules that need creating + namespace (str): Namespace for the schedules being compared + metadata (dict[str, Any]): Metadata for logging purposes """ for schedule_name, schedule in schedules.items(): if schedule_name in current_schedules: @@ -299,19 +308,23 @@ class Formatters(SientiaMonitoring): @activity.defn(name='create_schedule_config') async def create_schedule_config(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Creates a schedule config dictionary based on the input data. - Checks the existing schedule config and updates it with the new schedule config, - deleting unnecessary schedules, creating new schedules and updating existing schedules. + Create a schedule config dictionary based on the input data. + + Compares the current schedule configuration in Temporal with the new schedule + configuration to determine which schedules need to be created, updated, or deleted. + Uses timestamp comparison to identify schedules that have changed. Args: - - input_data (dict[str, Any]): The input data containing - the schedules to process. - - current_schedule_config (dict[str, Any]): The current schedule - config in Temporal server. - - schedule_config (dict[str, Any]): The schedule config to process. + input_data (dict[str, Any]): The input data containing: + - current_schedule_config (dict[str, Any]): The current schedule config in Temporal server + - schedule_config (dict[str, Any]): The schedule config to process + - metadata (dict): Metadata for logging purposes Returns: - - dict[str, Any]: A dictionary with keys 'to_update', 'to_create', and 'to_delete' + dict[str, Any]: A dictionary with keys: + - to_update (dict): Schedules that need updating + - to_create (dict): Schedules that need creating + - to_delete (dict): Schedules that need deleting """ metadata = input_data.get('metadata', {}) @@ -355,19 +368,22 @@ class Formatters(SientiaMonitoring): @activity.defn(name='create_slot_config') async def create_slot_config(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Creates a slot config dictionary based on the input data. - Checks the existing slot config and updates it with the new slot config, - deleting unnecessary slots. + Create a slot config dictionary based on the input data. + + Compares the current slot configuration in Redis with the new slot configuration + to determine which slots need to be inserted or deleted. Slots are identified + by numeric IDs, and excess slots are marked for deletion. Args: - - input_data (dict[str, Any]): The input data containing - the slots to process. - - current_slot_config (dict[str, Any]): The current slot - config in Temporal server. - - slot_config (dict[str, Any]): The slot config to process. + input_data (dict[str, Any]): The input data containing: + - current_slot_config (dict[str, Any]): The current slot config in Redis + - slot_config (dict[str, Any]): The slot config to process + - metadata (dict): Metadata for logging purposes Returns: - - dict[str, Any]: A dictionary containing 'to_insert' and 'to_delete' + dict[str, Any]: A dictionary containing: + - to_insert (dict): Slots that need to be inserted/updated + - to_delete (list[str]): Slot IDs that need to be deleted """ metadata = input_data.get('metadata', {}) @@ -399,13 +415,16 @@ class Formatters(SientiaMonitoring): attachment: Any | None = None, ) -> None: """ - Sends a success notification report. + Send a success notification report. + + Sends an INFO-level notification to the notification handler with success + details about orchestration operations. Args: - metadata (dict[str, Any]): Metadata for the notification. - message (str): The success message to send. - notification_id (str): The ID of the notification to send. - attachment (Any | None, optional): Optional attachment content to include. + metadata (dict[str, Any]): Metadata for the notification + message (str): The success message to send + notification_id (str): The ID of the notification to send + attachment (Any | None, optional): Optional attachment content to include """ await self.send_notification_async( metadata=metadata, @@ -420,13 +439,16 @@ class Formatters(SientiaMonitoring): self, metadata: dict[str, Any], message: str, notification_id: str, attachment: str ) -> None: """ - Sends an error notification report. + Send an error notification report. + + Sends an ERROR-level notification to the notification handler with error + details about orchestration operation failures. Args: - metadata (dict[str, Any]): Metadata for the notification. - message (str): The error message to send. - notification_id (str): The ID of the notification. - attachment (str): The attachment content for the notification. + metadata (dict[str, Any]): Metadata for the notification + message (str): The error message to send + notification_id (str): The ID of the notification + attachment (str): The attachment content for the notification """ await self.send_notification_async( metadata=metadata, @@ -441,12 +463,15 @@ class Formatters(SientiaMonitoring): self, input_data: list[dict[str, Any]] ) -> tuple[list[str], dict[str, Any]]: """ - Parses the report schedule data to extract success and error information. + Parse the report schedule data to extract success and error information. + + Processes schedule operation reports and separates successful operations + from failed ones, formatting keys as "namespace/schedule_name" for consistency. Args: input_data (list[dict[str, Any]]): The schedule reports. Each item must contain 'namespace', 'schedule_name', 'success', 'message', and optionally - 'attachment'. + 'attachment' Returns: tuple[list[str], dict[str, Any]]: A tuple of: @@ -472,11 +497,14 @@ class Formatters(SientiaMonitoring): def parse_report(self, input_data: dict[str, dict[str, Any]]) -> tuple[list[str], list[str]]: """ - Parses the report data to extract success and error keys. + Parse the report data to extract success and error keys. + + Processes operation reports and separates successful operations from failed ones + based on the 'success' field in each report item. Args: - input_data (dict[str, Any]): The input data containing report items. - Each item should have a 'success' field indicating success/failure. + input_data (dict[str, dict[str, Any]]): The input data containing report items. + Each item should have a 'success' field indicating success/failure Returns: tuple[list[str], list[str]]: A tuple containing: @@ -498,14 +526,17 @@ class Formatters(SientiaMonitoring): schedule_data: dict[str, Any], ): """ - Manages and sends success and error reports based on the provided keys and data. + Manage and send success and error reports based on the provided keys and data. + + Sends separate notifications for successful and failed operations, formatting + error messages with attachments when available. Args: - metadata (dict[str, Any]): Metadata for logging and notifications. - success_keys (list[str]): List of keys that were successful. - error_keys (dict[str, Any]): Dictionary of error keys mapped to error details. - schedule_type (str): The type of schedule being reported (e.g., 'created schedules'). - schedule_data (dict[str, Any]): The schedule data containing items and notification ID. + metadata (dict[str, Any]): Metadata for logging and notifications + success_keys (list[str]): List of keys that were successful + error_keys (dict[str, Any]): Dictionary of error keys mapped to error details + schedule_type (str): The type of schedule being reported (e.g., 'created schedules') + schedule_data (dict[str, Any]): The schedule data containing items and notification ID """ if len(success_keys) > 0: await self.send_success_report( @@ -533,14 +564,17 @@ class Formatters(SientiaMonitoring): @activity.defn(name='report_schedule_orchestration') async def report_schedule_orchestration(self, input_data: dict[str, Any]) -> None: """ - Reports the orchestration result to the notification handler. + Report the orchestration result to the notification handler. + + Processes schedule orchestration results and sends notifications for + created, updated, and deleted schedules with success and error details. Args: - - input_data (dict[str, Any]): The input data containing - the orchestration result. - - created_schedules (dict[str, Any]): The created schedules. - - updated_schedules (dict[str, Any]): The updated schedules. - - deleted_schedules (list[str]): The deleted schedules. + input_data (dict[str, Any]): The input data containing: + - created_schedules (list[dict[str, Any]]): The created schedules + - updated_schedules (list[dict[str, Any]]): The updated schedules + - deleted_schedules (list[dict[str, Any]]): The deleted schedules + - metadata (dict): Metadata for logging purposes """ metadata = input_data.get('metadata', {}) @@ -580,13 +614,16 @@ class Formatters(SientiaMonitoring): @activity.defn(name='report_slot_orchestration') async def report_slot_orchestration(self, input_data: dict[str, Any]) -> None: """ - Reports the orchestration result to the notification handler. + Report the slot orchestration result to the notification handler. + + Processes slot orchestration results and sends notifications for + inserted and deleted slots with success and error details. Args: - - input_data (dict[str, Any]): The input data containing - the orchestration result. - - inserted_slots (dict[str, Any]): The inserted slots. - - deleted_slots (list[str]): The deleted slots. + input_data (dict[str, Any]): The input data containing: + - inserted_slots (dict[str, Any]): The inserted slots + - deleted_slots (dict[str, Any]): The deleted slots + - metadata (dict): Metadata for logging purposes """ metadata = input_data.get('metadata', {}) @@ -635,16 +672,19 @@ class Formatters(SientiaMonitoring): @activity.defn(name='format_log_report') async def format_log_report(self, input_data: dict[str, Any]) -> dict[Hashable, Any]: """ - Formats the receiver_groups status to a dataframe to be stored in the database. + Format the receiver_groups status to a dataframe to be stored in the database. + + Transforms receiver group notification data into a structured format suitable + for database storage, aggregating notifications by unique notification ID and trigger. Args: input_data (dict[str, Any]): The input data containing: - - receiver_groups (dict): The receiver groups configuration. - - mail_type (str): The type of mail for the report. - - metadata (dict): Metadata for logging purposes. + - receiver_groups (dict): The receiver groups configuration with notifications + - mail_type (str): The type of mail for the report (Alerts/Reports) + - metadata (dict): Metadata for logging purposes Returns: - dict[str, Any]: The formatted log report as a dictionary representation of a DataFrame. + dict[Hashable, Any]: The formatted log report as a dictionary representation of a DataFrame """ metadata = input_data['metadata'] mail_type = input_data['mail_type'] diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 16dbb2d..0339ab5 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -69,6 +69,8 @@ class MongoDB(SientiaMonitoring): def close(self): """ Shutdown the MongoDB client and clean up resources. + + Closes the MongoDB repository connection and shuts down the SientiaMonitoring instance. """ self.mongo_db_repository.close() SientiaMonitoring.shutdown(self) @@ -220,8 +222,13 @@ class MongoDB(SientiaMonitoring): """ Update `updated_at` timestamps for successfully updated pipelines. - input_data: - - updated_pipelines (list[dict]): Pipelines with success flags to consider. + Updates the `updated_at` field in the `orchestrated_schedules` collection + for all pipelines that were successfully updated in Temporal. + + Args: + input_data (dict[str, Any]): Input data containing: + - updated_pipelines (list[dict]): Pipelines with success flags to consider + - metadata (dict): Metadata for logging purposes """ updated_pipelines = input_data.get('updated_pipelines', []) metadata = input_data.get('metadata', {}) @@ -263,8 +270,13 @@ class MongoDB(SientiaMonitoring): """ Insert `updated_at` timestamps for newly created pipelines. - input_data: - - created_pipelines (list[dict]): Pipelines with success flags to consider. + Inserts new documents into the `orchestrated_schedules` collection + for all pipelines that were successfully created in Temporal. + + Args: + input_data (dict[str, Any]): Input data containing: + - created_pipelines (list[dict]): Pipelines with success flags to consider + - metadata (dict): Metadata for logging purposes """ created_pipelines = input_data.get('created_pipelines', []) metadata = input_data.get('metadata', {}) @@ -312,8 +324,13 @@ class MongoDB(SientiaMonitoring): """ Delete timestamp rows for successfully deleted pipelines. - input_data: - - deleted_pipelines (list[dict]): Pipelines with success flags to consider. + Removes documents from the `orchestrated_schedules` collection + for all pipelines that were successfully deleted from Temporal. + + Args: + input_data (dict[str, Any]): Input data containing: + - deleted_pipelines (list[dict]): Pipelines with success flags to consider + - metadata (dict): Metadata for logging purposes """ deleted_pipelines = input_data.get('deleted_pipelines', []) metadata = input_data.get('metadata', {}) @@ -352,10 +369,15 @@ class MongoDB(SientiaMonitoring): @activity.defn(name='create_collection_with_ttl_index') async def create_collection_with_ttl_index(self, input_data: dict[str, Any]) -> None: """ - Create a collection with a TTL index. - input_data: - - collection_name (str): The name of the collection to create. - - ttl_index (str): The name of the TTL index to create. + Create collections with TTL indexes for pipeline topics. + + Creates MongoDB collections for scouter pipeline topics and sets up + TTL indexes on the `inserted_at` field to automatically expire old documents. + + Args: + input_data (dict[str, Any]): Input data containing: + - pipelines (dict[str, Any]): Pipeline configurations with topic names + - metadata (dict): Metadata for logging purposes """ pipelines = input_data.get('pipelines', {}) metadata = input_data.get('metadata', {}) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index 953c3a6..26b6813 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -71,6 +71,8 @@ class SlotManager(SientiaMonitoring): def close(self): """ Close the SlotManager connection and clean up resources. + + Closes the Redis repository connection and shuts down the SientiaMonitoring instance. """ self.redis_repository.close() SientiaMonitoring.shutdown(self) @@ -142,10 +144,16 @@ class SlotManager(SientiaMonitoring): @activity.defn(name='load_active_ingestors') 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. + + Retrieves all active ingestor heartbeat keys from Redis to determine + which ingestors are currently available for slot assignment. + + Args: + input_data (dict[str, Any]): Activity input containing metadata Returns: - list[str]: A list of active ingestors + list[str]: A list of active ingestor keys from Redis """ metadata = input_data.get('metadata', {}) @@ -259,15 +267,19 @@ class SlotManager(SientiaMonitoring): @activity.defn(name='get_last_data_timestamp') async def get_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None: """ - Gets the last data timestamp from redis. + Get the last data timestamp from Redis. + + Retrieves the last processed timestamp for a specific mail type from Redis. + This timestamp is used for incremental data loading to avoid reprocessing + already processed notifications. Args: input_data (dict[str, Any]): The input data containing: - - metadata (dict): Metadata for logging purposes. - - mail_type (str): The type of mail to get timestamp for. + - metadata (dict): Metadata for logging purposes + - mail_type (str): The type of mail to get timestamp for (Alerts/Reports) Returns: - str | None: The last data timestamp as a string, or None if no timestamp exists. + str | None: The last data timestamp as a string, or None if no timestamp exists """ metadata = input_data['metadata'] key = f'notification_last_timestamp:{input_data["mail_type"]}' @@ -293,18 +305,22 @@ class SlotManager(SientiaMonitoring): return data_hold @activity.defn(name='put_last_data_timestamp') - async def put_last_data_timestamp(self, input_data: dict[str, Any]): + async def put_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None: """ - Puts the last data timestamp into redis. + Store the last data timestamp in Redis. + + Extracts the maximum timestamp from the provided data and stores it in Redis + with a TTL for the specified mail type. This enables incremental processing + in subsequent workflow executions. Args: input_data (dict[str, Any]): The input data containing: - - metadata (dict): Metadata for logging purposes. - - data (list[dict]): The data to extract timestamp from. - - mail_type (str): The type of mail to store timestamp for. + - metadata (dict): Metadata for logging purposes + - data (list[dict]): The data to extract timestamp from + - mail_type (str): The type of mail to store timestamp for (Alerts/Reports) Returns: - str | None: The last data timestamp that was stored, or None if no data exists. + str | None: The last data timestamp that was stored, or None if no data exists """ metadata = input_data['metadata'] key = f'notification_last_timestamp:{input_data["mail_type"]}' @@ -426,11 +442,14 @@ class SlotManager(SientiaMonitoring): """ Store notification cache in Redis to track recently sent notifications. + Stores successfully sent notifications in Redis with a TTL to prevent + duplicate alert delivery. Only notifications with status 'sent' are cached. + Args: input_data (dict[str, Any]): The input data containing: - - metadata (dict): Metadata for logging purposes. - - log_report (list[dict]): The log report containing notification statuses. - - sent_ttl (int): Time to live for sent notification cache in seconds. + - metadata (dict): Metadata for logging purposes + - log_report (list[dict]): The log report containing notification statuses + - sent_ttl (int): Time to live for sent notification cache in seconds """ metadata = input_data['metadata'] log_report = DataFrame(input_data['log_report']) diff --git a/orchestrator/activities/temporal_manager.py b/orchestrator/activities/temporal_manager.py index 5fd8197..6b0932a 100644 --- a/orchestrator/activities/temporal_manager.py +++ b/orchestrator/activities/temporal_manager.py @@ -68,6 +68,8 @@ class TemporalManager(SientiaMonitoring): def close(self): """ Close the TemporalManager connection and clean up resources. + + Shuts down the SientiaMonitoring instance and releases all resources. """ SientiaMonitoring.shutdown(self) @@ -80,7 +82,9 @@ class TemporalManager(SientiaMonitoring): async def connect_to_temporal(self): """ Connect to Temporal server namespaces used by scouter and laborious workflows. + Creates and caches `Client` connections for both namespaces for later use. + The connections are stored in `temporal_clients` dictionary for efficient access. """ self.logger.info(f'Connecting to Temporal side namespaces at {self.temporal_host}') self.logger.info(f'Scouter namespace: {self.scouter_namespace}') @@ -108,8 +112,9 @@ class TemporalManager(SientiaMonitoring): Any schedule marked with search attribute `orchestrated=true` that does not exist in MongoDB collection `orchestrated_schedules` will be deleted. - input_data: - - orchestrated_schedules (dict[str, Any]): Current orchestrated schedules from MongoDB. + Args: + input_data (dict[str, Any]): Input data containing: + - orchestrated_schedules (dict[str, Any]): Current orchestrated schedules from MongoDB. """ metadata = input_data['metadata'] diff --git a/orchestrator/utils/email_builder.py b/orchestrator/utils/email_builder.py index db67591..c6e4d23 100644 --- a/orchestrator/utils/email_builder.py +++ b/orchestrator/utils/email_builder.py @@ -32,12 +32,15 @@ class EmailBuilder: """ Replace parameters in a Jinja2 template with provided values. + Renders a Jinja2 template string with the provided parameter dictionary, + replacing all template variables with their corresponding values. + Args: - template (str): The Jinja2 template string. - parameters (dict): Dictionary of parameters to replace in the template. + template (str): The Jinja2 template string + parameters (dict): Dictionary of parameters to replace in the template Returns: - str: The rendered template with parameters replaced. + str: The rendered template with parameters replaced """ # Create a Jinja2 template from the provided string template_obj = Template(template) @@ -48,13 +51,16 @@ class EmailBuilder: """ Build parameters dictionary for email templates based on general events and mail type. + Processes notification events organized by level and model, rendering HTML + sections for each notification level using the general template. + Args: general_events (dict): Dictionary containing events categorized by level (ERROR, WARNING, INFO). - Each level contains a 'models' key with model-specific event data. - mail_type (str): The type of email being sent. + Each level contains a 'models' key with model-specific event data + mail_type (str): The type of email being sent (Alerts/Reports) Returns: - dict: Dictionary with mail_type and rendered event sections for each notification level. + dict: Dictionary with mail_type and rendered event sections for each notification level """ error_events = general_events.get('ERROR', {}) warning_events = general_events.get('WARNING', {}) @@ -79,16 +85,20 @@ class EmailBuilder: def build_email(self, report_data: list[dict[str, Any]], mail_type: str) -> str: """ - Builds the email HTML by organizing report data by notification level and model. + Build the email HTML by organizing report data by notification level and model. + + Organizes notification data by level and model, then renders the complete + HTML email using the report template with all event sections. Args: - report_data (List[Dict[str, Any]]): List of notification reports, each containing: + report_data (list[dict[str, Any]]): List of notification reports, each containing: - level (str): Notification level (ERROR, WARNING, INFO) - model_name (str): Name of the model - Additional notification details + mail_type (str): The type of email being built (Alerts/Reports) Returns: - str: Complete HTML email content ready for sending. + str: Complete HTML email content ready for sending """ general_events: dict[str, dict[str, Any]] = {} diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 3aec5dd..ac50f32 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -35,6 +35,19 @@ def common_config(config: dict[str, Any]): def drift(config: dict[str, Any]): """ Build drift configuration from pipeline config. + + Creates a drift detection workflow configuration with database table mappings + and drift metric specifications for monitoring data distribution changes. + + Args: + config (dict[str, Any]): Pipeline configuration containing: + - interval_minutes (int, optional): Detection interval in minutes (default: 60) + - drift_metrics (list[str], optional): List of drift metrics to compute + (default: ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']) + - Additional fields from common_config + + Returns: + dict[str, Any]: Drift configuration with workflow type set to 'drift' """ return { **common_config(config), @@ -51,6 +64,19 @@ def drift(config: dict[str, Any]): def simple_metrics(config: dict[str, Any]): """ Build simple metrics configuration from pipeline config. + + Creates a simple metrics computation workflow configuration for calculating + model performance metrics like RMSE, MSE, MAE, and R². + + Args: + config (dict[str, Any]): Pipeline configuration containing: + - interval_minutes (int, optional): Computation interval in minutes (default: 60) + - metrics (list[str], optional): List of metrics to compute + (default: ['rmse', 'mse', 'mae', 'r2']) + - Additional fields from common_config + + Returns: + dict[str, Any]: Simple metrics configuration with workflow type set to 'simple_metrics' """ return { **common_config(config), @@ -245,13 +271,17 @@ def predictions_batch(config: dict[str, Any]): def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]: """ - Gathers all read tags from input pipelines. + Gather all read tags from input pipelines. + + Collects all read tags from scouter pipelines and organizes them by + server_id and tag_address, tracking which topics each tag is associated with. Args: - - pipelines (list[dict[str, Any]]): The schedules to process. + pipelines (list[dict[str, Any]]): The pipeline configurations to process Returns: - - dict[str, Any]: The read tags dictionary + dict[str, Any]: Dictionary of read tags keyed by "server_id:tag_address", + each containing tag configuration and associated topics """ tags = {} @@ -274,19 +304,24 @@ def build_tag_config( """ Build tag configuration for a specific slot and OPC server. + Organizes tags by OPC server and calculates the minimum subscription period + based on tag frequencies. Validates that all server IDs exist in the OPC + servers configuration. + Args: tags (list[dict[str, Any]]): List of tag configurations containing: - server_id (str): ID of the OPC server - tag_address (str): Address of the tag - slot_config (dict[str, Any]): Current slot configuration to update. - opc_servers (dict[str, Any]): Dictionary of OPC server configurations. - i (int): Slot number to configure. + - frequency (int): Tag read frequency in milliseconds + opc_servers (dict[str, Any]): Dictionary of OPC server configurations Returns: - dict[str, Any]: Updated slot configuration with the new tag. + tuple[dict[str, Any], list]: A tuple containing: + - Slot configuration dictionary organized by server name + - List of server IDs that were not found in opc_servers Raises: - ValueError: If the specified server_id is not found in opc_servers. + ValueError: If the specified server_id is not found in opc_servers """ slot_config = {} diff --git a/orchestrator/worker/worker.py b/orchestrator/worker/worker.py index 6283cbd..fe65d4f 100644 --- a/orchestrator/worker/worker.py +++ b/orchestrator/worker/worker.py @@ -39,7 +39,8 @@ async def main(): Sets up MongoDB connection, notification handler, Temporal client, and starts multiple workers for different task queues (orchestrator, alerts, reports). - Handles graceful shutdown and error handling. + Handles graceful shutdown and error handling. Initializes Prometheus metrics + server and SDK metrics for monitoring. """ host = os.getenv('TEMPORAL_HOST', 'localhost:7233') namespace = os.getenv('TEMPORAL_NAMESPACE', 'default') @@ -206,7 +207,11 @@ def start_prometheus_server(): Start the Prometheus metrics server on the configured port. Sets up HTTP server for metrics collection and marks the application as UP. - Exits the application if the server fails to start. + Exits the application if the server fails to start. The metrics server + exposes application metrics on the port specified by HTTP_METRICS_PORT. + + Raises: + SystemExit: If the metrics server fails to start """ try: port = int(os.getenv('HTTP_METRICS_PORT', 9090))