diff --git a/.env.example b/.env.example index c8e0e7e..22fea0c 100644 --- a/.env.example +++ b/.env.example @@ -1,52 +1,46 @@ -POSTGRES_HOST=paradedb-rw.paradedb.svc.cluster.local +POSTGRES_HOST=localhost POSTGRES_PORT=5432 -POSTGRES_USER=sientia -POSTGRES_PASSWORD=password +POSTGRES_USER=postgres +POSTGRES_PASSWORD=changeme POSTGRES_DBNAME=sientia POSTGRES_MIN_CONNECTIONS=10 POSTGRES_MAX_CONNECTIONS=30 -MLFLOW_URL=http://sientia-tracker-mlflow-tracking.sientia-tracker.svc.cluster.local:80 +MLFLOW_URL=http://localhost:5080 MLFLOW_USERNAME=aignosi -MLFLOW_PASSWORD=mlflow_password +MLFLOW_PASSWORD=changeme LOG_LEVEL=DEBUG HTTP_METRICS_PORT=9090 HTTP_SDK_METRICS_PORT=9091 PROJECT_NAME=sientia-model-manager -TEMPORAL_HOST=temporal-frontend.temporal.svc.cluster.local:7233 +TEMPORAL_HOST=localhost:7233 TEMPORAL_NAMESPACE=model-manager TEMPORAL_USE_TLS=false -RUNTIME=basic +RUNTIME=single -STORE_BASE_URL=http://gitea-http.gitea.svc.cluster.local -STORE_OWNER=aignosi -STORE_REPO=suse-model-store -STORE_BRANCH=main -STORE_USERNAME= -STORE_PASSWORD= -STORE_CACHE_TTL_SECONDS=3600 -PYPI_SERVER=http://library-distribution-server.library.svc.cluster.local:5000 -PYPI_USERNAME= -PYPI_PASSWORD= - -MONGODB_USERNAME=mongo_user -MONGODB_PASSWORD=mongo_db_password -MONGODB_URL=my-release-mongodb.mongodb.svc.cluster.local:27017 +MONGODB_USERNAME=root +MONGODB_PASSWORD=changeme +MONGODB_URL=localhost:27017 MONGODB_DATABASE=sientia MONGODB_TTL_INDEX_HOURS=1 -MINIO_ENDPOINT_URL=http://minio.minio.svc.cluster.local:9000 +MINIO_ENDPOINT_URL=http://localhost:9000 MINIO_ACCESS_KEY=minioadmin MINIO_SECRET_KEY=minioadmin -MINIO_REGION=us-east-1 -MINIO_USE_SSL=false -MINIO_MAX_RETRY_ATTEMPTS=3 -MINIO_RETRY_MODE=adaptive -MINIO_CONNECT_TIMEOUT=10 -MINIO_READ_TIMEOUT=60 MINIO_DEFAULT_BUCKET=model-training +MINIO_SECURE=false + +STORE_BASE_URL=http://localhost:3000 +STORE_OWNER=aignosi +STORE_REPO=suse-model-store +STORE_USERNAME= +STORE_PASSWORD= +STORE_CACHE_TTL_SECONDS=3600 +PYPI_SERVER=http://localhost:5000 +PYPI_USERNAME= +PYPI_PASSWORD= TIMEOUT_VALIDATE_PARAMS=30 TIMEOUT_TRAIN_MODEL=2700 @@ -61,5 +55,3 @@ CLEANUP_SCHEDULE_ID=cleanup-files-daily CLEANUP_CRON="0 0 * * *" CLEANUP_TIMEZONE=UTC CLEANUP_EXECUTION_TIMEOUT_HOURS=1 - -EXTRA_PIP_REQUIREMENTS=git+https://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@github.com/Aignosi/sientia-mlops-library.git diff --git a/README.md b/README.md index 60cd104..a50e437 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,7 @@ An enterprise-grade ML model training orchestration platform built on Temporal. - [Security Architecture](#security-architecture) - [Workflows](#workflows) - [Train Model Workflow](#train-model-workflow-train_modelpy) + - [Train model workflow input (sample)](#train-model-workflow-input-sample) - [Cleanup Files Workflow](#cleanup-files-workflow-cleanup_filespy) - [Installation & Setup](#installation--setup) - [Prerequisites](#prerequisites) @@ -154,9 +155,7 @@ The Model Manager system uses a Temporal-based workflow architecture with clear - Health check endpoints for Kubernetes liveness/readiness probes - Graceful shutdown with cleanup procedures - Multi-instance deployment support - - Two dedicated task queues: - - `train_model-queue`: ML model training workflows - - `cleanup-queue`: File cleanup workflows + - Task queues are derived from `RUNTIME` (default `single`): `train_model--queue` and `cleanup_files--queue` (see `prepare_worker.build_queue_name`) - Automated cleanup schedule management #### **Workflows (`model_manager/workflows/`)** @@ -276,39 +275,13 @@ The **TrainModel** workflow orchestrates the complete ML model training pipeline - **Structured Logging**: Rich context in error messages for debugging - **Business Validation**: 10 business rules including range checks, consistency validation, and data integrity -#### Input Parameters -```json -{ - "experiment_run_id": 123, - "target_variable": "price", - "variable_columns": ["feature1", "feature2"], - "train_size": 80, - "shuffle": true, - "use_scaler": true, - "include_ar": false, - "bucket_name": "ml-data", - "file_name": "training_data.csv", - "line_separator": "\n", - "decimal_separator": ".", - "lag_train": {"feature1": 0, "feature2": 2}, - "lag_val": {"feature1": 0, "feature2": 1}, - "rem_static_win": false, - "static_threshold": null, - "low_lim": {"feature1": 0.0, "feature2": 0.0}, - "upp_lim": {"feature1": 100.0, "feature2": 100.0}, - "window": 10, - "experiment_name": "production_model_v1", - "removed_intervals": [], - "model_name": "Linear Regression", - "degree": 1, - "interaction_only": false, - "nan_treatment": "drop", - "start_date": null, - "end_date": null, - "scaler_name": "Standard Scaler", - "support_filters": {} -} -``` +#### Train model workflow input (sample) + +The workflow receives **one argument**: a JSON-serializable object whose keys match `TrainModelParams` (`model_manager/utils/models/train_model_params.py`). All fields are passed at the **top level** (not nested under `train_params`). + +A **minimal valid example** (only keys required by `TrainModelParams.from_dict`, plus a minimal `model_metadata` for `validate_business_rules`) is in **`input-sample.json`**. See also **`input-sample.md`** for SQL/MinIO notes. Optional inputs include `date_column`, `date_format`, `random_state` (defaults to `42`), `val_file_name`, and `model_id`. + +When starting the workflow from a Temporal client, use the same **task queue** as the worker: `train_model--queue` (for example `train_model-single-queue` when `RUNTIME=single`). #### Architecture Diagram ```mermaid @@ -333,24 +306,17 @@ The workflow implements 5 different retry policies optimized for each operation | **No Retry** | - | - | - | 1 | Training/Validation (permanent data errors) | | **Database** | 2s | 20s | 2.0x | 5 | PostgreSQL updates (lock contention) | -#### Business Validation Rules +#### Business validation rules -The workflow validates comprehensive business rules beyond type checking: +`TrainModelParams.validate_business_rules()` runs after type coercion. Notable checks: -1. **train_size**: Must be between 10-100% -2. **variable_columns**: Cannot be empty -3. **lag_train, lag_val**: Per-variable dictionaries with non-negative values -4. **window**: Must be non-negative integer -5. **low_lim/upp_lim**: Must have same keys and low < upp for each variable -6. **target_variable**: Cannot be empty -7. **bucket_name, file_name, experiment_name**: Cannot be empty or whitespace -8. **degree**: Must be at least 1; must be >= 2 for Polynomial Regression -9. **nan_treatment**: Must be one of 'drop', 'linear interpolation', 'fill linear' -10. **scaler_name**: Must be 'Standard Scaler' or 'None' -11. **model_name**: Must be 'Linear Regression' or 'Polynomial Regression' -12. **Polynomial Regression requires Scaler**: Models with degree > 1 must have a scaler to prevent numerical overflow -13. **Linear Regression requires degree 1**: Linear models must have degree = 1 -14. **static_threshold**: When `rem_static_win` is true and `static_threshold` has a value, it must be between 1 and 1000 (inclusive). If null, defaults to 1. +1. **train_size**: Between 10 and 100 (percent). +2. **variable_columns**: Non-empty list. +3. **model_metadata**: Required (non-empty) for validation to succeed; may include JSON Schema definitions under `model_metadata.schemas.components.schemas` for `data_model`, `model`, and `opt_params` when you want schema validation of the corresponding kwargs. +4. **target_variable**, **bucket_name**, **file_name**, **model_name**: Non-empty strings (no whitespace-only values). +5. **date_format**: When set, must be an allowed frontend date format (see `validate_frontend_date_format`). + +Model-specific rules (for example polynomial degree and scaler requirements) live in the training stack and integration scenarios; see `docs/test-scenarios/` and `scripts/run_training_test.py` for scenario-based examples. ### Cleanup Files Workflow (`cleanup_files.py`) @@ -386,10 +352,10 @@ The cleanup schedule is automatically created when the worker starts: | Configuration | Environment Variable | Default | Description | |--------------|---------------------|---------|-------------| -| **Schedule ID** | `CLEANUP_SCHEDULE_ID` | `cleanup-files-daily` | Unique identifier for the schedule | +| **Schedule ID** | (derived) | `cleanup-files--daily` | Built from `RUNTIME` in `cleanup_schedule.build_cleanup_schedule_id` | | **Cron Expression** | `CLEANUP_CRON` | `0 0 * * *` | Daily at midnight UTC | | **Timezone** | `CLEANUP_TIMEZONE` | `UTC` | Timezone for cron execution | -| **Task Queue** | `CLEANUP_TASK_QUEUE` | `cleanup-queue` | Dedicated task queue | +| **Task Queue** | (derived) | `cleanup_files--queue` | Must match the cleanup worker queue (`build_queue_name('CleanupFiles', runtime)`) | | **Execution Timeout** | `CLEANUP_EXECUTION_TIMEOUT_HOURS` | `1` | Maximum execution time (hours) | | **Retention Period** | `CLEANUP_RETENTION_HOURS` | `24` | Files older than this are deleted | | **Dry Run** | `CLEANUP_DRY_RUN` | `false` | Test mode without actual deletion | @@ -739,6 +705,8 @@ fi python -m model_manager.worker.worker ``` +To start a **`train_model`** run from your own Temporal client, use the payload shape in **`input-sample.json`** (task queue `train_model--queue`, matching `RUNTIME` on the worker). For scripted tests that use the JSON scenarios under `docs/test-scenarios/`, see **`scripts/run_training_test.py`**. + ## Code Quality & Validation ### Overview @@ -1003,13 +971,10 @@ Test scenarios are defined as JSON files in `docs/test-scenarios/`. Each scenari 3. Save with a descriptive name: `XX-description.json` 4. Run with: `python scripts/run_training_test.py --scenario XX-description` -#### Important Validations +#### Important validations -The training workflow enforces several business rules: - -- **Polynomial Regression requires Scaler**: Models with `degree > 1` must have `useScaler: true` and a valid `scalerName` to prevent numerical overflow -- **Static Window Removal requires DatetimeIndex**: Scenarios with `remStaticWin: true` require data with a timestamp column for the `TimeSeriesDiscontinuityAnalyzer` -- **Variable Limits Consistency**: `lowLim` and `uppLim` must have matching keys, and `lowLim[key] < uppLim[key]` for all variables +- **Workflow payload** (`input-sample.json`, Temporal `execute_workflow`): snake_case fields validated by `TrainModelParams` (see [Business validation rules](#business-validation-rules) above). +- **Integration scenarios** (`docs/test-scenarios/*.json`): camelCase UI-oriented fields consumed by `scripts/run_training_test.py`, which maps them into `TrainModelParams` before running. Additional rules apply there (for example polynomial degree and scaler requirements, static window removal, variable limits); see scenario descriptions in the table above. ## Monitoring and Metrics @@ -1044,49 +1009,65 @@ The Model Manager system exposes comprehensive Prometheus metrics for operationa ### Environment Variables +Values are read in `model_manager/utils/connectors_config.py` and `model_manager/worker/worker.py`. Defaults below match the code. + | Variable | Description | Default | Required | |----------|-------------|---------|----------| -| `TEMPORAL_HOST` | Temporal server address | `localhost:7233` | Yes | +| `TEMPORAL_HOST` | Temporal server address (`host:port`) | `localhost:7233` | Yes | | `TEMPORAL_NAMESPACE` | Temporal namespace | `model-manager` | No | -| `TEMPORAL_USE_TLS` | Enable TLS for Temporal connection | `false` | No | -| `TRAIN_TASK_QUEUE` | Task queue for training workflows | `train_model-queue` | No | -| `CLEANUP_TASK_QUEUE` | Task queue for cleanup workflows | `cleanup-queue` | 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--queue`, `cleanup_files--queue`) | `single` (via `_get_runtime`) | 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 | | `POSTGRES_PORT` | PostgreSQL port | `5432` | Yes | | `POSTGRES_USER` | PostgreSQL username | `sientia` | Yes | | `POSTGRES_PASSWORD` | PostgreSQL password | `sientia` | Yes | | `POSTGRES_DBNAME` | PostgreSQL database | `sientia` | Yes | -| `POSTGRES_MIN_CONNECTIONS` | Minimum PostgreSQL connections | `5` | No | -| `POSTGRES_MAX_CONNECTIONS` | Maximum PostgreSQL connections | `20` | No | -| `MLFLOW_URL` | MLFlow server URL (full URL with protocol and port) | `http://localhost:5080` | Yes | -| `MLFLOW_USERNAME` | MLFlow username | `aignosi` | Yes | -| `MLFLOW_PASSWORD` | MLFlow password | `aignosi` | Yes | -| `MINIO_ENDPOINT_URL` | MinIO server endpoint | `http://minio.minio.svc.cluster.local:9000` | Yes | +| `POSTGRES_MIN_CONNECTIONS` | Minimum pool size | `5` | No | +| `POSTGRES_MAX_CONNECTIONS` | Maximum pool size | `20` | No | +| `MLFLOW_URL` | MLflow tracking URL (scheme, host, and port) | `http://localhost:5080` | Yes | +| `MLFLOW_USERNAME` | MLflow basic auth username | `aignosi` | Yes | +| `MLFLOW_PASSWORD` | MLflow basic auth password | `aignosi` | Yes | +| `MINIO_ENDPOINT_URL` | MinIO / S3 endpoint URL | `http://localhost:9000` | Yes | | `MINIO_ACCESS_KEY` | MinIO access key | `minioadmin` | Yes | | `MINIO_SECRET_KEY` | MinIO secret key | `minioadmin` | Yes | | `MINIO_REGION` | MinIO region | `us-east-1` | No | -| `MINIO_USE_SSL` | Enable SSL for MinIO | `false` | No | -| `MINIO_MAX_RETRY_ATTEMPTS` | Maximum retry attempts | `3` | No | -| `MINIO_RETRY_MODE` | Retry mode (standard/adaptive) | `adaptive` | No | +| `MINIO_SECURE` | Use TLS for MinIO client (`true`/`false`) | `false` | No | +| `MINIO_DEFAULT_BUCKET` | Default bucket for `MinioRepository` | `model-training` | No | +| `MINIO_MAX_RETRY_ATTEMPTS` | S3 retry attempts | `3` | No | +| `MINIO_RETRY_MODE` | Retry mode | `adaptive` | No | | `MINIO_CONNECT_TIMEOUT` | Connection timeout (seconds) | `10` | No | | `MINIO_READ_TIMEOUT` | Read timeout (seconds) | `60` | No | -| `MONGODB_URL` | MongoDB connection URI | `localhost:27018` | Yes | +| `MONGODB_URL` | MongoDB host:port (no scheme; used inside connection string) | `localhost:27018` | Yes | | `MONGODB_USERNAME` | MongoDB username | `root` | Yes | | `MONGODB_PASSWORD` | MongoDB password | `wKZDbMNU1c` | Yes | | `MONGODB_DATABASE` | MongoDB database name | `sientia` | Yes | -| `MONGODB_TTL_INDEX_HOURS` | MongoDB TTL index hours | `1` | No | -| `CLEANUP_SCHEDULE_ID` | Cleanup schedule identifier | `cleanup-files-daily` | No | -| `CLEANUP_CRON` | Cleanup cron expression | `0 0 * * *` | No | +| `MONGODB_TTL_INDEX_HOURS` | TTL index duration (hours) | `1` | No | +| `STORE_BASE_URL` | Plugin store Git server base URL | `http://localhost:3000` | No | +| `STORE_OWNER` | Git owner/org | `sientia` | No | +| `STORE_REPO` | Git repository | `model-library-store` | No | +| `STORE_BRANCH` | Optional branch | unset | No | +| `STORE_USERNAME` / `STORE_PASSWORD` | Git HTTP credentials | unset | No | +| `STORE_CACHE_TTL_SECONDS` | Plugin index cache TTL | unset | No | +| `PYPI_SERVER` | Custom PyPI index URL | `http://localhost:5000` | No | +| `PYPI_USERNAME` / `PYPI_PASSWORD` | PyPI credentials | unset | No | +| `CLEANUP_CRON` | Cleanup schedule cron | `0 0 * * *` | No | | `CLEANUP_TIMEZONE` | Cleanup schedule timezone | `UTC` | No | -| `CLEANUP_EXECUTION_TIMEOUT_HOURS` | Cleanup execution timeout | `1` | No | -| `CLEANUP_RETENTION_HOURS` | File retention period (hours) | `24` | No | -| `CLEANUP_DRY_RUN` | Dry-run mode (no actual deletion) | `false` | No | -| `LOG_LEVEL` | Application log level | `INFO` | No | -| `PROJECT_NAME` | Project name for metrics | `model-manager` | No | +| `CLEANUP_EXECUTION_TIMEOUT_HOURS` | Cleanup workflow timeout (hours) | `1` | No | +| `CLEANUP_RETENTION_HOURS` | Local temp retention (hours) | `24` | No | +| `CLEANUP_DRY_RUN` | Cleanup dry-run | `false` | No | +| `LOG_LEVEL` | Log level | `INFO` | No | +| `PROJECT_NAME` | Project name for notifications/metrics | `model-manager` | No | | `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No | | `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No | -| `POD_ID` | Kubernetes pod identifier | `None` | No | -| `EXTRA_PIP_REQUIREMENTS` | Extra pip requirements for MLFlow model serving | `None` | No | +| `POD_ID` | Pod label for metrics | unset | No | +| `EXTRA_PIP_REQUIREMENTS` | Extra `pip` packages for runtime installs | unset | No | +| `TIMEOUT_VALIDATE_PARAMS` | Activity timeout (seconds) | `30` | No | +| `TIMEOUT_TRAIN_MODEL` | Training activity timeout (seconds) | `2700` | No | +| `TIMEOUT_DELETE_FILE` | Delete/cleanup activity timeout (seconds) | `120` | No | +| `TIMEOUT_UPDATE_DATABASE` | DB update activity timeout (seconds) | `30` | No | +| `TIMEOUT_CLEANUP_LOCAL` | Cleanup workflow activity timeout (seconds) | `120` | No | #### Workflow Activity Timeouts @@ -1098,6 +1079,7 @@ These timeouts control how long each activity in workflows can run before timing |----------|-------------|---------|-------------------| | `TIMEOUT_VALIDATE_PARAMS` | Parameter validation timeout | `30` | Fast operation, no I/O | | `TIMEOUT_TRAIN_MODEL` | Model training timeout | `2700` | Large dataset processing (45 min) | +| `TIMEOUT_DELETE_FILE` | File delete / related I/O timeout | `120` | Network storage | | `TIMEOUT_UPDATE_DATABASE` | Database update timeout | `30` | PostgreSQL update query (30 sec) | **Cleanup Workflow Timeouts:** @@ -1246,6 +1228,7 @@ sientia-dataops-model-manager/ ├── requirements-dev.txt # Development dependencies ├── validate.sh # Code quality validation script ├── run_local.sh # Local execution script +├── input-sample.json # Example payload for the train_model workflow └── README.md # This file ``` @@ -1280,22 +1263,23 @@ sientia-dataops-model-manager/ ### Common Issues -1. **Temporal Connection Failures** - - Verify Temporal server is running and accessible +1. **Temporal connection failures** + - Verify Temporal server is running and reachable at `TEMPORAL_HOST` - Check namespace configuration and permissions - Review server logs for connection issues + - If you see **308 Permanent Redirect** or **invalid compression flag** on connect, the endpoint likely expects **TLS** while `TEMPORAL_USE_TLS` is `false`. Set `TEMPORAL_USE_TLS=true` and point `TEMPORAL_HOST` at the correct TLS gRPC address (host and port depend on your ingress or load balancer) -2. **MLFlow Connection Issues** +2. **MLFlow connection issues** - Verify MLFlow server is running and accessible - Check authentication credentials and permissions - Ensure model names and versions exist -3. **Database Connection Issues** +3. **Database connection issues** - Verify PostgreSQL service is running - Check connection credentials and network access - Ensure proper connection pool configuration -4. **Workflow Execution Failures** +4. **Workflow execution failures** - Review activity error logs and notifications - Check training parameter validation errors - Verify input data format and required fields @@ -1321,9 +1305,7 @@ export LOG_LEVEL=DEBUG ### Scaling Considerations - **Horizontal Scaling**: Deploy multiple worker instances -- **Task Queue Distribution**: Two dedicated task queues for workflow isolation - - `train_model-queue`: Training workflows - - `cleanup-queue`: Cleanup workflows +- **Task queue distribution**: Workers register `train_model--queue` and `cleanup_files--queue` (see `RUNTIME`) - **Database Performance**: Optimize indexes and connection pooling - **MLFlow Performance**: Configure appropriate model serving resources - **Storage Management**: Adjust cleanup retention period based on storage capacity and costs diff --git a/input-sample.md b/input-sample.md new file mode 100644 index 0000000..0ca71bb --- /dev/null +++ b/input-sample.md @@ -0,0 +1,56 @@ +##### Insert a new experiment run + +```sql +-- Optional: remove a previous run with the same id +DELETE FROM public.experiment_run WHERE experiment_run_id = 1001; +``` + +```sql +INSERT INTO public.experiment_run +(experiment_name, run_name, username, status, error_message, created_at, +updated_at, bucket_name, file_name, request_data, orchestrator_response_data) +VALUES( + 'test-experiment-name', + 'test-run-name', + 'test-username', + 'ORCHESTRATOR_WAITING_PROC', + null, + now(), + now(), + 'model-training', + 'training_data.csv', + '{"experiment_run_id":1001,"variable_columns":["feature_a","feature_b"],"target_variable":"target","bucket_name":"model-training","file_name":"training_data.csv","line_separator":",","decimal_separator":".","train_size":80,"shuffle":true,"model_name":"Linear Regression","model_type":"linear_regression","data_model_kwargs":{},"model_kwargs":{},"opt_params":{},"model_metadata":{"schemas":{"components":{"schemas":{}}}}}', + null +); +``` + +##### Upload the input dataset to MinIO + +```bash +mc cp input_dataset.csv suse/model-training/training-sample-dataset-1001.csv +``` + +##### Temporal input payload sample + +Keys match `TrainModelParams.from_dict` in `model_manager/utils/models/train_model_params.py`: every field passed to `_check_none` must be present; `model_metadata` must be non-empty for `validate_business_rules()`. Omit optional keys (`date_column`, `date_format`, `random_state`, `val_file_name`, `model_id`) when defaults or `None` apply. + +```json +{ + "experiment_run_id": 1001, + "variable_columns": ["feature_a", "feature_b"], + "target_variable": "target", + "bucket_name": "model-training", + "file_name": "training-sample-dataset-1001.csv", + "line_separator": ",", + "decimal_separator": ".", + "train_size": 80, + "shuffle": true, + "random_state": 42, + "model_name": "test-runtime-linear-regression-model", + "model_type": "linear_regression", + "model_id": 1001, + "data_model_kwargs": {}, + "model_kwargs": {}, + "opt_params": {} +} +``` diff --git a/input_dataset.csv b/input_dataset.csv new file mode 100644 index 0000000..1d9e1d6 --- /dev/null +++ b/input_dataset.csv @@ -0,0 +1,11 @@ +timestamp,feature_a,feature_b,target +2026-01-01 00:00:00,1.0,2.0,5.1 +2026-01-01 00:01:00,2.0,2.0,5.9 +2026-01-01 00:02:00,2.0,3.0,8.3 +2026-01-01 00:03:00,3.0,3.0,9.2 +2026-01-01 00:04:00,3.0,4.0,10.8 +2026-01-01 00:05:00,4.0,4.0,11.7 +2026-01-01 00:06:00,4.0,5.0,14.1 +2026-01-01 00:07:00,5.0,5.0,15.15 +2026-01-01 00:08:00,5.0,6.0,17.2 +2026-01-01 00:09:00,6.0,6.0,17.9 \ No newline at end of file diff --git a/model_manager/sientia/model_serving.py b/model_manager/sientia/model_serving.py deleted file mode 100644 index 160df07..0000000 --- a/model_manager/sientia/model_serving.py +++ /dev/null @@ -1,212 +0,0 @@ -import logging -import os -from collections.abc import Generator -from contextlib import contextmanager -from typing import Any - -import mlflow -import mlflow.sklearn -import pandas as pd - -from model_manager.sientia.exceptions import SientiaMlException - - -class ModelServing: - """ - MLflow model serving wrapper. - - Thread-safety note: This class modifies global state (MLflow tracking URI and - environment variables) during initialization. In multi-threaded environments, - ensure that: - 1. Instances are created with the same tracking_uri/credentials, OR - 2. Instance creation is synchronized (e.g., using a lock), OR - 3. Create a single instance and share it across threads - - The MLflow operations themselves (log_param, log_metric, etc.) are thread-safe - when operating on different runs. - """ - - def __init__( - self, - tracking_uri: str, - username: str | None = None, - password: str | None = None, - ): - """ - Initialize ModelServing client. - - WARNING: This modifies global state (MLflow config and environment variables). - Not thread-safe during initialization if different credentials are used. - - Args: - tracking_uri: MLflow tracking server URI - username: Optional MLflow username - password: Optional MLflow password - logger: Optional logger (currently unused) - """ - # Set tracking URI (modifies global MLflow state) - mlflow.set_tracking_uri(tracking_uri) - - # Set credentials in environment variables (global state) - if username is not None: - os.environ['MLFLOW_TRACKING_USERNAME'] = username - if password is not None: - os.environ['MLFLOW_TRACKING_PASSWORD'] = password - - # Function to list runs for a given experiment - def search_runs_by_name( - self, experiment_names: list[str], order_by: None | list[str] = None - ) -> pd.DataFrame | list: - """ - List runs for a specified MLflow experiment. - - Args: - experiment_names (list[str]): List with experiment_names to retrieve runs from. - - Returns: - Union[pd.DataFrame, list]: A DataFrame or list containing run information. - - Raises: - SientiaMlException: If unable to search runs. - """ - try: - runs = mlflow.search_runs(experiment_names=experiment_names, order_by=order_by) - except SientiaMlException as e: - logging.error(e) - raise SientiaMlException(str(e)) from e - return runs - - def set_experiment(self, experiment_identifier: str) -> None: - """ - Set the given experiment as the active experiment. - - Args: - experiment_identifier (str): name or id of the experiment to be setted - - Raises: - Exception: If setting the experiment fails. - """ - mlflow.set_experiment(experiment_identifier) - - def log_model(self, sk_model: Any, artifact_path: Any, **kwargs) -> None: - """ - Log a sklearn model. - - Args: - sk_model: scikit-learn model to be saved. - artifact_path: Run-relative artifact path. - - Returns: - None - - Security Warning: - The GitHub token is hardcoded. Consider moving to environment variable - or using a secure secret management solution (e.g., K8s secrets). - - Raises: - Exception: If logging the model fails. - """ - mlflow.sklearn.log_model( - sk_model, - artifact_path, - extra_pip_requirements=[os.getenv('EXTRA_PIP_REQUIREMENTS')], - **kwargs, - ) - - def log_param(self, key: str, value: Any) -> None: - """ - Log a param in the active run. - - Args: - key (str): Param name - value (any): Param value - - Returns: - None - - Raises: - Exception: If logging the parameter fails. - """ - mlflow.log_param(key, value) - - def log_metric(self, key: str, value: Any) -> None: - """ - Log a metric in the active run. - - Args: - key (str): Metric name - value (any): Metric value - - Returns: - None - - Raises: - Exception: If logging the metric fails. - """ - mlflow.log_metric(key, value) - - def log_artifact( - self, local_path: str, artifact_path: str | None = None, run_id: str | None = None - ) -> None: - """ - Log an artifact. - - Args: - local_path: Local path of the artifact to log. - artifact_path: If provided, the directory in artifact_uri to write to. - run_id: optional id of current run - - Returns: - None - - Raises: - Exception: If logging the artifact fails. - """ - mlflow.log_artifact(local_path=local_path, artifact_path=artifact_path, run_id=run_id) - - @contextmanager - def save_experiment( - self, - run_id: str | None = None, - experiment_id: str | None = None, - run_name: str | None = None, - nested: bool = False, - tags: dict[str, Any] | None = None, - description: str | None = None, - log_system_metrics: bool | None = None, - ) -> Generator[mlflow.ActiveRun, None, None]: - """ - Context manager to save an experiment, ensuring the run is properly closed. - - This prevents memory leaks by guaranteeing that MLflow runs are always ended, - even if an exception occurs. Thread-safe when used with proper MLflow configuration. - - Args: - run_id: If specified, get the run with the specified UUID and log parameters and metrics under that run. - experiment_id: ID of the experiment under which to create the current run (applicable only when run_id is not specified). - run_name: Name of new run. Used only when run_id is unspecified. - nested: Controls whether run is nested in parent run. True creates a nested run. - tags: An optional dictionary of string keys and values to set as tags on the run. If a run is being resumed, these tags are set on the resumed run. If a new run is being created, these tags are set on the new run. - description: An optional string that populates the description box of the run. - log_system_metrics: If True, system metrics will be logged. If None, we will check environment variable - - Yields: - ActiveRun: object that acts as a context manager wrapping the run's state. - - Raises: - Exception: If starting or ending the MLflow run fails. - """ - run = mlflow.start_run( - run_id=run_id, - experiment_id=experiment_id, - run_name=run_name, - nested=nested, - tags=tags, - description=description, - log_system_metrics=log_system_metrics, - ) - try: - yield run - finally: - # Ensure run is always ended, preventing resource leaks - mlflow.end_run() diff --git a/model_manager/utils/connectors_config.py b/model_manager/utils/connectors_config.py index f78b417..8b8d994 100644 --- a/model_manager/utils/connectors_config.py +++ b/model_manager/utils/connectors_config.py @@ -42,7 +42,8 @@ def build_mlflow_config() -> dict[str, Any]: It handles server connection and authentication parameters. Environment Variables: - MLFLOW_URL: MLFlow server hostname (default: http://localhost:5080) + MLFLOW_URL: Full MLflow tracking URL including scheme, host, and port + (default: http://localhost:5080) MLFLOW_USERNAME: MLFlow username (default: aignosi) MLFLOW_PASSWORD: MLFlow password (default: aignosi) @@ -68,7 +69,7 @@ def build_mongodb_config() -> dict[str, Any]: MONGODB_USERNAME: MongoDB username (default: root) MONGODB_PASSWORD: MongoDB password (default: wKZDbMNU1c) MONGODB_URL: MongoDB connection URI (default: localhost:27018) - MONGODB_DATABASE_NAME: MongoDB database name (default: sientia) + MONGODB_DATABASE: MongoDB database name (default: sientia) MONGODB_TTL_INDEX_HOURS: TTL index duration in hours (default: 1) Returns: @@ -82,7 +83,7 @@ def build_mongodb_config() -> dict[str, Any]: return { 'connection_string': connection_string, - 'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'), + 'database_name': getenv('MONGODB_DATABASE', 'sientia'), 'ttl_index_seconds': int(getenv('MONGODB_TTL_INDEX_HOURS', '1')) * 3600, 'uri': uri, } @@ -101,7 +102,7 @@ def build_minio_config() -> dict[str, Any]: MINIO_ACCESS_KEY: MinIO access key ID (default: minioadmin) MINIO_SECRET_KEY: MinIO secret access key (default: minioadmin) MINIO_REGION: MinIO region name (default: us-east-1) - MINIO_USE_SSL: Whether to use SSL/TLS (default: false) + MINIO_SECURE: Whether to use SSL/TLS (default: false) MINIO_MAX_RETRY_ATTEMPTS: Maximum number of retry attempts (default: 3) MINIO_RETRY_MODE: Retry mode - standard, legacy, or adaptive (default: adaptive) MINIO_CONNECT_TIMEOUT: Connection timeout in seconds (default: 10) @@ -116,7 +117,7 @@ def build_minio_config() -> dict[str, Any]: 'access_key': getenv('MINIO_ACCESS_KEY', 'minioadmin'), 'secret_key': getenv('MINIO_SECRET_KEY', 'minioadmin'), 'region': getenv('MINIO_REGION', 'us-east-1'), - 'use_ssl': getenv('MINIO_USE_SSL', 'false').lower() == 'true', + 'use_ssl': getenv('MINIO_SECURE', 'false').lower() == 'true', 'max_retry_attempts': int(getenv('MINIO_MAX_RETRY_ATTEMPTS', '3')), 'retry_mode': getenv('MINIO_RETRY_MODE', 'adaptive'), 'connect_timeout': int(getenv('MINIO_CONNECT_TIMEOUT', '10')), diff --git a/model_manager/worker/prepare_worker.py b/model_manager/worker/prepare_worker.py index 2855ce3..2dd4cbf 100644 --- a/model_manager/worker/prepare_worker.py +++ b/model_manager/worker/prepare_worker.py @@ -20,7 +20,7 @@ parameters = [ ('ACTIVITY_POLLER_BEHAVIOUR_MINIMUM', '10'), ('ACTIVITY_POLLER_BEHAVIOUR_INITIAL', '100'), ('ACTIVITY_POLLER_BEHAVIOUR_MAXIMUM', '200'), - ('ACTIVITY_EXECUTOR_MAX_WORKERS', '32'), + ('ACTIVITY_EXECUTOR_MAX_WORKERS', '200'), ] diff --git a/scripts/run_training_test.py b/scripts/run_training_test.py index a76d723..2cb3bb8 100644 --- a/scripts/run_training_test.py +++ b/scripts/run_training_test.py @@ -1,37 +1,31 @@ -#!/usr/bin/env python3 -"""Utility script to trigger the training workflow end-to-end for testing. +# --- +# jupyter: +# jupytext: +# formats: py:percent +# text_representation: +# extension: .py +# format_name: percent +# kernelspec: +# display_name: Python 3 +# language: python +# name: python3 +# --- -Steps performed (default): -1. Upload the CSV test dataset to MinIO using the configured `mc` alias. -2. Insert a new experiment_run record in Postgres and capture the generated ID. -3. Trigger the Temporal `train_model` workflow with the correct payload. - -Alternatives for local diagnosis (--local / --validate-only): -- example: python scripts/run_training_test.py --scenario 01-linear-regression-basic --local --csv docs/test-model-data.csv -- --validate-only: Validates scenario parameters only (no MinIO, Postgres, Temporal). -- --local: Runs the same training pipeline locally (validate + load CSV + train + - after_train_calculation). Use to get full Python tracebacks for debugging. - Does not upload to MinIO, insert DB, or start Temporal. - By default skips MLflow save; use --local-save-mlflow to also test saving to MLflow. - -Prerequisites (default flow): -- `mc` CLI configured with alias defined in MINIO_ALIAS. -- PostgreSQL accessible with credentials in environment variables or defaults. -- Temporal server reachable without TLS on TEMPORAL_HOST / TEMPORAL_NAMESPACE. -- Python dependencies installed (see requirements.txt / requirements-dev.txt). -""" +# %% [markdown] +# # Training smoke test (`input-sample.md`) +# +# Run cells top to bottom in VS Code / Cursor (**Run Cell** on each `# %%` block). +# +# Steps mirror `input-sample.md`: optional DB delete + insert, `mc cp` to MinIO, Temporal `train_model`. +# Set `POSTGRES_*`, `TEMPORAL_*`, `TRAIN_TASK_QUEUE`, and configure the `mc` alias (default `suse`). +# %% from __future__ import annotations -import argparse -import asyncio -import json import os import subprocess -import sys import uuid from datetime import datetime, timedelta -from io import BytesIO from pathlib import Path import psycopg2 @@ -39,27 +33,20 @@ from dotenv import load_dotenv from psycopg2.extras import Json from temporalio import client -# Carrega variáveis de ambiente do arquivo .env na raiz do projeto +# %% +# --- configuration (edit here or use `.env` at repo root) --- PROJECT_ROOT = Path(__file__).resolve().parent.parent -ENV_PATH = PROJECT_ROOT / '.env' -if ENV_PATH.exists(): - load_dotenv(dotenv_path=ENV_PATH) +load_dotenv(PROJECT_ROOT / '.env') +EXPERIMENT_RUN_ID = 1001 +MINIO_MC_ALIAS = os.getenv('MINIO_MC_ALIAS', 'suse') +MINIO_BUCKET = os.getenv('MINIO_DEFAULT_BUCKET', 'model-training') +OBJECT_NAME = f'training-sample-dataset-{EXPERIMENT_RUN_ID}.csv' +LOCAL_CSV = PROJECT_ROOT / 'input_dataset.csv' -DEFAULT_CSV_PATH = Path('docs/test-model-data.csv') -TEST_SCENARIOS_DIR = PROJECT_ROOT / 'docs' / 'test-scenarios' -MINIO_ALIAS = 'suse' -MINIO_BUCKET = 'model-training' - -# Mapeamento de CSV específico por cenário -SCENARIO_CSV_MAPPING = { - '12-angular-test-date-format': Path('docs/DB_CV022_WIT230.csv'), - '13-angular-test-double-date-column': Path('docs/DB_CV022_WIT230 _double_date_column.csv'), -} - -POSTGRES_CONFIG = { +PG = { 'host': os.getenv('POSTGRES_HOST'), - 'port': os.getenv('POSTGRES_PORT'), + 'port': int(os.getenv('POSTGRES_PORT', '5432')), 'user': os.getenv('POSTGRES_USER'), 'password': os.getenv('POSTGRES_PASSWORD'), 'dbname': os.getenv('POSTGRES_DBNAME'), @@ -68,597 +55,108 @@ POSTGRES_CONFIG = { TEMPORAL_HOST = os.getenv('TEMPORAL_HOST') TEMPORAL_NAMESPACE = os.getenv('TEMPORAL_NAMESPACE') TRAIN_TASK_QUEUE = os.getenv('TRAIN_TASK_QUEUE') -TEMPORAL_WORKFLOW = 'train_model' +TEMPORAL_TLS = os.getenv('TEMPORAL_USE_TLS', 'false').lower() in ('1', 'true', 'yes') +# %% +# --- 1) database: delete previous row (same id), then insert `experiment_run` --- +# Primary key column is `id` (see `experiment_tracking` updates). `request_data` matches the SQL sample in `input-sample.md`. -def list_available_scenarios() -> list[str]: - """List all available test scenario files.""" - if not TEST_SCENARIOS_DIR.exists(): - return [] - return sorted([f.stem for f in TEST_SCENARIOS_DIR.glob('*.json')]) +now = datetime.utcnow() +request_data = { + 'experiment_run_id': EXPERIMENT_RUN_ID, + 'variable_columns': ['feature_a', 'feature_b'], + 'target_variable': 'target', + 'bucket_name': MINIO_BUCKET, + 'file_name': OBJECT_NAME, + 'line_separator': ',', + 'decimal_separator': '.', + 'train_size': 80, + 'shuffle': True, + 'random_state': 42, + 'model_name': 'test-runtime-linear-regression-model', + 'model_type': 'linear_regression', + 'model_id': EXPERIMENT_RUN_ID, + 'data_model_kwargs': {}, + 'model_kwargs': {}, + 'opt_params': {}, + 'model_metadata': {'schemas': {'components': {'schemas': {}}}}, +} - -def load_scenario(scenario_name: str) -> dict: - """Load a test scenario from JSON file. - - Args: - scenario_name: Name of the scenario (without .json extension) - or full path to a JSON file. - - Returns: - Dictionary with scenario data. - - Raises: - FileNotFoundError: If scenario file doesn't exist. - """ - # Check if it's a full path - scenario_path = Path(scenario_name) - if scenario_path.suffix == '.json' and scenario_path.exists(): - with open(scenario_path) as f: - return json.load(f) - - # Otherwise, look in the test-scenarios directory - scenario_file = TEST_SCENARIOS_DIR / f'{scenario_name}.json' - if not scenario_file.exists(): - available = list_available_scenarios() - available_str = ', '.join(available) if available else 'none' - raise FileNotFoundError( - f"Scenario '{scenario_name}' not found at {scenario_file}.\n" - f'Available scenarios: {available_str}' - ) - - with open(scenario_file) as f: - return json.load(f) - - -def _resolve_csv_path(csv_path: Path) -> Path: - """Resolve CSV path; if not found in project root, try docs/.""" - if csv_path.is_absolute(): - return csv_path - resolved = PROJECT_ROOT / csv_path - if resolved.exists(): - return resolved - docs_path = PROJECT_ROOT / 'docs' / csv_path.name - if docs_path.exists(): - return docs_path - return resolved - - -def _ensure_source_file(path: Path) -> None: - if not path.exists(): - raise FileNotFoundError(f'Test dataset not found at {path.resolve()}') - - -def upload_to_minio(source_path: Path) -> str: - """Upload the CSV to MinIO using the mc CLI and return the object name.""" - _ensure_source_file(source_path) - timestamp = datetime.utcnow().strftime('%Y%m%d-%H%M%S') - object_name = f'test-model-data-{timestamp}.csv' - target_uri = f'{MINIO_ALIAS}/{MINIO_BUCKET}/{object_name}' - - subprocess.run( # noqa: S603 - ['mc', 'cp', str(source_path), target_uri], # noqa: S607 - check=True, - ) - return object_name - - -def insert_experiment_run(file_name: str, request_data: dict) -> int: - """Insert experiment_run record and return the generated ID.""" - now = datetime.utcnow() - payload = { - **request_data, - 'fileName': file_name, - 'bucketName': MINIO_BUCKET, - } - - insert_sql = """ - INSERT INTO experiment_run ( - experiment_name, - username, - status, - created_at, - updated_at, - bucket_name, - file_name, - request_data - ) - VALUES (%s, %s, %s, %s, %s, %s, %s, %s) - RETURNING id; - """ - - with psycopg2.connect(**POSTGRES_CONFIG) as conn: - with conn.cursor() as cur: - cur.execute( - insert_sql, - ( - request_data['experimentName'], - request_data['username'], - 'ORCHESTRATOR_WAITING_PROC', - now, - now, - MINIO_BUCKET, - file_name, - Json(payload), - ), +with psycopg2.connect(**PG) as conn: + with conn.cursor() as cur: + cur.execute('DELETE FROM public.experiment_run WHERE id = %s', (EXPERIMENT_RUN_ID,)) + cur.execute( + """ + INSERT INTO public.experiment_run ( + id, experiment_name, run_name, username, status, error_message, + created_at, updated_at, bucket_name, file_name, request_data, orchestrator_response_data ) - experiment_run_id = cur.fetchone()[0] - - return experiment_run_id - - -def build_workflow_payload( - experiment_run_id: int, - file_name: str, - request_data: dict, -) -> dict: - """Convert camelCase request data to snake_case and enrich with runtime values.""" - return { - 'experiment_run_id': experiment_run_id, - 'experiment_name': request_data['experimentName'], - 'username': request_data['username'], - 'target_variable': request_data['targetVariable'], - 'variable_columns': request_data['variableColumns'], - 'lag_train': request_data['lagTrain'], - 'lag_val': request_data['lagVal'], - 'rem_static_win': request_data['remStaticWin'], - 'low_lim': request_data['lowLim'], - 'upp_lim': request_data['uppLim'], - 'window': request_data['window'], - 'use_scaler': request_data['useScaler'], - 'include_ar': request_data['includeAr'], - 'train_size': request_data['trainSize'], - 'shuffle': request_data['shuffle'], - 'bucket_name': MINIO_BUCKET, - 'file_name': file_name, - 'line_separator': request_data['lineSeparator'], - 'decimal_separator': request_data['decimalSeparator'], - 'date_column': request_data.get('dateColumn'), - 'date_format': request_data.get('dateFormat'), - 'removed_intervals': request_data['removedIntervals'], - # New parameters - 'model_name': request_data.get('modelName', 'Linear Regression'), - 'degree': request_data.get('degree', 1), - 'interaction_only': request_data.get('interactionOnly', False), - 'nan_treatment': request_data.get('nanTreatment', 'drop'), - 'start_date': request_data.get('startDate'), - 'end_date': request_data.get('endDate'), - 'scaler_name': request_data.get('scalerName', 'None'), - 'support_filters': request_data.get('supportFilters', {}), - 'static_threshold': request_data.get('staticThreshold'), - } - - -async def trigger_temporal_workflow(workflow_input: dict) -> str: - """Connect to Temporal and trigger the training workflow.""" - temporal_client = await client.Client.connect( - target_host=TEMPORAL_HOST, - namespace=TEMPORAL_NAMESPACE, - tls=os.getenv('TEMPORAL_USE_TLS', False), - ) - - workflow_id = f'train-model-test-{uuid.uuid4()}' - await temporal_client.execute_workflow( - TEMPORAL_WORKFLOW, - workflow_input, - id=workflow_id, - task_queue=TRAIN_TASK_QUEUE, - execution_timeout=timedelta(minutes=5), - run_timeout=timedelta(minutes=5), - task_timeout=timedelta(minutes=5), - ) - return workflow_id - - -def _build_local_payload(request_data: dict, csv_path: Path) -> dict: - """Build workflow payload for local run (no real experiment_run_id).""" - return build_workflow_payload( - experiment_run_id=0, - file_name=csv_path.name, - request_data=request_data, - ) - - -def run_validate_only(scenario_name: str) -> dict: - """Validate scenario parameters only. No MinIO, Postgres, or Temporal. - - Returns: - dict: {'success': bool, 'error': str | None, 'scenario': str} - """ - from model_manager.utils.models.train_model_params import TrainModelParams - - result = {'scenario': scenario_name, 'success': False, 'error': None} - try: - request_data = load_scenario(scenario_name) - except FileNotFoundError as exc: - result['error'] = str(exc) - return result - - payload = _build_local_payload(request_data, Path('local.csv')) - try: - train_params = TrainModelParams.from_dict(payload) - train_params.validate_business_rules() - result['success'] = True - except (ValueError, TypeError, KeyError) as e: - result['error'] = str(e) - return result - - -def run_local_pipeline( - scenario_name: str, - csv_path: Path, - save_mlflow: bool = False, -) -> dict: - """Run the same training pipeline locally (validate + train + metrics). - - Reads CSV from disk, runs DataManagerRepository.prepare_training_data and - compute_regression_metrics. Optionally saves to MLflow if save_mlflow is - True (requires MLflow env). - - Returns: - dict: {'success': bool, 'error': str | None, 'scenario': str, ...} - """ - from model_manager.utils.logger_helper import get_logger - from model_manager.utils.models.train_model_params import TrainModelParams - from model_manager.utils.repository.data_manager_repository import DataManagerRepository - - result = { - 'scenario': scenario_name, - 'success': False, - 'error': None, - } - try: - request_data = load_scenario(scenario_name) - except FileNotFoundError as exc: - result['error'] = str(exc) - return result - - # Use scenario-specific CSV if mapped, otherwise use provided csv_path - if scenario_name in SCENARIO_CSV_MAPPING: - csv_path = SCENARIO_CSV_MAPPING[scenario_name] - - csv_path = _resolve_csv_path(csv_path) - _ensure_source_file(csv_path) - payload = _build_local_payload(request_data, csv_path) - try: - train_params = TrainModelParams.from_dict(payload) - train_params.validate_business_rules() - except (ValueError, TypeError, KeyError) as e: - result['error'] = f'Validation failed: {e}' - return result - - logger = get_logger(__name__) - data_manager_repository = DataManagerRepository(logger) - - with open(csv_path, 'rb') as f: - file_content = BytesIO(f.read()) - - try: - train_result = data_manager_repository.prepare_training_data( - train_file_bytes=file_content.getvalue(), - validation_file_bytes=None, - params=train_params, - metadata={'source': 'run_local_pipeline', 'scenario': scenario_name}, + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + """, + ( + EXPERIMENT_RUN_ID, + 'test-experiment-name', + 'test-run-name', + 'test-username', + 'ORCHESTRATOR_WAITING_PROC', + None, + now, + now, + MINIO_BUCKET, + OBJECT_NAME, + Json(request_data), + None, + ), ) - train_result = data_manager_repository.compute_regression_metrics( - train_result, - ) - except Exception as e: - result['error'] = str(e) - raise # re-raise so caller gets full traceback for diagnosis - if save_mlflow: - from model_manager.utils.connectors_config import build_mlflow_config - from model_manager.utils.repository.model_repository import ModelRepository +# %% +# --- 2) MinIO: upload local CSV (requires `mc` CLI and alias configured) --- +subprocess.run( + ['mc', 'cp', str(LOCAL_CSV), f'{MINIO_MC_ALIAS}/{MINIO_BUCKET}/{OBJECT_NAME}'], + check=True, +) - mlflow_config = build_mlflow_config() - model_repository = ModelRepository( - url=mlflow_config['url'], - username=mlflow_config['username'], - password=mlflow_config['password'], - logger=logger, - ) - train_result = model_repository.save_model(train_result) +# %% +# --- 3) Temporal: start `train_model` (flat payload; worker fills `model_metadata` in `load_model_metadata`) --- - result['success'] = True - result['run_name'] = getattr(train_result, 'run_name', None) - result['run_dir'] = getattr(train_result, 'run_dir', None) - return result +if not TEMPORAL_HOST or not TEMPORAL_NAMESPACE or not TRAIN_TASK_QUEUE: + raise RuntimeError('Set TEMPORAL_HOST, TEMPORAL_NAMESPACE, and TRAIN_TASK_QUEUE') +TH, TN, TQ = TEMPORAL_HOST, TEMPORAL_NAMESPACE, TRAIN_TASK_QUEUE +_workflow_input = { + 'experiment_run_id': EXPERIMENT_RUN_ID, + 'variable_columns': ['feature_a', 'feature_b'], + 'target_variable': 'target', + 'bucket_name': MINIO_BUCKET, + 'file_name': OBJECT_NAME, + 'line_separator': ',', + 'decimal_separator': '.', + 'train_size': 80, + 'shuffle': True, + 'random_state': 42, + 'model_name': 'test-runtime-linear-regression-model', + 'model_type': 'linear_regression', + 'model_id': EXPERIMENT_RUN_ID, + 'data_model_kwargs': {}, + 'model_kwargs': {}, + 'opt_params': {}, +} -def parse_args() -> argparse.Namespace: - """Parse command line arguments.""" - parser = argparse.ArgumentParser( - description='Run training workflow tests with different scenarios.', - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=""" -Examples: - # List available scenarios - python scripts/run_training_test.py --list +# %% - # Run a specific scenario - python scripts/run_training_test.py --scenario linear-regression-basic - - # Run with a custom JSON file - python scripts/run_training_test.py --scenario /path/to/custom-scenario.json - - # Run with a custom CSV data file - python scripts/run_training_test.py --scenario linear-regression-basic --csv docs/other-data.csv - - # Run all scenarios - python scripts/run_training_test.py --all - - # Validate scenario parameters only (no external services) - python scripts/run_training_test.py --scenario linear-regression-basic --validate-only - - # Run training pipeline locally to diagnose errors (full traceback) - python scripts/run_training_test.py --scenario linear-regression-basic --local --csv docs/test-model-data.csv - - # Local run and save to MLflow (requires MLflow env) - python scripts/run_training_test.py --scenario linear-regression-basic --local --local-save-mlflow -""", - ) - parser.add_argument( - '--scenario', - '-s', - type=str, - help='Name of the test scenario (without .json) or path to a JSON file.', - ) - parser.add_argument( - '--csv', - '-c', - type=Path, - default=DEFAULT_CSV_PATH, - help=f'Path to the CSV data file (default: {DEFAULT_CSV_PATH}).', - ) - parser.add_argument( - '--list', - '-l', - action='store_true', - help='List all available test scenarios and exit.', - ) - parser.add_argument( - '--all', - '-a', - action='store_true', - help='Run all available test scenarios sequentially.', - ) - parser.add_argument( - '--validate-only', - action='store_true', - help='Only validate scenario parameters (no MinIO, Postgres, Temporal).', - ) - parser.add_argument( - '--local', - action='store_true', - help='Run training pipeline locally (validate + train from CSV) to get full tracebacks.', - ) - parser.add_argument( - '--local-save-mlflow', - action='store_true', - help='With --local, also save the model to MLflow (requires MLflow env).', - ) - return parser.parse_args() - - -def run_single_scenario(scenario_name: str, csv_path: Path) -> dict: - """Run a single test scenario and return the result. - - Args: - scenario_name: Name of the scenario to run. - csv_path: Path to the CSV data file. - - Returns: - Dictionary with scenario result including success status and details. - """ - result = { - 'scenario': scenario_name, - 'success': False, - 'error': None, - 'experiment_run_id': None, - 's3_object_name': None, - 'workflow_id': None, - } - - # Load scenario - try: - experiment_request = load_scenario(scenario_name) - print(f' Loaded scenario: {scenario_name}') - except FileNotFoundError as exc: - result['error'] = str(exc) - return result - - # Use scenario-specific CSV if mapped, otherwise use provided csv_path - if scenario_name in SCENARIO_CSV_MAPPING: - csv_path = SCENARIO_CSV_MAPPING[scenario_name] - print(f' Using scenario-specific CSV: {csv_path}') - - csv_path = _resolve_csv_path(csv_path) - - # Upload CSV to MinIO - try: - uploaded_file_name = upload_to_minio(csv_path) - result['s3_object_name'] = uploaded_file_name - print(f' Uploaded CSV to MinIO: {uploaded_file_name}') - except subprocess.CalledProcessError as exc: - result['error'] = f'Failed to upload file to MinIO: {exc}' - return result - except FileNotFoundError as exc: - result['error'] = str(exc) - return result - - # Insert experiment run - try: - experiment_run_id = insert_experiment_run(uploaded_file_name, experiment_request) - result['experiment_run_id'] = experiment_run_id - print(f' Created experiment_run with ID: {experiment_run_id}') - except psycopg2.Error as exc: - result['error'] = f'Database error while inserting experiment_run: {exc}' - return result - - # Build and trigger workflow - workflow_payload = build_workflow_payload( - experiment_run_id=experiment_run_id, - file_name=uploaded_file_name, - request_data=experiment_request, - ) - - try: - workflow_id = asyncio.run(trigger_temporal_workflow(workflow_payload)) - result['workflow_id'] = workflow_id - result['success'] = True - print(f' Workflow started: {workflow_id}') - except Exception as exc: # noqa: BLE001 - result['error'] = f'Failed to start Temporal workflow: {exc}' - return result - - return result - - -def print_summary(results: list[dict]) -> None: - """Print a summary of all scenario results. - - Args: - results: List of result dictionaries from run_single_scenario. - """ - passed = [r for r in results if r['success']] - failed = [r for r in results if not r['success']] - - print('\n' + '=' * 60) - print('SUMMARY') - print('=' * 60) - print(f'Total: {len(results)} | Passed: {len(passed)} | Failed: {len(failed)}') - print('=' * 60) - - if passed: - print('\n✓ PASSED:') - for r in passed: - print(f' - {r["scenario"]}') - - if failed: - print('\n✗ FAILED:') - for r in failed: - print(f' - {r["scenario"]}') - if r['error']: - print(f' Error: {r["error"]}') - - print() - - -def _handle_list_scenarios() -> None: - """Print available scenarios and exit.""" - scenarios = list_available_scenarios() - if scenarios: - print('Available test scenarios:') - for scenario in scenarios: - print(f' - {scenario}') - else: - print(f'No scenarios found in {TEST_SCENARIOS_DIR}') - sys.exit(0) - - -def _handle_run_all(args: argparse.Namespace) -> None: - """Run all scenarios and exit with appropriate code.""" - scenarios = list_available_scenarios() - if not scenarios: - print(f'No scenarios found in {TEST_SCENARIOS_DIR}', file=sys.stderr) - sys.exit(1) - - print(f'Running {len(scenarios)} scenarios...\n') - results = [] - for i, scenario in enumerate(scenarios, 1): - print(f'[{i}/{len(scenarios)}] Running scenario: {scenario}') - result = run_single_scenario(scenario, args.csv) - results.append(result) - status = '✓' if result['success'] else '✗' - print(f'[{i}/{len(scenarios)}] {status} {scenario}\n') - - print_summary(results) - failed_count = sum(1 for r in results if not r['success']) - sys.exit(1 if failed_count > 0 else 0) - - -def _handle_validate_only(args: argparse.Namespace) -> None: - """Validate scenario parameters only and exit.""" - if not args.scenario: - print('Error: --scenario is required with --validate-only.', file=sys.stderr) - sys.exit(1) - result = run_validate_only(args.scenario) - if result['success']: - print(f'Validation OK: {result["scenario"]}') - else: - print(f'Validation failed: {result["error"]}', file=sys.stderr) - sys.exit(1) - sys.exit(0) - - -def _handle_local(args: argparse.Namespace) -> None: - """Run local pipeline and exit.""" - if not args.scenario: - print('Error: --scenario is required with --local.', file=sys.stderr) - sys.exit(1) - print(f'Running local pipeline: {args.scenario} (CSV: {args.csv})') - result = run_local_pipeline( - args.scenario, - args.csv, - save_mlflow=args.local_save_mlflow, - ) - if not result['success']: - print(f'Error: {result["error"]}', file=sys.stderr) - sys.exit(1) - out = {'scenario': result['scenario'], 'success': True} - if result.get('run_name') is not None: - out['run_name'] = result['run_name'] - if result.get('run_dir') is not None: - out['run_dir'] = result['run_dir'] - print(json.dumps(out, indent=2)) - - -def _handle_single_scenario(args: argparse.Namespace) -> None: - """Run one scenario (MinIO + Postgres + Temporal) and print result.""" - print(f'Running scenario: {args.scenario}') - result = run_single_scenario(args.scenario, args.csv) - if not result['success']: - print(f'Error: {result["error"]}', file=sys.stderr) - sys.exit(1) - print( - json.dumps( - { - 'scenario': result['scenario'], - 'experiment_run_id': result['experiment_run_id'], - 's3_object_name': result['s3_object_name'], - 'workflow_id': result['workflow_id'], - }, - indent=2, - ) - ) - - -def main() -> None: - args = parse_args() - - if args.list: - _handle_list_scenarios() - - if args.all: - _handle_run_all(args) - - if args.validate_only: - _handle_validate_only(args) - - if args.local: - _handle_local(args) - return - - if not args.scenario: - print( - 'Error: --scenario or --all is required. Use --list to see available scenarios.', - file=sys.stderr, - ) - sys.exit(1) - - _handle_single_scenario(args) - - -if __name__ == '__main__': - main() +c = await client.Client.connect( + target_host=TH, + namespace=TN, + tls=TEMPORAL_TLS, +) +wid = f'train-model-test-{uuid.uuid4()}' +await c.execute_workflow( # type: ignore[call-overload] + 'train_model', + _workflow_input, + id=wid, + task_queue=TQ, + execution_timeout=timedelta(minutes=5), + run_timeout=timedelta(minutes=5), + task_timeout=timedelta(minutes=5), +) +print(wid) diff --git a/tests/sientia/test_model_serving.py b/tests/sientia/test_model_serving.py deleted file mode 100644 index adfab27..0000000 --- a/tests/sientia/test_model_serving.py +++ /dev/null @@ -1,311 +0,0 @@ -"""Unit tests for ModelServing class.""" - -from unittest.mock import MagicMock, patch - -import pandas as pd -from pytest import raises - -from model_manager.sientia.exceptions import SientiaMlException -from model_manager.sientia.model_serving import ModelServing - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch.dict('os.environ', {}, clear=True) -def test_init_with_all_credentials(mock_set_tracking_uri): - """Test initialization with tracking URI, username, and password.""" - tracking_uri = 'http://mlflow.example.com' - username = 'test_user' - password = 'test_pass' - - ModelServing(tracking_uri=tracking_uri, username=username, password=password) - - mock_set_tracking_uri.assert_called_once_with(tracking_uri) - import os - - assert os.environ['MLFLOW_TRACKING_USERNAME'] == username - assert os.environ['MLFLOW_TRACKING_PASSWORD'] == password - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch.dict('os.environ', {}, clear=True) -def test_init_without_credentials(mock_set_tracking_uri): - """Test initialization without username and password.""" - tracking_uri = 'http://mlflow.example.com' - - ModelServing(tracking_uri=tracking_uri) - - mock_set_tracking_uri.assert_called_once_with(tracking_uri) - import os - - assert 'MLFLOW_TRACKING_USERNAME' not in os.environ - assert 'MLFLOW_TRACKING_PASSWORD' not in os.environ - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch.dict('os.environ', {}, clear=True) -def test_init_with_only_username(mock_set_tracking_uri): - """Test initialization with only username (no password).""" - tracking_uri = 'http://mlflow.example.com' - username = 'test_user' - - ModelServing(tracking_uri=tracking_uri, username=username) - - mock_set_tracking_uri.assert_called_once_with(tracking_uri) - import os - - assert os.environ['MLFLOW_TRACKING_USERNAME'] == username - assert 'MLFLOW_TRACKING_PASSWORD' not in os.environ - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch.dict('os.environ', {}, clear=True) -def test_init_with_only_password(mock_set_tracking_uri): - """Test initialization with only password (no username).""" - tracking_uri = 'http://mlflow.example.com' - password = 'test_pass' - - ModelServing(tracking_uri=tracking_uri, password=password) - - mock_set_tracking_uri.assert_called_once_with(tracking_uri) - import os - - assert 'MLFLOW_TRACKING_USERNAME' not in os.environ - assert os.environ['MLFLOW_TRACKING_PASSWORD'] == password - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch('model_manager.sientia.model_serving.mlflow.search_runs') -def test_search_runs_by_name_success(mock_search_runs, mock_set_tracking_uri): - """Test successful search_runs_by_name.""" - model_serving = ModelServing(tracking_uri='http://mlflow.example.com') - - expected_df = pd.DataFrame({'run_id': ['123', '456'], 'status': ['FINISHED', 'RUNNING']}) - mock_search_runs.return_value = expected_df - - experiment_names = ['experiment1', 'experiment2'] - result = model_serving.search_runs_by_name(experiment_names) - - mock_search_runs.assert_called_once_with(experiment_names=experiment_names, order_by=None) - pd.testing.assert_frame_equal(result, expected_df) - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch('model_manager.sientia.model_serving.mlflow.search_runs') -def test_search_runs_by_name_with_order_by(mock_search_runs, mock_set_tracking_uri): - """Test search_runs_by_name with order_by parameter.""" - model_serving = ModelServing(tracking_uri='http://mlflow.example.com') - - expected_df = pd.DataFrame({'run_id': ['123'], 'status': ['FINISHED']}) - mock_search_runs.return_value = expected_df - - experiment_names = ['experiment1'] - order_by = ['start_time DESC'] - result = model_serving.search_runs_by_name(experiment_names, order_by=order_by) - - mock_search_runs.assert_called_once_with(experiment_names=experiment_names, order_by=order_by) - pd.testing.assert_frame_equal(result, expected_df) - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch('model_manager.sientia.model_serving.mlflow.search_runs') -@patch('model_manager.sientia.model_serving.logging.error') -def test_search_runs_by_name_raises_exception( - mock_logging_error, mock_search_runs, mock_set_tracking_uri -): - """Test search_runs_by_name properly propagates SientiaMlException.""" - model_serving = ModelServing(tracking_uri='http://mlflow.example.com') - - exception = SientiaMlException(message='Search failed') - mock_search_runs.side_effect = exception - - with raises(SientiaMlException, match='Search failed'): - model_serving.search_runs_by_name(['experiment1']) - - mock_logging_error.assert_called_once_with(exception) - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch('model_manager.sientia.model_serving.mlflow.set_experiment') -def test_set_experiment(mock_set_experiment, mock_set_tracking_uri): - """Test set_experiment method.""" - model_serving = ModelServing(tracking_uri='http://mlflow.example.com') - - experiment_identifier = 'my_experiment' - model_serving.set_experiment(experiment_identifier) - - mock_set_experiment.assert_called_once_with(experiment_identifier) - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch('model_manager.sientia.model_serving.mlflow.sklearn.log_model') -def test_log_model(mock_log_model, mock_set_tracking_uri): - """Test log_model method.""" - model_serving = ModelServing(tracking_uri='http://mlflow.example.com') - - sk_model = MagicMock() - artifact_path = 'model' - model_serving.log_model(sk_model, artifact_path) - - mock_log_model.assert_called_once() - call_args = mock_log_model.call_args - assert call_args[0][0] == sk_model - assert call_args[0][1] == artifact_path - assert 'extra_pip_requirements' in call_args[1] - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch('model_manager.sientia.model_serving.mlflow.sklearn.log_model') -def test_log_model_with_kwargs(mock_log_model, mock_set_tracking_uri): - """Test log_model method with additional kwargs.""" - model_serving = ModelServing(tracking_uri='http://mlflow.example.com') - - sk_model = MagicMock() - artifact_path = 'model' - registered_model_name = 'my_model' - model_serving.log_model(sk_model, artifact_path, registered_model_name=registered_model_name) - - mock_log_model.assert_called_once() - call_args = mock_log_model.call_args - assert call_args[1]['registered_model_name'] == registered_model_name - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch('model_manager.sientia.model_serving.mlflow.log_param') -def test_log_param(mock_log_param, mock_set_tracking_uri): - """Test log_param method.""" - model_serving = ModelServing(tracking_uri='http://mlflow.example.com') - - key = 'learning_rate' - value = 0.01 - model_serving.log_param(key, value) - - mock_log_param.assert_called_once_with(key, value) - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch('model_manager.sientia.model_serving.mlflow.log_metric') -def test_log_metric(mock_log_metric, mock_set_tracking_uri): - """Test log_metric method.""" - model_serving = ModelServing(tracking_uri='http://mlflow.example.com') - - key = 'accuracy' - value = 0.95 - model_serving.log_metric(key, value) - - mock_log_metric.assert_called_once_with(key, value) - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch('model_manager.sientia.model_serving.mlflow.log_artifact') -def test_log_artifact_with_all_params(mock_log_artifact, mock_set_tracking_uri): - """Test log_artifact method with all parameters.""" - model_serving = ModelServing(tracking_uri='http://mlflow.example.com') - - local_path = '/path/to/artifact.txt' - artifact_path = 'artifacts' - run_id = 'run_123' - model_serving.log_artifact(local_path, artifact_path, run_id) - - mock_log_artifact.assert_called_once_with( - local_path=local_path, artifact_path=artifact_path, run_id=run_id - ) - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch('model_manager.sientia.model_serving.mlflow.log_artifact') -def test_log_artifact_with_minimal_params(mock_log_artifact, mock_set_tracking_uri): - """Test log_artifact method with only required parameter.""" - model_serving = ModelServing(tracking_uri='http://mlflow.example.com') - - local_path = '/path/to/artifact.txt' - model_serving.log_artifact(local_path) - - mock_log_artifact.assert_called_once_with( - local_path=local_path, artifact_path=None, run_id=None - ) - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch('model_manager.sientia.model_serving.mlflow.start_run') -@patch('model_manager.sientia.model_serving.mlflow.end_run') -def test_save_experiment_context_manager(mock_end_run, mock_start_run, mock_set_tracking_uri): - """Test save_experiment context manager.""" - model_serving = ModelServing(tracking_uri='http://mlflow.example.com') - - mock_run = MagicMock() - mock_start_run.return_value = mock_run - - with model_serving.save_experiment(run_name='test_run') as run: - assert run == mock_run - - mock_start_run.assert_called_once_with( - run_id=None, - experiment_id=None, - run_name='test_run', - nested=False, - tags=None, - description=None, - log_system_metrics=None, - ) - mock_end_run.assert_called_once() - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch('model_manager.sientia.model_serving.mlflow.start_run') -@patch('model_manager.sientia.model_serving.mlflow.end_run') -def test_save_experiment_with_all_params(mock_end_run, mock_start_run, mock_set_tracking_uri): - """Test save_experiment with all parameters.""" - model_serving = ModelServing(tracking_uri='http://mlflow.example.com') - - mock_run = MagicMock() - mock_start_run.return_value = mock_run - - run_id = 'run_123' - experiment_id = 'exp_456' - run_name = 'test_run' - nested = True - tags = {'key': 'value'} - description = 'Test description' - log_system_metrics = True - - with model_serving.save_experiment( - run_id=run_id, - experiment_id=experiment_id, - run_name=run_name, - nested=nested, - tags=tags, - description=description, - log_system_metrics=log_system_metrics, - ) as run: - assert run == mock_run - - mock_start_run.assert_called_once_with( - run_id=run_id, - experiment_id=experiment_id, - run_name=run_name, - nested=nested, - tags=tags, - description=description, - log_system_metrics=log_system_metrics, - ) - mock_end_run.assert_called_once() - - -@patch('model_manager.sientia.model_serving.mlflow.set_tracking_uri') -@patch('model_manager.sientia.model_serving.mlflow.start_run') -@patch('model_manager.sientia.model_serving.mlflow.end_run') -def test_save_experiment_ensures_end_run_on_exception( - mock_end_run, mock_start_run, mock_set_tracking_uri -): - """Test save_experiment ensures end_run is called even when exception occurs.""" - model_serving = ModelServing(tracking_uri='http://mlflow.example.com') - - mock_run = MagicMock() - mock_start_run.return_value = mock_run - - with raises(ValueError): - with model_serving.save_experiment(run_name='test_run'): - raise ValueError('Test exception') - - mock_start_run.assert_called_once() - mock_end_run.assert_called_once() diff --git a/tests/utils/test_connectors_config.py b/tests/utils/test_connectors_config.py index 84a2b6a..6411fb3 100644 --- a/tests/utils/test_connectors_config.py +++ b/tests/utils/test_connectors_config.py @@ -11,38 +11,31 @@ from model_manager.utils.connectors_config import ( def test_build_mlflow_config_with_env_vars(): - # Arrange + environ.pop('MLFLOW_URL', None) environ['MLFLOW_URL'] = 'http://test-host:8080' environ['MLFLOW_USERNAME'] = 'test-user' environ['MLFLOW_PASSWORD'] = 'test-pass' - # Act config = build_mlflow_config() - # Assert assert config['url'] == 'http://test-host:8080' assert config['username'] == 'test-user' assert config['password'] == 'test-pass' def test_build_mlflow_config_with_defaults(): - # Arrange - # Clear any existing env vars environ.pop('MLFLOW_URL', None) environ.pop('MLFLOW_USERNAME', None) environ.pop('MLFLOW_PASSWORD', None) - # Act config = build_mlflow_config() - # Assert assert config['url'] == 'http://localhost:5080' assert config['username'] == 'aignosi' assert config['password'] == 'aignosi' def test_build_postgres_config_with_env_vars(): - # Arrange environ['POSTGRES_HOST'] = 'test-host' environ['POSTGRES_PORT'] = '5433' environ['POSTGRES_USER'] = 'test-user' @@ -51,10 +44,8 @@ def test_build_postgres_config_with_env_vars(): environ['POSTGRES_MIN_CONNECTIONS'] = '10' environ['POSTGRES_MAX_CONNECTIONS'] = '30' - # Act config = build_postgres_config() - # Assert assert config['host'] == 'test-host' assert config['port'] == 5433 assert config['user'] == 'test-user' @@ -65,7 +56,6 @@ def test_build_postgres_config_with_env_vars(): def test_build_postgres_config_with_defaults(): - # Arrange environ.pop('POSTGRES_HOST', None) environ.pop('POSTGRES_PORT', None) environ.pop('POSTGRES_USER', None) @@ -74,10 +64,8 @@ def test_build_postgres_config_with_defaults(): environ.pop('POSTGRES_MIN_CONNECTIONS', None) environ.pop('POSTGRES_MAX_CONNECTIONS', None) - # Act config = build_postgres_config() - # Assert assert config['host'] == 'localhost' assert config['port'] == 5432 assert config['user'] == 'sientia' @@ -91,7 +79,7 @@ def test_build_mongo_db_config_with_env_vars(): environ['MONGODB_USERNAME'] = 'sientia1' environ['MONGODB_PASSWORD'] = 'sientia1' environ['MONGODB_URL'] = 'localhost:27018' - environ['MONGODB_DATABASE_NAME'] = 'test_db' + environ['MONGODB_DATABASE'] = 'test_db' environ['MONGODB_TTL_INDEX_HOURS'] = '1' assert build_mongodb_config() == { @@ -105,7 +93,7 @@ def test_build_mongo_db_config_with_env_vars(): def test_build_mongo_db_config_with_defaults(): environ.pop('MONGODB_USERNAME', None) environ.pop('MONGODB_PASSWORD', None) - environ.pop('MONGODB_DATABASE_NAME', None) + environ.pop('MONGODB_DATABASE', None) environ.pop('MONGODB_URL', None) environ.pop('MONGODB_TTL_INDEX_HOURS', None) assert build_mongodb_config() == { @@ -117,22 +105,19 @@ def test_build_mongo_db_config_with_defaults(): def test_build_minio_config_with_env_vars(): - # Arrange environ['MINIO_ENDPOINT_URL'] = 'http://test-minio:9000' environ['MINIO_ACCESS_KEY'] = 'test-access-key' environ['MINIO_SECRET_KEY'] = 'test-secret-key' environ['MINIO_REGION'] = 'eu-west-1' - environ['MINIO_USE_SSL'] = 'true' + environ['MINIO_SECURE'] = 'true' environ['MINIO_MAX_RETRY_ATTEMPTS'] = '5' environ['MINIO_RETRY_MODE'] = 'standard' environ['MINIO_CONNECT_TIMEOUT'] = '20' environ['MINIO_READ_TIMEOUT'] = '120' environ['MINIO_DEFAULT_BUCKET'] = 'my-bucket' - # Act config = build_minio_config() - # Assert assert config['endpoint_url'] == 'http://test-minio:9000' assert config['access_key'] == 'test-access-key' assert config['secret_key'] == 'test-secret-key' @@ -153,23 +138,19 @@ def test_build_plugin_store_config_cache_ttl_seconds(): def test_build_minio_config_with_defaults(): - # Arrange - # Clear any existing env vars environ.pop('MINIO_ENDPOINT_URL', None) environ.pop('MINIO_ACCESS_KEY', None) environ.pop('MINIO_SECRET_KEY', None) environ.pop('MINIO_REGION', None) - environ.pop('MINIO_USE_SSL', None) + environ.pop('MINIO_SECURE', None) environ.pop('MINIO_MAX_RETRY_ATTEMPTS', None) environ.pop('MINIO_RETRY_MODE', None) environ.pop('MINIO_CONNECT_TIMEOUT', None) environ.pop('MINIO_READ_TIMEOUT', None) environ.pop('MINIO_DEFAULT_BUCKET', None) - # Act config = build_minio_config() - # Assert assert config['endpoint_url'] == 'http://localhost:9000' assert config['access_key'] == 'minioadmin' assert config['secret_key'] == 'minioadmin' @@ -179,4 +160,4 @@ def test_build_minio_config_with_defaults(): assert config['retry_mode'] == 'adaptive' assert config['connect_timeout'] == 10 assert config['read_timeout'] == 60 - assert config['default_bucket'] == 'streamlit-connectors' + assert config['default_bucket'] == 'model-training' diff --git a/values.yaml b/values.yaml index cdb2bb3..74f1d54 100644 --- a/values.yaml +++ b/values.yaml @@ -150,7 +150,7 @@ global: value: "modelTrainingUser123" - name: MINIO_REGION value: "us-east-1" - - name: MINIO_USE_SSL + - name: MINIO_SECURE value: "false" - name: MINIO_MAX_RETRY_ATTEMPTS value: "3"