vitor-aignosi
cd1be2430a
Update .gitignore, requirements, and enhance OPC UA error handling - Added new entries to .gitignore for openspec and cursor directories. - Updated sientia-dataops-library dependency version in requirements-light.txt from 1.10.4 to 1.12.0. - Enhanced OPC UA communication by refining reconnect logic and error handling in opc_repository.py, including the introduction of a reconnect flag and improved session management. - Updated tests to cover new reconnect scenarios and ensure robust error handling for protocol states.
269 lines
9.4 KiB
Python
269 lines
9.4 KiB
Python
"""
|
|
Laborious Worker Module
|
|
|
|
This module provides the main worker implementation for the Sientia DataOps Laborious system.
|
|
It orchestrates Temporal workers, manages task queues, and handles the lifecycle of
|
|
prediction and retraining workflows.
|
|
|
|
The worker supports multiple task queues:
|
|
- predictions_batch-queue: Handles batch prediction workflows (heavy workload)
|
|
Includes activities for MLFlow, data quality gates, OPC export, PI Web API export, and PostgreSQL
|
|
- minimal_retrain-queue: Handles model retraining workflows
|
|
- drift-queue: Handles drift detection workflows
|
|
- simple_metrics-queue: Handles simple metrics calculation workflows
|
|
|
|
Key Features:
|
|
- Resource-based scaling with WorkerTuner (CPU and memory aware)
|
|
- Automatic polling scaling with PollerBehaviorAutoscaling
|
|
- Prometheus metrics integration
|
|
- Comprehensive error handling and logging
|
|
- Graceful shutdown with cleanup
|
|
- Multiple worker instances for different workflow types
|
|
|
|
Environment Variables:
|
|
- TEMPORAL_HOST: Temporal server address (default: localhost:7233)
|
|
- TEMPORAL_NAMESPACE: Temporal namespace (default: laborious)
|
|
- POD_ID: Kubernetes pod identifier for metrics
|
|
- HTTP_METRICS_PORT: Prometheus metrics server port (default: 9090)
|
|
- HTTP_SDK_METRICS_PORT: Temporal SDK metrics port (default: 9091)
|
|
- PROJECT_NAME: Project name for notifications (default: laborious)
|
|
"""
|
|
|
|
from temporalio import client, workflow
|
|
from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig
|
|
|
|
with workflow.unsafe.imports_passed_through():
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
|
|
from prometheus_client import start_http_server
|
|
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
|
|
from sientia_do.observability.logger import get_logger
|
|
from sientia_do.utils.connectors_config import (
|
|
build_api_config,
|
|
build_mongodb_config,
|
|
build_postgres_config,
|
|
)
|
|
|
|
from laborious import metrics
|
|
from laborious.activities.activities import Activities
|
|
from laborious.utils.connectors_config import (
|
|
build_minio_config,
|
|
build_mlflow_config,
|
|
build_opc_config,
|
|
)
|
|
from laborious.worker.prepare_worker import prepare_worker
|
|
from laborious.workflows.drift import Drift
|
|
from laborious.workflows.minimal_retrain import MinimalRetrain
|
|
from laborious.workflows.predictions_batch import PredictionsBatch
|
|
from laborious.workflows.simple_metrics import SimpleMetrics
|
|
from laborious.workflows.sub_workflows.format_and_export_prediction import (
|
|
FormatAndExportPrediction,
|
|
)
|
|
from laborious.workflows.sub_workflows.prediction_process import PredictionProcess
|
|
|
|
POD_ID = os.getenv('HOSTNAME')
|
|
SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', '9091'))
|
|
|
|
|
|
async def main():
|
|
"""
|
|
Main entry point for the Laborious worker application.
|
|
|
|
This function initializes and starts all components of the worker:
|
|
1. Sets up logging and metadata
|
|
2. Starts Prometheus metrics server
|
|
3. Initializes notification handler
|
|
4. Creates and configures activities
|
|
5. Initializes OPC connections
|
|
6. Starts Temporal client and workers
|
|
7. Manages worker lifecycle and graceful shutdown
|
|
|
|
The function runs indefinitely until interrupted or an error occurs.
|
|
On error, it performs cleanup and exits with a non-zero status code.
|
|
|
|
Raises:
|
|
Exception: Any unhandled exception during worker execution
|
|
SystemExit: On graceful shutdown or error conditions
|
|
"""
|
|
host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
|
|
logger = get_logger(__name__)
|
|
|
|
metadata = {
|
|
'pod_id': POD_ID,
|
|
'model_name': '-',
|
|
'model_id': '-',
|
|
'workflow_name': '-',
|
|
'schedule_name': '-',
|
|
}
|
|
|
|
logger.custom_info(f'Starting Worker with POD_ID: {POD_ID}', metadata)
|
|
|
|
logger.custom_info('Starting prometheus client...', metadata)
|
|
start_prometheus_server()
|
|
|
|
logger.custom_info('Starting Notification Handler...', metadata)
|
|
|
|
mongo_config = build_mongodb_config()
|
|
notification_handler = NotificationHandler(
|
|
connection_string=mongo_config['connection_string'],
|
|
database=mongo_config['database_name'],
|
|
logger=logger,
|
|
project_name=os.getenv('PROJECT_NAME', 'laborious'),
|
|
)
|
|
|
|
logger.custom_info('Starting Activities...', metadata)
|
|
|
|
activities = Activities(
|
|
postgres_config=build_postgres_config(),
|
|
mlflow_config=build_mlflow_config(),
|
|
minio_config=build_minio_config(),
|
|
opc_config=build_opc_config(),
|
|
pi_web_api_config=build_api_config(),
|
|
logger=logger,
|
|
notification_handler=notification_handler,
|
|
)
|
|
|
|
logger.custom_info('Initializing OPC...', metadata)
|
|
await activities.init_opc()
|
|
|
|
logger.custom_info(f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata)
|
|
|
|
new_runtime = Runtime(
|
|
telemetry=TelemetryConfig(
|
|
metrics=PrometheusConfig(bind_address=f'0.0.0.0:{SDK_METRICS_PORT}')
|
|
)
|
|
)
|
|
|
|
logger.custom_info(f'Starting Temporal Client at {host}...', metadata)
|
|
|
|
temporal_client = await client.Client.connect(
|
|
target_host=host,
|
|
namespace=os.getenv('TEMPORAL_NAMESPACE', 'laborious'),
|
|
runtime=new_runtime,
|
|
)
|
|
|
|
logger.custom_info('Starting Workers...', metadata)
|
|
|
|
workers = [
|
|
prepare_worker(
|
|
temporal_client=temporal_client,
|
|
main_workflow=MinimalRetrain,
|
|
other_workflows=[],
|
|
activities=[
|
|
activities.load_query_with_minio_offload,
|
|
activities.retrain_model,
|
|
activities.update_production_model,
|
|
activities.format_retrain_report,
|
|
activities.export_data_to_postgres,
|
|
],
|
|
logger=logger,
|
|
),
|
|
prepare_worker(
|
|
temporal_client=temporal_client,
|
|
main_workflow=SimpleMetrics,
|
|
other_workflows=[],
|
|
activities=[
|
|
activities.load_custom_query,
|
|
activities.calculate_simple_metrics,
|
|
activities.export_data_to_postgres,
|
|
],
|
|
logger=logger,
|
|
),
|
|
prepare_worker(
|
|
temporal_client=temporal_client,
|
|
main_workflow=Drift,
|
|
other_workflows=[],
|
|
activities=[
|
|
activities.load_custom_query,
|
|
activities.get_reference_data,
|
|
activities.calculate_drift,
|
|
activities.export_data_to_postgres,
|
|
],
|
|
logger=logger,
|
|
),
|
|
prepare_worker(
|
|
temporal_client=temporal_client,
|
|
main_workflow=PredictionsBatch,
|
|
other_workflows=[PredictionProcess, FormatAndExportPrediction],
|
|
activities=[
|
|
# MLFlow
|
|
activities.request_predict,
|
|
activities.request_transform,
|
|
# Gates
|
|
activities.input_gate,
|
|
activities.mlflow_response_gate,
|
|
activities.mlflow_content_gate,
|
|
activities.format_transformed_data,
|
|
activities.format_prediction,
|
|
activities.format_default_prediction,
|
|
# OPC
|
|
activities.write_opc_data,
|
|
# Postgres / MinIO offload
|
|
activities.load_query_with_minio_offload,
|
|
activities.cleanup_minio_objects_expired,
|
|
activities.repeat_last_prediction,
|
|
activities.export_data_to_postgres,
|
|
activities.export_payload_to_postgres,
|
|
activities.write_metrics,
|
|
# Pi Web API
|
|
activities.write_pi_web_api_data,
|
|
],
|
|
logger=logger,
|
|
),
|
|
]
|
|
|
|
handlers = []
|
|
for w in workers:
|
|
handlers.append(w.run())
|
|
|
|
logger.custom_info('Workers started successfully', metadata)
|
|
|
|
exit_code = 0
|
|
try:
|
|
# This will run the workers and wait for them to complete.
|
|
# If an exception occurs in any of the worker handlers, it will be propagated here.
|
|
await asyncio.gather(*handlers)
|
|
except BaseException as e: # NOSONAR
|
|
logger.custom_error(f'An unhandled exception occurred: {e}', metadata)
|
|
exit_code = 1
|
|
finally:
|
|
if notification_handler:
|
|
notification_handler.shutdown()
|
|
if activities:
|
|
await activities.shutdown()
|
|
metrics.APP_UP.labels(pod_id=POD_ID).set(0) # Mark app as DOWN
|
|
sys.exit(exit_code)
|
|
|
|
|
|
def start_prometheus_server():
|
|
"""
|
|
Starts the Prometheus metrics server for monitoring and observability.
|
|
|
|
This function initializes the Prometheus HTTP server on the configured port
|
|
and sets the application health metric to indicate the service is running.
|
|
|
|
The server exposes metrics that can be scraped by Prometheus for monitoring
|
|
the health and performance of the Laborious worker.
|
|
|
|
Environment Variables:
|
|
HTTP_METRICS_PORT: Port for the metrics server (default: 9090)
|
|
POD_ID: Pod identifier for metrics labeling
|
|
|
|
Raises:
|
|
SystemExit: If the metrics server fails to start
|
|
"""
|
|
try:
|
|
port = int(os.getenv('HTTP_METRICS_PORT', 9090))
|
|
start_http_server(port)
|
|
print(f'Prometheus server started on port {port}.')
|
|
metrics.APP_UP.labels(pod_id=POD_ID).set(1) # Mark app as UP
|
|
except Exception as e:
|
|
print(f'Failed to start Prometheus server: {e}')
|
|
os._exit(1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(main())
|