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

SIENTIAPDE-1110

Browse Source 
Update dependencies and enhance logging in activities; bump version in requirements and values.yaml
This commit is contained in:
vitor-aignosi
2025-06-27 09:17:24 -03:00
parent cefef0b1e9
commit 98ea2a7f75
10 changed files with 87 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
laborious/activities/activities.py
View File

@@ -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)

52
laborious/activities/gates.py
View File

@@ -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],

24
laborious/activities/mlflow.py
View File

@@ -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

15
laborious/activities/opc.py
View File

@@ -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()

4
laborious/worker/worker.py
View File

@@ -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()

17
laborious/workflows/predictions_batch.py
View File

@@ -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'],

5
laborious/workflows/sub_workflows/format_and_export_prediction.py
View File

@@ -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,

15
laborious/workflows/sub_workflows/prediction_process.py
View File

@@ -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,

2
requirements.txt
View File

@@ -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

6
values.yaml
View File

@@ -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 \