From 137367c96379ed92fc912507e8b08eb093bcacab Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 12 Sep 2025 14:39:42 -0300 Subject: [PATCH 01/12] SIENTIAPDE-1222 Refactor orchestration and email handling - Updated the orchestration queries to streamline model lookups and added conditions for active models. - Changed the entry point in the local run script to use the worker module. - Removed the deprecated samples.json file. - Enhanced the Email activity to skip sending if the SMTP server is not configured. - Added timestamp field handling in MongoDB activities for better data management. - Updated connectors configuration to allow for a None SMTP server. - Improved the common configuration function to include model configuration details. --- .env | 32 ++++ init_orchestration.ipynb | 56 +++---- orchestrator/activities/email.py | 14 +- orchestrator/activities/formatters.py | 8 +- orchestrator/activities/mongo_db.py | 14 ++ orchestrator/utils/connectors_config.py | 2 +- orchestrator/utils/orchestrator_functions.py | 4 +- orchestrator/workflows/orchestrator.py | 6 +- .../subworkflows/process_notifications.py | 3 + run_local.sh | 2 +- samples.json | 142 ------------------ 11 files changed, 92 insertions(+), 191 deletions(-) create mode 100644 .env delete mode 100644 samples.json diff --git a/.env b/.env new file mode 100644 index 0000000..5711d28 --- /dev/null +++ b/.env @@ -0,0 +1,32 @@ +REDIS_HOST="localhost" +REDIS_PORT="6379" +REDIS_USERNAME="default" +REDIS_PASSWORD="bdnZOpcyiL" + + +MONGODB_USERNAME="root" +MONGODB_PASSWORD="wKZDbMNU1c" +MONGODB_URL="localhost:27018" +MONGODB_DATABASE="sientia" +MONGODB_TTL_INDEX_HOURS="1" + +EMAIL_SENDER="aignosi@aignosi.com.br" +EMAIL_SENDER_PASSWORD="smtp_password" +EMAIL_SMTP_PORT="587" + +POSTGRES_HOST="localhost" +POSTGRES_PORT="5432" +POSTGRES_USER="sientia" +POSTGRES_PASSWORD="sientia" +POSTGRES_DBNAME="sientia" +POSTGRES_MIN_CONNECTIONS="10" +POSTGRES_MAX_CONNECTIONS="40" + +LOG_LEVEL="DEBUG" +HTTP_METRICS_PORT="9090" +PROJECT_NAME="sientia-orchestrator" + +TEMPORAL_HOST="localhost:7233" +TEMPORAL_NAMESPACE="default" +TEMPORAL_SCOUTER_NAMESPACE="scouter" +TEMPORAL_LABORIOUS_NAMESPACE="laborious" \ No newline at end of file diff --git a/init_orchestration.ipynb b/init_orchestration.ipynb index 7bfbd9f..9910317 100644 --- a/init_orchestration.ipynb +++ b/init_orchestration.ipynb @@ -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", diff --git a/orchestrator/activities/email.py b/orchestrator/activities/email.py index 783007c..5a61207 100644 --- a/orchestrator/activities/email.py +++ b/orchestrator/activities/email.py @@ -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) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index ba042b2..9b0dcb8 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -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 diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 5f32540..0be4e73 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -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) diff --git a/orchestrator/utils/connectors_config.py b/orchestrator/utils/connectors_config.py index 8ab0c76..a5e6b7c 100644 --- a/orchestrator/utils/connectors_config.py +++ b/orchestrator/utils/connectors_config.py @@ -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')) } diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index eeff899..fe2fc26 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -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', {}), } diff --git a/orchestrator/workflows/orchestrator.py b/orchestrator/workflows/orchestrator.py index 0b75851..d6b8ede 100644 --- a/orchestrator/workflows/orchestrator.py +++ b/orchestrator/workflows/orchestrator.py @@ -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) diff --git a/orchestrator/workflows/subworkflows/process_notifications.py b/orchestrator/workflows/subworkflows/process_notifications.py index 121de9e..f868515 100644 --- a/orchestrator/workflows/subworkflows/process_notifications.py +++ b/orchestrator/workflows/subworkflows/process_notifications.py @@ -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, diff --git a/run_local.sh b/run_local.sh index 9aaa688..4ff3c78 100755 --- a/run_local.sh +++ b/run_local.sh @@ -15,4 +15,4 @@ else fi echo "Starting orchestrator application..." -python -m orchestrator.app +python -m orchestrator.worker.worker diff --git a/samples.json b/samples.json deleted file mode 100644 index 882927f..0000000 --- a/samples.json +++ /dev/null @@ -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" - } - } -} \ No newline at end of file From 1f8d3e87777124866517d1664346bb18eabfd788 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 16 Sep 2025 15:24:04 -0300 Subject: [PATCH 02/12] SIENTIAPDE-1222 Update GITHUB_BRANCH in values.yaml to reflect new task for downloading the courier --- values.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/values.yaml b/values.yaml index 1e1ff41..9238b5d 100644 --- a/values.yaml +++ b/values.yaml @@ -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" From b816ad523b4515f40debd44d2cb22518bd777d18 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 16 Sep 2025 16:36:32 -0300 Subject: [PATCH 03/12] SIENTIAPDE-1222 Enhance README and activities with advanced notification filtering features - Updated README to include new intelligent notification filtering capabilities, including TTL-based duplicate prevention and ignore lists. - Enhanced `Formatters` activity to support notification filtering for scheduled reports and added detailed filtering logic. - Improved `SlotManager` to implement advanced notification filtering with persistent alert detection and group-based filtering. - Updated `Alerts` workflow to incorporate intelligent filtering for real-time error notifications, ensuring efficient alert delivery. --- README.md | 95 ++++++++++++++++++++++--- orchestrator/activities/formatters.py | 59 +++++++++++---- orchestrator/activities/slot_manager.py | 66 +++++++++++++---- orchestrator/workflows/alerts.py | 34 +++++++-- 4 files changed, 210 insertions(+), 44 deletions(-) diff --git a/README.md b/README.md index 8665909..eb75cfd 100644 --- a/README.md +++ b/README.md @@ -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 ``` diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 9b0dcb8..6928bd4 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -19,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 @@ -45,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 @@ -629,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'] diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index aea49cc..9accfd1 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -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'] diff --git a/orchestrator/workflows/alerts.py b/orchestrator/workflows/alerts.py index 90cef34..5f2ec31 100644 --- a/orchestrator/workflows/alerts.py +++ b/orchestrator/workflows/alerts.py @@ -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': { From dae585533ccf3197aeae8fc7748a77838828a5c7 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 16 Sep 2025 16:42:36 -0300 Subject: [PATCH 04/12] SIENTIAPDE-1222 Update image tag in values.yaml to version 0.4.6 --- values.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/values.yaml b/values.yaml index 9238b5d..90b8107 100644 --- a/values.yaml +++ b/values.yaml @@ -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: From e31c5eece1667585470d7eea726e9d25766ef06d Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 16 Sep 2025 17:36:39 -0300 Subject: [PATCH 05/12] SIENTIAPDE-1222 SIENTIAPDE-1222 Enhance tests for email and MongoDB activities - Added a test for sending emails when no SMTP server is configured, ensuring proper handling of such cases. - Updated MongoDB tests to include timestamp fields in the results, improving data accuracy in document retrieval. - Modified connector configuration tests to reflect changes in SMTP server settings. --- tests/orchestrator/activities/test_email.py | 13 ++++ .../orchestrator/activities/test_mongo_db.py | 42 ++++++++++--- .../utils/test_connectors_config.py | 2 +- .../utils/test_orchestrator_functions.py | 42 ++++++++++--- .../test_process_notifications.py | 50 ++++++++++++++- .../workflows/test_orchestrator.py | 62 ++++++++++--------- 6 files changed, 160 insertions(+), 51 deletions(-) diff --git a/tests/orchestrator/activities/test_email.py b/tests/orchestrator/activities/test_email.py index a01167d..19c3408 100644 --- a/tests/orchestrator/activities/test_email.py +++ b/tests/orchestrator/activities/test_email.py @@ -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') diff --git a/tests/orchestrator/activities/test_mongo_db.py b/tests/orchestrator/activities/test_mongo_db.py index 7dedc60..bce033d 100644 --- a/tests/orchestrator/activities/test_mongo_db.py +++ b/tests/orchestrator/activities/test_mongo_db.py @@ -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}}) diff --git a/tests/orchestrator/utils/test_connectors_config.py b/tests/orchestrator/utils/test_connectors_config.py index f1a62b9..13977fa 100644 --- a/tests/orchestrator/utils/test_connectors_config.py +++ b/tests/orchestrator/utils/test_connectors_config.py @@ -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 } diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index cd59478..de4a1d5 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -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", diff --git a/tests/orchestrator/workflows/subworkflows/test_process_notifications.py b/tests/orchestrator/workflows/subworkflows/test_process_notifications.py index 8a41654..3ffc3fe 100644 --- a/tests/orchestrator/workflows/subworkflows/test_process_notifications.py +++ b/tests/orchestrator/workflows/subworkflows/test_process_notifications.py @@ -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 diff --git a/tests/orchestrator/workflows/test_orchestrator.py b/tests/orchestrator/workflows/test_orchestrator.py index c3e0257..e89cf73 100644 --- a/tests/orchestrator/workflows/test_orchestrator.py +++ b/tests/orchestrator/workflows/test_orchestrator.py @@ -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 From 02b647931f05c7775b02f128dc8c2a2d223bbdf6 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 19 Sep 2025 08:56:40 -0300 Subject: [PATCH 06/12] SIENTIAPDE-1222 Enhance OPC server configuration in orchestrator_functions.py - Added optional fields for certificate paths in the build_tag_config function to improve security configuration flexibility. --- orchestrator/utils/orchestrator_functions.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index fe2fc26..79f1070 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -282,6 +282,9 @@ 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(): From c3f5ba660fda6377111c5b7cfb95f1f0a839d9e0 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 24 Sep 2025 09:13:22 -0300 Subject: [PATCH 07/12] SIENTIAPDE-1222 Enhance test cases for server configuration in test_formatters.py and test_orchestrator_functions.py - Added optional fields for certificate paths in the test cases to align with recent changes in server configuration, improving test coverage and ensuring proper handling of security configurations. --- tests/orchestrator/activities/test_formatters.py | 9 +++++++++ tests/orchestrator/utils/test_orchestrator_functions.py | 3 +++ 2 files changed, 12 insertions(+) diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index 6fd32fb..cf2e17a 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -239,6 +239,9 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma "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 +256,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 +275,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", diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index de4a1d5..78cfe1b 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -386,6 +386,9 @@ 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", From 52fd68f469961f91799ede7bc69274406c82267f Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 24 Sep 2025 10:26:09 -0300 Subject: [PATCH 08/12] Update orchestrator/workflows/subworkflows/process_notifications.py Co-authored-by: codeant-ai[bot] <151821869+codeant-ai[bot]@users.noreply.github.com> --- orchestrator/workflows/subworkflows/process_notifications.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/orchestrator/workflows/subworkflows/process_notifications.py b/orchestrator/workflows/subworkflows/process_notifications.py index f868515..7809ac1 100644 --- a/orchestrator/workflows/subworkflows/process_notifications.py +++ b/orchestrator/workflows/subworkflows/process_notifications.py @@ -70,7 +70,7 @@ class ProcessNotifications: ) if not log_report: - return + return {} # Format the log report to a dataframe to be stored in the database log_report = await workflow.execute_local_activity_method( From 0991cc74ab8e1db5931e96805f83cbabd1b5c983 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 24 Sep 2025 10:26:20 -0300 Subject: [PATCH 09/12] Update run_local.sh Co-authored-by: codeant-ai[bot] <151821869+codeant-ai[bot]@users.noreply.github.com> --- run_local.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/run_local.sh b/run_local.sh index 4ff3c78..af97775 100755 --- a/run_local.sh +++ b/run_local.sh @@ -15,4 +15,4 @@ else fi echo "Starting orchestrator application..." -python -m orchestrator.worker.worker +exec python -m orchestrator.worker.worker From 60cc55fdd0d982a657b6ba1c558d528c340fef2e Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 24 Sep 2025 10:36:35 -0300 Subject: [PATCH 10/12] SIENTIAPDE-1222 Remove .env file containing sensitive configuration details, enhancing security by eliminating hardcoded credentials. --- .env | 32 -------------------------------- 1 file changed, 32 deletions(-) delete mode 100644 .env diff --git a/.env b/.env deleted file mode 100644 index 5711d28..0000000 --- a/.env +++ /dev/null @@ -1,32 +0,0 @@ -REDIS_HOST="localhost" -REDIS_PORT="6379" -REDIS_USERNAME="default" -REDIS_PASSWORD="bdnZOpcyiL" - - -MONGODB_USERNAME="root" -MONGODB_PASSWORD="wKZDbMNU1c" -MONGODB_URL="localhost:27018" -MONGODB_DATABASE="sientia" -MONGODB_TTL_INDEX_HOURS="1" - -EMAIL_SENDER="aignosi@aignosi.com.br" -EMAIL_SENDER_PASSWORD="smtp_password" -EMAIL_SMTP_PORT="587" - -POSTGRES_HOST="localhost" -POSTGRES_PORT="5432" -POSTGRES_USER="sientia" -POSTGRES_PASSWORD="sientia" -POSTGRES_DBNAME="sientia" -POSTGRES_MIN_CONNECTIONS="10" -POSTGRES_MAX_CONNECTIONS="40" - -LOG_LEVEL="DEBUG" -HTTP_METRICS_PORT="9090" -PROJECT_NAME="sientia-orchestrator" - -TEMPORAL_HOST="localhost:7233" -TEMPORAL_NAMESPACE="default" -TEMPORAL_SCOUTER_NAMESPACE="scouter" -TEMPORAL_LABORIOUS_NAMESPACE="laborious" \ No newline at end of file From dacbcbf56a7577670d265c7b648b94d558e1007c Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 24 Sep 2025 10:37:05 -0300 Subject: [PATCH 11/12] SIENTIAPDE-1222 Update .gitignore to include .env file and ensure git_log is tracked, improving project configuration management. --- .gitignore | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index 9a3e1d1..c503b7f 100644 --- a/.gitignore +++ b/.gitignore @@ -43,4 +43,6 @@ htmlcov/ # git keys git_key* -git_log \ No newline at end of file +git_log + +.env \ No newline at end of file From 49f3429556a1cd4afc96c448db42b98764c3f868 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 24 Sep 2025 10:57:38 -0300 Subject: [PATCH 12/12] SIENTIAPDE-1222 SIENTIAPDE-1222 Refactor server configuration handling in orchestrator_functions.py and update related tests - Removed the security_spec field from server configuration in build_tag_config to streamline the configuration process. - Updated test cases in test_formatters.py and test_orchestrator_functions.py to reflect the removal of security_spec, ensuring alignment with the new configuration structure and improving test accuracy. --- orchestrator/utils/orchestrator_functions.py | 2 -- .../activities/test_formatters.py | 32 +++++++------------ .../utils/test_orchestrator_functions.py | 8 ++--- 3 files changed, 13 insertions(+), 29 deletions(-) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 79f1070..64499b0 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -287,8 +287,6 @@ def build_tag_config(tag: dict[str, Any], slot_config: dict[str, Any], "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, diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index cf2e17a..aec0e12 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -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,6 @@ 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, @@ -317,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": [ diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index 78cfe1b..28edc73 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -368,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 = { @@ -395,8 +392,7 @@ def test_build_tag_config(): "server_name": "test_server_name", "tag_address": "test_tag_address" } - }, - "test_name": "test_spec" + } } } }