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

Merge pull request #30 from Aignosi/feature/SIENTIAPDE-1273

Browse Source 
SIENTIAPDE-1273: Implement Drift Monitoring, Simple Metrics, and Enhanced Scheduling
This commit is contained in:
Matheus Demoner
2025-11-19 16:55:38 -03:00
committed by GitHub
parent b45aa33188 62e7a7dee9
commit bfc6991a8f
15 changed files with 2472 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

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

33
orchestrator/activities/email.py
View File

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

188
orchestrator/activities/formatters.py
View File

@@ -15,10 +15,12 @@ with workflow.unsafe.imports_passed_through():
from orchestrator.utils.orchestrator_functions import (
build_tag_config,
drift,
gather_read_tags,
minimal_retrain,
predictions_batch,
scouter,
simple_metrics,
)
topic_separator = '\n ========== \n'
@@ -34,7 +36,7 @@ class Formatters(SientiaMonitoring):
scheduled reports.
Key features:
- Pipeline schedule configuration formatting ("scouter", "predictions_batch", "minimal_retrain")
- Pipeline schedule configuration formatting ("scouter", "predictions_batch", "minimal_retrain", "drift")
- OPC slot distribution across active ingestors
- Notification filtering for comprehensive scheduled reports
- Group-based report filtering with ignore list support
@@ -67,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)
@@ -132,6 +136,20 @@ class Formatters(SientiaMonitoring):
'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
),
}
elif pipeline['workflow_type'] == 'drift':
schedule_config[self.laborious_namespace][pipeline['schedule_name']] = {
**drift(pipeline),
'updated_at': pipeline.get(
'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
),
}
elif pipeline['workflow_type'] == 'simple_metrics':
schedule_config[self.laborious_namespace][pipeline['schedule_name']] = {
**simple_metrics(pipeline),
'updated_at': pipeline.get(
'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
),
}
self.info('Processed schedules', metadata=metadata)
self.debug(json.dumps(schedule_config, indent=4, sort_keys=True), metadata=metadata)
@@ -213,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']
@@ -256,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:
@@ -283,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', {})
@@ -339,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', {})
@@ -383,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,
@@ -404,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,
@@ -425,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:
@@ -456,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:
@@ -482,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(
@@ -517,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', {})
@@ -564,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', {})
@@ -619,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']

42
orchestrator/activities/mongo_db.py
View File

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

49
orchestrator/activities/slot_manager.py
View File

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

15
orchestrator/activities/temporal_manager.py
View File

@@ -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']
@@ -320,8 +325,10 @@ class TemporalManager(SientiaMonitoring):
input_data.description.schedule.spec.intervals = [
ScheduleIntervalSpec(
every=timedelta(seconds=parse_frequency(
schedule.get('frequency', '1m')))
every=timedelta(
seconds=parse_frequency(schedule.get('frequency', '1m'))),
offset=timedelta(
seconds=parse_frequency(schedule.get('offset', '0m'))),
)
]

28
orchestrator/utils/email_builder.py
View File

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

86
orchestrator/utils/orchestrator_functions.py
View File

@@ -32,6 +32,63 @@ 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),
'schema': 'sientia_data',
'source_table_name': 'laborious_data',
'target_table_name': 'drift_metrics',
'interval': config.get('interval_minutes', 60),
'drift_metrics': config.get(
'drift_metrics', ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']
),
}
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),
'schema': 'sientia_data',
'predictions_table_name': 'predictions',
'data_table_name': 'laborious_data',
'target_table_name': 'simple_metrics',
'interval_minutes': config.get('interval_minutes', 60),
'metrics': config.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
}
def minimal_retrain(config: dict[str, Any]):
"""
Build minimal retrain configuration from pipeline config.
@@ -47,8 +104,6 @@ def minimal_retrain(config: dict[str, Any]):
"""
return {
**common_config(config),
'workflow_type': 'minimal_retrain',
'schedule_name': config['schedule_name'],
'query': config['query'],
'schema': 'sientia_data',
'table_name': 'log_retrain',
@@ -193,6 +248,8 @@ def predictions_batch(config: dict[str, Any]):
'datetime_columns': config.get('datetime_columns', []),
'schema': 'sientia_data',
'table_name': 'predictions',
'save_transform': config.get('save_transform', True),
'transform_table_name': 'transformed_data',
'retention_time': config.get('model_retention_minutes', 60) * 60,
'opc_output_config': tags,
'input_filters': overlap_filter_config(
@@ -216,13 +273,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 = {}
@@ -245,19 +306,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 = {}

9
orchestrator/worker/worker.py
View File

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

2
requirements.txt
View File

@@ -4,5 +4,5 @@ sqlalchemy
redis
pymongo
jinja2
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.3
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.6.1
prometheus-client

2040
test.ipynb
View File

File diff suppressed because it is too large Load Diff

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

@@ -44,8 +44,21 @@ metadata = {
'orchestrator.activities.formatters.minimal_retrain',
return_value={'test_minimal_retrain': 'test_minimal_retrain'},
)
@patch(
'orchestrator.activities.formatters.drift',
return_value={'test_drift': 'test_drift'},
)
@patch(
'orchestrator.activities.formatters.simple_metrics',
return_value={'test_simple_metrics': 'test_simple_metrics'},
)
async def test_process_schedules(
mock_minimal_retrain, mock_predictions_batch, mock_scouter, formatters
mock_simple_metrics,
mock_drift,
mock_minimal_retrain,
mock_predictions_batch,
mock_scouter,
formatters,
):
input_data = {
'pipelines': [
@@ -70,6 +83,20 @@ async def test_process_schedules(
'model_id': 'test_model_id',
'updated_at': '2021-01-03',
},
{
'schedule_name': 'test_schedule_name4',
'workflow_type': 'drift',
'model_name': 'test_model_name',
'model_id': 'test_model_id',
'updated_at': '2021-01-04',
},
{
'schedule_name': 'test_schedule_name5',
'workflow_type': 'simple_metrics',
'model_name': 'test_model_name',
'model_id': 'test_model_id',
'updated_at': '2021-01-05',
},
]
}
@@ -88,11 +115,22 @@ async def test_process_schedules(
'test_minimal_retrain': 'test_minimal_retrain',
'updated_at': '2021-01-03',
},
'test_schedule_name4': {
'test_drift': 'test_drift',
'updated_at': '2021-01-04',
},
'test_schedule_name5': {
'test_simple_metrics': 'test_simple_metrics',
'updated_at': '2021-01-05',
},
},
}
mock_scouter.assert_called_once_with(input_data['pipelines'][0])
mock_predictions_batch.assert_called_once_with(input_data['pipelines'][1])
mock_minimal_retrain.assert_called_once_with(input_data['pipelines'][2])
mock_drift.assert_called_once_with(input_data['pipelines'][3])
mock_simple_metrics.assert_called_once_with(input_data['pipelines'][4])
@mark.asyncio

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

@@ -18,8 +18,7 @@ metadata = {
@fixture
@patch('orchestrator.activities.temporal_manager.Client.connect')
def temporal_manager(connect_mock):
def temporal_manager():
temporal_manager = TemporalManager(
host='localhost:7233',
scouter_namespace='scouter',

65
tests/orchestrator/utils/test_orchestrator_functions.py
View File

@@ -3,12 +3,14 @@ from unittest.mock import call, patch
from orchestrator.utils.orchestrator_functions import (
build_tag_config,
common_config,
drift,
gather_read_tags,
minimal_retrain,
overlap_filter_config,
predictions_batch,
process_path_priority,
scouter,
simple_metrics,
)
@@ -35,6 +37,67 @@ def test_common_config():
assert result == expected
def test_drift():
config = {
'workflow_type': 'drift',
'schedule_name': 'test_schedule',
'model_id': 'test_model_id',
'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
'interval_minutes': 120,
'drift_metrics': ['kolmogorov_smirnov', 'jensen_shannon'],
}
result = drift(config)
expected = {
'workflow_type': 'drift',
'schedule_name': 'test_schedule',
'frequency': '1m',
'offset': '0m',
'max_retry_policy': 1,
'model_id': 'test_model_id',
'model_name': 'test_model_name',
'model_config': {'test_config': 'test_config'},
'schema': 'sientia_data',
'source_table_name': 'laborious_data',
'target_table_name': 'drift_metrics',
'interval': 120,
'drift_metrics': ['kolmogorov_smirnov', 'jensen_shannon'],
'execution_timeout_seconds': 300,
'task_timeout_seconds': 300,
}
assert result == expected
def test_simple_metrics():
config = {
'workflow_type': 'simple_metrics',
'schedule_name': 'test_schedule',
'model_id': 'test_model_id',
'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
'interval_minutes': 120,
'metrics': ['rmse', 'mse'],
}
result = simple_metrics(config)
expected = {
'workflow_type': 'simple_metrics',
'schedule_name': 'test_schedule',
'frequency': '1m',
'offset': '0m',
'max_retry_policy': 1,
'model_id': 'test_model_id',
'model_name': 'test_model_name',
'model_config': {'test_config': 'test_config'},
'schema': 'sientia_data',
'predictions_table_name': 'predictions',
'data_table_name': 'laborious_data',
'target_table_name': 'simple_metrics',
'interval_minutes': 120,
'metrics': ['rmse', 'mse'],
'execution_timeout_seconds': 300,
'task_timeout_seconds': 300,
}
assert result == expected
def test_minimal_retrain():
config = {
'workflow_type': 'minimal_retrain',
@@ -175,6 +238,8 @@ def test_predictions_batch(mock_process_path_priority, mock_overlap_filter_confi
'query': 'test_query',
'schema': 'sientia_data',
'table_name': 'predictions',
'save_transform': True,
'transform_table_name': 'transformed_data',
'retention_time': 60 * 60,
'opc_output_config': {
'test_server_id': {

6
values.yaml
View File

@@ -11,7 +11,7 @@ image:
# This sets the pull policy for images.
pullPolicy: Always
# Overrides the image tag whose default is the chart appVersion.
tag: "0.5.0"
tag: "0.5.2"
# This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
imagePullSecrets:
@@ -151,7 +151,7 @@ env:
- name: GITHUB_REPO_URL
value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
- name: GITHUB_BRANCH
value: "feature/SIENTIAPDE-1325-adicionar-metricas-especificas-de-operacoes-externas"
value: "feature/SIENTIAPDE-1273"
- name: PYTHON_APP
value: "orchestrator.worker.worker"
@@ -237,7 +237,7 @@ ssh:
# kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp
# helm upgrade --install sientia-orchestrator-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.5.0
# helm upgrade --install sientia-orchestrator-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.6.0
# kubectl create secret generic git-ssh-key-sientia-orchestrator-worker \
# --namespace sientia \