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

SIENTIAPDE-1172

Browse Source 
feat: enhance orchestrator activities with new email and Postgres integrations

- Added Email and Postgres classes to the Activities class for improved functionality.
- Introduced new methods in MongoDB and SlotManager for loading and managing data.
- Updated requirements.txt to include jinja2.
- Added new formatting activity for log reports in Formatters class.
- Enhanced test coverage for MongoDB and SlotManager activities.
This commit is contained in:
vitor-aignosi
2025-07-25 16:03:39 -03:00
parent b0171614aa
commit 2a5534cf64
19 changed files with 1660 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

247
email.html Normal file
View File

@@ -0,0 +1,247 @@
<!DOCTYPE html>
<html lang="pt-BR">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>SIENTIA™ Report</title>
<style>
body { font-family: Arial, sans-serif; margin: 0; padding: 20px; }
h1, h2 { color: #333; }
table { width: 100%; border-collapse: collapse; margin-top: 20px; }
th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
th { background-color: #f4f4f4; }
</style>
</head>
<body>
<h1>SIENTIA™ Alerts</h1>
<h3>Errors detected:</h3>
<h4>Model: <span>ut aliqua ut ipsum</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>TAG_node5:tag_87_LISTENNING_STOPPED</td>
<td>dolor_elit_aliqua</td>
<td>2025-07-25 16:50:48.925421+00:00</td>
<td>sit elit ipsum labore incididunt magna do et sit incididunt</td>
</tr>
</tbody>
</table>
<h4>Model: <span>elit dolore</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>OPC_LISTENNING_BACK__server_10</td>
<td>magna_magna_et</td>
<td>2025-07-25 16:50:48.925478+00:00</td>
<td>aliqua eiusmod ut eiusmod ut et ipsum</td>
</tr>
</tbody>
</table>
<h4>Model: <span>lorem lorem ipsum amet</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>TAG_node8:tag_11_LISTENNING_STOPPED</td>
<td>do_sit</td>
<td>2025-07-25 16:50:48.925506+00:00</td>
<td>labore do adipiscing adipiscing incididunt eiusmod</td>
</tr>
</tbody>
</table>
<h3>Warnings detected:</h3>
<h4>Model: <span>amet eiusmod dolor ipsum</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>REPORT_PARTITION_MANAGER</td>
<td>adipiscing_amet_ut_et</td>
<td>2025-07-25 16:50:48.925368+00:00</td>
<td>eiusmod et amet sit lorem incididunt</td>
</tr>
</tbody>
</table>
<h4>Model: <span>sit et labore</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>SIT_CONSECTETUR_EIUSMOD</td>
<td>adipiscing_elit_ut</td>
<td>2025-07-25 16:50:48.925392+00:00</td>
<td>magna do eiusmod et labore elit</td>
</tr>
</tbody>
</table>
<h4>Model: <span>consectetur labore do consectetur</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>SED_SIT_DOLOR_CONSECTETUR</td>
<td>dolor_dolore_ut</td>
<td>2025-07-25 16:50:48.925435+00:00</td>
<td>magna sed incididunt dolor tempor</td>
</tr>
</tbody>
</table>
<h4>Model: <span>et sed ut et</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>UT_UT_LABORE</td>
<td>sit_do</td>
<td>2025-07-25 16:50:48.925492+00:00</td>
<td>do sed sed elit do eiusmod</td>
</tr>
</tbody>
</table>
<h3>Infos detected:</h3>
<h4>Model: <span>eiusmod eiusmod</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>OPC_CONNECTION_RETRY__server_2</td>
<td>labore_sit_do</td>
<td>2025-07-25 16:50:48.925407+00:00</td>
<td>magna et ipsum amet dolore adipiscing adipiscing do</td>
</tr>
</tbody>
</table>
<h4>Model: <span>adipiscing consectetur</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>REPORT_PARTITION_MANAGER</td>
<td>incididunt_aliqua_ipsum_ut</td>
<td>2025-07-25 16:50:48.925447+00:00</td>
<td>labore magna adipiscing tempor elit et et magna elit lorem</td>
</tr>
</tbody>
</table>
<h4>Model: <span>et amet</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>OPC_LISTENNING_BACK__server_2</td>
<td>sed_elit_ut_do</td>
<td>2025-07-25 16:50:48.925462+00:00</td>
<td>dolor et adipiscing adipiscing dolore ipsum ipsum aliqua et elit</td>
</tr>
</tbody>
</table>
</body>
</html>

287
notification_generator.ipynb Normal file
View File

File diff suppressed because one or more lines are too long

26
orchestrator/activities/activities.py
View File

@@ -1,6 +1,7 @@
from temporalio import activity, workflow from temporalio import activity, workflow
from temporalio.client import Client from temporalio.client import Client
from orchestrator.activities.email import Email
from orchestrator.activities.mongo_db import MongoDB from orchestrator.activities.mongo_db import MongoDB
with workflow.unsafe.imports_passed_through(): with workflow.unsafe.imports_passed_through():
@@ -10,17 +11,21 @@ with workflow.unsafe.imports_passed_through():
from orchestrator.activities.formatters import Formatters from orchestrator.activities.formatters import Formatters
from typing import Any from typing import Any
from logging import Logger from logging import Logger
from sientia_do.temporal.activities.postgres import Postgres
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
class Activities( # Couchbase, class Activities( # Couchbase,
TemporalManager, SlotManager, Formatters, MongoDB): TemporalManager, SlotManager, Formatters, MongoDB, Email,
Postgres):
def __init__(self, def __init__(self,
temporal_config: dict[str, Any], temporal_config: dict[str, Any],
# couchbase_config: dict[str, Any], # couchbase_config: dict[str, Any],
redis_config: dict[str, Any], redis_config: dict[str, Any],
mongodb_config: dict[str, Any], mongodb_config: dict[str, Any],
email_config: dict[str, Any],
postgres_config: dict[str, Any],
logger: Logger, logger: Logger,
notification_handler: NotificationHandler): notification_handler: NotificationHandler):
@@ -59,5 +64,24 @@ class Activities( # Couchbase,
logger=logger, logger=logger,
notification_handler=notification_handler) notification_handler=notification_handler)
Email.__init__(self,
sender_email=email_config['sender_email'],
sender_password=email_config['sender_password'],
smpt_server=email_config['smpt_server'],
port=email_config['port'],
logger=logger,
notification_handler=notification_handler)
Postgres.__init__(self,
host=postgres_config['host'],
port=postgres_config['port'],
user=postgres_config['username'],
password=postgres_config['password'],
dbname=postgres_config['database_name'],
min_connections=postgres_config['min_connections'],
max_connections=postgres_config['max_connections'],
logger=logger,
notification_handler=notification_handler)
def shutdown(self): def shutdown(self):
MongoDB.shutdown(self) MongoDB.shutdown(self)

149
orchestrator/activities/email.py Normal file
View File

@@ -0,0 +1,149 @@
from email import encoders
from email.mime.base import MIMEBase
import traceback
from temporalio import workflow, activity
with workflow.unsafe.imports_passed_through():
import smtplib
from typing import Any
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.temporal.utils.logger import Logger
from sientia_do.notifications.handlers import NotificationHandler
from orchestrator.utils.email_builder import EmailBuilder
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
class Email(BaseActivity):
def __init__(self, sender_email: str, sender_password: str,
smpt_server: str, port: int,
logger: Logger, notification_handler: NotificationHandler):
self.email_builder = EmailBuilder(logger=logger)
self.sender_email = sender_email
self.sender_password = sender_password
self.port = port
self.logger = logger
if self.sender_password:
self.server = smtplib.SMTP_SSL(smpt_server, port)
self.server.login(self.sender_email, self.sender_password)
else:
self.server = smtplib.SMTP(smpt_server, port)
BaseActivity.__init__(self,
logger=logger,
notification_handler=notification_handler)
@activity.defn(name="build_email_html")
async def build_email_html(self, input_data: dict[str, Any]) -> str:
"""
Builds the email html for each receiver group.
input_data:
- receiver_groups (dict): The receiver groups.
- mail_type (str): The mail type.
"""
metadata = input_data['metadata']
receiver_groups = input_data['receiver_groups']
mail_type = input_data['mail_type']
self.info(f"Building email html for {mail_type} mail type.",
metadata=metadata)
for group_name, group_config in receiver_groups.items():
html = self.email_builder.build_email(
group_config['notifications'], mail_type)
group_config['html'] = html
self.info(f"Email html built for {mail_type} mail type.",
metadata=metadata)
return receiver_groups
def handle_attachments(self, attachments: list[dict], msg: MIMEMultipart) -> MIMEMultipart:
"""
Attaches a list of attachments to an email message.
Args:
attachments (List[Dict]): A list of dictionaries where each dictionary contains
the keys 'attachment_id', 'trigger', and 'timestamp' representing the attachment details.
msg (MIMEMultipart): The email message object to which the attachments will be added.
Returns:
MIMEMultipart: The email message object with the attachments added.
Raises:
Exception: If an attachment cannot be added, an error is logged.
"""
for attachment in attachments:
att_name = attachment['filename']
try:
# Create the attachment as a MIMEBase object
part = MIMEBase('application', 'octet-stream')
part.set_payload(
attachment['attachment_content'].encode('utf-8'))
encoders.encode_base64(part)
part.add_header(
'Content-Disposition',
f'attachment; filename="{att_name}"'
)
msg.attach(part)
except Exception as e:
self.logger.error(
f"Failed to attach content of {att_name}: {e}")
return msg
@activity.defn(name="send_email")
async def send_email(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Sends an email to the receivers of each group.
input_data:
- receiver_groups (dict): The receiver groups.
- mail_type (str): The mail type.
"""
metadata = input_data['metadata']
receiver_groups = input_data['receiver_groups']
mail_type = input_data['mail_type']
self.info(f"Sending email for {mail_type} mail type.",
metadata=metadata)
for group_name, group_config in receiver_groups.items():
receivers = ", ".join(group_config['members'])
self.info(f"Sending email to {group_name}: {receivers}",
metadata=metadata)
msg = MIMEMultipart()
msg.attach(MIMEText(group_config['html'], 'html'))
msg['From'] = self.sender_email
msg['To'] = receivers
msg['Subject'] = f"SIENTIA™ {mail_type}"
msg = self.handle_attachments(
[notification['attachment_content']
for notification in group_config['notifications']
if notification['attachment_content']],
msg)
try:
self.server.sendmail(
self.sender_email, receivers, msg.as_string())
except Exception as e:
self.error(f"Failed to send email to {group_name}: {e}",
metadata=metadata)
traceback.print_exc()
group_config['status'] = 'failed'
else:
group_config['status'] = 'sent'
self.info(f"Email sent to {group_name}.",
metadata=metadata)
self.info(f"Email sent for {mail_type} mail type.",
metadata=metadata)
return receiver_groups

45
orchestrator/activities/formatters.py
View File

@@ -1,4 +1,5 @@
from pandas import DataFrame
from temporalio import activity, workflow from temporalio import activity, workflow
from orchestrator.utils.orchestrator_functions import minimal_retrain from orchestrator.utils.orchestrator_functions import minimal_retrain
@@ -488,3 +489,47 @@ class Formatters(BaseActivity):
notification_id="REPORT_ORCHESTRATION_DELETED_SLOTS", notification_id="REPORT_ORCHESTRATION_DELETED_SLOTS",
attachment=deleted_slots attachment=deleted_slots
) )
@activity.defn(name="format_log_report")
async def format_log_report(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Formats the receiver_groups status to a dataframe to be stored in the database.
input_data:
- receiver_groups (dict): The receiver groups.
"""
metadata = input_data["metadata"]
mail_type = input_data["mail_type"]
self.info("Formatting log report...", metadata=metadata)
receiver_groups = input_data['receiver_groups']
data = {}
for group_name, group_config in receiver_groups.items():
notification_id = group_config['notification_id']
trigger = group_config['trigger']
key = f"{notification_id}:{trigger}"
if key not in data:
data[key] = {
'status': group_config['status'],
'timestamp': group_config['timestamp'],
'groups': [group_name],
'message': group_config['message'],
'level': group_config['level'],
'notification_id': notification_id,
'block': group_config['block'],
'schedule': trigger,
'pipeline': group_config['pipeline'],
'project': group_config['project'],
'model_name': group_config['model_name'],
'model_id': group_config['model_id'],
'mail_type': mail_type
}
else:
data[key]['groups'].append(group_name)
return DataFrame(list(data.values())).to_dict()

89
orchestrator/activities/mongo_db.py
View File

@@ -1,3 +1,4 @@
from pandas import DataFrame
from temporalio import workflow, activity from temporalio import workflow, activity
@@ -80,6 +81,15 @@ class MongoDB(BaseActivity):
""" """
self.shutdown() self.shutdown()
def find(self, collection_name: str, filters: dict[str, Any]) -> list[dict[str, Any]]:
collection = self.database[collection_name]
documents = list(collection.find(filters, {"_id": 0}))
documents = clear_mongo_id(documents)
return documents
@activity.defn(name="find_documents_in_mongodb",) @activity.defn(name="find_documents_in_mongodb",)
async def find_documents_in_mongodb(self, input_data: dict[str, Any]) -> list[dict[str, Any]]: async def find_documents_in_mongodb(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
""" """
@@ -106,11 +116,7 @@ class MongoDB(BaseActivity):
f"Loading documents from collection '{collection_name}' with filters: {filters}", metadata=metadata) f"Loading documents from collection '{collection_name}' with filters: {filters}", metadata=metadata)
try: try:
collection = self.database[collection_name] documents = self.find(collection_name, filters)
documents = list(collection.find(filters, {"_id": 0}))
documents = clear_mongo_id(documents)
self.info( self.info(
f"Loaded {len(documents)} documents from collection '{collection_name}'", metadata=metadata) f"Loaded {len(documents)} documents from collection '{collection_name}'", metadata=metadata)
@@ -352,3 +358,76 @@ class MongoDB(BaseActivity):
) )
self.error(trace, metadata=metadata) self.error(trace, metadata=metadata)
raise e raise e
@activity.defn(name="load_latest_data")
async def load_latest_data(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
"""
Loads the latest data from MongoDB.
input_data:
- metadata (dict): The metadata of the workflow.
- collection_name (str): The name of the collection to load data from.
- last_data_timestamp (str): The timestamp of the last data to load.
- base_data_filter (dict): The base data filter to apply to the query.
returns:
- data (list[dict]): The data loaded from MongoDB.
"""
metadata = input_data['metadata']
collection_name = input_data['collection_name']
last_data_timestamp = input_data['last_data_timestamp']
base_data_filter = input_data['base_data_filter']
self.debug(
f"Loading data from MongoDB: {input_data}",
metadata=metadata
)
try:
if last_data_timestamp is None:
data_filter = base_data_filter
else:
data_filter = {
**base_data_filter,
"timestamp": {
"$gt": datetime.strptime(last_data_timestamp, DEFAULT_DATE_FORMAT)
}
}
self.debug(
f"Data filter: {data_filter}",
metadata=metadata
)
data = self.find(collection_name, data_filter)
self.debug(
f"Collected: {data}",
metadata=metadata
)
for item in data:
item['timestamp'] = item['timestamp'].strftime(
DEFAULT_DATE_FORMAT)
self.info(
f"Loaded {len(data)} documents from MongoDB",
metadata=metadata
)
self.debug(
f"Loaded data: {data}",
metadata=metadata
)
return data
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
metadata=metadata,
notification_id="MONGO_LOAD_ERROR",
message=f"Error loading data from MongoDB: {e}",
block="load_latest_data",
level=NotificationLevel.ERROR,
attachment_content=trace
)
raise e

70
orchestrator/activities/slot_manager.py
View File

@@ -1,3 +1,4 @@
from pandas import DataFrame
from temporalio import activity, workflow from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through(): with workflow.unsafe.imports_passed_through():
@@ -193,3 +194,72 @@ class SlotManager(Redis):
f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}", metadata=metadata) f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}", metadata=metadata)
return report return report
@activity.defn(name="get_last_data_timestamp")
async def get_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None:
"""
Gets the last data timestamp from redis.
"""
metadata = input_data['metadata']
key = "notification_last_timestamp"
try:
data_hold = self.get(key)
except Exception as e:
self.send_notification(
metadata=metadata,
notification_id="REDIS_GET_ERROR",
message=f"Error getting last data timestamp: {e}",
block="get_last_data_timestamp",
level=NotificationLevel.ERROR,
attachment_content=traceback.format_exc()
)
raise e
self.debug(
f"Last collected timestamp: {data_hold}",
metadata=metadata
)
if not data_hold:
return None
return data_hold
@activity.defn(name="put_last_data_timestamp")
async def put_last_data_timestamp(self, input_data: dict[str, Any]):
"""
Puts the last data timestamp into redis.
"""
metadata = input_data['metadata']
key = "notification_last_timestamp"
data = DataFrame(input_data['data'])
if data.empty:
self.warning("No data to insert",
metadata=metadata
)
return None
last_data_timestamp = data['timestamp'].max()
self.debug(
f"Last collected timestamp to insert: {last_data_timestamp}",
metadata=metadata
)
try:
self.set(key, last_data_timestamp, ttl=60*60*5)
except Exception as e:
self.send_notification(
metadata=metadata,
notification_id="REDIS_SET_ERROR",
message=f"Error setting last data timestamp: {e}",
block="put_last_data_timestamp",
level=NotificationLevel.ERROR,
attachment_content=traceback.format_exc()
)
raise e
return last_data_timestamp

75
orchestrator/utils/email_builder.py Normal file
View File

@@ -0,0 +1,75 @@
import json
from sientia_do.temporal.utils.logger import Logger
from sientia_do.notifications.models import NotificationLevel
from jinja2 import Template
import re
class EmailBuilder:
def __init__(self, logger: Logger):
self.logger = logger
self.report_template_file = './orchestrator/utils/templates/email_template.html'
self.general_template_file = './orchestrator/utils/templates/general_template.html'
self.tag_template_file = './orchestrator/utils/templates/opc_tag_report_template.html'
self.opc_template_file = './orchestrator/utils/templates/opc_connection_report_template.html'
self.partition_manager_template_file = './orchestrator/utils/templates/partition_manager_report_template.html'
with open(self.report_template_file, 'r') as file:
self.report_template = file.read()
with open(self.general_template_file, 'r') as file:
self.general_template = file.read()
def replace_parameters(self, template: str, parameters: dict) -> str:
# Criar um template Jinja2
template = Template(template)
return template.render(parameters)
def parameters(self, report_data: dict, general_events: dict, mail_type: str) -> dict:
return {
'project_name': report_data[0]['project'],
'mail_type': mail_type,
'error_events': self.replace_parameters(self.general_template,
general_events['ERROR']) if general_events['ERROR']['models'] else '',
'warning_events': self.replace_parameters(self.general_template,
general_events['WARNING']) if general_events['WARNING']['models'] else '',
'info_events': self.replace_parameters(self.general_template,
general_events['INFO']) if general_events['INFO']['models'] else '',
}
def build_email(self, report_data: list[dict], mail_type: str) -> str:
"""
Builds the email html.
"""
general_events = {}
for report in report_data:
level = report['level']
model_name = report['model_name']
if level not in general_events:
general_events[level] = {
'section_name': f'{level.capitalize()}s detected:',
'models': {}
}
if model_name not in general_events[level]['models']:
general_events[level]['models'][model_name] = {
'model_name': model_name,
'events': []
}
general_events[level]['models'][model_name]['events'].append(
report)
print(general_events)
for _type, content in general_events.items():
content['models'] = list(content['models'].values())
return self.replace_parameters(self.report_template, self.parameters(
report_data, general_events, mail_type
))

24
orchestrator/utils/templates/email_template.html Normal file
View File

@@ -0,0 +1,24 @@
<!DOCTYPE html>
<html lang="pt-BR">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>SIENTIA™ Report</title>
<style>
body { font-family: Arial, sans-serif; margin: 0; padding: 20px; }
h1, h2 { color: #333; }
table { width: 100%; border-collapse: collapse; margin-top: 20px; }
th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
th { background-color: #f4f4f4; }
</style>
</head>
<body>
<h1>SIENTIA™ {{ mail_type }}</h1>
{{ error_events }}
{{ warning_events }}
{{ info_events }}
{{ special_events }}
</body>
</html>

24
orchestrator/utils/templates/general_template.html Normal file
View File

@@ -0,0 +1,24 @@
<h3>{{ section_name }}</h3>
{% for model in models %}
<h4>Model: <span>{{ model.model_name }}</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
{% for event in model.events %}
<tr>
<td>{{ event.notification_id }}</td>
<td>{{ event.block }}</td>
<td>{{ event.timestamp }}</td>
<td>{{ event.message }}</td>
</tr>
{% endfor %}
</tbody>
</table>
{% endfor %}

22
orchestrator/workflows/alerts.py Normal file
View File

@@ -0,0 +1,22 @@
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from orchestrator.activities.activities import Activities
from typing import Any
@workflow.defn(name="alerts")
class Alerts:
@workflow.run
async def run(self, input_data: dict[str, Any]):
# Call subworkflow "load_notification_package" passing the static filters
# (level = "ERROR" and timestamp > last timestamp)
# Filter notification package by groups custom configs, levels and
# timestamp cached
# Call subworkflow "process_notifications" passing the notification package
# Store the notification_id sendings to avoid sending them again
pass

9
orchestrator/workflows/orchestrator.py
View File

@@ -11,6 +11,15 @@ with workflow.unsafe.imports_passed_through():
class Orchestrator: class Orchestrator:
@workflow.run @workflow.run
async def run(self, input_data: dict[str, Any]): async def run(self, input_data: dict[str, Any]):
"""
Orchestrates the pipeline and slot management. Gets configuration from MongoDB and Redis,
creates the configuration and deploys the schedules and slots in the Temporal server and
Redis server.
input_data:
- schedule_name (str): The name of the schedule.
- pipelines_query (dict): The query to get the pipelines.
- opc_servers_query (dict): The query to get the OPC servers.
"""
input_data['workflow_name'] = 'orchestrator' input_data['workflow_name'] = 'orchestrator'

19
orchestrator/workflows/reports.py Normal file
View File

@@ -0,0 +1,19 @@
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from orchestrator.activities.activities import Activities
from typing import Any
@workflow.defn(name="reports")
class Reports:
@workflow.run
async def run(self, input_data: dict[str, Any]):
# Call subworkflow "load_notification_package" passing the static filters
# (timestamp > last timestamp)
# Filter notification package by groups custom configs
# Call subworkflow "process_notifications" passing the notification package
pass

93
orchestrator/workflows/subworkflows/load_notification_package.py Normal file
View File

@@ -0,0 +1,93 @@
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from orchestrator.activities.activities import Activities
from typing import Any
from sientia_do.temporal.utils.policies import retry_policy
from datetime import timedelta
@workflow.defn(name="load_notification_package")
class LoadNotificationPackage:
@workflow.run
async def run(self, input_data: dict[str, Any]):
"""
Loads the notification package from the MongoDB collection "notification_queue"
and the sending configs from the MongoDB collection "receiver_groups".
input_data:
- metadata (dict): The metadata of the workflow.
returns:
- last_timestamp (str): The last timestamp of the notification package.
- notification_package (list[dict]): The notification package.
- sending_configs (list[dict]): The sending configs.
"""
metadata = input_data['metadata']
# Load last timestamp from redis "notification_last_timestamp"
last_timestamp_handler = workflow.start_local_activity_method(
Activities.get_last_data_timestamp,
{
**metadata,
},
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
# In parallel, load sending configs from collection "receiver_groups"
sending_configs_handler = workflow.start_local_activity_method(
Activities.find_documents_in_mongodb,
{
**metadata,
'query': {
'collection': 'receiver_groups',
'filters': {
'active': True
}
}
},
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
last_timestamp = await last_timestamp_handler
# Load notification package from collection "notification_queue", using a
# static filter
notification_package = await workflow.start_local_activity_method(
Activities.load_latest_data,
{
**metadata,
'collection_name': 'notification_queue',
'last_data_timestamp': last_timestamp,
'base_data_filter': {
'level': 'ERROR'
}
},
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
# Put last collected timestamp in redis "notification_last_timestamp"
await workflow.start_activity_method(
Activities.put_last_data_timestamp,
{
**metadata,
'data': notification_package
},
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
# Return a dict with the following keys:
# - last_timestamp
# - notification_package
# - sending_configs
return {
'last_timestamp': last_timestamp,
'notification_package': notification_package,
'sending_configs': await sending_configs_handler
}

87
orchestrator/workflows/subworkflows/process_notifications.py Normal file
View File

@@ -0,0 +1,87 @@
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from orchestrator.activities.activities import Activities
from typing import Any
from datetime import timedelta
from sientia_do.temporal.utils.policies import retry_policy
@workflow.defn(name="process_notifications")
class ProcessNotifications:
@workflow.run
async def run(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Processes the notifications. Builds the report html for each group and each model,
sends the report html to the receivers of each group, stores the sending log in the
postgres database "log_report", and returns the log report to the caller.
input_data:
- metadata (dict): The metadata of the workflow.
- mail_type (str): The mail type.
- schema (str): The schema of the table.
- table_name (str): The name of the table.
- notification_package (list[dict]): The notification package. the format of each
notification package is:
{
'group_name' (str)
'group_members' (list[str])
'notifications' (dict)
{
'model_name' (dict[str, list[dict]])
}
}
returns:
- log_report (dict)
"""
metadata = input_data["metadata"]
# Use notification package to create the report html for each group and each model
data_to_sent = await workflow.execute_local_activity_method(
Activities.build_email_html,
{
**metadata,
"receiver_groups": input_data["notification_package"]
},
schedule_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
# Send the report html to the receivers of each group
log_report = await workflow.execute_activity_method(
Activities.send_email,
{
**metadata,
"receiver_groups": data_to_sent,
"mail_type": input_data["mail_type"]
},
schedule_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
# 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,
{
**metadata,
"receiver_groups": log_report,
"mail_type": input_data["mail_type"]
},
)
# Store sending log in postgres database "log_report"
await workflow.execute_activity_method(
Activities.export_data_to_postgres,
{
**metadata,
"schema": input_data["schema"],
"table_name": input_data["table_name"],
"data": log_report
},
schedule_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
# Return the log report to the caller
return log_report

1
requirements.txt
View File

@@ -4,4 +4,5 @@ sqlalchemy
redis redis
couchbase couchbase
pymongo pymongo
jinja2
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.4 git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.4

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

@@ -1,4 +1,5 @@
from curses import meta from curses import meta
from datetime import datetime
from unittest.mock import MagicMock, patch, ANY from unittest.mock import MagicMock, patch, ANY
from pytest import fixture, mark from pytest import fixture, mark
from orchestrator.activities.mongo_db import clear_mongo_id from orchestrator.activities.mongo_db import clear_mongo_id
@@ -518,3 +519,133 @@ async def test_create_collection_with_ttl_index_failure(mongo_db):
) )
else: else:
assert False, "Expected an exception to be raised" assert False, "Expected an exception to be raised"
@mark.asyncio
async def test_load_latest_data_none_last_data_timestamp(mongo_db):
"""Test load_latest_data"""
collection = MagicMock()
mongo_db.database.__getitem__.return_value = collection
collection.find.return_value = [
{
'name': 'test1',
'value': 1,
'timestamp': datetime.strptime(
'2023-01-01 12:00:00.000000', '%Y-%m-%d %H:%M:%S.%f')
}
]
result = await mongo_db.load_latest_data({
'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
'collection_name': 'test_collection',
'last_data_timestamp': None,
'base_data_filter': {
'level': 'ERROR'
}
})
mongo_db.database.__getitem__.assert_called_once_with(
'test_collection')
collection.find.assert_called_once_with(
{
'level': 'ERROR'
},
{"_id": 0}
)
assert result == {
'name': {
0: 'test1'
},
'value': {
0: 1
},
'timestamp': {
0: '2023-01-01 12:00:00.000000'
}
}
@mark.asyncio
async def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
"""Test load_latest_data"""
collection = MagicMock()
mongo_db.database.__getitem__.return_value = collection
collection.find.return_value = [
{
'name': 'test1',
'value': 1,
'timestamp': datetime.strptime(
'2023-01-01 12:00:00.000000', '%Y-%m-%d %H:%M:%S.%f')
}
]
result = await mongo_db.load_latest_data({
'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
'collection_name': 'test_collection',
'last_data_timestamp': '2023-01-01 12:00:00.000000',
'base_data_filter': {
'level': 'ERROR'
}
})
mongo_db.database.__getitem__.assert_called_once_with(
'test_collection')
collection.find.assert_called_once_with(
{
'level': 'ERROR',
'timestamp': {
'$gt': datetime.strptime(
'2023-01-01 12:00:00.000000', '%Y-%m-%d %H:%M:%S.%f')
}
},
{"_id": 0}
)
assert result == {
'name': {
0: 'test1'
},
'value': {
0: 1
},
'timestamp': {
0: '2023-01-01 12:00:00.000000'
}
}
@mark.asyncio
async def test_load_latest_data_error(mongo_db):
"""Test load_latest_data"""
collection = MagicMock()
mongo_db.send_notification = MagicMock()
mongo_db.database.__getitem__.return_value = collection
collection.find.side_effect = Exception('test')
try:
await mongo_db.load_latest_data({
'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
'collection_name': 'test_collection',
'last_data_timestamp': '2023-01-01 12:00:00.000000',
'base_data_filter': {
'level': 'ERROR'
}
})
except Exception as e:
assert str(e) == 'test'
mongo_db.send_notification.assert_called_once_with(
metadata={'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule'},
notification_id='MONGO_LOAD_ERROR',
message='Error loading data from MongoDB: test',
block='load_latest_data',
level=NotificationLevel.ERROR,
attachment_content=ANY
)

153
tests/orchestrator/activities/test_slot_manager.py
View File

@@ -1,4 +1,5 @@
from unittest.mock import MagicMock, patch, call, ANY from unittest.mock import MagicMock, patch, call, ANY
from pandas import DataFrame
from pytest import mark, fixture from pytest import mark, fixture
from orchestrator.activities.slot_manager import SlotManager from orchestrator.activities.slot_manager import SlotManager
from sientia_do.notifications.models import NotificationLevel from sientia_do.notifications.models import NotificationLevel
@@ -206,3 +207,155 @@ async def test_delete_slots(slot_manager):
"message": "Test exception" "message": "Test exception"
} }
} }
@mark.asyncio
async def test_get_last_data_timestamp_none(slot_manager):
"""Test get_last_data_timestamp"""
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule'
}
slot_manager.get = MagicMock(return_value=None)
result = await slot_manager.get_last_data_timestamp(test_data)
assert result is None
@mark.asyncio
async def test_get_last_data_timestamp_not_none(slot_manager):
"""Test get_last_data_timestamp"""
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule'
}
slot_manager.get = MagicMock(return_value='2023-01-01 12:00:00')
result = await slot_manager.get_last_data_timestamp(test_data)
slot_manager.get.assert_called_once_with(
'notification_last_timestamp'
)
assert result == '2023-01-01 12:00:00'
@mark.asyncio
async def test_get_last_data_timestamp_error(slot_manager):
"""Test get_last_data_timestamp error"""
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule'
}
slot_manager.send_notification = MagicMock()
slot_manager.get = MagicMock(side_effect=Exception('test'))
try:
await slot_manager.get_last_data_timestamp(test_data)
except Exception as e:
assert str(e) == 'test'
slot_manager.send_notification.assert_called_once_with(
metadata=metadata['metadata'],
notification_id="REDIS_GET_ERROR",
message="Error getting last data timestamp: test",
block="get_last_data_timestamp",
level=NotificationLevel.ERROR,
attachment_content=ANY
)
else:
assert False, "Expected exception"
@mark.asyncio
async def test_put_last_data_timestamp_empty_dataframe(slot_manager):
"""Test put_last_data_timestamp with empty dataframe"""
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule',
'data': DataFrame(columns=['name', 'value', 'timestamp']).to_dict('records')
}
slot_manager.set = MagicMock()
result = await slot_manager.put_last_data_timestamp(test_data)
assert result is None
slot_manager.set.assert_not_called()
@mark.asyncio
async def test_put_last_data_timestamp_not_empty_dataframe(slot_manager):
"""Test put_last_data_timestamp with not empty dataframe"""
data = DataFrame({
'name': ['sensor1', 'sensor2'],
'value': [25.5, 30.0],
'timestamp': ['2023-01-01 12:00:00', '2023-01-01 12:00:01']
})
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule',
'data': data.to_dict('records')
}
slot_manager.set = MagicMock()
result = await slot_manager.put_last_data_timestamp(test_data)
assert result == '2023-01-01 12:00:01'
slot_manager.set.assert_called_once_with(
'notification_last_timestamp',
'2023-01-01 12:00:01',
ttl=18000
)
@mark.asyncio
async def test_put_last_data_timestamp_error(slot_manager):
"""Test put_last_data_timestamp error"""
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule',
'data': DataFrame({
'name': ['sensor1', 'sensor2'],
'value': [25.5, 30.0],
'timestamp': ['2023-01-01 12:00:00'] * 2
}).to_dict('records')
}
slot_manager.send_notification = MagicMock()
slot_manager.set = MagicMock(side_effect=Exception('test'))
try:
await slot_manager.put_last_data_timestamp(test_data)
except Exception as e:
assert str(e) == 'test'
slot_manager.send_notification.assert_called_once_with(
metadata=metadata['metadata'],
notification_id="REDIS_SET_ERROR",
message="Error setting last data timestamp: test",
block="put_last_data_timestamp",
level=NotificationLevel.ERROR,
attachment_content=ANY
)
else:
assert False, "Expected exception"

115
tests/orchestrator/workflows/subworkflows/test_load_notification_package.py Normal file
View File

@@ -0,0 +1,115 @@
from unittest.mock import AsyncMock, patch, ANY, call
from pytest import fixture, mark
from orchestrator.workflows.subworkflows.load_notification_package import LoadNotificationPackage
from orchestrator.activities.activities import Activities
@fixture
def load_notification_package():
return LoadNotificationPackage()
metadata = {
'metadata': {
'schedule_name': 'test-schedule-name',
'workflow_name': 'test-workflow',
'model_name': '-',
'model_id': '-',
}
}
@mark.asyncio
@patch("orchestrator.workflows.subworkflows.load_notification_package.workflow", new_callable=AsyncMock)
async def test_run(workflow_mock, load_notification_package):
input_data = {
'metadata': metadata
}
workflow_mock.start_local_activity_method.side_effect = [
'2023-01-01 12:00:00',
[
{
'id': '1',
}
],
[
{
'id_r': '1',
}
]
]
output = await load_notification_package.run(input_data)
assert output == {
'last_timestamp': '2023-01-01 12:00:00',
'notification_package': [
{
'id': '1',
}
],
'sending_configs': [
{
'id_r': '1',
}
]
}
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.get_last_data_timestamp,
input_data['metadata'],
start_to_close_timeout=ANY,
retry_policy=ANY
)
])
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.find_documents_in_mongodb,
{
**input_data['metadata'],
'query': {
'collection': 'receiver_groups',
'filters': {
'active': True
}
}
},
start_to_close_timeout=ANY,
retry_policy=ANY
)
])
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.load_latest_data,
{
**input_data['metadata'],
'collection_name': 'notification_queue',
'last_data_timestamp': '2023-01-01 12:00:00',
'base_data_filter': {
'level': 'ERROR'
}
},
start_to_close_timeout=ANY,
retry_policy=ANY
)
])
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.put_last_data_timestamp,
{
**input_data['metadata'],
'data': [
{
'id': '1',
}
]
},
start_to_close_timeout=ANY,
retry_policy=ANY
)
])