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

Merge pull request #23 from Aignosi/SIENTIAPDE-1222-ajustar-a-library-para-fazer-o-download-do-courier

Browse Source 
SIENTIAPDE-1222: Enhance Orchestration, Notification Filtering, and Server Configuration
This commit is contained in:
Matheus Demoner
2025-09-24 12:22:31 -03:00
committed by GitHub
parent 5db411e5b5 49f3429556
commit 6091d2ce6b
22 changed files with 463 additions and 318 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -44,3 +44,5 @@ htmlcov/
git_key*
git_log
.env

95
README.md
View File

@@ -13,10 +13,15 @@ A high-performance, scalable workflow orchestration system built on Temporal.io
### Advanced Capabilities
- **Incremental Processing**: Timestamp-based data loading to avoid reprocessing
- **Configurable Filtering**: User group-based notification filtering with custom policies
- **Intelligent Notification Filtering**: Advanced filtering system with:
- User group-based notification filtering with custom policies
- TTL-based duplicate prevention for alerts
- Ignore lists for specific notifications
- Persistent alert detection for ongoing issues
- **Auto-scaling Workers**: Multiple worker instances with task queue isolation
- **Comprehensive Logging**: Structured logging with PostgreSQL audit trails
- **Prometheus Metrics**: Real-time monitoring and alerting integration
- **Template-based Email Generation**: Jinja2-powered HTML email templates
## Architecture
@@ -252,6 +257,12 @@ flowchart LR
- `notification_package` (list[dict]): Retrieved notifications
- `sending_configs` (list[dict]): Active receiver group configurations
**Key Activities**:
- `get_last_data_timestamp`: Retrieves last processed timestamp from Redis
- `find_documents_in_mongodb`: Loads receiver group configurations
- `load_latest_data`: Loads notifications with timestamp filtering
- `put_last_data_timestamp`: Updates last processed timestamp
**Architecture**:
```mermaid
@@ -277,10 +288,11 @@ flowchart LR
**Purpose**: Handles email generation, delivery, and audit logging for notification workflows.
**Key Features**:
- **HTML Generation**: Creates formatted email content for each receiver group
- **HTML Generation**: Creates formatted email content for each receiver group using Jinja2 templates
- **Email Delivery**: Sends emails with attachment support and error handling
- **Audit Logging**: Records delivery status and metrics in PostgreSQL
- **Error Recovery**: Handles SMTP failures with detailed error reporting
- **Template Support**: Uses customizable HTML templates for different email types
**Input Parameters**:
```json
@@ -300,6 +312,12 @@ flowchart LR
**Returns**:
- `log_report` (list[dict]): Detailed delivery status for each notification
**Key Activities**:
- `build_email_html`: Generates HTML content using Jinja2 templates
- `send_email`: Delivers emails to receiver groups with error handling
- `format_log_report`: Formats delivery results for database storage
- `export_data_to_postgres`: Stores audit logs in PostgreSQL
**Architecture**:
```mermaid
@@ -313,6 +331,29 @@ flowchart LR
```
## 🔔 Notification Filtering System
The orchestrator includes an advanced notification filtering system that prevents alert spam and ensures relevant notifications reach the appropriate user groups.
### **Alert Filtering (`filter_notification_alerts`)**
- **Purpose**: Filters error-level notifications for immediate alerts
- **TTL Management**: Prevents duplicate alerts using configurable time-to-live settings
- **Persistent Detection**: Identifies ongoing issues that require escalation
- **Group-based Filtering**: Routes notifications to appropriate receiver groups
- **Ignore Lists**: Supports notification exclusion per group
### **Report Filtering (`filter_notification_reports`)**
- **Purpose**: Filters all notification levels for comprehensive reports
- **Comprehensive Coverage**: Includes ERROR, WARNING, INFO, and DEBUG levels
- **Group Customization**: Applies different content policies per receiver group
- **Scheduled Processing**: Designed for regular report generation
### **Notification Caching (`store_notification_cache`)**
- **Purpose**: Manages Redis-based notification cache for TTL enforcement
- **TTL Support**: Configurable expiration times for different notification types
- **Duplicate Prevention**: Ensures notifications aren't sent repeatedly within TTL window
- **Key Management**: Uses structured keys for efficient cache lookups
### Key Components
#### **Worker (`orchestrator/worker/worker.py`)**
@@ -332,19 +373,23 @@ flowchart LR
#### **Activities (`orchestrator/activities/`)**
- **Activities**: Main activity orchestrator combining all operations
- **TemporalManager**: Temporal schedule CRUD operations across namespaces
- **SlotManager**: Redis-based OPC slot and cache management
- **SlotManager**: Redis-based OPC slot and cache management with notification filtering
- **MongoDB**: Document operations, aggregations, and TTL management
- **Email**: SMTP operations with HTML generation and attachment support
- **Formatters**: Configuration processing and slot distribution algorithms
- **Formatters**: Configuration processing, slot distribution algorithms, and notification filtering for reports
- **Postgres**: PostgreSQL operations for audit logging and data export (via sientia-dataops-library)
- **Couchbase**: Database operations (currently unused but maintained for future use)
#### **Utilities (`orchestrator/utils/`)**
- **Connectors Configuration**: Database and service configuration management
- **Email Builder**: HTML email template generation and formatting
- **Orchestrator Functions**: Pipeline configuration transformation utilities
- **Converters**: Data type conversion and validation utilities
- **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
- **Converters**: Data type conversion and validation utilities including frequency parsing
- **Templates**: HTML email templates for alerts and reports
## 📋 Prerequisites
### System Requirements
- Python 3.11+
- Temporal server/cluster
- Redis server
@@ -461,6 +506,12 @@ python -m orchestrator.worker.worker
| `EMAIL_SMTP_PORT` | SMTP port | `587` | No |
| `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No |
| `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No |
| `POSTGRES_MIN_CONNECTIONS` | PostgreSQL minimum connections | `10` | No |
| `POSTGRES_MAX_CONNECTIONS` | PostgreSQL maximum connections | `40` | No |
| `MONGODB_TTL_INDEX_HOURS` | MongoDB TTL index expiration in hours | `1` | No |
| `PROJECT_NAME` | Project name for metrics and logging | `sientia-orchestrator` | No |
| `LOG_LEVEL` | Application logging level | `INFO` | No |
| `KAFKA_BOOTSTRAP_SERVERS` | Kafka bootstrap servers | - | No |
### Workflow Configuration
@@ -560,11 +611,28 @@ The Orchestrator system exposes comprehensive Prometheus metrics:
```
tests/
├── orchestrator/ # Orchestrator workflow tests
│ ├── test_activities.py
│ ├── test_couchbase.py
│ ├── test_email.py
│ ├── test_formatters.py
│ ├── test_mongo_db.py
│ ├── test_slot_manager.py
│ ├── test_temporal_manager.py
│ └── test_workflows.py
├── activities/ # Activity implementation tests
├── utils/ # Utility function tests
└── integration/ # End-to-end workflow tests
```
### Test Coverage
The project maintains comprehensive test coverage including:
- **Activity Tests**: Unit tests for all activity classes
- **Workflow Tests**: Integration tests for workflow orchestration
- **Utility Tests**: Tests for configuration builders and converters
- **Database Tests**: Tests for MongoDB, Redis, and PostgreSQL operations
- **Email Tests**: Tests for email generation and delivery
- **Notification Tests**: Tests for filtering and caching logic
### Test Execution
```bash
# Install test dependencies
@@ -586,21 +654,28 @@ orchestrator/
├── activities/ # Temporal activity implementations
│ ├── activities.py # Main activities orchestrator
│ ├── temporal_manager.py # Temporal schedule operations
│ ├── slot_manager.py # Redis slot management
│ ├── slot_manager.py # Redis slot management and notification filtering
│ ├── mongo_db.py # MongoDB operations
│ ├── email.py # Email service operations
│ └── formatters.py # Configuration formatting
│ ├── formatters.py # Configuration formatting and report filtering
│ └── couchbase.py # Couchbase operations (currently unused)
├── workflows/ # Temporal workflow definitions
│ ├── orchestrator.py # Main orchestration workflow
│ ├── alerts.py # Error alert workflow
│ ├── reports.py # Scheduled report workflow
│ └── subworkflows/ # Sub-workflow implementations
│ ├── load_notification_package.py # Notification data loading
│ └── process_notifications.py # Email processing and delivery
├── worker/ # Worker implementation
│ └── worker.py # Main worker orchestrator
├── utils/ # Utility functions
│ ├── connectors_config.py # Database configuration
│ ├── email_builder.py # Email template generation
│ └── orchestrator_functions.py # Pipeline utilities
│ ├── orchestrator_functions.py # Pipeline utilities
│ ├── converters.py # Data type conversion utilities
│ └── templates/ # Email HTML templates
│ ├── email_template.html # Report email template
│ └── general_template.html # General email template
└── metrics.py # Prometheus metrics definitions
```

56
init_orchestration.ipynb
View File

@@ -23,7 +23,7 @@
},
{
"cell_type": "code",
"execution_count": 4,
"execution_count": null,
"id": "d9d2a242",
"metadata": {},
"outputs": [
@@ -75,40 +75,28 @@
" \"pipelines_query\": {\n",
" \"collection\": \"pipelines\",\n",
" \"aggregation\": [\n",
" {\n",
" \"$lookup\": {\n",
" \"from\": \"models\",\n",
" \"localField\": \"model_id\",\n",
" \"foreignField\": \"id\",\n",
" \"as\": \"model_docs\"\n",
" }\n",
" },\n",
" {\n",
" \"$match\": {\n",
" \"active\": True\n",
" }\n",
" },\n",
" {\n",
" \"$addFields\": {\n",
" \"models\": {\n",
" \"$arrayElemAt\": [\n",
" \"$model_docs\",\n",
" 0\n",
" ]\n",
" {\n",
" \"$lookup\": {\n",
" \"from\": \"models\",\n",
" \"localField\": \"model_id\",\n",
" \"foreignField\": \"id\",\n",
" \"as\": \"model\"\n",
" }\n",
" },\n",
" {\n",
" \"$unwind\": \"$model\"\n",
" },\n",
" {\n",
" \"$match\": {\n",
" \"active\": True\n",
" }\n",
" },\n",
" {\n",
" \"$match\": {\n",
" \"model.active\": True\n",
" }\n",
" }\n",
" }\n",
" },\n",
" {\n",
" \"$match\": {\n",
" \"models.active\": True\n",
" }\n",
" },\n",
" {\n",
" \"$project\": {\n",
" \"model_docs\": 0\n",
" }\n",
" }\n",
" ]\n",
" ]\n",
" },\n",
" \"opc_servers_query\": {\n",
" \"collection\": \"opc-servers\",\n",

14
orchestrator/activities/email.py
View File

@@ -48,11 +48,12 @@ class Email(BaseActivity):
logger.info(f"Initializing Email with {smtp_server}:{smtp_port}")
self.server = smtplib.SMTP(smtp_server, smtp_port, timeout=20)
if smtp_server is not None:
self.server = smtplib.SMTP(smtp_server, smtp_port, timeout=20)
if self.sender_password:
self.server.starttls()
self.server.login(self.sender_email, self.sender_password)
if self.sender_password:
self.server.starttls()
self.server.login(self.sender_email, self.sender_password)
BaseActivity.__init__(self,
logger=logger,
@@ -200,6 +201,11 @@ class Email(BaseActivity):
receiver_groups = input_data['receiver_groups']
mail_type = input_data['mail_type']
if self.smtp_server is None:
self.info(f"Skipping email sending for {mail_type} mail type.",
metadata=metadata)
return {}
self.info(f"Sending email for {mail_type} mail type.",
metadata=metadata)

67
orchestrator/activities/formatters.py
View File

@@ -1,19 +1,15 @@
from pandas import DataFrame
from temporalio import activity, workflow
from orchestrator.utils.orchestrator_functions import minimal_retrain
with workflow.unsafe.imports_passed_through():
import json
from typing import Any
from logging import Logger
from pandas import DataFrame
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.notifications.models import NotificationLevel
from orchestrator.utils.orchestrator_functions import (
scouter, predictions_batch, gather_read_tags, build_tag_config
scouter, predictions_batch, gather_read_tags, build_tag_config, minimal_retrain
)
from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now
from math import ceil
@@ -23,12 +19,20 @@ topic_separator = "\n ========== \n"
class Formatters(BaseActivity):
"""
Schedule and slot configuration formatting activity.
Schedule and slot configuration formatting and notification filtering 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.
This class provides comprehensive formatting operations for schedules and OPC slots,
converting pipeline configurations into Temporal-compatible formats, managing slot
distribution across active ingestors, and implementing notification filtering for
scheduled reports.
Key Features:
- Pipeline schedule configuration formatting (scouter, predictions_batch, minimal_retrain)
- OPC slot distribution across active ingestors
- Notification filtering for comprehensive reports
- Group-based report filtering with ignore list support
- Resource optimization algorithms
- Configuration validation and transformation
Args:
scouter_namespace (str): Scouter workflow namespace
@@ -49,8 +53,16 @@ class Formatters(BaseActivity):
@activity.defn(name="process_schedules")
async def process_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Process schedules. Generates a schedule config dictionary
based on the input data workflow type.
Process pipeline configurations into Temporal-compatible schedule configurations.
This method transforms pipeline configurations from MongoDB into properly formatted
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
Args:
- input_data (dict[str, Any]): The input data containing
@@ -633,17 +645,34 @@ class Formatters(BaseActivity):
@activity.defn(name="filter_notification_reports")
async def filter_notification_reports(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Filters notification reports based on sending configurations and notification package.
Filter notifications for comprehensive scheduled reports.
This method filters notifications of all levels (ERROR, WARNING, INFO, DEBUG)
for scheduled report generation. Unlike alert filtering, this method does not
implement TTL-based duplicate prevention since reports are meant to provide
comprehensive coverage of system activity within a time window.
Filtering Logic:
- Processes all notification levels (not just ERROR)
- Applies group-specific content filtering for "reports" type
- Respects ignore lists for each receiver group
- Prevents duplicate notifications within the same report
- Groups notifications by receiver group configurations
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.
- metadata (dict): Workflow execution metadata for logging
- notification_package (list): All notifications to filter (any level)
- sending_configs (list): Receiver group configurations with:
- group_name (str): Name of the receiver group
- contents (list): Content types to include (must contain "reports")
- ignore (list, optional): Notification IDs to exclude
Returns:
dict[str, Any]: The filtered receiver groups with their notifications.
dict[str, Any]: Filtered receiver groups with their notifications, keyed by group_name.
Each group contains:
- All receiver group configuration fields
- notifications (list): Filtered notifications for this group
"""
metadata = input_data['metadata']
notification_package = input_data['notification_package']

14
orchestrator/activities/mongo_db.py
View File

@@ -130,6 +130,7 @@ class MongoDB(BaseActivity):
query = input_data.get("query", {})
metadata = input_data.get("metadata", {})
timestamp_fields = input_data.get("timestamp_fields", [])
collection_name = query.get("collection")
if not collection_name:
@@ -146,6 +147,12 @@ class MongoDB(BaseActivity):
self.info(
f"Loaded {len(documents)} documents from collection '{collection_name}'", metadata=metadata)
for document in documents:
for timestamp_field in timestamp_fields:
if timestamp_field in document:
document[timestamp_field] = document[timestamp_field].replace(tzinfo=timezone.utc).strftime(
DATETIME_FORMAT_MS_WITH_TZ)
self.debug(
f"Documents loaded: {documents}", metadata=metadata)
@@ -181,6 +188,7 @@ class MongoDB(BaseActivity):
query = input_data.get("query", {})
metadata = input_data.get("metadata", {})
timestamp_fields = input_data.get("timestamp_fields", [])
collection_name = query.get("collection")
if not collection_name:
@@ -204,6 +212,12 @@ class MongoDB(BaseActivity):
self.info(
f"Aggregated {len(aggregated_documents)} documents from collection '{collection_name}'", metadata=metadata)
for document in aggregated_documents:
for timestamp_field in timestamp_fields:
if timestamp_field in document:
document[timestamp_field] = document[timestamp_field].replace(tzinfo=timezone.utc).strftime(
DATETIME_FORMAT_MS_WITH_TZ)
self.debug(
f"Aggregation result: {aggregated_documents}", metadata=metadata)

66
orchestrator/activities/slot_manager.py
View File

@@ -16,12 +16,20 @@ with workflow.unsafe.imports_passed_through():
class SlotManager(Redis):
"""
Redis-based OPC slot management activity.
Redis-based OPC slot management and notification filtering 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.
This class manages OPC server slots and provides advanced notification
filtering capabilities through Redis operations. It handles OPC slot
lifecycle management, active ingestor tracking, and implements intelligent
notification filtering with TTL-based duplicate prevention.
Key Features:
- OPC slot loading, updating, and deletion
- Active ingestor management
- Notification filtering for alerts with TTL management
- Persistent alert detection for ongoing issues
- Notification caching with configurable expiration
- Group-based filtering with ignore list support
Args:
host (str): Redis server hostname
@@ -42,10 +50,23 @@ class SlotManager(Redis):
@activity.defn(name="load_opc_slots")
async def load_opc_slots(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Load all OPC slots from Redis
Load all OPC slots from Redis for current system state assessment.
This method retrieves all OPC server slot configurations from Redis,
which are used to determine current resource allocation and identify
changes needed for pipeline orchestration.
Args:
input_data (dict[str, Any]): Activity input containing metadata
Returns:
dict[str, Any]: A dictionary of OPC slots
dict[str, Any]: Dictionary of OPC slots keyed by server ID, where each slot contains:
- Configuration parameters for OPC server connections
- Active pipeline assignments
- Resource allocation details
Raises:
Exception: If Redis connection fails or data retrieval errors occur
"""
metadata = input_data.get("metadata", {})
@@ -313,18 +334,35 @@ 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 based on sending configurations and notification package.
Filter notification alerts with intelligent TTL-based duplicate prevention.
This method implements advanced notification filtering for ERROR-level alerts,
preventing spam through TTL management and detecting persistent issues that
require escalation. It applies user group-based filtering with configurable
ignore lists and content policies.
Filtering Logic:
- Checks Redis cache for recently sent notifications
- Identifies "core_alerts" for new notifications
- Detects "persistent_alerts" for ongoing issues beyond TTL
- Applies group-specific content filtering and ignore lists
- Prevents duplicate notifications within the same TTL window
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.
- metadata (dict): Workflow execution metadata for logging
- notification_package (list): ERROR-level notifications to filter
- sending_configs (list): Receiver group configurations with:
- group_name (str): Name of the receiver group
- contents (list): Alert types to include (core_alerts, persistent_alerts)
- ignore (list, optional): Notification IDs to exclude
- notification_ttl (int): Seconds before considering notification persistent
Returns:
dict[str, Any]: The filtered receiver groups with their notifications.
dict[str, Any]: Filtered receiver groups with their notifications, keyed by group_name.
Each group contains:
- All receiver group configuration fields
- notifications (list): Filtered notifications for this group
"""
metadata = input_data['metadata']
notification_package = input_data['notification_package']

2
orchestrator/utils/connectors_config.py
View File

@@ -92,6 +92,6 @@ def build_email_config():
return {
'sender_email': getenv('EMAIL_SENDER', 'sientia-alerts@aignosi.com'),
'sender_password': getenv('EMAIL_SENDER_PASSWORD', 'sientia'),
'smtp_server': getenv('EMAIL_SMTP_SERVER', 'smtp.gmail.com'),
'smtp_server': getenv('EMAIL_SMTP_SERVER', None),
'smtp_port': int(getenv('EMAIL_SMTP_PORT', '587'))
}

9
orchestrator/utils/orchestrator_functions.py
View File

@@ -17,6 +17,7 @@ def common_config(config: dict[str, Any]):
Returns:
dict[str, Any]: Common configuration dictionary with extracted parameters.
"""
model = config['model']
return {
"workflow_type": config['workflow_type'],
"schedule_name": config['schedule_name'],
@@ -24,7 +25,8 @@ def common_config(config: dict[str, Any]):
"max_retry_policy": config.get('max_retry_policy', 1),
"model_id": config['model_id'],
"model_name": config['models']['name'],
"model_name": model['name'],
"model_config": model.get('model_config', {}),
}
@@ -280,10 +282,11 @@ def build_tag_config(tag: dict[str, Any], slot_config: dict[str, Any],
"name": server_name,
"url": opc_servers[server_id]['url'],
"server_uri": opc_servers[server_id]['uri'],
"cert_path": opc_servers[server_id].get('cert_path', None),
"private_key_path": opc_servers[server_id].get('private_key_path', None),
"server_cert_path": opc_servers[server_id].get('server_cert_path', None),
"tags": {}
}
for name, spec in opc_servers[server_id].get('security_spec', {}).items():
slot_config[f"{i}"][server_name][name] = spec
slot_config[f"{i}"][server_name]["tags"][tag['tag_address']] = {
**tag,

34
orchestrator/workflows/alerts.py
View File

@@ -9,22 +9,42 @@ with workflow.unsafe.imports_passed_through():
@workflow.defn(name="alerts")
class Alerts:
"""
Alerts workflow for real-time error notification delivery.
This workflow processes ERROR-level notifications from the notification queue
and sends immediate alerts to configured user groups. It implements intelligent
filtering with TTL-based duplicate prevention and persistent alert detection
for ongoing issues.
"""
@workflow.run
async def run(self, input_data: dict[str, Any]):
"""
Workflow to send alerts to the users
Execute the alerts workflow for real-time error notification delivery.
This workflow loads ERROR-level notifications from MongoDB, applies
intelligent filtering with TTL management to prevent alert spam,
and sends immediate email alerts to configured receiver groups.
The workflow implements:
- TTL-based duplicate prevention for notifications
- Persistent alert detection for ongoing issues
- User group-based filtering with ignore lists
- Audit logging of alert delivery status
Args:
input_data (dict[str, Any]): Input data. It contains the following keys:
- schedule_name: str - Name of the schedule
- notification_ttl: int - Period before consider some notification persistent
- sent_ttl: int - Time to live for the sent notification
input_data (dict[str, Any]): Workflow input parameters.
Required fields:
- schedule_name (str): Name of the alert schedule
- notification_ttl (int): Seconds before considering notification persistent
- sent_ttl (int): Time-to-live for sent notification cache
Returns:
None
None: Workflow completes without return value
Raises:
Exception: If the workflow fails
Exception: If alert processing or delivery fails
"""
metadata = {
'metadata': {

6
orchestrator/workflows/orchestrator.py
View File

@@ -57,7 +57,8 @@ class Orchestrator:
Activities.aggregate_documents_in_mongodb,
{
**metadata,
'query': input_data['pipelines_query']
'query': input_data['pipelines_query'],
"timestamp_fields": ["updated_at"]
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
@@ -79,7 +80,8 @@ class Orchestrator:
**metadata,
'query': {
'collection': 'orchestrated_schedules'
}
},
"timestamp_fields": ["updated_at"]
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)

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

@@ -69,6 +69,9 @@ class ProcessNotifications:
retry_policy=retry_policy
)
if not log_report:
return {}
# Format the log report to a dataframe to be stored in the database
log_report = await workflow.execute_local_activity_method(
Activities.format_log_report,

2
run_local.sh
View File

@@ -15,4 +15,4 @@ else
fi
echo "Starting orchestrator application..."
python -m orchestrator.app
exec python -m orchestrator.worker.worker

142
samples.json
View File

@@ -1,142 +0,0 @@
{
"models": {
"1": {
"name": "Demo Model-Demo2"
}
},
"pipelines": {
"1": {
"schedule_name": "scouter-opcua-orchestrated-pipeline",
"model_id": "1",
"workflow_type": "scouter",
"frequency": "5s",
"max_retry_policy": 1,
"read_tags": [
{
"tag_name": "Counter",
"server_id": "1",
"aggr_func": "avg",
"tag_address": "ns=2;i=2",
"frequency": "15000",
"data_range": [
-100,
100
]
},
{
"tag_name": "Rollout",
"server_id": "1",
"aggr_func": "mdn",
"tag_address": "ns=2;i=3",
"frequency": "15000",
"data_range": [
-100,
100
]
},
{
"tag_name": "Square",
"server_id": "1",
"aggr_func": "lts",
"tag_address": "ns=2;i=4",
"frequency": "15000",
"data_range": [
-100,
100
]
}
],
"filters": [
{
"filter_name": "OUT_OF_BOUNDS_FILTER",
"policy": "DISCARD"
},
{
"filter_name": "NULL_VALUES_FILTER",
"policy": "DISCARD"
}
],
"tag_retention_minutes": 60,
"active": true,
"updated_at": "2025-07-14 10:00:00.000000"
},
"2": {
"schedule_name": "laborious-orchestrated-pipeline",
"model_id": "1",
"workflow_type": "predictions_batch",
"frequency": "30s",
"max_retry_policy": 1,
"query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;",
"retention_time": 60,
"write_tags": [
{
"server_id": "1",
"type": "prediction",
"addr": "ns=2;i=2",
"data_type": "float"
},
{
"server_id": "1",
"type": "confidence",
"addr": "ns=2;i=2",
"data_type": "float"
}
],
"input_filters": [
{
"filter_name": "EMPTY_DATA",
"policy": "STOP"
},
{
"filter_name": "SPECIFIC_VARIABLES_NULL_VALUES",
"policy": "CONTINUE",
"config": {
"variables": [
"Counter"
]
}
}
],
"mlflow_transform_filters": [
{
"filter_name": "API_ERROR",
"policy": "REPEAT"
},
{
"filter_name": "NAN_VALUES",
"policy": "STOP"
}
],
"mlflow_predict_filters": [
{
"filter_name": "API_ERROR",
"policy": "CONTINUE"
}
],
"path_priority": [
"STOP",
"CONTINUE",
"REPEAT"
],
"active": true,
"updated_at": "2025-07-14 10:00:00.000000"
},
"3": {
"schedule_name": "minimal-retrain-pipeline",
"model_id": "1",
"workflow_type": "minimal_retrain",
"frequency": "5m",
"max_retry_policy": 1,
"query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;",
"active": true,
"updated_at": "2025-07-23 10:00:00.000000"
}
},
"opc-servers": {
"1": {
"server_name": "default_server",
"url": "opc.tcp://sientia-opc-simulator.sientia.svc.cluster.local:4840",
"uri": "http://opcua-server.simulator"
}
}
}

13
tests/orchestrator/activities/test_email.py
View File

@@ -282,6 +282,19 @@ def test_try_send_email_reconnect_quit_failure(smtp, email):
assert False, "Expected exception"
@mark.asyncio
async def test_send_email_without_smtp_server(email):
email.smtp_server = None
input_data = {
**metadata,
"receiver_groups": {},
"mail_type": "test_TYPE"
}
response = await email.send_email(input_data)
assert response == {}
@mark.asyncio
@patch('orchestrator.activities.email.MIMEText')
@patch('orchestrator.activities.email.MIMEMultipart')

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

@@ -121,10 +121,7 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma
"id": "1",
"server_name": "test_server_name",
"url": "test_url",
"uri": "test_uri",
"security_spec": {
"test_name": "test_spec"
}
"uri": "test_uri"
},
{
"id": "2",
@@ -157,10 +154,7 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma
"id": "1",
"server_name": "test_server_name",
"url": "test_url",
"uri": "test_uri",
"security_spec": {
"test_name": "test_spec"
}
"uri": "test_uri"
},
"2": {
"id": "2",
@@ -186,10 +180,7 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma
"id": "1",
"server_name": "test_server_name",
"url": "test_url",
"uri": "test_uri",
"security_spec": {
"test_name": "test_spec"
}
"uri": "test_uri"
},
"2": {
"id": "2",
@@ -215,10 +206,7 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma
"id": "1",
"server_name": "test_server_name",
"url": "test_url",
"uri": "test_uri",
"security_spec": {
"test_name": "test_spec"
}
"uri": "test_uri"
},
"2": {
"id": "2",
@@ -238,7 +226,9 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma
"name": "test_server_name",
"url": "test_url",
"server_uri": "test_uri",
"test_name": "test_spec",
"cert_path": None,
"private_key_path": None,
"server_cert_path": None,
"tags": {
"test_tag_address": {
"server_id": "1",
@@ -253,6 +243,9 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma
"name": "test_server_name2",
"url": "test_url2",
"server_uri": "test_uri2",
"cert_path": None,
"private_key_path": None,
"server_cert_path": None,
"tags": {
"test_tag_address2": {
"server_id": "2",
@@ -269,6 +262,9 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma
"name": "test_server_name2",
"url": "test_url2",
"server_uri": "test_uri2",
"cert_path": None,
"private_key_path": None,
"server_cert_path": None,
"tags": {
"test_tag_address3": {
"server_id": "2",
@@ -308,15 +304,18 @@ async def test_process_slots_exception(mock_build_tag_config, mock_gather_read_t
"server_name": "test_server_name",
"url": "test_url",
"uri": "test_uri",
"security_spec": {
"test_name": "test_spec"
}
"cert_path": 'test_cert_path',
"private_key_path": 'test_private_key_path',
"server_cert_path": 'test_server_cert_path'
},
{
"id": "2",
"server_name": "test_server_name2",
"url": "test_url2",
"uri": "test_uri2"
"uri": "test_uri2",
"cert_path": 'test_cert_path2',
"private_key_path": 'test_private_key_path2',
"server_cert_path": 'test_server_cert_path2'
}
],
"active_ingestors": [

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

@@ -107,19 +107,30 @@ async def test_find_documents_in_mongodb_success(mongo_db):
}}
mock_collection = MagicMock()
mock_collection.find.return_value = [
{"_id": "12345", "name": "test1"},
{"_id": "67890", "name": "test2"}
{
"_id": "12345",
"name": "test1",
"timestamp": datetime.strptime(
"2023-01-01 12:00:00.000000+0000", DATETIME_FORMAT_MS_WITH_TZ)},
{
"_id": "67890",
"name": "test2",
"timestamp": datetime.strptime(
"2023-01-01 12:00:00.000000+0000", DATETIME_FORMAT_MS_WITH_TZ)}
]
mongo_db.database.__getitem__.return_value = mock_collection
result = await mongo_db.find_documents_in_mongodb(
{
"query": input_data
"query": input_data,
"timestamp_fields": ["timestamp"]
})
assert len(result) == 2
assert result[0] == {"name": "test1"}
assert result[1] == {"name": "test2"}
assert result[0] == {"name": "test1",
"timestamp": "2023-01-01 12:00:00.000000+0000"}
assert result[1] == {"name": "test2",
"timestamp": "2023-01-01 12:00:00.000000+0000"}
mock_collection.find.assert_called_once_with(
{"name": {"$exists": True}}, {"_id": 0}
)
@@ -186,19 +197,30 @@ async def test_aggregate_documents_in_mongodb_success(mongo_db):
]}
mock_collection = MagicMock()
mock_collection.aggregate.return_value = [
{"_id": "asdad", "name": "test1"},
{"_id": "adzx", "name": "test2"}
{
"_id": "asdad",
"name": "test1",
"timestamp": datetime.strptime(
"2023-01-01 12:00:00.000000+0000", DATETIME_FORMAT_MS_WITH_TZ)},
{
"_id": "adzx",
"name": "test2",
"timestamp": datetime.strptime(
"2023-01-01 12:00:00.000000+0000", DATETIME_FORMAT_MS_WITH_TZ)}
]
mongo_db.database.__getitem__.return_value = mock_collection
result = await mongo_db.aggregate_documents_in_mongodb(
{
"query": input_data
"query": input_data,
"timestamp_fields": ["timestamp"]
})
assert len(result) == 2
assert result[0] == {"name": "test1"}
assert result[1] == {"name": "test2"}
assert result[0] == {"name": "test1",
"timestamp": "2023-01-01 12:00:00.000000+0000"}
assert result[1] == {"name": "test2",
"timestamp": "2023-01-01 12:00:00.000000+0000"}
expected_pipeline = input_data["aggregation"]
expected_pipeline.append({"$project": {"_id": 0}})

2
tests/orchestrator/utils/test_connectors_config.py
View File

@@ -127,7 +127,7 @@ def test_build_email_config_with_defaults():
assert build_email_config() == {
'sender_email': 'sientia-alerts@aignosi.com',
'sender_password': 'sientia',
'smtp_server': 'smtp.gmail.com',
'smtp_server': None,
'smtp_port': 587
}

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

@@ -16,8 +16,11 @@ def test_common_config():
"workflow_type": "scouter",
"schedule_name": "test_schedule",
"model_id": "test_model_id",
"models": {
"name": "test_model_name"
"model": {
"name": "test_model_name",
"model_config": {
"test_config": "test_config"
}
}
}
result = common_config(config)
@@ -27,7 +30,10 @@ def test_common_config():
"frequency": "1m",
"max_retry_policy": 1,
"model_id": "test_model_id",
"model_name": "test_model_name"
"model_name": "test_model_name",
"model_config": {
"test_config": "test_config"
}
}
assert result == expected
@@ -37,8 +43,11 @@ def test_minimal_retrain():
"workflow_type": "minimal_retrain",
"schedule_name": "test_schedule",
"model_id": "test_model_id",
"models": {
"name": "test_model_name"
"model": {
"name": "test_model_name",
"model_config": {
"test_config": "test_config"
}
},
"query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;",
"datetime_columns": ["timestamp"]
@@ -51,6 +60,9 @@ def test_minimal_retrain():
"max_retry_policy": 1,
"model_id": "test_model_id",
"model_name": "test_model_name",
"model_config": {
"test_config": "test_config"
},
"query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;",
"schema": "sientia_data",
"table_name": "log_retrain",
@@ -64,8 +76,11 @@ def test_scouter():
"workflow_type": "scouter",
"schedule_name": "test_schedule",
"model_id": "test_model_id",
"models": {
"name": "test_model_name"
"model": {
"name": "test_model_name",
"model_config": {
"test_config": "test_config"
}
},
"filters": [
{
@@ -90,6 +105,9 @@ def test_scouter():
"max_retry_policy": 1,
"model_id": "test_model_id",
"model_name": "test_model_name",
"model_config": {
"test_config": "test_config"
},
"topic": "raw_test_schedule",
"trigger_laborious": False,
"filters": {
@@ -162,8 +180,11 @@ def test_predictions_batch(mock_process_path_priority,
"schedule_name": "test_schedule",
"workflow_type": "predictions_batch",
"model_id": "test_model_id",
"models": {
"name": "test_model_name"
"model": {
"name": "test_model_name",
"model_config": {
"test_config": "test_config"
}
},
"query": "test_query",
"write_tags": [
@@ -236,6 +257,9 @@ def test_predictions_batch(mock_process_path_priority,
"max_retry_policy": 1,
"model_id": "test_model_id",
"model_name": "test_model_name",
"model_config": {
"test_config": "test_config"
},
"query": "test_query",
"schema": "sientia_data",
"table_name": "predictions",
@@ -344,10 +368,7 @@ def test_build_tag_config():
"1": {
"server_name": "test_server_name",
"url": "test_url",
"uri": "test_uri",
"security_spec": {
"test_name": "test_spec"
}
"uri": "test_uri"
}
}
slot_config = {
@@ -362,14 +383,16 @@ def test_build_tag_config():
"name": "test_server_name",
"url": "test_url",
"server_uri": "test_uri",
"cert_path": None,
"private_key_path": None,
"server_cert_path": None,
"tags": {
"test_tag_address": {
"server_id": "1",
"server_name": "test_server_name",
"tag_address": "test_tag_address"
}
},
"test_name": "test_spec"
}
}
}
}

50
tests/orchestrator/workflows/subworkflows/test_process_notifications.py
View File

@@ -70,7 +70,10 @@ async def test_run(workflow_mock, process_notifications):
},
schedule_to_close_timeout=ANY,
retry_policy=ANY
),
)]
)
workflow_mock.execute_activity_method.assert_has_calls([
call(
Activities.export_data_to_postgres,
{
@@ -87,3 +90,48 @@ async def test_run(workflow_mock, process_notifications):
retry_policy=ANY
)
])
@mark.asyncio
@patch("orchestrator.workflows.subworkflows.process_notifications.workflow", new_callable=AsyncMock)
async def test_run_send_email_return_empty(workflow_mock, process_notifications):
input_data = {
'metadata': metadata,
'notification_package': ["content"],
'mail_type': 'test_mail_type',
'schema': 'test_schema',
'table_name': 'test_table_name',
}
workflow_mock.execute_activity_method.return_value = []
response = await process_notifications.run(input_data)
workflow_mock.execute_local_activity_method.assert_has_calls([
call(
Activities.build_email_html,
{
**metadata,
'receiver_groups': input_data['notification_package'],
'mail_type': input_data['mail_type']
},
schedule_to_close_timeout=ANY,
retry_policy=ANY
)
])
workflow_mock.execute_activity_method.assert_has_calls([
call(
Activities.send_email,
{
**metadata,
'receiver_groups': workflow_mock.execute_local_activity_method.return_value,
'mail_type': input_data['mail_type']
},
schedule_to_close_timeout=ANY,
retry_policy=ANY
)]
)
assert workflow_mock.execute_activity_method.call_count == 1
assert workflow_mock.execute_local_activity_method.call_count == 1

62
tests/orchestrator/workflows/test_orchestrator.py
View File

@@ -34,8 +34,9 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.aggregate_documents_in_mongodb,
{
**metadata,
"query": input_data["pipelines_query"],
**metadata
"timestamp_fields": ["updated_at"]
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -46,8 +47,8 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.find_documents_in_mongodb,
{
"query": input_data["opc_servers_query"],
**metadata
**metadata,
"query": input_data["opc_servers_query"]
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -58,10 +59,11 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.find_documents_in_mongodb,
{
**metadata,
"query": {
"collection": "orchestrated_schedules"
},
**metadata
"timestamp_fields": ["updated_at"]
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -94,8 +96,8 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.format_schedule_config,
{
'schedule_config': workflow_mock.start_local_activity_method.return_value,
**metadata
**metadata,
'schedule_config': workflow_mock.start_local_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -106,8 +108,8 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.process_schedules,
{
'pipelines': workflow_mock.start_local_activity_method.return_value,
**metadata
**metadata,
'pipelines': workflow_mock.start_local_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -118,10 +120,10 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.process_slots,
{
**metadata,
'opc_servers': workflow_mock.start_local_activity_method.return_value,
'active_ingestors': workflow_mock.start_local_activity_method.return_value,
'pipelines': workflow_mock.start_local_activity_method.return_value,
**metadata
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -132,9 +134,9 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.create_schedule_config,
{
**metadata,
'current_schedule_config': workflow_mock.start_local_activity_method.return_value,
'schedule_config': workflow_mock.start_local_activity_method.return_value,
**metadata
'schedule_config': workflow_mock.start_local_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -145,9 +147,9 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.create_slot_config,
{
**metadata,
'current_slot_config': workflow_mock.start_local_activity_method.return_value,
'slot_config': workflow_mock.start_local_activity_method.return_value,
**metadata
'slot_config': workflow_mock.start_local_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -158,8 +160,8 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.normalize_schedules,
{
'orchestrated_schedules': workflow_mock.start_local_activity_method.return_value,
**metadata
**metadata,
'orchestrated_schedules': workflow_mock.start_local_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -182,9 +184,9 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.delete_slots,
{
**metadata,
'to_delete':
workflow_mock.start_local_activity_method.return_value['to_delete'],
**metadata
workflow_mock.start_local_activity_method.return_value['to_delete']
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -195,9 +197,9 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.update_slots,
{
**metadata,
'to_insert':
workflow_mock.start_local_activity_method.return_value['to_insert'],
**metadata
workflow_mock.start_local_activity_method.return_value['to_insert']
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -208,9 +210,9 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.delete_schedules,
{
**metadata,
'schedules':
workflow_mock.start_local_activity_method.return_value['to_delete'],
**metadata
workflow_mock.start_local_activity_method.return_value['to_delete']
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -221,9 +223,9 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.create_schedules,
{
**metadata,
'schedules':
workflow_mock.start_local_activity_method.return_value['to_create'],
**metadata
workflow_mock.start_local_activity_method.return_value['to_create']
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -234,9 +236,9 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.update_schedules,
{
**metadata,
'schedules':
workflow_mock.start_local_activity_method.return_value['to_update'],
**metadata
workflow_mock.start_local_activity_method.return_value['to_update']
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -247,10 +249,10 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.report_schedule_orchestration,
{
**metadata,
'created_schedules': workflow_mock.start_activity_method.return_value,
'updated_schedules': workflow_mock.start_activity_method.return_value,
'deleted_schedules': workflow_mock.start_activity_method.return_value,
**metadata
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -261,9 +263,9 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.report_slot_orchestration,
{
**metadata,
'inserted_slots': workflow_mock.start_activity_method.return_value,
'deleted_slots': workflow_mock.start_activity_method.return_value,
**metadata
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -274,8 +276,8 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.update_pipelines_timestamps,
{
**metadata,
'updated_pipelines': workflow_mock.start_activity_method.return_value,
**metadata
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -286,8 +288,8 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.delete_pipelines_timestamps,
{
**metadata,
'deleted_pipelines': workflow_mock.start_activity_method.return_value,
**metadata
},
retry_policy=ANY,
start_to_close_timeout=ANY
@@ -298,8 +300,8 @@ async def test_run(workflow_mock, orchestrator):
call(
Activities.create_pipelines_timestamps,
{
**metadata,
'created_pipelines': workflow_mock.start_activity_method.return_value,
**metadata
},
retry_policy=ANY,
start_to_close_timeout=ANY

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.4.5"
tag: "0.4.6"
# 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: "SIENTIAPDE-1182-ajustar-laborious-para-pegar-timestamp-da-resposta-do-mlflow"
value: "SIENTIAPDE-1222-ajustar-a-library-para-fazer-o-download-do-courier"
- name: PYTHON_APP
value: "orchestrator.worker.worker"