From 7fd38e1d52990ff8d3bfef30e368f18635b675cb Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 12 Aug 2025 13:36:13 -0300 Subject: [PATCH 01/11] SIENTIAPDE-1184 fix: add conditional check for data_filter in MongoDB activity - Updated the insert_many method to only execute if data_filter is not empty, preventing unnecessary database operations and potential errors. --- init_orchestration.ipynb | 332 ++++++++++++++++++++++++++++ orchestrator/activities/mongo_db.py | 4 +- 2 files changed, 334 insertions(+), 2 deletions(-) create mode 100644 init_orchestration.ipynb diff --git a/init_orchestration.ipynb b/init_orchestration.ipynb new file mode 100644 index 0000000..14f66c0 --- /dev/null +++ b/init_orchestration.ipynb @@ -0,0 +1,332 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": 2, + "id": "3f8b77a4", + "metadata": {}, + "outputs": [], + "source": [ + "from temporalio import client\n", + "from orchestrator.activities.temporal_manager import TemporalManager\n", + "import os\n", + "from unittest.mock import MagicMock\n", + "\n", + "host = \"localhost:7233\"\n", + "logger = MagicMock(info=MagicMock(side_effect=print), debug=MagicMock(side_effect=print))\n", + "\n", + "temporal_client = await client.Client.connect(\n", + " target_host=host,\n", + " namespace=os.getenv('TEMPORAL_NAMESPACE', 'default')\n", + ")" + ] + }, + { + "cell_type": "code", + "execution_count": 3, + "id": "6ee4b5a7", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "" + ] + }, + "execution_count": 3, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "from datetime import timedelta\n", + "from temporalio.client import (\n", + " Client,\n", + " Schedule,\n", + " ScheduleActionStartWorkflow,\n", + " ScheduleIntervalSpec,\n", + " ScheduleSpec,\n", + ")\n", + "from temporalio.common import TypedSearchAttributes, SearchAttributeKey, SearchAttributePair\n", + "\n", + "# temporal operator search-attribute create --namespace scouter --name model_id --type Text && temporal operator search-attribute create --namespace scouter --name orchestrated --type Text && temporal operator search-attribute create --namespace scouter --name model_name --type Text && temporal operator search-attribute create --namespace laborious --name model_id --type Text && temporal operator search-attribute create --namespace laborious --name orchestrated --type Text && temporal operator search-attribute create --namespace laborious --name model_name --type Text\n", + "\n", + "\n", + "\n", + "await temporal_client.create_schedule(\n", + " \"orchestrator\",\n", + " Schedule(\n", + " action=ScheduleActionStartWorkflow(\n", + " 'orchestrator',\n", + " {\n", + " \"schedule_name\": \"orchestrator-test\",\n", + " \"pipelines_query\": {\n", + " \"collection\": \"pipelines\",\n", + " \"aggregation\": [\n", + " {\n", + " \"$lookup\": {\n", + " \"from\": \"models\",\n", + " \"localField\": \"model_id\",\n", + " \"foreignField\": \"id\",\n", + " \"as\": \"model_docs\"\n", + " }\n", + " },\n", + " {\n", + " \"$match\": {\n", + " \"active\": True\n", + " }\n", + " },\n", + " {\n", + " \"$addFields\": {\n", + " \"models\": {\n", + " \"$arrayElemAt\": [\n", + " \"$model_docs\",\n", + " 0\n", + " ]\n", + " }\n", + " }\n", + " },\n", + " {\n", + " \"$match\": {\n", + " \"models.active\": True\n", + " }\n", + " },\n", + " {\n", + " \"$project\": {\n", + " \"model_docs\": 0\n", + " }\n", + " }\n", + " ]\n", + " },\n", + " \"opc_servers_query\": {\n", + " \"collection\": \"opc-servers\",\n", + " \"filters\": {\n", + "\n", + " }\n", + " }\n", + " },\n", + " id=\"orchestrator\",\n", + " task_queue=\"orchestrator-queue\",\n", + " execution_timeout=timedelta(minutes=600)\n", + " ),\n", + " spec=ScheduleSpec(\n", + " intervals=[ScheduleIntervalSpec(every=timedelta(minutes=60))]\n", + " )\n", + " )\n", + ")" + ] + }, + { + "cell_type": "code", + "execution_count": 4, + "id": "d9d2a242", + "metadata": {}, + "outputs": [ + { + "ename": "ScheduleAlreadyRunningError", + "evalue": "Schedule already running", + "output_type": "error", + "traceback": [ + "\u001b[31m---------------------------------------------------------------------------\u001b[39m", + "\u001b[31mRPCError\u001b[39m Traceback (most recent call last)", + "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/temporalio/service.py:1243\u001b[39m, in \u001b[36m_BridgeServiceClient._rpc_call\u001b[39m\u001b[34m(self, rpc, req, resp_type, service, retry, metadata, timeout)\u001b[39m\n\u001b[32m 1242\u001b[39m client = \u001b[38;5;28;01mawait\u001b[39;00m \u001b[38;5;28mself\u001b[39m._connected_client()\n\u001b[32m-> \u001b[39m\u001b[32m1243\u001b[39m resp = \u001b[38;5;28;01mawait\u001b[39;00m client.call(\n\u001b[32m 1244\u001b[39m service=service,\n\u001b[32m 1245\u001b[39m rpc=rpc,\n\u001b[32m 1246\u001b[39m req=req,\n\u001b[32m 1247\u001b[39m resp_type=resp_type,\n\u001b[32m 1248\u001b[39m retry=retry,\n\u001b[32m 1249\u001b[39m metadata=metadata,\n\u001b[32m 1250\u001b[39m timeout=timeout,\n\u001b[32m 1251\u001b[39m )\n\u001b[32m 1252\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m LOG_PROTOS:\n", + "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/temporalio/bridge/client.py:151\u001b[39m, in \u001b[36mClient.call\u001b[39m\u001b[34m(self, service, rpc, req, resp_type, retry, metadata, timeout)\u001b[39m\n\u001b[32m 150\u001b[39m resp = resp_type()\n\u001b[32m--> \u001b[39m\u001b[32m151\u001b[39m resp.ParseFromString(\u001b[38;5;28;01mawait\u001b[39;00m resp_fut)\n\u001b[32m 152\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m resp\n", + "\u001b[31mRPCError\u001b[39m: (6, 'Workflow execution is already running. WorkflowId: temporal-sys-scheduler:orchestrator, RunId: 01989edb-863c-7585-8653-945d00384e2f.', b'\\x08\\x06\\x12\\x84\\x01Workflow execution is already running. WorkflowId: temporal-sys-scheduler:orchestrator, RunId: 01989edb-863c-7585-8653-945d00384e2f.\\x1a\\xa7\\x01\\nWtype.googleapis.com/temporal.api.errordetails.v1.WorkflowExecutionAlreadyStartedFailure\\x12L\\n$d248fcc1-8071-40fc-b5f3-e70b8cb1dbb5\\x12$01989edb-863c-7585-8653-945d00384e2f')", + "\nDuring handling of the above exception, another exception occurred:\n", + "\u001b[31mRPCError\u001b[39m Traceback (most recent call last)", + "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/temporalio/client.py:6430\u001b[39m, in \u001b[36m_ClientImpl.create_schedule\u001b[39m\u001b[34m(self, input)\u001b[39m\n\u001b[32m 6427\u001b[39m temporalio.converter.encode_search_attributes(\n\u001b[32m 6428\u001b[39m \u001b[38;5;28minput\u001b[39m.search_attributes, request.search_attributes\n\u001b[32m 6429\u001b[39m )\n\u001b[32m-> \u001b[39m\u001b[32m6430\u001b[39m \u001b[38;5;28;01mawait\u001b[39;00m \u001b[38;5;28mself\u001b[39m._client.workflow_service.create_schedule(\n\u001b[32m 6431\u001b[39m request,\n\u001b[32m 6432\u001b[39m retry=\u001b[38;5;28;01mTrue\u001b[39;00m,\n\u001b[32m 6433\u001b[39m metadata=\u001b[38;5;28minput\u001b[39m.rpc_metadata,\n\u001b[32m 6434\u001b[39m timeout=\u001b[38;5;28minput\u001b[39m.rpc_timeout,\n\u001b[32m 6435\u001b[39m )\n\u001b[32m 6436\u001b[39m \u001b[38;5;28;01mexcept\u001b[39;00m RPCError \u001b[38;5;28;01mas\u001b[39;00m err:\n", + "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/temporalio/service.py:1170\u001b[39m, in \u001b[36mServiceCall.__call__\u001b[39m\u001b[34m(self, req, retry, metadata, timeout)\u001b[39m\n\u001b[32m 1155\u001b[39m \u001b[38;5;250m\u001b[39m\u001b[33;03m\"\"\"Invoke underlying client with the given request.\u001b[39;00m\n\u001b[32m 1156\u001b[39m \n\u001b[32m 1157\u001b[39m \u001b[33;03mArgs:\u001b[39;00m\n\u001b[32m (...)\u001b[39m\u001b[32m 1168\u001b[39m \u001b[33;03m RPCError: Any RPC error that occurs during the call.\u001b[39;00m\n\u001b[32m 1169\u001b[39m \u001b[33;03m\"\"\"\u001b[39;00m\n\u001b[32m-> \u001b[39m\u001b[32m1170\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28;01mawait\u001b[39;00m \u001b[38;5;28mself\u001b[39m.service_client._rpc_call(\n\u001b[32m 1171\u001b[39m \u001b[38;5;28mself\u001b[39m.name,\n\u001b[32m 1172\u001b[39m req,\n\u001b[32m 1173\u001b[39m \u001b[38;5;28mself\u001b[39m.resp_type,\n\u001b[32m 1174\u001b[39m service=\u001b[38;5;28mself\u001b[39m.service,\n\u001b[32m 1175\u001b[39m retry=retry,\n\u001b[32m 1176\u001b[39m metadata=metadata,\n\u001b[32m 1177\u001b[39m timeout=timeout,\n\u001b[32m 1178\u001b[39m )\n", + "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/temporalio/service.py:1258\u001b[39m, in \u001b[36m_BridgeServiceClient._rpc_call\u001b[39m\u001b[34m(self, rpc, req, resp_type, service, retry, metadata, timeout)\u001b[39m\n\u001b[32m 1257\u001b[39m status, message, details = err.args\n\u001b[32m-> \u001b[39m\u001b[32m1258\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m RPCError(message, RPCStatusCode(status), details)\n", + "\u001b[31mRPCError\u001b[39m: Workflow execution is already running. WorkflowId: temporal-sys-scheduler:orchestrator, RunId: 01989edb-863c-7585-8653-945d00384e2f.", + "\nDuring handling of the above exception, another exception occurred:\n", + "\u001b[31mScheduleAlreadyRunningError\u001b[39m Traceback (most recent call last)", + "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[4]\u001b[39m\u001b[32m, line 13\u001b[39m\n\u001b[32m 2\u001b[39m \u001b[38;5;28;01mfrom\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34;01mtemporalio\u001b[39;00m\u001b[34;01m.\u001b[39;00m\u001b[34;01mclient\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;28;01mimport\u001b[39;00m (\n\u001b[32m 3\u001b[39m Schedule,\n\u001b[32m 4\u001b[39m ScheduleActionStartWorkflow,\n\u001b[32m 5\u001b[39m ScheduleIntervalSpec,\n\u001b[32m 6\u001b[39m ScheduleSpec,\n\u001b[32m 7\u001b[39m )\n\u001b[32m 9\u001b[39m \u001b[38;5;66;03m# temporal operator search-attribute create --namespace scouter --name model_id --type Text && temporal operator search-attribute create --namespace scouter --name orchestrated --type Text && temporal operator search-attribute create --namespace scouter --name model_name --type Text && temporal operator search-attribute create --namespace laborious --name model_id --type Text && temporal operator search-attribute create --namespace laborious --name orchestrated --type Text && temporal operator search-attribute create --namespace laborious --name model_name --type Text\u001b[39;00m\n\u001b[32m---> \u001b[39m\u001b[32m13\u001b[39m \u001b[38;5;28;01mawait\u001b[39;00m temporal_client.create_schedule(\n\u001b[32m 14\u001b[39m \u001b[33m\"\u001b[39m\u001b[33morchestrator\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 15\u001b[39m Schedule(\n\u001b[32m 16\u001b[39m action=ScheduleActionStartWorkflow(\n\u001b[32m 17\u001b[39m \u001b[33m'\u001b[39m\u001b[33morchestrator\u001b[39m\u001b[33m'\u001b[39m,\n\u001b[32m 18\u001b[39m {\n\u001b[32m 19\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mschedule_name\u001b[39m\u001b[33m\"\u001b[39m: \u001b[33m\"\u001b[39m\u001b[33morchestrator-test\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 20\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mpipelines_query\u001b[39m\u001b[33m\"\u001b[39m: {\n\u001b[32m 21\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mcollection\u001b[39m\u001b[33m\"\u001b[39m: \u001b[33m\"\u001b[39m\u001b[33mpipelines\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 22\u001b[39m \u001b[33m\"\u001b[39m\u001b[33maggregation\u001b[39m\u001b[33m\"\u001b[39m: [\n\u001b[32m 23\u001b[39m {\n\u001b[32m 24\u001b[39m \u001b[33m\"\u001b[39m\u001b[33m$lookup\u001b[39m\u001b[33m\"\u001b[39m: {\n\u001b[32m 25\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mfrom\u001b[39m\u001b[33m\"\u001b[39m: \u001b[33m\"\u001b[39m\u001b[33mmodels\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 26\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mlocalField\u001b[39m\u001b[33m\"\u001b[39m: \u001b[33m\"\u001b[39m\u001b[33mmodel_id\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 27\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mforeignField\u001b[39m\u001b[33m\"\u001b[39m: \u001b[33m\"\u001b[39m\u001b[33mid\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 28\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mas\u001b[39m\u001b[33m\"\u001b[39m: \u001b[33m\"\u001b[39m\u001b[33mmodel_docs\u001b[39m\u001b[33m\"\u001b[39m\n\u001b[32m 29\u001b[39m }\n\u001b[32m 30\u001b[39m },\n\u001b[32m 31\u001b[39m {\n\u001b[32m 32\u001b[39m \u001b[33m\"\u001b[39m\u001b[33m$match\u001b[39m\u001b[33m\"\u001b[39m: {\n\u001b[32m 33\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mactive\u001b[39m\u001b[33m\"\u001b[39m: \u001b[38;5;28;01mTrue\u001b[39;00m\n\u001b[32m 34\u001b[39m }\n\u001b[32m 35\u001b[39m },\n\u001b[32m 36\u001b[39m {\n\u001b[32m 37\u001b[39m \u001b[33m\"\u001b[39m\u001b[33m$addFields\u001b[39m\u001b[33m\"\u001b[39m: {\n\u001b[32m 38\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mmodels\u001b[39m\u001b[33m\"\u001b[39m: {\n\u001b[32m 39\u001b[39m \u001b[33m\"\u001b[39m\u001b[33m$arrayElemAt\u001b[39m\u001b[33m\"\u001b[39m: [\n\u001b[32m 40\u001b[39m \u001b[33m\"\u001b[39m\u001b[33m$model_docs\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 41\u001b[39m \u001b[32m0\u001b[39m\n\u001b[32m 42\u001b[39m ]\n\u001b[32m 43\u001b[39m }\n\u001b[32m 44\u001b[39m }\n\u001b[32m 45\u001b[39m },\n\u001b[32m 46\u001b[39m {\n\u001b[32m 47\u001b[39m \u001b[33m\"\u001b[39m\u001b[33m$match\u001b[39m\u001b[33m\"\u001b[39m: {\n\u001b[32m 48\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mmodels.active\u001b[39m\u001b[33m\"\u001b[39m: \u001b[38;5;28;01mTrue\u001b[39;00m\n\u001b[32m 49\u001b[39m }\n\u001b[32m 50\u001b[39m },\n\u001b[32m 51\u001b[39m {\n\u001b[32m 52\u001b[39m \u001b[33m\"\u001b[39m\u001b[33m$project\u001b[39m\u001b[33m\"\u001b[39m: {\n\u001b[32m 53\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mmodel_docs\u001b[39m\u001b[33m\"\u001b[39m: \u001b[32m0\u001b[39m\n\u001b[32m 54\u001b[39m }\n\u001b[32m 55\u001b[39m }\n\u001b[32m 56\u001b[39m ]\n\u001b[32m 57\u001b[39m },\n\u001b[32m 58\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mopc_servers_query\u001b[39m\u001b[33m\"\u001b[39m: {\n\u001b[32m 59\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mcollection\u001b[39m\u001b[33m\"\u001b[39m: \u001b[33m\"\u001b[39m\u001b[33mopc-servers\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 60\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mfilters\u001b[39m\u001b[33m\"\u001b[39m: {\n\u001b[32m 61\u001b[39m \n\u001b[32m 62\u001b[39m }\n\u001b[32m 63\u001b[39m }\n\u001b[32m 64\u001b[39m },\n\u001b[32m 65\u001b[39m \u001b[38;5;28mid\u001b[39m=\u001b[33m\"\u001b[39m\u001b[33morchestrator\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 66\u001b[39m task_queue=\u001b[33m\"\u001b[39m\u001b[33morchestrator-queue\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 67\u001b[39m execution_timeout=timedelta(minutes=\u001b[32m600\u001b[39m)\n\u001b[32m 68\u001b[39m ),\n\u001b[32m 69\u001b[39m spec=ScheduleSpec(\n\u001b[32m 70\u001b[39m intervals=[ScheduleIntervalSpec(every=timedelta(minutes=\u001b[32m60\u001b[39m))]\n\u001b[32m 71\u001b[39m )\n\u001b[32m 72\u001b[39m )\n\u001b[32m 73\u001b[39m )\n", + "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/temporalio/client.py:1308\u001b[39m, in \u001b[36mClient.create_schedule\u001b[39m\u001b[34m(self, id, schedule, trigger_immediately, backfill, memo, search_attributes, static_summary, static_details, rpc_metadata, rpc_timeout)\u001b[39m\n\u001b[32m 1275\u001b[39m \u001b[38;5;250m\u001b[39m\u001b[33;03m\"\"\"Create a schedule and return its handle.\u001b[39;00m\n\u001b[32m 1276\u001b[39m \n\u001b[32m 1277\u001b[39m \u001b[33;03mArgs:\u001b[39;00m\n\u001b[32m (...)\u001b[39m\u001b[32m 1305\u001b[39m \u001b[33;03m running.\u001b[39;00m\n\u001b[32m 1306\u001b[39m \u001b[33;03m\"\"\"\u001b[39;00m\n\u001b[32m 1307\u001b[39m temporalio.common._warn_on_deprecated_search_attributes(search_attributes)\n\u001b[32m-> \u001b[39m\u001b[32m1308\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28;01mawait\u001b[39;00m \u001b[38;5;28mself\u001b[39m._impl.create_schedule(\n\u001b[32m 1309\u001b[39m CreateScheduleInput(\n\u001b[32m 1310\u001b[39m \u001b[38;5;28mid\u001b[39m=\u001b[38;5;28mid\u001b[39m,\n\u001b[32m 1311\u001b[39m schedule=schedule,\n\u001b[32m 1312\u001b[39m trigger_immediately=trigger_immediately,\n\u001b[32m 1313\u001b[39m backfill=backfill,\n\u001b[32m 1314\u001b[39m memo=memo,\n\u001b[32m 1315\u001b[39m search_attributes=search_attributes,\n\u001b[32m 1316\u001b[39m rpc_metadata=rpc_metadata,\n\u001b[32m 1317\u001b[39m rpc_timeout=rpc_timeout,\n\u001b[32m 1318\u001b[39m )\n\u001b[32m 1319\u001b[39m )\n", + "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/temporalio/client.py:6445\u001b[39m, in \u001b[36m_ClientImpl.create_schedule\u001b[39m\u001b[34m(self, input)\u001b[39m\n\u001b[32m 6437\u001b[39m already_started = (\n\u001b[32m 6438\u001b[39m err.status == RPCStatusCode.ALREADY_EXISTS\n\u001b[32m 6439\u001b[39m \u001b[38;5;129;01mand\u001b[39;00m err.grpc_status.details\n\u001b[32m (...)\u001b[39m\u001b[32m 6442\u001b[39m )\n\u001b[32m 6443\u001b[39m )\n\u001b[32m 6444\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m already_started:\n\u001b[32m-> \u001b[39m\u001b[32m6445\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m ScheduleAlreadyRunningError()\n\u001b[32m 6446\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m\n\u001b[32m 6447\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m ScheduleHandle(\u001b[38;5;28mself\u001b[39m._client, \u001b[38;5;28minput\u001b[39m.id)\n", + "\u001b[31mScheduleAlreadyRunningError\u001b[39m: Schedule already running" + ] + } + ], + "source": [ + "from datetime import timedelta\n", + "from temporalio.client import (\n", + " Schedule,\n", + " ScheduleActionStartWorkflow,\n", + " ScheduleIntervalSpec,\n", + " ScheduleSpec,\n", + ")\n", + "\n", + "# temporal operator search-attribute create --namespace scouter --name model_id --type Text && temporal operator search-attribute create --namespace scouter --name orchestrated --type Text && temporal operator search-attribute create --namespace scouter --name model_name --type Text && temporal operator search-attribute create --namespace laborious --name model_id --type Text && temporal operator search-attribute create --namespace laborious --name orchestrated --type Text && temporal operator search-attribute create --namespace laborious --name model_name --type Text\n", + "\n", + "\n", + "\n", + "await temporal_client.create_schedule(\n", + " \"orchestrator\",\n", + " Schedule(\n", + " action=ScheduleActionStartWorkflow(\n", + " 'orchestrator',\n", + " {\n", + " \"schedule_name\": \"orchestrator-test\",\n", + " \"pipelines_query\": {\n", + " \"collection\": \"pipelines\",\n", + " \"aggregation\": [\n", + " {\n", + " \"$lookup\": {\n", + " \"from\": \"models\",\n", + " \"localField\": \"model_id\",\n", + " \"foreignField\": \"id\",\n", + " \"as\": \"model_docs\"\n", + " }\n", + " },\n", + " {\n", + " \"$match\": {\n", + " \"active\": True\n", + " }\n", + " },\n", + " {\n", + " \"$addFields\": {\n", + " \"models\": {\n", + " \"$arrayElemAt\": [\n", + " \"$model_docs\",\n", + " 0\n", + " ]\n", + " }\n", + " }\n", + " },\n", + " {\n", + " \"$match\": {\n", + " \"models.active\": True\n", + " }\n", + " },\n", + " {\n", + " \"$project\": {\n", + " \"model_docs\": 0\n", + " }\n", + " }\n", + " ]\n", + " },\n", + " \"opc_servers_query\": {\n", + " \"collection\": \"opc-servers\",\n", + " \"filters\": {\n", + "\n", + " }\n", + " }\n", + " },\n", + " id=\"orchestrator\",\n", + " task_queue=\"orchestrator-queue\",\n", + " execution_timeout=timedelta(minutes=600)\n", + " ),\n", + " spec=ScheduleSpec(\n", + " intervals=[ScheduleIntervalSpec(every=timedelta(minutes=60))]\n", + " )\n", + " )\n", + ")" + ] + }, + { + "cell_type": "code", + "execution_count": 5, + "id": "6ea0f616", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "" + ] + }, + "execution_count": 5, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "await temporal_client.create_schedule(\n", + " \"alerts\",\n", + " Schedule(\n", + " action=ScheduleActionStartWorkflow(\n", + " 'alerts',\n", + " {\n", + " \"schedule_name\": \"alerts\",\n", + " \"notification_ttl\": 5*60,\n", + " \"sent_ttl\": 10*60\n", + " },\n", + " id=\"alerts\",\n", + " task_queue=\"alerts-queue\",\n", + " execution_timeout=timedelta(minutes=600)\n", + " ),\n", + " spec=ScheduleSpec(\n", + " intervals=[ScheduleIntervalSpec(every=timedelta(seconds=30))]\n", + " )\n", + " )\n", + ")" + ] + }, + { + "cell_type": "code", + "execution_count": 6, + "id": "05c46ec8", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "" + ] + }, + "execution_count": 6, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "await temporal_client.create_schedule(\n", + " \"reports\",\n", + " Schedule(\n", + " action=ScheduleActionStartWorkflow(\n", + " 'reports',\n", + " {\n", + " \"schedule_name\": \"reports\",\n", + " },\n", + " id=\"reports\",\n", + " task_queue=\"reports-queue\",\n", + " execution_timeout=timedelta(minutes=600)\n", + " ),\n", + " spec=ScheduleSpec(\n", + " intervals=[ScheduleIntervalSpec(every=timedelta(minutes=20))]\n", + " )\n", + " )\n", + ")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "76e3d8a7", + "metadata": {}, + "outputs": [], + "source": [] + } + ], + "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/mongo_db.py b/orchestrator/activities/mongo_db.py index 105264e..4f9ba78 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -1,4 +1,3 @@ -from pandas import DataFrame from temporalio import workflow, activity @@ -257,7 +256,8 @@ class MongoDB(BaseActivity): data_filter = argument if argument else {} try: - collection.insert_many(data_filter) + if data_filter: + collection.insert_many(data_filter) except Exception as e: trace = traceback.format_exc() self.send_notification( From 8537eb515b4e74940511c6ce6a321edd02330802 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 12 Aug 2025 13:42:30 -0300 Subject: [PATCH 02/11] SIENTIAPDE-1184 chore: update image tag in values.yaml from 0.3.2 to 0.4.0 and change GITHUB_BRANCH to SIENTIAPDE-1184-investigar-bugs-detectados-no-grafana --- values.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/values.yaml b/values.yaml index 1ae2b9b..7a7aba0 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.3.2" + tag: "0.4.0" # 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: @@ -144,7 +144,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-1174-mapear-e-implementar-metricas-a-serem-criadas" + value: "SIENTIAPDE-1184-investigar-bugs-detectados-no-grafana" - name: PYTHON_APP value: "orchestrator.worker.worker" From 5477f68b6bd19caca88abca2724644de7855bc16 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 12 Aug 2025 14:12:52 -0300 Subject: [PATCH 03/11] SIENTIAPDE-1184 fix: update notification IDs for error reporting in Formatters class - Changed notification IDs for error reports related to schedule creation, update, and deletion to include the suffix '_ERROR' for better clarity in error handling. --- orchestrator/activities/formatters.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 04a983d..2e8cf1b 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -391,7 +391,7 @@ class Formatters(BaseActivity): self.send_error_report( metadata=metadata, message=f"Failed to create schedules: \n {', '.join(error_keys)}", - notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES", + notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES_ERROR", attachment=created_schedules ) @@ -411,7 +411,7 @@ class Formatters(BaseActivity): self.send_error_report( metadata=metadata, message=f"Failed to update schedules: \n {', '.join(error_keys)}", - notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES", + notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES_ERROR", attachment=updated_schedules ) @@ -430,7 +430,7 @@ class Formatters(BaseActivity): self.send_error_report( metadata=metadata, message=f"Failed to delete schedules: \n {', '.join(error_keys)}", - notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES", + notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES_ERROR", attachment=deleted_schedules ) From 1306bed21de85f601f41a981dc85b2da438a7559 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 12 Aug 2025 16:26:24 -0300 Subject: [PATCH 04/11] SIENTIAPDE-1184 fix: enhance error reporting in TemporalManager class - Added typed_search_attributes to the schedule creation for improved search capabilities. - Included traceback in error messages when schedule deletion fails, providing better context for debugging. --- orchestrator/activities/temporal_manager.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/orchestrator/activities/temporal_manager.py b/orchestrator/activities/temporal_manager.py index f8461eb..eb7d361 100644 --- a/orchestrator/activities/temporal_manager.py +++ b/orchestrator/activities/temporal_manager.py @@ -167,7 +167,8 @@ class TemporalManager(BaseActivity): schedule, id=schedule_name, task_queue=f"{workflow_type}-queue", - execution_timeout=timedelta(minutes=2) + execution_timeout=timedelta(minutes=2), + typed_search_attributes=search_attributes ), spec=ScheduleSpec( intervals=[ @@ -333,13 +334,15 @@ class TemporalManager(BaseActivity): "message": "Schedule deleted successfully" }) except Exception as e: + trace = traceback.format_exc() self.error( f"Failed to delete schedule {schedule_name}: {str(e)}", metadata=metadata) report.append({ "namespace": namespace, "schedule_name": schedule_name, "success": False, - "message": str(e) + "message": str(e), + "attachment": trace }) self.info( From 4d75d581f8c5f235befcbf7fec6d887bfd1f9309 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 13 Aug 2025 16:34:02 -0300 Subject: [PATCH 05/11] SIENTIAPDE-1184 Changing "updated_at" type from str to datetime --- orchestrator/activities/formatters.py | 2 +- orchestrator/activities/mongo_db.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 2e8cf1b..9bb17ff 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -198,7 +198,7 @@ class Formatters(BaseActivity): if schedule_name in current_schedules: update_timestamp = schedule.get( - 'updated_at', datetime.now().strftime(DEFAULT_DATE_FORMAT)) + 'updated_at', datetime.now()) old_timestamp = current_schedules[schedule_name] diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 4f9ba78..80acd0c 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -206,7 +206,7 @@ class MongoDB(BaseActivity): """ updated_pipelines = input_data.get("updated_pipelines", []) metadata = input_data.get("metadata", {}) - now = datetime.now().strftime(DEFAULT_DATE_FORMAT) + now = datetime.now() collection = self.database["orchestrated_schedules"] argument = [ @@ -245,7 +245,7 @@ class MongoDB(BaseActivity): metadata = input_data.get("metadata", {}) collection = self.database["orchestrated_schedules"] - now = datetime.now().strftime(DEFAULT_DATE_FORMAT) + now = datetime.now() argument = [ {"schedule_name": pipeline["schedule_name"], From e8ce3e75d55403bc2b7da61345a178d475bd77e0 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 14 Aug 2025 08:50:21 -0300 Subject: [PATCH 06/11] SIENTIAPDE-1184 fix: enhance error reporting in Formatters class - Updated the parse_report_schedule method to return a dictionary for error keys, including both message and attachment. - Improved error report formatting in send_error_report method to include attachments for better context in failure notifications. --- orchestrator/activities/formatters.py | 48 +++++++++-- .../activities/test_formatters.py | 83 +++++++++++++------ .../orchestrator/activities/test_mongo_db.py | 6 +- .../activities/test_temporal_manager.py | 12 ++- 4 files changed, 109 insertions(+), 40 deletions(-) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 9bb17ff..bf6f0cf 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -336,12 +336,15 @@ class Formatters(BaseActivity): attachment_content=json.dumps(attachment, indent=4, sort_keys=True) ) - def parse_report_schedule(self, input_data: dict[str, Any]) -> tuple[list[str], list[str]]: + def parse_report_schedule(self, input_data: dict[str, Any]) -> tuple[list[str], dict[str, Any]]: success_keys = [f"{value['namespace']}/{value['schedule_name']}" for value in input_data if value['success']] - error_keys = [f"{value['namespace']}/{value['schedule_name']}: {value['message']}" - for value in input_data if not value['success']] + error_keys = {f"{value['namespace']}/{value['schedule_name']}": { + 'message': value['message'], + 'attachment': value.get('attachment', None) + } + for value in input_data if not value['success']} return success_keys, error_keys @@ -384,15 +387,24 @@ class Formatters(BaseActivity): self.send_success_report( metadata=metadata, message=f"Created schedules: \n {', '.join(success_keys)}", - notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES" + notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES", + attachment=created_schedules ) if len(error_keys) > 0: + attachment = [] + for value in error_keys.values(): + if value['attachment'] is not None: + attachment.append( + f"{value['message']}\n{value['attachment']}") + else: + attachment.append(value['message']) + self.send_error_report( metadata=metadata, message=f"Failed to create schedules: \n {', '.join(error_keys)}", notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES_ERROR", - attachment=created_schedules + attachment="\n ========== \n".join(attachment) ) # Send report for updated schedules @@ -404,15 +416,24 @@ class Formatters(BaseActivity): self.send_success_report( metadata=metadata, message=f"Updated schedules: \n {', '.join(success_keys)}", - notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES" + notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES", + attachment=updated_schedules ) if len(error_keys) > 0: + attachment = [] + for value in error_keys.values(): + if value['attachment'] is not None: + attachment.append( + f"{value['message']}\n{value['attachment']}") + else: + attachment.append(value['message']) + self.send_error_report( metadata=metadata, message=f"Failed to update schedules: \n {', '.join(error_keys)}", notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES_ERROR", - attachment=updated_schedules + attachment="\n ========== \n".join(attachment) ) if len(deleted_schedules) > 0: @@ -423,15 +444,24 @@ class Formatters(BaseActivity): self.send_success_report( metadata=metadata, message=f"Deleted schedules: \n {', '.join(success_keys)}", - notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES" + notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES", + attachment=deleted_schedules ) if len(error_keys) > 0: + attachment = [] + for value in error_keys.values(): + if value['attachment'] is not None: + attachment.append( + f"{value['message']}\n{value['attachment']}") + else: + attachment.append(value['message']) + self.send_error_report( metadata=metadata, message=f"Failed to delete schedules: \n {', '.join(error_keys)}", notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES_ERROR", - attachment=deleted_schedules + attachment="\n ========== \n".join(attachment) ) @activity.defn(name="report_slot_orchestration") diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index f3c1397..c480997 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -527,14 +527,18 @@ def test_parse_report_schedule(formatters): "namespace": "test_namespace", "schedule_name": "test_schedule_name_to_create_error", "success": False, - "message": "test_error" + "message": "test_error", + "attachment": "test_attachment" } ] result = formatters.parse_report_schedule(input_data) assert result == ( ["test_namespace/test_schedule_name_to_create"], - ["test_namespace/test_schedule_name_to_create_error: test_error"] + {"test_namespace/test_schedule_name_to_create_error": { + 'message': 'test_error', + 'attachment': 'test_attachment' + }} ) @@ -558,7 +562,14 @@ async def test_report_schedule_orchestration(formatters): "namespace": "test_namespace", "schedule_name": "test_schedule_name_to_create_error", "success": False, - "message": "test_error" + "message": "test_error1", + "attachment": "test_attachment1" + }, + { + "namespace": "test_namespace", + "schedule_name": "test_schedule_name_to_create_error2", + "success": False, + "message": "test_error2" } ], "updated_schedules": [ @@ -571,7 +582,21 @@ async def test_report_schedule_orchestration(formatters): "namespace": "test_namespace", "schedule_name": "test_schedule_name_to_update_error", "success": False, - "message": "test_error" + "message": "test_error2", + "attachment": "test_attachment2" + }, + { + "namespace": "test_namespace", + "schedule_name": "test_schedule_name_to_update_error2", + "success": False, + "message": "test_error3", + "attachment": "test_attachment3" + }, + { + "namespace": "test_namespace", + "schedule_name": "test_schedule_name_to_update_error3", + "success": False, + "message": "test_error4" } ], "deleted_schedules": [ @@ -584,7 +609,14 @@ async def test_report_schedule_orchestration(formatters): "namespace": "test_namespace", "schedule_name": "test_schedule_name_to_delete_error", "success": False, - "message": "test_error" + "message": "test_error4", + "attachment": "test_attachment4" + }, + { + "namespace": "test_namespace", + "schedule_name": "test_schedule_name_to_delete_error2", + "success": False, + "message": "test_error5" } ] } @@ -600,39 +632,42 @@ async def test_report_schedule_orchestration(formatters): call( metadata=metadata['metadata'], message="Created schedules: \n test_namespace/test_schedule_name_to_create", - notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES" - ), - call( - metadata=metadata['metadata'], - message="Updated schedules: \n test_namespace/test_schedule_name_to_update", - notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES" - ), - call( - metadata=metadata['metadata'], - message="Deleted schedules: \n test_namespace/test_schedule_name_to_delete", - notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES" - ) - ]) - formatters.send_error_report.assert_has_calls([ - call( - metadata=metadata['metadata'], - message="Failed to create schedules: \n test_namespace/test_schedule_name_to_create_error: test_error", notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES", attachment=input_data['created_schedules'] ), call( metadata=metadata['metadata'], - message="Failed to update schedules: \n test_namespace/test_schedule_name_to_update_error: test_error", + message="Updated schedules: \n test_namespace/test_schedule_name_to_update", notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES", attachment=input_data['updated_schedules'] ), call( metadata=metadata['metadata'], - message="Failed to delete schedules: \n test_namespace/test_schedule_name_to_delete_error: test_error", + message="Deleted schedules: \n test_namespace/test_schedule_name_to_delete", notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES", attachment=input_data['deleted_schedules'] ) ]) + formatters.send_error_report.assert_has_calls([ + call( + metadata=metadata['metadata'], + message="Failed to create schedules: \n test_namespace/test_schedule_name_to_create_error, test_namespace/test_schedule_name_to_create_error2", + notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES_ERROR", + attachment="test_error1\ntest_attachment1\n ========== \ntest_error2" + ), + call( + metadata=metadata['metadata'], + message="Failed to update schedules: \n test_namespace/test_schedule_name_to_update_error, test_namespace/test_schedule_name_to_update_error2, test_namespace/test_schedule_name_to_update_error3", + notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES_ERROR", + attachment="test_error2\ntest_attachment2\n ========== \ntest_error3\ntest_attachment3\n ========== \ntest_error4" + ), + call( + metadata=metadata['metadata'], + message="Failed to delete schedules: \n test_namespace/test_schedule_name_to_delete_error, test_namespace/test_schedule_name_to_delete_error2", + notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES_ERROR", + attachment="test_error4\ntest_attachment4\n ========== \ntest_error5" + ) + ]) @mark.asyncio diff --git a/tests/orchestrator/activities/test_mongo_db.py b/tests/orchestrator/activities/test_mongo_db.py index e3fa686..742d42f 100644 --- a/tests/orchestrator/activities/test_mongo_db.py +++ b/tests/orchestrator/activities/test_mongo_db.py @@ -280,7 +280,7 @@ async def test_update_pipelines_timestamps_success(datetime_mock, mongo_db): {"schedule_name": "test2", "namespace": "test2"} ]}, {"$set": { - "updated_at": datetime_mock.now.return_value.strftime.return_value}} + "updated_at": datetime_mock.now.return_value}} ) @@ -325,9 +325,9 @@ async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db): mongo_db.database["pipelines"].insert_many.assert_called_once_with( [ {"schedule_name": "test1", "namespace": "test1", - "updated_at": datetime_mock.now.return_value.strftime.return_value}, + "updated_at": datetime_mock.now.return_value}, {"schedule_name": "test2", "namespace": "test2", - "updated_at": datetime_mock.now.return_value.strftime.return_value} + "updated_at": datetime_mock.now.return_value} ] ) diff --git a/tests/orchestrator/activities/test_temporal_manager.py b/tests/orchestrator/activities/test_temporal_manager.py index 058aa70..dbb1258 100644 --- a/tests/orchestrator/activities/test_temporal_manager.py +++ b/tests/orchestrator/activities/test_temporal_manager.py @@ -213,21 +213,24 @@ async def test_create_schedule( input_data['schedules']['scouter']['test-schedule'], id="test-schedule", task_queue="test-workflow-queue", - execution_timeout=ANY + execution_timeout=ANY, + typed_search_attributes=mock_typed_search_attributes.return_value, ), call( "test-workflow", input_data['schedules']['scouter']['test-schedule-invalid-frequency'], id="test-schedule-invalid-frequency", task_queue="test-workflow-queue", - execution_timeout=ANY + execution_timeout=ANY, + typed_search_attributes=mock_typed_search_attributes.return_value, ), call( "test-workflow", input_data['schedules']['laborious']['test-schedule-laborious'], id="test-schedule-laborious", task_queue="test-workflow-queue", - execution_timeout=ANY + execution_timeout=ANY, + typed_search_attributes=mock_typed_search_attributes.return_value, ) ]) @@ -546,7 +549,8 @@ async def test_delete_schedules(temporal_manager): "schedule_name": "test-schedule_no_handler", "namespace": "scouter", "success": False, - "message": "Schedule test-schedule_no_handler not found" + "message": "Schedule test-schedule_no_handler not found", + "attachment": ANY }, { "schedule_name": "test-schedule-laborious", From 479166564202cc22cc15b376675f992a2200ec89 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 14 Aug 2025 09:19:48 -0300 Subject: [PATCH 07/11] SIENTIAPDE-1184 SIENTIAPDE-1184 refactor: improve schedule reporting in Formatters class - Introduced a unified reporting structure for created, updated, and deleted schedules. - Enhanced success and error report messages for clarity, including specific schedule types in notifications. - Updated attachment handling in success and error reports to improve context and readability. --- orchestrator/activities/formatters.py | 131 ++++++------------ .../activities/test_formatters.py | 52 +++---- 2 files changed, 73 insertions(+), 110 deletions(-) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index bf6f0cf..a3122fc 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -19,6 +19,8 @@ with workflow.unsafe.imports_passed_through(): from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT from math import ceil +topic_separator = "\n ========== \n" + class Formatters(BaseActivity): def __init__(self, @@ -316,24 +318,25 @@ class Formatters(BaseActivity): return output - def send_success_report(self, metadata: dict[str, Any], message: str, notification_id: str) -> None: + def send_success_report(self, metadata: dict[str, Any], message: str, notification_id: str, attachment: str = None) -> None: self.send_notification( metadata=metadata, notification_id=notification_id, message=message, block="report_orchestration", - level=NotificationLevel.INFO + level=NotificationLevel.INFO, + attachment_content=json.dumps(attachment, indent=4, sort_keys=True) ) def send_error_report(self, metadata: dict[str, Any], message: str, notification_id: str, - attachment: dict[str, Any]) -> None: + attachment: str) -> None: self.send_notification( metadata=metadata, notification_id=notification_id, message=message, block="report_orchestration", level=NotificationLevel.ERROR, - attachment_content=json.dumps(attachment, indent=4, sort_keys=True) + attachment_content=attachment ) def parse_report_schedule(self, input_data: dict[str, Any]) -> tuple[list[str], dict[str, Any]]: @@ -378,91 +381,49 @@ class Formatters(BaseActivity): updated_schedules = input_data['updated_schedules'] deleted_schedules = input_data['deleted_schedules'] - # Send report for created schedules - if len(created_schedules) > 0: - success_keys, error_keys = self.parse_report_schedule( - created_schedules) + schedules_report = { + 'created schedules': { + 'items': created_schedules, + 'id': 'REPORT_ORCHESTRATION_CREATED_SCHEDULES' + }, + 'updated schedules': { + 'items': updated_schedules, + 'id': 'REPORT_ORCHESTRATION_UPDATED_SCHEDULES' + }, + 'deleted schedules': { + 'items': deleted_schedules, + 'id': 'REPORT_ORCHESTRATION_DELETED_SCHEDULES' + } + } - if len(success_keys) > 0: - self.send_success_report( - metadata=metadata, - message=f"Created schedules: \n {', '.join(success_keys)}", - notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES", - attachment=created_schedules - ) + for schedule_type, schedule_data in schedules_report.items(): + if len(schedule_data['items']) > 0: + success_keys, error_keys = self.parse_report_schedule( + schedule_data['items']) - if len(error_keys) > 0: - attachment = [] - for value in error_keys.values(): - if value['attachment'] is not None: - attachment.append( - f"{value['message']}\n{value['attachment']}") - else: - attachment.append(value['message']) + if len(success_keys) > 0: + self.send_success_report( + metadata=metadata, + message=f"Successfully {schedule_type}: \n {', '.join(success_keys)}", + notification_id=schedule_data['id'], + attachment=schedule_data['items'] + ) - self.send_error_report( - metadata=metadata, - message=f"Failed to create schedules: \n {', '.join(error_keys)}", - notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES_ERROR", - attachment="\n ========== \n".join(attachment) - ) + if len(error_keys) > 0: + attachment = [] + for key, value in error_keys.items(): + if value['attachment'] is not None: + attachment.append( + f"{key}:\n{value['message']}\n{value['attachment']}") + else: + attachment.append(f"{key}:\n{value['message']}") - # Send report for updated schedules - if len(updated_schedules) > 0: - success_keys, error_keys = self.parse_report_schedule( - updated_schedules) - - if len(success_keys) > 0: - self.send_success_report( - metadata=metadata, - message=f"Updated schedules: \n {', '.join(success_keys)}", - notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES", - attachment=updated_schedules - ) - - if len(error_keys) > 0: - attachment = [] - for value in error_keys.values(): - if value['attachment'] is not None: - attachment.append( - f"{value['message']}\n{value['attachment']}") - else: - attachment.append(value['message']) - - self.send_error_report( - metadata=metadata, - message=f"Failed to update schedules: \n {', '.join(error_keys)}", - notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES_ERROR", - attachment="\n ========== \n".join(attachment) - ) - - if len(deleted_schedules) > 0: - success_keys, error_keys = self.parse_report_schedule( - deleted_schedules) - - if len(success_keys) > 0: - self.send_success_report( - metadata=metadata, - message=f"Deleted schedules: \n {', '.join(success_keys)}", - notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES", - attachment=deleted_schedules - ) - - if len(error_keys) > 0: - attachment = [] - for value in error_keys.values(): - if value['attachment'] is not None: - attachment.append( - f"{value['message']}\n{value['attachment']}") - else: - attachment.append(value['message']) - - self.send_error_report( - metadata=metadata, - message=f"Failed to delete schedules: \n {', '.join(error_keys)}", - notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES_ERROR", - attachment="\n ========== \n".join(attachment) - ) + self.send_error_report( + metadata=metadata, + message=f"Fails on {schedule_type}: \n {', '.join(error_keys)}", + notification_id=f"{schedule_data['id']}_ERROR", + attachment=topic_separator.join(attachment) + ) @activity.defn(name="report_slot_orchestration") async def report_slot_orchestration(self, diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index c480997..950c70f 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -466,21 +466,6 @@ async def test_create_slot_config(formatters): def test_send_success_report(formatters): formatters.send_success_report( - metadata=metadata, - message="test_message", - notification_id="test_notification_id" - ) - formatters.send_notification.assert_called_once_with( - metadata=metadata, - notification_id="test_notification_id", - message="test_message", - block="report_orchestration", - level=NotificationLevel.INFO - ) - - -def test_send_error_report(formatters): - formatters.send_error_report( metadata=metadata, message="test_message", notification_id="test_notification_id", @@ -491,12 +476,29 @@ def test_send_error_report(formatters): notification_id="test_notification_id", message="test_message", block="report_orchestration", - level=NotificationLevel.ERROR, + level=NotificationLevel.INFO, attachment_content=json.dumps( {"test": "test"}, indent=4, sort_keys=True) ) +def test_send_error_report(formatters): + formatters.send_error_report( + metadata=metadata, + message="test_message", + notification_id="test_notification_id", + attachment="test_attachment" + ) + formatters.send_notification.assert_called_once_with( + metadata=metadata, + notification_id="test_notification_id", + message="test_message", + block="report_orchestration", + level=NotificationLevel.ERROR, + attachment_content="test_attachment" + ) + + def test_parse_report(formatters): input_data = { "test_key": { @@ -631,19 +633,19 @@ async def test_report_schedule_orchestration(formatters): formatters.send_success_report.assert_has_calls([ call( metadata=metadata['metadata'], - message="Created schedules: \n test_namespace/test_schedule_name_to_create", + message="Successfully created schedules: \n test_namespace/test_schedule_name_to_create", notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES", attachment=input_data['created_schedules'] ), call( metadata=metadata['metadata'], - message="Updated schedules: \n test_namespace/test_schedule_name_to_update", + message="Successfully updated schedules: \n test_namespace/test_schedule_name_to_update", notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES", attachment=input_data['updated_schedules'] ), call( metadata=metadata['metadata'], - message="Deleted schedules: \n test_namespace/test_schedule_name_to_delete", + message="Successfully deleted schedules: \n test_namespace/test_schedule_name_to_delete", notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES", attachment=input_data['deleted_schedules'] ) @@ -651,21 +653,21 @@ async def test_report_schedule_orchestration(formatters): formatters.send_error_report.assert_has_calls([ call( metadata=metadata['metadata'], - message="Failed to create schedules: \n test_namespace/test_schedule_name_to_create_error, test_namespace/test_schedule_name_to_create_error2", + message="Fails on created schedules: \n test_namespace/test_schedule_name_to_create_error, test_namespace/test_schedule_name_to_create_error2", notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES_ERROR", - attachment="test_error1\ntest_attachment1\n ========== \ntest_error2" + attachment="test_namespace/test_schedule_name_to_create_error:\ntest_error1\ntest_attachment1\n ========== \ntest_namespace/test_schedule_name_to_create_error2:\ntest_error2" ), call( metadata=metadata['metadata'], - message="Failed to update schedules: \n test_namespace/test_schedule_name_to_update_error, test_namespace/test_schedule_name_to_update_error2, test_namespace/test_schedule_name_to_update_error3", + message="Fails on updated schedules: \n test_namespace/test_schedule_name_to_update_error, test_namespace/test_schedule_name_to_update_error2, test_namespace/test_schedule_name_to_update_error3", notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES_ERROR", - attachment="test_error2\ntest_attachment2\n ========== \ntest_error3\ntest_attachment3\n ========== \ntest_error4" + attachment="test_namespace/test_schedule_name_to_update_error:\ntest_error2\ntest_attachment2\n ========== \ntest_namespace/test_schedule_name_to_update_error2:\ntest_error3\ntest_attachment3\n ========== \ntest_namespace/test_schedule_name_to_update_error3:\ntest_error4" ), call( metadata=metadata['metadata'], - message="Failed to delete schedules: \n test_namespace/test_schedule_name_to_delete_error, test_namespace/test_schedule_name_to_delete_error2", + message="Fails on deleted schedules: \n test_namespace/test_schedule_name_to_delete_error, test_namespace/test_schedule_name_to_delete_error2", notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES_ERROR", - attachment="test_error4\ntest_attachment4\n ========== \ntest_error5" + attachment="test_namespace/test_schedule_name_to_delete_error:\ntest_error4\ntest_attachment4\n ========== \ntest_namespace/test_schedule_name_to_delete_error2:\ntest_error5" ) ]) From 36c34d32087e586b8ecb0390dae7beefba8ff95d Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 14 Aug 2025 09:26:10 -0300 Subject: [PATCH 08/11] SIENTIAPDE-1184 refactor: consolidate report handling in Formatters class - Introduced a new method, manage_and_send_report, to streamline success and error report generation. - Improved code readability by reducing duplication in report handling logic. - Maintained existing functionality while enhancing the structure of report messages. --- orchestrator/activities/formatters.py | 55 ++++++++++++++++----------- 1 file changed, 32 insertions(+), 23 deletions(-) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index a3122fc..e0452b0 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -360,6 +360,31 @@ class Formatters(BaseActivity): return success_keys, error_keys + def manage_and_send_report(self, metadata: dict[str, Any], success_keys: list[str], error_keys: dict[str, Any], schedule_type: str, schedule_data: dict[str, Any]): + if len(success_keys) > 0: + self.send_success_report( + metadata=metadata, + message=f"Successfully {schedule_type}: \n {', '.join(success_keys)}", + notification_id=schedule_data['id'], + attachment=schedule_data['items'] + ) + + if len(error_keys) > 0: + attachment = [] + for key, value in error_keys.items(): + if value['attachment'] is not None: + attachment.append( + f"{key}:\n{value['message']}\n{value['attachment']}") + else: + attachment.append(f"{key}:\n{value['message']}") + + self.send_error_report( + metadata=metadata, + message=f"Fails on {schedule_type}: \n {', '.join(error_keys)}", + notification_id=f"{schedule_data['id']}_ERROR", + attachment=topic_separator.join(attachment) + ) + @activity.defn(name="report_schedule_orchestration") async def report_schedule_orchestration(self, input_data: dict[str, Any]) -> None: @@ -401,29 +426,13 @@ class Formatters(BaseActivity): success_keys, error_keys = self.parse_report_schedule( schedule_data['items']) - if len(success_keys) > 0: - self.send_success_report( - metadata=metadata, - message=f"Successfully {schedule_type}: \n {', '.join(success_keys)}", - notification_id=schedule_data['id'], - attachment=schedule_data['items'] - ) - - if len(error_keys) > 0: - attachment = [] - for key, value in error_keys.items(): - if value['attachment'] is not None: - attachment.append( - f"{key}:\n{value['message']}\n{value['attachment']}") - else: - attachment.append(f"{key}:\n{value['message']}") - - self.send_error_report( - metadata=metadata, - message=f"Fails on {schedule_type}: \n {', '.join(error_keys)}", - notification_id=f"{schedule_data['id']}_ERROR", - attachment=topic_separator.join(attachment) - ) + self.manage_and_send_report( + metadata=metadata, + success_keys=success_keys, + error_keys=error_keys, + schedule_type=schedule_type, + schedule_data=schedule_data + ) @activity.defn(name="report_slot_orchestration") async def report_slot_orchestration(self, From bfb633c9d7e3b98f94ea35b44e79554ce87c4da1 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 14 Aug 2025 09:29:48 -0300 Subject: [PATCH 09/11] SIENTIAPDE-1184 docs: enhance docstrings in Formatters class - Improved clarity and consistency of docstrings across multiple methods in the Formatters class. - Added detailed descriptions for method arguments and return values to facilitate better understanding and usage. - Ensured that all public methods now have comprehensive documentation, enhancing maintainability and readability. --- orchestrator/activities/formatters.py | 95 +++++++++++++++++++++++++-- 1 file changed, 88 insertions(+), 7 deletions(-) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index e0452b0..0bfbc15 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -168,11 +168,13 @@ class Formatters(BaseActivity): async def format_schedule_config(self, input_data: dict[str, Any]) -> dict[str, Any]: """ Formats the schedule config to a dictionary with the schedule name as the key. - input_data: - - schedule_config (list[dict[str, Any]]): The schedule config to format. + + Args: + input_data (dict[str, Any]): The input data containing the schedule config to format. + - schedule_config (list[dict[str, Any]]): The schedule config to format. Returns: - - dict[str, Any]: The formatted schedule config. + dict[str, Any]: The formatted schedule config. """ schedule_config = input_data['schedule_config'] @@ -194,7 +196,16 @@ class Formatters(BaseActivity): to_update: dict[str, Any], to_create: dict[str, Any], namespace: str, metadata: dict[str, Any]): """ - Compares the timestamps of the schedule and the current schedule. + Compares the timestamps of the schedule and the current schedule to determine + which schedules need to be updated or created. + + Args: + schedules (dict[str, Any]): The new schedules to compare. + current_schedules (dict[str, Any]): The existing schedules to compare against. + to_update (dict[str, Any]): Dictionary to populate with schedules that need updating. + to_create (dict[str, Any]): Dictionary to populate with schedules that need creating. + namespace (str): The namespace for the schedules. + metadata (dict[str, Any]): Metadata for logging purposes. """ for schedule_name, schedule in schedules.items(): if schedule_name in current_schedules: @@ -319,6 +330,15 @@ class Formatters(BaseActivity): return output def send_success_report(self, metadata: dict[str, Any], message: str, notification_id: str, attachment: str = None) -> None: + """ + Sends a success notification report. + + Args: + metadata (dict[str, Any]): Metadata for the notification. + message (str): The success message to send. + notification_id (str): The ID of the notification. + attachment (str, optional): Optional attachment content for the notification. + """ self.send_notification( metadata=metadata, notification_id=notification_id, @@ -330,6 +350,15 @@ class Formatters(BaseActivity): def send_error_report(self, metadata: dict[str, Any], message: str, notification_id: str, attachment: str) -> None: + """ + Sends an error notification report. + + Args: + metadata (dict[str, Any]): Metadata for the notification. + message (str): The error message to send. + notification_id (str): The ID of the notification. + attachment (str): The attachment content for the notification. + """ self.send_notification( metadata=metadata, notification_id=notification_id, @@ -340,6 +369,19 @@ class Formatters(BaseActivity): ) def parse_report_schedule(self, input_data: dict[str, Any]) -> tuple[list[str], dict[str, Any]]: + """ + Parses the report schedule data to extract success and error information. + + Args: + input_data (dict[str, Any]): The input data containing schedule reports. + Each item should have 'namespace', 'schedule_name', 'success', 'message', + and optionally 'attachment' fields. + + Returns: + tuple[list[str], dict[str, Any]]: A tuple containing: + - List of successful schedule keys in format "namespace/schedule_name" + - Dictionary of error keys mapped to their error details + """ success_keys = [f"{value['namespace']}/{value['schedule_name']}" for value in input_data if value['success']] @@ -352,6 +394,18 @@ class Formatters(BaseActivity): return success_keys, error_keys def parse_report(self, input_data: dict[str, Any]) -> tuple[list[str], list[str]]: + """ + Parses the report data to extract success and error keys. + + Args: + input_data (dict[str, Any]): The input data containing report items. + Each item should have a 'success' field indicating success/failure. + + Returns: + tuple[list[str], list[str]]: A tuple containing: + - List of successful keys + - List of error keys + """ success_keys = [key for key, value in input_data.items() if value['success']] @@ -361,6 +415,16 @@ class Formatters(BaseActivity): return success_keys, error_keys def manage_and_send_report(self, metadata: dict[str, Any], success_keys: list[str], error_keys: dict[str, Any], schedule_type: str, schedule_data: dict[str, Any]): + """ + Manages and sends success and error reports based on the provided keys and data. + + Args: + metadata (dict[str, Any]): Metadata for logging and notifications. + success_keys (list[str]): List of keys that were successful. + error_keys (dict[str, Any]): Dictionary of error keys mapped to error details. + schedule_type (str): The type of schedule being reported (e.g., 'created schedules'). + schedule_data (dict[str, Any]): The schedule data containing items and notification ID. + """ if len(success_keys) > 0: self.send_success_report( metadata=metadata, @@ -494,8 +558,15 @@ class Formatters(BaseActivity): 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. + + Args: + input_data (dict[str, Any]): The input data containing: + - receiver_groups (dict): The receiver groups configuration. + - mail_type (str): The type of mail for the report. + - metadata (dict): Metadata for logging purposes. + + Returns: + dict[str, Any]: The formatted log report as a dictionary representation of a DataFrame. """ metadata = input_data["metadata"] mail_type = input_data["mail_type"] @@ -540,7 +611,17 @@ class Formatters(BaseActivity): @activity.defn(name="filter_notification_reports") async def filter_notification_reports(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Filter notification reports. + Filters notification reports based on sending configurations and notification package. + + Args: + input_data (dict[str, Any]): The input data containing: + - metadata (dict): Metadata for logging purposes. + - notification_package (list): The package of notifications to filter. + - sending_configs (list): The configurations for sending notifications. + Each config should have 'group_name', 'contents', and optionally 'ignore' fields. + + Returns: + dict[str, Any]: The filtered receiver groups with their notifications. """ metadata = input_data['metadata'] notification_package = input_data['notification_package'] From 31f2ff2439f37492712ae4ea6855e18eeff4c0b1 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 14 Aug 2025 09:48:19 -0300 Subject: [PATCH 10/11] SIENTIAPDE-1184 refactor: enhance shutdown procedures and documentation across activities - Added shutdown methods to MongoDB and Email classes to ensure proper resource cleanup. - Updated docstrings for shutdown methods to clarify their purpose and functionality. - Enhanced documentation for various methods across multiple classes, improving clarity on parameters and return values. - Improved the main function and other utility functions with detailed docstrings for better understanding and maintainability. --- orchestrator/activities/activities.py | 5 + orchestrator/activities/email.py | 16 ++- orchestrator/activities/mongo_db.py | 12 ++- orchestrator/activities/slot_manager.py | 38 ++++++- orchestrator/activities/temporal_manager.py | 4 + orchestrator/utils/connectors_config.py | 36 +++++++ orchestrator/utils/email_builder.py | 32 +++++- orchestrator/utils/orchestrator_functions.py | 105 +++++++++++++++++++ orchestrator/worker/worker.py | 13 +++ 9 files changed, 255 insertions(+), 6 deletions(-) diff --git a/orchestrator/activities/activities.py b/orchestrator/activities/activities.py index 2bf0388..ff36b17 100644 --- a/orchestrator/activities/activities.py +++ b/orchestrator/activities/activities.py @@ -82,4 +82,9 @@ class Activities( # Couchbase, notification_handler=notification_handler) def shutdown(self): + """ + Shutdown the MongoDB connection and clean up resources. + """ MongoDB.shutdown(self) + Postgres.close(self) + Email.shutdown(self) diff --git a/orchestrator/activities/email.py b/orchestrator/activities/email.py index e47ec1e..1401876 100644 --- a/orchestrator/activities/email.py +++ b/orchestrator/activities/email.py @@ -42,6 +42,12 @@ class Email(BaseActivity): logger=logger, notification_handler=notification_handler) + def shutdown(self): + """ + Shutdown the Email connection and clean up resources. + """ + self.server.quit() + @activity.defn(name="build_email_html") async def build_email_html(self, input_data: dict[str, Any]) -> dict[str, Any]: """ @@ -105,9 +111,15 @@ class Email(BaseActivity): def try_send_email(self, msg: MIMEMultipart, receivers: str): """ - Sends an email to the receivers. - """ + Sends an email to the receivers with automatic reconnection handling. + Args: + msg (MIMEMultipart): The email message to send. + receivers (str): Comma-separated list of email addresses to send to. + + Raises: + Exception: If email sending fails after reconnection attempts. + """ try: self.server.sendmail( self.sender_email, receivers, msg.as_string()) diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 80acd0c..be0d40f 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -64,7 +64,7 @@ class MongoDB(BaseActivity): def shutdown(self): """ - Close the MongoDB client connection. + Shutdown the MongoDB connection and clean up resources. """ try: if self.client: @@ -81,6 +81,16 @@ class MongoDB(BaseActivity): self.shutdown() def find(self, collection_name: str, filters: dict[str, Any]) -> list[dict[str, Any]]: + """ + Find documents in a MongoDB collection based on the provided filters. + + Args: + collection_name (str): The name of the collection to search in. + filters (dict[str, Any]): The query filters to apply. + + Returns: + list[dict[str, Any]]: List of documents matching the filters, with _id fields removed. + """ collection = self.database[collection_name] documents = list(collection.find(filters, {"_id": 0})) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index 18e96d0..14aa70c 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -202,6 +202,14 @@ class SlotManager(Redis): async def get_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None: """ Gets the last data timestamp from redis. + + Args: + input_data (dict[str, Any]): The input data containing: + - metadata (dict): Metadata for logging purposes. + - mail_type (str): The type of mail to get timestamp for. + + Returns: + str | None: The last data timestamp as a string, or None if no timestamp exists. """ metadata = input_data['metadata'] key = f"notification_last_timestamp:{input_data['mail_type']}" @@ -233,6 +241,15 @@ class SlotManager(Redis): async def put_last_data_timestamp(self, input_data: dict[str, Any]): """ Puts the last data timestamp into redis. + + Args: + input_data (dict[str, Any]): The input data containing: + - metadata (dict): Metadata for logging purposes. + - data (list[dict]): The data to extract timestamp from. + - mail_type (str): The type of mail to store timestamp for. + + Returns: + str | None: The last data timestamp that was stored, or None if no data exists. """ metadata = input_data['metadata'] key = f"notification_last_timestamp:{input_data['mail_type']}" @@ -270,7 +287,18 @@ class SlotManager(Redis): @activity.defn(name="filter_notification_alerts") async def filter_notification_alerts(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Filter notification alerts + Filter notification alerts based on sending configurations and notification package. + + Args: + input_data (dict[str, Any]): The input data containing: + - metadata (dict): Metadata for logging purposes. + - notification_package (list): The package of notifications to filter. + - sending_configs (list): The configurations for sending notifications. + Each config should have 'group_name', 'contents', and optionally 'ignore' fields. + - notification_ttl (int): Time to live for notifications in seconds. + + Returns: + dict[str, Any]: The filtered receiver groups with their notifications. """ metadata = input_data['metadata'] notification_package = input_data['notification_package'] @@ -330,7 +358,13 @@ class SlotManager(Redis): @activity.defn(name="store_notification_cache") async def store_notification_cache(self, input_data: dict[str, Any]) -> None: """ - Store notification cache + Store notification cache in Redis to track recently sent notifications. + + Args: + input_data (dict[str, Any]): The input data containing: + - metadata (dict): Metadata for logging purposes. + - log_report (list[dict]): The log report containing notification statuses. + - sent_ttl (int): Time to live for sent notification cache in seconds. """ metadata = input_data['metadata'] log_report = DataFrame(input_data['log_report']) diff --git a/orchestrator/activities/temporal_manager.py b/orchestrator/activities/temporal_manager.py index eb7d361..d6f64bf 100644 --- a/orchestrator/activities/temporal_manager.py +++ b/orchestrator/activities/temporal_manager.py @@ -42,6 +42,10 @@ class TemporalManager(BaseActivity): notification_handler=notification_handler) async def connect_to_temporal(self): + """ + Connect to Temporal server namespaces for scouter and laborious workflows. + Creates client connections to both namespaces and stores them for later use. + """ self.logger.info( f"Connecting to Temporal side namespaces at {self.temporal_host}") self.logger.info(f"Scouter namespace: {self.scouter_namespace}") diff --git a/orchestrator/utils/connectors_config.py b/orchestrator/utils/connectors_config.py index 7237a05..8ab0c76 100644 --- a/orchestrator/utils/connectors_config.py +++ b/orchestrator/utils/connectors_config.py @@ -2,6 +2,12 @@ from os import getenv def build_redis_config(): + """ + Build Redis configuration from environment variables. + + Returns: + dict: Redis configuration with host, port, username, and password. + """ return { 'host': getenv('REDIS_HOST', 'localhost'), 'port': int(getenv('REDIS_PORT', '6379')), @@ -11,6 +17,12 @@ def build_redis_config(): def build_mongodb_config(): + """ + Build MongoDB configuration from environment variables. + + Returns: + dict: MongoDB configuration with connection string, database name, and TTL index seconds. + """ username = getenv('MONGODB_USERNAME', 'root') password = getenv('MONGODB_PASSWORD', 'wKZDbMNU1c') uri = getenv('MONGODB_URL', 'localhost:27018') @@ -24,6 +36,12 @@ def build_mongodb_config(): def build_couchbase_config(): + """ + Build Couchbase configuration from environment variables. + + Returns: + dict: Couchbase configuration with connection string, username, and password. + """ return { 'connection_string': getenv('COUCHBASE_CONNECTION_STRING', 'couchbase://localhost'), 'username': getenv('COUCHBASE_USERNAME', 'sientia'), @@ -32,6 +50,12 @@ def build_couchbase_config(): def build_temporal_config(): + """ + Build Temporal configuration from environment variables. + + Returns: + dict: Temporal configuration with host and namespace settings. + """ return { 'temporal_host': getenv('TEMPORAL_HOST', 'localhost:7233'), 'temporal_namespace': getenv('TEMPORAL_NAMESPACE', 'default'), @@ -41,6 +65,12 @@ def build_temporal_config(): def build_postgres_config(): + """ + Build PostgreSQL configuration from environment variables. + + Returns: + dict: PostgreSQL configuration with connection details and connection pool settings. + """ return { 'host': getenv('POSTGRES_HOST', 'localhost'), 'port': int(getenv('POSTGRES_PORT', '5432')), @@ -53,6 +83,12 @@ def build_postgres_config(): def build_email_config(): + """ + Build email configuration from environment variables. + + Returns: + dict: Email configuration with SMTP server settings and sender credentials. + """ return { 'sender_email': getenv('EMAIL_SENDER', 'sientia-alerts@aignosi.com'), 'sender_password': getenv('EMAIL_SENDER_PASSWORD', 'sientia'), diff --git a/orchestrator/utils/email_builder.py b/orchestrator/utils/email_builder.py index 88a765f..64b5963 100644 --- a/orchestrator/utils/email_builder.py +++ b/orchestrator/utils/email_builder.py @@ -18,12 +18,33 @@ class EmailBuilder: self.general_template = file.read() def replace_parameters(self, template: str, parameters: dict) -> str: + """ + Replace parameters in a Jinja2 template with provided values. + + Args: + template (str): The Jinja2 template string. + parameters (dict): Dictionary of parameters to replace in the template. + + Returns: + str: The rendered template with parameters replaced. + """ # Criar um template Jinja2 template = Template(template) return template.render(parameters) def parameters(self, general_events: dict, mail_type: str) -> dict: + """ + Build parameters dictionary for email templates based on general events and mail type. + + Args: + general_events (dict): Dictionary containing events categorized by level (ERROR, WARNING, INFO). + Each level contains a 'models' key with model-specific event data. + mail_type (str): The type of email being sent. + + Returns: + dict: Dictionary with mail_type and rendered event sections for each notification level. + """ error_models = general_events.get('ERROR', {}).get('models', []) warning_models = general_events.get('WARNING', {}).get('models', []) info_models = general_events.get('INFO', {}).get('models', []) @@ -43,7 +64,16 @@ class EmailBuilder: def build_email(self, report_data: list[dict], mail_type: str) -> str: """ - Builds the email html. + Builds the email HTML by organizing report data by notification level and model. + + Args: + report_data (list[dict]): List of notification reports, each containing: + - level (str): Notification level (ERROR, WARNING, INFO) + - model_name (str): Name of the model + - Additional notification details + + Returns: + str: Complete HTML email content ready for sending. """ general_events = {} diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 7b6289f..38bde51 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -2,6 +2,21 @@ from typing import Any def common_config(config: dict[str, Any]): + """ + Extract common configuration parameters from a pipeline configuration. + + Args: + config (dict[str, Any]): Pipeline configuration containing: + - workflow_type (str): Type of workflow (e.g., 'scouter', 'predictions_batch') + - schedule_name (str): Name of the schedule + - frequency (str, optional): Frequency of execution (default: '1m') + - max_retry_policy (int, optional): Maximum retry attempts (default: 1) + - model_id (str): ID of the model + - models (dict): Model configuration containing 'name' field + + Returns: + dict[str, Any]: Common configuration dictionary with extracted parameters. + """ return { "workflow_type": config['workflow_type'], "schedule_name": config['schedule_name'], @@ -14,6 +29,18 @@ def common_config(config: dict[str, Any]): def minimal_retrain(config: dict[str, Any]): + """ + Build minimal retrain configuration from pipeline config. + + Args: + config (dict[str, Any]): Pipeline configuration containing: + - schedule_name (str): Name of the schedule + - query (str): SQL query for retraining + - Additional fields from common_config + + Returns: + dict[str, Any]: Minimal retrain configuration with workflow type set to 'minimal_retrain'. + """ return { **common_config(config), "workflow_type": "minimal_retrain", @@ -25,6 +52,25 @@ def minimal_retrain(config: dict[str, Any]): def scouter(config: dict[str, Any]): + """ + Build scouter configuration from pipeline config. + + Args: + config (dict[str, Any]): Pipeline configuration containing: + - filters (list[dict], optional): List of filter configurations + - read_tags (list[dict]): List of tag configurations with: + - filter_name (str): Name of the filter + - policy (str): Filter policy + - tag_name (str): Name of the tag + - aggr_func (str, optional): Aggregation function (default: 'lts') + - data_range (list[int], optional): Data range limits (default: [-100, 100]) + - tag_retention_minutes (int, optional): Tag retention time in minutes (default: 60) + - debug_data_package (bool, optional): Enable debug data package (default: False) + - Additional fields from common_config + + Returns: + dict[str, Any]: Scouter configuration with topic, filters, tags, and retention settings. + """ filters = {} for f in config.get('filters', []): filters[f['filter_name']] = { @@ -53,6 +99,19 @@ def scouter(config: dict[str, Any]): def overlap_filter_config(base_filter_config: dict[str, Any], config: list[dict[str, Any]]): + """ + Overlap filter configuration with base filter config. + + Args: + base_filter_config (dict[str, Any]): Base filter configuration to extend. + config (list[dict[str, Any]]): List of filter configurations to add, each containing: + - filter_name (str): Name of the filter + - policy (str): Filter policy + - config (dict, optional): Additional filter configuration + + Returns: + dict[str, Any]: Extended filter configuration with new filters added. + """ for fil in config: base_filter_config[fil['filter_name']] = { "policy": fil['policy'], @@ -63,6 +122,15 @@ def overlap_filter_config(base_filter_config: dict[str, Any], config: list[dict[ def process_path_priority(path_priority: list[str]): + """ + Process and normalize path priority list to ensure it contains the required priorities. + + Args: + path_priority (list[str]): List of path priorities to process. + + Returns: + list[str]: Normalized path priority list with exactly 3 elements: ["STOP", "CONTINUE", "REPEAT"]. + """ for priority in path_priority[:]: if priority not in ["STOP", "CONTINUE", "REPEAT"]: path_priority.remove(priority) @@ -75,6 +143,26 @@ def process_path_priority(path_priority: list[str]): def predictions_batch(config: dict[str, Any]): + """ + Build predictions batch configuration from pipeline config. + + Args: + config (dict[str, Any]): Pipeline configuration containing: + - write_tags (list[dict]): List of tag configurations with: + - server_id (str): ID of the OPC server + - type (str): Tag type ('prediction' or 'confidence') + - addr (str): Tag address + - data_type (str, optional): Data type (default: 'float') + - path_priority (list[str], optional): List of path priorities (default: ["STOP", "CONTINUE", "REPEAT"]) + - input_filters (list[dict], optional): List of input filter configurations + - mlflow_transform_filters (list[dict], optional): List of MLflow transform filter configurations + - mlflow_predict_filters (list[dict], optional): List of MLflow predict filter configurations + - model_retention_minutes (int, optional): Model retention time in minutes (default: 60) + - Additional fields from common_config + + Returns: + dict[str, Any]: Predictions batch configuration with OPC output config, filters, and path priority. + """ tags = {} for tag in config.get('write_tags', []): if tag['server_id'] not in tags: @@ -156,6 +244,23 @@ def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]: def build_tag_config(tag: dict[str, Any], slot_config: dict[str, Any], opc_servers: dict[str, Any], i: int): + """ + Build tag configuration for a specific slot and OPC server. + + Args: + tag (dict[str, Any]): Tag configuration containing: + - server_id (str): ID of the OPC server + - tag_address (str): Address of the tag + slot_config (dict[str, Any]): Current slot configuration to update. + opc_servers (dict[str, Any]): Dictionary of OPC server configurations. + i (int): Slot number to configure. + + Returns: + dict[str, Any]: Updated slot configuration with the new tag. + + Raises: + ValueError: If the specified server_id is not found in opc_servers. + """ server_id = tag['server_id'] if server_id not in opc_servers: diff --git a/orchestrator/worker/worker.py b/orchestrator/worker/worker.py index 89959e3..49fc31e 100644 --- a/orchestrator/worker/worker.py +++ b/orchestrator/worker/worker.py @@ -29,6 +29,13 @@ POD_ID = os.getenv("POD_ID") async def main(): + """ + Main function to initialize and run the Temporal worker. + + Sets up MongoDB connection, notification handler, Temporal client, and starts + multiple workers for different task queues (orchestrator, alerts, reports). + Handles graceful shutdown and error handling. + """ host = os.getenv('TEMPORAL_HOST', 'localhost:7233') namespace = os.getenv('TEMPORAL_NAMESPACE', 'default') logger = get_logger(__name__) @@ -176,6 +183,12 @@ async def main(): def start_prometheus_server(): + """ + Start the Prometheus metrics server on the configured port. + + Sets up HTTP server for metrics collection and marks the application as UP. + Exits the application if the server fails to start. + """ try: port = int(os.getenv("HTTP_METRICS_PORT", 9090)) start_http_server(port) From 229dcb1ac87e63f9d8e5decbea965ed58fb395c0 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 14 Aug 2025 09:58:51 -0300 Subject: [PATCH 11/11] SIENTIAPDE-1184 docs: update README.md to provide comprehensive overview of SIENTIA DataOps Orchestrator Temporal - Added detailed project overview, goals, and architecture sections. - Documented main workflows including orchestrator, alerts, and reports workflows with input parameters and functionalities. - Included environment variable configurations for Redis, MongoDB, PostgreSQL, Couchbase, email, Temporal, application, and Kafka. - Described deployment strategies and monitoring capabilities, including Prometheus metrics. - Enhanced usage instructions for triggering workflows and logging operations. --- README.md | 188 +++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 186 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 01ebf1d..160ec29 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,190 @@ -#PR shortcut +# SIENTIA DataOps Orchestrator Temporal + +## Overview + +The SIENTIA DataOps Orchestrator Temporal is a comprehensive workflow orchestration system built on Temporal.io that manages data pipelines, notifications, and system orchestration for the SIENTIA platform. It provides automated scheduling, monitoring, and execution of data processing workflows with integrated alerting and reporting capabilities. + +## Project Goals + +- **Pipeline Orchestration**: Automate the deployment and management of data processing pipelines +- **Notification Management**: Handle real-time alerts and scheduled reports for system events +- **Resource Management**: Manage OPC server slots and data ingestion resources +- **Workflow Automation**: Coordinate complex workflows across multiple services and databases +- **Monitoring & Reporting**: Provide comprehensive logging and metrics for system health + +## Architecture + +The system is built around three main worker queues, each handling specific types of workflows: + +### 1. Orchestrator Queue (`orchestrator-queue`) +Handles pipeline orchestration and resource management workflows. + +### 2. Alerts Queue (`alerts-queue`) +Manages real-time alert notifications and error reporting. + +### 3. Reports Queue (`reports-queue`) +Handles scheduled reports and data summaries. + +## Workflows + +### Main Workflows + +#### 1. Orchestrator Workflow +**Purpose**: Main orchestration workflow that manages pipeline deployment and resource allocation. + +**Input Parameters**: +- `schedule_name` (str): Name of the orchestration schedule +- `pipelines_query` (dict): MongoDB query to retrieve pipeline configurations +- `opc_servers_query` (dict): MongoDB query to retrieve OPC server configurations + +**What it does**: +- Retrieves pipeline configurations from MongoDB +- Loads current OPC server slots and active ingestors from Redis +- Processes schedules and creates slot configurations +- Deploys schedules to Temporal server (scouter and laborious namespaces) +- Updates OPC slots in Redis +- Generates orchestration reports + +#### 2. Alerts Workflow +**Purpose**: Sends real-time error alerts to configured user groups. + +**Input Parameters**: +- `schedule_name` (str): Name of the alert schedule +- `notification_ttl` (int): Time period before considering notifications persistent +- `sent_ttl` (int): Time to live for sent notification cache + +**What it does**: +- Filters notifications by ERROR level +- Loads notification packages from MongoDB +- Applies user group filtering and notification TTL rules +- Sends HTML email alerts +- Stores notification logs in PostgreSQL +- Caches sent notifications to prevent duplicates + +#### 3. Reports Workflow +**Purpose**: Sends scheduled reports to configured user groups. + +**Input Parameters**: +- `schedule_name` (str): Name of the report schedule + +**What it does**: +- Loads all notifications (any level) from MongoDB +- Applies user group filtering +- Generates HTML report emails +- Stores report logs in PostgreSQL + +### Subworkflows + +#### 1. Load Notification Package +**Purpose**: Loads notification data and configuration from various sources. + +**Input Parameters**: +- `metadata` (dict): Workflow metadata +- `mail_type` (str): Type of mail (Alerts/Reports) +- `base_data_filter` (dict): Base filters for data retrieval + +**Returns**: +- `last_timestamp` (str): Last processed timestamp +- `notification_package` (list): Package of notifications to process +- `sending_configs` (list): Email sending configurations + +#### 2. Process Notifications +**Purpose**: Processes notifications and sends emails with logging. + +**Input Parameters**: +- `metadata` (dict): Workflow metadata +- `mail_type` (str): Type of mail being sent +- `schema` (str): Database schema name +- `table_name` (str): Database table name +- `notification_package` (list): Notifications to process + +**Returns**: +- `log_report` (dict): Report of processed notifications + +## Environment Variables + +### Database Connections + +#### Redis Configuration +- `REDIS_HOST`: Redis server hostname (default: localhost) +- `REDIS_PORT`: Redis server port (default: 6379) +- `REDIS_USERNAME`: Redis username (default: default) +- `REDIS_PASSWORD`: Redis password (from secret) + +#### MongoDB Configuration +- `MONGODB_USERNAME`: MongoDB username (default: root) +- `MONGODB_PASSWORD`: MongoDB password +- `MONGODB_URL`: MongoDB server URL (default: localhost:27017) +- `MONGODB_DATABASE`: Database name (default: sientia) +- `MONGODB_TTL_INDEX_HOURS`: TTL index duration in hours (default: 1) + +#### PostgreSQL Configuration +- `POSTGRES_HOST`: PostgreSQL server hostname +- `POSTGRES_PORT`: PostgreSQL server port (default: 5432) +- `POSTGRES_USER`: Database username (default: sientia) +- `POSTGRES_PASSWORD`: Database password (default: sientia) +- `POSTGRES_DBNAME`: Database name (default: sientia) +- `POSTGRES_MIN_CONNECTIONS`: Minimum connection pool size (default: 10) +- `POSTGRES_MAX_CONNECTIONS`: Maximum connection pool size (default: 40) + +#### Couchbase Configuration +- `COUCHBASE_CONNECTION_STRING`: Couchbase server connection string +- `COUCHBASE_USERNAME`: Couchbase username (default: sientia) +- `COUCHBASE_PASSWORD`: Couchbase password (default: sientia) + +### Email Configuration +- `EMAIL_SENDER`: Sender email address +- `EMAIL_SENDER_PASSWORD`: App password for SMTP authentication +- `EMAIL_SMTP_SERVER`: SMTP server hostname (default: smtp.gmail.com) +- `EMAIL_SMTP_PORT`: SMTP server port (default: 587) + +### Temporal Configuration +- `TEMPORAL_HOST`: Temporal server hostname and port +- `TEMPORAL_NAMESPACE`: Default Temporal namespace (default: default) +- `TEMPORAL_SCOUTER_NAMESPACE`: Scouter workflow namespace (default: scouter) +- `TEMPORAL_LABORIOUS_NAMESPACE`: Laborious workflow namespace (default: laborious) + +### Application Configuration +- `LOG_LEVEL`: Logging level (default: DEBUG) +- `HTTP_METRICS_PORT`: Prometheus metrics port (default: 9090) +- `PROJECT_NAME`: Project identifier (default: sientia-orchestrator) +- `POD_ID`: Kubernetes pod identifier for metrics + +### Kafka Configuration +- `KAFKA_BOOTSTRAP_SERVERS`: Kafka bootstrap servers + +## Deployment + +The system is designed for Kubernetes deployment using Helm charts with: +- Health checks and readiness probes +- Prometheus metrics endpoint +- ServiceMonitor integration for Prometheus Operator +- Configurable resource limits and scaling +- SSH key management for Git operations + +## Usage + +Workflows can be triggered via Temporal client calls with appropriate input parameters. The system automatically handles: +- Pipeline configuration retrieval +- Resource allocation +- Schedule deployment +- Notification processing +- Email delivery +- Logging and monitoring + +## Monitoring + +The system exposes Prometheus metrics at `/metrics` endpoint including: +- Application status (up/down) +- Email sent counts +- Workflow execution metrics +- Custom business metrics + +All operations are logged with structured metadata for debugging and auditing purposes. + +# PR shortcut ``` git log origin/main..HEAD --no-merges > git_log ``` Prompt: -Write a summary of PR changes in markdown. Be objective and direct. Write to file \ No newline at end of file +Write a summary of PR changes in markdown. Be objective and direct. Write to file