diff --git a/laborious/activities/activities.py b/laborious/activities/activities.py index d3c9436..2156009 100644 --- a/laborious/activities/activities.py +++ b/laborious/activities/activities.py @@ -3,11 +3,11 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): from sientia_do.temporal.activities.postgres import Postgres from sientia_do.notifications.handlers import NotificationHandler + from sientia_do.temporal.utils.logger import Logger from laborious.activities.mlflow import MLFlow from laborious.activities.gates import Gates from laborious.activities.opc import OPC from typing import Any - from logging import Logger class Activities(Postgres, MLFlow, Gates, OPC): @@ -44,10 +44,6 @@ class Activities(Postgres, MLFlow, Gates, OPC): logger=logger, notification_handler=notification_handler) - @activity.defn(name="prepare_activity") - async def prepare_activity(self, input_data: dict[str, Any]): - await super().prepare_activity(input_data) - def shutdown(self): Postgres.close(self) OPC.shutdown(self) diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index d744578..b9ab505 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -3,10 +3,10 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): import traceback - from logging import Logger from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.temporal.activities.base import BaseActivity + from sientia_do.temporal.utils.logger import Logger from laborious.utils.filters.mlflow_filters import nan_values_filter, api_error_filter from typing import Any from laborious.utils.filters.conditional_filters import ( @@ -65,7 +65,11 @@ class Gates(BaseActivity): list and filter configuration and functions. """ - self.logger.debug("Performing input gate...") + metadata = input_data['metadata'] + + self.debug("Performing input gate...", metadata) + + self.debug(f"Input data: {input_data}", metadata) filters = input_data['filters'] data = DataFrame(input_data['data']) @@ -73,17 +77,17 @@ class Gates(BaseActivity): filter_output = [] - self.logger.debug(f"Input data:\n {data}") - self.logger.debug(f"Filters: {filters}") + self.debug(f"Input data:\n {data}", metadata) + self.debug(f"Filters: {filters}", metadata) for fil, config in filters.items(): if fil not in input_filter_functions: - self.logger.error(f"Filter {fil} not found") + self.error(f"Filter {fil} not found", metadata) continue try: if input_filter_functions[fil](data, config['config']): - self.logger.debug( - f"Data not passed the input filter {fil}:{config}") + self.debug( + f"Data not passed the input filter {fil}:{config}", metadata) filter_output.append(config['policy']) except Exception as e: trace = traceback.format_exc() @@ -97,11 +101,11 @@ class Gates(BaseActivity): for path_flag in path_priority: if path_flag in filter_output: - self.logger.debug(f"Input gate result: {path_flag}") + self.debug(f"Input gate result: {path_flag}", metadata) return path_flag, input_filter_functions['path_confidence'][path_flag], \ "Input data with bad quality" - self.logger.debug("Nothing was filtered by the input gate") + self.debug("Nothing was filtered by the input gate", metadata) return None, 0, "" @activity.defn(name="mlflow_response_gate") @@ -121,7 +125,8 @@ class Gates(BaseActivity): and filter configuration and functions. """ - self.logger.debug("Performing mlflow response gate...") + metadata = input_data['metadata'] + self.debug("Performing mlflow response gate...", metadata) filters = input_data['filters'] data = input_data['data'] @@ -130,8 +135,8 @@ class Gates(BaseActivity): filter_output = [] - self.logger.debug(f"Input data:\n {data}") - self.logger.debug(f"Filters: {filters}") + self.debug(f"Input data:\n {data}", metadata) + self.debug(f"Filters: {filters}", metadata) comments = [] for fil, config in filters.items(): @@ -160,11 +165,12 @@ class Gates(BaseActivity): for path_flag in path_priority: if path_flag in filter_output: - self.logger.debug(f"Mlflow response gate result: {path_flag}") + self.debug( + f"Mlflow response gate result: {path_flag}", metadata) return path_flag, mlflow_response_filter_functions['path_confidence'][path_flag], \ ", ".join(comments) - self.logger.debug("Nothing was filtered by the mlflow response gate") + self.debug("Nothing was filtered by the mlflow response gate", metadata) return None, 0, "" @activity.defn(name="mlflow_content_gate") @@ -184,7 +190,8 @@ class Gates(BaseActivity): list and filter configuration and functions. """ - self.logger.debug("Performing mlflow content gate...") + metadata = input_data['metadata'] + self.debug("Performing mlflow content gate...", metadata) filters = input_data['filters'] data = DataFrame(input_data['data']) @@ -193,8 +200,8 @@ class Gates(BaseActivity): filter_output = [] - self.logger.debug(f"Input data:\n {data}") - self.logger.debug(f"Filters: {filters}") + self.debug(f"Input data:\n {data}", metadata) + self.debug(f"Filters: {filters}", metadata) for fil, config in filters.items(): if fil not in mlflow_content_filter_functions: @@ -221,11 +228,12 @@ class Gates(BaseActivity): for path_flag in path_priority: if path_flag in filter_output: - self.logger.debug(f"Mlflow content gate result: {path_flag}") + self.debug( + f"Mlflow content gate result: {path_flag}", metadata) return path_flag, mlflow_content_filter_functions['path_confidence'][path_flag], \ "Transformed data not passed the content filter" - self.logger.debug("Nothing was filtered by the mlflow content gate") + self.debug("Nothing was filtered by the mlflow content gate", metadata) return None, 0, "" @activity.defn(name="format_prediction") @@ -241,7 +249,8 @@ class Gates(BaseActivity): Returns: dict: The formatted data. """ - self.logger.debug("Formatting prediction...") + metadata = input_data['metadata'] + self.debug("Formatting prediction...", metadata) data = DataFrame(input_data['data']) data['timestamp'] = input_data['timestamp'] @@ -269,7 +278,8 @@ class Gates(BaseActivity): dict: The formatted data. """ - self.logger.debug("Formatting default prediction...") + metadata = input_data['metadata'] + self.debug("Formatting default prediction...", metadata) return DataFrame({ 'prediction': [0], diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index b336f45..4b8c129 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -6,9 +6,9 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): from sientia_do.temporal.activities.base import BaseActivity from sientia_do.notifications.handlers import NotificationHandler + from sientia_do.temporal.utils.logger import Logger from laborious.utils.repository.model_repository import MLFlowRepository from typing import Any - from logging import Logger class MLFlow(BaseActivity): @@ -36,13 +36,14 @@ class MLFlow(BaseActivity): Returns: dict[str, Any]: The transformed data. """ - self.logger.info('Transforming data...') + metadata = input_data['metadata'] + self.debug('Transforming data...', metadata) data = DataFrame(input_data['data']) model_name = input_data['model_name'] model_retention = input_data['model_retention'] - self.logger.debug("Raw input data:") - self.logger.debug(data) + self.debug("Raw input data:", metadata) + self.debug(data, metadata) # Sort by created_at in descending order and keep first occurrence of each variable/timestamp pair data = data.sort_values('created_at', ascending=False).drop_duplicates( @@ -56,14 +57,14 @@ class MLFlow(BaseActivity): data.reset_index(inplace=True) data.columns.name = None - self.logger.debug("Processed input data:") - self.logger.debug(data) + self.debug("Processed input data:", metadata) + self.debug(data, metadata) response_data = self.model_monitoring_repository.transform( model_name, data, model_retention) - self.logger.debug("Response data:") - self.logger.debug(response_data) + self.debug("Response data:", metadata) + self.debug(response_data, metadata) return response_data @@ -79,18 +80,19 @@ class MLFlow(BaseActivity): Returns: dict[str, Any]: The predicted data. """ - self.logger.info('Predicting data...') + metadata = input_data['metadata'] + self.debug('Predicting data...', metadata) data = DataFrame(input_data['data']) model_name = input_data['model_name'] model_retention = input_data['model_retention'] - self.logger.debug(data) + self.debug(data, metadata) data.replace(np.nan, None, inplace=True) response_data = self.model_monitoring_repository.predict( model_name, data, model_retention) - self.logger.debug(response_data) + self.debug(response_data, metadata) return response_data diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 93c6b7e..3150288 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -2,10 +2,10 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): - from logging import Logger from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.temporal.activities.base import BaseActivity + from sientia_do.temporal.utils.logger import Logger from laborious.utils.repository.opc_repository import OpcRepository from typing import Any import traceback @@ -89,16 +89,17 @@ class OPC(BaseActivity): - dict[Any, Any]: The data that was written to the OPC servers. """ - self.logger.debug("Writing data to OPC servers...") + metadata = input_data['metadata'] + self.debug("Writing data to OPC servers...", metadata) data = DataFrame(input_data['data']) opc_output_config = input_data['opc_output_config'] - self.logger.debug(data) + self.debug(data, metadata) success = True for server, config in opc_output_config.items(): if self.opc_repository.get(server) is None: - self.logger.error(f"OPC server {server} not found") + self.error(f"OPC server {server} not found", metadata) continue if 'prediction_tags' in config: @@ -137,15 +138,17 @@ class OPC(BaseActivity): Returns: dict[Any, Any]: The processed data as a dictionary. """ + metadata = data['metadata'] + if not success: data['prediction_confidence'] = OPC_WRITTING_ERROR_CONFIDENCE - self.logger.debug( + self.debug( "Some data could not be written to OPC servers, setting confidence to " f"{OPC_WRITTING_ERROR_CONFIDENCE}." ) else: - self.logger.info("Data written to OPC servers successfully.") + self.debug("Data written to OPC servers successfully.", metadata) return data.to_dict() diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index ae5e47c..e847108 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -60,8 +60,6 @@ async def main(): workflows=[PredictionsBatch, PredictionProcess, FormatAndExportPrediction], activities=[ - # Base - activities.prepare_activity, # MLFlow activities.request_predict, activities.request_transform, @@ -93,7 +91,7 @@ async def main(): # If an exception occurs in any of the worker handlers, it will be propagated here. await asyncio.gather(*handlers) except BaseException as e: - logger.error("An unhandled exception occurred: %s", e, exc_info=True) + logger.error(f"An unhandled exception occurred: {e}") finally: if notification_handler: notification_handler.shutdown() diff --git a/laborious/workflows/predictions_batch.py b/laborious/workflows/predictions_batch.py index d80a9d4..f643c7d 100644 --- a/laborious/workflows/predictions_batch.py +++ b/laborious/workflows/predictions_batch.py @@ -40,27 +40,28 @@ class PredictionsBatch(): Exception: If any of the required parameters are missing or if the workflow fails. """ - await workflow.execute_local_activity_method( - Activities.prepare_activity, - { + metadata = { + 'metadata': { 'schedule_name': input_data['schedule_name'], 'model_name': input_data['model_name'], 'model_id': input_data['model_id'], 'workflow_name': 'predictions_batch' - }, - retry_policy=retry_policy, - start_to_close_timeout=timedelta(seconds=60) - ) + } + } data = await workflow.execute_local_activity_method( Activities.load_custom_query, - input_data['query'], + { + **metadata, + 'query': input_data['query'], + }, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60) ) # Prepare input for prediction_process workflow prediction_input = { + **metadata, 'data': data, 'schema': input_data['schema'], 'table_name': input_data['table_name'], diff --git a/laborious/workflows/sub_workflows/format_and_export_prediction.py b/laborious/workflows/sub_workflows/format_and_export_prediction.py index a33f71f..b021bfd 100644 --- a/laborious/workflows/sub_workflows/format_and_export_prediction.py +++ b/laborious/workflows/sub_workflows/format_and_export_prediction.py @@ -36,6 +36,7 @@ class FormatAndExportPrediction(): Returns: bool: True if the workflow was successful, False otherwise. """ + metadata = input_data['metadata'] path_flag = input_data['path_flag'] data = input_data['data'] prediction_confidence = input_data['prediction_confidence'] @@ -45,6 +46,7 @@ class FormatAndExportPrediction(): prediction = await workflow.execute_local_activity_method( Activities.format_prediction, { + **metadata, 'data': data, 'timestamp': input_data['timestamp'], 'model_id': input_data['model_id'], @@ -59,6 +61,7 @@ class FormatAndExportPrediction(): prediction = await workflow.execute_local_activity_method( Activities.format_default_prediction, { + **metadata, 'timestamp': input_data['timestamp'], 'model_id': input_data['model_id'], 'prediction_confidence': prediction_confidence, @@ -72,6 +75,7 @@ class FormatAndExportPrediction(): prediction = await workflow.execute_activity_method( Activities.write_opc_data, { + **metadata, 'opc_output_config': input_data['opc_output_config'], 'data': prediction }, @@ -83,6 +87,7 @@ class FormatAndExportPrediction(): await workflow.execute_activity_method( Activities.export_data_to_postgres, { + **metadata, 'schema': input_data['schema'], 'table_name': input_data['table_name'], 'data': prediction, diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index e3b38d8..3d5bef8 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -41,6 +41,7 @@ class PredictionProcess(): Exception: If any of the required parameters are missing or if the workflow fails. """ + metadata = input_data['metadata'] data = input_data['data'] model_id = input_data['model_id'] model_name = input_data['model_name'] @@ -49,6 +50,7 @@ class PredictionProcess(): last_timestamp = await workflow.execute_local_activity_method( Activities.get_last_timestamp, { + **metadata, 'data': data }, retry_policy=retry_policy, @@ -58,6 +60,7 @@ class PredictionProcess(): path_flag, confidence, comment = await workflow.execute_local_activity_method( Activities.input_gate, { + **metadata, 'filters': input_data['input_filters'], 'data': data, 'path_priority': input_data['path_priority'] @@ -74,6 +77,7 @@ class PredictionProcess(): response_data = await workflow.execute_local_activity_method( Activities.request_transform, { + **metadata, 'data': data, 'model_name': model_name, 'model_retention': model_retention @@ -85,6 +89,7 @@ class PredictionProcess(): path_flag, confidence, comment = await workflow.execute_local_activity_method( Activities.mlflow_response_gate, { + **metadata, 'filters': input_data['mlflow_transform_filters'], 'data': response_data, 'type': 'transform', @@ -104,6 +109,7 @@ class PredictionProcess(): path_flag, confidence, comment = await workflow.execute_local_activity_method( Activities.mlflow_content_gate, { + **metadata, 'filters': input_data['mlflow_transform_filters'], 'data': transformed_data, 'type': 'transform', @@ -121,6 +127,7 @@ class PredictionProcess(): response_data = await workflow.execute_local_activity_method( Activities.request_predict, { + **metadata, 'data': transformed_data, 'model_name': model_name, 'model_retention': model_retention @@ -132,6 +139,7 @@ class PredictionProcess(): path_flag, confidence, comment = await workflow.execute_local_activity_method( Activities.mlflow_response_gate, { + **metadata, 'filters': input_data['mlflow_predict_filters'], 'data': response_data, 'type': 'predict', @@ -149,6 +157,7 @@ class PredictionProcess(): await workflow.execute_child_workflow( 'format_and_export_prediction', { + **metadata, 'path_flag': path_flag, 'data': response_data['content'], 'prediction_confidence': confidence, @@ -187,13 +196,15 @@ class PredictionProcess(): bool: True if the prediction should be stopped, False otherwise. """ + metadata = input_data['metadata'] + schema = input_data['schema'] table_name = input_data['table_name'] model_id = input_data['model_id'] model_name = input_data['model_name'] model_retention = input_data['model_retention'] - path_flag = path_flag.upper() if path_flag else None + path_flag = path_flag.upper() if path_flag else '' if path_flag == 'STOP': return True @@ -203,6 +214,7 @@ class PredictionProcess(): await workflow.execute_activity_method( Activities.repeat_last_prediction, { + **metadata, 'schema': schema, 'table_name': table_name, 'model': model_id, @@ -218,6 +230,7 @@ class PredictionProcess(): await workflow.execute_child_workflow( 'format_and_export_prediction', { + **metadata, 'path_flag': path_flag, 'data': data, 'prediction_confidence': confidence, diff --git a/requirements.txt b/requirements.txt index 599917b..c8054ba 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.17 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.2.0 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.1 diff --git a/values.yaml b/values.yaml index ba03752..7b657e4 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.1" + tag: "0.2.2" # 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-1097-realizar-testes-basicos-no-cluster-suse-linux" + value: "SIENTIAPDE-1110-criar-testes-e-2-e" - name: PYTHON_APP value: "laborious.worker.worker" @@ -178,7 +178,7 @@ ssh: # kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp -# helm upgrade --install sientia-laborious-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.1.0-uat +# helm upgrade --install sientia-laborious-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.4.0-uat # kubectl create secret generic git-ssh-key-sientia-laborious-worker \ # --namespace sientia \