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 df2ff5e..631e29d 100644 --- a/model_manager/worker/worker.py +++ b/model_manager/worker/worker.py @@ -32,6 +32,7 @@ from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig with workflow.unsafe.imports_passed_through(): import asyncio + import importlib.metadata import os import sys @@ -76,6 +77,36 @@ def _get_runtime(runtime: str | None) -> str: return normalized_runtime or 'single' +def _loaded_distribution_versions() -> dict[str, str]: + """Version on disk of every installed distribution whose top-level module is imported.""" + # Iterate a copy: another thread importing while we iterate would raise RuntimeError. + loaded = {name.partition('.')[0] for name in sys.modules.copy()} + return { + dist: importlib.metadata.version(dist) + for module, dists in importlib.metadata.packages_distributions().items() + if module in loaded + for dist in dists + } + + +def _runtime_changed_loaded_packages(before: dict[str, str]) -> dict[str, str]: + """ + Packages that were imported before ``install_runtime`` and whose version changed on disk. + + The process keeps the old code in memory, so a later lazy import mixes two versions + (SIENTIAPDE-2214: sklearn 1.9.1 in memory, 1.8.0 on disk, ``_cyutility`` import error). + """ + changed = {} + for dist, old in before.items(): + try: + new = importlib.metadata.version(dist) + except importlib.metadata.PackageNotFoundError: + new = 'removed' + if new != old: + changed[dist] = f'{old} -> {new}' + return changed + + async def main(): """ Main entry point for the Model Manager worker application. @@ -83,7 +114,9 @@ async def main(): 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 + 3. Initializes notification handler, then installs the runtime; exits if the install + 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 @@ -144,8 +177,23 @@ async def main(): metrics_controller=metrics_controller, ) + loaded_before_install = _loaded_distribution_versions() await plugin_store.install_runtime(runtime_name=runtime) + changed = _runtime_changed_loaded_packages(loaded_before_install) + if changed: + logger.custom_critical( + f'Runtime {runtime} changed packages already imported by the worker: {changed}. ' + 'Pin them in requirements.txt to the runtime versions.', + metadata, + ) + 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(), @@ -232,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. @@ -249,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/pyproject.toml b/pyproject.toml index b3c62e1..8409ec4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "model-manager" -version = "2.0.3" +version = "2.0.4" description = "Sientia DataOps Model Manager - ML Model Orchestration System" readme = "README.md" requires-python = ">=3.11" 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 46894ca..ace4c48 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,8 +4,19 @@ 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 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-local.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/tests/worker/test_worker.py b/tests/worker/test_worker.py index b127853..07fd18c 100644 --- a/tests/worker/test_worker.py +++ b/tests/worker/test_worker.py @@ -3,7 +3,7 @@ import asyncio import os import sys -from unittest.mock import AsyncMock, Mock, patch +from unittest.mock import AsyncMock, MagicMock, Mock, patch import pytest @@ -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 @@ -848,12 +848,10 @@ async def test_main_missing_runtime_uses_single_fallback( def test_main_entrypoint(mock_asyncio_run): """Test the __main__ entrypoint.""" # Import and execute the main block - with patch.object(sys, 'argv', ['worker.py']): - import model_manager.worker.worker as worker_module - - # Simulate running the module - worker_module.main = AsyncMock() + import model_manager.worker.worker as worker_module + # patch.object restores main afterwards; assigning it leaked the mock into later tests. + with patch.object(sys, 'argv', ['worker.py']), patch.object(worker_module, 'main', AsyncMock()): # This would normally be called by asyncio.run(main()) # We just verify the pattern is correct assert callable(worker_module.main) @@ -905,3 +903,105 @@ def test_start_prometheus_server_prints_failure( # Agora o erro é logado via logger crítico mock_logger.custom_critical.assert_called_once() mock_exit.assert_called_once_with(1) + + +def test_loaded_distribution_versions_reads_imported_packages(): + """An imported module maps to its distribution's installed version.""" + import importlib.metadata + + from model_manager.worker.worker import _loaded_distribution_versions + + assert _loaded_distribution_versions()['pytest'] == importlib.metadata.version('pytest') + + +def test_runtime_changed_loaded_packages_reports_only_changes(): + """Only distributions whose version on disk differs after the install are reported.""" + import importlib.metadata + + from model_manager.worker.worker import _runtime_changed_loaded_packages + + on_disk = {'scikit-learn': '1.8.0', 'numpy': '1.26.4'} + + def fake_version(dist): + if dist not in on_disk: + raise importlib.metadata.PackageNotFoundError(dist) + return on_disk[dist] + + before = {'scikit-learn': '1.9.1', 'numpy': '1.26.4', 'gone': '1.0'} + with patch('model_manager.worker.worker.importlib.metadata.version', side_effect=fake_version): + changed = _runtime_changed_loaded_packages(before) + + assert changed == {'scikit-learn': '1.9.1 -> 1.8.0', 'gone': '1.0 -> removed'} + + +@pytest.mark.asyncio +async def test_main_runtime_changed_loaded_package_exits(mock_env_vars, mock_logger): + """Boot stops before Activities when install_runtime changed an imported package.""" + import model_manager.worker.worker as worker_module + + 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') 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, + ): + with pytest.raises(SystemExit): + await worker_module.main() + + 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'}