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

@@ -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
 

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())

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']

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', {})

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'])

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']
 

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]] = {}
 

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 = {}

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))