From dcbcc7065166859e46f7836caffea2fd5a63ba46 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 4 Jun 2025 15:06:54 -0300 Subject: [PATCH] fix: update search attribute keys and improve schedule handling - Changed search attribute key from "Orchestrated" to "orchestrated" in temporal_manager.py and test cases. - Enhanced logging for schedule creation and updates in temporal_manager.py. - Updated schedule creation to use workflow_type directly instead of a hardcoded string. - Modified gather_read_tags function to use server_id instead of server_name for tag identification. - Adjusted test cases to reflect changes in server_id usage and ensure consistency across tests. - Fixed model_name retrieval in common_config to access nested models dictionary. - Updated test cases to align with new data structures and ensure accurate assertions. --- input_sample.json | 35 +----- orchestrator/activities/formatters.py | 38 +++++- orchestrator/activities/temporal_manager.py | 20 +++- orchestrator/utils/orchestrator_functions.py | 14 ++- test.ipynb | 111 +++++++----------- .../activities/test_formatters.py | 38 ++++-- .../activities/test_slot_manager.py | 9 +- .../activities/test_temporal_manager.py | 18 +-- .../utils/test_orchestrator_functions.py | 32 +++-- 9 files changed, 176 insertions(+), 139 deletions(-) diff --git a/input_sample.json b/input_sample.json index 9fffe37..4a89997 100644 --- a/input_sample.json +++ b/input_sample.json @@ -1,34 +1,5 @@ { - "schedule_name": "scouter-opcua-pipeline", - "model_name": "Demo Model", - "model_id": 1, - "query": "SELECT * FROM sientia_data.laborious_data order by \"timestamp\" desc limit 30;", - "schema": "sientia_data", - "table_name": "predictions", - "retention_time": 3600, - "model_retention": 120, - "path_priority": ["STOP", "CONTINUE", "REPEAT"], - "input_filters": { - "SPECIFIC_VARIABLES_NULL_VALUES": { - "POLICY": "STOP", - "VARIABLES": ["Counter"] - }, - "EMPTY_DATA": { - "POLICY": "STOP" - } - }, - "mlflow_transform_filters": { - "API_ERROR": { - "POLICY": "CONTINUE" - }, - "NAN_VALUES": { - "POLICY": "CONTINUE" - } - }, - "mlflow_predict_filters": { - "API_ERROR": { - "POLICY": "CONTINUE" - } - }, - "opc_output_config": {} + "schedule_name": "orchestrator-test", + "pipelines_query": "SELECT pipelines.*, models FROM `pipelines` JOIN `models` ON KEYS pipelines.model_id;", + "opc_servers_query": "select META().id, opc_servers.* from `opc_servers`;" } \ No newline at end of file diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index d88bd5b..8a06865 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -46,6 +46,11 @@ class Formatters(BaseActivity): schedule_config[pipeline['schedule_name'] ] = predictions_batch(pipeline) + self.logger.info("Processed schedules") + + self.logger.debug(json.dumps( + schedule_config, indent=4, sort_keys=True)) + return schedule_config @activity.defn(name="process_slots") @@ -74,7 +79,7 @@ class Formatters(BaseActivity): opc_servers = {} for server in opc_servers_list: - opc_servers[server['server_name']] = { + opc_servers[server['id']] = { **server, } @@ -99,6 +104,10 @@ class Formatters(BaseActivity): slot_config = build_tag_config( tag, slot_config.copy(), opc_servers, number_of_slots) + self.logger.info("Processed slots") + self.logger.debug(json.dumps( + slot_config, indent=4, sort_keys=True)) + return slot_config @activity.defn(name="create_schedule_config") @@ -131,7 +140,16 @@ class Formatters(BaseActivity): for schedule_name, schedule in schedule_config.items(): if schedule_name in current_schedule_config: - if schedule != current_schedule_config[schedule_name]['data']: + self.logger.debug(f"{current_schedule_config[schedule_name]}") + old_config = current_schedule_config[schedule_name]['data'] + + self.logger.debug(f"Comparing {schedule_name}:") + self.logger.debug(json.dumps( + old_config, indent=4, sort_keys=True)) + self.logger.debug(json.dumps( + schedule, indent=4, sort_keys=True)) + + if schedule != old_config: to_update[schedule_name] = schedule elif schedule_name not in current_schedule_config: @@ -141,12 +159,18 @@ class Formatters(BaseActivity): if schedule_name not in schedule_config: to_delete.append(schedule_name) - return { + output = { "to_update": to_update, "to_create": to_create, "to_delete": to_delete } + self.logger.info("Created schedule config") + self.logger.debug(json.dumps( + output, indent=4, sort_keys=True)) + + return output + @activity.defn(name="create_slot_config") async def create_slot_config(self, input_data: dict[str, Any]) -> dict[str, Any]: @@ -179,11 +203,17 @@ class Formatters(BaseActivity): to_delete = [str(i) for i in range( number_of_slots + 1, number_of_current_slots + 1)] - return { + output = { "to_delete": to_delete, "to_insert": slot_config } + self.logger.info("Created slot config") + self.logger.debug(json.dumps( + output, indent=4, sort_keys=True)) + + return output + def send_success_report(self, message: str, notification_id: str) -> None: self.notification_handler.build_and_send_notification( notification_id, diff --git a/orchestrator/activities/temporal_manager.py b/orchestrator/activities/temporal_manager.py index 71fc641..7fd520e 100644 --- a/orchestrator/activities/temporal_manager.py +++ b/orchestrator/activities/temporal_manager.py @@ -49,7 +49,7 @@ class TemporalManager(BaseActivity): async for schedule in await self.temporal_client.list_schedules(): search_attrs = getattr(schedule, "search_attributes", {}) - if search_attrs.get("Orchestrated", ["false"]) == ["true"]: + if search_attrs.get("orchestrated", ["false"]) == ["true"]: schedule_id = schedule.id handle = self.temporal_client.get_schedule_handle(schedule_id) @@ -113,14 +113,18 @@ class TemporalManager(BaseActivity): workflow_type = schedule['workflow_type'] try: + self.logger.debug(f"Creating schedule {schedule_name}:") + self.logger.debug(json.dumps( + schedule, indent=4, sort_keys=True)) await self.temporal_client.create_schedule( schedule_name, Schedule( action=ScheduleActionStartWorkflow( - workflow=workflow_type, - args=schedule, + workflow_type, + schedule, id=schedule_name, - task_queue=f"{workflow_type}-queue" + task_queue=f"{workflow_type}-queue", + execution_timeout=timedelta(minutes=2) ), spec=ScheduleSpec( intervals=[ @@ -181,8 +185,14 @@ class TemporalManager(BaseActivity): async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate: schedule_action = input_data.description.schedule.action + self.logger.debug("Updating schedule:") + if hasattr(schedule_action, "args"): - schedule_action.args = schedule + self.logger.debug("New schedule:") + self.logger.debug(json.dumps( + schedule, indent=4, sort_keys=True)) + + schedule_action.args = [schedule] input_data.description.schedule.spec.intervals = [ ScheduleIntervalSpec( diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 5e57968..22e5444 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -9,7 +9,7 @@ def common_config(config: dict[str, Any]): "max_retry_policy": config.get('max_retry_policy', 1), "model_id": config['model_id'], - "model_name": config['model_name'], + "model_name": config['models']['name'], } @@ -129,7 +129,7 @@ def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]: # Get all read tags from pipelines for pipeline in pipelines: for tag in pipeline['read_tags']: - tag_string = f"{tag['server_name']}:{tag['tag_address']}" + tag_string = f"{tag['server_id']}:{tag['tag_address']}" if tag_string not in tags: tags[tag_string] = { **tag, @@ -144,15 +144,17 @@ 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): - server_name = tag['server_name'] + server_id = tag['server_id'] + server_name = opc_servers[server_id]['server_name'] if server_name not in slot_config[f"{i}"]: slot_config[f"{i}"][server_name] = { + "server_id": server_id, "name": server_name, - "url": opc_servers[server_name]['url'], - "server_uri": opc_servers[server_name]['uri'], + "url": opc_servers[server_id]['url'], + "server_uri": opc_servers[server_id]['uri'], "tags": {} } - for name, spec in opc_servers[server_name].get('security_spec', {}).items(): + for name, spec in opc_servers[server_id].get('security_spec', {}).items(): slot_config[f"{i}"][server_name][name] = spec slot_config[f"{i}"][server_name]["tags"][tag['tag_address']] = { diff --git a/test.ipynb b/test.ipynb index 9683267..ecf11d1 100644 --- a/test.ipynb +++ b/test.ipynb @@ -136,33 +136,19 @@ }, { "cell_type": "code", - "execution_count": 2, + "execution_count": 6, "id": "bb750ae6", "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:meu-schedule-id5, RunId: 01971d9e-b19b-7e25-8f60-ba4ed95e12e4.', b'\\x08\\x06\\x12\\x88\\x01Workflow execution is already running. WorkflowId: temporal-sys-scheduler:meu-schedule-id5, RunId: 01971d9e-b19b-7e25-8f60-ba4ed95e12e4.\\x1a\\xa7\\x01\\nWtype.googleapis.com/temporal.api.errordetails.v1.WorkflowExecutionAlreadyStartedFailure\\x12L\\n$6ae80236-ccbf-45b5-8282-1ef14a559b59\\x12$01971d9e-b19b-7e25-8f60-ba4ed95e12e4')", - "\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:meu-schedule-id5, RunId: 01971d9e-b19b-7e25-8f60-ba4ed95e12e4.", - "\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[2]\u001b[39m\u001b[32m, line 17\u001b[39m\n\u001b[32m 13\u001b[39m customer_id_key = SearchAttributeKey.for_keyword(\u001b[33m\"\u001b[39m\u001b[33mOrchestrated\u001b[39m\u001b[33m\"\u001b[39m)\n\u001b[32m 14\u001b[39m search_attributes = TypedSearchAttributes([\n\u001b[32m 15\u001b[39m SearchAttributePair(customer_id_key, \u001b[33m\"\u001b[39m\u001b[33mtrue\u001b[39m\u001b[33m\"\u001b[39m)\n\u001b[32m 16\u001b[39m ])\n\u001b[32m---> \u001b[39m\u001b[32m17\u001b[39m \u001b[38;5;28;01mawait\u001b[39;00m temporal_client.create_schedule(\n\u001b[32m 18\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mmeu-schedule-id5\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 19\u001b[39m Schedule(\n\u001b[32m 20\u001b[39m action=ScheduleActionStartWorkflow(\n\u001b[32m 21\u001b[39m \u001b[33m'\u001b[39m\u001b[33mscouter-test2\u001b[39m\u001b[33m'\u001b[39m,\n\u001b[32m 22\u001b[39m {\n\u001b[32m 23\u001b[39m \u001b[33m'\u001b[39m\u001b[33margs\u001b[39m\u001b[33m'\u001b[39m: {\n\u001b[32m 24\u001b[39m \u001b[33m'\u001b[39m\u001b[33marg1\u001b[39m\u001b[33m'\u001b[39m: \u001b[33m'\u001b[39m\u001b[33mvalue1\u001b[39m\u001b[33m'\u001b[39m\n\u001b[32m 25\u001b[39m }\n\u001b[32m 26\u001b[39m },\n\u001b[32m 27\u001b[39m \u001b[38;5;28mid\u001b[39m=\u001b[33m\"\u001b[39m\u001b[33mworkflow-id-unico\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 28\u001b[39m task_queue=\u001b[33m\"\u001b[39m\u001b[33mnome-da-task-queue\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 29\u001b[39m ),\n\u001b[32m 30\u001b[39m spec=ScheduleSpec(\n\u001b[32m 31\u001b[39m intervals=[ScheduleIntervalSpec(every=timedelta(minutes=\u001b[32m10\u001b[39m))]\n\u001b[32m 32\u001b[39m )\n\u001b[32m 33\u001b[39m ),\n\u001b[32m 34\u001b[39m search_attributes=search_attributes,\n\u001b[32m 35\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" - ] + "data": { + "text/plain": [ + "" + ] + }, + "execution_count": 6, + "metadata": {}, + "output_type": "execute_result" } ], "source": [ @@ -178,7 +164,7 @@ "from temporalio.common import TypedSearchAttributes, SearchAttributeKey, SearchAttributePair\n", "\n", "\n", - "customer_id_key = SearchAttributeKey.for_keyword(\"Orchestrated\")\n", + "customer_id_key = SearchAttributeKey.for_keyword(\"orchestrated\")\n", "search_attributes = TypedSearchAttributes([\n", " SearchAttributePair(customer_id_key, \"true\")\n", "])\n", @@ -213,9 +199,7 @@ "name": "stdout", "output_type": "stream", "text": [ - "Getting orchestrated schedules...\n", - "Found %d orchestrated schedules 5\n", - "Orchestrated schedules: %s {'meu-schedule-id5': {'frequency': 600, 'data': {'args': {'arg1': 'value1'}}, 'handle': }, 'meu-schedule-id4': {'frequency': 600, 'data': {}, 'handle': }, 'meu-schedule-id3': {'frequency': 600, 'data': {}, 'handle': }, 'meu-schedule-id2': {'frequency': 600, 'data': {}, 'handle': }, 'meu-schedule-id': {'frequency': 600, 'data': {}, 'handle': }}\n" + "Getting orchestrated schedules...\n" ] } ], @@ -225,7 +209,22 @@ }, { "cell_type": "code", - "execution_count": 87, + "execution_count": 24, + "id": "1948670e", + "metadata": {}, + "outputs": [], + "source": [ + "from google.protobuf.json_format import MessageToDict\n", + "import base64\n", + "import json\n", + "for arg in schedules.schedule.action.args:\n", + " data = MessageToDict(arg)\n", + " data = base64.b64decode(data['data']).decode('utf-8')" + ] + }, + { + "cell_type": "code", + "execution_count": 12, "id": "a4c777dd", "metadata": {}, "outputs": [], @@ -234,9 +233,8 @@ "import base64\n", "import json\n", "\n", - "\n", "schedules_config = {}\n", - "for schedule in schedules:\n", + "async for schedule in await temporal_client.list_schedules():\n", " id = schedule.id\n", "\n", " handle = temporal_client.get_schedule_handle(id)\n", @@ -262,42 +260,19 @@ }, { "cell_type": "code", - "execution_count": 88, + "execution_count": 13, "id": "988d1718", "metadata": {}, - "outputs": [ - { - "data": { - "text/plain": [ - "{'meu-schedule-id5': {'frequency': 600,\n", - " 'data': {'args': {'arg1': 'value1'}},\n", - " 'handle': },\n", - " 'meu-schedule-id4': {'frequency': 600,\n", - " 'data': {},\n", - " 'handle': },\n", - " 'meu-schedule-id3': {'frequency': 600,\n", - " 'data': {},\n", - " 'handle': },\n", - " 'meu-schedule-id2': {'frequency': 600,\n", - " 'data': {},\n", - " 'handle': },\n", - " 'meu-schedule-id': {'frequency': 600,\n", - " 'data': {},\n", - " 'handle': }}" - ] - }, - "execution_count": 88, - "metadata": {}, - "output_type": "execute_result" - } - ], + "outputs": [], "source": [ - "schedules_config" + "schedules_config = {\n", + " \"scouter-opcua-orchestrated-pipeline\": schedules_config['scouter-opcua-orchestrated-pipeline'],\n", + "}" ] }, { "cell_type": "code", - "execution_count": 85, + "execution_count": 14, "id": "c12f5e75", "metadata": {}, "outputs": [ @@ -305,12 +280,14 @@ "name": "stdout", "output_type": "stream", "text": [ - "meu-schedule-id5 {'frequency': 600, 'data': {'args': {'arg1': 'value1'}}, 'handle': }\n", - "meu-schedule-id4 {'frequency': 1200, 'data': {'args': {'arg1': 'value1'}}, 'handle': }\n", - "meu-schedule-id4\n", - "meu-schedule-id3 {'frequency': 600, 'data': {}, 'handle': }\n", - "meu-schedule-id2 {'frequency': 600, 'data': {}, 'handle': }\n", - "meu-schedule-id {'frequency': 600, 'data': {}, 'handle': }\n" + "scouter-opcua-orchestrated-pipeline {'frequency': 5, 'data': {'filters': {'NULL_VALUES_FILTER': {'policy': 'DISCARD'}, 'OUT_OF_BOUNDS_FILTER': {'policy': 'DISCARD'}}, 'frequency': '5s', 'max_retry_policy': 1, 'model_id': '1', 'model_name': 'Demo Model-Demo2', 'model_tags': {'Counter': {'aggr_func': 'avg', 'data_range': [-100, 100]}}, 'retention_time': 3600, 'schedule_name': 'scouter-opcua-orchestrated-pipeline', 'schema': 'sientia_data', 'table_name': 'laborious_data', 'topic': 'raw_scouter-opcua-orchestrated-pipeline', 'trigger_laborious': False, 'workflow_type': 'scouter'}, 'handle': }\n", + "scouter-opcua-orchestrated-pipeline\n", + "{'workflow': 'scouter', 'args': [metadata {\n", + " key: \"encoding\"\n", + " value: \"json/plain\"\n", + "}\n", + "data: \"{\\\"filters\\\":{\\\"NULL_VALUES_FILTER\\\":{\\\"policy\\\":\\\"DISCARD\\\"},\\\"OUT_OF_BOUNDS_FILTER\\\":{\\\"policy\\\":\\\"DISCARD\\\"}},\\\"frequency\\\":\\\"5s\\\",\\\"max_retry_policy\\\":1,\\\"model_id\\\":\\\"1\\\",\\\"model_name\\\":\\\"Demo Model-Demo2\\\",\\\"model_tags\\\":{\\\"Counter\\\":{\\\"aggr_func\\\":\\\"avg\\\",\\\"data_range\\\":[-100,100]}},\\\"retention_time\\\":3600,\\\"schedule_name\\\":\\\"scouter-opcua-orchestrated-pipeline\\\",\\\"schema\\\":\\\"sientia_data\\\",\\\"table_name\\\":\\\"laborious_data\\\",\\\"topic\\\":\\\"raw_scouter-opcua-orchestrated-pipeline\\\",\\\"trigger_laborious\\\":false,\\\"workflow_type\\\":\\\"scouter\\\"}\"\n", + "], 'id': 'scouter-opcua-orchestrated-pipeline', 'task_queue': 'scouter-queue', 'execution_timeout': datetime.timedelta(seconds=120), 'run_timeout': None, 'task_timeout': None, 'retry_policy': None, 'memo': None, 'typed_search_attributes': TypedSearchAttributes(search_attributes=[]), 'headers': None, 'untyped_search_attributes': {}, 'static_summary': None, 'static_details': None, 'priority': Priority(priority_key=None)}\n" ] } ], @@ -328,6 +305,8 @@ " \n", " async def update_schedule(input: ScheduleUpdateInput) -> ScheduleUpdate:\n", " schedule_action = input.description.schedule.action\n", + "\n", + " print(schedule_action.__dict__)\n", " \n", " if hasattr(schedule_action, 'args'):\n", " schedule_action.args = [{}]\n", @@ -421,7 +400,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.11.12" + "version": "3.11.13" } }, "nbformat": 4, diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index 6ba74b9..73f29a7 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -51,17 +51,20 @@ async def test_process_schedules(mock_predictions_batch, mock_scouter, formatter @mark.asyncio @patch("orchestrator.activities.formatters.gather_read_tags", return_value={ - "test_server_name:test_tag_address": { + "1:test_tag_address": { + "server_id": "1", "server_name": "test_server_name", "tag_address": "test_tag_address", "topics": ["raw_test_schedule"] }, - "test_server_name2:test_tag_address2": { + "2:test_tag_address2": { + "server_id": "2", "server_name": "test_server_name2", "tag_address": "test_tag_address2", "topics": ["raw_test_schedule2"] }, - "test_server_name2:test_tag_address3": { + "2:test_tag_address3": { + "server_id": "2", "server_name": "test_server_name2", "tag_address": "test_tag_address3", "topics": ["raw_test_schedule2"] @@ -72,6 +75,7 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma input_data = { "opc_servers": [ { + "id": "1", "server_name": "test_server_name", "url": "test_url", "uri": "test_uri", @@ -80,6 +84,7 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma } }, { + "id": "2", "server_name": "test_server_name2", "url": "test_url2", "uri": "test_uri2" @@ -98,13 +103,15 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma mock_build_tag_config.assert_has_calls([ call( { + "server_id": "1", "server_name": "test_server_name", "tag_address": "test_tag_address", "topics": ["raw_test_schedule"] }, ANY, { - "test_server_name": { + "1": { + "id": "1", "server_name": "test_server_name", "url": "test_url", "uri": "test_uri", @@ -112,7 +119,8 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma "test_name": "test_spec" } }, - "test_server_name2": { + "2": { + "id": "2", "server_name": "test_server_name2", "url": "test_url2", "uri": "test_uri2" @@ -124,13 +132,15 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma mock_build_tag_config.assert_has_calls([ call( { + "server_id": "2", "server_name": "test_server_name2", "tag_address": "test_tag_address2", "topics": ["raw_test_schedule2"] }, ANY, { - "test_server_name": { + "1": { + "id": "1", "server_name": "test_server_name", "url": "test_url", "uri": "test_uri", @@ -138,7 +148,8 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma "test_name": "test_spec" } }, - "test_server_name2": { + "2": { + "id": "2", "server_name": "test_server_name2", "url": "test_url2", "uri": "test_uri2" @@ -150,13 +161,15 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma mock_build_tag_config.assert_has_calls([ call( { + "server_id": "2", "server_name": "test_server_name2", "tag_address": "test_tag_address3", "topics": ["raw_test_schedule2"] }, ANY, { - "test_server_name": { + "1": { + "id": "1", "server_name": "test_server_name", "url": "test_url", "uri": "test_uri", @@ -164,7 +177,8 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma "test_name": "test_spec" } }, - "test_server_name2": { + "2": { + "id": "2", "server_name": "test_server_name2", "url": "test_url2", "uri": "test_uri2" @@ -177,12 +191,14 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma assert result == { "1": { "test_server_name": { + "server_id": "1", "name": "test_server_name", "url": "test_url", "server_uri": "test_uri", "test_name": "test_spec", "tags": { "test_tag_address": { + "server_id": "1", "server_name": "test_server_name", "tag_address": "test_tag_address", "topics": ["raw_test_schedule"] @@ -190,11 +206,13 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma } }, "test_server_name2": { + "server_id": "2", "name": "test_server_name2", "url": "test_url2", "server_uri": "test_uri2", "tags": { "test_tag_address2": { + "server_id": "2", "server_name": "test_server_name2", "tag_address": "test_tag_address2", "topics": ["raw_test_schedule2"] @@ -204,11 +222,13 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma }, "2": { "test_server_name2": { + "server_id": "2", "name": "test_server_name2", "url": "test_url2", "server_uri": "test_uri2", "tags": { "test_tag_address3": { + "server_id": "2", "server_name": "test_server_name2", "tag_address": "test_tag_address3", "topics": ["raw_test_schedule2"] diff --git a/tests/orchestrator/activities/test_slot_manager.py b/tests/orchestrator/activities/test_slot_manager.py index 75919d4..0fdd410 100644 --- a/tests/orchestrator/activities/test_slot_manager.py +++ b/tests/orchestrator/activities/test_slot_manager.py @@ -34,8 +34,13 @@ async def test_load_opc_slots(slot_manager): slot_manager.redis_client.keys.return_value = [ b"slot:opc_tags:1", b"slot:opc_tags:2", b"slot:opc_tags:3"] - slot_manager.redis_client.mget.return_value = [ - b"value1", "value2", None] + slot_manager.get = MagicMock( + side_effect=[ + "value1", + "value2", + None + ] + ) response = await slot_manager.load_opc_slots() diff --git a/tests/orchestrator/activities/test_temporal_manager.py b/tests/orchestrator/activities/test_temporal_manager.py index 810e562..7e504ef 100644 --- a/tests/orchestrator/activities/test_temporal_manager.py +++ b/tests/orchestrator/activities/test_temporal_manager.py @@ -1,4 +1,4 @@ -from unittest.mock import MagicMock, patch, AsyncMock, call +from unittest.mock import MagicMock, patch, AsyncMock, call, ANY from datetime import timedelta import base64 import json @@ -25,7 +25,7 @@ async def test_load_schedule(_mock_message_to_dict, temporal_manager): yield MagicMock( id="test-schedule-id", search_attributes={ - "Orchestrated": ["true"] + "orchestrated": ["true"] } ) yield MagicMock( @@ -136,16 +136,18 @@ async def test_create_schedule( mock_schedule_action_start_workflow.assert_has_calls([ call( - workflow="test-workflow", - args=input_data['schedules']['test-schedule'], + "test-workflow", + input_data['schedules']['test-schedule'], id="test-schedule", - task_queue="test-workflow-queue" + task_queue="test-workflow-queue", + execution_timeout=ANY ), call( - workflow="test-workflow", - args=input_data['schedules']['test-schedule-invalid-frequency'], + "test-workflow", + input_data['schedules']['test-schedule-invalid-frequency'], id="test-schedule-invalid-frequency", - task_queue="test-workflow-queue" + task_queue="test-workflow-queue", + execution_timeout=ANY ) ]) diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index 5b27250..789c374 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -12,9 +12,12 @@ from orchestrator.utils.orchestrator_functions import ( def test_common_config(): config = { + "workflow_type": "scouter", "schedule_name": "test_schedule", "model_id": "test_model_id", - "model_name": "test_model_name" + "models": { + "name": "test_model_name" + } } result = common_config(config) expected = { @@ -30,9 +33,12 @@ def test_common_config(): def test_scouter(): config = { + "workflow_type": "scouter", "schedule_name": "test_schedule", "model_id": "test_model_id", - "model_name": "test_model_name", + "models": { + "name": "test_model_name" + }, "filters": [ { "filter_name": "test_filter_name", @@ -127,7 +133,9 @@ def test_predictions_batch(mock_process_path_priority, "schedule_name": "test_schedule", "workflow_type": "predictions_batch", "model_id": "test_model_id", - "model_name": "test_model_name", + "models": { + "name": "test_model_name" + }, "query": "test_query", "write_tags": [ { @@ -244,6 +252,7 @@ def test_gather_read_tags(): "schedule_name": "test_schedule", "read_tags": [ { + "server_id": "1", "server_name": "test_server_name", "tag_address": "test_tag_address" } @@ -253,10 +262,12 @@ def test_gather_read_tags(): "schedule_name": "test_schedule2", "read_tags": [ { + "server_id": "2", "server_name": "test_server_name2", "tag_address": "test_tag_address2" }, { + "server_id": "2", "server_name": "test_server_name2", "tag_address": "test_tag_address3" } @@ -267,17 +278,20 @@ def test_gather_read_tags(): result = gather_read_tags(pipelines) expected = { - "test_server_name:test_tag_address": { + "1:test_tag_address": { + "server_id": "1", "server_name": "test_server_name", "tag_address": "test_tag_address", "topics": ["raw_test_schedule"] }, - "test_server_name2:test_tag_address2": { + "2:test_tag_address2": { + "server_id": "2", "server_name": "test_server_name2", "tag_address": "test_tag_address2", "topics": ["raw_test_schedule2"] }, - "test_server_name2:test_tag_address3": { + "2:test_tag_address3": { + "server_id": "2", "server_name": "test_server_name2", "tag_address": "test_tag_address3", "topics": ["raw_test_schedule2"] @@ -289,11 +303,13 @@ def test_gather_read_tags(): def test_build_tag_config(): tag = { + "server_id": "1", "server_name": "test_server_name", "tag_address": "test_tag_address" } opc_servers = { - "test_server_name": { + "1": { + "server_name": "test_server_name", "url": "test_url", "uri": "test_uri", "security_spec": { @@ -309,11 +325,13 @@ def test_build_tag_config(): expected = { "1": { "test_server_name": { + "server_id": "1", "name": "test_server_name", "url": "test_url", "server_uri": "test_uri", "tags": { "test_tag_address": { + "server_id": "1", "server_name": "test_server_name", "tag_address": "test_tag_address" }