diff --git a/CLAUDE.md b/CLAUDE.md index 3ced319..88ef953 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -45,7 +45,7 @@ Coverage gate in CI is 80% minimum; this repo targets 100% in practice — don't ## Architecture ### Temporal layering -- **Worker** (`model_manager/worker/worker.py`): entrypoint. Builds config from env, connects Postgres/MLFlow/MinIO/Plugin Store, installs the `RUNTIME` model runtime via `PluginStore.install_runtime`, starts two Temporal workers (one per task queue) plus Prometheus metrics server and the cleanup schedule. All heavy imports are wrapped in `workflow.unsafe.imports_passed_through()` — required because Temporal workflow sandboxing forbids non-deterministic imports at module scope outside that block. +- **Worker** (`model_manager/worker/worker.py`): entrypoint. Builds config from env, connects Postgres/MLFlow/MinIO/Plugin Store, installs the `RUNTIME` model runtime via `PluginStore.install_runtime` (boot fails if that changes the version of an already-imported package, so the `requirements.txt` pins must match the factory runtimes; `app_up` goes to 1 only after this check), starts two Temporal workers (one per task queue) plus Prometheus metrics server and the cleanup schedule. All heavy imports are wrapped in `workflow.unsafe.imports_passed_through()` — required because Temporal workflow sandboxing forbids non-deterministic imports at module scope outside that block. - **Workflows** (`model_manager/workflows/`): pure orchestration, no I/O beyond `execute_activity_method` calls. - `train_model.py` (`TrainModel`, workflow name `train_model`): validate `experiment_run_id` → `load_model_metadata` → `validate_train_params` → update DB status `ORCHESTRATOR_WAITING_PROC` → `train_model` activity → update DB status `TRAINING_SUCCESS`/`TRAINING_ERROR` → `cleanup_resources` (always runs in `finally` if a `run_dir` exists, even on failure). Uses distinct `RetryPolicy`s per operation class: `network_retry_policy` (transient I/O), `no_retry_policy` (training/validation — data errors are permanent, and retrying would duplicate MLflow writes), `database_retry_policy` (Postgres). Timeouts are env-driven (`TIMEOUT_*`), extend `TIMEOUT_TRAIN_MODEL` instead of adding retries for slow training. - `cleanup_files.py` (`CleanupFiles`): scheduled (`cleanup_schedule.py`, default daily at midnight UTC via `CLEANUP_CRON`) deletion of stale local temp dirs matching `name_YYYYMMDD_HHMMSS_microseconds`, older than `CLEANUP_RETENTION_HOURS`. diff --git a/README.md b/README.md index 4483917..e53e7f0 100644 --- a/README.md +++ b/README.md @@ -149,6 +149,7 @@ The Model Manager system uses a Temporal-based workflow architecture with clear - Prometheus metrics server initialization - Notification handler setup and configuration - **Runtime Installation**: Automatically installs the required model runtime from the Plugin Store + - **Runtime version guard**: boot fails if the install changes the version of a package the worker already imported, so the `requirements.txt` pins must match `model-plugin-factory` `runtime/basic.yaml` and `runtime/xgboost.yaml`; `app_up` goes to 1 only after this check - **Key Features**: - Automatic scaling with `PollerBehaviorAutoscaling` - Health check endpoints for Kubernetes liveness/readiness probes @@ -1249,6 +1250,10 @@ sientia-dataops-model-manager/ - Check training parameter validation errors - Verify input data format and required fields +5. **Worker exits at boot with `Runtime ... changed packages already imported by the worker`** + - A runtime pin in `model-plugin-factory` `runtime/*.yaml` no longer matches `requirements.txt`; the message lists each package as `old -> new` + - Set the runtime version in `requirements.txt` and `requirements-local.txt` here and in `sientia-dataops-laborious`, then redeploy + ### Debug Mode Enable debug logging by setting the log level: diff --git a/model_manager/worker/worker.py b/model_manager/worker/worker.py index 70bdb32..8e94429 100644 --- a/model_manager/worker/worker.py +++ b/model_manager/worker/worker.py @@ -79,7 +79,8 @@ def _get_runtime(runtime: str | None) -> str: def _loaded_distribution_versions() -> dict[str, str]: """Version on disk of every installed distribution whose top-level module is imported.""" - loaded = {name.partition('.')[0] for name in sys.modules} + # list() copies the keys: another thread importing while we iterate would raise RuntimeError. + loaded = {name.partition('.')[0] for name in list(sys.modules)} return { dist: importlib.metadata.version(dist) for module, dists in importlib.metadata.packages_distributions().items() @@ -114,7 +115,8 @@ async def main(): 1. Sets up logging and metadata 2. Starts Prometheus metrics server 3. Initializes notification handler, then installs the runtime; exits if the install - changed the version of a package the worker already imported + changed the version of a package the worker already imported, and only then sets + ``app_up`` (what the probes check) to 1 4. Creates and configures activities 5. Starts Temporal client and workers 6. Manages worker lifecycle and graceful shutdown @@ -188,6 +190,10 @@ async def main(): metrics.APP_UP.labels(pod_id=POD_ID).set(0) sys.exit(1) + # The probes only check app_up, so it goes to 1 after the guard: a pod that would fail it + # never turns Ready, and a rollout keeps the old pod. + metrics.APP_UP.labels(pod_id=POD_ID).set(1) + activities = Activities( postgres_config=build_postgres_config(), mlflow_config=build_mlflow_config(), @@ -274,8 +280,8 @@ def start_prometheus_server(logger: SientiaLogger, metadata: dict[str, str | Non """ 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. + This function initializes the Prometheus HTTP server on the configured port. + ``main`` sets ``app_up`` to 1 later, once the runtime install passed the version guard. The server exposes metrics that can be scraped by Prometheus for monitoring the health and performance of the Model Manager worker. @@ -291,7 +297,6 @@ def start_prometheus_server(logger: SientiaLogger, metadata: dict[str, str | Non port = int(os.getenv('HTTP_METRICS_PORT', 9090)) start_http_server(port) logger.custom_info(f'Prometheus server initialized on port {port}.', metadata) - metrics.APP_UP.labels(pod_id=POD_ID).set(1) # Mark app as UP except Exception as e: # noqa: BLE001 logger.custom_critical(f'Failed to start Prometheus server: {e}', metadata) os._exit(1) diff --git a/requirements-local.txt b/requirements-local.txt index d517cec..4e3af07 100644 --- a/requirements-local.txt +++ b/requirements-local.txt @@ -8,4 +8,15 @@ git+ssh://git@github.com/Aignosi/sientia-model-library.git@0.14.0 prometheus-client==0.23.1 beautifulsoup4==4.12.3 evidently==0.6.7 -jsonschema==4.26.0 \ No newline at end of file +jsonschema==4.26.0 +# Same versions as the plugin runtimes (model-plugin-factory runtime/basic.yaml and xgboost.yaml). +# The worker imports these before install_runtime; if the runtime installed a different version, +# the old code would stay in memory, so the worker refuses to boot (SIENTIAPDE-2214). +# Changing a runtime pin means changing it here and in requirements.txt too. +mlflow==3.8.1 +pathspec==1.0.4 +pandas==2.3.3 +cloudpickle==3.1.2 +scikit-learn==1.8.0 +joblib==1.5.3 +rich==14.3.2 diff --git a/requirements.txt b/requirements.txt index 9ae8fa6..ace4c48 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,7 +4,7 @@ sqlalchemy==2.0.50 boto3==1.42.70 botocore==1.42.70 sientia_do>=1.12.2 -sientia_model>=0.8.1 +sientia_model>=0.14.0 prometheus-client==0.23.1 beautifulsoup4==4.12.3 evidently==0.6.7 @@ -12,7 +12,7 @@ jsonschema==4.26.0 # Same versions as the plugin runtimes (model-plugin-factory runtime/basic.yaml and xgboost.yaml). # The worker imports these before install_runtime; if the runtime installed a different version, # the old code would stay in memory, so the worker refuses to boot (SIENTIAPDE-2214). -# Changing a runtime pin means changing it here too. +# Changing a runtime pin means changing it here and in requirements-local.txt too. mlflow==3.8.1 pathspec==1.0.4 pandas==2.3.3 diff --git a/tests/worker/test_worker.py b/tests/worker/test_worker.py index 9fb43ff..07fd18c 100644 --- a/tests/worker/test_worker.py +++ b/tests/worker/test_worker.py @@ -132,9 +132,8 @@ def test_start_prometheus_server_success( # Verify HTTP server started mock_start_http_server.assert_called_once_with(9090) - # Verify APP_UP metric was set to 1 - mock_metrics.APP_UP.labels.assert_called_once_with(pod_id='test-pod-123') - mock_app_up.set.assert_called_once_with(1) + # app_up goes to 1 in main(), after the runtime version guard. + mock_app_up.set.assert_not_called() mock_logger.custom_info.assert_called_once() @@ -284,7 +283,8 @@ async def test_main_successful_startup( # Verify cleanup was performed mock_notification_handler.shutdown.assert_called_once() mock_activities.shutdown.assert_called_once() - mock_app_up.set.assert_called_with(0) + # Ready after the guard, down again on shutdown. + assert [c.args[0] for c in mock_app_up.set.call_args_list] == [1, 0] @pytest.mark.asyncio @@ -941,22 +941,30 @@ async def test_main_runtime_changed_loaded_package_exits(mock_env_vars, mock_log plugin_store = Mock() plugin_store.install_runtime = AsyncMock(return_value=None) + before = {'scikit-learn': '1.9.1'} + + def snapshot(): + # Taken after install_runtime, it would already see 1.8.0 and the guard would pass. + assert plugin_store.install_runtime.await_count == 0 + return before + mongo = {'connection_string': 'mongodb://test', 'database_name': 'test_db', 'uri': 'x'} with ( patch.object(worker_module, 'RUNTIME', 'basic'), patch.object(worker_module, 'ensure_runtime_directories'), - patch.object(worker_module, 'metrics'), + patch.object(worker_module, 'metrics') as mock_metrics, patch.object(worker_module, 'start_prometheus_server'), patch.object(worker_module, 'get_logger', return_value=mock_logger), patch.object(worker_module, 'build_mongodb_config', return_value=mongo), patch.object(worker_module, 'NotificationHandler'), patch.object(worker_module, 'build_plugin_store_config', return_value=MagicMock()), patch.object(worker_module, 'PluginStore', return_value=plugin_store), + patch.object(worker_module, '_loaded_distribution_versions', side_effect=snapshot), patch.object( worker_module, '_runtime_changed_loaded_packages', return_value={'scikit-learn': '1.9.1 -> 1.8.0'}, - ), + ) as mock_changed, patch.object(worker_module, 'Activities') as mock_activities_class, patch.object(worker_module.sys, 'exit', side_effect=SystemExit(1)) as mock_exit, ): @@ -965,5 +973,35 @@ async def test_main_runtime_changed_loaded_package_exits(mock_env_vars, mock_log mock_exit.assert_called_once_with(1) mock_activities_class.assert_not_called() + plugin_store.install_runtime.assert_awaited_once() + mock_changed.assert_called_once_with(before) + mock_metrics.APP_UP.labels.return_value.set.assert_called_once_with(0) message = mock_logger.custom_critical.call_args.args[0] assert "'scikit-learn': '1.9.1 -> 1.8.0'" in message + + +def test_guard_detects_version_change_on_disk(tmp_path, monkeypatch): + """The real importlib.metadata path: a package swapped on disk after import is reported.""" + import types + + from model_manager.worker.worker import ( + _loaded_distribution_versions, + _runtime_changed_loaded_packages, + ) + + info = tmp_path / 'fakepkg_2214-1.0.dist-info' + info.mkdir() + (info / 'METADATA').write_text('Metadata-Version: 2.1\nName: fakepkg-2214\nVersion: 1.0\n') + (info / 'top_level.txt').write_text('fakepkg_2214\n') + monkeypatch.syspath_prepend(str(tmp_path)) + # The guard only reads sys.modules keys; setitem removes the entry again on teardown. + monkeypatch.setitem(sys.modules, 'fakepkg_2214', types.ModuleType('fakepkg_2214')) + + before = _loaded_distribution_versions() + assert before['fakepkg-2214'] == '1.0' + + # What pip does on an upgrade: the old dist-info goes away and a new one takes its place. + new_info = info.rename(tmp_path / 'fakepkg_2214-2.0.dist-info') + (new_info / 'METADATA').write_text('Metadata-Version: 2.1\nName: fakepkg-2214\nVersion: 2.0\n') + + assert _runtime_changed_loaded_packages(before) == {'fakepkg-2214': '1.0 -> 2.0'}