diff --git a/email.html b/email.html new file mode 100644 index 0000000..5e90b26 --- /dev/null +++ b/email.html @@ -0,0 +1,267 @@ + + + + + + SIENTIA™ Report + + + +

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
+ + + + + \ No newline at end of file diff --git a/notification_generator.ipynb b/notification_generator.ipynb new file mode 100644 index 0000000..8d24f92 --- /dev/null +++ b/notification_generator.ipynb @@ -0,0 +1,416 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": 58, + "id": "1786cbf0", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "Generated 10 notifications\n" + ] + } + ], + "source": [ + "from unittest.mock import MagicMock\n", + "from orchestrator.activities.mongo_db import MongoDB\n", + "from orchestrator.utils.connectors_config import build_mongodb_config\n", + "import random\n", + "from datetime import datetime, timezone\n", + "import json\n", + "\n", + "word_list = [\n", + " 'lorem', 'ipsum', 'dolor', 'sit', 'amet', 'consectetur',\n", + " 'adipiscing', 'elit', 'sed', 'do', 'eiusmod', 'tempor',\n", + " 'incididunt', 'ut', 'labore', 'et', 'dolore', 'magna', 'aliqua'\n", + " ]\n", + "\n", + "config = build_mongodb_config()\n", + "\n", + "mongo_db = MongoDB(\n", + " connection_string=config['connection_string'],\n", + " database_name=config['database_name'],\n", + " ttl_index_seconds=config['ttl_index_seconds'],\n", + " logger=MagicMock(),\n", + " notification_handler=MagicMock()\n", + ")\n", + "\n", + "AMOUT = 10\n", + "\n", + "notification_ids = [\n", + " \"TAG_node:_LISTENNING_STOPPED\",\n", + " \"OPC___\",\n", + " \"REPORT_PARTITION_MANAGER\",\n", + " \"\"\n", + "]\n", + "\n", + "notifications = []\n", + "\n", + "for _ in range(AMOUT):\n", + " # Choose random notification ID and fill in placeholders\n", + " notification_id = random.choice(notification_ids)\n", + " \n", + " if \"\" in notification_id:\n", + " notification_id = notification_id.replace(\"\", str(random.randint(1,100)))\n", + " if \"\" in notification_id:\n", + " notification_id = notification_id.replace(\"\", f\"tag_{random.randint(1,100)}\")\n", + " if \"\" in notification_id:\n", + " events = [\"LISTENNING_STOPPED\", \"CONNECTION_RETRY\", \"LISTENNING_BACK\", \"SUBSCRIPTION\", \"QUEUE_RETRIEVE\"]\n", + " notification_id = notification_id.replace(\"\", random.choice(events))\n", + " if \"\" in notification_id:\n", + " notification_id = notification_id.replace(\"\", f\"server_{random.randint(1,10)}\")\n", + " if \"\" in notification_id:\n", + " notification_id = notification_id.replace(\"\", \"_\".join(random.choices(word_list, k=random.randint(3,5))).upper())\n", + "\n", + " notification = {\n", + " \"project\": \"sientia-laborious\",\n", + " \"pipeline\": \"_\".join(random.choices(word_list, k=random.randint(2,4))), \n", + " \"trigger\": \"_\".join(random.choices(word_list, k=random.randint(3,6))),\n", + " \"model_name\": \" \".join(random.choices(word_list, k=random.randint(2,4))),\n", + " \"model_id\": str(random.randint(1,100)),\n", + " \"block\": \"_\".join(random.choices(word_list, k=random.randint(2,4))),\n", + " \"level\": random.choice([\"INFO\", \"WARNING\", \"ERROR\"]),\n", + " \"message\": \" \".join(random.choices(word_list, k=random.randint(5,10))),\n", + " \"attachment_content\": \" \".join(random.choices(word_list, k=random.randint(20,30))),\n", + " \"notification_id\": notification_id,\n", + " \"timestamp\": datetime.now(timezone.utc)\n", + " }\n", + " \n", + " notifications.append(notification)\n", + "\n", + "print(f\"Generated {len(notifications)} notifications\")" + ] + }, + { + "cell_type": "code", + "execution_count": 59, + "id": "8a67c90d", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "{'WARNING': {'section_name': 'Warnings detected:', 'models': [{'model_name': 'amet sed consectetur', 'events': [{'project': 'sientia-laborious', 'pipeline': 'elit_adipiscing_magna', 'trigger': 'eiusmod_do_incididunt_dolor_aliqua', 'model_name': 'amet sed consectetur', 'model_id': '8', 'block': 'do_amet_sit', 'level': 'WARNING', 'message': 'aliqua lorem elit magna eiusmod', 'attachment_content': 'sit ut consectetur dolore ut ipsum eiusmod tempor incididunt sit dolore elit aliqua amet amet aliqua magna dolore ut amet lorem aliqua dolore eiusmod', 'notification_id': 'OPC_CONNECTION_RETRY__server_9', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121908, tzinfo=datetime.timezone.utc)}]}, {'model_name': 'sit dolor lorem', 'events': [{'project': 'sientia-laborious', 'pipeline': 'consectetur_adipiscing_incididunt', 'trigger': 'labore_ut_amet_dolor', 'model_name': 'sit dolor lorem', 'model_id': '44', 'block': 'adipiscing_amet_aliqua_eiusmod', 'level': 'WARNING', 'message': 'consectetur elit incididunt dolore magna ut sit sit adipiscing', 'attachment_content': 'lorem ipsum elit eiusmod do lorem et do do eiusmod eiusmod dolore eiusmod lorem consectetur amet ipsum eiusmod dolor dolore aliqua adipiscing dolore magna ut', 'notification_id': 'OPC_LISTENNING_BACK__server_5', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121982, tzinfo=datetime.timezone.utc)}]}, {'model_name': 'ut labore ipsum ut', 'events': [{'project': 'sientia-laborious', 'pipeline': 'lorem_elit', 'trigger': 'magna_sed_tempor_labore', 'model_name': 'ut labore ipsum ut', 'model_id': '99', 'block': 'sed_dolor_do', 'level': 'WARNING', 'message': 'magna magna incididunt ut incididunt incididunt consectetur', 'attachment_content': 'sed lorem ut amet labore incididunt tempor incididunt do lorem aliqua dolore adipiscing labore labore tempor sit eiusmod tempor aliqua dolor ut sed incididunt adipiscing', 'notification_id': 'OPC_QUEUE_RETRIEVE__server_2', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 122025, tzinfo=datetime.timezone.utc)}]}]}, 'ERROR': {'section_name': 'Errors detected:', 'models': [{'model_name': 'ut elit et eiusmod', 'events': [{'project': 'sientia-laborious', 'pipeline': 'do_magna_aliqua_aliqua', 'trigger': 'dolore_sit_sed', 'model_name': 'ut elit et eiusmod', 'model_id': '20', 'block': 'do_magna', 'level': 'ERROR', 'message': 'dolor aliqua labore labore eiusmod do', 'attachment_content': 'eiusmod sed magna amet tempor labore magna adipiscing incididunt tempor ipsum ipsum aliqua magna tempor do tempor consectetur amet aliqua lorem sed lorem ut consectetur', 'notification_id': 'REPORT_PARTITION_MANAGER', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121924, tzinfo=datetime.timezone.utc)}]}, {'model_name': 'dolore aliqua eiusmod eiusmod', 'events': [{'project': 'sientia-laborious', 'pipeline': 'aliqua_dolor_elit', 'trigger': 'consectetur_incididunt_adipiscing', 'model_name': 'dolore aliqua eiusmod eiusmod', 'model_id': '97', 'block': 'consectetur_magna_do', 'level': 'ERROR', 'message': 'lorem eiusmod et aliqua aliqua sit', 'attachment_content': 'amet lorem labore amet eiusmod do do sit adipiscing ut adipiscing sed sed do ipsum tempor incididunt dolor sed et ipsum amet incididunt sed', 'notification_id': 'REPORT_PARTITION_MANAGER', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121953, tzinfo=datetime.timezone.utc)}]}, {'model_name': 'incididunt eiusmod amet eiusmod', 'events': [{'project': 'sientia-laborious', 'pipeline': 'ipsum_et', 'trigger': 'elit_amet_lorem_dolore_elit', 'model_name': 'incididunt eiusmod amet eiusmod', 'model_id': '31', 'block': 'amet_tempor_tempor', 'level': 'ERROR', 'message': 'ut dolor consectetur amet elit do elit', 'attachment_content': 'amet ut adipiscing lorem adipiscing ut et amet adipiscing incididunt elit magna et aliqua incididunt lorem incididunt lorem incididunt eiusmod sit amet tempor lorem', 'notification_id': 'ELIT_ALIQUA_TEMPOR', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 122040, tzinfo=datetime.timezone.utc)}]}]}, 'INFO': {'section_name': 'Infos detected:', 'models': [{'model_name': 'do magna', 'events': [{'project': 'sientia-laborious', 'pipeline': 'tempor_amet', 'trigger': 'incididunt_adipiscing_incididunt_eiusmod_incididunt_tempor', 'model_name': 'do magna', 'model_id': '29', 'block': 'incididunt_ipsum', 'level': 'INFO', 'message': 'sit labore consectetur labore magna labore eiusmod ipsum', 'attachment_content': 'ipsum amet et ipsum lorem consectetur incididunt dolore consectetur et elit eiusmod elit do eiusmod amet consectetur adipiscing adipiscing lorem eiusmod magna dolor eiusmod sit eiusmod lorem amet magna consectetur', 'notification_id': 'OPC_SUBSCRIPTION__server_4', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121940, tzinfo=datetime.timezone.utc)}]}, {'model_name': 'amet adipiscing ut', 'events': [{'project': 'sientia-laborious', 'pipeline': 'aliqua_dolore_adipiscing_adipiscing', 'trigger': 'ut_incididunt_ut_et_tempor_incididunt', 'model_name': 'amet adipiscing ut', 'model_id': '28', 'block': 'do_sed', 'level': 'INFO', 'message': 'consectetur elit amet dolor incididunt dolor dolor labore adipiscing', 'attachment_content': 'do incididunt aliqua tempor dolor sed lorem elit consectetur et tempor labore elit sed elit dolor sed amet ipsum do eiusmod aliqua adipiscing et sit', 'notification_id': 'TAG_node56:tag_30_LISTENNING_STOPPED', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121968, tzinfo=datetime.timezone.utc)}]}, {'model_name': 'sed lorem et lorem', 'events': [{'project': 'sientia-laborious', 'pipeline': 'do_ipsum_consectetur_ut', 'trigger': 'dolore_incididunt_consectetur_et', 'model_name': 'sed lorem et lorem', 'model_id': '82', 'block': 'tempor_incididunt', 'level': 'INFO', 'message': 'lorem tempor eiusmod dolore tempor aliqua amet sed', 'attachment_content': 'sed dolore amet ut eiusmod aliqua sed dolore incididunt consectetur adipiscing et elit magna consectetur ipsum ipsum ipsum dolor eiusmod do', 'notification_id': 'REPORT_PARTITION_MANAGER', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121994, tzinfo=datetime.timezone.utc)}]}, {'model_name': 'eiusmod elit incididunt', 'events': [{'project': 'sientia-laborious', 'pipeline': 'sit_dolore', 'trigger': 'eiusmod_sed_amet_ut', 'model_name': 'eiusmod elit incididunt', 'model_id': '85', 'block': 'amet_incididunt', 'level': 'INFO', 'message': 'do consectetur incididunt magna dolore et et elit incididunt', 'attachment_content': 'tempor eiusmod sit elit consectetur ut consectetur eiusmod labore labore consectetur ut eiusmod sit dolor ipsum sit do et amet lorem lorem et do ut magna dolor', 'notification_id': 'REPORT_PARTITION_MANAGER', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 122010, tzinfo=datetime.timezone.utc)}]}]}}\n" + ] + } + ], + "source": [ + "from orchestrator.utils.email_builder import EmailBuilder\n", + "from unittest.mock import MagicMock\n", + "\n", + "\n", + "email_builder = EmailBuilder(logger=MagicMock())\n", + "\n", + "\n", + "general_events = {}\n", + "\n", + "for report in notifications:\n", + "\n", + " level = report['level']\n", + " model_name = report['model_name']\n", + "\n", + " if level not in general_events:\n", + " general_events[level] = {\n", + " 'section_name': f'{level.capitalize()}s detected:',\n", + " 'models': {}\n", + " }\n", + "\n", + " if model_name not in general_events[level]['models']:\n", + " general_events[level]['models'][model_name] = {\n", + " 'model_name': model_name,\n", + " 'events': []\n", + " }\n", + "\n", + " general_events[level]['models'][model_name]['events'].append(\n", + " report)\n", + "\n", + "for _type, content in general_events.items():\n", + " content['models'] = list(content['models'].values())\n", + "\n", + "print(general_events)" + ] + }, + { + "cell_type": "code", + "execution_count": 60, + "id": "6944697a", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "[{'model_name': 'ut elit et eiusmod', 'events': [{'project': 'sientia-laborious', 'pipeline': 'do_magna_aliqua_aliqua', 'trigger': 'dolore_sit_sed', 'model_name': 'ut elit et eiusmod', 'model_id': '20', 'block': 'do_magna', 'level': 'ERROR', 'message': 'dolor aliqua labore labore eiusmod do', 'attachment_content': 'eiusmod sed magna amet tempor labore magna adipiscing incididunt tempor ipsum ipsum aliqua magna tempor do tempor consectetur amet aliqua lorem sed lorem ut consectetur', 'notification_id': 'REPORT_PARTITION_MANAGER', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121924, tzinfo=datetime.timezone.utc)}]}, {'model_name': 'dolore aliqua eiusmod eiusmod', 'events': [{'project': 'sientia-laborious', 'pipeline': 'aliqua_dolor_elit', 'trigger': 'consectetur_incididunt_adipiscing', 'model_name': 'dolore aliqua eiusmod eiusmod', 'model_id': '97', 'block': 'consectetur_magna_do', 'level': 'ERROR', 'message': 'lorem eiusmod et aliqua aliqua sit', 'attachment_content': 'amet lorem labore amet eiusmod do do sit adipiscing ut adipiscing sed sed do ipsum tempor incididunt dolor sed et ipsum amet incididunt sed', 'notification_id': 'REPORT_PARTITION_MANAGER', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121953, tzinfo=datetime.timezone.utc)}]}, {'model_name': 'incididunt eiusmod amet eiusmod', 'events': [{'project': 'sientia-laborious', 'pipeline': 'ipsum_et', 'trigger': 'elit_amet_lorem_dolore_elit', 'model_name': 'incididunt eiusmod amet eiusmod', 'model_id': '31', 'block': 'amet_tempor_tempor', 'level': 'ERROR', 'message': 'ut dolor consectetur amet elit do elit', 'attachment_content': 'amet ut adipiscing lorem adipiscing ut et amet adipiscing incididunt elit magna et aliqua incididunt lorem incididunt lorem incididunt eiusmod sit amet tempor lorem', 'notification_id': 'ELIT_ALIQUA_TEMPOR', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 122040, tzinfo=datetime.timezone.utc)}]}]\n", + "[{'model_name': 'amet sed consectetur', 'events': [{'project': 'sientia-laborious', 'pipeline': 'elit_adipiscing_magna', 'trigger': 'eiusmod_do_incididunt_dolor_aliqua', 'model_name': 'amet sed consectetur', 'model_id': '8', 'block': 'do_amet_sit', 'level': 'WARNING', 'message': 'aliqua lorem elit magna eiusmod', 'attachment_content': 'sit ut consectetur dolore ut ipsum eiusmod tempor incididunt sit dolore elit aliqua amet amet aliqua magna dolore ut amet lorem aliqua dolore eiusmod', 'notification_id': 'OPC_CONNECTION_RETRY__server_9', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121908, tzinfo=datetime.timezone.utc)}]}, {'model_name': 'sit dolor lorem', 'events': [{'project': 'sientia-laborious', 'pipeline': 'consectetur_adipiscing_incididunt', 'trigger': 'labore_ut_amet_dolor', 'model_name': 'sit dolor lorem', 'model_id': '44', 'block': 'adipiscing_amet_aliqua_eiusmod', 'level': 'WARNING', 'message': 'consectetur elit incididunt dolore magna ut sit sit adipiscing', 'attachment_content': 'lorem ipsum elit eiusmod do lorem et do do eiusmod eiusmod dolore eiusmod lorem consectetur amet ipsum eiusmod dolor dolore aliqua adipiscing dolore magna ut', 'notification_id': 'OPC_LISTENNING_BACK__server_5', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121982, tzinfo=datetime.timezone.utc)}]}, {'model_name': 'ut labore ipsum ut', 'events': [{'project': 'sientia-laborious', 'pipeline': 'lorem_elit', 'trigger': 'magna_sed_tempor_labore', 'model_name': 'ut labore ipsum ut', 'model_id': '99', 'block': 'sed_dolor_do', 'level': 'WARNING', 'message': 'magna magna incididunt ut incididunt incididunt consectetur', 'attachment_content': 'sed lorem ut amet labore incididunt tempor incididunt do lorem aliqua dolore adipiscing labore labore tempor sit eiusmod tempor aliqua dolor ut sed incididunt adipiscing', 'notification_id': 'OPC_QUEUE_RETRIEVE__server_2', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 122025, tzinfo=datetime.timezone.utc)}]}]\n", + "[{'model_name': 'do magna', 'events': [{'project': 'sientia-laborious', 'pipeline': 'tempor_amet', 'trigger': 'incididunt_adipiscing_incididunt_eiusmod_incididunt_tempor', 'model_name': 'do magna', 'model_id': '29', 'block': 'incididunt_ipsum', 'level': 'INFO', 'message': 'sit labore consectetur labore magna labore eiusmod ipsum', 'attachment_content': 'ipsum amet et ipsum lorem consectetur incididunt dolore consectetur et elit eiusmod elit do eiusmod amet consectetur adipiscing adipiscing lorem eiusmod magna dolor eiusmod sit eiusmod lorem amet magna consectetur', 'notification_id': 'OPC_SUBSCRIPTION__server_4', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121940, tzinfo=datetime.timezone.utc)}]}, {'model_name': 'amet adipiscing ut', 'events': [{'project': 'sientia-laborious', 'pipeline': 'aliqua_dolore_adipiscing_adipiscing', 'trigger': 'ut_incididunt_ut_et_tempor_incididunt', 'model_name': 'amet adipiscing ut', 'model_id': '28', 'block': 'do_sed', 'level': 'INFO', 'message': 'consectetur elit amet dolor incididunt dolor dolor labore adipiscing', 'attachment_content': 'do incididunt aliqua tempor dolor sed lorem elit consectetur et tempor labore elit sed elit dolor sed amet ipsum do eiusmod aliqua adipiscing et sit', 'notification_id': 'TAG_node56:tag_30_LISTENNING_STOPPED', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121968, tzinfo=datetime.timezone.utc)}]}, {'model_name': 'sed lorem et lorem', 'events': [{'project': 'sientia-laborious', 'pipeline': 'do_ipsum_consectetur_ut', 'trigger': 'dolore_incididunt_consectetur_et', 'model_name': 'sed lorem et lorem', 'model_id': '82', 'block': 'tempor_incididunt', 'level': 'INFO', 'message': 'lorem tempor eiusmod dolore tempor aliqua amet sed', 'attachment_content': 'sed dolore amet ut eiusmod aliqua sed dolore incididunt consectetur adipiscing et elit magna consectetur ipsum ipsum ipsum dolor eiusmod do', 'notification_id': 'REPORT_PARTITION_MANAGER', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121994, tzinfo=datetime.timezone.utc)}]}, {'model_name': 'eiusmod elit incididunt', 'events': [{'project': 'sientia-laborious', 'pipeline': 'sit_dolore', 'trigger': 'eiusmod_sed_amet_ut', 'model_name': 'eiusmod elit incididunt', 'model_id': '85', 'block': 'amet_incididunt', 'level': 'INFO', 'message': 'do consectetur incididunt magna dolore et et elit incididunt', 'attachment_content': 'tempor eiusmod sit elit consectetur ut consectetur eiusmod labore labore consectetur ut eiusmod sit dolor ipsum sit do et amet lorem lorem et do ut magna dolor', 'notification_id': 'REPORT_PARTITION_MANAGER', 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 122010, tzinfo=datetime.timezone.utc)}]}]\n" + ] + } + ], + "source": [ + "error_models = general_events.get('ERROR', {}).get('models', [])\n", + "warning_models = general_events.get('WARNING', {}).get('models', [])\n", + "info_models = general_events.get('INFO', {}).get('models', [])\n", + "mail_type = 'Alerts'\n", + "\n", + "print(error_models)\n", + "print(warning_models)\n", + "print(info_models)" + ] + }, + { + "cell_type": "code", + "execution_count": 61, + "id": "02e7dabf", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "'

Errors detected:

\\n\\n

Model: ut elit et eiusmod

\\n\\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n
Notification IDScheduleBlockTimestampMessage
REPORT_PARTITION_MANAGERdolore_sit_seddo_magna2025-07-29 12:21:45.121924+00:00dolor aliqua labore labore eiusmod do
\\n\\n

Model: dolore aliqua eiusmod eiusmod

\\n\\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n
Notification IDScheduleBlockTimestampMessage
REPORT_PARTITION_MANAGERconsectetur_incididunt_adipiscingconsectetur_magna_do2025-07-29 12:21:45.121953+00:00lorem eiusmod et aliqua aliqua sit
\\n\\n

Model: incididunt eiusmod amet eiusmod

\\n\\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n \\n
Notification IDScheduleBlockTimestampMessage
ELIT_ALIQUA_TEMPORelit_amet_lorem_dolore_elitamet_tempor_tempor2025-07-29 12:21:45.122040+00:00ut dolor consectetur amet elit do elit
\\n'" + ] + }, + "execution_count": 61, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "email_builder.replace_parameters(email_builder.general_template,\n", + " general_events.get('ERROR', {})) if general_events else ''" + ] + }, + { + "cell_type": "code", + "execution_count": 62, + "id": "bebde0ea", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "{'mail_type': 'Alerts',\n", + " 'error_events': '

\\n',\n", + " 'warning_events': '

\\n',\n", + " 'info_events': '

\\n'}" + ] + }, + "execution_count": 62, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "\n", + "\n", + "\n", + "\n", + "d = {\n", + " 'mail_type': mail_type,\n", + " 'error_events': email_builder.replace_parameters(email_builder.general_template,\n", + " error_models) if error_models else '',\n", + " 'warning_events': email_builder.replace_parameters(email_builder.general_template,\n", + " warning_models) if warning_models else '',\n", + " 'info_events': email_builder.replace_parameters(email_builder.general_template,\n", + " info_models) if info_models else '',\n", + "}\n", + "\n", + "d" + ] + }, + { + "cell_type": "code", + "execution_count": 63, + "id": "c1467b72", + "metadata": {}, + "outputs": [], + "source": [ + "from orchestrator.utils.email_builder import EmailBuilder\n", + "from unittest.mock import MagicMock\n", + "\n", + "\n", + "email_builder = EmailBuilder(logger=MagicMock())\n", + "\n", + "html = email_builder.build_email(notifications, \"Alerts\")\n", + "\n", + "with open('email.html', 'w') as file:\n", + " file.write(html)" + ] + }, + { + "cell_type": "code", + "execution_count": 64, + "id": "ffd7be77", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "[{'project': 'sientia-laborious',\n", + " 'pipeline': 'elit_adipiscing_magna',\n", + " 'trigger': 'eiusmod_do_incididunt_dolor_aliqua',\n", + " 'model_name': 'amet sed consectetur',\n", + " 'model_id': '8',\n", + " 'block': 'do_amet_sit',\n", + " 'level': 'WARNING',\n", + " 'message': 'aliqua lorem elit magna eiusmod',\n", + " 'attachment_content': 'sit ut consectetur dolore ut ipsum eiusmod tempor incididunt sit dolore elit aliqua amet amet aliqua magna dolore ut amet lorem aliqua dolore eiusmod',\n", + " 'notification_id': 'OPC_CONNECTION_RETRY__server_9',\n", + " 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121908, tzinfo=datetime.timezone.utc)},\n", + " {'project': 'sientia-laborious',\n", + " 'pipeline': 'do_magna_aliqua_aliqua',\n", + " 'trigger': 'dolore_sit_sed',\n", + " 'model_name': 'ut elit et eiusmod',\n", + " 'model_id': '20',\n", + " 'block': 'do_magna',\n", + " 'level': 'ERROR',\n", + " 'message': 'dolor aliqua labore labore eiusmod do',\n", + " 'attachment_content': 'eiusmod sed magna amet tempor labore magna adipiscing incididunt tempor ipsum ipsum aliqua magna tempor do tempor consectetur amet aliqua lorem sed lorem ut consectetur',\n", + " 'notification_id': 'REPORT_PARTITION_MANAGER',\n", + " 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121924, tzinfo=datetime.timezone.utc)},\n", + " {'project': 'sientia-laborious',\n", + " 'pipeline': 'tempor_amet',\n", + " 'trigger': 'incididunt_adipiscing_incididunt_eiusmod_incididunt_tempor',\n", + " 'model_name': 'do magna',\n", + " 'model_id': '29',\n", + " 'block': 'incididunt_ipsum',\n", + " 'level': 'INFO',\n", + " 'message': 'sit labore consectetur labore magna labore eiusmod ipsum',\n", + " 'attachment_content': 'ipsum amet et ipsum lorem consectetur incididunt dolore consectetur et elit eiusmod elit do eiusmod amet consectetur adipiscing adipiscing lorem eiusmod magna dolor eiusmod sit eiusmod lorem amet magna consectetur',\n", + " 'notification_id': 'OPC_SUBSCRIPTION__server_4',\n", + " 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121940, tzinfo=datetime.timezone.utc)},\n", + " {'project': 'sientia-laborious',\n", + " 'pipeline': 'aliqua_dolor_elit',\n", + " 'trigger': 'consectetur_incididunt_adipiscing',\n", + " 'model_name': 'dolore aliqua eiusmod eiusmod',\n", + " 'model_id': '97',\n", + " 'block': 'consectetur_magna_do',\n", + " 'level': 'ERROR',\n", + " 'message': 'lorem eiusmod et aliqua aliqua sit',\n", + " 'attachment_content': 'amet lorem labore amet eiusmod do do sit adipiscing ut adipiscing sed sed do ipsum tempor incididunt dolor sed et ipsum amet incididunt sed',\n", + " 'notification_id': 'REPORT_PARTITION_MANAGER',\n", + " 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121953, tzinfo=datetime.timezone.utc)},\n", + " {'project': 'sientia-laborious',\n", + " 'pipeline': 'aliqua_dolore_adipiscing_adipiscing',\n", + " 'trigger': 'ut_incididunt_ut_et_tempor_incididunt',\n", + " 'model_name': 'amet adipiscing ut',\n", + " 'model_id': '28',\n", + " 'block': 'do_sed',\n", + " 'level': 'INFO',\n", + " 'message': 'consectetur elit amet dolor incididunt dolor dolor labore adipiscing',\n", + " 'attachment_content': 'do incididunt aliqua tempor dolor sed lorem elit consectetur et tempor labore elit sed elit dolor sed amet ipsum do eiusmod aliqua adipiscing et sit',\n", + " 'notification_id': 'TAG_node56:tag_30_LISTENNING_STOPPED',\n", + " 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121968, tzinfo=datetime.timezone.utc)},\n", + " {'project': 'sientia-laborious',\n", + " 'pipeline': 'consectetur_adipiscing_incididunt',\n", + " 'trigger': 'labore_ut_amet_dolor',\n", + " 'model_name': 'sit dolor lorem',\n", + " 'model_id': '44',\n", + " 'block': 'adipiscing_amet_aliqua_eiusmod',\n", + " 'level': 'WARNING',\n", + " 'message': 'consectetur elit incididunt dolore magna ut sit sit adipiscing',\n", + " 'attachment_content': 'lorem ipsum elit eiusmod do lorem et do do eiusmod eiusmod dolore eiusmod lorem consectetur amet ipsum eiusmod dolor dolore aliqua adipiscing dolore magna ut',\n", + " 'notification_id': 'OPC_LISTENNING_BACK__server_5',\n", + " 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121982, tzinfo=datetime.timezone.utc)},\n", + " {'project': 'sientia-laborious',\n", + " 'pipeline': 'do_ipsum_consectetur_ut',\n", + " 'trigger': 'dolore_incididunt_consectetur_et',\n", + " 'model_name': 'sed lorem et lorem',\n", + " 'model_id': '82',\n", + " 'block': 'tempor_incididunt',\n", + " 'level': 'INFO',\n", + " 'message': 'lorem tempor eiusmod dolore tempor aliqua amet sed',\n", + " 'attachment_content': 'sed dolore amet ut eiusmod aliqua sed dolore incididunt consectetur adipiscing et elit magna consectetur ipsum ipsum ipsum dolor eiusmod do',\n", + " 'notification_id': 'REPORT_PARTITION_MANAGER',\n", + " 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 121994, tzinfo=datetime.timezone.utc)},\n", + " {'project': 'sientia-laborious',\n", + " 'pipeline': 'sit_dolore',\n", + " 'trigger': 'eiusmod_sed_amet_ut',\n", + " 'model_name': 'eiusmod elit incididunt',\n", + " 'model_id': '85',\n", + " 'block': 'amet_incididunt',\n", + " 'level': 'INFO',\n", + " 'message': 'do consectetur incididunt magna dolore et et elit incididunt',\n", + " 'attachment_content': 'tempor eiusmod sit elit consectetur ut consectetur eiusmod labore labore consectetur ut eiusmod sit dolor ipsum sit do et amet lorem lorem et do ut magna dolor',\n", + " 'notification_id': 'REPORT_PARTITION_MANAGER',\n", + " 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 122010, tzinfo=datetime.timezone.utc)},\n", + " {'project': 'sientia-laborious',\n", + " 'pipeline': 'lorem_elit',\n", + " 'trigger': 'magna_sed_tempor_labore',\n", + " 'model_name': 'ut labore ipsum ut',\n", + " 'model_id': '99',\n", + " 'block': 'sed_dolor_do',\n", + " 'level': 'WARNING',\n", + " 'message': 'magna magna incididunt ut incididunt incididunt consectetur',\n", + " 'attachment_content': 'sed lorem ut amet labore incididunt tempor incididunt do lorem aliqua dolore adipiscing labore labore tempor sit eiusmod tempor aliqua dolor ut sed incididunt adipiscing',\n", + " 'notification_id': 'OPC_QUEUE_RETRIEVE__server_2',\n", + " 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 122025, tzinfo=datetime.timezone.utc)},\n", + " {'project': 'sientia-laborious',\n", + " 'pipeline': 'ipsum_et',\n", + " 'trigger': 'elit_amet_lorem_dolore_elit',\n", + " 'model_name': 'incididunt eiusmod amet eiusmod',\n", + " 'model_id': '31',\n", + " 'block': 'amet_tempor_tempor',\n", + " 'level': 'ERROR',\n", + " 'message': 'ut dolor consectetur amet elit do elit',\n", + " 'attachment_content': 'amet ut adipiscing lorem adipiscing ut et amet adipiscing incididunt elit magna et aliqua incididunt lorem incididunt lorem incididunt eiusmod sit amet tempor lorem',\n", + " 'notification_id': 'ELIT_ALIQUA_TEMPOR',\n", + " 'timestamp': datetime.datetime(2025, 7, 29, 12, 21, 45, 122040, tzinfo=datetime.timezone.utc)}]" + ] + }, + "execution_count": 64, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "notifications" + ] + }, + { + "cell_type": "code", + "execution_count": 65, + "id": "55aed216", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "InsertManyResult([ObjectId('6888bcd90ccc3300e4a6c2e5'), ObjectId('6888bcd90ccc3300e4a6c2e6'), ObjectId('6888bcd90ccc3300e4a6c2e7'), ObjectId('6888bcd90ccc3300e4a6c2e8'), ObjectId('6888bcd90ccc3300e4a6c2e9'), ObjectId('6888bcd90ccc3300e4a6c2ea'), ObjectId('6888bcd90ccc3300e4a6c2eb'), ObjectId('6888bcd90ccc3300e4a6c2ec'), ObjectId('6888bcd90ccc3300e4a6c2ed'), ObjectId('6888bcd90ccc3300e4a6c2ee')], acknowledged=True)" + ] + }, + "execution_count": 65, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "mongo_db.database['notification_queue'].insert_many(notifications)" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "venv", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.11.13" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/orchestrator/activities/activities.py b/orchestrator/activities/activities.py index 6c769cc..2bf0388 100644 --- a/orchestrator/activities/activities.py +++ b/orchestrator/activities/activities.py @@ -1,26 +1,29 @@ -from temporalio import activity, workflow -from temporalio.client import Client +from temporalio import workflow -from orchestrator.activities.mongo_db import MongoDB with workflow.unsafe.imports_passed_through(): - from orchestrator.activities.couchbase import Couchbase from orchestrator.activities.temporal_manager import TemporalManager from orchestrator.activities.slot_manager import SlotManager from orchestrator.activities.formatters import Formatters + from orchestrator.activities.email import Email + from orchestrator.activities.mongo_db import MongoDB from typing import Any from logging import Logger + from sientia_do.temporal.activities.postgres import Postgres from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler class Activities( # Couchbase, - TemporalManager, SlotManager, Formatters, MongoDB): + TemporalManager, SlotManager, Formatters, MongoDB, Email, + Postgres): def __init__(self, temporal_config: dict[str, Any], # couchbase_config: dict[str, Any], redis_config: dict[str, Any], mongodb_config: dict[str, Any], + email_config: dict[str, Any], + postgres_config: dict[str, Any], logger: Logger, notification_handler: NotificationHandler): @@ -59,5 +62,24 @@ class Activities( # Couchbase, logger=logger, notification_handler=notification_handler) + Email.__init__(self, + sender_email=email_config['sender_email'], + sender_password=email_config['sender_password'], + smtp_server=email_config['smtp_server'], + smtp_port=email_config['smtp_port'], + logger=logger, + notification_handler=notification_handler) + + Postgres.__init__(self, + host=postgres_config['host'], + port=postgres_config['port'], + user=postgres_config['user'], + password=postgres_config['password'], + dbname=postgres_config['dbname'], + min_connections=postgres_config['min_connections'], + max_connections=postgres_config['max_connections'], + logger=logger, + notification_handler=notification_handler) + def shutdown(self): MongoDB.shutdown(self) diff --git a/orchestrator/activities/email.py b/orchestrator/activities/email.py new file mode 100644 index 0000000..b59ff91 --- /dev/null +++ b/orchestrator/activities/email.py @@ -0,0 +1,187 @@ + +from smtplib import SMTPServerDisconnected +from temporalio import workflow, activity + +with workflow.unsafe.imports_passed_through(): + import traceback + 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 + from email.mime.base import MIMEBase + from email import encoders + + +class Email(BaseActivity): + def __init__(self, sender_email: str, sender_password: str, + smtp_server: str, smtp_port: int, + logger: Logger, notification_handler: NotificationHandler): + + self.email_builder = EmailBuilder(logger=logger) + + self.sender_email = sender_email + self.sender_password = sender_password + self.smtp_port = smtp_port + self.smtp_server = smtp_server + + logger.info(f"Initializing Email with {smtp_server}:{smtp_port}") + + 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) + + 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]) -> dict[str, Any]: + """ + 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 'filename' and 'content' 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}") + + raise e + + return msg + + def try_send_email(self, msg: MIMEMultipart, receivers: str): + """ + Sends an email to the receivers. + """ + + try: + self.server.sendmail( + self.sender_email, receivers, msg.as_string()) + except SMTPServerDisconnected as e: + self.logger.error(f"SMTP server disconnected: {e}") + self.logger.info( + f"Reconnecting to {self.smtp_server}:{self.smtp_port}") + + if self.server: + try: + self.server.quit() + except SMTPServerDisconnected as e: + self.logger.info(f"Server already disconnected: {e}") + except Exception as e: + self.logger.error(f"Failed to quit server: {e}") + raise e + + self.server = smtplib.SMTP( + self.smtp_server, self.smtp_port, timeout=20) + if self.sender_password: + self.server.starttls() + self.server.login(self.sender_email, self.sender_password) + self.server.sendmail(self.sender_email, receivers, msg.as_string()) + + @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(): + try: + + 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( + [ + { + "filename": f"{notification['trigger']}_{notification['notification_id']}.txt", + "attachment_content": notification['attachment_content'] + } + for notification in group_config['notifications'] + if notification.get('attachment_content') is not None], + msg) + + self.try_send_email(msg, receivers) + 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}: {receivers}", + metadata=metadata) + + self.info(f"Email sent for {mail_type} mail type.", + metadata=metadata) + + return receiver_groups diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 5078b12..1ebcc6b 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -1,4 +1,5 @@ +from pandas import DataFrame from temporalio import activity, workflow from orchestrator.utils.orchestrator_functions import minimal_retrain @@ -488,3 +489,89 @@ class Formatters(BaseActivity): notification_id="REPORT_ORCHESTRATION_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(): + + for notification in group_config['notifications']: + + notification_id = notification['notification_id'] + trigger = notification['trigger'] + + key = f"{notification_id}:{trigger}" + + if key not in data: + data[key] = { + 'status': group_config['status'], + 'timestamp': notification['timestamp'], + 'groups': [group_name], + 'message': notification['message'], + 'level': notification['level'], + 'notification_id': notification_id, + 'block': notification['block'], + 'schedule': trigger, + 'pipeline': notification['pipeline'], + 'project': notification['project'], + 'model_name': notification['model_name'], + 'model_id': notification['model_id'], + 'mail_type': mail_type + } + else: + 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'] = [] + + already_added_keys = [] + + ignore_list = receiver_group.get('ignore', []) + + for notification in notification_package: + alert_type = "reports" + notification_id = notification['notification_id'] + + key = f"{notification['trigger']}:{notification_id}" + + # Check if this group must be notified + if alert_type in receiver_group['contents'] and notification_id not in ignore_list and key not in already_added_keys: + receiver_groups[group_name]["notifications"].append( + notification) + already_added_keys.append(key) + + return receiver_groups diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 515a9b4..105264e 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -1,3 +1,4 @@ +from pandas import DataFrame from temporalio import workflow, activity @@ -80,6 +81,15 @@ class MongoDB(BaseActivity): """ 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",) 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) try: - collection = self.database[collection_name] - - documents = list(collection.find(filters, {"_id": 0})) - - documents = clear_mongo_id(documents) + documents = self.find(collection_name, filters) self.info( f"Loaded {len(documents)} documents from collection '{collection_name}'", metadata=metadata) @@ -352,3 +358,76 @@ class MongoDB(BaseActivity): ) self.error(trace, metadata=metadata) 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 diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index 404c8d8..076a1a8 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -1,3 +1,4 @@ + from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): @@ -8,6 +9,9 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.temporal.activities.redis_base import Redis from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel + from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT + from pandas import DataFrame + from datetime import datetime, timedelta class SlotManager(Redis): @@ -193,3 +197,146 @@ class SlotManager(Redis): f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}", metadata=metadata) 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 = f"notification_last_timestamp:{input_data['mail_type']}" + + 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 = f"notification_last_timestamp:{input_data['mail_type']}" + + 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 + + @activity.defn(name="filter_notification_alerts") + async def filter_notification_alerts(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Filter notification alerts + """ + metadata = input_data['metadata'] + notification_package = input_data['notification_package'] + sending_configs = input_data['sending_configs'] + notification_ttl = input_data['notification_ttl'] + + self.info("Filtering notification alerts...", 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'] = [] + + already_added_keys = [] + + ignore_list = receiver_group.get('ignore', []) + + for notification in notification_package: + alert_type = "do_nothing" + notification_id = notification['notification_id'] + # Check if notification was recently sent + key = f"{notification['trigger']}:{notification_id}" + + last_sent = self.get(key) + + if last_sent is None: + alert_type = "core_alerts" + + else: + last_sent = datetime.strptime( + last_sent, DEFAULT_DATE_FORMAT) + + # Check if "notification_ttl" seconds has passed since last sent + if (datetime.now() - last_sent) > timedelta(seconds=notification_ttl): + alert_type = "persistent_alerts" + + # Check if this group must be notified + if alert_type in receiver_group['contents'] and notification_id not in ignore_list and key not in already_added_keys: + receiver_groups[group_name]["notifications"].append( + notification) + already_added_keys.append(key) + + return receiver_groups + + @activity.defn(name="store_notification_cache") + async def store_notification_cache(self, input_data: dict[str, Any]) -> None: + """ + Store notification cache + """ + metadata = input_data['metadata'] + log_report = DataFrame(input_data['log_report']) + sent_ttl = input_data['sent_ttl'] + + self.info("Storing notification cache...", metadata=metadata) + + now = datetime.now().strftime(DEFAULT_DATE_FORMAT) + + for index, row in log_report.iterrows(): + status = row['status'] + if status == 'sent': + key = f"{row['schedule']}:{row['notification_id']}" + self.set(key, now, ttl=sent_ttl) + + self.info("Notification cache stored...", metadata=metadata) diff --git a/orchestrator/utils/connectors_config.py b/orchestrator/utils/connectors_config.py index 519731a..7237a05 100644 --- a/orchestrator/utils/connectors_config.py +++ b/orchestrator/utils/connectors_config.py @@ -38,3 +38,24 @@ def build_temporal_config(): 'temporal_scouter_namespace': getenv('TEMPORAL_SCOUTER_NAMESPACE', 'scouter'), 'temporal_laborious_namespace': getenv('TEMPORAL_LABORIOUS_NAMESPACE', 'laborious') } + + +def build_postgres_config(): + return { + 'host': getenv('POSTGRES_HOST', 'localhost'), + 'port': int(getenv('POSTGRES_PORT', '5432')), + 'user': getenv('POSTGRES_USER', 'sientia'), + 'password': getenv('POSTGRES_PASSWORD', 'sientia'), + 'dbname': getenv('POSTGRES_DBNAME', 'sientia'), + 'min_connections': int(getenv('POSTGRES_MIN_CONNECTIONS', '5')), + 'max_connections': int(getenv('POSTGRES_MAX_CONNECTIONS', '20')) + } + + +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_port': int(getenv('EMAIL_SMTP_PORT', '587')) + } diff --git a/orchestrator/utils/email_builder.py b/orchestrator/utils/email_builder.py new file mode 100644 index 0000000..88a765f --- /dev/null +++ b/orchestrator/utils/email_builder.py @@ -0,0 +1,76 @@ +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' + + 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, general_events: dict, mail_type: str) -> dict: + error_models = general_events.get('ERROR', {}).get('models', []) + warning_models = general_events.get('WARNING', {}).get('models', []) + info_models = general_events.get('INFO', {}).get('models', []) + + return { + 'mail_type': mail_type, + 'error_events': self.replace_parameters(self.general_template, + general_events.get( + 'ERROR')) if error_models else '', + 'warning_events': self.replace_parameters(self.general_template, + general_events.get( + 'WARNING')) if warning_models else '', + 'info_events': self.replace_parameters(self.general_template, + general_events.get( + 'INFO')) if 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) + + for _type, content in general_events.items(): + content['models'] = list(content['models'].values()) + + return self.replace_parameters( + self.report_template, self.parameters( + general_events, mail_type + )) diff --git a/orchestrator/utils/templates/email_template.html b/orchestrator/utils/templates/email_template.html new file mode 100644 index 0000000..e269e53 --- /dev/null +++ b/orchestrator/utils/templates/email_template.html @@ -0,0 +1,24 @@ + + + + + + SIENTIA™ Report + + + +

SIENTIA™ {{ mail_type }}

+ + {{ error_events }} + {{ warning_events }} + {{ info_events }} + + {{ special_events }} + + diff --git a/orchestrator/utils/templates/general_template.html b/orchestrator/utils/templates/general_template.html new file mode 100644 index 0000000..0ae39fa --- /dev/null +++ b/orchestrator/utils/templates/general_template.html @@ -0,0 +1,26 @@ +

{{ section_name }}

+{% for model in models %} +

Model: {{ model.model_name }}

+ + + + + + + + + + + + {% for event in model.events %} + + + + + + + + {% endfor %} + +
Notification IDScheduleBlockTimestampMessage
{{ event.notification_id }}{{ event.trigger }}{{ event.block }}{{ event.timestamp }}{{ event.message }}
+{% endfor %} \ No newline at end of file diff --git a/orchestrator/worker/worker.py b/orchestrator/worker/worker.py index 525a376..fcd815d 100644 --- a/orchestrator/worker/worker.py +++ b/orchestrator/worker/worker.py @@ -1,18 +1,24 @@ from temporalio import workflow, client from temporalio.worker import Worker -from orchestrator.utils.connectors_config import build_temporal_config with workflow.unsafe.imports_passed_through(): import os 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 from orchestrator.activities.activities import Activities from orchestrator.utils.connectors_config import ( # build_couchbase_config, build_redis_config, - build_mongodb_config + build_mongodb_config, + build_temporal_config, + build_email_config, + build_postgres_config ) from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.temporal.utils.logger import get_logger @@ -48,6 +54,8 @@ async def main(): temporal_config=build_temporal_config(), redis_config=build_redis_config(), mongodb_config=build_mongodb_config(), + email_config=build_email_config(), + postgres_config=build_postgres_config(), logger=logger, notification_handler=notification_handler ) @@ -90,6 +98,51 @@ async def main(): activities.report_slot_orchestration, activities.format_schedule_config, ] + ), + Worker( + temporal_client, + task_queue='alerts-queue', + workflows=[Alerts, 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_alerts, + + # Send email and export data to postgres + activities.build_email_html, + activities.send_email, + activities.format_log_report, + activities.export_data_to_postgres, + + # 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 new file mode 100644 index 0000000..6fbe231 --- /dev/null +++ b/orchestrator/workflows/alerts.py @@ -0,0 +1,95 @@ +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="alerts") +class Alerts: + @workflow.run + async def run(self, input_data: dict[str, Any]): + """ + Workflow to send alerts to the users + + 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 + + Returns: + None + + Raises: + Exception: If the workflow fails + """ + metadata = { + 'metadata': { + 'schedule_name': input_data['schedule_name'], + 'workflow_name': 'alerts', + 'model_name': '-', + 'model_id': '-' + } + } + + mail_type = "Alerts" + + input_data['metadata'] = metadata + input_data['mail_type'] = mail_type + + input_data['base_data_filter'] = { + 'level': 'ERROR' + } + + # Call subworkflow "load_notification_package" passing the static filters + # (level = "ERROR" and timestamp > last timestamp) + + 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_alerts, + { + **metadata, + 'notification_package': package['notification_package'], + 'sending_configs': package['sending_configs'], + 'notification_ttl': input_data['notification_ttl'] + }, + schedule_to_close_timeout=timedelta(seconds=60), + retry_policy=retry_policy + ) + + # Call subworkflow "process_notifications" passing the notification package + log_report = await workflow.execute_child_workflow( + 'process_notifications', + { + 'metadata': metadata, + 'mail_type': mail_type, + 'notification_package': receiver_groups, + 'schema': 'sientia_data', + 'table_name': 'log_report' + } + ) + + # Store the notification_id sendings to avoid sending them again + await workflow.execute_activity_method( + Activities.store_notification_cache, + { + **metadata, + 'log_report': log_report, + 'sent_ttl': input_data['sent_ttl'] + }, + schedule_to_close_timeout=timedelta(seconds=60), + retry_policy=retry_policy + ) diff --git a/orchestrator/workflows/orchestrator.py b/orchestrator/workflows/orchestrator.py index f61587d..cb81d13 100644 --- a/orchestrator/workflows/orchestrator.py +++ b/orchestrator/workflows/orchestrator.py @@ -11,6 +11,15 @@ with workflow.unsafe.imports_passed_through(): class Orchestrator: @workflow.run 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' diff --git a/orchestrator/workflows/reports.py b/orchestrator/workflows/reports.py new file mode 100644 index 0000000..75e2851 --- /dev/null +++ b/orchestrator/workflows/reports.py @@ -0,0 +1,78 @@ +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) + + 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 + 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 new file mode 100644 index 0000000..15d3560 --- /dev/null +++ b/orchestrator/workflows/subworkflows/load_notification_package.py @@ -0,0 +1,102 @@ +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. + - mail_type (str): The mail type. + """ + 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, + 'mail_type': input_data['mail_type'] + }, + 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': input_data['base_data_filter'] + }, + start_to_close_timeout=timedelta(seconds=60), + retry_policy=retry_policy + ) + sending_configs = await sending_configs_handler + + if not sending_configs or not notification_package: + return { + 'last_timestamp': last_timestamp, + 'notification_package': notification_package, + 'sending_configs': sending_configs + } + + # Put last collected timestamp in redis "notification_last_timestamp" + + await workflow.start_activity_method( + Activities.put_last_data_timestamp, + { + **metadata, + 'data': notification_package, + 'mail_type': input_data['mail_type'] + }, + 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': sending_configs + } diff --git a/orchestrator/workflows/subworkflows/process_notifications.py b/orchestrator/workflows/subworkflows/process_notifications.py new file mode 100644 index 0000000..f2f36c8 --- /dev/null +++ b/orchestrator/workflows/subworkflows/process_notifications.py @@ -0,0 +1,90 @@ +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"], + "mail_type": input_data["mail_type"] + }, + 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"] + }, + schedule_to_close_timeout=timedelta(seconds=60), + retry_policy=retry_policy + ) + + # 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 diff --git a/requirements.txt b/requirements.txt index 23d31f9..166f96d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,4 +4,5 @@ sqlalchemy redis couchbase pymongo -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.4 +jinja2 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.5 diff --git a/tests/orchestrator/activities/test_activities.py b/tests/orchestrator/activities/test_activities.py index 1d76bae..d2640d1 100644 --- a/tests/orchestrator/activities/test_activities.py +++ b/tests/orchestrator/activities/test_activities.py @@ -12,7 +12,12 @@ from orchestrator.activities.formatters import Formatters @patch('orchestrator.activities.temporal_manager.TemporalManager.__init__') @patch('orchestrator.activities.slot_manager.SlotManager.__init__') @patch('orchestrator.activities.formatters.Formatters.__init__') -def test___init__(mock_formatters_init, mock_slot_manager_init, +@patch('orchestrator.activities.email.Email.__init__') +@patch('sientia_do.temporal.activities.postgres.Postgres.__init__') +def test___init__(mock_postgres_init, + mock_email_init, + mock_formatters_init, + mock_slot_manager_init, mock_temporal_manager_init, mock_mongodb_init): @@ -35,6 +40,23 @@ def test___init__(mock_formatters_init, mock_slot_manager_init, 'temporal_laborious_namespace': 'laborious' } + email_config = { + 'sender_email': 'test@test.com', + 'sender_password': 'test', + 'smtp_server': 'test', + 'smtp_port': 587 + } + + postgres_config = { + 'host': 'localhost', + 'port': 5432, + 'user': 'admin', + 'password': 'password', + 'dbname': 'test_db', + 'min_connections': 1, + 'max_connections': 10, + } + logger = MagicMock() notification_handler = MagicMock() @@ -42,6 +64,8 @@ def test___init__(mock_formatters_init, mock_slot_manager_init, temporal_config=temporal_config, redis_config=redis_config, mongodb_config=mongo_db_config, + email_config=email_config, + postgres_config=postgres_config, logger=logger, notification_handler=notification_handler ) diff --git a/tests/orchestrator/activities/test_email.py b/tests/orchestrator/activities/test_email.py new file mode 100644 index 0000000..1c82145 --- /dev/null +++ b/tests/orchestrator/activities/test_email.py @@ -0,0 +1,359 @@ +from smtplib import SMTPServerDisconnected +from unittest.mock import MagicMock, call, patch +from pytest import mark, fixture + +from orchestrator.activities.email import Email + + +@fixture +@patch('orchestrator.activities.email.EmailBuilder') +@patch('orchestrator.activities.email.smtplib') +def email(smtplib, email_builder): + email = Email( + sender_email="test@test.com", + sender_password="test", + smtp_server="test", + smtp_port=587, + logger=MagicMock(), + notification_handler=MagicMock() + ) + email_builder.send_notification = MagicMock() + + return email + + +@patch('orchestrator.activities.email.EmailBuilder') +@patch('orchestrator.activities.email.smtplib') +def test___init___with_password(smtplib, email_builder): + email = Email( + sender_email="test@test.com", + sender_password="test", + smtp_server="test", + smtp_port=587, + logger=MagicMock(), + notification_handler=MagicMock() + ) + + assert email.sender_email == "test@test.com" + assert email.sender_password == "test" + assert email.smtp_port == 587 + + smtplib.SMTP.assert_called_once_with("test", 587, timeout=20) + smtplib.SMTP.return_value.starttls.assert_called_once() + smtplib.SMTP.return_value.login.assert_called_once_with( + "test@test.com", "test") + + assert email.server == smtplib.SMTP.return_value + + +@patch('orchestrator.activities.email.EmailBuilder') +@patch('orchestrator.activities.email.smtplib') +def test___init___without_password(smtplib, email_builder): + email = Email( + sender_email="test@test.com", + sender_password=None, + smtp_server="test", + smtp_port=587, + logger=MagicMock(), + notification_handler=MagicMock() + ) + + assert email.sender_email == "test@test.com" + assert email.sender_password is None + assert email.smtp_port == 587 + + smtplib.SMTP.assert_called_once_with("test", 587, timeout=20) + assert email.server == smtplib.SMTP.return_value + + +metadata = { + "metadata": { + "schedule_name": "test", + "model_name": "test", + "model_id": "test", + "workflow_name": "test", + } +} + + +@mark.asyncio +async def test_build_email_html(email): + email.email_builder.build_email = MagicMock( + return_value="test" + ) + input_data = { + **metadata, + "receiver_groups": { + "group_1": { + "notifications": [ + { + "type": "test", + "subject": "test", + "body": "test" + }, + { + "type": "test", + "subject": "test", + "body": "test" + } + ] + } + }, + "mail_type": "test" + } + + response = await email.build_email_html(input_data) + + assert response == { + "group_1": { + "notifications": [ + { + "type": "test", + "subject": "test", + "body": "test", + }, + { + "type": "test", + "subject": "test", + "body": "test", + } + ], + "html": "test" + } + } + + email.email_builder.build_email.assert_called_once_with( + input_data['receiver_groups']['group_1']['notifications'], + input_data['mail_type'] + ) + + +@patch('orchestrator.activities.email.MIMEBase') +@patch('orchestrator.activities.email.encoders') +def test_handle_attachments_success(encoders, mime_base, email): + message = MagicMock() + + attachments = [ + { + "filename": "file_1", + "attachment_content": "test_content_1" + }, + { + "filename": "file_2", + "attachment_content": "test_content_2" + }, + { + "filename": "file_3", + "attachment_content": "test_content_3" + } + ] + + response = email.handle_attachments(attachments, message) + + assert response == message + + mime_base.assert_called_with('application', 'octet-stream') + assert mime_base.return_value.set_payload.call_count == 3 + + mime_base.return_value.set_payload.assert_has_calls( + [ + call("test_content_1".encode('utf-8')), + call("test_content_2".encode('utf-8')), + call("test_content_3".encode('utf-8')) + ] + ) + + encoders.encode_base64.assert_called_with(mime_base.return_value) + assert encoders.encode_base64.call_count == 3 + + mime_base.return_value.add_header.assert_has_calls( + [ + call('Content-Disposition', 'attachment; filename="file_1"'), + call('Content-Disposition', 'attachment; filename="file_2"'), + call('Content-Disposition', 'attachment; filename="file_3"') + ] + ) + + message.attach.assert_called_with(mime_base.return_value) + assert message.attach.call_count == 3 + + +@patch('orchestrator.activities.email.MIMEBase') +def test_handle_attachments_failure(mime_base, email): + message = MagicMock() + + mime_base.side_effect = Exception("test") + + attachments = [ + { + "filename": "file_1", + "attachment_content": "test_content_1" + } + ] + + try: + email.handle_attachments(attachments, message) + except Exception as e: + assert str(e) == "test" + + assert message.attach.call_count == 0 + + +def test_try_send_email_success(email): + email.server.sendmail = MagicMock() + + msg = MagicMock() + + email.try_send_email(msg, "test") + + email.server.sendmail.assert_called_once_with( + "test@test.com", "test", msg.as_string.return_value) + + +@patch('orchestrator.activities.email.smtplib.SMTP') +def test_try_send_email_reconnect_quit_success(smtp, email): + email.server.sendmail = MagicMock( + side_effect=SMTPServerDisconnected("test") + ) + email.server.quit = MagicMock() + + msg = MagicMock() + + email.try_send_email(msg, "test") + + smtp.assert_has_calls([ + call("test", 587, timeout=20) + ]) + + smtp.return_value.starttls.assert_called_once() + smtp.return_value.login.assert_called_once_with( + "test@test.com", "test") + + smtp.return_value.sendmail.assert_called_once_with( + "test@test.com", "test", msg.as_string.return_value) + + +@patch('orchestrator.activities.email.smtplib.SMTP') +def test_try_send_email_reconnect_quit_failure_disconnect(smtp, email): + email.server.sendmail = MagicMock( + side_effect=SMTPServerDisconnected("test") + ) + email.server.quit = MagicMock( + side_effect=SMTPServerDisconnected("test") + ) + + msg = MagicMock() + + email.try_send_email(msg, "test") + + smtp.assert_has_calls([ + call("test", 587, timeout=20) + ]) + smtp.return_value.starttls.assert_called_once() + smtp.return_value.login.assert_called_once_with( + "test@test.com", "test") + + smtp.return_value.sendmail.assert_called_once_with( + "test@test.com", "test", msg.as_string.return_value) + + +@patch('orchestrator.activities.email.smtplib.SMTP') +def test_try_send_email_reconnect_quit_failure(smtp, email): + email.server.sendmail = MagicMock( + side_effect=SMTPServerDisconnected("test") + ) + email.server.quit = MagicMock( + side_effect=Exception("test") + ) + + msg = MagicMock() + + try: + email.try_send_email(msg, "test") + except Exception as e: + assert str(e) == "test" + else: + assert False, "Expected exception" + + +@mark.asyncio +@patch('orchestrator.activities.email.MIMEText') +@patch('orchestrator.activities.email.MIMEMultipart') +async def test_send_email(mimemultipart, mimetext, email): + side_effect_1 = MagicMock() + side_effect_2 = MagicMock() + mimemultipart.side_effect = [ + side_effect_1, + side_effect_2 + ] + + email.email_builder.send_notification = MagicMock() + + email.try_send_email = MagicMock( + side_effect=[ + None, + Exception("test") + ] + ) + + input_data = { + **metadata, + "receiver_groups": { + "group_1": { + "members": ["test@test.com", "test2@test.com"], + "notifications": [ + { + "attachment_content": "test_content_1", + "trigger": "test_trigger", + "notification_id": "test_notification_id" + } + ], + "html": "test_html1" + }, + "group_2": { + "members": ["test3@test.com", "test4@test.com"], + "notifications": [], + "html": "test_html2" + } + }, + "mail_type": "test_TYPE" + } + + response = await email.send_email(input_data) + + assert response['group_1']['status'] == 'sent' + assert response['group_2']['status'] == 'failed' + + assert mimemultipart.call_count == 2 + + mimetext.assert_has_calls( + [ + call("test_html1", "html"), + call("test_html2", "html") + ] + ) + + side_effect_1.__setitem__.assert_has_calls( + [ + call('From', 'test@test.com'), + call('To', 'test@test.com, test2@test.com'), + call('Subject', 'SIENTIA™ test_TYPE') + ] + ) + + side_effect_2.__setitem__.assert_has_calls( + [ + call('From', 'test@test.com'), + call('To', 'test3@test.com, test4@test.com'), + call('Subject', 'SIENTIA™ test_TYPE') + ] + ) + + email.try_send_email.assert_has_calls( + [ + call(side_effect_1, 'test@test.com, test2@test.com'), + call(side_effect_2, 'test3@test.com, test4@test.com') + ] + ) + + assert email.try_send_email.call_count == 2 diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index 04ada97..ef89e9b 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -1,5 +1,6 @@ from unittest.mock import MagicMock, patch, call, ANY import json +from pandas import DataFrame from pytest import fixture, mark from sientia_do.notifications.models import NotificationLevel from orchestrator.activities.formatters import Formatters @@ -696,3 +697,129 @@ async def test_report_slot_orchestration(formatters): attachment=input_data['deleted_slots'] ) ]) + + +@mark.asyncio +async def test_format_log_report(formatters): + input_data = { + **metadata, + "receiver_groups": { + "test_receiver_group": { + "notifications": [ + { + "notification_id": "test_notification_id", + "trigger": "test_trigger", + "timestamp": "2021-01-01", + "message": "test_message", + "level": "test_level", + "block": "test_block", + "pipeline": "test_pipeline", + "project": "test_project", + "model_name": "test_model_name", + "model_id": "test_model_id" + } + ], + "status": "sent", + }, + "test_receiver_group2": { + "notifications": [ + { + "notification_id": "test_notification_id", + "trigger": "test_trigger", + "timestamp": "2021-01-01", + "message": "test_message", + "level": "test_level", + "block": "test_block", + "pipeline": "test_pipeline", + "project": "test_project", + "model_name": "test_model_name", + "model_id": "test_model_id" + } + ], + "status": "sent", + } + }, + "mail_type": "test_mail_type" + } + + result = await formatters.format_log_report(input_data) + + expected_result = DataFrame( + [ + { + "status": "sent", + "timestamp": "2021-01-01", + "groups": ["test_receiver_group", "test_receiver_group2"], + "message": "test_message", + "level": "test_level", + "notification_id": "test_notification_id", + "block": "test_block", + "schedule": "test_trigger", + "pipeline": "test_pipeline", + "project": "test_project", + "model_name": "test_model_name", + "model_id": "test_model_id", + "mail_type": "test_mail_type" + } + ] + ) + + 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/activities/test_mongo_db.py b/tests/orchestrator/activities/test_mongo_db.py index 707b866..e3fa686 100644 --- a/tests/orchestrator/activities/test_mongo_db.py +++ b/tests/orchestrator/activities/test_mongo_db.py @@ -1,4 +1,5 @@ from curses import meta +from datetime import datetime from unittest.mock import MagicMock, patch, ANY from pytest import fixture, mark from orchestrator.activities.mongo_db import clear_mongo_id @@ -518,3 +519,121 @@ async def test_create_collection_with_ttl_index_failure(mongo_db): ) else: 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': 'test1', + 'value': 1, + 'timestamp': '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': 'test1', + 'value': 1, + 'timestamp': '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 + ) diff --git a/tests/orchestrator/activities/test_slot_manager.py b/tests/orchestrator/activities/test_slot_manager.py index 9be7686..debf0a7 100644 --- a/tests/orchestrator/activities/test_slot_manager.py +++ b/tests/orchestrator/activities/test_slot_manager.py @@ -1,7 +1,10 @@ from unittest.mock import MagicMock, patch, call, ANY +from datetime import datetime, timedelta +from pandas import DataFrame from pytest import mark, fixture from orchestrator.activities.slot_manager import SlotManager from sientia_do.notifications.models import NotificationLevel +from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT metadata = { "metadata": { @@ -206,3 +209,257 @@ async def test_delete_slots(slot_manager): "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', + 'mail_type': 'test_mail_type' + } + + 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', + 'mail_type': 'test_mail_type' + } + + 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:test_mail_type' + ) + + 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', + 'mail_type': 'test_mail_type' + } + + 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'), + 'mail_type': 'test_mail_type' + } + + 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'), + 'mail_type': 'test_mail_type' + } + + 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:test_mail_type', + '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'), + 'mail_type': 'test_mail_type' + } + + 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" + + +@mark.asyncio +async def test_filter_notification_alerts(slot_manager): + slot_manager.get = MagicMock(side_effect=[ + None, + (datetime.now() - timedelta(seconds=600) + ).strftime(DEFAULT_DATE_FORMAT), + datetime.now().strftime(DEFAULT_DATE_FORMAT), + None, + (datetime.now() - timedelta(seconds=600) + ).strftime(DEFAULT_DATE_FORMAT), + datetime.now().strftime(DEFAULT_DATE_FORMAT)]) + + 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': ['core_alerts', 'persistent_alerts'] + }, + { + 'group_name': 'test_group_2', + 'contents': ['core_alerts'] + } + ], + 'notification_ttl': 300, + 'mail_type': 'test_mail_type' + } + + response = await slot_manager.filter_notification_alerts(input_data) + + assert response == { + 'test_group_1': { + 'group_name': 'test_group_1', + 'contents': ['core_alerts', 'persistent_alerts'], + 'notifications': [ + { + 'trigger': 'test_trigger_1', + 'notification_id': 'test_notification_id_1' + }, + { + 'trigger': 'test_trigger_2', + 'notification_id': 'test_notification_id_2' + } + ] + }, + 'test_group_2': { + 'group_name': 'test_group_2', + 'contents': ['core_alerts'], + 'notifications': [ + { + 'trigger': 'test_trigger_1', + 'notification_id': 'test_notification_id_1' + } + ] + } + } + + +@mark.asyncio +async def test_store_notification_cache(slot_manager): + """Test store_notification_cache""" + test_data = { + **metadata, + 'log_report': DataFrame({ + 'status': ['sent', 'error'], + 'schedule': ['test_schedule_1', 'test_schedule_2'], + 'notification_id': ['test_notification_id_1', 'test_notification_id_2'] + }).to_dict(), + 'sent_ttl': 600 + } + + slot_manager.set = MagicMock() + + await slot_manager.store_notification_cache(test_data) + + slot_manager.set.assert_called_once_with( + "test_schedule_1:test_notification_id_1", + ANY, + ttl=600 + ) diff --git a/tests/orchestrator/utils/test_connectors_config.py b/tests/orchestrator/utils/test_connectors_config.py index a1c219e..f1a62b9 100644 --- a/tests/orchestrator/utils/test_connectors_config.py +++ b/tests/orchestrator/utils/test_connectors_config.py @@ -1,7 +1,10 @@ from os import environ from orchestrator.utils.connectors_config import (build_redis_config, build_couchbase_config, - build_mongodb_config, build_temporal_config) + build_mongodb_config, + build_temporal_config, + build_email_config, + build_postgres_config) def test_build_redis_config_with_env_vars(): @@ -100,3 +103,69 @@ def test_build_temporal_config_with_defaults(): 'temporal_scouter_namespace': 'scouter', 'temporal_laborious_namespace': 'laborious' } + + +def test_build_email_config_with_env_vars(): + environ['EMAIL_SENDER'] = 'test@test.com' + environ['EMAIL_SENDER_PASSWORD'] = 'test' + environ['EMAIL_SMTP_SERVER'] = 'test' + environ['EMAIL_SMTP_PORT'] = '587' + assert build_email_config() == { + 'sender_email': 'test@test.com', + 'sender_password': 'test', + 'smtp_server': 'test', + 'smtp_port': 587 + } + + +def test_build_email_config_with_defaults(): + environ.pop('EMAIL_SENDER', None) + environ.pop('EMAIL_SENDER_PASSWORD', None) + environ.pop('EMAIL_SMTP_SERVER', None) + environ.pop('EMAIL_SMTP_PORT', None) + + assert build_email_config() == { + 'sender_email': 'sientia-alerts@aignosi.com', + 'sender_password': 'sientia', + 'smtp_server': 'smtp.gmail.com', + 'smtp_port': 587 + } + + +def test_build_postgres_config_with_env_vars(): + environ['POSTGRES_HOST'] = 'localhost' + environ['POSTGRES_PORT'] = '5432' + environ['POSTGRES_USER'] = 'sientia' + environ['POSTGRES_PASSWORD'] = 'sientia' + environ['POSTGRES_DBNAME'] = 'sientia' + environ['POSTGRES_MIN_CONNECTIONS'] = '5' + environ['POSTGRES_MAX_CONNECTIONS'] = '20' + assert build_postgres_config() == { + 'host': 'localhost', + 'port': 5432, + 'user': 'sientia', + 'password': 'sientia', + 'dbname': 'sientia', + 'min_connections': 5, + 'max_connections': 20 + } + + +def test_build_postgres_config_with_defaults(): + environ.pop('POSTGRES_HOST', None) + environ.pop('POSTGRES_PORT', None) + environ.pop('POSTGRES_USER', None) + environ.pop('POSTGRES_PASSWORD', None) + environ.pop('POSTGRES_DBNAME', None) + environ.pop('POSTGRES_MIN_CONNECTIONS', None) + environ.pop('POSTGRES_MAX_CONNECTIONS', None) + + assert build_postgres_config() == { + 'host': 'localhost', + 'port': 5432, + 'user': 'sientia', + 'password': 'sientia', + 'dbname': 'sientia', + 'min_connections': 5, + 'max_connections': 20 + } diff --git a/tests/orchestrator/utils/test_email_builder.py b/tests/orchestrator/utils/test_email_builder.py new file mode 100644 index 0000000..d2145be --- /dev/null +++ b/tests/orchestrator/utils/test_email_builder.py @@ -0,0 +1,145 @@ +import json +from unittest.mock import MagicMock, patch +from pytest import fixture + +from orchestrator.utils.email_builder import EmailBuilder + + +@fixture +@patch('orchestrator.utils.email_builder.open') +def report_builder(open): + return EmailBuilder(MagicMock()) + + +@patch('orchestrator.utils.email_builder.Template') +def test_replace_parameters(template, report_builder): + output = report_builder.replace_parameters('template', {'key': 'value'}) + + assert output == template.return_value.render.return_value + + template.assert_called_once_with('template') + template.return_value.render.assert_called_once_with({'key': 'value'}) + + +def test_parameters(report_builder): + report_builder.replace_parameters = MagicMock() + general_events = { + 'ERROR': { + 'models': [ + {'model_name': 'model_name', 'events': [ + {'notification_id': 'ID_1', 'level': 'ERROR', 'project': 'project'}]}, + ] + }, + 'WARNING': { + 'models': [ + {'model_name': 'model_name', 'events': [ + {'notification_id': 'ID_2', 'level': 'WARNING', 'project': 'project'}]}, + ] + }, + 'INFO': { + 'models': [ + {'model_name': 'model_name', 'events': [ + {'notification_id': 'ID_3', 'level': 'INFO', 'project': 'project'}]}, + ] + } + } + + output = report_builder.parameters(general_events, 'model_name') + + assert output == { + 'mail_type': 'model_name', + 'error_events': report_builder.replace_parameters.return_value, + 'warning_events': report_builder.replace_parameters.return_value, + 'info_events': report_builder.replace_parameters.return_value, + } + + report_builder.replace_parameters.assert_any_call( + report_builder.general_template, + {'models': [ + {'model_name': 'model_name', 'events': [ + {'notification_id': 'ID_1', 'level': 'ERROR', 'project': 'project'}]}, + ]} + ) + report_builder.replace_parameters.assert_any_call( + report_builder.general_template, + {'models': [ + {'model_name': 'model_name', 'events': [ + {'notification_id': 'ID_2', 'level': 'WARNING', 'project': 'project'}]}, + ]} + ) + report_builder.replace_parameters.assert_any_call( + report_builder.general_template, + {'models': [ + {'model_name': 'model_name', 'events': [ + {'notification_id': 'ID_3', 'level': 'INFO', 'project': 'project'}]}, + ]} + ) + + +def test_build_email(report_builder): + report_builder.parameters = MagicMock() + report_builder.replace_parameters = MagicMock() + + report_data = [ + {'notification_id': 'ID_1', 'level': 'ERROR', + 'project': 'project', 'model_name': 'model_name'}, + {'notification_id': 'ID_2', 'level': 'WARNING', + 'project': 'project', 'model_name': 'model_name'}, + {'notification_id': 'ID_2', 'level': 'INFO', + 'project': 'project', 'model_name': 'model_name'}, + {'notification_id': 'ID_3', 'level': 'ERROR', + 'project': 'project', 'model_name': 'model_name'} + ] + + html = report_builder.build_email(report_data, 'type_1') + + report_builder.replace_parameters.assert_called_once_with( + report_builder.report_template, + report_builder.parameters.return_value + ) + + assert html == report_builder.replace_parameters.return_value + + report_builder.parameters.assert_called_once_with( + { + 'ERROR': { + 'section_name': 'Errors detected:', + 'models': [ + { + 'model_name': 'model_name', + 'events': [ + {'notification_id': 'ID_1', 'level': 'ERROR', + 'project': 'project', 'model_name': 'model_name'}, + {'notification_id': 'ID_3', 'level': 'ERROR', + 'project': 'project', 'model_name': 'model_name'} + ] + } + ] + }, + 'WARNING': { + 'section_name': 'Warnings detected:', + 'models': [ + { + 'model_name': 'model_name', + 'events': [ + {'notification_id': 'ID_2', 'level': 'WARNING', + 'project': 'project', 'model_name': 'model_name'} + ] + } + ] + }, + 'INFO': { + 'section_name': 'Infos detected:', + 'models': [ + { + 'model_name': 'model_name', + 'events': [ + {'notification_id': 'ID_2', 'level': 'INFO', + 'project': 'project', 'model_name': 'model_name'} + ] + } + ] + } + }, + 'type_1' + ) diff --git a/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py b/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py new file mode 100644 index 0000000..dc77662 --- /dev/null +++ b/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py @@ -0,0 +1,179 @@ +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, + 'mail_type': 'test_mail_type', + 'base_data_filter': { + 'level': 'ERROR' + } + } + + 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'], + 'mail_type': 'test_mail_type' + }, + 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', + } + ], + 'mail_type': 'test_mail_type' + }, + start_to_close_timeout=ANY, + retry_policy=ANY + ) + ]) + + +@mark.asyncio +@patch("orchestrator.workflows.subworkflows.load_notification_package.workflow", new_callable=AsyncMock) +async def test_run_no_data(workflow_mock, load_notification_package): + input_data = { + 'metadata': metadata, + 'mail_type': 'test_mail_type', + 'base_data_filter': { + 'level': 'ERROR' + } + } + + workflow_mock.start_local_activity_method.side_effect = [ + '2023-01-01 12:00:00', + [], + [] + ] + + output = await load_notification_package.run(input_data) + + assert output == { + 'last_timestamp': '2023-01-01 12:00:00', + 'notification_package': [], + 'sending_configs': [] + } + + workflow_mock.start_activity_method.assert_not_called() + + +@mark.asyncio +@patch("orchestrator.workflows.subworkflows.load_notification_package.workflow", new_callable=AsyncMock) +async def test_run_no_data(workflow_mock, load_notification_package): + input_data = { + 'metadata': metadata, + 'base_data_filter': { + 'level': 'ERROR' + }, + 'mail_type': 'test_mail_type' + } + + workflow_mock.start_local_activity_method.side_effect = [ + '2023-01-01 12:00:00', + ["data"], + [] + ] + + output = await load_notification_package.run(input_data) + + assert output == { + 'last_timestamp': '2023-01-01 12:00:00', + 'notification_package': ["data"], + 'sending_configs': [] + } + + workflow_mock.start_activity_method.assert_not_called() diff --git a/tests/orchestrator/workflows/subworkflows/test_process_notifications.py b/tests/orchestrator/workflows/subworkflows/test_process_notifications.py new file mode 100644 index 0000000..df25c84 --- /dev/null +++ b/tests/orchestrator/workflows/subworkflows/test_process_notifications.py @@ -0,0 +1,84 @@ +from unittest.mock import AsyncMock, patch, ANY, call +from pytest import fixture, mark +from orchestrator.workflows.subworkflows.process_notifications import ProcessNotifications +from orchestrator.activities.activities import Activities + + +@fixture +def process_notifications(): + return ProcessNotifications() + + +metadata = { + 'metadata': { + 'schedule_name': 'test-schedule-name', + 'workflow_name': 'test-workflow', + 'model_name': '-', + 'model_id': '-', + } +} + + +@mark.asyncio +@patch("orchestrator.workflows.subworkflows.process_notifications.workflow", new_callable=AsyncMock) +async def test_run(workflow_mock, process_notifications): + input_data = { + 'metadata': metadata, + 'notification_package': ["content"], + 'mail_type': 'test_mail_type', + 'schema': 'test_schema', + 'table_name': 'test_table_name', + } + + response = await process_notifications.run(input_data) + + assert response == workflow_mock.execute_local_activity_method.return_value + + 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_local_activity_method.assert_has_calls([ + call( + Activities.format_log_report, + { + **metadata, + 'receiver_groups': workflow_mock.execute_activity_method.return_value, + '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 + ), + call( + Activities.export_data_to_postgres, + { + **metadata, + 'schema': input_data['schema'], + 'table_name': input_data['table_name'], + 'data': workflow_mock.execute_local_activity_method.return_value + }, + schedule_to_close_timeout=ANY, + retry_policy=ANY + ) + ]) diff --git a/tests/orchestrator/workflows/test_alerts.py b/tests/orchestrator/workflows/test_alerts.py new file mode 100644 index 0000000..0dfffd3 --- /dev/null +++ b/tests/orchestrator/workflows/test_alerts.py @@ -0,0 +1,117 @@ +from unittest.mock import AsyncMock, patch, ANY, call +from pytest import fixture, mark +from orchestrator.workflows.alerts import Alerts +from orchestrator.activities.activities import Activities + + +@fixture +def alerts(): + return Alerts() + + +metadata = { + 'metadata': { + 'schedule_name': 'test-schedule-name', + 'workflow_name': 'alerts', + 'model_name': '-', + 'model_id': '-', + } +} + + +@mark.asyncio +@patch("orchestrator.workflows.alerts.workflow", new_callable=AsyncMock) +async def test_run_full_flow(workflow_mock, alerts): + input_data = { + 'schedule_name': 'test-schedule-name', + 'notification_ttl': 300, + 'sent_ttl': 600 + } + + await alerts.run(input_data) + + workflow_mock.execute_child_workflow.assert_has_calls([ + call( + 'load_notification_package', + { + **input_data, + 'metadata': metadata, + 'base_data_filter': { + 'level': 'ERROR' + } + } + ) + ]) + + workflow_mock.execute_child_workflow.assert_has_calls([ + call( + 'process_notifications', + { + 'metadata': metadata, + 'mail_type': 'Alerts', + '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_alerts, + { + **metadata, + 'notification_package': workflow_mock.execute_child_workflow.return_value['notification_package'], + 'sending_configs': workflow_mock.execute_child_workflow.return_value['sending_configs'], + 'notification_ttl': input_data['notification_ttl'] + }, + schedule_to_close_timeout=ANY, + retry_policy=ANY + ) + ]) + + workflow_mock.execute_activity_method.assert_has_calls([ + call( + Activities.store_notification_cache, + { + **metadata, + 'log_report': workflow_mock.execute_child_workflow.return_value, + 'sent_ttl': input_data['sent_ttl'] + }, + schedule_to_close_timeout=ANY, + retry_policy=ANY + ) + ]) + + +@mark.asyncio +@patch("orchestrator.workflows.alerts.workflow", new_callable=AsyncMock) +async def test_run_no_data(workflow_mock, alerts): + 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 alerts.run(input_data) + + workflow_mock.execute_child_workflow.assert_has_calls([ + call( + 'load_notification_package', + { + **input_data, + 'metadata': metadata, + 'base_data_filter': { + 'level': 'ERROR' + } + } + ) + ]) + + workflow_mock.execute_local_activity_method.assert_not_called() 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() diff --git a/values.yaml b/values.yaml index 4614016..570ae29 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.2.5" + tag: "0.2.7" # 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: @@ -132,7 +132,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-1171-criar-pipeline-de-retreino-laborious" + value: "SIENTIAPDE-1172-criar-pipeline-de-alertas-orquestrador" - name: PYTHON_APP value: "orchestrator.worker.worker" @@ -170,6 +170,34 @@ env: - name: MONGODB_TTL_INDEX_HOURS value: "1" + - name: EMAIL_SENDER + value: "vitor.santos@aignosi.com.br" + - name: EMAIL_SENDER_PASSWORD + valueFrom: + secretKeyRef: + name: smtp-credentials + key: app_password + - name: EMAIL_SMTP_SERVER + value: "smtp.gmail.com" + - name: EMAIL_SMTP_PORT + value: "587" + + # Application variables + - name: POSTGRES_HOST + value: "paradedb-rw.paradedb.svc.cluster.local" + - name: POSTGRES_PORT + value: "5432" + - name: POSTGRES_USER + value: "sientia" + - name: POSTGRES_PASSWORD + value: "sientia" + - name: POSTGRES_DBNAME + value: "sientia" + - name: POSTGRES_MIN_CONNECTIONS + value: "10" + - name: POSTGRES_MAX_CONNECTIONS + value: "40" + - name: KAFKA_BOOTSTRAP_SERVERS value: "kafka.kafka.svc.cluster.local:9092" @@ -201,3 +229,7 @@ ssh: # --namespace sientia \ # --from-file=ssh-privatekey=git_key \ # --type=kubernetes.io/ssh-auth + +# kubectl create secret generic smtp-credentials \ +# --namespace sientia \ +# --from-literal=app_password='sua-senha-de-app-de-16-digitos'