From 17b98392a7de434d6e9aff2e3078c955bcfe8817 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 10 Jun 2025 08:39:19 -0300 Subject: [PATCH 1/7] SIENTIAPDE-1094 Update notification handler parameters and change task queue name --- laborious/worker/worker.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index f10716e..ae5e47c 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -32,10 +32,6 @@ async def main(): servers=os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'http://localhost:9092'), logger=logger, project_name=os.getenv('PROJECT_NAME', 'laborious'), - pipeline_name='-', - trigger_name='-', - model_name='-', - model='-' ) logger.info('Starting Activities...') @@ -60,7 +56,7 @@ async def main(): workers = [ Worker( temporal_client, - task_queue='predictions-queue', + task_queue='predictions_batch-queue', workflows=[PredictionsBatch, PredictionProcess, FormatAndExportPrediction], activities=[ From 881e7232e92b0d81d9467c4ecda4ce4ab5ba0735 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 10 Jun 2025 08:48:44 -0300 Subject: [PATCH 2/7] SIENTIAPDE-1094 fix: update variable names for consistency and bump image tag version --- laborious/activities/gates.py | 2 +- laborious/utils/filters/conditional_filters.py | 2 +- values.yaml | 4 ++-- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index f0796ff..7bab5f5 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -139,7 +139,7 @@ class Gates(BaseActivity): continue try: if mlflow_response_filter_functions[fil](data, config): - filter_output.append(config['POLICY']) + filter_output.append(config['policy']) comments.append(data['content']['message']) self.notification_handler.build_and_send_notification( notification_id=f"{gate_type.upper()}_GATE_RESPONSE_FILTER__{fil}", diff --git a/laborious/utils/filters/conditional_filters.py b/laborious/utils/filters/conditional_filters.py index d98cebe..cdc6bf6 100644 --- a/laborious/utils/filters/conditional_filters.py +++ b/laborious/utils/filters/conditional_filters.py @@ -13,7 +13,7 @@ def filter_specific_variables_null_values(data: DataFrame, config: dict) -> bool bool: True if the specific columns have null values, False otherwise. """ return not data[ - data['variable'].isin(config['VARIABLES']) & data['value'].isna()].empty + data['variable'].isin(config['variables']) & data['value'].isna()].empty def filter_empty_data(data: DataFrame, _config: dict) -> bool: diff --git a/values.yaml b/values.yaml index b164156..4e47334 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.0.2" + tag: "0.1.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: @@ -123,7 +123,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-994-implementar-os-workflows-mapeados-utilizando-as-workers-e-activities-apropriadas" + value: "SIENTIAPDE-1097-realizar-testes-basicos-no-cluster-suse-linux" - name: PYTHON_APP value: "laborious.worker.worker" From ba96ea988ef1bdd34eaca0aa54436ce8bd1c6243 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 10 Jun 2025 08:53:05 -0300 Subject: [PATCH 3/7] Update model_id key to model in prediction_process.py --- laborious/workflows/sub_workflows/prediction_process.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index e4fc8cc..1e01f6a 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -205,7 +205,7 @@ class PredictionProcess(): { 'schema': schema, 'table_name': table_name, - 'model_id': model_id + 'model': model_id }, retry_policy=retry_policy, start_to_close_timeout=timedelta(minutes=1), From c02dae19a00dc0727570d28ad9eb59f90520c95c Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 10 Jun 2025 09:07:36 -0300 Subject: [PATCH 4/7] SIENTIAPDE-1097 Update input filter function call and add last_timestamp to prediction_process --- laborious/activities/gates.py | 2 +- laborious/workflows/sub_workflows/prediction_process.py | 4 +++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 7bab5f5..e78c42b 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -81,7 +81,7 @@ class Gates(BaseActivity): self.logger.error(f"Filter {fil} not found") continue try: - if input_filter_functions[fil](data, config): + if input_filter_functions[fil](data, config['config']): self.logger.debug( f"Data not passed the input filter {fil}:{config}") filter_output.append(config['POLICY']) diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index 1e01f6a..1ee59ed 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -192,6 +192,7 @@ class PredictionProcess(): model_id = input_data['model_id'] model_name = input_data['model_name'] model_retention = input_data['model_retention'] + last_timestamp = input_data['last_timestamp'] path_flag = path_flag.upper() if path_flag else None @@ -205,7 +206,8 @@ class PredictionProcess(): { 'schema': schema, 'table_name': table_name, - 'model': model_id + 'model': model_id, + 'last_timestamp': last_timestamp }, retry_policy=retry_policy, start_to_close_timeout=timedelta(minutes=1), From 9430bc9424a200dc62143fab4c4163f914d17cfb Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 10 Jun 2025 09:10:09 -0300 Subject: [PATCH 5/7] Remove unused last_timestamp from input_data in prediction_process.py --- laborious/workflows/sub_workflows/prediction_process.py | 1 - 1 file changed, 1 deletion(-) diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index 1ee59ed..e3b38d8 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -192,7 +192,6 @@ class PredictionProcess(): model_id = input_data['model_id'] model_name = input_data['model_name'] model_retention = input_data['model_retention'] - last_timestamp = input_data['last_timestamp'] path_flag = path_flag.upper() if path_flag else None From 98a931393be33d432a1efacf51423218c25f8396 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 10 Jun 2025 11:38:17 -0300 Subject: [PATCH 6/7] SIENTIAPDE-1097 Update library versions in requirements.txt and bump image tag in values.yaml to 0.1.1 --- requirements.txt | 4 ++-- values.yaml | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/requirements.txt b/requirements.txt index 5604fc7..599917b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,5 +3,5 @@ psycopg2-binary sqlalchemy asyncua redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.14 -git+ssh://git@github.com/Aignosi/sientia-mlops-library.git +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.17 +git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.1 diff --git a/values.yaml b/values.yaml index 4e47334..bd46925 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.1.0" + tag: "0.1.1" # 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: From 948fecc44493ec10c911309aa431a027cc9a4819 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 10 Jun 2025 11:51:27 -0300 Subject: [PATCH 7/7] SIENTIAPDE-1097 Refactor filter configuration keys from 'POLICY' to 'policy' for consistency --- laborious/activities/gates.py | 4 ++-- tests/laborious/activities/test_gates.py | 10 +++++----- .../utils/filters/test_conditional_filters.py | 4 ++-- .../workflows/subworkflows/test_prediction_process.py | 3 ++- 4 files changed, 11 insertions(+), 10 deletions(-) diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index e78c42b..4b183a0 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -84,7 +84,7 @@ class Gates(BaseActivity): if input_filter_functions[fil](data, config['config']): self.logger.debug( f"Data not passed the input filter {fil}:{config}") - filter_output.append(config['POLICY']) + filter_output.append(config['policy']) except Exception as e: trace = traceback.format_exc() self.notification_handler.build_and_send_notification( @@ -201,7 +201,7 @@ class Gates(BaseActivity): continue try: if mlflow_content_filter_functions[fil](data, config): - filter_output.append(config['POLICY']) + filter_output.append(config['policy']) self.notification_handler.build_and_send_notification( notification_id=f"{gate_type.upper()}_GATE_CONTENT_FILTER__{fil}", message=f"Data not passed the content filter {fil}:{config}", diff --git a/tests/laborious/activities/test_gates.py b/tests/laborious/activities/test_gates.py index 6b61c81..28ec1d2 100644 --- a/tests/laborious/activities/test_gates.py +++ b/tests/laborious/activities/test_gates.py @@ -42,7 +42,7 @@ async def test_input_gate_filter_exception(mock_input_filter_functions, gates_ac side_effect=Exception("Test error")) input_data = { 'filters': { - 'EMPTY_DATA': {'POLICY': 'STOP'} + 'EMPTY_DATA': {'policy': 'STOP', 'config': {}} }, 'data': {'value': []}, 'path_priority': ['STOP', 'CONTINUE', 'REPEAT'] @@ -55,7 +55,7 @@ async def test_input_gate_filter_exception(mock_input_filter_functions, gates_ac assert result == (None, 0, "") gates_activity.notification_handler.build_and_send_notification.assert_called_once_with( notification_id="INTPUT_GATE_ERROR__EMPTY_DATA", - message="Error in filter EMPTY_DATA:{'POLICY': 'STOP'}: \n Test error", + message="Error in filter EMPTY_DATA:{'policy': 'STOP', 'config': {}}: \n Test error", block="input_gate", level=NotificationLevel.ERROR, attachment_content=ANY @@ -84,7 +84,7 @@ async def test_input_gate_with_filter(gates_activity): # Arrange input_data = { 'filters': { - 'EMPTY_DATA': {'POLICY': 'STOP'} + 'EMPTY_DATA': {'policy': 'STOP', 'config': {}} }, 'data': {'value': []}, 'path_priority': ['STOP', 'CONTINUE', 'REPEAT'] @@ -171,7 +171,7 @@ async def test_mlflow_response_gate_with_filter(gates_activity): # Arrange input_data = { 'filters': { - 'API_ERROR': {'POLICY': 'STOP'} + 'API_ERROR': {'policy': 'STOP'} }, 'data': { 'success': False, @@ -273,7 +273,7 @@ async def test_mlflow_content_gate_with_filter(gates_activity): # Arrange input_data = { 'filters': { - 'NAN_VALUES': {'POLICY': 'STOP'} + 'NAN_VALUES': {'policy': 'STOP', 'config': {}} }, 'data': {'value': [None, None, None]}, 'type': 'test', diff --git a/tests/laborious/utils/filters/test_conditional_filters.py b/tests/laborious/utils/filters/test_conditional_filters.py index edcbcd6..405bc9b 100644 --- a/tests/laborious/utils/filters/test_conditional_filters.py +++ b/tests/laborious/utils/filters/test_conditional_filters.py @@ -10,14 +10,14 @@ def test_filter_specific_variables_null_values(): assert filter_specific_variables_null_values( DataFrame( {'variable': ['variable1', 'variable2'], 'value': [1, 2]}), - config={'VARIABLES': ['variable2']}) is False + config={'variables': ['variable2']}) is False def test_filter_specific_variables_null_values_with_null_values(): assert filter_specific_variables_null_values( DataFrame( {'variable': ['variable1', 'variable2'], 'value': [1, None]}), - config={'VARIABLES': ['variable2']}) is True + config={'variables': ['variable2']}) is True def test_filter_empty_data(): diff --git a/tests/laborious/workflows/subworkflows/test_prediction_process.py b/tests/laborious/workflows/subworkflows/test_prediction_process.py index 4318379..560565e 100644 --- a/tests/laborious/workflows/subworkflows/test_prediction_process.py +++ b/tests/laborious/workflows/subworkflows/test_prediction_process.py @@ -426,7 +426,8 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process): { 'schema': schema, 'table_name': table_name, - 'model_id': model + 'model': model, + 'last_timestamp': last_timestamp, }, retry_policy=ANY, start_to_close_timeout=ANY