vitor-aignosi
4c51e7c2b3
SIENTIAPDE-1646 Merge origin/main into release/SIENTIAPDE-1646 Bring main up to date while keeping this branch's sync execution, PluginStore runtime and sientia_model (SientiaMLflowRepository/SientiaModel) structure. Resolution decisions: - OPC: keep the sync asyncua.sync implementation; port main's server_uri validation, e2e race fix (lock held until released) and the new unit tests converted to sync. - Model import workflow (.sientia bundles), joblib/transform flavors and the local MLFlowRepository (model_repository.py) are dropped, with their tests, docs, metrics, env vars and connectors config. - Worker: drift/simple_metrics stay pinned to the 'core' runtime; port main's HOSTNAME pod id, logger-based prometheus startup and sys.exit after finally. - values.yaml removed, following main (SIENTIAPDE-1945). - Deps: keep sientia_model; take main's fsspec/s3fs/botocore/boto3 pins and sientia_do>=1.12.2; drop import-only deps (pynacl, cryptography, pymongo, python-dotenv, pycaret, joblib, gcsfs). - e2e: keep sientia_data schema and scenario input files. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
324 lines
12 KiB
Python
324 lines
12 KiB
Python
"""
|
|
Laborious Worker Module
|
|
|
|
Entry process that connects to Temporal, registers Laborious activities, and runs four workers in
|
|
parallel. Each worker shares the same ``Activities`` instance (single Postgres pool, single MLflow
|
|
repository, single PluginStore handle) but polls a different task queue.
|
|
|
|
Task queues (see ``sientia_do.temporal.worker.prepare_worker``):
|
|
- ``predictions_batch-{runtime}-queue`` + sub-workflows on the same queue (ML-heavy path).
|
|
- ``minimal_retrain-{runtime}-queue`` (retrain + promote + export).
|
|
- ``drift-core-queue`` and ``simple_metrics-core-queue``: pinned to the ``core`` runtime because
|
|
drift and metrics do not load model code, so every runtime shares the same queues.
|
|
|
|
Bootstrap order:
|
|
1. Prometheus app metrics and Mongo-backed notification handler.
|
|
2. ``RUNTIME`` validation and ``PluginStore.install_runtime`` so ``SientiaModel`` code is importable.
|
|
3. ``Activities`` construction (builds ``SientiaMLflowRepository`` internally from env).
|
|
4. OPC client initialization inside activities.
|
|
5. Temporal ``Runtime`` with SDK Prometheus bind, client connect, then ``prepare_worker`` per workflow.
|
|
|
|
Shutdown closes workers, notifications, activities (pools + OPC), and clears ``app_up``.
|
|
|
|
Environment Variables:
|
|
- RUNTIME: Required non-empty string passed to ``install_runtime``.
|
|
- STORE_* / PYPI_*: Plugin store and private index (see ``build_plugin_store_config``).
|
|
- TEMPORAL_HOST, TEMPORAL_NAMESPACE: Cluster connection.
|
|
- HOSTNAME (pod id), HTTP_METRICS_PORT, HTTP_SDK_METRICS_PORT: Observability.
|
|
- PROJECT_NAME, MONGODB_*: Notifications (via ``build_mongodb_config`` in handler).
|
|
- POSTGRES_*, MINIO_*, OPC_*, PI_WEB_API_*, MLFLOW_*: Passed through ``Activities`` helpers.
|
|
"""
|
|
|
|
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.observability.metrics_controller import MetricsController
|
|
from sientia_do.temporal.worker.prepare_worker import prepare_worker
|
|
from sientia_do.utils.connectors_config import (
|
|
build_api_config,
|
|
build_mongodb_config,
|
|
build_postgres_config,
|
|
)
|
|
from sientia_model.model_repository.plugin_store import PluginStore
|
|
|
|
from laborious import metrics
|
|
from laborious.activities.activities import Activities
|
|
from laborious.utils.connectors_config import (
|
|
build_minio_config,
|
|
build_opc_config,
|
|
build_plugin_store_config,
|
|
)
|
|
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():
|
|
"""
|
|
Run the full worker lifecycle: metrics, notifications, runtime install, workers, gather.
|
|
|
|
Exits the process with code 0 on normal completion of all worker tasks, or 1 after logging
|
|
if any worker raises. ``finally`` always shuts down notifications and activities and sets
|
|
``app_up`` to 0; ``sys.exit`` runs after it.
|
|
|
|
Raises:
|
|
Exception: Propagated from ``asyncio.gather`` only before ``finally`` handling; typically
|
|
workers run until cancelled.
|
|
|
|
Return:
|
|
None (process terminates via ``sys.exit`` after the ``finally`` block).
|
|
"""
|
|
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, metadata)
|
|
|
|
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'),
|
|
)
|
|
|
|
metrics_controller = MetricsController(logger=logger)
|
|
|
|
runtime = os.getenv('RUNTIME', '').strip()
|
|
if not runtime:
|
|
logger.custom_critical(
|
|
'RUNTIME environment variable is required and must be non-empty',
|
|
metadata,
|
|
)
|
|
metrics.APP_UP.labels(pod_id=POD_ID).set(0)
|
|
sys.exit(1)
|
|
|
|
metadata_runtime = {**metadata, 'runtime': runtime}
|
|
logger.custom_info(f'Installing PluginStore runtime: {runtime}', metadata_runtime)
|
|
|
|
ps_cfg = build_plugin_store_config()
|
|
plugin_store = PluginStore(
|
|
base_url=ps_cfg['base_url'],
|
|
owner=ps_cfg['owner'],
|
|
repo=ps_cfg['repo'],
|
|
username=ps_cfg['username'],
|
|
password=ps_cfg['password'],
|
|
branch=ps_cfg['branch'],
|
|
cache_ttl_seconds=ps_cfg['cache_ttl_seconds'],
|
|
pypi_index_url=ps_cfg['pypi_index_url'],
|
|
pypi_username=ps_cfg['pypi_username'],
|
|
pypi_password=ps_cfg['pypi_password'],
|
|
logger=logger,
|
|
notification_handler=notification_handler,
|
|
metrics_controller=metrics_controller,
|
|
)
|
|
|
|
if runtime == 'legacy':
|
|
to_install_runtime = 'single'
|
|
else:
|
|
to_install_runtime = runtime
|
|
|
|
try:
|
|
await plugin_store.install_runtime(
|
|
runtime_name=to_install_runtime, metadata=metadata_runtime
|
|
)
|
|
except Exception as exc:
|
|
logger.custom_critical(
|
|
f'Failed to install runtime {to_install_runtime}: {exc}', metadata_runtime
|
|
)
|
|
metrics.APP_UP.labels(pod_id=POD_ID).set(0)
|
|
sys.exit(1)
|
|
|
|
logger.custom_info('Starting Activities...', metadata_runtime)
|
|
|
|
activities = Activities(
|
|
postgres_config=build_postgres_config(),
|
|
plugin_store=plugin_store,
|
|
minio_config=build_minio_config(),
|
|
opc_config=build_opc_config(),
|
|
pi_web_api_config=build_api_config(),
|
|
logger=logger,
|
|
notification_handler=notification_handler,
|
|
metrics_controller=metrics_controller,
|
|
)
|
|
|
|
logger.custom_info('Initializing OPC...', metadata_runtime)
|
|
activities.init_opc()
|
|
|
|
logger.custom_info(
|
|
f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...',
|
|
metadata_runtime,
|
|
)
|
|
|
|
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_runtime)
|
|
|
|
temporal_client = await client.Client.connect(
|
|
target_host=host,
|
|
namespace=os.getenv('TEMPORAL_NAMESPACE', 'laborious'),
|
|
runtime=new_runtime,
|
|
)
|
|
|
|
logger.custom_info(f'Starting Workers (runtime={runtime})...', metadata_runtime)
|
|
|
|
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,
|
|
runtime=runtime,
|
|
),
|
|
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,
|
|
runtime='core',
|
|
),
|
|
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,
|
|
runtime='core',
|
|
),
|
|
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,
|
|
runtime=runtime,
|
|
),
|
|
]
|
|
|
|
handlers = []
|
|
for w in workers:
|
|
handlers.append(w.run())
|
|
|
|
logger.custom_info('Workers started successfully', metadata_runtime)
|
|
|
|
exit_code = 0
|
|
try:
|
|
await asyncio.gather(*handlers)
|
|
except Exception as e:
|
|
logger.custom_error(f'An unhandled exception occurred: {e}', metadata_runtime)
|
|
exit_code = 1
|
|
finally:
|
|
# No `except (SystemExit, KeyboardInterrupt)` on purpose: `finally` runs on every path,
|
|
# caught or not, so cleanup still happens for a deliberate shutdown (Ctrl+C, a handler's
|
|
# own sys.exit) without us having to catch and re-raise it ourselves — which would only
|
|
# let us relabel its exit code before `sys.exit(exit_code)` below discards it anyway.
|
|
# Letting it propagate on its own keeps the original code (e.g. a handler's sys.exit(N))
|
|
# intact instead of collapsing it to 0.
|
|
notification_handler.shutdown()
|
|
activities.shutdown()
|
|
metrics.APP_UP.labels(pod_id=POD_ID).set(0) # Mark app as DOWN
|
|
|
|
sys.exit(exit_code)
|
|
|
|
|
|
def start_prometheus_server(logger, metadata: dict):
|
|
"""
|
|
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)
|
|
HOSTNAME: 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)
|
|
logger.custom_info(f'Prometheus server started on port {port}.', metadata)
|
|
metrics.APP_UP.labels(pod_id=POD_ID).set(1) # Mark app as UP
|
|
except Exception as e:
|
|
logger.custom_critical(f'Failed to start Prometheus server: {e}', metadata)
|
|
os._exit(1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(main())
|