Merge pull request #37 from Aignosi/feature/SIENTIAPDE-2226
Enhance Quality Gate Workflow for Template Management and Runtime Checks
This commit is contained in:
GitHub
1
.github/workflows/quality-gate.yml
vendored
1
.github/workflows/quality-gate.yml
vendored
@@ -26,4 +26,5 @@ jobs:
|
||||
with:
|
||||
project_name: 'model_manager'
|
||||
repositories: 'sientia-dataops-library,sientia-model-library'
|
||||
runtime_check: true
|
||||
secrets: inherit
|
||||
@@ -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` (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.
|
||||
- **Worker** (`model_manager/worker/worker.py`): entrypoint. Builds config from env, connects Postgres/MLFlow/MinIO/Plugin Store, does **not** install the model runtime (the `sientia-module` entrypoint merges `requirements.txt` with the Plugin Store runtime and installs both before boot, SIENTIAPDE-2223; leave runtime libs unpinned in `requirements.txt` so the runtime version wins, and the quality gate's `runtime_check` fails a PR that conflicts with any warehouse runtime), sets `app_up` to 1 only after initialization, 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`.
|
||||
|
||||
24
README.md
24
README.md
@@ -90,7 +90,7 @@ An enterprise-grade ML model training orchestration platform built on Temporal.
|
||||
- **Scheduled Maintenance**: Automated lifecycle jobs for filesystem hygiene and stale file cleanup
|
||||
|
||||
### Advanced Capabilities
|
||||
- **Dynamic Runtime Provisioning**: Automated installation of required model runtimes from the Plugin Store
|
||||
- **Dynamic Runtime Provisioning**: The `sientia-module` entrypoint merges `requirements.txt` with the Plugin Store runtime and installs both before the worker starts
|
||||
- **Granular Retry Policies**: Tailored strategies for network, MLFlow, database, and filesystem operations
|
||||
- **Scalable Infrastructure**: Kubernetes-ready design with support for horizontal scaling and poller autoscaling
|
||||
- **Secure Configuration**: Environment-driven connection management with fallback to sensible defaults
|
||||
@@ -148,8 +148,7 @@ The Model Manager system uses a Temporal-based workflow architecture with clear
|
||||
- Task queue configuration and load balancing
|
||||
- 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
|
||||
- **Runtime libraries**: not installed by the worker. The `sientia-module` entrypoint merges `requirements.txt` with the Plugin Store runtime (`PLUGIN_RUNTIME`), writes the audited `requirements.resolved.txt` and installs it before Python starts; on a conflict the pod does not start (SIENTIAPDE-2223). `app_up` goes to 1 only after initialization
|
||||
- **Key Features**:
|
||||
- Automatic scaling with `PollerBehaviorAutoscaling`
|
||||
- Health check endpoints for Kubernetes liveness/readiness probes
|
||||
@@ -578,6 +577,18 @@ kubectl exec -n temporal <temporal-admin-tools-pod-name> -- \
|
||||
pip install -r requirements-dev.txt
|
||||
```
|
||||
|
||||
4. **Install the runtime libraries**
|
||||
|
||||
In the cluster, the `sientia-module` entrypoint installs the runtime libraries before the
|
||||
worker starts; the worker itself installs nothing. Locally, install them by hand from
|
||||
`runtime/<RUNTIME>.yaml` of the Plugin Store (`Aignosi/model-plugin-warehouse`), skipping
|
||||
`sientia_do`/`sientia_model`, which the requirements file already brings. For `basic`:
|
||||
```bash
|
||||
pip install mlflow==3.8.1 pathspec==1.0.4 pandas==2.3.3 cloudpickle==3.1.2 scikit-learn==1.8.0
|
||||
```
|
||||
Copy the versions from the YAML, not from this example: they change with the runtime. With
|
||||
`RUNTIME=single` there is nothing to install.
|
||||
|
||||
4. **Create environment configuration file**
|
||||
```bash
|
||||
cp .env.example .env
|
||||
@@ -812,6 +823,7 @@ Runs automatically on each Pull Request to `main`:
|
||||
- ✅ Tests must pass with minimum 80% coverage
|
||||
- ✅ SonarQube analysis for code quality metrics
|
||||
- ✅ Automatic version calculation based on branch pattern
|
||||
- ✅ Runtime requirements check (`runtime_check: true`): `requirements.txt` is merged with every `runtime/*.yaml` of `model-plugin-warehouse@main`, using the `sientia-module` entrypoint's own merge script; a conflict fails the gate, "runtime prevaleceu" is a warning. The warehouse changes on its own, so a push with no code change can turn red
|
||||
|
||||
#### Deploy (`.github/workflows/deploy.yml`)
|
||||
Triggered manually via `workflow_dispatch`, with a required `chart_ref` input (no
|
||||
@@ -1250,9 +1262,9 @@ 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
|
||||
5. **Pod does not start, with `# CONFLICTS` in the log**
|
||||
- The `sientia-module` entrypoint found a library whose version in `requirements.txt` conflicts with the Plugin Store runtime; the log lists each library with both specs
|
||||
- Leave libraries the runtime already defines without a version in `requirements.txt` (the runtime's version wins), or align the version, then redeploy. The quality gate catches this on the PR against the warehouse `main`
|
||||
|
||||
### Debug Mode
|
||||
|
||||
|
||||
@@ -64,3 +64,15 @@ These scenarios test the business rule validations inside `validate_train_params
|
||||
| `test_scenario_3_1_1_cleanup_with_no_temp_dirs` | Temp directory is empty. Activity completes without error. |
|
||||
| `test_scenario_3_1_2_cleanup_removes_old_temp_dirs` | Two stale directories matching `name_YYYYMMDD_HHMMSS_microseconds` are removed when older than retention. |
|
||||
| `test_scenario_3_1_3_cleanup_nonexistent_temp_path` | Target path does not exist. Handled gracefully without error. |
|
||||
|
||||
## 4. PluginStore Wiring (`test_plugin_store_wiring.py`)
|
||||
|
||||
`Activities` receives `plugin_store_config` and builds the `PluginStore` itself. These scenarios run `01-linear-regression-basic.json` on a worker built from a per-test config (`activities_factory`) against the Gitea `model-store` repo.
|
||||
|
||||
| Test Function | Config Change | Expected Status | Description |
|
||||
|---|---|---|---|
|
||||
| `test_scenario_4_1_1_plugin_store_from_env_config` | `STORE_*` env vars → `build_plugin_store_config()` | `TRAINING_SUCCESS` | Same path as the worker. With `STORE_BRANCH` unset the store resolves the branch to `main`. |
|
||||
| `test_scenario_4_1_2_explicit_branch` | `branch = 'main'` | `TRAINING_SUCCESS` | Explicit branch skips resolution and is used for every read. |
|
||||
| `test_scenario_4_1_3_explicit_branch_without_model` | `branch = 'e2e-no-linear-regression'` | `ORCHESTRATOR_VALIDATION_ERROR` | Branch forked from `main` without `models/linear_regression/index.yaml`; error mentions `linear_regression/index.yaml`. |
|
||||
| `test_scenario_4_1_4_nonexistent_repository` | `repo = 'no-such-store'` | `ORCHESTRATOR_VALIDATION_ERROR` | `Activities` still builds (branch falls back to `main`); fails when loading model metadata. |
|
||||
| `test_scenario_4_1_5_wrong_credentials` | `password = 'wrong-password'` | `ORCHESTRATOR_VALIDATION_ERROR` | Gitea rejects the read; error mentions `401`. |
|
||||
|
||||
@@ -32,7 +32,6 @@ from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig
|
||||
|
||||
with workflow.unsafe.imports_passed_through():
|
||||
import asyncio
|
||||
import importlib.metadata
|
||||
import os
|
||||
import sys
|
||||
|
||||
@@ -77,36 +76,6 @@ 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.
|
||||
@@ -114,9 +83,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, 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
|
||||
3. Initializes notification handler and Plugin Store client, then sets ``app_up``
|
||||
(what the probes check) to 1. The runtime libraries are already installed: the
|
||||
sientia-module entrypoint merges them into the resolved requirements before boot
|
||||
4. Creates and configures activities
|
||||
5. Starts Temporal client and workers
|
||||
6. Manages worker lifecycle and graceful shutdown
|
||||
@@ -158,8 +127,6 @@ async def main():
|
||||
|
||||
metrics_controller = MetricsController(logger=logger)
|
||||
|
||||
logger.custom_info(f'Installing runtime {runtime}', metadata)
|
||||
|
||||
plugin_store_parameters = build_plugin_store_config()
|
||||
plugin_store = PluginStore(
|
||||
base_url=plugin_store_parameters['base_url'],
|
||||
@@ -177,21 +144,7 @@ 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.
|
||||
# The probes only check app_up, so it goes to 1 only after initialization.
|
||||
metrics.APP_UP.labels(pod_id=POD_ID).set(1)
|
||||
|
||||
activities = Activities(
|
||||
@@ -281,7 +234,7 @@ 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.
|
||||
``main`` sets ``app_up`` to 1 later, once the runtime install passed the version guard.
|
||||
``main`` sets ``app_up`` to 1 later, once initialization is done.
|
||||
|
||||
The server exposes metrics that can be scraped by Prometheus for monitoring
|
||||
the health and performance of the Model Manager worker.
|
||||
|
||||
@@ -131,7 +131,9 @@ class TrainModel:
|
||||
training_succeeded = True
|
||||
finally:
|
||||
try:
|
||||
if train_result is not None:
|
||||
# train_result is only None when training raised, so the false branch
|
||||
# always leaves through the re-raise and never reaches the return below.
|
||||
if train_result is not None: # pragma: no branch
|
||||
await self._cleanup_resources(
|
||||
run_dir=train_result.get('run_dir'),
|
||||
metadata=metadata,
|
||||
|
||||
@@ -1,22 +1,19 @@
|
||||
temporalio==1.23.0
|
||||
psycopg2-binary==2.9.11
|
||||
sqlalchemy==2.0.49
|
||||
boto3==1.42.70
|
||||
botocore==1.42.70
|
||||
temporalio
|
||||
psycopg2-binary
|
||||
# SQLAlchemy 2.1 makes psycopg (v3) the default driver for postgresql:// URLs; sientia_do 1.12.2
|
||||
# builds that URL and ships only psycopg2-binary, so 2.1 fails with "No module named psycopg".
|
||||
sqlalchemy<2.1
|
||||
boto3
|
||||
botocore
|
||||
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2
|
||||
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
|
||||
# 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
|
||||
prometheus-client
|
||||
beautifulsoup4
|
||||
# reports.py uses the Report/metric_preset API that 0.7 moved to evidently.legacy.
|
||||
evidently<0.7
|
||||
jsonschema
|
||||
# Imported directly by the worker, without a version: the sientia-module entrypoint merges this
|
||||
# file with the Plugin Store runtime, and the runtime's version wins (SIENTIAPDE-2223).
|
||||
mlflow
|
||||
pandas
|
||||
scikit-learn
|
||||
|
||||
@@ -1,22 +1,19 @@
|
||||
temporalio==1.23.0
|
||||
psycopg2-binary==2.9.11
|
||||
sqlalchemy==2.0.50
|
||||
boto3==1.42.70
|
||||
botocore==1.42.70
|
||||
temporalio
|
||||
psycopg2-binary
|
||||
# SQLAlchemy 2.1 makes psycopg (v3) the default driver for postgresql:// URLs; sientia_do 1.12.2
|
||||
# builds that URL and ships only psycopg2-binary, so 2.1 fails with "No module named psycopg".
|
||||
sqlalchemy<2.1
|
||||
boto3
|
||||
botocore
|
||||
sientia_do>=1.12.2
|
||||
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
|
||||
prometheus-client
|
||||
beautifulsoup4
|
||||
# reports.py uses the Report/metric_preset API that 0.7 moved to evidently.legacy.
|
||||
evidently<0.7
|
||||
jsonschema
|
||||
# Imported directly by the worker, without a version: the sientia-module entrypoint merges this
|
||||
# file with the Plugin Store runtime, and the runtime's version wins (SIENTIAPDE-2223).
|
||||
mlflow
|
||||
pandas
|
||||
scikit-learn
|
||||
|
||||
@@ -92,6 +92,53 @@ def test_ensure_date_column_parsed_invalid_raises():
|
||||
dmr._ensure_date_column_parsed(df, p)
|
||||
|
||||
|
||||
def test_ensure_date_column_parsed_unmapped_format_raises():
|
||||
df = pd.DataFrame({'ts': ['2024-01-01 10:00:00']})
|
||||
p = _params(date_column='ts')
|
||||
# Bypasses validate_business_rules, which would reject it earlier
|
||||
p.date_format = 'dd/MM/yyyy HH:mm'
|
||||
with pytest.raises(ValueError, match='is not mapped to a strftime pattern'):
|
||||
dmr._ensure_date_column_parsed(df, p)
|
||||
|
||||
|
||||
def test_drop_rows_with_missing_timestamp_without_date_column_returns_input():
|
||||
repo = dmr.DataManagerRepository(MagicMock())
|
||||
df = pd.DataFrame({'v1': [1.0, 2.0]})
|
||||
assert repo._drop_rows_with_missing_timestamp(df, _params(), {}) is df
|
||||
|
||||
|
||||
def test_drop_rows_with_missing_timestamp_non_object_dtype_drops_nat():
|
||||
repo = dmr.DataManagerRepository(MagicMock())
|
||||
df = pd.DataFrame(
|
||||
{'timestamp': pd.to_datetime(['2024-01-01', None, '2024-01-03']), 'v1': [1, 2, 3]}
|
||||
)
|
||||
out = repo._drop_rows_with_missing_timestamp(df, _params(), {})
|
||||
assert list(out['v1']) == [1, 3]
|
||||
|
||||
|
||||
def test_coerce_non_timestamp_columns_to_numeric_skips_date_column():
|
||||
repo = dmr.DataManagerRepository(MagicMock())
|
||||
df = pd.DataFrame({'timestamp': ['2024-01-01 10:00:00'], 'v1': ['1.5']})
|
||||
out = repo._coerce_non_timestamp_columns_to_numeric(df, _params(), {})
|
||||
assert out['timestamp'].iloc[0] == '2024-01-01 10:00:00'
|
||||
assert float(out['v1'].iloc[0]) == 1.5
|
||||
|
||||
|
||||
def test_set_timezone_on_index_converts_aware_index_to_utc():
|
||||
repo = dmr.DataManagerRepository(MagicMock())
|
||||
idx = pd.DatetimeIndex(['2024-01-01 00:00'], tz='America/Sao_Paulo')
|
||||
out = repo._set_timezone_on_index(pd.DataFrame({'v1': [1]}, index=idx))
|
||||
assert str(out.index.tz) == 'UTC'
|
||||
assert out.index[0] == pd.Timestamp('2024-01-01 03:00', tz='UTC')
|
||||
|
||||
|
||||
def test_set_timezone_on_index_non_datetime_index_raises():
|
||||
repo = dmr.DataManagerRepository(MagicMock())
|
||||
df = pd.DataFrame({'v1': [1]})
|
||||
with pytest.raises(ValueError, match='Index is not a DatetimeIndex'):
|
||||
repo._set_timezone_on_index(df)
|
||||
|
||||
|
||||
def test_prepare_training_data_csv_load_failure():
|
||||
repo = dmr.DataManagerRepository(MagicMock())
|
||||
p = _params()
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
from unittest.mock import AsyncMock, MagicMock, Mock, patch
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -132,7 +132,7 @@ def test_start_prometheus_server_success(
|
||||
# Verify HTTP server started
|
||||
mock_start_http_server.assert_called_once_with(9090)
|
||||
|
||||
# app_up goes to 1 in main(), after the runtime version guard.
|
||||
# app_up goes to 1 in main(), after initialization.
|
||||
mock_app_up.set.assert_not_called()
|
||||
mock_logger.custom_info.assert_called_once()
|
||||
|
||||
@@ -232,9 +232,6 @@ async def test_main_successful_startup(
|
||||
mock_activities_class.return_value = mock_activities
|
||||
|
||||
mock_plugin_store_instance = AsyncMock()
|
||||
mock_plugin_store_instance.install_runtime = AsyncMock(
|
||||
return_value={'runtime': 'model-manager-worker', 'installed': []},
|
||||
)
|
||||
mock_plugin_store_class.return_value = mock_plugin_store_instance
|
||||
mock_build_plugin_store_config.return_value = {
|
||||
'base_url': 'http://sientia-plugin-store.svc.cluster.local',
|
||||
@@ -283,8 +280,10 @@ async def test_main_successful_startup(
|
||||
# Verify cleanup was performed
|
||||
mock_notification_handler.shutdown.assert_called_once()
|
||||
mock_activities.shutdown.assert_called_once()
|
||||
# Ready after the guard, down again on shutdown.
|
||||
# Ready after initialization, down again on shutdown.
|
||||
assert [c.args[0] for c in mock_app_up.set.call_args_list] == [1, 0]
|
||||
# The runtime libraries come from the sientia-module entrypoint, not from the worker.
|
||||
mock_plugin_store_instance.install_runtime.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -360,9 +359,6 @@ async def test_main_handles_exception(
|
||||
mock_metrics.APP_UP.labels.return_value = mock_app_up
|
||||
|
||||
mock_plugin_store_instance = AsyncMock()
|
||||
mock_plugin_store_instance.install_runtime = AsyncMock(
|
||||
return_value={'runtime': 'model-manager-worker', 'installed': []},
|
||||
)
|
||||
mock_plugin_store_class.return_value = mock_plugin_store_instance
|
||||
mock_build_plugin_store_config.return_value = {
|
||||
'base_url': 'http://sientia-plugin-store.svc.cluster.local',
|
||||
@@ -459,9 +455,6 @@ async def test_main_temporal_client_configuration(
|
||||
mock_build_minio.return_value = {}
|
||||
|
||||
mock_plugin_store_instance = AsyncMock()
|
||||
mock_plugin_store_instance.install_runtime = AsyncMock(
|
||||
return_value={'runtime': 'model-manager-worker', 'installed': []},
|
||||
)
|
||||
mock_plugin_store_class.return_value = mock_plugin_store_instance
|
||||
mock_build_plugin_store_config.return_value = {
|
||||
'base_url': 'http://sientia-plugin-store.svc.cluster.local',
|
||||
@@ -594,9 +587,6 @@ async def test_main_worker_configuration(
|
||||
mock_metrics.APP_UP.labels.return_value = mock_app_up
|
||||
|
||||
mock_plugin_store_instance = AsyncMock()
|
||||
mock_plugin_store_instance.install_runtime = AsyncMock(
|
||||
return_value={'runtime': 'model-manager-worker', 'installed': []},
|
||||
)
|
||||
mock_plugin_store_class.return_value = mock_plugin_store_instance
|
||||
mock_build_plugin_store_config.return_value = {
|
||||
'base_url': 'http://sientia-plugin-store.svc.cluster.local',
|
||||
@@ -716,9 +706,6 @@ async def test_main_schedule_creation_failure_does_not_stop_worker(
|
||||
mock_metrics.APP_UP.labels.return_value = mock_app_up
|
||||
|
||||
mock_plugin_store_instance = AsyncMock()
|
||||
mock_plugin_store_instance.install_runtime = AsyncMock(
|
||||
return_value={'runtime': 'model-manager-worker', 'installed': []},
|
||||
)
|
||||
mock_plugin_store_class.return_value = mock_plugin_store_instance
|
||||
mock_build_plugin_store_config.return_value = {
|
||||
'base_url': 'http://sientia-plugin-store.svc.cluster.local',
|
||||
@@ -816,9 +803,6 @@ async def test_main_missing_runtime_uses_single_fallback(
|
||||
mock_prepare_worker.return_value = mock_worker_instance
|
||||
|
||||
mock_plugin_store_instance = AsyncMock()
|
||||
mock_plugin_store_instance.install_runtime = AsyncMock(
|
||||
return_value={'runtime': 'single', 'installed': []},
|
||||
)
|
||||
mock_plugin_store_class.return_value = mock_plugin_store_instance
|
||||
mock_build_plugin_store_config.return_value = {
|
||||
'base_url': 'http://sientia-plugin-store.svc.cluster.local',
|
||||
@@ -903,105 +887,3 @@ 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'}
|
||||
|
||||
Reference in New Issue
Block a user