docs(worker): document SIENTIA_RUNTIME_READY, log the skipped install, test install failure (SIENTIAPDE-2214)
From the code-review-python pass on #34: - README, CLAUDE.md and the module docstring: the worker re-executes itself after install_runtime, and SIENTIA_RUNTIME_READY is internal; setting it in .env/the chart would skip the install silently. - "Installing runtime" is now logged only on the pass that installs; the re-executed pass logs that the install was skipped. - The ponytail comment notes that the install success metrics are lost to the exec. - New test: a failed install propagates without setting the flag or calling os.execv (locks the install -> flag -> exec order).
This commit is contained in:
@@ -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` and then re-executes itself once (`os.execv`, flagged by the internal `SIENTIA_RUNTIME_READY=1`) so every import loads the runtime's package versions (SIENTIAPDE-2214), 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`.
|
||||
|
||||
@@ -988,6 +988,7 @@ Values are read in `model_manager/utils/connectors_config.py` and `model_manager
|
||||
| `TEMPORAL_NAMESPACE` | Temporal namespace | `model-manager` | No |
|
||||
| `TEMPORAL_USE_TLS` | Use TLS for Temporal gRPC (`true`/`false`). Set `true` when the endpoint serves TLS or you get HTTP redirects (for example 308) to HTTPS | `false` | No |
|
||||
| `RUNTIME` | Suffix for worker task queues (`train_model-<runtime>-queue`, `cleanup_files-<runtime>-queue`) | `single` (via `_get_runtime`) | No |
|
||||
| `SIENTIA_RUNTIME_READY` | Internal: set by the worker itself before re-executing after `install_runtime` (SIENTIAPDE-2214). **Never set it in `.env` or the chart**: it skips the runtime install | unset | No |
|
||||
| `TRAIN_TASK_QUEUE` | Used by **clients** (for example `scripts/run_training_test.py`), not by the worker process | unset | No |
|
||||
| `CLEANUP_TASK_QUEUE` | Used by **clients** (for example `scripts/run_cleanup_test.py`), not by the worker | unset | No |
|
||||
| `POSTGRES_HOST` | PostgreSQL hostname | `localhost` | Yes |
|
||||
|
||||
@@ -21,6 +21,8 @@ Environment Variables:
|
||||
- TEMPORAL_NAMESPACE: Temporal namespace (default: model-manager)
|
||||
- TEMPORAL_USE_TLS: Enable TLS for Temporal connection (default: false)
|
||||
- RUNTIME: Runtime identifier used in queue naming (default: single)
|
||||
- SIENTIA_RUNTIME_READY: Internal, set by the worker itself before re-executing after
|
||||
install_runtime. Never set it in the deployment: it skips the runtime install.
|
||||
- POD_ID: Kubernetes pod identifier for metrics
|
||||
- HTTP_METRICS_PORT: Prometheus metrics server port (default: 9090)
|
||||
- HTTP_SDK_METRICS_PORT: Temporal SDK metrics port (default: 9091)
|
||||
@@ -127,8 +129,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'],
|
||||
@@ -147,11 +147,18 @@ async def main():
|
||||
)
|
||||
|
||||
if os.environ.get('SIENTIA_RUNTIME_READY') != '1':
|
||||
logger.custom_info(f'Installing runtime {runtime}', metadata)
|
||||
await plugin_store.install_runtime(runtime_name=runtime)
|
||||
os.environ['SIENTIA_RUNTIME_READY'] = '1'
|
||||
logger.custom_info(f'Runtime {runtime} installed, re-executing worker', metadata)
|
||||
# ponytail: fresh interpreter so loaded modules match the runtime on disk (SIENTIAPDE-2214)
|
||||
# ponytail: fresh interpreter so loaded modules match the runtime on disk (SIENTIAPDE-2214).
|
||||
# The install_runtime success metrics (count/lag) are lost to the exec; no dashboard uses them.
|
||||
os.execv(sys.executable, [sys.executable, *sys.orig_argv[1:]]) # noqa: S606
|
||||
else:
|
||||
logger.custom_info(
|
||||
f'SIENTIA_RUNTIME_READY=1, runtime {runtime} already installed, skipping install',
|
||||
metadata,
|
||||
)
|
||||
|
||||
activities = Activities(
|
||||
postgres_config=build_postgres_config(),
|
||||
|
||||
@@ -967,6 +967,49 @@ async def test_main_skips_install_when_runtime_ready(
|
||||
mock_activities_class.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch('model_manager.worker.worker.RUNTIME', 'basic')
|
||||
@patch('model_manager.worker.worker.Activities', side_effect=_StopBootError)
|
||||
@patch('model_manager.worker.worker.NotificationHandler')
|
||||
@patch('model_manager.worker.worker.PluginStore')
|
||||
@patch('model_manager.worker.worker.build_plugin_store_config', return_value=_PLUGIN_STORE_CONFIG)
|
||||
@patch(
|
||||
'model_manager.worker.worker.build_mongodb_config',
|
||||
return_value={'connection_string': 'mongodb://test', 'database_name': 'db', 'uri': 'u'},
|
||||
)
|
||||
@patch('model_manager.worker.worker.get_logger')
|
||||
@patch('model_manager.worker.worker.start_prometheus_server')
|
||||
@patch('model_manager.worker.worker.ensure_runtime_directories')
|
||||
async def test_main_install_failure_does_not_reexec(
|
||||
_mock_dirs,
|
||||
_mock_prometheus,
|
||||
_mock_get_logger,
|
||||
_mock_mongo,
|
||||
_mock_ps_config,
|
||||
mock_plugin_store_class,
|
||||
_mock_notification,
|
||||
mock_activities_class,
|
||||
monkeypatch,
|
||||
):
|
||||
"""A failed install propagates without marking the runtime ready or replacing the process."""
|
||||
from model_manager.worker import worker as worker_module
|
||||
|
||||
monkeypatch.setenv('SIENTIA_RUNTIME_READY', '0')
|
||||
mock_plugin_store_class.return_value.install_runtime = AsyncMock(
|
||||
side_effect=RuntimeError('pip failed')
|
||||
)
|
||||
|
||||
with (
|
||||
patch('model_manager.worker.worker.os.execv') as m_execv,
|
||||
pytest.raises(RuntimeError, match='pip failed'),
|
||||
):
|
||||
await worker_module.main()
|
||||
|
||||
assert os.environ['SIENTIA_RUNTIME_READY'] == '0'
|
||||
m_execv.assert_not_called()
|
||||
mock_activities_class.assert_not_called()
|
||||
|
||||
|
||||
@patch('model_manager.worker.worker.asyncio.run')
|
||||
def test_main_entrypoint(mock_asyncio_run):
|
||||
"""Test the __main__ entrypoint."""
|
||||
|
||||
Reference in New Issue
Block a user