diff --git a/email.html b/email.html index c75e731..5e90b26 100644 --- a/email.html +++ b/email.html @@ -15,11 +15,251 @@

SIENTIA™ Alerts

-

+

Errors detected:

-

+

Model: ipsum et

+ + + + + + + + + + + + + + + + + + + + + +
Notification IDScheduleBlockTimestampMessage
TAG_node34:tag_8_LISTENNING_STOPPEDadipiscing_eiusmod_dodolor_labore_consectetur_ipsum2025-07-29 14:39:52.952316+00:00sed elit dolore incididunt incididunt aliqua
-

+

Model: incididunt et do

+ + + + + + + + + + + + + + + + + + + + + +
Notification IDScheduleBlockTimestampMessage
OPC_LISTENNING_STOPPED__server_5eiusmod_adipiscing_dolore_ipsum_incididunt_incididuntaliqua_dolore2025-07-29 14:39:52.952579+00:00dolor elit do ipsum consectetur amet ut do amet et
+ +

Model: magna adipiscing aliqua

+ + + + + + + + + + + + + + + + + + + + + +
Notification IDScheduleBlockTimestampMessage
REPORT_PARTITION_MANAGERdo_dolore_ut_ametet_eiusmod2025-07-29 14:39:52.952688+00:00incididunt lorem eiusmod do et ipsum et ut
+ +

Warnings detected:

+ +

Model: adipiscing magna lorem

+ + + + + + + + + + + + + + + + + + + + + +
Notification IDScheduleBlockTimestampMessage
REPORT_PARTITION_MANAGERipsum_et_incididunt_dout_sed_incididunt2025-07-29 14:39:52.952121+00:00aliqua tempor aliqua sit ipsum amet elit ipsum
+ +

Model: sed amet adipiscing incididunt

+ + + + + + + + + + + + + + + + + + + + + +
Notification IDScheduleBlockTimestampMessage
ALIQUA_SIT_INCIDIDUNT_EIUSMODconsectetur_dolore_sit_sed_seddolore_ipsum2025-07-29 14:39:52.952259+00:00do aliqua ut incididunt consectetur consectetur sed et labore
+ +

Model: ipsum ipsum amet tempor

+ + + + + + + + + + + + + + + + + + + + + +
Notification IDScheduleBlockTimestampMessage
OPC_LISTENNING_STOPPED__server_6sit_labore_sed_dolor_et_elitdolor_do_magna_et2025-07-29 14:39:52.952376+00:00magna consectetur do et dolor aliqua
+ +

Model: sed ut

+ + + + + + + + + + + + + + + + + + + + + +
Notification IDScheduleBlockTimestampMessage
REPORT_PARTITION_MANAGERelit_et_ipsum_doloresed_ipsum2025-07-29 14:39:52.952426+00:00elit dolor dolore dolor eiusmod
+ +

Model: eiusmod labore

+ + + + + + + + + + + + + + + + + + + + + +
Notification IDScheduleBlockTimestampMessage
OPC_CONNECTION_RETRY__server_3dolor_ut_dolor_sit_adipiscing_incididuntsit_tempor_dolore2025-07-29 14:39:52.952483+00:00ipsum consectetur magna elit dolore
+ +

Model: do dolor tempor amet

+ + + + + + + + + + + + + + + + + + + + + +
Notification IDScheduleBlockTimestampMessage
REPORT_PARTITION_MANAGERsed_sit_do_tempor_uteiusmod_adipiscing2025-07-29 14:39:52.952637+00:00incididunt do do adipiscing dolore ut ut elit sit labore
+ +

Infos detected:

+ +

Model: sed consectetur ut

+ + + + + + + + + + + + + + + + + + + + + +
Notification IDScheduleBlockTimestampMessage
TEMPOR_DO_DOLOR_TEMPOR_DOLORelit_tempor_dolore_consectetur_dolore_adipiscingmagna_labore_lorem2025-07-29 14:39:52.952199+00:00et et adipiscing lorem magna eiusmod labore do
diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 635c613..17a091b 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -532,6 +532,41 @@ class Formatters(BaseActivity): 'mail_type': mail_type } else: - data[key]['groups'].append(group_name) + if group_name not in data[key]['groups']: + data[key]['groups'].append(group_name) return DataFrame(list(data.values())).to_dict() + + @activity.defn(name="filter_notification_reports") + async def filter_notification_reports(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Filter notification reports. + """ + metadata = input_data['metadata'] + notification_package = input_data['notification_package'] + sending_configs = input_data['sending_configs'] + + self.info("Filtering notification reports...", metadata=metadata) + + receiver_groups = {} + + for receiver_group in sending_configs: + group_name = receiver_group['group_name'] + receiver_groups[group_name] = { + **receiver_group, + "notifications": [] + } + receiver_groups[group_name]['notifications'] = [] + + ignore_list = receiver_group.get('ignore', []) + + for notification in notification_package: + alert_type = "reports" + notification_id = notification['notification_id'] + + # Check if this group must be notified + if alert_type in receiver_group['contents'] and notification_id not in ignore_list: + receiver_groups[group_name]["notifications"].append( + notification) + + return receiver_groups diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index 17dbd07..16cf26f 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -204,7 +204,7 @@ class SlotManager(Redis): Gets the last data timestamp from redis. """ metadata = input_data['metadata'] - key = "notification_last_timestamp" + key = f"notification_last_timestamp:{input_data['mail_type']}" try: data_hold = self.get(key) @@ -235,7 +235,7 @@ class SlotManager(Redis): Puts the last data timestamp into redis. """ metadata = input_data['metadata'] - key = "notification_last_timestamp" + key = f"notification_last_timestamp:{input_data['mail_type']}" data = DataFrame(input_data['data']) diff --git a/orchestrator/worker/worker.py b/orchestrator/worker/worker.py index 21f9e96..fcd815d 100644 --- a/orchestrator/worker/worker.py +++ b/orchestrator/worker/worker.py @@ -7,6 +7,7 @@ with workflow.unsafe.imports_passed_through(): import sys import asyncio from orchestrator.workflows.alerts import Alerts + from orchestrator.workflows.reports import Reports from orchestrator.workflows.subworkflows.load_notification_package import LoadNotificationPackage from orchestrator.workflows.subworkflows.process_notifications import ProcessNotifications from orchestrator.workflows.orchestrator import Orchestrator @@ -121,6 +122,27 @@ async def main(): # Store notification cache activities.store_notification_cache ] + ), + Worker( + temporal_client, + task_queue='reports-queue', + workflows=[Reports, LoadNotificationPackage, ProcessNotifications], + activities=[ + # Load notifications + activities.get_last_data_timestamp, + activities.find_documents_in_mongodb, + activities.load_latest_data, + activities.put_last_data_timestamp, + + # Format and filter notifications + activities.filter_notification_reports, + + # Send email and export data to postgres + activities.build_email_html, + activities.send_email, + activities.format_log_report, + activities.export_data_to_postgres + ] ) ] diff --git a/orchestrator/workflows/alerts.py b/orchestrator/workflows/alerts.py index 441390b..6fbe231 100644 --- a/orchestrator/workflows/alerts.py +++ b/orchestrator/workflows/alerts.py @@ -38,6 +38,7 @@ class Alerts: mail_type = "Alerts" input_data['metadata'] = metadata + input_data['mail_type'] = mail_type input_data['base_data_filter'] = { 'level': 'ERROR' diff --git a/orchestrator/workflows/reports.py b/orchestrator/workflows/reports.py index 002034f..75e2851 100644 --- a/orchestrator/workflows/reports.py +++ b/orchestrator/workflows/reports.py @@ -3,17 +3,76 @@ 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="reports") class Reports: @workflow.run async def run(self, input_data: dict[str, Any]): + """ + Workflow to send reports to the users + + Args: + input_data (dict[str, Any]): Input data. It contains the following keys: + - schedule_name: str - Name of the schedule + + Returns: + None + + Raises: + Exception: If the workflow fails + """ + metadata = { + 'metadata': { + 'schedule_name': input_data['schedule_name'], + 'workflow_name': 'reports', + 'model_name': '-', + 'model_id': '-' + } + } + + mail_type = "Reports" + + input_data['metadata'] = metadata + input_data['mail_type'] = mail_type + + input_data['base_data_filter'] = {} + # Call subworkflow "load_notification_package" passing the static filters # (timestamp > last timestamp) - # Filter notification package by groups custom configs + package = await workflow.execute_child_workflow( + 'load_notification_package', + input_data + ) + + if not package['notification_package'] or not package['sending_configs']: + return + + # Filter notification package by groups custom configs, levels and + # timestamp cached + + receiver_groups = await workflow.execute_local_activity_method( + Activities.filter_notification_reports, + { + **metadata, + 'notification_package': package['notification_package'], + 'sending_configs': package['sending_configs'] + }, + schedule_to_close_timeout=timedelta(seconds=60), + retry_policy=retry_policy + ) # Call subworkflow "process_notifications" passing the notification package - - pass + await workflow.execute_child_workflow( + 'process_notifications', + { + 'metadata': metadata, + 'mail_type': mail_type, + 'notification_package': receiver_groups, + 'schema': 'sientia_data', + 'table_name': 'log_report' + } + ) diff --git a/orchestrator/workflows/subworkflows/load_notification_package.py b/orchestrator/workflows/subworkflows/load_notification_package.py index 83a1c0f..15d3560 100644 --- a/orchestrator/workflows/subworkflows/load_notification_package.py +++ b/orchestrator/workflows/subworkflows/load_notification_package.py @@ -22,6 +22,7 @@ class LoadNotificationPackage: - last_timestamp (str): The last timestamp of the notification package. - notification_package (list[dict]): The notification package. - sending_configs (list[dict]): The sending configs. + - mail_type (str): The mail type. """ metadata = input_data['metadata'] @@ -30,6 +31,7 @@ class LoadNotificationPackage: Activities.get_last_data_timestamp, { **metadata, + 'mail_type': input_data['mail_type'] }, start_to_close_timeout=timedelta(seconds=60), retry_policy=retry_policy @@ -82,7 +84,8 @@ class LoadNotificationPackage: Activities.put_last_data_timestamp, { **metadata, - 'data': notification_package + 'data': notification_package, + 'mail_type': input_data['mail_type'] }, start_to_close_timeout=timedelta(seconds=60), retry_policy=retry_policy diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index 8440e19..ef89e9b 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -765,3 +765,61 @@ async def test_format_log_report(formatters): ) assert DataFrame(result).equals(expected_result) + + +@mark.asyncio +async def test_filter_notification_reports(formatters): + + input_data = { + **metadata, + 'notification_package': [ + { + 'trigger': 'test_trigger_1', + 'notification_id': 'test_notification_id_1' + }, + { + 'trigger': 'test_trigger_2', + 'notification_id': 'test_notification_id_2' + }, + { + 'trigger': 'test_trigger_3', + 'notification_id': 'test_notification_id_3' + } + ], + 'sending_configs': [ + { + 'group_name': 'test_group_1', + 'contents': ['reports'], + 'ignore': ['test_notification_id_1'] + }, + { + 'group_name': 'test_group_2', + 'contents': ['core_alerts'] + } + ] + } + + response = await formatters.filter_notification_reports(input_data) + + assert response == { + 'test_group_1': { + 'group_name': 'test_group_1', + 'contents': ['reports'], + 'ignore': ['test_notification_id_1'], + 'notifications': [ + { + 'trigger': 'test_trigger_2', + 'notification_id': 'test_notification_id_2' + }, + { + 'trigger': 'test_trigger_3', + 'notification_id': 'test_notification_id_3' + } + ] + }, + 'test_group_2': { + 'group_name': 'test_group_2', + 'contents': ['core_alerts'], + 'notifications': [] + } + } diff --git a/tests/orchestrator/workflows/test_reports.py b/tests/orchestrator/workflows/test_reports.py new file mode 100644 index 0000000..1b1983d --- /dev/null +++ b/tests/orchestrator/workflows/test_reports.py @@ -0,0 +1,99 @@ +from unittest.mock import AsyncMock, patch, ANY, call +from pytest import fixture, mark +from orchestrator.workflows.reports import Reports +from orchestrator.activities.activities import Activities + + +@fixture +def reports(): + return Reports() + + +metadata = { + 'metadata': { + 'schedule_name': 'test-schedule-name', + 'workflow_name': 'reports', + 'model_name': '-', + 'model_id': '-', + } +} + + +@mark.asyncio +@patch("orchestrator.workflows.reports.workflow", new_callable=AsyncMock) +async def test_run_full_flow(workflow_mock, reports): + input_data = { + 'schedule_name': 'test-schedule-name', + 'notification_ttl': 300, + 'sent_ttl': 600 + } + + await reports.run(input_data) + + workflow_mock.execute_child_workflow.assert_has_calls([ + call( + 'load_notification_package', + { + **input_data, + 'metadata': metadata, + 'base_data_filter': {} + } + ) + ]) + + workflow_mock.execute_child_workflow.assert_has_calls([ + call( + 'process_notifications', + { + 'metadata': metadata, + 'mail_type': 'Reports', + 'notification_package': workflow_mock.execute_local_activity_method.return_value, + 'schema': 'sientia_data', + 'table_name': 'log_report' + } + ) + ]) + + workflow_mock.execute_local_activity_method.assert_has_calls([ + call( + Activities.filter_notification_reports, + { + **metadata, + 'notification_package': workflow_mock.execute_child_workflow.return_value['notification_package'], + 'sending_configs': workflow_mock.execute_child_workflow.return_value['sending_configs'] + }, + schedule_to_close_timeout=ANY, + retry_policy=ANY + ) + ]) + + +@mark.asyncio +@patch("orchestrator.workflows.reports.workflow", new_callable=AsyncMock) +async def test_run_no_data(workflow_mock, reports): + workflow_mock.execute_child_workflow.return_value = { + 'last_timestamp': '2023-01-01 12:00:00.000000', + 'notification_package': [], + 'sending_configs': [] + } + + input_data = { + 'schedule_name': 'test-schedule-name', + 'notification_ttl': 300, + 'sent_ttl': 600 + } + + await reports.run(input_data) + + workflow_mock.execute_child_workflow.assert_has_calls([ + call( + 'load_notification_package', + { + **input_data, + 'metadata': metadata, + 'base_data_filter': {} + } + ) + ]) + + workflow_mock.execute_local_activity_method.assert_not_called()