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

SIENTIAPDE-1231

Browse Source 
Enhance README and code documentation for clarity and structure

- Added a comprehensive Table of Contents to the README for easier navigation.
- Introduced a new section on Code Quality & Validation, detailing the validation workflow and tools used.
- Improved descriptions and consistency in docstrings across various modules, including metrics, formatters, and MongoDB activities.
- Clarified function return types and arguments in the MongoDB and TemporalManager classes for better understanding.
This commit is contained in:
vitor-aignosi
2025-10-16 10:33:58 -03:00
parent c9f83186c8
commit 9031b4c88f
6 changed files with 164 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
orchestrator/activities/formatters.py
View File

@@ -34,12 +34,11 @@ class Formatters(BaseActivity):
distribution across active ingestors, and implementing notification filtering for
scheduled reports.
Key Features:
- Pipeline schedule configuration formatting (scouter, predictions_batch, minimal_retrain)
Key features:
- Pipeline schedule configuration formatting ("scouter", "predictions_batch", "minimal_retrain")
- OPC slot distribution across active ingestors
- Notification filtering for comprehensive reports
- Notification filtering for comprehensive scheduled reports
- Group-based report filtering with ignore list support
- Resource optimization algorithms
- Configuration validation and transformation
Args:
@@ -69,10 +68,10 @@ class Formatters(BaseActivity):
Temporal schedule configurations, organizing them by workflow type (scouter and
laborious) and applying the appropriate configuration builders for each pipeline type.
Pipeline Types Supported:
- scouter: Data collection workflows with OPC tag configurations
- predictions_batch: ML prediction workflows with OPC write configurations
- minimal_retrain: Model retraining workflows with SQL query configurations
Pipeline types supported:
- "scouter": Data collection workflows with OPC tag configurations
- "predictions_batch": ML prediction workflows with OPC write configurations
- "minimal_retrain": Model retraining workflows with SQL query configurations
Args:
- input_data (dict[str, Any]): The input data containing
@@ -80,7 +79,7 @@ class Formatters(BaseActivity):
- pipelines (list[dict[str, Any]]): The schedules to process.
Returns:
- dict[str, Any]: The schedule config dictionary
- dict[str, Any]: The schedule configuration dictionary keyed by namespace
"""
metadata = input_data.get('metadata', {})
@@ -137,7 +136,7 @@ class Formatters(BaseActivity):
- active_ingestors (list[str]): The active ingestors to divide into slots.
Returns:
- dict[str, Any]: The slot config dictionary
- dict[str, Any]: The slot configuration dictionary keyed by slot id (as string)
"""
metadata = input_data.get('metadata', {})
@@ -243,15 +242,14 @@ class Formatters(BaseActivity):
metadata: dict[str, Any],
):
"""
Compares the timestamps of the schedule and the current schedule to determine
which schedules need to be updated or created.
Compare new and current schedules to determine which should be updated or created.
Args:
schedules (dict[str, Any]): The new schedules to compare.
current_schedules (dict[str, Any]): The existing schedules to compare against.
to_update (dict[str, Any]): Dictionary to populate with schedules that need updating.
to_create (dict[str, Any]): Dictionary to populate with schedules that need creating.
namespace (str): The namespace for the schedules.
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():
@@ -285,7 +283,7 @@ class Formatters(BaseActivity):
- schedule_config (dict[str, Any]): The schedule config to process.
Returns:
- dict[str, Any]: The schedule config dictionary
- dict[str, Any]: A dictionary with keys 'to_update', 'to_create', and 'to_delete'
"""
metadata = input_data.get('metadata', {})
@@ -341,7 +339,7 @@ class Formatters(BaseActivity):
- slot_config (dict[str, Any]): The slot config to process.
Returns:
- dict[str, Any]: The slot config dictionary
- dict[str, Any]: A dictionary containing 'to_insert' and 'to_delete'
"""
metadata = input_data.get('metadata', {})
@@ -378,8 +376,8 @@ class Formatters(BaseActivity):
Args:
metadata (dict[str, Any]): Metadata for the notification.
message (str): The success message to send.
notification_id (str): The ID of the notification.
attachment (str, optional): Optional attachment content for the notification.
notification_id (str): The ID of the notification to send.
attachment (Any | None, optional): Optional attachment content to include.
"""
self.send_notification(
metadata=metadata,
@@ -418,14 +416,14 @@ class Formatters(BaseActivity):
Parses the report schedule data to extract success and error information.
Args:
input_data (dict[str, Any]): The input data containing schedule reports.
Each item should have 'namespace', 'schedule_name', 'success', 'message',
and optionally 'attachment' fields.
input_data (list[dict[str, Any]]): The schedule reports. Each item must
contain 'namespace', 'schedule_name', 'success', 'message', and optionally
'attachment'.
Returns:
tuple[list[str], dict[str, Any]]: A tuple containing:
- List of successful schedule keys in format "namespace/schedule_name"
- Dictionary of error keys mapped to their error details
tuple[list[str], dict[str, Any]]: A tuple of:
- Successful schedule keys in the form "namespace/schedule_name"
- Error map keyed by the same string to error details
"""
success_keys = [
f'{value["namespace"]}/{value["schedule_name"]}'

41
orchestrator/activities/mongo_db.py
View File

@@ -17,13 +17,13 @@ with workflow.unsafe.imports_passed_through():
def clear_mongo_id(docs: list) -> list:
"""
Remove the MongoDB internal `_id` field from the document.
Remove MongoDB internal `_id` fields from nested structures.
Args:
docs (list): The document to clear.
docs (list): The list of documents or nested structures to clean.
Returns:
list: The documents without the `_id` field.
list: The cleaned documents with `_id` fields removed wherever present.
"""
for doc in docs:
if isinstance(doc, list):
@@ -48,8 +48,8 @@ class MongoDB(BaseActivity):
This class provides MongoDB database operations including document
querying, aggregation, timestamp management, and collection management
with TTL indexes. It handles all MongoDB interactions required by
the orchestration system.
with TTL indexes. It centralizes all MongoDB interactions required by
the orchestration system lifecycle.
Args:
connection_string (str): MongoDB connection string
@@ -73,7 +73,7 @@ class MongoDB(BaseActivity):
self.client: MongoClient = MongoClient(
self.connection_string, serverSelectionTimeoutMS=5000
)
self.client.server_info() # Trigger an exception if connection fails
self.client.server_info() # Force early failure if connection is invalid
self.database = self.client[self.database_name]
@@ -86,7 +86,7 @@ class MongoDB(BaseActivity):
def shutdown(self):
"""
Shutdown the MongoDB connection and clean up resources.
Shutdown the MongoDB client and clean up resources.
"""
try:
if self.client:
@@ -98,7 +98,7 @@ class MongoDB(BaseActivity):
def __del__(self):
"""
Destructor to ensure MongoDB client is closed when the object is deleted.
Ensure the MongoDB client is closed when the object is garbage-collected.
"""
self.shutdown()
@@ -111,7 +111,7 @@ class MongoDB(BaseActivity):
filters (dict[str, Any]): The query filters to apply.
Returns:
list[dict[str, Any]]: List of documents matching the filters, with _id fields removed.
list[dict[str, Any]]: Documents matching the filters (with `_id` removed).
"""
collection = self.database[collection_name]
@@ -131,9 +131,10 @@ class MongoDB(BaseActivity):
Args:
- input_data (dict): Input data containing query parameters. Contains:
- query (dict): Query parameters to filter documents.
- timestamp_fields (list[str], optional): Fields to format as RFC3339 with TZ.
Returns:
list[dict]: List of documents matching the query.
list[dict]: Documents matching the query with timestamp fields normalized.
"""
query = input_data.get('query', {})
@@ -195,10 +196,11 @@ class MongoDB(BaseActivity):
Args:
- input_data (dict): Input data containing aggregation parameters. Contains:
- query (dict): Query parameters to filter documents.
- query (dict): Aggregation parameters including 'collection' and 'aggregation'.
- timestamp_fields (list[str], optional): Fields to format as RFC3339 with TZ.
Returns:
list[dict]: List of aggregated documents.
list[dict]: Aggregated documents with timestamp fields normalized.
"""
query = input_data.get('query', {})
@@ -260,9 +262,10 @@ class MongoDB(BaseActivity):
@activity.defn(name='update_pipelines_timestamps')
async def update_pipelines_timestamps(self, input_data: dict[str, Any]) -> None:
"""
Update the timestamps of the pipelines in the MongoDB collection.
Update `updated_at` timestamps for successfully updated pipelines.
input_data:
- updated_pipelines (list): List of updated pipelines.
- updated_pipelines (list[dict]): Pipelines with success flags to consider.
"""
updated_pipelines = input_data.get('updated_pipelines', [])
metadata = input_data.get('metadata', {})
@@ -304,9 +307,10 @@ class MongoDB(BaseActivity):
@activity.defn(name='create_pipelines_timestamps')
async def create_pipelines_timestamps(self, input_data: dict[str, Any]) -> None:
"""
Create the timestamps of the pipelines in the MongoDB collection.
Insert `updated_at` timestamps for newly created pipelines.
input_data:
- created_pipelines (list): List of created pipelines.
- created_pipelines (list[dict]): Pipelines with success flags to consider.
"""
created_pipelines = input_data.get('created_pipelines', [])
metadata = input_data.get('metadata', {})
@@ -354,9 +358,10 @@ class MongoDB(BaseActivity):
@activity.defn(name='delete_pipelines_timestamps')
async def delete_pipelines_timestamps(self, input_data: dict[str, Any]) -> None:
"""
Delete the timestamps of the pipelines in the MongoDB collection.
Delete timestamp rows for successfully deleted pipelines.
input_data:
- deleted_pipelines (list): List of deleted pipelines.
- deleted_pipelines (list[dict]): Pipelines with success flags to consider.
"""
deleted_pipelines = input_data.get('deleted_pipelines', [])
metadata = input_data.get('metadata', {})

24
orchestrator/activities/temporal_manager.py
View File

@@ -62,8 +62,8 @@ class TemporalManager(BaseActivity):
async def connect_to_temporal(self):
"""
Connect to Temporal server namespaces for scouter and laborious workflows.
Creates client connections to both namespaces and stores them for later use.
Connect to Temporal server namespaces used by scouter and laborious workflows.
Creates and caches `Client` connections for both namespaces for later use.
"""
self.logger.info(f'Connecting to Temporal side namespaces at {self.temporal_host}')
self.logger.info(f'Scouter namespace: {self.scouter_namespace}')
@@ -86,9 +86,13 @@ class TemporalManager(BaseActivity):
@activity.defn(name='normalize_schedules')
async def normalize_schedules(self, input_data: dict[str, Any]):
"""
Normalize schedules. Removes schedules with no update time in mongo db collection "orchestrated_schedules".
Normalize schedules by removing orphaned schedules from Temporal.
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]): The orchestrated schedules to compare.
- orchestrated_schedules (dict[str, Any]): Current orchestrated schedules from MongoDB.
"""
metadata = input_data['metadata']
@@ -139,7 +143,7 @@ class TemporalManager(BaseActivity):
@activity.defn(name='create_schedules')
async def create_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
"""
Create schedules in Temporal
Create schedules in Temporal.
Args:
- input_data (dict[str, Any]): The input data containing
@@ -147,7 +151,7 @@ class TemporalManager(BaseActivity):
- schedules (dict[str, Any]): The schedules to create.
Returns:
- dict[str, Any]: A report of the created schedules.
- list[dict[str, Any]]: Report entries for each attempted schedule creation.
"""
schedules_to_create = input_data['schedules']
@@ -247,7 +251,7 @@ class TemporalManager(BaseActivity):
@activity.defn(name='update_schedules')
async def update_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
"""
Update schedules in Temporal
Update schedules in Temporal.
Args:
- input_data (dict[str, Any]): The input data containing
@@ -255,7 +259,7 @@ class TemporalManager(BaseActivity):
- schedules (dict[str, Any]): The schedules to update.
Returns:
- dict[str, Any]: A report of the updated schedules.
- list[dict[str, Any]]: Report entries for each attempted schedule update.
"""
schedules_to_update = input_data['schedules']
@@ -342,7 +346,7 @@ class TemporalManager(BaseActivity):
@activity.defn(name='delete_schedules')
async def delete_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
"""
Delete schedules in Temporal
Delete schedules in Temporal.
Args:
- input_data (dict[str, Any]): The input data containing
@@ -350,7 +354,7 @@ class TemporalManager(BaseActivity):
- schedules (list[str]): The schedules to delete.
Returns:
- dict[str, Any]: A report of the deleted schedules.
- list[dict[str, Any]]: Report entries for each attempted schedule deletion.
"""
schedules_to_delete = input_data['schedules']

9
orchestrator/metrics.py
View File

@@ -1,9 +1,8 @@
"""
Prometheus metrics definitions for the orchestrator application.
Prometheus metric definitions for the orchestrator application.
This module defines all Prometheus metrics used for monitoring the
orchestrator system including application health, email delivery,
and workflow execution metrics.
This module exposes Prometheus counters and gauges for monitoring the
orchestrator, including application health and email delivery metrics.
"""
from prometheus_client import Counter, Gauge
@@ -19,6 +18,6 @@ CORE_LABELS = ['pod_id', 'model_name', 'pipeline_name']
EMAIL_SENT_COUNT = Counter(
'email_sent_count',
'Number of emails sent',
'Total number of emails sent by the orchestrator',
[*CORE_LABELS, 'email_group'],
)

2
orchestrator/utils/email_builder.py
View File

@@ -39,7 +39,7 @@ class EmailBuilder:
Returns:
str: The rendered template with parameters replaced.
"""
# Criar um template Jinja2
# Create a Jinja2 template from the provided string
template_obj = Template(template)
return template_obj.render(parameters)