From a0d8f96a99b28bc4010374d61f84ce5181194c8a Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 9 Jun 2025 17:19:53 -0300 Subject: [PATCH 1/5] SIENTIAPDE-1095 Fix decoding of active ingestors to handle both bytes and string types --- orchestrator/activities/slot_manager.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index 45d77d4..d8484cb 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -67,7 +67,15 @@ class SlotManager(Redis): self.logger.debug("Active ingestors: \n %s", active_ingestors) - return [ingestor.decode('utf-8') for ingestor in active_ingestors] + ingestors = [] + + for ingestor in active_ingestors: + if isinstance(ingestor, bytes): + ingestors.append(ingestor.decode('utf-8')) + else: + ingestors.append(ingestor) + + return ingestors @activity.defn(name="update_slots") async def update_slots(self, input_data: dict[str, Any]) -> dict[str, Any]: From caa7b125f98f81f86725938665b68ec9a7a6e18d Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 10 Jun 2025 08:25:50 -0300 Subject: [PATCH 2/5] SIENTIAPDE-1095 Refactor predictions_batch to use server_id instead of server_name; update values.yaml to set GITHUB_BRANCH to main; add new laborious pipeline configuration in samples.json --- orchestrator/utils/orchestrator_functions.py | 10 ++-- samples.json | 59 ++++++++++++++++++++ values.yaml | 4 +- 3 files changed, 66 insertions(+), 7 deletions(-) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 22e5444..6d18011 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -65,18 +65,18 @@ def process_path_priority(path_priority: list[str]): def predictions_batch(config: dict[str, Any]): tags = {} for tag in config['write_tags']: - if tag['server_name'] not in tags: - tags[tag['server_name']] = {} + if tag['server_id'] not in tags: + tags[tag['server_id']] = {} tag_type = tag['type'] if tag_type == 'prediction' or tag_type == 'confidence': tag_type_str = f"{tag_type}_tags" - if tag_type_str not in tags[tag['server_name']]: - tags[tag['server_name']][tag_type_str] = {} + if tag_type_str not in tags[tag['server_id']]: + tags[tag['server_id']][tag_type_str] = {} - tags[tag['server_name']][tag_type_str][tag['addr']] = { + tags[tag['server_id']][tag_type_str][tag['addr']] = { "data_type": tag.get('data_type', 'float'), } diff --git a/samples.json b/samples.json index f38194e..336ef01 100644 --- a/samples.json +++ b/samples.json @@ -57,6 +57,65 @@ } ], "tag_retention_minutes": 60 + }, + "2": { + "schedule_name": "laborious-orchestrated-pipeline", + "model_id": "1", + "workflow_type": "predictions_batch", + "frequency": "30s", + "max_retry_policy": 1, + "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;", + "retention_time": 60, + "opc_output_config": [ + { + "server_id": "1", + "type": "prediction", + "addr": "ns=2;i=2", + "data_type": "float" + }, + { + "server_id": "1", + "type": "confidence", + "addr": "ns=2;i=2", + "data_type": "float" + } + ], + "input_filters": [ + { + "filter_name": "EMPTY_DATA", + "policy": "STOP" + }, + { + "filter_name": "SPECIFIC_VARIABLES_NULL_VALUES", + "policy": "CONTINUE", + "config": { + "VARIABLES": [ + "Counter" + ] + } + } + ], + "mlflow_transform_filters": [ + { + "filter_name": "API_ERROR", + "policy": "REPEAT" + }, + { + "filter_name": "NAN_VALUES", + "policy": "STOP" + } + ], + "mlflow_predict_filters": [ + { + "filter_name": "API_ERROR", + "policy": "CONTINUE" + } + ], + "path_priority": [ + "STOP", + "CONTINUE", + "REPEAT" + ] } }, "opc-servers": { diff --git a/values.yaml b/values.yaml index 016b99a..59e971a 100644 --- a/values.yaml +++ b/values.yaml @@ -123,7 +123,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-1030-implementar-os-workflows-mapeados-utilizando-as-workers-e-activities-apropriadas" + value: "main" - name: PYTHON_APP value: "orchestrator.worker.worker" @@ -176,4 +176,4 @@ ssh: # kubectl create secret generic git-ssh-key-sientia-orchestrator-worker \ # --namespace sientia \ # --from-file=ssh-privatekey=git_key \ -# --type=kubernetes.io/ssh-auth \ No newline at end of file +# --type=kubernetes.io/ssh-auth From 89ac81e5e543b1cc234029007c5159f41b8dae94 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 10 Jun 2025 08:32:16 -0300 Subject: [PATCH 3/5] SIENTIAPDE-1095 Fix read_tags access in gather_read_tags function and update samples.json to use write_tags --- orchestrator/utils/orchestrator_functions.py | 2 +- samples.json | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 6d18011..5375e14 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -128,7 +128,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']: + for tag in pipeline.get('read_tags', []): tag_string = f"{tag['server_id']}:{tag['tag_address']}" if tag_string not in tags: tags[tag_string] = { diff --git a/samples.json b/samples.json index 336ef01..18c7177 100644 --- a/samples.json +++ b/samples.json @@ -66,7 +66,7 @@ "max_retry_policy": 1, "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;", "retention_time": 60, - "opc_output_config": [ + "write_tags": [ { "server_id": "1", "type": "prediction", From 5e2634f062b04051a2890bfad13b04c83ef8afb3 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 10 Jun 2025 11:38:35 -0300 Subject: [PATCH 4/5] fix: update key casing for VARIABLES to variables in filter configuration --- samples.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/samples.json b/samples.json index 18c7177..249d968 100644 --- a/samples.json +++ b/samples.json @@ -89,7 +89,7 @@ "filter_name": "SPECIFIC_VARIABLES_NULL_VALUES", "policy": "CONTINUE", "config": { - "VARIABLES": [ + "variables": [ "Counter" ] } From 0a612f6a5b5cc028e2ffa04adcc3e097332716af Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 10 Jun 2025 13:06:13 -0300 Subject: [PATCH 5/5] fix: update server_name to server_id in predictions batch test --- tests/orchestrator/activities/test_slot_manager.py | 2 +- tests/orchestrator/utils/test_orchestrator_functions.py | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/orchestrator/activities/test_slot_manager.py b/tests/orchestrator/activities/test_slot_manager.py index ca2adfa..39d65aa 100644 --- a/tests/orchestrator/activities/test_slot_manager.py +++ b/tests/orchestrator/activities/test_slot_manager.py @@ -76,7 +76,7 @@ async def test_load_opc_slots_no_decode(slot_manager): @mark.asyncio async def test_load_active_ingestors(slot_manager): slot_manager.redis_client.keys.return_value = [ - b"heartbeat:ingestor:1", b"heartbeat:ingestor:2", b"heartbeat:ingestor:3"] + b"heartbeat:ingestor:1", b"heartbeat:ingestor:2", "heartbeat:ingestor:3"] response = await slot_manager.load_active_ingestors() diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index 789c374..67a05d5 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -139,12 +139,12 @@ def test_predictions_batch(mock_process_path_priority, "query": "test_query", "write_tags": [ { - "server_name": "test_server_name", + "server_id": "test_server_id", "type": "prediction", "addr": "test_addr" }, { - "server_name": "test_server_name", + "server_id": "test_server_id", "type": "confidence", "addr": "test_addr" } @@ -210,7 +210,7 @@ def test_predictions_batch(mock_process_path_priority, "table_name": "predictions", "retention_time": 60 * 60, "opc_output_config": { - "test_server_name": { + "test_server_id": { "prediction_tags": { "test_addr": { "data_type": "float"