From 137367c96379ed92fc912507e8b08eb093bcacab Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 12 Sep 2025 14:39:42 -0300 Subject: [PATCH] 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