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

fix: update search attribute keys and improve schedule handling

Browse Source 
- 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.
This commit is contained in:
vitor-aignosi
2025-06-04 15:06:54 -03:00
parent 5dc49b476a
commit dcbcc70651
9 changed files with 176 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
input_sample.json
View File

@@ -1,34 +1,5 @@
{ {
"schedule_name": "scouter-opcua-pipeline", "schedule_name": "orchestrator-test",
"model_name": "Demo Model", "pipelines_query": "SELECT pipelines.*, models FROM `pipelines` JOIN `models` ON KEYS pipelines.model_id;",
"model_id": 1, "opc_servers_query": "select META().id, opc_servers.* from `opc_servers`;"
"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": {}
} }

38
orchestrator/activities/formatters.py
View File

@@ -46,6 +46,11 @@ class Formatters(BaseActivity):
schedule_config[pipeline['schedule_name'] schedule_config[pipeline['schedule_name']
] = predictions_batch(pipeline) ] = predictions_batch(pipeline)
self.logger.info("Processed schedules")
self.logger.debug(json.dumps(
schedule_config, indent=4, sort_keys=True))
return schedule_config return schedule_config
@activity.defn(name="process_slots") @activity.defn(name="process_slots")
@@ -74,7 +79,7 @@ class Formatters(BaseActivity):
opc_servers = {} opc_servers = {}
for server in opc_servers_list: for server in opc_servers_list:
opc_servers[server['server_name']] = { opc_servers[server['id']] = {
**server, **server,
} }
@@ -99,6 +104,10 @@ class Formatters(BaseActivity):
slot_config = build_tag_config( slot_config = build_tag_config(
tag, slot_config.copy(), opc_servers, number_of_slots) 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 return slot_config
@activity.defn(name="create_schedule_config") @activity.defn(name="create_schedule_config")
@@ -131,7 +140,16 @@ class Formatters(BaseActivity):
for schedule_name, schedule in schedule_config.items(): for schedule_name, schedule in schedule_config.items():
if schedule_name in current_schedule_config: 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 to_update[schedule_name] = schedule
elif schedule_name not in current_schedule_config: elif schedule_name not in current_schedule_config:
@@ -141,12 +159,18 @@ class Formatters(BaseActivity):
if schedule_name not in schedule_config: if schedule_name not in schedule_config:
to_delete.append(schedule_name) to_delete.append(schedule_name)
return { output = {
"to_update": to_update, "to_update": to_update,
"to_create": to_create, "to_create": to_create,
"to_delete": to_delete "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") @activity.defn(name="create_slot_config")
async def create_slot_config(self, async def create_slot_config(self,
input_data: dict[str, Any]) -> dict[str, Any]: input_data: dict[str, Any]) -> dict[str, Any]:
@@ -179,11 +203,17 @@ class Formatters(BaseActivity):
to_delete = [str(i) for i in range( to_delete = [str(i) for i in range(
number_of_slots + 1, number_of_current_slots + 1)] number_of_slots + 1, number_of_current_slots + 1)]
return { output = {
"to_delete": to_delete, "to_delete": to_delete,
"to_insert": slot_config "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: def send_success_report(self, message: str, notification_id: str) -> None:
self.notification_handler.build_and_send_notification( self.notification_handler.build_and_send_notification(
notification_id, notification_id,

20
orchestrator/activities/temporal_manager.py
View File

@@ -49,7 +49,7 @@ class TemporalManager(BaseActivity):
async for schedule in await self.temporal_client.list_schedules(): async for schedule in await self.temporal_client.list_schedules():
search_attrs = getattr(schedule, "search_attributes", {}) search_attrs = getattr(schedule, "search_attributes", {})
if search_attrs.get("Orchestrated", ["false"]) == ["true"]: if search_attrs.get("orchestrated", ["false"]) == ["true"]:
schedule_id = schedule.id schedule_id = schedule.id
handle = self.temporal_client.get_schedule_handle(schedule_id) handle = self.temporal_client.get_schedule_handle(schedule_id)
@@ -113,14 +113,18 @@ class TemporalManager(BaseActivity):
workflow_type = schedule['workflow_type'] workflow_type = schedule['workflow_type']
try: 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( await self.temporal_client.create_schedule(
schedule_name, schedule_name,
Schedule( Schedule(
action=ScheduleActionStartWorkflow( action=ScheduleActionStartWorkflow(
workflow=workflow_type, workflow_type,
args=schedule, schedule,
id=schedule_name, id=schedule_name,
task_queue=f"{workflow_type}-queue" task_queue=f"{workflow_type}-queue",
execution_timeout=timedelta(minutes=2)
), ),
spec=ScheduleSpec( spec=ScheduleSpec(
intervals=[ intervals=[
@@ -181,8 +185,14 @@ class TemporalManager(BaseActivity):
async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate: async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate:
schedule_action = input_data.description.schedule.action schedule_action = input_data.description.schedule.action
self.logger.debug("Updating schedule:")
if hasattr(schedule_action, "args"): 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 = [ input_data.description.schedule.spec.intervals = [
ScheduleIntervalSpec( ScheduleIntervalSpec(

14
orchestrator/utils/orchestrator_functions.py
View File

@@ -9,7 +9,7 @@ def common_config(config: dict[str, Any]):
"max_retry_policy": config.get('max_retry_policy', 1), "max_retry_policy": config.get('max_retry_policy', 1),
"model_id": config['model_id'], "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 # Get all read tags from pipelines
for pipeline in pipelines: for pipeline in pipelines:
for tag in pipeline['read_tags']: 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: if tag_string not in tags:
tags[tag_string] = { tags[tag_string] = {
**tag, **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], def build_tag_config(tag: dict[str, Any], slot_config: dict[str, Any],
opc_servers: dict[str, Any], i: int): 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}"]: if server_name not in slot_config[f"{i}"]:
slot_config[f"{i}"][server_name] = { slot_config[f"{i}"][server_name] = {
"server_id": server_id,
"name": server_name, "name": server_name,
"url": opc_servers[server_name]['url'], "url": opc_servers[server_id]['url'],
"server_uri": opc_servers[server_name]['uri'], "server_uri": opc_servers[server_id]['uri'],
"tags": {} "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][name] = spec
slot_config[f"{i}"][server_name]["tags"][tag['tag_address']] = { slot_config[f"{i}"][server_name]["tags"][tag['tag_address']] = {

111
test.ipynb
View File

@@ -136,33 +136,19 @@
}, },
{ {
"cell_type": "code", "cell_type": "code",
"execution_count": 2, "execution_count": 6,
"id": "bb750ae6", "id": "bb750ae6",
"metadata": {}, "metadata": {},
"outputs": [ "outputs": [
{ {
"ename": "ScheduleAlreadyRunningError", "data": {
"evalue": "Schedule already running", "text/plain": [
"output_type": "error", "<temporalio.client.ScheduleHandle at 0x7dfad094e610>"
"traceback": [ ]
"\u001b[31m---------------------------------------------------------------------------\u001b[39m", },
"\u001b[31mRPCError\u001b[39m Traceback (most recent call last)", "execution_count": 6,
"\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", "metadata": {},
"\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", "output_type": "execute_result"
"\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"
]
} }
], ],
"source": [ "source": [
@@ -178,7 +164,7 @@
"from temporalio.common import TypedSearchAttributes, SearchAttributeKey, SearchAttributePair\n", "from temporalio.common import TypedSearchAttributes, SearchAttributeKey, SearchAttributePair\n",
"\n", "\n",
"\n", "\n",
"customer_id_key = SearchAttributeKey.for_keyword(\"Orchestrated\")\n", "customer_id_key = SearchAttributeKey.for_keyword(\"orchestrated\")\n",
"search_attributes = TypedSearchAttributes([\n", "search_attributes = TypedSearchAttributes([\n",
" SearchAttributePair(customer_id_key, \"true\")\n", " SearchAttributePair(customer_id_key, \"true\")\n",
"])\n", "])\n",
@@ -213,9 +199,7 @@
"name": "stdout", "name": "stdout",
"output_type": "stream", "output_type": "stream",
"text": [ "text": [
"Getting orchestrated schedules...\n", "Getting orchestrated schedules...\n"
"Found %d orchestrated schedules 5\n",
"Orchestrated schedules: %s {'meu-schedule-id5': {'frequency': 600, 'data': {'args': {'arg1': 'value1'}}, 'handle': <temporalio.client.ScheduleHandle object at 0x701334390f10>}, 'meu-schedule-id4': {'frequency': 600, 'data': {}, 'handle': <temporalio.client.ScheduleHandle object at 0x7013243d0b10>}, 'meu-schedule-id3': {'frequency': 600, 'data': {}, 'handle': <temporalio.client.ScheduleHandle object at 0x7013243d0a90>}, 'meu-schedule-id2': {'frequency': 600, 'data': {}, 'handle': <temporalio.client.ScheduleHandle object at 0x7013243d0ed0>}, 'meu-schedule-id': {'frequency': 600, 'data': {}, 'handle': <temporalio.client.ScheduleHandle object at 0x7013243d1190>}}\n"
] ]
} }
], ],
@@ -225,7 +209,22 @@
}, },
{ {
"cell_type": "code", "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", "id": "a4c777dd",
"metadata": {}, "metadata": {},
"outputs": [], "outputs": [],
@@ -234,9 +233,8 @@
"import base64\n", "import base64\n",
"import json\n", "import json\n",
"\n", "\n",
"\n",
"schedules_config = {}\n", "schedules_config = {}\n",
"for schedule in schedules:\n", "async for schedule in await temporal_client.list_schedules():\n",
" id = schedule.id\n", " id = schedule.id\n",
"\n", "\n",
" handle = temporal_client.get_schedule_handle(id)\n", " handle = temporal_client.get_schedule_handle(id)\n",
@@ -262,42 +260,19 @@
}, },
{ {
"cell_type": "code", "cell_type": "code",
"execution_count": 88, "execution_count": 13,
"id": "988d1718", "id": "988d1718",
"metadata": {}, "metadata": {},
"outputs": [ "outputs": [],
{
"data": {
"text/plain": [
"{'meu-schedule-id5': {'frequency': 600,\n",
" 'data': {'args': {'arg1': 'value1'}},\n",
" 'handle': <temporalio.client.ScheduleHandle at 0x76f30f5828d0>},\n",
" 'meu-schedule-id4': {'frequency': 600,\n",
" 'data': {},\n",
" 'handle': <temporalio.client.ScheduleHandle at 0x76f30f5cf4d0>},\n",
" 'meu-schedule-id3': {'frequency': 600,\n",
" 'data': {},\n",
" 'handle': <temporalio.client.ScheduleHandle at 0x76f3141f13d0>},\n",
" 'meu-schedule-id2': {'frequency': 600,\n",
" 'data': {},\n",
" 'handle': <temporalio.client.ScheduleHandle at 0x76f30f64c710>},\n",
" 'meu-schedule-id': {'frequency': 600,\n",
" 'data': {},\n",
" 'handle': <temporalio.client.ScheduleHandle at 0x76f31417c350>}}"
]
},
"execution_count": 88,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [ "source": [
"schedules_config" "schedules_config = {\n",
" \"scouter-opcua-orchestrated-pipeline\": schedules_config['scouter-opcua-orchestrated-pipeline'],\n",
"}"
] ]
}, },
{ {
"cell_type": "code", "cell_type": "code",
"execution_count": 85, "execution_count": 14,
"id": "c12f5e75", "id": "c12f5e75",
"metadata": {}, "metadata": {},
"outputs": [ "outputs": [
@@ -305,12 +280,14 @@
"name": "stdout", "name": "stdout",
"output_type": "stream", "output_type": "stream",
"text": [ "text": [
"meu-schedule-id5 {'frequency': 600, 'data': {'args': {'arg1': 'value1'}}, 'handle': <temporalio.client.ScheduleHandle object at 0x76f30f5ce590>}\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': <temporalio.client.ScheduleHandle object at 0x7dfab3b2ab50>}\n",
"meu-schedule-id4 {'frequency': 1200, 'data': {'args': {'arg1': 'value1'}}, 'handle': <temporalio.client.ScheduleHandle object at 0x76f30f723910>}\n", "scouter-opcua-orchestrated-pipeline\n",
"meu-schedule-id4\n", "{'workflow': 'scouter', 'args': [metadata {\n",
"meu-schedule-id3 {'frequency': 600, 'data': {}, 'handle': <temporalio.client.ScheduleHandle object at 0x76f30f64fe50>}\n", " key: \"encoding\"\n",
"meu-schedule-id2 {'frequency': 600, 'data': {}, 'handle': <temporalio.client.ScheduleHandle object at 0x76f31417b9d0>}\n", " value: \"json/plain\"\n",
"meu-schedule-id {'frequency': 600, 'data': {}, 'handle': <temporalio.client.ScheduleHandle object at 0x76f30f64dad0>}\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", " \n",
" async def update_schedule(input: ScheduleUpdateInput) -> ScheduleUpdate:\n", " async def update_schedule(input: ScheduleUpdateInput) -> ScheduleUpdate:\n",
" schedule_action = input.description.schedule.action\n", " schedule_action = input.description.schedule.action\n",
"\n",
" print(schedule_action.__dict__)\n",
" \n", " \n",
" if hasattr(schedule_action, 'args'):\n", " if hasattr(schedule_action, 'args'):\n",
" schedule_action.args = [{}]\n", " schedule_action.args = [{}]\n",
@@ -421,7 +400,7 @@
"name": "python", "name": "python",
"nbconvert_exporter": "python", "nbconvert_exporter": "python",
"pygments_lexer": "ipython3", "pygments_lexer": "ipython3",
"version": "3.11.12" "version": "3.11.13"
} }
}, },
"nbformat": 4, "nbformat": 4,

38
tests/orchestrator/activities/test_formatters.py
View File

@@ -51,17 +51,20 @@ async def test_process_schedules(mock_predictions_batch, mock_scouter, formatter
@mark.asyncio @mark.asyncio
@patch("orchestrator.activities.formatters.gather_read_tags", @patch("orchestrator.activities.formatters.gather_read_tags",
return_value={ return_value={
"test_server_name:test_tag_address": { "1:test_tag_address": {
"server_id": "1",
"server_name": "test_server_name", "server_name": "test_server_name",
"tag_address": "test_tag_address", "tag_address": "test_tag_address",
"topics": ["raw_test_schedule"] "topics": ["raw_test_schedule"]
}, },
"test_server_name2:test_tag_address2": { "2:test_tag_address2": {
"server_id": "2",
"server_name": "test_server_name2", "server_name": "test_server_name2",
"tag_address": "test_tag_address2", "tag_address": "test_tag_address2",
"topics": ["raw_test_schedule2"] "topics": ["raw_test_schedule2"]
}, },
"test_server_name2:test_tag_address3": { "2:test_tag_address3": {
"server_id": "2",
"server_name": "test_server_name2", "server_name": "test_server_name2",
"tag_address": "test_tag_address3", "tag_address": "test_tag_address3",
"topics": ["raw_test_schedule2"] "topics": ["raw_test_schedule2"]
@@ -72,6 +75,7 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma
input_data = { input_data = {
"opc_servers": [ "opc_servers": [
{ {
"id": "1",
"server_name": "test_server_name", "server_name": "test_server_name",
"url": "test_url", "url": "test_url",
"uri": "test_uri", "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", "server_name": "test_server_name2",
"url": "test_url2", "url": "test_url2",
"uri": "test_uri2" "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([ mock_build_tag_config.assert_has_calls([
call( call(
{ {
"server_id": "1",
"server_name": "test_server_name", "server_name": "test_server_name",
"tag_address": "test_tag_address", "tag_address": "test_tag_address",
"topics": ["raw_test_schedule"] "topics": ["raw_test_schedule"]
}, },
ANY, ANY,
{ {
"test_server_name": { "1": {
"id": "1",
"server_name": "test_server_name", "server_name": "test_server_name",
"url": "test_url", "url": "test_url",
"uri": "test_uri", "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_name": "test_spec"
} }
}, },
"test_server_name2": { "2": {
"id": "2",
"server_name": "test_server_name2", "server_name": "test_server_name2",
"url": "test_url2", "url": "test_url2",
"uri": "test_uri2" "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([ mock_build_tag_config.assert_has_calls([
call( call(
{ {
"server_id": "2",
"server_name": "test_server_name2", "server_name": "test_server_name2",
"tag_address": "test_tag_address2", "tag_address": "test_tag_address2",
"topics": ["raw_test_schedule2"] "topics": ["raw_test_schedule2"]
}, },
ANY, ANY,
{ {
"test_server_name": { "1": {
"id": "1",
"server_name": "test_server_name", "server_name": "test_server_name",
"url": "test_url", "url": "test_url",
"uri": "test_uri", "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_name": "test_spec"
} }
}, },
"test_server_name2": { "2": {
"id": "2",
"server_name": "test_server_name2", "server_name": "test_server_name2",
"url": "test_url2", "url": "test_url2",
"uri": "test_uri2" "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([ mock_build_tag_config.assert_has_calls([
call( call(
{ {
"server_id": "2",
"server_name": "test_server_name2", "server_name": "test_server_name2",
"tag_address": "test_tag_address3", "tag_address": "test_tag_address3",
"topics": ["raw_test_schedule2"] "topics": ["raw_test_schedule2"]
}, },
ANY, ANY,
{ {
"test_server_name": { "1": {
"id": "1",
"server_name": "test_server_name", "server_name": "test_server_name",
"url": "test_url", "url": "test_url",
"uri": "test_uri", "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_name": "test_spec"
} }
}, },
"test_server_name2": { "2": {
"id": "2",
"server_name": "test_server_name2", "server_name": "test_server_name2",
"url": "test_url2", "url": "test_url2",
"uri": "test_uri2" "uri": "test_uri2"
@@ -177,12 +191,14 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma
assert result == { assert result == {
"1": { "1": {
"test_server_name": { "test_server_name": {
"server_id": "1",
"name": "test_server_name", "name": "test_server_name",
"url": "test_url", "url": "test_url",
"server_uri": "test_uri", "server_uri": "test_uri",
"test_name": "test_spec", "test_name": "test_spec",
"tags": { "tags": {
"test_tag_address": { "test_tag_address": {
"server_id": "1",
"server_name": "test_server_name", "server_name": "test_server_name",
"tag_address": "test_tag_address", "tag_address": "test_tag_address",
"topics": ["raw_test_schedule"] "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": { "test_server_name2": {
"server_id": "2",
"name": "test_server_name2", "name": "test_server_name2",
"url": "test_url2", "url": "test_url2",
"server_uri": "test_uri2", "server_uri": "test_uri2",
"tags": { "tags": {
"test_tag_address2": { "test_tag_address2": {
"server_id": "2",
"server_name": "test_server_name2", "server_name": "test_server_name2",
"tag_address": "test_tag_address2", "tag_address": "test_tag_address2",
"topics": ["raw_test_schedule2"] "topics": ["raw_test_schedule2"]
@@ -204,11 +222,13 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma
}, },
"2": { "2": {
"test_server_name2": { "test_server_name2": {
"server_id": "2",
"name": "test_server_name2", "name": "test_server_name2",
"url": "test_url2", "url": "test_url2",
"server_uri": "test_uri2", "server_uri": "test_uri2",
"tags": { "tags": {
"test_tag_address3": { "test_tag_address3": {
"server_id": "2",
"server_name": "test_server_name2", "server_name": "test_server_name2",
"tag_address": "test_tag_address3", "tag_address": "test_tag_address3",
"topics": ["raw_test_schedule2"] "topics": ["raw_test_schedule2"]

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

@@ -34,8 +34,13 @@ async def test_load_opc_slots(slot_manager):
slot_manager.redis_client.keys.return_value = [ slot_manager.redis_client.keys.return_value = [
b"slot:opc_tags:1", b"slot:opc_tags:2", b"slot:opc_tags:3"] b"slot:opc_tags:1", b"slot:opc_tags:2", b"slot:opc_tags:3"]
slot_manager.redis_client.mget.return_value = [ slot_manager.get = MagicMock(
b"value1", "value2", None] side_effect=[
"value1",
"value2",
None
]
)
response = await slot_manager.load_opc_slots() response = await slot_manager.load_opc_slots()

18
tests/orchestrator/activities/test_temporal_manager.py
View File

@@ -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 from datetime import timedelta
import base64 import base64
import json import json
@@ -25,7 +25,7 @@ async def test_load_schedule(_mock_message_to_dict, temporal_manager):
yield MagicMock( yield MagicMock(
id="test-schedule-id", id="test-schedule-id",
search_attributes={ search_attributes={
"Orchestrated": ["true"] "orchestrated": ["true"]
} }
) )
yield MagicMock( yield MagicMock(
@@ -136,16 +136,18 @@ async def test_create_schedule(
mock_schedule_action_start_workflow.assert_has_calls([ mock_schedule_action_start_workflow.assert_has_calls([
call( call(
workflow="test-workflow", "test-workflow",
args=input_data['schedules']['test-schedule'], input_data['schedules']['test-schedule'],
id="test-schedule", id="test-schedule",
task_queue="test-workflow-queue" task_queue="test-workflow-queue",
execution_timeout=ANY
), ),
call( call(
workflow="test-workflow", "test-workflow",
args=input_data['schedules']['test-schedule-invalid-frequency'], input_data['schedules']['test-schedule-invalid-frequency'],
id="test-schedule-invalid-frequency", id="test-schedule-invalid-frequency",
task_queue="test-workflow-queue" task_queue="test-workflow-queue",
execution_timeout=ANY
) )
]) ])

32
tests/orchestrator/utils/test_orchestrator_functions.py
View File

@@ -12,9 +12,12 @@ from orchestrator.utils.orchestrator_functions import (
def test_common_config(): def test_common_config():
config = { config = {
"workflow_type": "scouter",
"schedule_name": "test_schedule", "schedule_name": "test_schedule",
"model_id": "test_model_id", "model_id": "test_model_id",
"model_name": "test_model_name" "models": {
"name": "test_model_name"
}
} }
result = common_config(config) result = common_config(config)
expected = { expected = {
@@ -30,9 +33,12 @@ def test_common_config():
def test_scouter(): def test_scouter():
config = { config = {
"workflow_type": "scouter",
"schedule_name": "test_schedule", "schedule_name": "test_schedule",
"model_id": "test_model_id", "model_id": "test_model_id",
"model_name": "test_model_name", "models": {
"name": "test_model_name"
},
"filters": [ "filters": [
{ {
"filter_name": "test_filter_name", "filter_name": "test_filter_name",
@@ -127,7 +133,9 @@ def test_predictions_batch(mock_process_path_priority,
"schedule_name": "test_schedule", "schedule_name": "test_schedule",
"workflow_type": "predictions_batch", "workflow_type": "predictions_batch",
"model_id": "test_model_id", "model_id": "test_model_id",
"model_name": "test_model_name", "models": {
"name": "test_model_name"
},
"query": "test_query", "query": "test_query",
"write_tags": [ "write_tags": [
{ {
@@ -244,6 +252,7 @@ def test_gather_read_tags():
"schedule_name": "test_schedule", "schedule_name": "test_schedule",
"read_tags": [ "read_tags": [
{ {
"server_id": "1",
"server_name": "test_server_name", "server_name": "test_server_name",
"tag_address": "test_tag_address" "tag_address": "test_tag_address"
} }
@@ -253,10 +262,12 @@ def test_gather_read_tags():
"schedule_name": "test_schedule2", "schedule_name": "test_schedule2",
"read_tags": [ "read_tags": [
{ {
"server_id": "2",
"server_name": "test_server_name2", "server_name": "test_server_name2",
"tag_address": "test_tag_address2" "tag_address": "test_tag_address2"
}, },
{ {
"server_id": "2",
"server_name": "test_server_name2", "server_name": "test_server_name2",
"tag_address": "test_tag_address3" "tag_address": "test_tag_address3"
} }
@@ -267,17 +278,20 @@ def test_gather_read_tags():
result = gather_read_tags(pipelines) result = gather_read_tags(pipelines)
expected = { expected = {
"test_server_name:test_tag_address": { "1:test_tag_address": {
"server_id": "1",
"server_name": "test_server_name", "server_name": "test_server_name",
"tag_address": "test_tag_address", "tag_address": "test_tag_address",
"topics": ["raw_test_schedule"] "topics": ["raw_test_schedule"]
}, },
"test_server_name2:test_tag_address2": { "2:test_tag_address2": {
"server_id": "2",
"server_name": "test_server_name2", "server_name": "test_server_name2",
"tag_address": "test_tag_address2", "tag_address": "test_tag_address2",
"topics": ["raw_test_schedule2"] "topics": ["raw_test_schedule2"]
}, },
"test_server_name2:test_tag_address3": { "2:test_tag_address3": {
"server_id": "2",
"server_name": "test_server_name2", "server_name": "test_server_name2",
"tag_address": "test_tag_address3", "tag_address": "test_tag_address3",
"topics": ["raw_test_schedule2"] "topics": ["raw_test_schedule2"]
@@ -289,11 +303,13 @@ def test_gather_read_tags():
def test_build_tag_config(): def test_build_tag_config():
tag = { tag = {
"server_id": "1",
"server_name": "test_server_name", "server_name": "test_server_name",
"tag_address": "test_tag_address" "tag_address": "test_tag_address"
} }
opc_servers = { opc_servers = {
"test_server_name": { "1": {
"server_name": "test_server_name",
"url": "test_url", "url": "test_url",
"uri": "test_uri", "uri": "test_uri",
"security_spec": { "security_spec": {
@@ -309,11 +325,13 @@ def test_build_tag_config():
expected = { expected = {
"1": { "1": {
"test_server_name": { "test_server_name": {
"server_id": "1",
"name": "test_server_name", "name": "test_server_name",
"url": "test_url", "url": "test_url",
"server_uri": "test_uri", "server_uri": "test_uri",
"tags": { "tags": {
"test_tag_address": { "test_tag_address": {
"server_id": "1",
"server_name": "test_server_name", "server_name": "test_server_name",
"tag_address": "test_tag_address" "tag_address": "test_tag_address"
} }