diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index bcb39de..79249d6 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -8,7 +8,7 @@ on: jobs: quality-gate: - uses: Aignosi/github_workflow_templates/.github/workflows/dataops-module-quality-gate.yml@main + uses: Aignosi/github_workflow_templates/.github/workflows/python-quality-gate.yml@main permissions: write-all with: project_name: 'laborious' diff --git a/README.md b/README.md index 0de066f..0b5c844 100644 --- a/README.md +++ b/README.md @@ -126,10 +126,15 @@ Laborious uses a Temporal-based architecture with strong separation of concerns ### Key Components #### **Worker (`laborious/worker/worker.py`)** -- Temporal client setup, worker lifecycle, task queues +- Temporal client setup, four workers via `sientia_do.temporal.worker.prepare_worker` +- Runtime-scoped task queues: `{workflow}-{RUNTIME}-queue` for all workflows - Metrics server initialization, notification handler setup - Graceful shutdown and autoscaling-friendly behavior +**Breaking (schedulers):** drift and simple_metrics queues are no longer `drift-queue` / +`simple_metrics-queue`. Use `drift-{RUNTIME}-queue` and `simple_metrics-{RUNTIME}-queue` +matching the worker pod `RUNTIME` env (same as `predictions_batch` / `minimal_retrain`). + #### **Workflows (`laborious/workflows/`)** - `predictions_batch.py`: Batch prediction entry point - `sub_workflows/prediction_process.py`: Core prediction pipeline @@ -802,6 +807,7 @@ See [OPC UA Communication](#opc-ua-communication) for semantics, concurrency, an |----------|-------------|---------|----------| | `TEMPORAL_HOST` | Temporal server address | `localhost:7233` | Yes | | `TEMPORAL_NAMESPACE` | Temporal namespace | `laborious` | No | +| `RUNTIME` | Task queue suffix for all workflows (`{workflow}-{RUNTIME}-queue`) | _(none)_ | Yes | | `POSTGRES_HOST` | PostgreSQL hostname | `localhost` | Yes | | `POSTGRES_PORT` | PostgreSQL port | `5432` | Yes | | `POSTGRES_USER` | PostgreSQL username | `sientia` | Yes | @@ -1097,8 +1103,7 @@ laborious/ │ ├── prediction_process.py # Core prediction workflow │ └── format_and_export_prediction.py # Export workflow ├── worker/ # Worker implementation -│ ├── worker.py # Main worker orchestrator -│ └── prepare_worker.py # Worker factory with autoscaling config +│ └── worker.py # Main worker orchestrator (uses sientia_do prepare_worker) ├── utils/ # Utility functions │ ├── connectors_config.py # Environment-driven config builders │ ├── models/ # Data models @@ -1179,7 +1184,9 @@ export LOG_LEVEL=DEBUG ### Scaling Considerations - **Horizontal Scaling**: Deploy multiple worker instances -- **Task Queue Distribution**: Use multiple task queues for different workflow types +- **Task Queue Distribution**: One worker pod per `RUNTIME`; queues are + `predictions_batch-{RUNTIME}-queue`, `minimal_retrain-{RUNTIME}-queue`, + `drift-{RUNTIME}-queue`, `simple_metrics-{RUNTIME}-queue` - **Database Performance**: Optimize indexes and connection pooling - **MLFlow Performance**: Configure appropriate model serving resources diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index b4fdad5..6a25b8c 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -170,8 +170,19 @@ class MLFlowRepository(SientiaMonitoring): # Sort by version number to get the latest latest_version = max(stage_versions, key=lambda v: int(v.version)) - run_id = latest_version.source.split('/') - return run_id[2] + source = latest_version.source + if source is None: + raise mlflow.exceptions.MlflowException( + f"Model '{model_name}' version '{latest_version.version}' in stage '{stage}' " + 'has no source URI to resolve run ID.' + ) + parts = source.split('/') + if len(parts) <= 2 or not parts[2]: + raise mlflow.exceptions.MlflowException( + f"Model '{model_name}' version '{latest_version.version}' in stage '{stage}' " + f"has invalid source URI '{source}' for run ID resolution." + ) + return parts[2] def get_next_run_name(self, model_name: str) -> str: """ diff --git a/laborious/worker/prepare_worker.py b/laborious/worker/prepare_worker.py deleted file mode 100644 index 9766af2..0000000 --- a/laborious/worker/prepare_worker.py +++ /dev/null @@ -1,73 +0,0 @@ -import os -import re -from collections.abc import Sequence -from typing import Any - -from sientia_do.observability.logger import Logger -from temporalio.client import Client -from temporalio.worker import PollerBehaviorAutoscaling, Worker - -parameters = [ - ('MAX_CONCURRENT_WORKFLOW_TASKS', '200'), - ('MAX_CONCURRENT_ACTIVITIES', '200'), - ('MAX_CONCURRENT_LOCAL_ACTIVITIES', '200'), - ('MAX_CACHED_WORKFLOWS', '200'), - ('WORKFLOW_POLLER_BEHAVIOUR_MINIMUM', '10'), - ('WORKFLOW_POLLER_BEHAVIOUR_INITIAL', '100'), - ('WORKFLOW_POLLER_BEHAVIOUR_MAXIMUM', '200'), - ('ACTIVITY_POLLER_BEHAVIOUR_MINIMUM', '10'), - ('ACTIVITY_POLLER_BEHAVIOUR_INITIAL', '100'), - ('ACTIVITY_POLLER_BEHAVIOUR_MAXIMUM', '200'), -] - - -def camel_to_snake(text: str) -> str: - """Convert camelCase or PascalCase to snake_case.""" - text = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', text) - text = re.sub('([a-z0-9])([A-Z])', r'\1_\2', text) - return text.lower() - - -def prepare_worker( - main_workflow: type, - other_workflows: Sequence[type], - activities: Sequence[Any], - temporal_client: Client, - logger: Logger, -) -> Worker: - main_workflow_name = main_workflow.__name__.upper() - - queue_name = f'{camel_to_snake(main_workflow.__name__)}-queue' - - local_workflow_parameters = {} - - for parameter in parameters: - local_workflow_parameters[parameter[0]] = int( - os.getenv(main_workflow_name + '_' + parameter[0], parameter[1]) - ) - - logger.info(f'Preparing worker for {main_workflow_name} with queue {queue_name}') - logger.info(f'Worker runtime config: {local_workflow_parameters}') - - return Worker( - temporal_client, - task_queue=queue_name, - workflows=[main_workflow, *other_workflows], - activities=[*activities], - max_concurrent_workflow_tasks=local_workflow_parameters['MAX_CONCURRENT_WORKFLOW_TASKS'], - max_concurrent_activities=local_workflow_parameters['MAX_CONCURRENT_ACTIVITIES'], - max_concurrent_local_activities=local_workflow_parameters[ - 'MAX_CONCURRENT_LOCAL_ACTIVITIES' - ], - max_cached_workflows=local_workflow_parameters['MAX_CACHED_WORKFLOWS'], - workflow_task_poller_behavior=PollerBehaviorAutoscaling( - minimum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIOUR_MINIMUM'], - initial=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIOUR_INITIAL'], - maximum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIOUR_MAXIMUM'], - ), - activity_task_poller_behavior=PollerBehaviorAutoscaling( - minimum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIOUR_MINIMUM'], - initial=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIOUR_INITIAL'], - maximum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIOUR_MAXIMUM'], - ), - ) diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 69e6d2f..7a420ba 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -5,12 +5,14 @@ This module provides the main worker implementation for the Sientia DataOps Labo 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 +The worker supports multiple runtime-scoped task queues (via ``sientia_do.temporal.worker.prepare_worker``): +- predictions_batch-{runtime}-queue: Batch prediction workflows (heavy workload) +- minimal_retrain-{runtime}-queue: Model retraining workflows +- drift-{runtime}-queue: Drift detection workflows +- simple_metrics-{runtime}-queue: Simple metrics workflows + +``RUNTIME`` must be set; it is passed to every ``prepare_worker`` call. Schedulers must use the +same queue names (breaking change vs legacy ``drift-queue`` / ``simple_metrics-queue``). Key Features: - Resource-based scaling with WorkerTuner (CPU and memory aware) @@ -21,6 +23,7 @@ Key Features: - Multiple worker instances for different workflow types Environment Variables: +- RUNTIME: Required non-empty string; suffix for all task queue names - TEMPORAL_HOST: Temporal server address (default: localhost:7233) - TEMPORAL_NAMESPACE: Temporal namespace (default: laborious) - POD_ID: Kubernetes pod identifier for metrics @@ -40,6 +43,7 @@ with workflow.unsafe.imports_passed_through(): 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.temporal.worker.prepare_worker import prepare_worker from sientia_do.utils.connectors_config import ( build_api_config, build_mongodb_config, @@ -53,7 +57,6 @@ with workflow.unsafe.imports_passed_through(): 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 @@ -100,10 +103,21 @@ async def main(): logger.custom_info(f'Starting Worker with POD_ID: {POD_ID}', metadata) - logger.custom_info('Starting prometheus client...', metadata) + 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('Starting prometheus client...', metadata_runtime) start_prometheus_server() - logger.custom_info('Starting Notification Handler...', metadata) + logger.custom_info('Starting Notification Handler...', metadata_runtime) mongo_config = build_mongodb_config() notification_handler = NotificationHandler( @@ -113,7 +127,7 @@ async def main(): project_name=os.getenv('PROJECT_NAME', 'laborious'), ) - logger.custom_info('Starting Activities...', metadata) + logger.custom_info('Starting Activities...', metadata_runtime) activities = Activities( postgres_config=build_postgres_config(), @@ -125,10 +139,13 @@ async def main(): notification_handler=notification_handler, ) - logger.custom_info('Initializing OPC...', metadata) + logger.custom_info('Initializing OPC...', metadata_runtime) await activities.init_opc() - logger.custom_info(f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata) + logger.custom_info( + f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', + metadata_runtime, + ) new_runtime = Runtime( telemetry=TelemetryConfig( @@ -136,7 +153,7 @@ async def main(): ) ) - logger.custom_info(f'Starting Temporal Client at {host}...', metadata) + logger.custom_info(f'Starting Temporal Client at {host}...', metadata_runtime) temporal_client = await client.Client.connect( target_host=host, @@ -144,7 +161,7 @@ async def main(): runtime=new_runtime, ) - logger.custom_info('Starting Workers...', metadata) + logger.custom_info(f'Starting Workers (runtime={runtime})...', metadata_runtime) workers = [ prepare_worker( @@ -159,6 +176,7 @@ async def main(): activities.export_data_to_postgres, ], logger=logger, + runtime=runtime, ), prepare_worker( temporal_client=temporal_client, @@ -170,6 +188,7 @@ async def main(): activities.export_data_to_postgres, ], logger=logger, + runtime=runtime, ), prepare_worker( temporal_client=temporal_client, @@ -182,6 +201,7 @@ async def main(): activities.export_data_to_postgres, ], logger=logger, + runtime=runtime, ), prepare_worker( temporal_client=temporal_client, @@ -211,6 +231,7 @@ async def main(): activities.write_pi_web_api_data, ], logger=logger, + runtime=runtime, ), ] @@ -218,7 +239,7 @@ async def main(): for w in workers: handlers.append(w.run()) - logger.custom_info('Workers started successfully', metadata) + logger.custom_info('Workers started successfully', metadata_runtime) exit_code = 0 try: diff --git a/requirements-light.txt b/requirements-light.txt index f105052..302fb79 100644 --- a/requirements-light.txt +++ b/requirements-light.txt @@ -1,12 +1,18 @@ temporalio psycopg2-binary sqlalchemy -asyncua +asyncua==1.0.6 redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.0 +sientia_do>=1.12.1 +mlflow prometheus-client botocore boto3 s3fs pyarrow -mlflow \ No newline at end of file +kaleido +hyperopt +shap +pycurl +scipy<1.14.0 +scikit-learn==1.5.2 diff --git a/requirements-local.txt b/requirements-local.txt new file mode 100644 index 0000000..73a9a47 --- /dev/null +++ b/requirements-local.txt @@ -0,0 +1,18 @@ +temporalio +psycopg2-binary +sqlalchemy +asyncua==1.0.6 +redis +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.1 +git+ssh://git@github.com/Aignosi/sientia-model-library.git@0.10.0 +prometheus-client +botocore +boto3 +s3fs +pyarrow +kaleido +hyperopt +shap +pycurl +scipy<1.14.0 +scikit-learn==1.5.2 \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 6c5c242..4ad005d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,10 +1,10 @@ temporalio psycopg2-binary sqlalchemy -asyncua +asyncua==1.0.6 redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.10.4 -git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.41.0 +sientia_do>=1.12.1 +sientia>0.40.0 prometheus-client botocore boto3 @@ -15,4 +15,4 @@ hyperopt shap pycurl scipy<1.14.0 -scikit-learn==1.5.2 \ No newline at end of file +scikit-learn==1.5.2 diff --git a/tests/laborious/utils/repository/test_model_repository.py b/tests/laborious/utils/repository/test_model_repository.py index 15f8c6b..54a75bb 100644 --- a/tests/laborious/utils/repository/test_model_repository.py +++ b/tests/laborious/utils/repository/test_model_repository.py @@ -142,6 +142,41 @@ def test_get_model_run_id_success(mlflow_repository): assert output == '1' +def test_get_model_run_id_missing_source(mlflow_repository): + mlflow_repository.client.search_registered_models.return_value = [MagicMock(name='test')] + + mlflow_repository.client.search_model_versions.return_value = [ + MagicMock(current_stage='Production', version='1', source='runs/test/0'), + MagicMock(current_stage='Production', version='2', source=None), + ] + + with pytest.raises(mlflow_lib.exceptions.MlflowException) as exc_info: + mlflow_repository.get_model_run_id('test') + + assert ( + str(exc_info.value) + == "Model 'test' version '2' in stage 'Production' has no source URI to resolve run ID." + ) + + +def test_get_model_run_id_invalid_source(mlflow_repository): + mlflow_repository.client.search_registered_models.return_value = [MagicMock(name='test')] + + mlflow_repository.client.search_model_versions.return_value = [ + MagicMock(current_stage='Production', version='1', source='runs/test/0'), + MagicMock(current_stage='Production', version='2', source='runs/test'), + ] + + with pytest.raises(mlflow_lib.exceptions.MlflowException) as exc_info: + mlflow_repository.get_model_run_id('test') + + assert ( + str(exc_info.value) + == "Model 'test' version '2' in stage 'Production' has invalid source URI " + "'runs/test' for run ID resolution." + ) + + def test_get_next_run_name(mlflow, mlflow_repository): mlflow.search_runs.return_value = [1, 2, 3] output = mlflow_repository.get_next_run_name('run') diff --git a/tests/laborious/worker/test_runtime_task_queues.py b/tests/laborious/worker/test_runtime_task_queues.py new file mode 100644 index 0000000..8b40d4a --- /dev/null +++ b/tests/laborious/worker/test_runtime_task_queues.py @@ -0,0 +1,14 @@ +from sientia_do.temporal.worker.prepare_worker import build_queue_name + +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 + + +def test_runtime_scoped_queue_names(): + runtime = 'prod-a' + assert build_queue_name(PredictionsBatch.__name__, runtime) == 'predictions_batch-prod-a-queue' + assert build_queue_name(MinimalRetrain.__name__, runtime) == 'minimal_retrain-prod-a-queue' + assert build_queue_name(Drift.__name__, runtime) == 'drift-prod-a-queue' + assert build_queue_name(SimpleMetrics.__name__, runtime) == 'simple_metrics-prod-a-queue' diff --git a/values.yaml b/values.yaml index b07d60b..25d3418 100644 --- a/values.yaml +++ b/values.yaml @@ -209,6 +209,9 @@ env: value: "temporal-frontend.temporal.svc.cluster.local:7233" - name: TEMPORAL_NAMESPACE value: "laborious" + # Suffix for all Temporal task queues: {workflow}-{RUNTIME}-queue + - name: RUNTIME + value: "legacy" - name: MONGODB_USERNAME value: "root"