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

SIENTIAPDE-1222

Browse Source 
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.
This commit is contained in:
vitor-aignosi
2025-09-12 14:39:42 -03:00
parent 5db411e5b5
commit 137367c963
11 changed files with 92 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
.env Normal file
View File

@@ -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"

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)

8
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

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)

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'))
}

4
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', {}),
}

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
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"
}
}
}