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

SIENTIAPDE-1182

Browse Source 
Remove deprecated files and configurations

- Deleted coverage.sh, Dockerfile, docker-compose.yml, email.html, input_sample.json, and Makefile as part of the cleanup process.
- Updated README.md to reflect the removal of these components and provide a clearer overview of the project structure and functionality.
- Added module-level docstrings to orchestrator, activities, and workflows for improved documentation and clarity on system architecture.
This commit is contained in:
vitor-aignosi
2025-08-29 16:13:57 -03:00
parent 36f1388301
commit 01b4c7a126
25 changed files with 987 additions and 735 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
orchestrator/__init__.py
View File

@@ -0,0 +1,15 @@
"""
SIENTIA DataOps Orchestrator Temporal.
A high-performance, scalable workflow orchestration system built on Temporal.io
for automated pipeline management, notification delivery, and resource coordination.
The orchestrator provides enterprise-grade workflow automation, real-time alerting,
and comprehensive monitoring capabilities for the SIENTIA platform.
Modules:
activities: Temporal activity implementations for database, email, and resource operations
workflows: Temporal workflow definitions for orchestration, alerts, and reports
worker: Main worker implementation and application lifecycle management
utils: Utility functions for configuration, email building, and data conversion
metrics: Prometheus metrics definitions for monitoring and observability
"""

12
orchestrator/activities/__init__.py
View File

@@ -0,0 +1,12 @@
"""
Temporal activity implementations for the orchestrator.
This package contains all Temporal activity classes that implement
specific operations for the orchestration workflows including:
- Database operations (MongoDB, PostgreSQL, Redis)
- Email services and notification delivery
- Temporal schedule and resource management
- Configuration formatting and slot distribution
- Data validation and processing
"""

23
orchestrator/activities/activities.py
View File

@@ -16,6 +16,23 @@ with workflow.unsafe.imports_passed_through():
class Activities( # Couchbase,
TemporalManager, SlotManager, Formatters, MongoDB, Email,
Postgres):
"""
Central activities orchestrator for Temporal workflow operations.
This class combines multiple activity components including temporal management,
slot management, data formatting, MongoDB operations, email services, and
PostgreSQL operations. It provides a unified interface for all activity
operations required by the orchestration workflows.
Args:
temporal_config (dict[str, Any]): Temporal server configuration
redis_config (dict[str, Any]): Redis server configuration
mongodb_config (dict[str, Any]): MongoDB connection configuration
email_config (dict[str, Any]): Email service configuration
postgres_config (dict[str, Any]): PostgreSQL database configuration
logger (Logger): Application logger instance
notification_handler (NotificationHandler): Notification management handler
"""
def __init__(self,
temporal_config: dict[str, Any],
@@ -83,7 +100,11 @@ class Activities( # Couchbase,
def shutdown(self):
"""
Shutdown the MongoDB connection and clean up resources.
Shutdown all connections and clean up resources.
This method gracefully shuts down all database connections, email
services, and other resources to ensure proper cleanup when the
application terminates.
"""
MongoDB.shutdown(self)
Postgres.close(self)

19
orchestrator/activities/couchbase.py
View File

@@ -15,6 +15,25 @@ with workflow.unsafe.imports_passed_through():
class Couchbase(BaseActivity):
"""
Couchbase database operations activity (currently unused).
This class provides Couchbase database connectivity and query operations
for Temporal workflows. It handles connection management, query execution,
and error reporting with automatic connection lifecycle management.
Args:
connection_string (str): Couchbase cluster connection string
username (str): Couchbase authentication username
password (str): Couchbase authentication password
logger (Logger): Application logger instance
notification_handler (NotificationHandler): Notification management handler
Note:
This class is currently commented out in the main Activities class
but maintained for potential future use.
"""
def __init__(self, connection_string: str, username: str,
password: str, logger: Logger,
notification_handler: NotificationHandler):

60
orchestrator/activities/email.py
View File

@@ -19,6 +19,22 @@ with workflow.unsafe.imports_passed_through():
class Email(BaseActivity):
"""
Email service activity for sending workflow notifications.
This class provides email sending capabilities including HTML email
generation, attachment handling, and SMTP connection management with
automatic reconnection for workflow notification delivery.
Args:
sender_email (str): Email address for sending messages
sender_password (str): SMTP authentication password
smtp_server (str): SMTP server hostname
smtp_port (int): SMTP server port number
logger (Logger): Application logger instance
notification_handler (NotificationHandler): Notification management handler
"""
def __init__(self, sender_email: str, sender_password: str,
smtp_server: str, smtp_port: int,
logger: Logger, notification_handler: NotificationHandler):
@@ -51,10 +67,24 @@ class Email(BaseActivity):
@activity.defn(name="build_email_html")
async def build_email_html(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Builds the email html for each receiver group.
input_data:
- receiver_groups (dict): The receiver groups.
- mail_type (str): The mail type.
Build HTML email content for configured receiver groups.
This activity generates HTML email content for each receiver group
based on notification data and mail type. It processes notification
data through the email builder to create formatted HTML messages.
Args:
input_data (dict[str, Any]): Activity input parameters.
Required fields:
- metadata (dict[str, Any]): Workflow execution metadata
- receiver_groups (dict[str, Any]): Receiver group configurations with notifications
- mail_type (str): Type of email (Alerts/Reports)
Returns:
dict[str, Any]: Updated receiver groups with generated HTML content
Raises:
Exception: If HTML generation fails
"""
metadata = input_data['metadata']
receiver_groups = input_data['receiver_groups']
@@ -147,10 +177,24 @@ class Email(BaseActivity):
@activity.defn(name="send_email")
async def send_email(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Sends an email to the receivers of each group.
input_data:
- receiver_groups (dict): The receiver groups.
- mail_type (str): The mail type.
Send email notifications to configured receiver groups.
This activity sends HTML emails with attachments to all configured
receiver groups. It handles SMTP connection management, attachment
processing, and error reporting with automatic reconnection support.
Args:
input_data (dict[str, Any]): Activity input parameters.
Required fields:
- metadata (dict[str, Any]): Workflow execution metadata
- receiver_groups (dict[str, Any]): Receiver groups with HTML content
- mail_type (str): Type of email being sent (Alerts/Reports)
Returns:
dict[str, Any]: Updated receiver groups with sending status
Raises:
Exception: If email sending fails for all groups
"""
metadata = input_data['metadata']
receiver_groups = input_data['receiver_groups']

15
orchestrator/activities/formatters.py
View File

@@ -22,6 +22,21 @@ topic_separator = "\n ========== \n"
class Formatters(BaseActivity):
"""
Schedule and slot configuration formatting activity.
This class provides formatting operations for schedules and OPC slots,
converting pipeline configurations into Temporal-compatible formats
and managing slot distribution across active ingestors for optimal
resource utilization.
Args:
scouter_namespace (str): Scouter workflow namespace
laborious_namespace (str): Laborious workflow namespace
logger (Logger): Application logger instance
notification_handler (NotificationHandler): Notification management handler
"""
def __init__(self,
scouter_namespace: str,
laborious_namespace: str,

43
orchestrator/activities/mongo_db.py
View File

@@ -41,6 +41,22 @@ def clear_mongo_id(docs: list) -> list:
class MongoDB(BaseActivity):
"""
MongoDB operations activity for Temporal workflows.
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.
Args:
connection_string (str): MongoDB connection string
database_name (str): Target database name
ttl_index_seconds (int): TTL index duration in seconds
logger (Logger): Application logger instance
notification_handler (NotificationHandler): Notification management handler
"""
def __init__(self, connection_string: str, database_name: str, ttl_index_seconds: int,
logger: Logger,
notification_handler: NotificationHandler):
@@ -415,14 +431,25 @@ class MongoDB(BaseActivity):
@activity.defn(name="load_latest_data")
async def load_latest_data(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
"""
Loads the latest data from MongoDB.
input_data:
- metadata (dict): The metadata of the workflow.
- collection_name (str): The name of the collection to load data from.
- last_data_timestamp (str): The timestamp of the last data to load.
- base_data_filter (dict): The base data filter to apply to the query.
returns:
- data (list[dict]): The data loaded from MongoDB.
Load the latest data from MongoDB collection since a specified timestamp.
This activity retrieves data from a MongoDB collection, optionally
filtering by timestamp to enable incremental data processing. It
handles connection management and provides comprehensive error reporting.
Args:
input_data (dict[str, Any]): Activity input parameters.
Required fields:
- metadata (dict[str, Any]): Workflow execution metadata
- collection_name (str): Name of the MongoDB collection
- last_data_timestamp (str | None): Last processed timestamp for filtering
- base_data_filter (dict[str, Any]): Base query filter conditions
Returns:
list[dict[str, Any]]: Retrieved data, or empty list if no data found
Raises:
Exception: If MongoDB operation fails
"""
metadata = input_data['metadata']
collection_name = input_data['collection_name']

16
orchestrator/activities/slot_manager.py
View File

@@ -15,6 +15,22 @@ with workflow.unsafe.imports_passed_through():
class SlotManager(Redis):
"""
Redis-based OPC slot management activity.
This class manages OPC server slots and notification processing through
Redis operations. It provides functionality for loading, updating, and
deleting OPC slots, managing active ingestors, and handling notification
caching with timestamp management.
Args:
host (str): Redis server hostname
port (int): Redis server port number
username (str): Redis authentication username
password (str): Redis authentication password
logger (Logger): Application logger instance
notification_handler (NotificationHandler): Notification management handler
"""
def __init__(self, host: str, port: int,
username: str, password: str,

16
orchestrator/activities/temporal_manager.py
View File

@@ -19,6 +19,22 @@ with workflow.unsafe.imports_passed_through():
class TemporalManager(BaseActivity):
"""
Temporal workflow and schedule management activity.
This class manages Temporal schedules across multiple namespaces,
providing operations for schedule creation, updates, deletion, and
normalization. It handles connections to both scouter and laborious
namespaces for comprehensive workflow orchestration.
Args:
host (str): Temporal server host address
scouter_namespace (str): Scouter workflow namespace
laborious_namespace (str): Laborious workflow namespace
logger (Logger): Application logger instance
notification_handler (NotificationHandler): Notification management handler
"""
def __init__(self, host: str, scouter_namespace: str, laborious_namespace: str,
logger: Logger, notification_handler: NotificationHandler):

8
orchestrator/metrics.py
View File

@@ -1,3 +1,11 @@
"""
Prometheus metrics 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.
"""
from prometheus_client import Gauge, Counter
APP_UP = Gauge(

19
orchestrator/utils/converters.py
View File

@@ -1,6 +1,23 @@
def parse_frequency(frequency: str) -> int:
"""
Parse frequency string to seconds
Parse frequency string into seconds for Temporal schedule intervals.
This function converts human-readable frequency strings into seconds
for use in Temporal schedule configurations. Supports seconds, minutes,
hours, and days notation.
Args:
frequency (str): Frequency string with suffix:
- 's' for seconds (e.g., '30s')
- 'm' for minutes (e.g., '5m')
- 'h' for hours (e.g., '2h')
- 'd' for days (e.g., '1d')
Returns:
int: Frequency converted to seconds
Raises:
ValueError: If frequency format is invalid
"""
if frequency.endswith("s"):
return int(frequency[:-1])

12
orchestrator/utils/email_builder.py
View File

@@ -6,6 +6,18 @@ import re
class EmailBuilder:
"""
HTML email template builder for notification emails.
This class handles the generation of HTML email content from notification
data using Jinja2 templates. It supports different email types (alerts,
reports) and notification levels (ERROR, WARNING, INFO) with customizable
templates and parameter replacement.
Args:
logger (Logger): Application logger instance for error reporting
"""
def __init__(self, logger: Logger):
self.logger = logger

11
orchestrator/workflows/__init__.py
View File

@@ -0,0 +1,11 @@
"""
Temporal workflow definitions for the orchestrator.
This package contains all Temporal workflow classes that define
the business logic and coordination patterns for:
- Main orchestration workflow for pipeline and resource management
- Alert workflows for real-time error notification delivery
- Report workflows for scheduled notification summaries
- Subworkflows for notification loading and processing
"""

35
orchestrator/workflows/orchestrator.py
View File

@@ -9,16 +9,37 @@ with workflow.unsafe.imports_passed_through():
@workflow.defn(name="orchestrator")
class Orchestrator:
"""
Main orchestrator workflow for pipeline and resource management.
This workflow coordinates pipeline deployment and OPC server slot
management by retrieving configurations from MongoDB and Redis,
processing schedules, and deploying them to the Temporal server
and Redis infrastructure.
"""
@workflow.run
async def run(self, input_data: dict[str, Any]):
"""
Orchestrates the pipeline and slot management. Gets configuration from MongoDB and Redis,
creates the configuration and deploys the schedules and slots in the Temporal server and
Redis server.
input_data:
- schedule_name (str): The name of the schedule.
- pipelines_query (dict): The query to get the pipelines.
- opc_servers_query (dict): The query to get the OPC servers.
Execute the orchestration workflow for pipeline and slot management.
This workflow retrieves pipeline configurations and OPC server data,
processes schedules and slot configurations, and deploys them to
the appropriate services. It handles creation, updates, and deletion
of schedules and slots based on current system state.
Args:
input_data (dict[str, Any]): Workflow input parameters.
Required fields:
- schedule_name (str): Name of the orchestration schedule
- pipelines_query (dict[str, Any]): MongoDB query for pipeline configurations
- opc_servers_query (dict[str, Any]): MongoDB query for OPC server data
Returns:
None: Workflow completes without return value
Raises:
Exception: If orchestration operations fail
"""
input_data['workflow_name'] = 'orchestrator'

24
orchestrator/workflows/reports.py
View File

@@ -9,20 +9,34 @@ with workflow.unsafe.imports_passed_through():
@workflow.defn(name="reports")
class Reports:
"""
Reports workflow for sending scheduled notification summaries.
This workflow processes and sends scheduled reports to configured
user groups. It loads notification data from MongoDB, filters it
by receiver group configurations, and sends formatted HTML reports
via email.
"""
@workflow.run
async def run(self, input_data: dict[str, Any]):
"""
Workflow to send reports to the users
Execute the reports workflow for scheduled notification delivery.
This workflow loads all notifications from the notification queue,
applies receiver group filtering, and sends comprehensive HTML
reports to configured user groups.
Args:
input_data (dict[str, Any]): Input data. It contains the following keys:
- schedule_name: str - Name of the schedule
input_data (dict[str, Any]): Workflow input parameters.
Required fields:
- schedule_name (str): Name of the report schedule
Returns:
None
None: Workflow completes without return value
Raises:
Exception: If the workflow fails
Exception: If report generation or delivery fails
"""
metadata = {
'metadata': {

37
orchestrator/workflows/subworkflows/load_notification_package.py
View File

@@ -9,20 +9,39 @@ with workflow.unsafe.imports_passed_through():
@workflow.defn(name="load_notification_package")
class LoadNotificationPackage:
"""
Subworkflow for loading notification data and configuration.
This subworkflow retrieves notification packages from MongoDB and
loads receiver group configurations. It handles timestamp-based
filtering for incremental data processing and manages the data
required for notification workflows.
"""
@workflow.run
async def run(self, input_data: dict[str, Any]):
"""
Loads the notification package from the MongoDB collection "notification_queue"
and the sending configs from the MongoDB collection "receiver_groups".
Load notification package and sending configurations.
input_data:
- metadata (dict): The metadata of the workflow.
This subworkflow loads notifications from the MongoDB notification
queue using timestamp-based filtering and retrieves active receiver
group configurations. It updates the last processed timestamp in Redis.
returns:
- last_timestamp (str): The last timestamp of the notification package.
- notification_package (list[dict]): The notification package.
- sending_configs (list[dict]): The sending configs.
- mail_type (str): The mail type.
Args:
input_data (dict[str, Any]): Workflow input parameters.
Required fields:
- metadata (dict[str, Any]): Workflow execution metadata
- mail_type (str): Type of mail (Alerts/Reports)
- base_data_filter (dict[str, Any]): Base filter for notification query
Returns:
dict[str, Any]: Package containing:
- last_timestamp (str | None): Last processed timestamp
- notification_package (list[dict]): Retrieved notifications
- sending_configs (list[dict]): Active receiver group configurations
Raises:
Exception: If data loading fails
"""
metadata = input_data['metadata']

47
orchestrator/workflows/subworkflows/process_notifications.py
View File

@@ -10,30 +10,37 @@ with workflow.unsafe.imports_passed_through():
@workflow.defn(name="process_notifications")
class ProcessNotifications:
"""
Subworkflow for processing and sending notification emails.
This subworkflow handles the email delivery process including HTML
generation, email sending, and logging to PostgreSQL. It processes
receiver groups and generates delivery reports for monitoring.
"""
@workflow.run
async def run(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Processes the notifications. Builds the report html for each group and each model,
sends the report html to the receivers of each group, stores the sending log in the
postgres database "log_report", and returns the log report to the caller.
Process notifications and send emails to configured receiver groups.
input_data:
- metadata (dict): The metadata of the workflow.
- mail_type (str): The mail type.
- schema (str): The schema of the table.
- table_name (str): The name of the table.
- notification_package (list[dict]): The notification package. the format of each
notification package is:
{
'group_name' (str)
'group_members' (list[str])
'notifications' (dict)
{
'model_name' (dict[str, list[dict]])
}
}
returns:
- log_report (dict)
This subworkflow builds HTML email content, sends emails to all
receiver groups, and logs the delivery results to PostgreSQL for
monitoring and audit purposes.
Args:
input_data (dict[str, Any]): Workflow input parameters.
Required fields:
- metadata (dict[str, Any]): Workflow execution metadata
- mail_type (str): Type of email being sent
- schema (str): PostgreSQL schema name for logging
- table_name (str): PostgreSQL table name for logging
- notification_package (dict[str, Any]): Receiver groups with notifications
Returns:
dict[str, Any]: Log report of email delivery results
Raises:
Exception: If notification processing fails
"""
metadata = input_data["metadata"]