From 5a4d6c57060b7047b7d8837850b3532e6292a138 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 23 Jul 2025 12:26:36 -0300 Subject: [PATCH 1/5] SIENTIAPDE-1171 feat: add minimal_retrain function to orchestrator_functions.py for retraining workflows - Introduced minimal_retrain function to streamline retraining configuration. - The function integrates common configuration with specific parameters for retraining workflows. --- orchestrator/utils/orchestrator_functions.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index f1d6510..6fb3810 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -13,6 +13,17 @@ def common_config(config: dict[str, Any]): } +def minimal_retrain(config: dict[str, Any]): + return { + **common_config(config), + "workflow_type": "retrain", + "schedule_name": config['schedule_name'], + "query": config['query'], + "schema": "sientia_data", + "table_name": "log_retrain", + } + + def scouter(config: dict[str, Any]): filters = {} for f in config.get('filters', []): From 115a174a315ec25adaf0903952d367bcb581c30f Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 23 Jul 2025 12:32:20 -0300 Subject: [PATCH 2/5] SIENTIAPDE-1171 Update GITHUB_BRANCH in values.yaml to SIENTIAPDE-1171-criar-pipeline-de-retreino-laborious --- values.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/values.yaml b/values.yaml index bb7ac0d..4614016 100644 --- a/values.yaml +++ b/values.yaml @@ -132,7 +132,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-1166-alterar-orquestrador-para-criar-collections-com-ttl-no-mongo]" + value: "SIENTIAPDE-1171-criar-pipeline-de-retreino-laborious" - name: PYTHON_APP value: "orchestrator.worker.worker" From 66e9cf58e46828ec43670ac3b67ebfc0b63651d2 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 23 Jul 2025 12:51:55 -0300 Subject: [PATCH 3/5] SIENTIAPDE-1171 SIENTIAPDE-1171 feat: add minimal retrain schedule configuration to samples.json and update formatters.py - Added a new schedule configuration for the minimal retrain pipeline in samples.json. - Updated the Formatters class in formatters.py to handle the new minimal retrain workflow type, integrating the minimal_retrain function for streamlined processing. --- orchestrator/activities/formatters.py | 9 +++++++++ samples.json | 12 +++++++++++- 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 67a6773..5078b12 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -1,6 +1,8 @@ from temporalio import activity, workflow +from orchestrator.utils.orchestrator_functions import minimal_retrain + with workflow.unsafe.imports_passed_through(): import json @@ -67,6 +69,13 @@ class Formatters(BaseActivity): "updated_at": pipeline.get( "updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT)) } + elif pipeline['workflow_type'] == 'minimal_retrain': + schedule_config[self.laborious_namespace][pipeline['schedule_name'] + ] = { + **minimal_retrain(pipeline), + "updated_at": pipeline.get( + "updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT)) + } self.info("Processed schedules", metadata=metadata) self.debug(json.dumps( diff --git a/samples.json b/samples.json index 94274ca..882927f 100644 --- a/samples.json +++ b/samples.json @@ -120,7 +120,17 @@ ], "active": true, "updated_at": "2025-07-14 10:00:00.000000" - } + }, + "3": { + "schedule_name": "minimal-retrain-pipeline", + "model_id": "1", + "workflow_type": "minimal_retrain", + "frequency": "5m", + "max_retry_policy": 1, + "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;", + "active": true, + "updated_at": "2025-07-23 10:00:00.000000" + } }, "opc-servers": { "1": { From c1e6807ac3413576664d2dfe437cd3547f705e80 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 23 Jul 2025 13:01:37 -0300 Subject: [PATCH 4/5] SIENTIAPDE-1171 fix: update workflow_type in minimal_retrain function for consistency - Changed workflow_type from "retrain" to "minimal_retrain" in the minimal_retrain function to align with the new schedule configuration. --- orchestrator/utils/orchestrator_functions.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 6fb3810..7b6289f 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -16,7 +16,7 @@ def common_config(config: dict[str, Any]): def minimal_retrain(config: dict[str, Any]): return { **common_config(config), - "workflow_type": "retrain", + "workflow_type": "minimal_retrain", "schedule_name": config['schedule_name'], "query": config['query'], "schema": "sientia_data", From 1bdfc48cb2c9fdd373892ca98e101173c5060ca1 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 23 Jul 2025 16:56:24 -0300 Subject: [PATCH 5/5] SIENTIAPDE-1171 feat: enhance test coverage for minimal_retrain workflow - Added a new test for the minimal_retrain function to validate its configuration output. - Updated the test_process_schedules to include a new schedule entry for minimal_retrain, ensuring comprehensive testing of the workflow integration. --- .../activities/test_formatters.py | 15 ++++++++++- .../utils/test_orchestrator_functions.py | 26 +++++++++++++++++++ 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index b75cdfc..04ada97 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -34,7 +34,9 @@ metadata = { return_value={"test_scouter": "test_scouter"}) @patch("orchestrator.activities.formatters.predictions_batch", return_value={"test_predictions_batch": "test_predictions_batch"}) -async def test_process_schedules(mock_predictions_batch, mock_scouter, formatters): +@patch("orchestrator.activities.formatters.minimal_retrain", + return_value={"test_minimal_retrain": "test_minimal_retrain"}) +async def test_process_schedules(mock_minimal_retrain, mock_predictions_batch, mock_scouter, formatters): input_data = { "pipelines": [ { @@ -50,6 +52,13 @@ async def test_process_schedules(mock_predictions_batch, mock_scouter, formatter "model_name": "test_model_name", "model_id": "test_model_id", "updated_at": "2021-01-02" + }, + { + "schedule_name": "test_schedule_name3", + "workflow_type": "minimal_retrain", + "model_name": "test_model_name", + "model_id": "test_model_id", + "updated_at": "2021-01-03" } ] } @@ -67,6 +76,10 @@ async def test_process_schedules(mock_predictions_batch, mock_scouter, formatter "test_schedule_name2": { "test_predictions_batch": "test_predictions_batch", "updated_at": "2021-01-02" + }, + "test_schedule_name3": { + "test_minimal_retrain": "test_minimal_retrain", + "updated_at": "2021-01-03" } } } diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index e0b08a9..eb4b056 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -1,6 +1,7 @@ from unittest.mock import patch, call from orchestrator.utils.orchestrator_functions import ( common_config, + minimal_retrain, scouter, predictions_batch, overlap_filter_config, @@ -31,6 +32,31 @@ def test_common_config(): assert result == expected +def test_minimal_retrain(): + config = { + "workflow_type": "minimal_retrain", + "schedule_name": "test_schedule", + "model_id": "test_model_id", + "models": { + "name": "test_model_name" + }, + "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;", + } + result = minimal_retrain(config) + expected = { + "workflow_type": "minimal_retrain", + "schedule_name": "test_schedule", + "frequency": "1m", + "max_retry_policy": 1, + "model_id": "test_model_id", + "model_name": "test_model_name", + "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;", + "schema": "sientia_data", + "table_name": "log_retrain", + } + assert result == expected + + def test_scouter(): config = { "workflow_type": "scouter",