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

Merge pull request #14 from Aignosi/SIENTIAPDE-1184-investigar-bugs-detectados-no-grafana

Browse Source 
Sientiapde 1184 investigar bugs detectados no grafana
This commit is contained in:
Bruno Domingues
2025-08-18 17:14:57 +00:00
committed by GitHub
parent da603d1b1f 229dcb1ac8
commit 552e1d7087
16 changed files with 1024 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

184
README.md
View File

@@ -1,3 +1,187 @@
# SIENTIA DataOps Orchestrator Temporal
## Overview
The SIENTIA DataOps Orchestrator Temporal is a comprehensive workflow orchestration system built on Temporal.io that manages data pipelines, notifications, and system orchestration for the SIENTIA platform. It provides automated scheduling, monitoring, and execution of data processing workflows with integrated alerting and reporting capabilities.
## Project Goals
- **Pipeline Orchestration**: Automate the deployment and management of data processing pipelines
- **Notification Management**: Handle real-time alerts and scheduled reports for system events
- **Resource Management**: Manage OPC server slots and data ingestion resources
- **Workflow Automation**: Coordinate complex workflows across multiple services and databases
- **Monitoring & Reporting**: Provide comprehensive logging and metrics for system health
## Architecture
The system is built around three main worker queues, each handling specific types of workflows:
### 1. Orchestrator Queue (`orchestrator-queue`)
Handles pipeline orchestration and resource management workflows.
### 2. Alerts Queue (`alerts-queue`)
Manages real-time alert notifications and error reporting.
### 3. Reports Queue (`reports-queue`)
Handles scheduled reports and data summaries.
## Workflows
### Main Workflows
#### 1. Orchestrator Workflow
**Purpose**: Main orchestration workflow that manages pipeline deployment and resource allocation.
**Input Parameters**:
- `schedule_name` (str): Name of the orchestration schedule
- `pipelines_query` (dict): MongoDB query to retrieve pipeline configurations
- `opc_servers_query` (dict): MongoDB query to retrieve OPC server configurations
**What it does**:
- Retrieves pipeline configurations from MongoDB
- Loads current OPC server slots and active ingestors from Redis
- Processes schedules and creates slot configurations
- Deploys schedules to Temporal server (scouter and laborious namespaces)
- Updates OPC slots in Redis
- Generates orchestration reports
#### 2. Alerts Workflow
**Purpose**: Sends real-time error alerts to configured user groups.
**Input Parameters**:
- `schedule_name` (str): Name of the alert schedule
- `notification_ttl` (int): Time period before considering notifications persistent
- `sent_ttl` (int): Time to live for sent notification cache
**What it does**:
- Filters notifications by ERROR level
- Loads notification packages from MongoDB
- Applies user group filtering and notification TTL rules
- Sends HTML email alerts
- Stores notification logs in PostgreSQL
- Caches sent notifications to prevent duplicates
#### 3. Reports Workflow
**Purpose**: Sends scheduled reports to configured user groups.
**Input Parameters**:
- `schedule_name` (str): Name of the report schedule
**What it does**:
- Loads all notifications (any level) from MongoDB
- Applies user group filtering
- Generates HTML report emails
- Stores report logs in PostgreSQL
### Subworkflows
#### 1. Load Notification Package
**Purpose**: Loads notification data and configuration from various sources.
**Input Parameters**:
- `metadata` (dict): Workflow metadata
- `mail_type` (str): Type of mail (Alerts/Reports)
- `base_data_filter` (dict): Base filters for data retrieval
**Returns**:
- `last_timestamp` (str): Last processed timestamp
- `notification_package` (list): Package of notifications to process
- `sending_configs` (list): Email sending configurations
#### 2. Process Notifications
**Purpose**: Processes notifications and sends emails with logging.
**Input Parameters**:
- `metadata` (dict): Workflow metadata
- `mail_type` (str): Type of mail being sent
- `schema` (str): Database schema name
- `table_name` (str): Database table name
- `notification_package` (list): Notifications to process
**Returns**:
- `log_report` (dict): Report of processed notifications
## Environment Variables
### Database Connections
#### Redis Configuration
- `REDIS_HOST`: Redis server hostname (default: localhost)
- `REDIS_PORT`: Redis server port (default: 6379)
- `REDIS_USERNAME`: Redis username (default: default)
- `REDIS_PASSWORD`: Redis password (from secret)
#### MongoDB Configuration
- `MONGODB_USERNAME`: MongoDB username (default: root)
- `MONGODB_PASSWORD`: MongoDB password
- `MONGODB_URL`: MongoDB server URL (default: localhost:27017)
- `MONGODB_DATABASE`: Database name (default: sientia)
- `MONGODB_TTL_INDEX_HOURS`: TTL index duration in hours (default: 1)
#### PostgreSQL Configuration
- `POSTGRES_HOST`: PostgreSQL server hostname
- `POSTGRES_PORT`: PostgreSQL server port (default: 5432)
- `POSTGRES_USER`: Database username (default: sientia)
- `POSTGRES_PASSWORD`: Database password (default: sientia)
- `POSTGRES_DBNAME`: Database name (default: sientia)
- `POSTGRES_MIN_CONNECTIONS`: Minimum connection pool size (default: 10)
- `POSTGRES_MAX_CONNECTIONS`: Maximum connection pool size (default: 40)
#### Couchbase Configuration
- `COUCHBASE_CONNECTION_STRING`: Couchbase server connection string
- `COUCHBASE_USERNAME`: Couchbase username (default: sientia)
- `COUCHBASE_PASSWORD`: Couchbase password (default: sientia)
### Email Configuration
- `EMAIL_SENDER`: Sender email address
- `EMAIL_SENDER_PASSWORD`: App password for SMTP authentication
- `EMAIL_SMTP_SERVER`: SMTP server hostname (default: smtp.gmail.com)
- `EMAIL_SMTP_PORT`: SMTP server port (default: 587)
### Temporal Configuration
- `TEMPORAL_HOST`: Temporal server hostname and port
- `TEMPORAL_NAMESPACE`: Default Temporal namespace (default: default)
- `TEMPORAL_SCOUTER_NAMESPACE`: Scouter workflow namespace (default: scouter)
- `TEMPORAL_LABORIOUS_NAMESPACE`: Laborious workflow namespace (default: laborious)
### Application Configuration
- `LOG_LEVEL`: Logging level (default: DEBUG)
- `HTTP_METRICS_PORT`: Prometheus metrics port (default: 9090)
- `PROJECT_NAME`: Project identifier (default: sientia-orchestrator)
- `POD_ID`: Kubernetes pod identifier for metrics
### Kafka Configuration
- `KAFKA_BOOTSTRAP_SERVERS`: Kafka bootstrap servers
## Deployment
The system is designed for Kubernetes deployment using Helm charts with:
- Health checks and readiness probes
- Prometheus metrics endpoint
- ServiceMonitor integration for Prometheus Operator
- Configurable resource limits and scaling
- SSH key management for Git operations
## Usage
Workflows can be triggered via Temporal client calls with appropriate input parameters. The system automatically handles:
- Pipeline configuration retrieval
- Resource allocation
- Schedule deployment
- Notification processing
- Email delivery
- Logging and monitoring
## Monitoring
The system exposes Prometheus metrics at `/metrics` endpoint including:
- Application status (up/down)
- Email sent counts
- Workflow execution metrics
- Custom business metrics
All operations are logged with structured metadata for debugging and auditing purposes.
# PR shortcut
```
git log origin/main..HEAD --no-merges > git_log

332
init_orchestration.ipynb Normal file
View File

File diff suppressed because one or more lines are too long

5
orchestrator/activities/activities.py
View File

@@ -82,4 +82,9 @@ class Activities( # Couchbase,
notification_handler=notification_handler)
def shutdown(self):
"""
Shutdown the MongoDB connection and clean up resources.
"""
MongoDB.shutdown(self)
Postgres.close(self)
Email.shutdown(self)

16
orchestrator/activities/email.py
View File

@@ -42,6 +42,12 @@ class Email(BaseActivity):
logger=logger,
notification_handler=notification_handler)
def shutdown(self):
"""
Shutdown the Email connection and clean up resources.
"""
self.server.quit()
@activity.defn(name="build_email_html")
async def build_email_html(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
@@ -105,9 +111,15 @@ class Email(BaseActivity):
def try_send_email(self, msg: MIMEMultipart, receivers: str):
"""
Sends an email to the receivers.
"""
Sends an email to the receivers with automatic reconnection handling.
Args:
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.
"""
try:
self.server.sendmail(
self.sender_email, receivers, msg.as_string())

217
orchestrator/activities/formatters.py
View File

@@ -19,6 +19,8 @@ with workflow.unsafe.imports_passed_through():
from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT
from math import ceil
topic_separator = "\n ========== \n"
class Formatters(BaseActivity):
def __init__(self,
@@ -166,11 +168,13 @@ class Formatters(BaseActivity):
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.
input_data:
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.
Returns:
- dict[str, Any]: The formatted schedule config.
dict[str, Any]: The formatted schedule config.
"""
schedule_config = input_data['schedule_config']
@@ -192,13 +196,22 @@ class Formatters(BaseActivity):
to_update: dict[str, Any], to_create: dict[str, Any],
namespace: str, metadata: dict[str, Any]):
"""
Compares the timestamps of the schedule and the current schedule.
Compares the timestamps of the schedule and the current schedule to determine
which schedules need to 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.
metadata (dict[str, Any]): Metadata for logging purposes.
"""
for schedule_name, schedule in schedules.items():
if schedule_name in current_schedules:
update_timestamp = schedule.get(
'updated_at', datetime.now().strftime(DEFAULT_DATE_FORMAT))
'updated_at', datetime.now())
old_timestamp = current_schedules[schedule_name]
@@ -316,36 +329,83 @@ class Formatters(BaseActivity):
return output
def send_success_report(self, metadata: dict[str, Any], message: str, notification_id: str) -> None:
def send_success_report(self, metadata: dict[str, Any], message: str, notification_id: str, attachment: str = None) -> None:
"""
Sends a success notification report.
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.
"""
self.send_notification(
metadata=metadata,
notification_id=notification_id,
message=message,
block="report_orchestration",
level=NotificationLevel.INFO
level=NotificationLevel.INFO,
attachment_content=json.dumps(attachment, indent=4, sort_keys=True)
)
def send_error_report(self, metadata: dict[str, Any], message: str, notification_id: str,
attachment: dict[str, Any]) -> None:
attachment: str) -> None:
"""
Sends an error notification report.
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.
"""
self.send_notification(
metadata=metadata,
notification_id=notification_id,
message=message,
block="report_orchestration",
level=NotificationLevel.ERROR,
attachment_content=json.dumps(attachment, indent=4, sort_keys=True)
attachment_content=attachment
)
def parse_report_schedule(self, input_data: dict[str, Any]) -> tuple[list[str], list[str]]:
def parse_report_schedule(self, input_data: dict[str, Any]) -> tuple[list[str], dict[str, Any]]:
"""
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.
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
"""
success_keys = [f"{value['namespace']}/{value['schedule_name']}"
for value in input_data if value['success']]
error_keys = [f"{value['namespace']}/{value['schedule_name']}: {value['message']}"
for value in input_data if not value['success']]
error_keys = {f"{value['namespace']}/{value['schedule_name']}": {
'message': value['message'],
'attachment': value.get('attachment', None)
}
for value in input_data if not value['success']}
return success_keys, error_keys
def parse_report(self, input_data: dict[str, Any]) -> tuple[list[str], list[str]]:
"""
Parses the report data to extract success and error keys.
Args:
input_data (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:
- List of successful keys
- List of error keys
"""
success_keys = [key for key, value
in input_data.items() if value['success']]
@@ -354,6 +414,41 @@ class Formatters(BaseActivity):
return success_keys, error_keys
def manage_and_send_report(self, metadata: dict[str, Any], success_keys: list[str], error_keys: dict[str, Any], schedule_type: str, schedule_data: dict[str, Any]):
"""
Manages and sends success and error reports based on the provided keys and data.
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.
"""
if len(success_keys) > 0:
self.send_success_report(
metadata=metadata,
message=f"Successfully {schedule_type}: \n {', '.join(success_keys)}",
notification_id=schedule_data['id'],
attachment=schedule_data['items']
)
if len(error_keys) > 0:
attachment = []
for key, value in error_keys.items():
if value['attachment'] is not None:
attachment.append(
f"{key}:\n{value['message']}\n{value['attachment']}")
else:
attachment.append(f"{key}:\n{value['message']}")
self.send_error_report(
metadata=metadata,
message=f"Fails on {schedule_type}: \n {', '.join(error_keys)}",
notification_id=f"{schedule_data['id']}_ERROR",
attachment=topic_separator.join(attachment)
)
@activity.defn(name="report_schedule_orchestration")
async def report_schedule_orchestration(self,
input_data: dict[str, Any]) -> None:
@@ -375,63 +470,32 @@ class Formatters(BaseActivity):
updated_schedules = input_data['updated_schedules']
deleted_schedules = input_data['deleted_schedules']
# Send report for created schedules
if len(created_schedules) > 0:
schedules_report = {
'created schedules': {
'items': created_schedules,
'id': 'REPORT_ORCHESTRATION_CREATED_SCHEDULES'
},
'updated schedules': {
'items': updated_schedules,
'id': 'REPORT_ORCHESTRATION_UPDATED_SCHEDULES'
},
'deleted schedules': {
'items': deleted_schedules,
'id': 'REPORT_ORCHESTRATION_DELETED_SCHEDULES'
}
}
for schedule_type, schedule_data in schedules_report.items():
if len(schedule_data['items']) > 0:
success_keys, error_keys = self.parse_report_schedule(
created_schedules)
schedule_data['items'])
if len(success_keys) > 0:
self.send_success_report(
self.manage_and_send_report(
metadata=metadata,
message=f"Created schedules: \n {', '.join(success_keys)}",
notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES"
)
if len(error_keys) > 0:
self.send_error_report(
metadata=metadata,
message=f"Failed to create schedules: \n {', '.join(error_keys)}",
notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES",
attachment=created_schedules
)
# Send report for updated schedules
if len(updated_schedules) > 0:
success_keys, error_keys = self.parse_report_schedule(
updated_schedules)
if len(success_keys) > 0:
self.send_success_report(
metadata=metadata,
message=f"Updated schedules: \n {', '.join(success_keys)}",
notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES"
)
if len(error_keys) > 0:
self.send_error_report(
metadata=metadata,
message=f"Failed to update schedules: \n {', '.join(error_keys)}",
notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES",
attachment=updated_schedules
)
if len(deleted_schedules) > 0:
success_keys, error_keys = self.parse_report_schedule(
deleted_schedules)
if len(success_keys) > 0:
self.send_success_report(
metadata=metadata,
message=f"Deleted schedules: \n {', '.join(success_keys)}",
notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES"
)
if len(error_keys) > 0:
self.send_error_report(
metadata=metadata,
message=f"Failed to delete schedules: \n {', '.join(error_keys)}",
notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES",
attachment=deleted_schedules
success_keys=success_keys,
error_keys=error_keys,
schedule_type=schedule_type,
schedule_data=schedule_data
)
@activity.defn(name="report_slot_orchestration")
@@ -494,8 +558,15 @@ class Formatters(BaseActivity):
async def format_log_report(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Formats the receiver_groups status to a dataframe to be stored in the database.
input_data:
- receiver_groups (dict): The receiver groups.
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.
Returns:
dict[str, Any]: The formatted log report as a dictionary representation of a DataFrame.
"""
metadata = input_data["metadata"]
mail_type = input_data["mail_type"]
@@ -540,7 +611,17 @@ class Formatters(BaseActivity):
@activity.defn(name="filter_notification_reports")
async def filter_notification_reports(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Filter notification reports.
Filters notification reports based on sending configurations and notification package.
Args:
input_data (dict[str, Any]): The input data containing:
- metadata (dict): Metadata for logging purposes.
- notification_package (list): The package of notifications to filter.
- sending_configs (list): The configurations for sending notifications.
Each config should have 'group_name', 'contents', and optionally 'ignore' fields.
Returns:
dict[str, Any]: The filtered receiver groups with their notifications.
"""
metadata = input_data['metadata']
notification_package = input_data['notification_package']

18
orchestrator/activities/mongo_db.py
View File

@@ -1,4 +1,3 @@
from pandas import DataFrame
from temporalio import workflow, activity
@@ -65,7 +64,7 @@ class MongoDB(BaseActivity):
def shutdown(self):
"""
Close the MongoDB client connection.
Shutdown the MongoDB connection and clean up resources.
"""
try:
if self.client:
@@ -82,6 +81,16 @@ class MongoDB(BaseActivity):
self.shutdown()
def find(self, collection_name: str, filters: dict[str, Any]) -> list[dict[str, Any]]:
"""
Find documents in a MongoDB collection based on the provided filters.
Args:
collection_name (str): The name of the collection to search in.
filters (dict[str, Any]): The query filters to apply.
Returns:
list[dict[str, Any]]: List of documents matching the filters, with _id fields removed.
"""
collection = self.database[collection_name]
documents = list(collection.find(filters, {"_id": 0}))
@@ -207,7 +216,7 @@ class MongoDB(BaseActivity):
"""
updated_pipelines = input_data.get("updated_pipelines", [])
metadata = input_data.get("metadata", {})
now = datetime.now().strftime(DEFAULT_DATE_FORMAT)
now = datetime.now()
collection = self.database["orchestrated_schedules"]
argument = [
@@ -246,7 +255,7 @@ class MongoDB(BaseActivity):
metadata = input_data.get("metadata", {})
collection = self.database["orchestrated_schedules"]
now = datetime.now().strftime(DEFAULT_DATE_FORMAT)
now = datetime.now()
argument = [
{"schedule_name": pipeline["schedule_name"],
@@ -257,6 +266,7 @@ class MongoDB(BaseActivity):
data_filter = argument if argument else {}
try:
if data_filter:
collection.insert_many(data_filter)
except Exception as e:
trace = traceback.format_exc()

38
orchestrator/activities/slot_manager.py
View File

@@ -202,6 +202,14 @@ class SlotManager(Redis):
async def get_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None:
"""
Gets the last data timestamp from redis.
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.
Returns:
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']}"
@@ -233,6 +241,15 @@ class SlotManager(Redis):
async def put_last_data_timestamp(self, input_data: dict[str, Any]):
"""
Puts the last data timestamp into redis.
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.
Returns:
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']}"
@@ -270,7 +287,18 @@ class SlotManager(Redis):
@activity.defn(name="filter_notification_alerts")
async def filter_notification_alerts(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Filter notification alerts
Filter notification alerts based on sending configurations and notification package.
Args:
input_data (dict[str, Any]): The input data containing:
- metadata (dict): Metadata for logging purposes.
- notification_package (list): The package of notifications to filter.
- sending_configs (list): The configurations for sending notifications.
Each config should have 'group_name', 'contents', and optionally 'ignore' fields.
- notification_ttl (int): Time to live for notifications in seconds.
Returns:
dict[str, Any]: The filtered receiver groups with their notifications.
"""
metadata = input_data['metadata']
notification_package = input_data['notification_package']
@@ -330,7 +358,13 @@ class SlotManager(Redis):
@activity.defn(name="store_notification_cache")
async def store_notification_cache(self, input_data: dict[str, Any]) -> None:
"""
Store notification cache
Store notification cache in Redis to track recently sent notifications.
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 = input_data['metadata']
log_report = DataFrame(input_data['log_report'])

11
orchestrator/activities/temporal_manager.py
View File

@@ -42,6 +42,10 @@ class TemporalManager(BaseActivity):
notification_handler=notification_handler)
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.
"""
self.logger.info(
f"Connecting to Temporal side namespaces at {self.temporal_host}")
self.logger.info(f"Scouter namespace: {self.scouter_namespace}")
@@ -167,7 +171,8 @@ class TemporalManager(BaseActivity):
schedule,
id=schedule_name,
task_queue=f"{workflow_type}-queue",
execution_timeout=timedelta(minutes=2)
execution_timeout=timedelta(minutes=2),
typed_search_attributes=search_attributes
),
spec=ScheduleSpec(
intervals=[
@@ -333,13 +338,15 @@ class TemporalManager(BaseActivity):
"message": "Schedule deleted successfully"
})
except Exception as e:
trace = traceback.format_exc()
self.error(
f"Failed to delete schedule {schedule_name}: {str(e)}", metadata=metadata)
report.append({
"namespace": namespace,
"schedule_name": schedule_name,
"success": False,
"message": str(e)
"message": str(e),
"attachment": trace
})
self.info(

36
orchestrator/utils/connectors_config.py
View File

@@ -2,6 +2,12 @@ from os import getenv
def build_redis_config():
"""
Build Redis configuration from environment variables.
Returns:
dict: Redis configuration with host, port, username, and password.
"""
return {
'host': getenv('REDIS_HOST', 'localhost'),
'port': int(getenv('REDIS_PORT', '6379')),
@@ -11,6 +17,12 @@ def build_redis_config():
def build_mongodb_config():
"""
Build MongoDB configuration from environment variables.
Returns:
dict: MongoDB configuration with connection string, database name, and TTL index seconds.
"""
username = getenv('MONGODB_USERNAME', 'root')
password = getenv('MONGODB_PASSWORD', 'wKZDbMNU1c')
uri = getenv('MONGODB_URL', 'localhost:27018')
@@ -24,6 +36,12 @@ def build_mongodb_config():
def build_couchbase_config():
"""
Build Couchbase configuration from environment variables.
Returns:
dict: Couchbase configuration with connection string, username, and password.
"""
return {
'connection_string': getenv('COUCHBASE_CONNECTION_STRING', 'couchbase://localhost'),
'username': getenv('COUCHBASE_USERNAME', 'sientia'),
@@ -32,6 +50,12 @@ def build_couchbase_config():
def build_temporal_config():
"""
Build Temporal configuration from environment variables.
Returns:
dict: Temporal configuration with host and namespace settings.
"""
return {
'temporal_host': getenv('TEMPORAL_HOST', 'localhost:7233'),
'temporal_namespace': getenv('TEMPORAL_NAMESPACE', 'default'),
@@ -41,6 +65,12 @@ def build_temporal_config():
def build_postgres_config():
"""
Build PostgreSQL configuration from environment variables.
Returns:
dict: PostgreSQL configuration with connection details and connection pool settings.
"""
return {
'host': getenv('POSTGRES_HOST', 'localhost'),
'port': int(getenv('POSTGRES_PORT', '5432')),
@@ -53,6 +83,12 @@ def build_postgres_config():
def build_email_config():
"""
Build email configuration from environment variables.
Returns:
dict: Email configuration with SMTP server settings and sender credentials.
"""
return {
'sender_email': getenv('EMAIL_SENDER', 'sientia-alerts@aignosi.com'),
'sender_password': getenv('EMAIL_SENDER_PASSWORD', 'sientia'),

32
orchestrator/utils/email_builder.py
View File

@@ -18,12 +18,33 @@ class EmailBuilder:
self.general_template = file.read()
def replace_parameters(self, template: str, parameters: dict) -> str:
"""
Replace parameters in a Jinja2 template with provided values.
Args:
template (str): The Jinja2 template string.
parameters (dict): Dictionary of parameters to replace in the template.
Returns:
str: The rendered template with parameters replaced.
"""
# Criar um template Jinja2
template = Template(template)
return template.render(parameters)
def parameters(self, general_events: dict, mail_type: str) -> dict:
"""
Build parameters dictionary for email templates based on general events and mail type.
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.
Returns:
dict: Dictionary with mail_type and rendered event sections for each notification level.
"""
error_models = general_events.get('ERROR', {}).get('models', [])
warning_models = general_events.get('WARNING', {}).get('models', [])
info_models = general_events.get('INFO', {}).get('models', [])
@@ -43,7 +64,16 @@ class EmailBuilder:
def build_email(self, report_data: list[dict], mail_type: str) -> str:
"""
Builds the email html.
Builds the email HTML by organizing report data by notification level and model.
Args:
report_data (list[dict]): List of notification reports, each containing:
- level (str): Notification level (ERROR, WARNING, INFO)
- model_name (str): Name of the model
- Additional notification details
Returns:
str: Complete HTML email content ready for sending.
"""
general_events = {}

105
orchestrator/utils/orchestrator_functions.py
View File

@@ -2,6 +2,21 @@ from typing import Any
def common_config(config: dict[str, Any]):
"""
Extract common configuration parameters from a pipeline configuration.
Args:
config (dict[str, Any]): Pipeline configuration containing:
- workflow_type (str): Type of workflow (e.g., 'scouter', 'predictions_batch')
- schedule_name (str): Name of the schedule
- frequency (str, optional): Frequency of execution (default: '1m')
- max_retry_policy (int, optional): Maximum retry attempts (default: 1)
- model_id (str): ID of the model
- models (dict): Model configuration containing 'name' field
Returns:
dict[str, Any]: Common configuration dictionary with extracted parameters.
"""
return {
"workflow_type": config['workflow_type'],
"schedule_name": config['schedule_name'],
@@ -14,6 +29,18 @@ def common_config(config: dict[str, Any]):
def minimal_retrain(config: dict[str, Any]):
"""
Build minimal retrain configuration from pipeline config.
Args:
config (dict[str, Any]): Pipeline configuration containing:
- schedule_name (str): Name of the schedule
- query (str): SQL query for retraining
- Additional fields from common_config
Returns:
dict[str, Any]: Minimal retrain configuration with workflow type set to 'minimal_retrain'.
"""
return {
**common_config(config),
"workflow_type": "minimal_retrain",
@@ -25,6 +52,25 @@ def minimal_retrain(config: dict[str, Any]):
def scouter(config: dict[str, Any]):
"""
Build scouter configuration from pipeline config.
Args:
config (dict[str, Any]): Pipeline configuration containing:
- filters (list[dict], optional): List of filter configurations
- read_tags (list[dict]): List of tag configurations with:
- filter_name (str): Name of the filter
- policy (str): Filter policy
- tag_name (str): Name of the tag
- aggr_func (str, optional): Aggregation function (default: 'lts')
- data_range (list[int], optional): Data range limits (default: [-100, 100])
- tag_retention_minutes (int, optional): Tag retention time in minutes (default: 60)
- debug_data_package (bool, optional): Enable debug data package (default: False)
- Additional fields from common_config
Returns:
dict[str, Any]: Scouter configuration with topic, filters, tags, and retention settings.
"""
filters = {}
for f in config.get('filters', []):
filters[f['filter_name']] = {
@@ -53,6 +99,19 @@ def scouter(config: dict[str, Any]):
def overlap_filter_config(base_filter_config: dict[str, Any], config: list[dict[str, Any]]):
"""
Overlap filter configuration with base filter config.
Args:
base_filter_config (dict[str, Any]): Base filter configuration to extend.
config (list[dict[str, Any]]): List of filter configurations to add, each containing:
- filter_name (str): Name of the filter
- policy (str): Filter policy
- config (dict, optional): Additional filter configuration
Returns:
dict[str, Any]: Extended filter configuration with new filters added.
"""
for fil in config:
base_filter_config[fil['filter_name']] = {
"policy": fil['policy'],
@@ -63,6 +122,15 @@ def overlap_filter_config(base_filter_config: dict[str, Any], config: list[dict[
def process_path_priority(path_priority: list[str]):
"""
Process and normalize path priority list to ensure it contains the required priorities.
Args:
path_priority (list[str]): List of path priorities to process.
Returns:
list[str]: Normalized path priority list with exactly 3 elements: ["STOP", "CONTINUE", "REPEAT"].
"""
for priority in path_priority[:]:
if priority not in ["STOP", "CONTINUE", "REPEAT"]:
path_priority.remove(priority)
@@ -75,6 +143,26 @@ def process_path_priority(path_priority: list[str]):
def predictions_batch(config: dict[str, Any]):
"""
Build predictions batch configuration from pipeline config.
Args:
config (dict[str, Any]): Pipeline configuration containing:
- write_tags (list[dict]): List of tag configurations with:
- server_id (str): ID of the OPC server
- type (str): Tag type ('prediction' or 'confidence')
- addr (str): Tag address
- data_type (str, optional): Data type (default: 'float')
- path_priority (list[str], optional): List of path priorities (default: ["STOP", "CONTINUE", "REPEAT"])
- input_filters (list[dict], optional): List of input filter configurations
- mlflow_transform_filters (list[dict], optional): List of MLflow transform filter configurations
- mlflow_predict_filters (list[dict], optional): List of MLflow predict filter configurations
- model_retention_minutes (int, optional): Model retention time in minutes (default: 60)
- Additional fields from common_config
Returns:
dict[str, Any]: Predictions batch configuration with OPC output config, filters, and path priority.
"""
tags = {}
for tag in config.get('write_tags', []):
if tag['server_id'] not in tags:
@@ -156,6 +244,23 @@ def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]:
def build_tag_config(tag: dict[str, Any], slot_config: dict[str, Any],
opc_servers: dict[str, Any], i: int):
"""
Build tag configuration for a specific slot and OPC server.
Args:
tag (dict[str, Any]): Tag configuration 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.
Returns:
dict[str, Any]: Updated slot configuration with the new tag.
Raises:
ValueError: If the specified server_id is not found in opc_servers.
"""
server_id = tag['server_id']
if server_id not in opc_servers:

13
orchestrator/worker/worker.py
View File

@@ -29,6 +29,13 @@ POD_ID = os.getenv("POD_ID")
async def main():
"""
Main function to initialize and run the Temporal worker.
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.
"""
host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
namespace = os.getenv('TEMPORAL_NAMESPACE', 'default')
logger = get_logger(__name__)
@@ -176,6 +183,12 @@ async def main():
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.
"""
try:
port = int(os.getenv("HTTP_METRICS_PORT", 9090))
start_http_server(port)

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

@@ -466,21 +466,6 @@ async def test_create_slot_config(formatters):
def test_send_success_report(formatters):
formatters.send_success_report(
metadata=metadata,
message="test_message",
notification_id="test_notification_id"
)
formatters.send_notification.assert_called_once_with(
metadata=metadata,
notification_id="test_notification_id",
message="test_message",
block="report_orchestration",
level=NotificationLevel.INFO
)
def test_send_error_report(formatters):
formatters.send_error_report(
metadata=metadata,
message="test_message",
notification_id="test_notification_id",
@@ -491,12 +476,29 @@ def test_send_error_report(formatters):
notification_id="test_notification_id",
message="test_message",
block="report_orchestration",
level=NotificationLevel.ERROR,
level=NotificationLevel.INFO,
attachment_content=json.dumps(
{"test": "test"}, indent=4, sort_keys=True)
)
def test_send_error_report(formatters):
formatters.send_error_report(
metadata=metadata,
message="test_message",
notification_id="test_notification_id",
attachment="test_attachment"
)
formatters.send_notification.assert_called_once_with(
metadata=metadata,
notification_id="test_notification_id",
message="test_message",
block="report_orchestration",
level=NotificationLevel.ERROR,
attachment_content="test_attachment"
)
def test_parse_report(formatters):
input_data = {
"test_key": {
@@ -527,14 +529,18 @@ def test_parse_report_schedule(formatters):
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_create_error",
"success": False,
"message": "test_error"
"message": "test_error",
"attachment": "test_attachment"
}
]
result = formatters.parse_report_schedule(input_data)
assert result == (
["test_namespace/test_schedule_name_to_create"],
["test_namespace/test_schedule_name_to_create_error: test_error"]
{"test_namespace/test_schedule_name_to_create_error": {
'message': 'test_error',
'attachment': 'test_attachment'
}}
)
@@ -558,7 +564,14 @@ async def test_report_schedule_orchestration(formatters):
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_create_error",
"success": False,
"message": "test_error"
"message": "test_error1",
"attachment": "test_attachment1"
},
{
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_create_error2",
"success": False,
"message": "test_error2"
}
],
"updated_schedules": [
@@ -571,7 +584,21 @@ async def test_report_schedule_orchestration(formatters):
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_update_error",
"success": False,
"message": "test_error"
"message": "test_error2",
"attachment": "test_attachment2"
},
{
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_update_error2",
"success": False,
"message": "test_error3",
"attachment": "test_attachment3"
},
{
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_update_error3",
"success": False,
"message": "test_error4"
}
],
"deleted_schedules": [
@@ -584,7 +611,14 @@ async def test_report_schedule_orchestration(formatters):
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_delete_error",
"success": False,
"message": "test_error"
"message": "test_error4",
"attachment": "test_attachment4"
},
{
"namespace": "test_namespace",
"schedule_name": "test_schedule_name_to_delete_error2",
"success": False,
"message": "test_error5"
}
]
}
@@ -599,40 +633,43 @@ async def test_report_schedule_orchestration(formatters):
formatters.send_success_report.assert_has_calls([
call(
metadata=metadata['metadata'],
message="Created schedules: \n test_namespace/test_schedule_name_to_create",
notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES"
),
call(
metadata=metadata['metadata'],
message="Updated schedules: \n test_namespace/test_schedule_name_to_update",
notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES"
),
call(
metadata=metadata['metadata'],
message="Deleted schedules: \n test_namespace/test_schedule_name_to_delete",
notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES"
)
])
formatters.send_error_report.assert_has_calls([
call(
metadata=metadata['metadata'],
message="Failed to create schedules: \n test_namespace/test_schedule_name_to_create_error: test_error",
message="Successfully created schedules: \n test_namespace/test_schedule_name_to_create",
notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES",
attachment=input_data['created_schedules']
),
call(
metadata=metadata['metadata'],
message="Failed to update schedules: \n test_namespace/test_schedule_name_to_update_error: test_error",
message="Successfully updated schedules: \n test_namespace/test_schedule_name_to_update",
notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES",
attachment=input_data['updated_schedules']
),
call(
metadata=metadata['metadata'],
message="Failed to delete schedules: \n test_namespace/test_schedule_name_to_delete_error: test_error",
message="Successfully deleted schedules: \n test_namespace/test_schedule_name_to_delete",
notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES",
attachment=input_data['deleted_schedules']
)
])
formatters.send_error_report.assert_has_calls([
call(
metadata=metadata['metadata'],
message="Fails on created schedules: \n test_namespace/test_schedule_name_to_create_error, test_namespace/test_schedule_name_to_create_error2",
notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES_ERROR",
attachment="test_namespace/test_schedule_name_to_create_error:\ntest_error1\ntest_attachment1\n ========== \ntest_namespace/test_schedule_name_to_create_error2:\ntest_error2"
),
call(
metadata=metadata['metadata'],
message="Fails on updated schedules: \n test_namespace/test_schedule_name_to_update_error, test_namespace/test_schedule_name_to_update_error2, test_namespace/test_schedule_name_to_update_error3",
notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES_ERROR",
attachment="test_namespace/test_schedule_name_to_update_error:\ntest_error2\ntest_attachment2\n ========== \ntest_namespace/test_schedule_name_to_update_error2:\ntest_error3\ntest_attachment3\n ========== \ntest_namespace/test_schedule_name_to_update_error3:\ntest_error4"
),
call(
metadata=metadata['metadata'],
message="Fails on deleted schedules: \n test_namespace/test_schedule_name_to_delete_error, test_namespace/test_schedule_name_to_delete_error2",
notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES_ERROR",
attachment="test_namespace/test_schedule_name_to_delete_error:\ntest_error4\ntest_attachment4\n ========== \ntest_namespace/test_schedule_name_to_delete_error2:\ntest_error5"
)
])
@mark.asyncio

6
tests/orchestrator/activities/test_mongo_db.py
View File

@@ -280,7 +280,7 @@ async def test_update_pipelines_timestamps_success(datetime_mock, mongo_db):
{"schedule_name": "test2", "namespace": "test2"}
]},
{"$set": {
"updated_at": datetime_mock.now.return_value.strftime.return_value}}
"updated_at": datetime_mock.now.return_value}}
)
@@ -325,9 +325,9 @@ async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db):
mongo_db.database["pipelines"].insert_many.assert_called_once_with(
[
{"schedule_name": "test1", "namespace": "test1",
"updated_at": datetime_mock.now.return_value.strftime.return_value},
"updated_at": datetime_mock.now.return_value},
{"schedule_name": "test2", "namespace": "test2",
"updated_at": datetime_mock.now.return_value.strftime.return_value}
"updated_at": datetime_mock.now.return_value}
]
)

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

@@ -213,21 +213,24 @@ async def test_create_schedule(
input_data['schedules']['scouter']['test-schedule'],
id="test-schedule",
task_queue="test-workflow-queue",
execution_timeout=ANY
execution_timeout=ANY,
typed_search_attributes=mock_typed_search_attributes.return_value,
),
call(
"test-workflow",
input_data['schedules']['scouter']['test-schedule-invalid-frequency'],
id="test-schedule-invalid-frequency",
task_queue="test-workflow-queue",
execution_timeout=ANY
execution_timeout=ANY,
typed_search_attributes=mock_typed_search_attributes.return_value,
),
call(
"test-workflow",
input_data['schedules']['laborious']['test-schedule-laborious'],
id="test-schedule-laborious",
task_queue="test-workflow-queue",
execution_timeout=ANY
execution_timeout=ANY,
typed_search_attributes=mock_typed_search_attributes.return_value,
)
])
@@ -546,7 +549,8 @@ async def test_delete_schedules(temporal_manager):
"schedule_name": "test-schedule_no_handler",
"namespace": "scouter",
"success": False,
"message": "Schedule test-schedule_no_handler not found"
"message": "Schedule test-schedule_no_handler not found",
"attachment": ANY
},
{
"schedule_name": "test-schedule-laborious",

4
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.3.2"
tag: "0.4.0"
# 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:
@@ -144,7 +144,7 @@ env:
- name: GITHUB_REPO_URL
value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
- name: GITHUB_BRANCH
value: "SIENTIAPDE-1174-mapear-e-implementar-metricas-a-serem-criadas"
value: "SIENTIAPDE-1184-investigar-bugs-detectados-no-grafana"
- name: PYTHON_APP
value: "orchestrator.worker.worker"