16 Commits

Author SHA1 Message Date
PedroHMCosme
29700169ad refactor(worker): iterate a sys.modules copy instead of list() (Sonar S7504) (SIENTIAPDE-2214) 2026-10-22 12:37:11 -03:00
PedroHMCosme
dfbdd4b605 fix(worker): set app_up only after the runtime version guard (SIENTIAPDE-2214)
The probes only check app_up, which was set to 1 when the metrics server started,
before install_runtime. A pod that would fail the guard could turn Ready first, so a
rollout would remove the old pod. app_up now goes to 1 right after the guard.

Also: copy sys.modules keys before iterating, raise the sientia_model floor to the
runtime's >=0.14.0, pin the same runtime versions in requirements-local.txt, test
that the snapshot is taken before the install and the real importlib.metadata path,
and document the boot failure in README and CLAUDE.md.
2026-10-22 12:17:12 -03:00
PedroHMCosme
0f5e4a5cef fix(worker): pin the runtime versions and refuse to boot when install_runtime swaps an imported package (SIENTIAPDE-2214)
Replaces the re-exec. requirements.txt now pins the union of the basic and xgboost
runtime libraries, so install_runtime has nothing to change in the packages the worker
already imported. If a runtime pin changes and requirements.txt is not updated, the
worker logs which packages changed and exits before starting the workers.

test_main_entrypoint assigned worker.main = AsyncMock() without restoring it, so every
later test of main() ran the mock; it now uses patch.object.
2026-10-22 11:34:04 -03:00
PedroHMCosme
993c7d7662 Merge remote-tracking branch 'origin/main' into fix/SIENTIAPDE-2214 2026-10-21 21:46:49 -03:00
dc9fd871bb fix(ci): restaura pull-requests:read no deploy.yml, exigido pelo release-tarball
reusable-deploy-dataops.yml encadeia release-tarball -> dataops-module-release.yml,
que pede pull-requests:read; GitHub valida a interseção de permissions em todo o
workflow_call encadeado, independente do if: que gate o job. Removido por engano
em 586a5b9, achando que só existia por causa do trigger pull_request antigo.

[skip ci]
2026-10-21 14:10:25 -03:00
242566f9b1 fix(ci): grant contents:write permission to version-bump workflow
Caller workflow had no permissions block, so GitHub's permission-
intersection rule capped the nested reusable-version-bump.yml call
to read-only, breaking dispatch ("requesting 'contents: write', but
is only allowed 'contents: read'").

[skip ci]
2026-10-21 11:44:53 -03:00
PedroHMCosme
7903dda730 Merge branch 'main' into fix/SIENTIAPDE-2214 2026-10-21 09:28:48 -03:00
d0b4644f63 chore(ci): add version-bump.yml, decoupled from deploy
New workflow bumps pyproject.toml's version on PR merge, calling the
shared reusable-version-bump.yml (SIENTIAPDE-2217 version-bump reform) —
zero build/image/chart/deploy side effects. deploy.yml (workflow_dispatch-
only) will read this committed value instead of calculating one.

[skip ci]
2026-10-21 00:18:51 -03:00
ced847bbe7 chore(ci): switch deploy.yml to manual workflow_dispatch with required chart_ref
Removes auto-deploy-on-PR-merge trigger and the old PR-merge gate. Deploy now
triggers only via workflow_dispatch, with chart_ref required (no default) as
the literal sientia-helm-chart branch bumped and deployed — hard error if it
doesn't exist, no silent fallback to development/main.

- .github/workflows/deploy.yml: drop pull_request trigger, add chart_ref
  input, branch_name now github.ref_name, drop pull-requests:read permission
- Delete .github/workflows/require-chart-ref-label.yml
- README.md: update Deploy section to describe manual trigger

[skip ci]
2026-10-20 17:03:33 -03:00
PedroHMCosme
262ba8d658 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).
2026-10-20 16:30:23 -03:00
PedroHMCosme
0c9c5f300b fix(worker): re-exec after install_runtime so imports match the runtime (SIENTIAPDE-2214)
The worker imports sklearn (metrics, evidently) before
PluginStore.install_runtime replaces it on disk with the runtime's pinned
version, the same order that breaks every sklearn prediction in laborious.
After installing the runtime, the worker now replaces its process once
(SIENTIA_RUNTIME_READY=1 skips the second install), so every module is
loaded from the runtime's packages.
2026-10-20 15:40:38 -03:00
49c8942a87 chore(ci): remove feature/** do push trigger (evita run duplicado) [skip ci] 2026-10-19 15:38:47 -03:00
github-actions[bot]
3b90812c12 chore: bump version to 2.0.3 2026-10-16 15:10:26 +00:00
0b673b35a1 Merge pull request #33 from Aignosi/fix/SIENTIAPDE-2201
SIENTIAPDE-2201: Keep the timestamp column in train/test data CSVs
2026-10-16 12:10:01 -03:00
PedroHMCosme
37f16f34ff SIENTIAPDE-2201: fix: keep the timestamp column in train/test data CSVs
generate_report wrote train_data.csv and test_data.csv with index=False, which
dropped the UTC time index. laborious drift reads test_data.csv from MLflow as
its reference and needs a timestamp column; without it, every model trained here
drifted against the first 30% of its own target window. Both CSVs now write the
index as timestamp. If a column already has that name, the CSV is written as
before and a warning is logged, so training never fails over drift.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 10:09:06 -03:00
dd981b8027 fix(ci): Update env_list_path to array format in deploy workflow
This commit changes the env_list_path parameter in the deploy GitHub Actions workflow to an array format, aligning with expected input types for the action.
2026-09-11 00:41:42 -03:00
13 changed files with 303 additions and 44 deletions

View File

@@ -1,10 +1,15 @@
name: Deploy DataOps Worker name: Deploy DataOps Worker
on: on:
pull_request:
types: [closed]
branches: [main, 'release/**', 'feature/**']
workflow_dispatch: workflow_dispatch:
inputs:
chart_ref:
description: >
sientia-helm-chart branch to bump GITHUB_BRANCH on AND deploy — the literal
deploy target. Validated via git ls-remote; a branch that does not exist in
sientia-helm-chart is a hard error, not a silent fallback to main.
required: true
type: string
permissions: permissions:
contents: write contents: write
@@ -14,13 +19,13 @@ permissions:
jobs: jobs:
deploy: deploy:
name: Deploy name: Deploy
if: github.event_name == 'workflow_dispatch' || github.event.pull_request.merged == true
uses: Aignosi/github_workflow_templates/.github/workflows/reusable-deploy-dataops.yml@main uses: Aignosi/github_workflow_templates/.github/workflows/reusable-deploy-dataops.yml@main
with: with:
branch_name: ${{ github.event.pull_request.head.ref }} branch_name: ${{ github.ref_name }}
chart_ref: ${{ inputs.chart_ref }}
project_name: 'model-manager' project_name: 'model-manager'
chart_name: 'model-manager' chart_name: 'model-manager'
env_list_path: '.global.env' env_list_path: '[".global.env"]'
update_version_in: '["pyproject"]' update_version_in: '["pyproject"]'
create_tarball: ${{ github.event_name == 'pull_request' }} create_tarball: ${{ github.event_name == 'pull_request' }}
mirror: true mirror: true

View File

@@ -5,7 +5,6 @@ on:
branches: branches:
- main - main
- 'release/**' - 'release/**'
- 'feature/**'
pull_request: pull_request:
branches: branches:
- main - main

View File

@@ -1,16 +0,0 @@
name: Require chart-ref label
on:
pull_request:
types: [opened, synchronize, reopened, labeled, unlabeled]
branches:
- main
- 'release/**'
- 'feature/**'
permissions:
pull-requests: write
jobs:
require-chart-ref-label:
uses: Aignosi/github_workflow_templates/.github/workflows/require-chart-ref-label.yml@main

17
.github/workflows/version-bump.yml vendored Normal file
View File

@@ -0,0 +1,17 @@
name: Version Bump
permissions:
contents: write
on:
pull_request:
types: [closed]
branches: [main, 'release/**', 'feature/**']
jobs:
bump:
if: github.event.pull_request.merged == true
uses: Aignosi/github_workflow_templates/.github/workflows/reusable-version-bump.yml@main
with:
update_version_in: '["pyproject"]'
secrets: inherit

View File

@@ -45,7 +45,7 @@ Coverage gate in CI is 80% minimum; this repo targets 100% in practice — don't
## Architecture ## Architecture
### Temporal layering ### 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. - **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. - `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`. - `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`.

View File

@@ -149,6 +149,7 @@ The Model Manager system uses a Temporal-based workflow architecture with clear
- Prometheus metrics server initialization - Prometheus metrics server initialization
- Notification handler setup and configuration - Notification handler setup and configuration
- **Runtime Installation**: Automatically installs the required model runtime from the Plugin Store - **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**: - **Key Features**:
- Automatic scaling with `PollerBehaviorAutoscaling` - Automatic scaling with `PollerBehaviorAutoscaling`
- Health check endpoints for Kubernetes liveness/readiness probes - Health check endpoints for Kubernetes liveness/readiness probes
@@ -813,7 +814,9 @@ Runs automatically on each Pull Request to `main`:
- ✅ Automatic version calculation based on branch pattern - ✅ Automatic version calculation based on branch pattern
#### Deploy (`.github/workflows/deploy.yml`) #### Deploy (`.github/workflows/deploy.yml`)
Runs automatically when a PR is merged to `main`: Triggered manually via `workflow_dispatch`, with a required `chart_ref` input (no
default — the sientia-helm-chart branch to bump and deploy; a branch that doesn't
exist is a hard error, not a silent fallback):
- ✅ Builds and pushes Docker image to Azure Container Registry - ✅ Builds and pushes Docker image to Azure Container Registry
- ✅ Creates GitHub release with calculated version - ✅ Creates GitHub release with calculated version
- ✅ Deploys to Kubernetes using Helm - ✅ Deploys to Kubernetes using Helm
@@ -1247,6 +1250,10 @@ sientia-dataops-model-manager/
- Check training parameter validation errors - Check training parameter validation errors
- Verify input data format and required fields - 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 ### Debug Mode
Enable debug logging by setting the log level: Enable debug logging by setting the log level:

View File

@@ -527,6 +527,32 @@ class DataManagerRepository(SientiaMonitoring):
self.error(error_msg, metadata) self.error(error_msg, metadata)
raise OSError(error_msg) from e raise OSError(error_msg) from e
def _save_data_csv(
self, frame: pd.DataFrame, file_path: str, metadata: dict[str, Any] | None = None
) -> None:
"""
Write a train/test frame to CSV with its time index as a ``timestamp`` column.
laborious drift reads ``test_data.csv`` from MLflow as its reference and needs
``timestamp`` (SIENTIAPDE-2201). If a column already has that name, the frame is
written without the index and that model's drift keeps falling back to the target
window: training never fails over drift.
Args:
frame: Frame indexed by the UTC training timestamps.
file_path: Destination CSV path.
metadata: Workflow metadata for logging.
"""
if 'timestamp' in frame.columns:
self.warning(
f"Column 'timestamp' already exists; {path.basename(file_path)} saved without "
'the time column (drift reference falls back to the target window)',
metadata,
)
frame.to_csv(file_path, index=False)
return
frame.to_csv(file_path, index_label='timestamp')
def generate_report( def generate_report(
self, data: TrainModelResult, metadata: dict[str, Any] | None = None self, data: TrainModelResult, metadata: dict[str, Any] | None = None
) -> TrainModelResult: ) -> TrainModelResult:
@@ -598,11 +624,11 @@ class DataManagerRepository(SientiaMonitoring):
# Save training / validation CSVs using the same frames as the report (includes # Save training / validation CSVs using the same frames as the report (includes
# literal `target` alias for Evidently, plus predictions and float-cast features). # literal `target` alias for Evidently, plus predictions and float-cast features).
data.train_data_path = path.join(data.run_dir, 'train_data.csv') data.train_data_path = path.join(data.run_dir, 'train_data.csv')
reference_data_float.to_csv(data.train_data_path, index=False) self._save_data_csv(reference_data_float, data.train_data_path, metadata)
# Save test data CSV # Save test data CSV
data.test_data_path = path.join(data.run_dir, 'test_data.csv') data.test_data_path = path.join(data.run_dir, 'test_data.csv')
current_data_float.to_csv(data.test_data_path, index=False) self._save_data_csv(current_data_float, data.test_data_path, metadata)
# Save equation as JSON # Save equation as JSON
if data.equation is not None and data.params.model_type == 'linear_regression': if data.equation is not None and data.params.model_type == 'linear_regression':

View File

@@ -32,6 +32,7 @@ from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig
with workflow.unsafe.imports_passed_through(): with workflow.unsafe.imports_passed_through():
import asyncio import asyncio
import importlib.metadata
import os import os
import sys import sys
@@ -76,6 +77,36 @@ def _get_runtime(runtime: str | None) -> str:
return normalized_runtime or 'single' 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(): async def main():
""" """
Main entry point for the Model Manager worker application. 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: This function initializes and starts all components of the worker:
1. Sets up logging and metadata 1. Sets up logging and metadata
2. Starts Prometheus metrics server 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 4. Creates and configures activities
5. Starts Temporal client and workers 5. Starts Temporal client and workers
6. Manages worker lifecycle and graceful shutdown 6. Manages worker lifecycle and graceful shutdown
@@ -144,8 +177,23 @@ async def main():
metrics_controller=metrics_controller, metrics_controller=metrics_controller,
) )
loaded_before_install = _loaded_distribution_versions()
await plugin_store.install_runtime(runtime_name=runtime) 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( activities = Activities(
postgres_config=build_postgres_config(), postgres_config=build_postgres_config(),
mlflow_config=build_mlflow_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. Starts the Prometheus metrics server for monitoring and observability.
This function initializes the Prometheus HTTP server on the configured port This function initializes the Prometheus HTTP server on the configured port.
and sets the application health metric to indicate the service is running. ``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 server exposes metrics that can be scraped by Prometheus for monitoring
the health and performance of the Model Manager worker. 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)) port = int(os.getenv('HTTP_METRICS_PORT', 9090))
start_http_server(port) start_http_server(port)
logger.custom_info(f'Prometheus server initialized on port {port}.', metadata) 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 except Exception as e: # noqa: BLE001
logger.custom_critical(f'Failed to start Prometheus server: {e}', metadata) logger.custom_critical(f'Failed to start Prometheus server: {e}', metadata)
os._exit(1) os._exit(1)

View File

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "model-manager" name = "model-manager"
version = "2.0.2" version = "2.0.3"
description = "Sientia DataOps Model Manager - ML Model Orchestration System" description = "Sientia DataOps Model Manager - ML Model Orchestration System"
readme = "README.md" readme = "README.md"
requires-python = ">=3.11" requires-python = ">=3.11"

View File

@@ -9,3 +9,14 @@ prometheus-client==0.23.1
beautifulsoup4==4.12.3 beautifulsoup4==4.12.3
evidently==0.6.7 evidently==0.6.7
jsonschema==4.26.0 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

View File

@@ -4,8 +4,19 @@ sqlalchemy==2.0.50
boto3==1.42.70 boto3==1.42.70
botocore==1.42.70 botocore==1.42.70
sientia_do>=1.12.2 sientia_do>=1.12.2
sientia_model>=0.8.1 sientia_model>=0.14.0
prometheus-client==0.23.1 prometheus-client==0.23.1
beautifulsoup4==4.12.3 beautifulsoup4==4.12.3
evidently==0.6.7 evidently==0.6.7
jsonschema==4.26.0 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

View File

@@ -514,6 +514,58 @@ def test_generate_report_adds_target_alias_for_reports(tmp_path):
assert test_csv['target'].equals(test_csv['t']) assert test_csv['target'].equals(test_csv['t'])
_UTC_INDEX = pd.DatetimeIndex(['2024-01-01 02:00', '2024-01-01 01:00'], tz='UTC')
def _dated_result(**train_columns: list[float]) -> TrainModelResult:
"""Frames indexed by UTC time, like prepare_training_data + predict produce them."""
columns = train_columns or {'v1': [1.0, 2.0], 't': [1.0, 2.0]}
return TrainModelResult(
params=TrainModelParams.from_dict(_minimal_dict_for_prepare()),
train_data=pd.DataFrame(columns, index=_UTC_INDEX),
val_data=pd.DataFrame(columns, index=_UTC_INDEX),
y_train_pred=pd.DataFrame({'t': [1.0, 2.0]}, index=_UTC_INDEX),
y_pred=pd.DataFrame({'t': [1.0, 2.0]}, index=_UTC_INDEX),
run_name='testrun',
)
def _generate_report(repo, tmr: TrainModelResult, tmp_path) -> TrainModelResult:
with (
patch.object(repo, '_get_reports_directory', return_value=str(tmp_path)),
patch('model_manager.utils.repository.data_manager_repository.Reports'),
):
return repo.generate_report(tmr, {})
def test_generate_report_csvs_keep_the_time_index_as_timestamp(tmp_path):
"""laborious drift reads test_data.csv as its reference and needs ``timestamp``."""
repo = dmr.DataManagerRepository(MagicMock())
out = _generate_report(repo, _dated_result(), tmp_path)
for csv_path in (out.train_data_path, out.test_data_path):
csv = pd.read_csv(csv_path)
assert list(csv.columns) == ['timestamp', 'prediction', 'v1', 't', 'target']
assert pd.to_datetime(csv['timestamp']).tolist() == list(_UTC_INDEX)
def test_generate_report_csvs_skip_timestamp_when_a_column_already_has_the_name(tmp_path):
"""A variable literally named ``timestamp`` wins; training must not fail over drift."""
repo = dmr.DataManagerRepository(MagicMock())
tmr = _dated_result(timestamp=[5.0, 6.0], v1=[1.0, 2.0], t=[1.0, 2.0])
with patch.object(repo, 'warning') as warning:
out = _generate_report(repo, tmr, tmp_path)
for csv_path in (out.train_data_path, out.test_data_path):
csv = pd.read_csv(csv_path)
assert list(csv.columns) == ['prediction', 'timestamp', 'v1', 't', 'target']
assert csv['timestamp'].tolist() == [5.0, 6.0]
assert warning.call_count == 2
assert "Column 'timestamp' already exists" in warning.call_args.args[0]
def test_generate_report_skips_equation_file_when_not_linear(tmp_path): def test_generate_report_skips_equation_file_when_not_linear(tmp_path):
repo = dmr.DataManagerRepository(MagicMock()) repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare()) p = TrainModelParams.from_dict(_minimal_dict_for_prepare())

View File

@@ -3,7 +3,7 @@
import asyncio import asyncio
import os import os
import sys import sys
from unittest.mock import AsyncMock, Mock, patch from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest import pytest
@@ -132,9 +132,8 @@ def test_start_prometheus_server_success(
# Verify HTTP server started # Verify HTTP server started
mock_start_http_server.assert_called_once_with(9090) mock_start_http_server.assert_called_once_with(9090)
# Verify APP_UP metric was set to 1 # app_up goes to 1 in main(), after the runtime version guard.
mock_metrics.APP_UP.labels.assert_called_once_with(pod_id='test-pod-123') mock_app_up.set.assert_not_called()
mock_app_up.set.assert_called_once_with(1)
mock_logger.custom_info.assert_called_once() mock_logger.custom_info.assert_called_once()
@@ -284,7 +283,8 @@ async def test_main_successful_startup(
# Verify cleanup was performed # Verify cleanup was performed
mock_notification_handler.shutdown.assert_called_once() mock_notification_handler.shutdown.assert_called_once()
mock_activities.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 @pytest.mark.asyncio
@@ -848,12 +848,10 @@ async def test_main_missing_runtime_uses_single_fallback(
def test_main_entrypoint(mock_asyncio_run): def test_main_entrypoint(mock_asyncio_run):
"""Test the __main__ entrypoint.""" """Test the __main__ entrypoint."""
# Import and execute the main block # Import and execute the main block
with patch.object(sys, 'argv', ['worker.py']): import model_manager.worker.worker as worker_module
import model_manager.worker.worker as worker_module
# Simulate running the module
worker_module.main = AsyncMock()
# 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()) # This would normally be called by asyncio.run(main())
# We just verify the pattern is correct # We just verify the pattern is correct
assert callable(worker_module.main) assert callable(worker_module.main)
@@ -905,3 +903,105 @@ def test_start_prometheus_server_prints_failure(
# Agora o erro é logado via logger crítico # Agora o erro é logado via logger crítico
mock_logger.custom_critical.assert_called_once() mock_logger.custom_critical.assert_called_once()
mock_exit.assert_called_once_with(1) 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'}