From 53f49d7a974d31c80f467dc91e18d88dcfcc4e4f Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Wed, 19 Nov 2025 18:38:00 -0300 Subject: [PATCH 01/16] SIENTIAPDE-1350: Implement file cleanup workflow and activities, including MinIO and local directory cleanup, configuration, and metrics. --- .env.example | 83 +++-- model_manager/activities/activities.py | 11 +- model_manager/activities/cleanup.py | 338 ++++++++++++++++++ .../activities/experiment_tracking.py | 3 +- model_manager/activities/training.py | 2 +- .../utils/repository/storage_repository.py | 33 ++ model_manager/worker/worker.py | 25 +- model_manager/workflows/cleanup_files.py | 83 +++++ scripts/run_cleanup_test.py | 76 ++++ tests/worker/test_worker.py | 40 ++- todo-list.txt | 8 +- 11 files changed, 641 insertions(+), 61 deletions(-) create mode 100644 model_manager/activities/cleanup.py create mode 100644 model_manager/workflows/cleanup_files.py create mode 100644 scripts/run_cleanup_test.py diff --git a/.env.example b/.env.example index f2740cb..b106f15 100644 --- a/.env.example +++ b/.env.example @@ -1,44 +1,57 @@ -POSTGRES_HOST="paradedb-rw.paradedb.svc.cluster.local" -POSTGRES_PORT="5432" -POSTGRES_USER="sientia" -POSTGRES_PASSWORD="password" -POSTGRES_DBNAME="sientia" -POSTGRES_MIN_CONNECTIONS="10" -POSTGRES_MAX_CONNECTIONS="30" +POSTGRES_HOST=paradedb-rw.paradedb.svc.cluster.local +POSTGRES_PORT=5432 +POSTGRES_USER=sientia +POSTGRES_PASSWORD=password +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_USERNAME="aignosi" -MLFLOW_PASSWORD="mlflow_password" +MLFLOW_URL=http://sientia-tracker-mlflow-tracking.sientia-tracker.svc.cluster.local:80 +MLFLOW_USERNAME=aignosi +MLFLOW_PASSWORD=mlflow_password -LOG_LEVEL="DEBUG" -HTTP_METRICS_PORT="9090" -HTTP_SDK_METRICS_PORT="9091" -PROJECT_NAME="sientia-model-manager" +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_NAMESPACE="model-manager" +TEMPORAL_HOST=temporal-frontend.temporal.svc.cluster.local:7233 +TEMPORAL_NAMESPACE=model-manager -MONGODB_USERNAME="mongo_user" -MONGODB_PASSWORD="mongo_db_password" -MONGODB_URL="my-release-mongodb.mongodb.svc.cluster.local:27017" -MONGODB_DATABASE="sientia" -MONGODB_TTL_INDEX_HOURS="1" +MONGODB_USERNAME=mongo_user +MONGODB_PASSWORD=mongo_db_password +MONGODB_URL=my-release-mongodb.mongodb.svc.cluster.local:27017 +MONGODB_DATABASE=sientia +MONGODB_TTL_INDEX_HOURS=1 -MINIO_ENDPOINT_URL="http://minio.minio.svc.cluster.local: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_ENDPOINT_URL=http://minio.minio.svc.cluster.local: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 # Workflow Activity Timeouts (in seconds) # These timeouts are designed to handle large files (up to 200MB) -TIMEOUT_VALIDATE_PARAMS="30" # Parameter validation (fast operation) -TIMEOUT_TRAIN_MODEL="2700" # Model training (30 min for large datasets) -TIMEOUT_DELETE_FILE="120" # Delete file from MinIO (1 min) -TIMEOUT_UPDATE_DATABASE="30" # Database update operations (30 sec) +TIMEOUT_VALIDATE_PARAMS=30 # Parameter validation (fast operation) +TIMEOUT_TRAIN_MODEL=2700 # Model training (30 min for large datasets) +TIMEOUT_DELETE_FILE=120 # Delete file from MinIO (1 min) +TIMEOUT_UPDATE_DATABASE=30 # Database update operations (30 sec) -EXTRA_PIP_REQUIREMENTS="git+https://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@github.com/Aignosi/sientia-mlops-library.git" +# Cleanup Configuration +# Cleanup retention period in hours (files older than this will be deleted) +CLEANUP_RETENTION_HOURS=24 # Default: 24 hours +# Enable dry run mode to test without actually deleting files +CLEANUP_DRY_RUN=false # Set to true for testing without deletion +# Cleanup operation timeouts +TIMEOUT_CLEANUP_MINIO=300 # MinIO cleanup timeout (5 minutes) +TIMEOUT_CLEANUP_LOCAL=120 # Local directory cleanup timeout (2 minutes) +# MinIO list operation page size for cleanup +MAX_KEYS_CLEANUP=1000 # Maximum keys per page when listing objects +# Default bucket for cleanup operations +DEFAULT_CLEANUP_BUCKET=model-training # Default bucket to clean + +EXTRA_PIP_REQUIREMENTS=git+https://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@github.com/Aignosi/sientia-mlops-library.git diff --git a/model_manager/activities/activities.py b/model_manager/activities/activities.py index 4bdcfee..177f9ed 100644 --- a/model_manager/activities/activities.py +++ b/model_manager/activities/activities.py @@ -7,13 +7,14 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.observability.logger import Logger from sientia_do.observability.metrics_controller import MetricsController + from model_manager.activities.cleanup import Cleanup from model_manager.activities.experiment_tracking import ExperimentTracking from model_manager.activities.training import Training from model_manager.utils.repository.model_repository import ModelRepository from model_manager.utils.repository.storage_repository import StorageRepository -class Activities(ExperimentTracking, Training): +class Activities(ExperimentTracking, Training, Cleanup): """ Main activities orchestrator for the Model Manager system. @@ -109,6 +110,14 @@ class Activities(ExperimentTracking, Training): metrics_controller=metrics_controller, ) + Cleanup.__init__( + self, + storage_repository=self.storage_repository, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + def __del__(self): """ Destructor to safely handle cleanup during garbage collection. diff --git a/model_manager/activities/cleanup.py b/model_manager/activities/cleanup.py new file mode 100644 index 0000000..bd1e1ec --- /dev/null +++ b/model_manager/activities/cleanup.py @@ -0,0 +1,338 @@ +""" +Cleanup activities for removing stale files from MinIO and local filesystem. + +This module provides activities for cleaning up temporary files and directories +that are older than the configured retention period. It operates independently +of the database, using timestamps embedded in filenames. +""" + +from temporalio import activity, workflow + +with workflow.unsafe.imports_passed_through(): + import os + import re + import shutil + import traceback + from datetime import UTC, datetime, timedelta + from typing import Any + + from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler + from sientia_do.notifications.models import NotificationLevel + from sientia_do.observability.logger import Logger + from sientia_do.observability.metrics_controller import MetricsController + from sientia_do.observability.sientia_monitoring import SientiaMonitoring + + from model_manager.metrics import ACTIVITY_EXECUTION_TOTAL, WORKFLOW_EXECUTION_TOTAL + from model_manager.utils.repository.storage_repository import StorageRepository + + RETENTION_HOURS = int(os.getenv('CLEANUP_RETENTION_HOURS', '24')) + DRY_RUN = os.getenv('CLEANUP_DRY_RUN', 'false').lower() == 'true' + MAX_KEYS_CLEANUP = int(os.getenv('MAX_KEYS_CLEANUP', '1000')) + + +class Cleanup(SientiaMonitoring): + """ + Activity for cleaning up stale files and directories. + + This activity extends SientiaMonitoring and handles cleanup of: + - MinIO files with timestamp prefixes (timestamp-filename pattern) + - Local temporary directories with timestamp suffixes + """ + + def __init__( + self, + storage_repository: StorageRepository, + logger: Logger, + notification_handler: NotificationHandler, + metrics_controller: MetricsController, + ): + """ + Initialize Cleanup activity. + + Args: + storage_repository: Repository for MinIO operations + logger: Logger instance for observability + notification_handler: Handler for sending notifications + metrics_controller: Controller for metrics emission + """ + SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller) + self.storage_repository = storage_repository + + # Configuration from environment variables + self.retention_hours = RETENTION_HOURS + self.dry_run = DRY_RUN + + # MinIO list operation page size + self.max_keys_cleanup = MAX_KEYS_CLEANUP + + # Regex patterns for timestamp extraction + self.minio_timestamp_pattern = re.compile(r'^(\d{13})-(.+)') # timestamp-filename + self.dir_timestamp_pattern = re.compile( + r'^(.+)_(\d{8}_\d{6}_\d{6})$' + ) # name_YYYYMMDD_HHMMSS_microseconds + + @activity.defn(name='cleanup_minio_files') + async def cleanup_minio_files(self, input_data: dict[str, Any]) -> None: + """ + Clean up stale files from MinIO based on timestamp in filename. + + This activity scans a single MinIO bucket for files following the pattern + '{timestamp}-{filename}' where timestamp is milliseconds since epoch. + Files older than the retention period are deleted. + + Args: + input_data: Cleanup configuration containing: + - metadata (dict): Workflow execution metadata + - bucket_name (str): Name of the bucket to scan + + Returns: + None: Results are logged and tracked via metrics + + Raises: + Exception: If cleanup fails (after sending notification) + """ + metadata = input_data.get('metadata', {}) + bucket_name = input_data.get('bucket_name') + metrics_status = 'success' + + if not bucket_name: + raise ValueError('bucket_name must be provided') + + cutoff_time = datetime.now(UTC) - timedelta(hours=self.retention_hours) + cutoff_timestamp_ms = int(cutoff_time.timestamp() * 1000) + + try: + self.info( + f'Starting MinIO cleanup - Bucket: {bucket_name}, ' + f'Retention: {self.retention_hours}h, Dry run: {self.dry_run}, ' + f'Cutoff: {cutoff_time.isoformat()}', + metadata, + ) + + files_scanned = 0 + files_deleted = 0 + errors = [] + + # List objects in the specified bucket + max_keys = self.max_keys_cleanup # Use environment variable for page size + objects = self.storage_repository.list_bucket_objects(bucket_name, max_keys) + + for obj_key in objects: + files_scanned += 1 + + # Extract timestamp from filename + match = self.minio_timestamp_pattern.match(obj_key) + if not match: + self.debug(f'Skipping file without timestamp pattern: {obj_key}', metadata) + continue + + file_timestamp_ms = int(match.group(1)) + + if file_timestamp_ms < cutoff_timestamp_ms: + if self.dry_run: + self.info( + f'[DRY RUN] Would delete: {obj_key} (age: {(cutoff_time.timestamp() - file_timestamp_ms / 1000) / 3600:.1f}h)', + metadata, + ) + files_deleted += 1 + else: + try: + self.storage_repository.delete_file(bucket_name, obj_key) + self.info(f'Deleted stale file: {obj_key}', metadata) + files_deleted += 1 + except (OSError, PermissionError, ConnectionError) as e: + error_msg = f'Failed to delete {obj_key}: {str(e)}' + errors.append(error_msg) + self.error(error_msg, metadata) + else: + self.debug( + f'Keeping recent file: {obj_key} (age: {(cutoff_time.timestamp() - file_timestamp_ms / 1000) / 3600:.1f}h)', + metadata, + ) + + self.info( + f'MinIO cleanup completed - Bucket: {bucket_name}, ' + f'Scanned: {files_scanned}, Deleted: {files_deleted}, Errors: {len(errors)}', + metadata, + ) + + except Exception as e: + metrics_status = 'error' + error_msg = f'Error in MinIO cleanup: {str(e)}' + trace = traceback.format_exc() + + self.send_notification( + metadata=metadata, + notification_id='CLEANUP_MINIO_ERROR', + message=error_msg, + block='cleanup_minio_files', + level=NotificationLevel.ERROR, + attachment_content=trace, + ) + + raise + finally: + await self._emit_metrics( + metadata=metadata, + metrics_status=metrics_status, + activity_name='cleanup_minio_files', + emit_workflow_metric=(metrics_status == 'error'), + ) + + @activity.defn(name='cleanup_temp_directories') + async def cleanup_temp_directories(self, input_data: dict[str, Any]) -> None: + """ + Clean up stale temporary directories based on timestamp in directory name. + + This activity scans the reports/temp directory for subdirectories following + the pattern '{name}_{timestamp}' where timestamp is in YYYYMMDD_HHMMSS_microseconds format. + Directories older than the retention period are deleted. + + Args: + input_data: Cleanup configuration containing: + - metadata (dict): Workflow execution metadata + - temp_path (str): Path to temp directory (optional, defaults to reports/temp) + + Returns: + None: Results are logged and tracked via metrics + + Raises: + Exception: If cleanup fails (after sending notification) + """ + metadata = input_data.get('metadata', {}) + temp_path = input_data.get('temp_path', 'model_manager/reports/temp') + metrics_status = 'success' + + cutoff_time = datetime.now() - timedelta(hours=self.retention_hours) + + try: + self.info( + f'Starting local directory cleanup - Path: {temp_path}, ' + f'Retention: {self.retention_hours}h, Dry run: {self.dry_run}', + metadata, + ) + + if not os.path.exists(temp_path): + self.warning(f'Temp directory does not exist: {temp_path}', metadata) + return + + directories_scanned = 0 + directories_deleted = 0 + errors = [] + + for item_name in os.listdir(temp_path): + item_path = os.path.join(temp_path, item_name) + + if not os.path.isdir(item_path): + continue + + directories_scanned += 1 + + # Extract timestamp from directory name + match = self.dir_timestamp_pattern.match(item_name) + if not match: + self.debug( + f'Skipping directory without timestamp pattern: {item_name}', metadata + ) + continue + + timestamp_str = match.group(2) + try: + # Parse YYYYMMDD_HHMMSS_microseconds + dir_time = datetime.strptime(timestamp_str, '%Y%m%d_%H%M%S_%f') + + if dir_time < cutoff_time: + age_hours = (datetime.now() - dir_time).total_seconds() / 3600 + + if self.dry_run: + self.info( + f'[DRY RUN] Would delete directory: {item_name} (age: {age_hours:.1f}h)', + metadata, + ) + directories_deleted += 1 + else: + try: + shutil.rmtree(item_path) + self.info( + f'Deleted stale directory: {item_name} (age: {age_hours:.1f}h)', + metadata, + ) + directories_deleted += 1 + except (OSError, PermissionError) as e: + error_msg = f'Failed to delete directory {item_name}: {str(e)}' + errors.append(error_msg) + self.error(error_msg, metadata) + else: + age_hours = (datetime.now() - dir_time).total_seconds() / 3600 + self.debug( + f'Keeping recent directory: {item_name} (age: {age_hours:.1f}h)', + metadata, + ) + + except ValueError as e: + error_msg = f'Invalid timestamp format in directory {item_name}: {str(e)}' + errors.append(error_msg) + self.error(error_msg, metadata) + + self.info( + f'Directory cleanup completed - Scanned: {directories_scanned}, ' + f'Deleted: {directories_deleted}, Errors: {len(errors)}', + metadata, + ) + + except Exception as e: + metrics_status = 'error' + error_msg = f'Error in directory cleanup: {str(e)}' + trace = traceback.format_exc() + + self.send_notification( + metadata=metadata, + notification_id='CLEANUP_DIRECTORIES_ERROR', + message=error_msg, + block='cleanup_temp_directories', + level=NotificationLevel.ERROR, + attachment_content=trace, + ) + + raise + finally: + await self._emit_metrics( + metadata=metadata, + metrics_status=metrics_status, + activity_name='cleanup_temp_directories', + emit_workflow_metric=True, + ) + + async def _emit_metrics( + self, + metadata: dict[str, Any], + metrics_status: str, + activity_name: str, + emit_workflow_metric: bool, + ) -> None: + """ + Emit workflow and activity execution metrics. + + Args: + metadata: Activity metadata containing pod_id and workflow_name + metrics_status: Execution status ('success' or 'error') + activity_name: Name of the activity being executed + """ + if emit_workflow_metric: + await self.emit_metric( + metric_object=WORKFLOW_EXECUTION_TOTAL, + tags={ + 'pod_id': metadata.get('pod_id'), + 'workflow_name': metadata.get('workflow_name'), + 'status': metrics_status, + }, + ) + + await self.emit_metric( + metric_object=ACTIVITY_EXECUTION_TOTAL, + tags={ + 'pod_id': metadata.get('pod_id'), + 'activity_name': activity_name, + 'status': metrics_status, + }, + ) diff --git a/model_manager/activities/experiment_tracking.py b/model_manager/activities/experiment_tracking.py index d61573c..c9686de 100644 --- a/model_manager/activities/experiment_tracking.py +++ b/model_manager/activities/experiment_tracking.py @@ -76,7 +76,8 @@ class ExperimentTracking(Postgres): Raises: ConnectionError: If database connection cannot be established """ - super().__init__( + Postgres.__init__( + self, host=host, port=port, user=user, diff --git a/model_manager/activities/training.py b/model_manager/activities/training.py index 6e7d167..aaae6b2 100644 --- a/model_manager/activities/training.py +++ b/model_manager/activities/training.py @@ -51,7 +51,7 @@ class Training(SientiaMonitoring): logger: Logger instance for observability notification_handler: Handler for sending notifications """ - super().__init__(logger, notification_handler, metrics_controller) + SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller) self.training_repository = TrainingRepository(logger) self.model_repository = model_repository self.storage_repository = storage_repository diff --git a/model_manager/utils/repository/storage_repository.py b/model_manager/utils/repository/storage_repository.py index 71f3110..3c655f4 100644 --- a/model_manager/utils/repository/storage_repository.py +++ b/model_manager/utils/repository/storage_repository.py @@ -125,3 +125,36 @@ class StorageRepository: """ self.minio_client.delete_object(Bucket=bucket_name, Key=file_name) self.logger.info(f'File deleted successfully: {bucket_name}/{file_name}') + + def list_bucket_objects(self, bucket_name: str, max_keys: int = 1000) -> list[str]: + """ + List objects in a MinIO bucket. + + This method uses the MinIO/S3 list_objects_v2 API to retrieve objects + from the specified bucket. This is optimized for cleanup operations + by using configurable pagination. + + Args: + bucket_name: Name of the bucket to list objects from. + max_keys: Maximum number of keys per page (default: 1000). + + Returns: + List[str]: List of object keys (file names). + """ + # Use list_objects_v2 for efficient pagination + paginator = self.minio_client.get_paginator('list_objects_v2') + + pages = paginator.paginate(Bucket=bucket_name, MaxKeys=max_keys) + + objects = [] + total_count = 0 + + for page in pages: + if 'Contents' in page: + for obj in page['Contents']: + objects.append(obj['Key']) + total_count += 1 + + self.logger.info(f'Listed {total_count} objects from bucket {bucket_name}') + + return objects diff --git a/model_manager/worker/worker.py b/model_manager/worker/worker.py index cb96773..ff98693 100644 --- a/model_manager/worker/worker.py +++ b/model_manager/worker/worker.py @@ -43,6 +43,7 @@ with workflow.unsafe.imports_passed_through(): build_postgres_config, ) from model_manager.utils.logger_helper import get_logger + from model_manager.workflows.cleanup_files import CleanupFiles from model_manager.workflows.train_model import TrainModel POD_ID = os.getenv('POD_ID') @@ -73,7 +74,6 @@ async def main(): metadata = { 'pod_id': POD_ID, - 'workflow_name': 'train_model', } logger.custom_info(f'Starting Worker with POD_ID: {POD_ID}', metadata) @@ -130,10 +130,25 @@ async def main(): activities.train_model, activities.cleanup_resources, ], - max_concurrent_workflow_tasks=50, - max_concurrent_activities=50, - max_concurrent_local_activities=50, - max_cached_workflows=200, + max_concurrent_workflow_tasks=10, + max_concurrent_activities=10, + max_concurrent_local_activities=10, + max_cached_workflows=100, + workflow_task_poller_behavior=PollerBehaviorAutoscaling(), + activity_task_poller_behavior=PollerBehaviorAutoscaling(), + ), + Worker( + temporal_client, + task_queue='cleanup-queue', + workflows=[CleanupFiles], + activities=[ + activities.cleanup_minio_files, + activities.cleanup_temp_directories, + ], + max_concurrent_workflow_tasks=20, + max_concurrent_activities=20, + max_concurrent_local_activities=20, + max_cached_workflows=100, workflow_task_poller_behavior=PollerBehaviorAutoscaling(), activity_task_poller_behavior=PollerBehaviorAutoscaling(), ), diff --git a/model_manager/workflows/cleanup_files.py b/model_manager/workflows/cleanup_files.py new file mode 100644 index 0000000..685f946 --- /dev/null +++ b/model_manager/workflows/cleanup_files.py @@ -0,0 +1,83 @@ +""" +Cleanup workflow for removing stale files from MinIO and local filesystem. + +This module provides a Temporal cron workflow that runs daily to clean up +temporary files and directories older than the configured retention period. +""" + +from temporalio import workflow + +with workflow.unsafe.imports_passed_through(): + import os + from datetime import timedelta + from typing import Any + + from model_manager.activities.activities import Activities + from model_manager.workflows.train_model import POD_ID, network_retry_policy, no_retry_policy + + TIMEOUT_CLEANUP_MINIO = int(os.getenv('TIMEOUT_CLEANUP_MINIO', '300')) + TIMEOUT_CLEANUP_LOCAL = int(os.getenv('TIMEOUT_CLEANUP_LOCAL', '120')) + DEFAULT_CLEANUP_BUCKET = os.getenv('DEFAULT_CLEANUP_BUCKET', 'model-training') + + +@workflow.defn(name='cleanup_files') +class CleanupFiles: + """ + Cleanup workflow for removing stale files. + + This workflow cleans up: + - MinIO files with timestamp prefixes + - Local temporary directories with timestamp suffixes + + The workflow is designed to be simple and robust, with error handling + delegated to the individual activities. + """ + + @workflow.run + async def run(self, input_data: dict[str, Any]) -> None: + """ + Execute the cleanup workflow. + + This method orchestrates the cleanup of MinIO files and local directories + in sequence. No exception handling is needed as activities handle their + own errors and notifications. + + Args: + input_data: Workflow configuration containing optional: + - bucket_name (str): Bucket to clean (defaults to environment variable) + """ + # Get bucket name from input or environment + bucket_name = input_data.get('bucket_name', DEFAULT_CLEANUP_BUCKET) + + # Default temp path for local cleanup + temp_path = 'model_manager/reports/temp' + + # Metadata for tracking + metadata = { + 'metadata': { + 'pod_id': POD_ID, + 'workflow_name': 'cleanup_files', + } + } + + # Execute MinIO cleanup + await workflow.execute_activity_method( + Activities.cleanup_minio_files, + { + **metadata, + 'bucket_name': bucket_name, + }, + retry_policy=network_retry_policy, + start_to_close_timeout=timedelta(seconds=TIMEOUT_CLEANUP_MINIO), + ) + + # Execute local directory cleanup + await workflow.execute_activity_method( + Activities.cleanup_temp_directories, + { + **metadata, + 'temp_path': temp_path, + }, + retry_policy=no_retry_policy, + start_to_close_timeout=timedelta(seconds=TIMEOUT_CLEANUP_LOCAL), + ) diff --git a/scripts/run_cleanup_test.py b/scripts/run_cleanup_test.py new file mode 100644 index 0000000..6084daa --- /dev/null +++ b/scripts/run_cleanup_test.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +"""Run cleanup_files workflow once for manual testing. + +This script starts the Temporal workflow `cleanup_files` a single time, +using the same Temporal namespace and task queue as the main worker. + +It is intended only for local/manual testing; scheduling (cron) must be +configured separately in Temporal. +""" + +from __future__ import annotations + +import asyncio +import os +import sys +from datetime import timedelta +from typing import Any + +from temporalio.client import Client + +# Ensure project root is on PYTHONPATH when running directly +ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if ROOT_DIR not in sys.path: + sys.path.insert(0, ROOT_DIR) + +from model_manager.workflows.cleanup_files import CleanupFiles # noqa: E402 + + +async def main(argv: list[str]) -> None: + """Entry point for manual cleanup workflow execution. + + Args: + argv: Command-line arguments (excluding program name). + """ + + # Config from environment / defaults + temporal_host = os.getenv('TEMPORAL_HOST', 'localhost:37463') + temporal_namespace = os.getenv('TEMPORAL_NAMESPACE', 'model-manager') + task_queue = 'cleanup-queue' + + default_bucket = os.getenv('DEFAULT_CLEANUP_BUCKET', 'model-training') + + # Optional CLI: bucket name override + bucket_name = default_bucket + if argv: + bucket_name = argv[0] + + print(f"Connecting to Temporal at {temporal_host} (namespace={temporal_namespace})...") + client = await Client.connect(temporal_host, namespace=temporal_namespace) + + input_data: dict[str, Any] = { + 'bucket_name': bucket_name, + } + + workflow_id = f"cleanup-files-manual-{int(asyncio.get_event_loop().time())}" + + print(f"Starting cleanup_files workflow once...\n" + f" workflow_id = {workflow_id}\n" + f" task_queue = {task_queue}\n" + f" bucket_name = {bucket_name}") + + handle = await client.start_workflow( + CleanupFiles.run, + input_data, + id=workflow_id, + task_queue=task_queue, + run_timeout=timedelta(minutes=10), + ) + + print("Workflow started, waiting for completion...") + await handle.result() + print("cleanup_files workflow completed successfully.") + + +if __name__ == '__main__': # pragma: no cover - manual utility script + asyncio.run(main(sys.argv[1:])) diff --git a/tests/worker/test_worker.py b/tests/worker/test_worker.py index ff49482..8883824 100644 --- a/tests/worker/test_worker.py +++ b/tests/worker/test_worker.py @@ -234,7 +234,8 @@ async def test_main_successful_startup( mock_notification_handler_class.assert_called_once() mock_activities_class.assert_called_once() mock_client_class.connect.assert_called_once() - mock_worker_class.assert_called_once() + # Agora sao criados dois Workers: um para train_model-queue e outro para cleanup-queue + assert mock_worker_class.call_count == 2 # Verify cleanup was performed mock_notification_handler.shutdown.assert_called_once() @@ -468,22 +469,31 @@ async def test_main_worker_configuration( await main() # Verify Worker was created with correct configuration - mock_worker_class.assert_called_once() - call_args = mock_worker_class.call_args + assert mock_worker_class.call_count == 2 - assert call_args[0][0] == mock_client_instance # temporal_client - assert call_args[1]['task_queue'] == 'train_model-queue' - assert call_args[1]['max_concurrent_workflow_tasks'] == 50 - assert call_args[1]['max_concurrent_activities'] == 50 - assert call_args[1]['max_concurrent_local_activities'] == 50 - assert call_args[1]['max_cached_workflows'] == 200 + # Primeira chamada: worker de treinamento (train_model-queue) + train_call_args = mock_worker_class.call_args_list[0] + assert train_call_args[0][0] == mock_client_instance # temporal_client + assert train_call_args[1]['task_queue'] == 'train_model-queue' + assert train_call_args[1]['max_concurrent_workflow_tasks'] == 10 + assert train_call_args[1]['max_concurrent_activities'] == 10 + assert train_call_args[1]['max_concurrent_local_activities'] == 10 + assert train_call_args[1]['max_cached_workflows'] == 100 - # Verify activities are included - activities_list = call_args[1]['activities'] - assert mock_activities.update_experiment_run in activities_list - assert mock_activities.validate_train_params in activities_list - assert mock_activities.train_model in activities_list - assert mock_activities.cleanup_resources in activities_list + train_activities_list = train_call_args[1]['activities'] + assert mock_activities.update_experiment_run in train_activities_list + assert mock_activities.validate_train_params in train_activities_list + assert mock_activities.train_model in train_activities_list + assert mock_activities.cleanup_resources in train_activities_list + + # Segunda chamada: worker de cleanup (cleanup-queue) + cleanup_call_args = mock_worker_class.call_args_list[1] + assert cleanup_call_args[0][0] == mock_client_instance # temporal_client + assert cleanup_call_args[1]['task_queue'] == 'cleanup-queue' + assert cleanup_call_args[1]['max_concurrent_workflow_tasks'] == 20 + assert cleanup_call_args[1]['max_concurrent_activities'] == 20 + assert cleanup_call_args[1]['max_concurrent_local_activities'] == 20 + assert cleanup_call_args[1]['max_cached_workflows'] == 100 @patch('model_manager.worker.worker.asyncio.run') diff --git a/todo-list.txt b/todo-list.txt index f9fee32..ba89408 100644 --- a/todo-list.txt +++ b/todo-list.txt @@ -1,7 +1,9 @@ -- Criar o dashboard do grafana. +- Atualizar o sientia-dataops-library para a versão 1.6.1 +- Refatorar o arquivo .dockerignore para só deixar copiar os arquivos que forem necessários para a execução do container, pois ele está copiando muitos arquivos desnecessários. +- atualizar as variáveis de ambiente no helm chart +- Criar um gráfico no grafana para cada nova atividade. +- Atualizar a documentação do projeto. - Atualizar o .github/workflows/quality-gate.yml para usar os pipelines genéricos do github; Criar um workflow para fazer o deploy no suse. Criar um workflow para criar o release no github. - -- Atualizar a documentação do projeto. From 2ee66552d2b55e85b22637223990359f52775ce6 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Mon, 24 Nov 2025 11:21:43 -0300 Subject: [PATCH 02/16] SIENTIAPDE-1350: Configure separate task queues for train and cleanup workers, update sientia-dataops-library to 1.6.1 and refactor prometheus server startup to use logger. --- .env.example | 2 ++ model_manager/worker/worker.py | 21 ++++++++------- requirements.txt | 2 +- tests/worker/test_worker.py | 48 ++++++++++++++++++++++------------ 4 files changed, 45 insertions(+), 28 deletions(-) diff --git a/.env.example b/.env.example index b106f15..f977184 100644 --- a/.env.example +++ b/.env.example @@ -17,6 +17,8 @@ PROJECT_NAME=sientia-model-manager TEMPORAL_HOST=temporal-frontend.temporal.svc.cluster.local:7233 TEMPORAL_NAMESPACE=model-manager +TRAIN_TASK_QUEUE=train_model-queue +CLEANUP_TASK_QUEUE=cleanup-queue MONGODB_USERNAME=mongo_user MONGODB_PASSWORD=mongo_db_password diff --git a/model_manager/worker/worker.py b/model_manager/worker/worker.py index ff98693..d290deb 100644 --- a/model_manager/worker/worker.py +++ b/model_manager/worker/worker.py @@ -33,6 +33,7 @@ with workflow.unsafe.imports_passed_through(): from prometheus_client import start_http_server from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler + from sientia_do.observability.logger import Logger as SientiaLogger from model_manager import metrics from model_manager.activities.activities import Activities @@ -48,6 +49,8 @@ with workflow.unsafe.imports_passed_through(): POD_ID = os.getenv('POD_ID') SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', '9091')) +TRAIN_TASK_QUEUE = os.getenv('TRAIN_TASK_QUEUE', 'train_model-queue') +CLEANUP_TASK_QUEUE = os.getenv('CLEANUP_TASK_QUEUE', 'cleanup-queue') async def main(): @@ -77,13 +80,10 @@ async def main(): } logger.custom_info(f'Starting Worker with POD_ID: {POD_ID}', metadata) - - logger.custom_info('Starting prometheus client...', metadata) - start_prometheus_server() - + start_prometheus_server(logger, metadata) logger.custom_info('Starting Notification Handler...', metadata) - mongo_config = build_mongodb_config() + notification_handler = NotificationHandler( connection_string=mongo_config['connection_string'], database=mongo_config['database_name'], @@ -122,7 +122,7 @@ async def main(): workers = [ Worker( temporal_client, - task_queue='train_model-queue', + task_queue=TRAIN_TASK_QUEUE, workflows=[TrainModel], activities=[ activities.update_experiment_run, @@ -139,7 +139,7 @@ async def main(): ), Worker( temporal_client, - task_queue='cleanup-queue', + task_queue=CLEANUP_TASK_QUEUE, workflows=[CleanupFiles], activities=[ activities.cleanup_minio_files, @@ -155,6 +155,7 @@ async def main(): ] handlers = [] + for w in workers: handlers.append(w.run()) @@ -174,7 +175,7 @@ async def main(): sys.exit(1) -def start_prometheus_server(): +def start_prometheus_server(logger: SientiaLogger, metadata: dict[str, str | None]): """ Starts the Prometheus metrics server for monitoring and observability. @@ -194,10 +195,10 @@ def start_prometheus_server(): try: port = int(os.getenv('HTTP_METRICS_PORT', 9090)) start_http_server(port) - print(f'Prometheus server started on port {port}.') + logger.custom_info(f'Prometheus server started on port {port}.', metadata) metrics.APP_UP.labels(pod_id=POD_ID).set(1) # Mark app as UP except Exception as e: # noqa: BLE001 - print(f'Failed to start Prometheus server: {e}') + logger.custom_critical(f'Failed to start Prometheus server: {e}', metadata) os._exit(1) diff --git a/requirements.txt b/requirements.txt index be6b7b6..362a16b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,7 +3,7 @@ psycopg2-binary==2.9.11 sqlalchemy==2.0.44 boto3==1.40.55 botocore==1.40.55 -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.2 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.6.1 prometheus-client==0.23.1 mlflow==2.10.1 evidently==0.4.21 diff --git a/tests/worker/test_worker.py b/tests/worker/test_worker.py index 8883824..0e036c2 100644 --- a/tests/worker/test_worker.py +++ b/tests/worker/test_worker.py @@ -109,14 +109,18 @@ def test_sdk_metrics_port_from_env(): @patch('model_manager.worker.worker.POD_ID', 'test-pod-123') @patch('model_manager.worker.worker.start_http_server') @patch('model_manager.worker.worker.metrics') -def test_start_prometheus_server_success(mock_metrics, mock_start_http_server, mock_env_vars): +def test_start_prometheus_server_success( + mock_metrics, mock_start_http_server, mock_env_vars, mock_logger +): """Test successful Prometheus server startup.""" from model_manager.worker.worker import start_prometheus_server mock_app_up = Mock() mock_metrics.APP_UP.labels.return_value = mock_app_up - start_prometheus_server() + metadata = {'pod_id': 'test-pod-123', 'workflow_name': 'train_model'} + + start_prometheus_server(mock_logger, metadata) # Verify HTTP server started mock_start_http_server.assert_called_once_with(9090) @@ -124,11 +128,12 @@ def test_start_prometheus_server_success(mock_metrics, mock_start_http_server, m # Verify APP_UP metric was set to 1 mock_metrics.APP_UP.labels.assert_called_once_with(pod_id='test-pod-123') mock_app_up.set.assert_called_once_with(1) + mock_logger.custom_info.assert_called_once() @patch('model_manager.worker.worker.start_http_server') @patch('model_manager.worker.worker.metrics') -def test_start_prometheus_server_custom_port(mock_metrics, mock_start_http_server): +def test_start_prometheus_server_custom_port(mock_metrics, mock_start_http_server, mock_logger): """Test Prometheus server startup with custom port.""" from model_manager.worker.worker import start_prometheus_server @@ -136,7 +141,9 @@ def test_start_prometheus_server_custom_port(mock_metrics, mock_start_http_serve mock_app_up = Mock() mock_metrics.APP_UP.labels.return_value = mock_app_up - start_prometheus_server() + metadata = {'pod_id': 'custom-pod', 'workflow_name': 'train_model'} + + start_prometheus_server(mock_logger, metadata) mock_start_http_server.assert_called_once_with(8080) @@ -145,17 +152,20 @@ def test_start_prometheus_server_custom_port(mock_metrics, mock_start_http_serve @patch('model_manager.worker.worker.metrics') @patch('model_manager.worker.worker.os._exit') def test_start_prometheus_server_failure( - mock_exit, mock_metrics, mock_start_http_server, mock_env_vars + mock_exit, mock_metrics, mock_start_http_server, mock_env_vars, mock_logger ): """Test Prometheus server startup failure.""" from model_manager.worker.worker import start_prometheus_server mock_start_http_server.side_effect = OSError('Port already in use') - start_prometheus_server() + metadata = {'pod_id': 'test-pod-123', 'workflow_name': 'train_model'} - # Verify exit was called with code 1 + start_prometheus_server(mock_logger, metadata) + + # Verify exit was called with code 1 e log crítico emitido mock_exit.assert_called_once_with(1) + mock_logger.custom_critical.assert_called_once() @pytest.mark.asyncio @@ -234,7 +244,7 @@ async def test_main_successful_startup( mock_notification_handler_class.assert_called_once() mock_activities_class.assert_called_once() mock_client_class.connect.assert_called_once() - # Agora sao criados dois Workers: um para train_model-queue e outro para cleanup-queue + # Agora são criados dois Workers: um para train_model-queue e outro para cleanup-queue assert mock_worker_class.call_count == 2 # Verify cleanup was performed @@ -523,7 +533,7 @@ def test_worker_module_docstring(): @patch('model_manager.worker.worker.start_http_server') @patch('model_manager.worker.worker.metrics') def test_start_prometheus_server_prints_success( - mock_metrics, mock_start_http_server, capsys, mock_env_vars + mock_metrics, mock_start_http_server, capsys, mock_env_vars, mock_logger ): """Test that start_prometheus_server prints success message.""" from model_manager.worker.worker import start_prometheus_server @@ -531,25 +541,29 @@ def test_start_prometheus_server_prints_success( mock_app_up = Mock() mock_metrics.APP_UP.labels.return_value = mock_app_up - start_prometheus_server() + metadata = {'pod_id': 'test-pod-123', 'workflow_name': 'train_model'} - captured = capsys.readouterr() - assert 'Prometheus server started on port 9090' in captured.out + start_prometheus_server(mock_logger, metadata) + + # Agora a mensagem é enviada via logger + mock_logger.custom_info.assert_called_once() @patch('model_manager.worker.worker.start_http_server') @patch('model_manager.worker.worker.metrics') @patch('model_manager.worker.worker.os._exit') def test_start_prometheus_server_prints_failure( - mock_exit, mock_metrics, mock_start_http_server, capsys, mock_env_vars + mock_exit, mock_metrics, mock_start_http_server, capsys, mock_env_vars, mock_logger ): """Test that start_prometheus_server prints failure message.""" from model_manager.worker.worker import start_prometheus_server mock_start_http_server.side_effect = Exception('Test error') - start_prometheus_server() + metadata = {'pod_id': 'test-pod-123', 'workflow_name': 'train_model'} - captured = capsys.readouterr() - assert 'Failed to start Prometheus server' in captured.out - assert 'Test error' in captured.out + start_prometheus_server(mock_logger, metadata) + + # Agora o erro é logado via logger crítico + mock_logger.custom_critical.assert_called_once() + mock_exit.assert_called_once_with(1) From 463906073e15e3ea42ea7246e46db70dfec8a0f0 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Mon, 24 Nov 2025 23:08:29 -0300 Subject: [PATCH 03/16] SIENTIAPDE-1350: Refactor: Improve logging, configuration, and resource management. Includes .env updates, client initialization logging, and resource closing. --- .env.example | 29 ++++++--------- model_manager/activities/activities.py | 2 + .../activities/experiment_tracking.py | 2 + model_manager/utils/connectors_config.py | 1 + .../utils/repository/model_repository.py | 2 +- .../utils/repository/storage_repository.py | 6 ++- model_manager/worker/worker.py | 17 ++++----- run_local.sh | 8 ---- scripts/run_training_test.py | 37 +++++++++++-------- todo-list.txt | 2 +- 10 files changed, 53 insertions(+), 53 deletions(-) diff --git a/.env.example b/.env.example index f977184..6efb9cb 100644 --- a/.env.example +++ b/.env.example @@ -19,6 +19,7 @@ TEMPORAL_HOST=temporal-frontend.temporal.svc.cluster.local:7233 TEMPORAL_NAMESPACE=model-manager TRAIN_TASK_QUEUE=train_model-queue CLEANUP_TASK_QUEUE=cleanup-queue +TEMPORAL_USE_TLS=false MONGODB_USERNAME=mongo_user MONGODB_PASSWORD=mongo_db_password @@ -36,24 +37,16 @@ MINIO_RETRY_MODE=adaptive MINIO_CONNECT_TIMEOUT=10 MINIO_READ_TIMEOUT=60 -# Workflow Activity Timeouts (in seconds) -# These timeouts are designed to handle large files (up to 200MB) -TIMEOUT_VALIDATE_PARAMS=30 # Parameter validation (fast operation) -TIMEOUT_TRAIN_MODEL=2700 # Model training (30 min for large datasets) -TIMEOUT_DELETE_FILE=120 # Delete file from MinIO (1 min) -TIMEOUT_UPDATE_DATABASE=30 # Database update operations (30 sec) +TIMEOUT_VALIDATE_PARAMS=30 +TIMEOUT_TRAIN_MODEL=2700 +TIMEOUT_DELETE_FILE=120 +TIMEOUT_UPDATE_DATABASE=30 -# Cleanup Configuration -# Cleanup retention period in hours (files older than this will be deleted) -CLEANUP_RETENTION_HOURS=24 # Default: 24 hours -# Enable dry run mode to test without actually deleting files -CLEANUP_DRY_RUN=false # Set to true for testing without deletion -# Cleanup operation timeouts -TIMEOUT_CLEANUP_MINIO=300 # MinIO cleanup timeout (5 minutes) -TIMEOUT_CLEANUP_LOCAL=120 # Local directory cleanup timeout (2 minutes) -# MinIO list operation page size for cleanup -MAX_KEYS_CLEANUP=1000 # Maximum keys per page when listing objects -# Default bucket for cleanup operations -DEFAULT_CLEANUP_BUCKET=model-training # Default bucket to clean +CLEANUP_RETENTION_HOURS=24 +CLEANUP_DRY_RUN=false +TIMEOUT_CLEANUP_MINIO=300 +TIMEOUT_CLEANUP_LOCAL=120 +MAX_KEYS_CLEANUP=1000 +DEFAULT_CLEANUP_BUCKET=model-training EXTRA_PIP_REQUIREMENTS=git+https://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@github.com/Aignosi/sientia-mlops-library.git diff --git a/model_manager/activities/activities.py b/model_manager/activities/activities.py index 177f9ed..972c27e 100644 --- a/model_manager/activities/activities.py +++ b/model_manager/activities/activities.py @@ -152,3 +152,5 @@ class Activities(ExperimentTracking, Training, Cleanup): Prefer calling this method explicitly rather than relying on __del__. """ ExperimentTracking.close(self) + self.info('Postgres client closed') + self.storage_repository.close() diff --git a/model_manager/activities/experiment_tracking.py b/model_manager/activities/experiment_tracking.py index c9686de..7ab6d59 100644 --- a/model_manager/activities/experiment_tracking.py +++ b/model_manager/activities/experiment_tracking.py @@ -90,6 +90,8 @@ class ExperimentTracking(Postgres): metrics_controller=metrics_controller, ) + self.info(f'Postgres client initialized at {host}:{port}') + def __del__(self): """ Destructor to safely handle cleanup during garbage collection. diff --git a/model_manager/utils/connectors_config.py b/model_manager/utils/connectors_config.py index b42d017..9a1e40a 100644 --- a/model_manager/utils/connectors_config.py +++ b/model_manager/utils/connectors_config.py @@ -84,6 +84,7 @@ def build_mongodb_config() -> dict[str, Any]: 'connection_string': connection_string, 'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'), 'ttl_index_seconds': int(getenv('MONGODB_TTL_INDEX_HOURS', '1')) * 3600, + 'uri': uri, } diff --git a/model_manager/utils/repository/model_repository.py b/model_manager/utils/repository/model_repository.py index 5154f84..803e7da 100644 --- a/model_manager/utils/repository/model_repository.py +++ b/model_manager/utils/repository/model_repository.py @@ -31,8 +31,8 @@ warnings.filterwarnings('ignore', category=FutureWarning, message=".*'squared' i class ModelRepository: def __init__(self, url, username, password, logger: Logger): self.model_serving = ModelServing(tracking_uri=url, username=username, password=password) - self.logger = logger + self.logger.info(f'MLFlow client initialized at {url}') def save_model(self, train_result: TrainModelResult) -> TrainModelResult: """ diff --git a/model_manager/utils/repository/storage_repository.py b/model_manager/utils/repository/storage_repository.py index 3c655f4..c8639fb 100644 --- a/model_manager/utils/repository/storage_repository.py +++ b/model_manager/utils/repository/storage_repository.py @@ -86,7 +86,11 @@ class StorageRepository: use_ssl=use_ssl, ) - self.logger.info(f'MinIO client initialized successfully: {endpoint_url}') + self.logger.info(f'MinIO client initialized at {endpoint_url}') + + def close(self) -> None: + self.minio_client.close() + self.logger.info('MinIO client closed') def fetch_file(self, bucket_name: str, file_name: str) -> BytesIO: """ diff --git a/model_manager/worker/worker.py b/model_manager/worker/worker.py index d290deb..29200aa 100644 --- a/model_manager/worker/worker.py +++ b/model_manager/worker/worker.py @@ -73,15 +73,14 @@ async def main(): SystemExit: On graceful shutdown or error conditions """ host = os.getenv('TEMPORAL_HOST', 'localhost:7233') + use_tls = os.getenv('TEMPORAL_USE_TLS', 'false').lower() == 'true' logger = get_logger(__name__) metadata = { 'pod_id': POD_ID, } - logger.custom_info(f'Starting Worker with POD_ID: {POD_ID}', metadata) start_prometheus_server(logger, metadata) - logger.custom_info('Starting Notification Handler...', metadata) mongo_config = build_mongodb_config() notification_handler = NotificationHandler( @@ -91,7 +90,7 @@ async def main(): project_name=os.getenv('PROJECT_NAME', 'model-manager'), ) - logger.custom_info('Starting Activities...', metadata) + logger.custom_info(f'MongoDB client initialized at {mongo_config["uri"]}', metadata) activities = Activities( postgres_config=build_postgres_config(), @@ -101,23 +100,22 @@ async def main(): notification_handler=notification_handler, ) - logger.custom_info(f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata) - new_runtime = Runtime( telemetry=TelemetryConfig( metrics=PrometheusConfig(bind_address=f'0.0.0.0:{SDK_METRICS_PORT}') ) ) - logger.custom_info(f'Starting Temporal Client at {host}...', metadata) + logger.custom_info(f'SDK metrics server initialized on port {SDK_METRICS_PORT}', metadata) temporal_client = await client.Client.connect( target_host=host, namespace=os.getenv('TEMPORAL_NAMESPACE', 'model-manager'), runtime=new_runtime, + tls=use_tls, ) - logger.custom_info('Starting Workers...', metadata) + logger.custom_info(f'Temporal client initialized at {host}', metadata) workers = [ Worker( @@ -159,7 +157,7 @@ async def main(): for w in workers: handlers.append(w.run()) - logger.custom_info('Workers started successfully', metadata) + logger.custom_info('Model manager workers initialized', metadata) try: # This will run the workers and wait for them to complete. @@ -169,6 +167,7 @@ async def main(): logger.custom_error(f'An unhandled exception occurred: {e}', metadata) finally: notification_handler.shutdown() + logger.custom_info('MongoDB client closed', metadata) await activities.shutdown() # Exit with a non-zero status code to indicate failure to Kubernetes metrics.APP_UP.labels(pod_id=POD_ID).set(0) # Mark app as DOWN @@ -195,7 +194,7 @@ def start_prometheus_server(logger: SientiaLogger, metadata: dict[str, str | Non try: port = int(os.getenv('HTTP_METRICS_PORT', 9090)) start_http_server(port) - logger.custom_info(f'Prometheus server started 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 logger.custom_critical(f'Failed to start Prometheus server: {e}', metadata) diff --git a/run_local.sh b/run_local.sh index d4c8c10..4c88728 100755 --- a/run_local.sh +++ b/run_local.sh @@ -3,12 +3,6 @@ # Exit on any error set -e -#echo "Activating virtual environment..." - -#conda activate ./venv - -echo "Loading environment variables from .env..." - if [ -f .env ]; then export $(cat .env | grep -v '^#' | xargs) echo "Environment variables loaded from .env" @@ -16,6 +10,4 @@ else echo "Warning: .env file not found. Continuing without environment variables." fi -echo "Starting ingestor application..." - python -m model_manager.worker.worker diff --git a/scripts/run_training_test.py b/scripts/run_training_test.py index dbb641f..a978daa 100644 --- a/scripts/run_training_test.py +++ b/scripts/run_training_test.py @@ -24,29 +24,35 @@ import uuid from datetime import datetime, timedelta from pathlib import Path +from dotenv import load_dotenv import psycopg2 from psycopg2.extras import Json from temporalio import client + +# Carrega variáveis de ambiente do arquivo .env na raiz do projeto +PROJECT_ROOT = Path(__file__).resolve().parent.parent +ENV_PATH = PROJECT_ROOT / '.env' +if ENV_PATH.exists(): + load_dotenv(dotenv_path=ENV_PATH) + + DOCS_PATH = Path('docs/test-model-data.csv') -MINIO_ALIAS = os.getenv('MINIO_ALIAS', 'suse') -MINIO_BUCKET = os.getenv('MINIO_BUCKET', 'model-training') +MINIO_ALIAS = 'suse' +MINIO_BUCKET = 'model-training' POSTGRES_CONFIG = { - 'host': os.getenv('POSTGRES_HOST', 'localhost'), - 'port': os.getenv('POSTGRES_PORT', '55432'), - 'user': os.getenv('POSTGRES_USER', 'postgres'), - 'password': os.getenv( - 'POSTGRES_PASSWORD', - 'nFqc81y6kwmr2zuAIx43DhiOosFCVPpeEfTtTWZflkNjB2j1KtEeIANkhFR9mAX3', - ), - 'dbname': os.getenv('POSTGRES_DBNAME', 'sientia-core-mlops-bff'), + 'host': os.getenv('POSTGRES_HOST'), + 'port': os.getenv('POSTGRES_PORT'), + 'user': os.getenv('POSTGRES_USER'), + 'password': os.getenv('POSTGRES_PASSWORD'), + 'dbname': os.getenv('POSTGRES_DBNAME'), } -TEMPORAL_HOST = os.getenv('TEMPORAL_HOST', 'localhost:37463') -TEMPORAL_NAMESPACE = os.getenv('TEMPORAL_NAMESPACE', 'model-manager') -TEMPORAL_TASK_QUEUE = os.getenv('TEMPORAL_TASK_QUEUE', 'train_model-queue') -TEMPORAL_WORKFLOW = os.getenv('TEMPORAL_WORKFLOW', 'train_model') +TEMPORAL_HOST = os.getenv('TEMPORAL_HOST') +TEMPORAL_NAMESPACE = os.getenv('TEMPORAL_NAMESPACE') +TRAIN_TASK_QUEUE = os.getenv('TRAIN_TASK_QUEUE') +TEMPORAL_WORKFLOW = 'train_model' BASE_REQUEST_DATA = { 'experimentName': 'model-manager-test-01', @@ -169,6 +175,7 @@ async def trigger_temporal_workflow(workflow_input: dict) -> str: 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()}' @@ -176,7 +183,7 @@ async def trigger_temporal_workflow(workflow_input: dict) -> str: TEMPORAL_WORKFLOW, workflow_input, id=workflow_id, - task_queue=TEMPORAL_TASK_QUEUE, + task_queue=TRAIN_TASK_QUEUE, execution_timeout=timedelta(minutes=5), run_timeout=timedelta(minutes=5), task_timeout=timedelta(minutes=5), diff --git a/todo-list.txt b/todo-list.txt index ba89408..a2d8850 100644 --- a/todo-list.txt +++ b/todo-list.txt @@ -1,7 +1,7 @@ -- Atualizar o sientia-dataops-library para a versão 1.6.1 - Refatorar o arquivo .dockerignore para só deixar copiar os arquivos que forem necessários para a execução do container, pois ele está copiando muitos arquivos desnecessários. - atualizar as variáveis de ambiente no helm chart - Criar um gráfico no grafana para cada nova atividade. +- Atualizar a documentação dos métodos alterados. - Atualizar a documentação do projeto. - Atualizar o .github/workflows/quality-gate.yml para usar os pipelines genéricos do github; From f3c88885ea57423cd31a4d96436cd1ae7d2f4e3a Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Tue, 25 Nov 2025 00:44:55 -0300 Subject: [PATCH 04/16] SIENTIAPDE-1350: Configure cleanup test script and improve test isolation. This commit configures the cleanup test script to load environment variables from a .env file and use them for Temporal connection. It also adds a fixture to clean up temporary directories created by tests, ensuring better test isolation and preventing potential conflicts. Additionally, it adds the 'uri' property to the MongoDB config. --- scripts/run_cleanup_test.py | 28 +++++++--- .../utils/repository/test_model_repository.py | 53 +++++++++++++++++-- tests/utils/test_connectors_config.py | 2 + tests/worker/test_worker.py | 9 +++- 4 files changed, 79 insertions(+), 13 deletions(-) diff --git a/scripts/run_cleanup_test.py b/scripts/run_cleanup_test.py index 6084daa..16e8a87 100644 --- a/scripts/run_cleanup_test.py +++ b/scripts/run_cleanup_test.py @@ -15,7 +15,8 @@ import os import sys from datetime import timedelta from typing import Any - +from dotenv import load_dotenv +from pathlib import Path from temporalio.client import Client # Ensure project root is on PYTHONPATH when running directly @@ -23,7 +24,14 @@ ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if ROOT_DIR not in sys.path: sys.path.insert(0, ROOT_DIR) -from model_manager.workflows.cleanup_files import CleanupFiles # noqa: E402 +from model_manager.workflows.cleanup_files import CleanupFiles + + +# Carrega variáveis de ambiente do arquivo .env na raiz do projeto +PROJECT_ROOT = Path(__file__).resolve().parent.parent +ENV_PATH = PROJECT_ROOT / '.env' +if ENV_PATH.exists(): + load_dotenv(dotenv_path=ENV_PATH) async def main(argv: list[str]) -> None: @@ -34,11 +42,11 @@ async def main(argv: list[str]) -> None: """ # Config from environment / defaults - temporal_host = os.getenv('TEMPORAL_HOST', 'localhost:37463') - temporal_namespace = os.getenv('TEMPORAL_NAMESPACE', 'model-manager') - task_queue = 'cleanup-queue' - - default_bucket = os.getenv('DEFAULT_CLEANUP_BUCKET', 'model-training') + temporal_host = os.getenv('TEMPORAL_HOST') + temporal_namespace = os.getenv('TEMPORAL_NAMESPACE') + task_queue = os.getenv('CLEANUP_TASK_QUEUE') + default_bucket = os.getenv('DEFAULT_CLEANUP_BUCKET') + use_tls = os.getenv('TEMPORAL_USE_TLS', 'false').lower() == 'true' # Optional CLI: bucket name override bucket_name = default_bucket @@ -46,7 +54,11 @@ async def main(argv: list[str]) -> None: bucket_name = argv[0] print(f"Connecting to Temporal at {temporal_host} (namespace={temporal_namespace})...") - client = await Client.connect(temporal_host, namespace=temporal_namespace) + client = await Client.connect( + target_host=temporal_host, + namespace=temporal_namespace, + tls=use_tls, + ) input_data: dict[str, Any] = { 'bucket_name': bucket_name, diff --git a/tests/utils/repository/test_model_repository.py b/tests/utils/repository/test_model_repository.py index a31cb42..61d1cea 100644 --- a/tests/utils/repository/test_model_repository.py +++ b/tests/utils/repository/test_model_repository.py @@ -1,6 +1,7 @@ """Unit tests for ModelRepository with 100% coverage.""" import os +import shutil from unittest.mock import MagicMock, patch import numpy as np @@ -8,6 +9,29 @@ import pandas as pd import pytest +@pytest.fixture(autouse=True) +def cleanup_temp_directories(): + """Clean up temporary directories after each test.""" + # Get the temp directory path + current_file_dir = os.path.dirname(os.path.abspath(__file__)) + model_manager_dir = os.path.dirname(os.path.dirname(os.path.dirname(current_file_dir))) + temp_dir = os.path.join(model_manager_dir, 'reports', 'temp') + + # Run the test + yield + + # Clean up after test + if os.path.exists(temp_dir): + for item in os.listdir(temp_dir): + item_path = os.path.join(temp_dir, item) + if os.path.isdir(item_path) and item.startswith('test_run_'): + try: + shutil.rmtree(item_path) + except (OSError, PermissionError): + # Ignore cleanup errors + pass + + @pytest.fixture def mock_logger(): """Create a mock logger.""" @@ -79,6 +103,9 @@ def test_save_model_success(mock_model_serving_class, mock_logger, mock_train_re url='http://mlflow.test', username='user', password='pass', logger=mock_logger ) + # Reset mock after initialization to focus on method-specific calls + mock_logger.reset_mock() + repo._get_next_run_name = MagicMock(return_value='test_experiment-1') repo._generate_artifacts = MagicMock(return_value=mock_train_result) repo._save_run = MagicMock() @@ -105,6 +132,9 @@ def test_cleanup_run_directory_exists( url='http://mlflow.test', username='user', password='pass', logger=mock_logger ) + # Reset mock after initialization to focus on method-specific calls + mock_logger.reset_mock() + mock_exists.return_value = True repo.cleanup_run_directory('/tmp/test_run') # noqa: S108 @@ -124,6 +154,9 @@ def test_cleanup_run_directory_not_exists(mock_exists, mock_model_serving_class, url='http://mlflow.test', username='user', password='pass', logger=mock_logger ) + # Reset mock after initialization to focus on method-specific calls + mock_logger.reset_mock() + mock_exists.return_value = False repo.cleanup_run_directory('/tmp/test_run') # noqa: S108 @@ -141,6 +174,9 @@ def test_cleanup_run_directory_empty_path(mock_model_serving_class, mock_logger) url='http://mlflow.test', username='user', password='pass', logger=mock_logger ) + # Reset mock after initialization to focus on method-specific calls + mock_logger.reset_mock() + repo.cleanup_run_directory('') mock_logger.info.assert_called_once_with('No run directory specified, skipping cleanup') @@ -756,9 +792,14 @@ def test_generate_artifacts_no_run_name( ) mock_train_result.run_name = None + mock_exists.return_value = True # Mock reports directory exists - with pytest.raises(ValueError, match='run_name must be set'): - repo._generate_artifacts(mock_train_result) + # Mock _create_run_directory to avoid creating real directories + with patch.object(repo, '_create_run_directory') as mock_create_dir: + mock_create_dir.return_value = '/mock/run/dir' + + with pytest.raises(ValueError, match='run_name must be set'): + repo._generate_artifacts(mock_train_result) @patch('model_manager.utils.repository.model_repository.ModelServing') @@ -800,8 +841,12 @@ def test_generate_artifacts_header_not_found( mock_exists.side_effect = exists_side_effect - with pytest.raises(FileNotFoundError, match='Header file does not exist'): - repo._generate_artifacts(mock_train_result) + # Mock _create_run_directory to avoid creating real directories + with patch.object(repo, '_create_run_directory') as mock_create_dir: + mock_create_dir.return_value = '/mock/run/dir' + + with pytest.raises(FileNotFoundError, match='Header file does not exist'): + repo._generate_artifacts(mock_train_result) @patch('model_manager.utils.repository.model_repository.ModelServing') diff --git a/tests/utils/test_connectors_config.py b/tests/utils/test_connectors_config.py index 15adb7d..2679020 100644 --- a/tests/utils/test_connectors_config.py +++ b/tests/utils/test_connectors_config.py @@ -96,6 +96,7 @@ def test_build_mongo_db_config_with_env_vars(): 'connection_string': 'mongodb://sientia1:sientia1@localhost:27018', 'database_name': 'test_db', 'ttl_index_seconds': 3600, + 'uri': 'localhost:27018', } @@ -109,6 +110,7 @@ def test_build_mongo_db_config_with_defaults(): 'connection_string': 'mongodb://root:wKZDbMNU1c@localhost:27018', 'database_name': 'sientia', 'ttl_index_seconds': 3600, + 'uri': 'localhost:27018', } diff --git a/tests/worker/test_worker.py b/tests/worker/test_worker.py index 0e036c2..1a3d5d0 100644 --- a/tests/worker/test_worker.py +++ b/tests/worker/test_worker.py @@ -209,6 +209,7 @@ async def test_main_successful_startup( mock_build_mongodb.return_value = { 'connection_string': 'mongodb://test', 'database_name': 'test_db', + 'uri': 'localhost:27018', } mock_build_postgres.return_value = {} mock_build_mlflow.return_value = {} @@ -290,6 +291,7 @@ async def test_main_handles_exception( mock_build_mongodb.return_value = { 'connection_string': 'mongodb://test', 'database_name': 'test_db', + 'uri': 'localhost:27018', } mock_build_postgres.return_value = {} mock_build_mlflow.return_value = {} @@ -372,6 +374,7 @@ async def test_main_temporal_client_configuration( mock_build_mongodb.return_value = { 'connection_string': 'mongodb://test', 'database_name': 'test_db', + 'uri': 'localhost:27018', } mock_build_postgres.return_value = {} mock_build_mlflow.return_value = {} @@ -404,7 +407,10 @@ async def test_main_temporal_client_configuration( # Verify Temporal client was configured with correct parameters mock_client_class.connect.assert_called_once_with( - target_host='temporal.example.com:7233', namespace='production', runtime=mock_runtime + target_host='temporal.example.com:7233', + namespace='production', + runtime=mock_runtime, + tls=False, ) @@ -445,6 +451,7 @@ async def test_main_worker_configuration( mock_build_mongodb.return_value = { 'connection_string': 'mongodb://test', 'database_name': 'test_db', + 'uri': 'localhost:27018', } mock_build_postgres.return_value = {} mock_build_mlflow.return_value = {} From d2c4803a7595dcfb3527445588af462c5b7e0f79 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Tue, 25 Nov 2025 16:01:10 -0300 Subject: [PATCH 05/16] SIENTIAPDE-1350: Add unit tests for the Cleanup activity, ensuring 100% code coverage, and update todo-list. --- tests/activities/test_cleanup.py | 554 +++++++++++++++++++++++++++++++ todo-list.txt | 1 + 2 files changed, 555 insertions(+) create mode 100644 tests/activities/test_cleanup.py diff --git a/tests/activities/test_cleanup.py b/tests/activities/test_cleanup.py new file mode 100644 index 0000000..79379ee --- /dev/null +++ b/tests/activities/test_cleanup.py @@ -0,0 +1,554 @@ +"""Unit tests for the Cleanup activity, ensuring 100% code coverage.""" + +import asyncio +import os +import shutil +import tempfile +from datetime import UTC, datetime, timedelta +from importlib import reload +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +# Define mocks at the top level to be accessible by all tests + + +@pytest.fixture +def mock_logger(): + """Fixture for a mock logger.""" + return MagicMock() + + +@pytest.fixture +def mock_notification_handler(): + """Fixture for a mock notification handler.""" + return MagicMock() + + +@pytest.fixture +def mock_metrics_controller(): + """Fixture for a mock metrics controller with async methods.""" + controller = MagicMock() + controller.shutdown = AsyncMock() + controller.emit = AsyncMock() + return controller + + +@pytest.fixture +def mock_storage_repository(): + """Fixture for a mock storage repository.""" + repo = MagicMock() + repo.delete_file = MagicMock() + repo.list_bucket_objects = MagicMock() + return repo + + +@pytest.fixture +def temp_dir(): + """Fixture to create and clean up a temporary directory.""" + path = tempfile.mkdtemp() + yield path + shutil.rmtree(path) + + +# --- Initialization Tests --- + + +def test_cleanup_init_default_values( + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test Cleanup initialization uses default environment values.""" + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + + assert cleanup.retention_hours == 24 + assert cleanup.dry_run is False + assert cleanup.max_keys_cleanup == 1000 + + +def test_cleanup_init_custom_env_values( + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test Cleanup initialization with custom environment values.""" + with patch.dict( + os.environ, + { + 'CLEANUP_RETENTION_HOURS': '48', + 'CLEANUP_DRY_RUN': 'true', + 'MAX_KEYS_CLEANUP': '500', + }, + ): + import model_manager.activities.cleanup + + reload(model_manager.activities.cleanup) + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + + assert cleanup.retention_hours == 48 + assert cleanup.dry_run is True + assert cleanup.max_keys_cleanup == 500 + + +@patch.dict(os.environ, {'CLEANUP_RETENTION_HOURS': 'invalid'}) +def test_cleanup_init_invalid_env_value_raises_error(): + """Test Cleanup module raises ValueError for invalid environment variables on import.""" + import model_manager.activities.cleanup + + with pytest.raises(ValueError): + reload(model_manager.activities.cleanup) + + +# --- MinIO Cleanup Tests --- + + +def test_cleanup_minio_files_missing_bucket_name( + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test cleanup_minio_files raises ValueError if bucket_name is missing.""" + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + cleanup._emit_metrics = AsyncMock() + + with pytest.raises(ValueError, match='bucket_name must be provided'): + asyncio.run(cleanup.cleanup_minio_files({'metadata': {}})) + + +def test_cleanup_minio_files_success_with_deletions( + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test successful deletion of old files from MinIO.""" + import model_manager.activities.cleanup + + reload(model_manager.activities.cleanup) + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + cleanup._emit_metrics = AsyncMock() + + old_ts = int((datetime.now(UTC) - timedelta(hours=48)).timestamp() * 1000) + recent_ts = int((datetime.now(UTC) - timedelta(hours=1)).timestamp() * 1000) + + mock_storage_repository.list_bucket_objects.return_value = [ + f'{old_ts}-old-file.txt', + f'{recent_ts}-recent-file.txt', + 'no-timestamp-file.txt', + ] + + asyncio.run(cleanup.cleanup_minio_files({'bucket_name': 'test-bucket', 'metadata': {}})) + + mock_storage_repository.delete_file.assert_called_once_with( + 'test-bucket', f'{old_ts}-old-file.txt' + ) + cleanup._emit_metrics.assert_called_once() + + +def test_cleanup_minio_files_dry_run( + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test MinIO cleanup in dry_run mode does not delete files.""" + with patch.dict(os.environ, {'CLEANUP_DRY_RUN': 'true'}): + import model_manager.activities.cleanup + + reload(model_manager.activities.cleanup) + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + cleanup._emit_metrics = AsyncMock() + + old_ts = int((datetime.now(UTC) - timedelta(hours=48)).timestamp() * 1000) + mock_storage_repository.list_bucket_objects.return_value = [f'{old_ts}-old-file.txt'] + + asyncio.run(cleanup.cleanup_minio_files({'bucket_name': 'test-bucket', 'metadata': {}})) + + mock_storage_repository.delete_file.assert_not_called() + cleanup._emit_metrics.assert_called_once() + + +def test_cleanup_minio_files_delete_error( + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test error during MinIO file deletion is handled gracefully.""" + import model_manager.activities.cleanup + + reload(model_manager.activities.cleanup) + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + cleanup._emit_metrics = AsyncMock() + cleanup.error = AsyncMock() + + old_ts = int((datetime.now(UTC) - timedelta(hours=48)).timestamp() * 1000) + mock_storage_repository.list_bucket_objects.return_value = [f'{old_ts}-old-file.txt'] + mock_storage_repository.delete_file.side_effect = OSError('Permission Denied') + + asyncio.run(cleanup.cleanup_minio_files({'bucket_name': 'test-bucket', 'metadata': {}})) + + cleanup.error.assert_called_once() + cleanup._emit_metrics.assert_called_once() + + +def test_cleanup_minio_files_exception_handling( + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test exception during MinIO cleanup triggers notification and metrics.""" + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + cleanup.send_notification = AsyncMock() + cleanup._emit_metrics = AsyncMock() + + mock_storage_repository.list_bucket_objects.side_effect = Exception('Connection Error') + + with pytest.raises(Exception, match='Connection Error'): + asyncio.run(cleanup.cleanup_minio_files({'bucket_name': 'test-bucket', 'metadata': {}})) + + cleanup.send_notification.assert_called_once() + cleanup._emit_metrics.assert_called_once() + + +# --- Temp Directory Cleanup Tests --- + + +def test_cleanup_temp_directories_nonexistent_path( + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test temp directory cleanup with a non-existent path.""" + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + cleanup._emit_metrics = AsyncMock() + cleanup.warning = AsyncMock() + + asyncio.run( + cleanup.cleanup_temp_directories({'temp_path': '/nonexistent/path', 'metadata': {}}) + ) + + cleanup.warning.assert_called_once() + cleanup._emit_metrics.assert_called_once() + + +def test_cleanup_temp_directories_success_with_deletions( + temp_dir, + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test successful deletion of old temporary directories.""" + import model_manager.activities.cleanup + + reload(model_manager.activities.cleanup) + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + cleanup._emit_metrics = AsyncMock() + + old_time = (datetime.now() - timedelta(hours=48)).strftime('%Y%m%d_%H%M%S_000000') + old_dir = os.path.join(temp_dir, f'old_dir_{old_time}') + os.makedirs(old_dir) + + recent_time = (datetime.now() - timedelta(hours=1)).strftime('%Y%m%d_%H%M%S_000000') + recent_dir = os.path.join(temp_dir, f'recent_dir_{recent_time}') + os.makedirs(recent_dir) + + asyncio.run(cleanup.cleanup_temp_directories({'temp_path': temp_dir, 'metadata': {}})) + + assert not os.path.exists(old_dir) + assert os.path.exists(recent_dir) + cleanup._emit_metrics.assert_called_once() + + +@patch.dict(os.environ, {'CLEANUP_DRY_RUN': 'true'}) +def test_cleanup_temp_directories_dry_run( + temp_dir, + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test temp directory cleanup in dry_run mode does not delete.""" + import model_manager.activities.cleanup + + reload(model_manager.activities.cleanup) + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + cleanup._emit_metrics = AsyncMock() + + old_time = (datetime.now() - timedelta(hours=48)).strftime('%Y%m%d_%H%M%S_000000') + old_dir = os.path.join(temp_dir, f'old_dir_{old_time}') + os.makedirs(old_dir) + + asyncio.run(cleanup.cleanup_temp_directories({'temp_path': temp_dir, 'metadata': {}})) + + assert os.path.exists(old_dir) + cleanup._emit_metrics.assert_called_once() + + +def test_cleanup_temp_directories_delete_error( + temp_dir, + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test graceful handling of errors during directory deletion.""" + import model_manager.activities.cleanup + + reload(model_manager.activities.cleanup) + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + cleanup._emit_metrics = AsyncMock() + cleanup.error = AsyncMock() + + old_time = (datetime.now() - timedelta(hours=48)).strftime('%Y%m%d_%H%M%S_000000') + old_dir = os.path.join(temp_dir, f'old_dir_{old_time}') + os.makedirs(old_dir) + + with patch('shutil.rmtree', side_effect=OSError('Permission Denied')): + asyncio.run(cleanup.cleanup_temp_directories({'temp_path': temp_dir, 'metadata': {}})) + + cleanup.error.assert_called_once() + cleanup._emit_metrics.assert_called_once() + + +# --- Metrics and Utility Tests --- + + +def test_emit_metrics( + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test that _emit_metrics calls the public emit_metric method.""" + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + cleanup.emit_metric = AsyncMock() + + asyncio.run( + cleanup._emit_metrics( + metadata={'pod_id': 'p1', 'workflow_name': 'wf1'}, + metrics_status='success', + activity_name='test_activity', + emit_workflow_metric=True, + ) + ) + + assert cleanup.emit_metric.call_count == 2 + + +def test_cleanup_temp_directories_with_files_and_unmatched_dirs( + temp_dir, + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test that files and directories with non-matching names are skipped.""" + import model_manager.activities.cleanup + + reload(model_manager.activities.cleanup) + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + cleanup._emit_metrics = AsyncMock() + cleanup.debug = AsyncMock() + + # Create a file and a directory with a non-matching name + with open(os.path.join(temp_dir, 'a_file.txt'), 'w') as f: + f.write('hello') + os.makedirs(os.path.join(temp_dir, 'a_directory_with_no_timestamp')) + + asyncio.run(cleanup.cleanup_temp_directories({'temp_path': temp_dir, 'metadata': {}})) + + # Ensure the debug message for skipping was called for the unmatched directory + cleanup.debug.assert_called_with( + 'Skipping directory without timestamp pattern: a_directory_with_no_timestamp', {} + ) + cleanup._emit_metrics.assert_called_once() + + +def test_cleanup_temp_directories_invalid_timestamp_format( + temp_dir, + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test that a directory with an invalid timestamp format is handled correctly.""" + import model_manager.activities.cleanup + + reload(model_manager.activities.cleanup) + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + cleanup._emit_metrics = AsyncMock() + cleanup.error = AsyncMock() + + # Create a directory with a malformed timestamp that matches the regex but fails parsing + malformed_dir_name = 'dir_20239999_999999_999999' + os.makedirs(os.path.join(temp_dir, malformed_dir_name)) + + asyncio.run(cleanup.cleanup_temp_directories({'temp_path': temp_dir, 'metadata': {}})) + + cleanup.error.assert_called_once() + cleanup._emit_metrics.assert_called_once() + + +def test_cleanup_temp_directories_generic_exception( + temp_dir, + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test that a generic exception during directory cleanup is handled.""" + import model_manager.activities.cleanup + + reload(model_manager.activities.cleanup) + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + cleanup._emit_metrics = AsyncMock() + cleanup.send_notification = AsyncMock() + + with patch('os.listdir', side_effect=Exception('Unexpected OS Error')): + with pytest.raises(Exception, match='Unexpected OS Error'): + asyncio.run(cleanup.cleanup_temp_directories({'temp_path': temp_dir, 'metadata': {}})) + + cleanup.send_notification.assert_called_once() + cleanup._emit_metrics.assert_called_once() + + +def test_emit_metrics_activity_only( + mock_storage_repository, + mock_logger, + mock_notification_handler, + mock_metrics_controller, +): + """Test that _emit_metrics can emit only the activity metric.""" + from model_manager.activities.cleanup import Cleanup + + cleanup = Cleanup( + storage_repository=mock_storage_repository, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + cleanup.emit_metric = AsyncMock() + + asyncio.run( + cleanup._emit_metrics( + metadata={'pod_id': 'p1', 'workflow_name': 'wf1'}, + metrics_status='success', + activity_name='test_activity', + emit_workflow_metric=False, + ) + ) + + cleanup.emit_metric.assert_called_once() diff --git a/todo-list.txt b/todo-list.txt index a2d8850..d6338b8 100644 --- a/todo-list.txt +++ b/todo-list.txt @@ -1,3 +1,4 @@ +- Adcionar configuração do cron job ao helm chart - Refatorar o arquivo .dockerignore para só deixar copiar os arquivos que forem necessários para a execução do container, pois ele está copiando muitos arquivos desnecessários. - atualizar as variáveis de ambiente no helm chart - Criar um gráfico no grafana para cada nova atividade. From d0b5b74f8881ef631ad42f8c02d1e6f679dc2e61 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Tue, 25 Nov 2025 16:04:59 -0300 Subject: [PATCH 06/16] SIENTIAPDE-1350: Add tests for close method and paginated list_bucket_objects --- .../repository/test_storage_repository.py | 49 +++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/tests/utils/repository/test_storage_repository.py b/tests/utils/repository/test_storage_repository.py index d8e22ee..3aaeec7 100644 --- a/tests/utils/repository/test_storage_repository.py +++ b/tests/utils/repository/test_storage_repository.py @@ -462,3 +462,52 @@ def test_fetch_file_logs_file_size(mock_boto3, mock_logger, storage_config): # Verify logging includes file size log_calls = [str(call) for call in mock_logger.info.call_args_list] assert any('12345 bytes' in str(call) for call in log_calls) + + +@patch('model_manager.utils.repository.storage_repository.boto3') +def test_close_method(mock_boto3, mock_logger, storage_config): + """Test that the close method calls the underlying client's close method.""" + from model_manager.utils.repository.storage_repository import StorageRepository + + mock_s3_client = Mock() + mock_boto3.client.return_value = mock_s3_client + + repo = StorageRepository(logger=mock_logger, **storage_config) + repo.close() + + mock_s3_client.close.assert_called_once() + mock_logger.info.assert_called_with('MinIO client closed') + + +@patch('model_manager.utils.repository.storage_repository.boto3') +def test_list_bucket_objects_with_pagination(mock_boto3, mock_logger, storage_config): + """Test list_bucket_objects with a paginated response.""" + from model_manager.utils.repository.storage_repository import StorageRepository + + mock_s3_client = Mock() + mock_paginator = Mock() + page1 = { + 'Contents': [ + {'Key': 'file1.txt'}, + {'Key': 'file2.txt'}, + ] + } + page2 = { + 'Contents': [ + {'Key': 'file3.txt'}, + ] + } + page3 = {} + + mock_paginator.paginate.return_value = [page1, page2, page3] + mock_s3_client.get_paginator.return_value = mock_paginator + mock_boto3.client.return_value = mock_s3_client + + repo = StorageRepository(logger=mock_logger, **storage_config) + objects = repo.list_bucket_objects('test-bucket', max_keys=2) + + assert objects == ['file1.txt', 'file2.txt', 'file3.txt'] + assert len(objects) == 3 + mock_s3_client.get_paginator.assert_called_once_with('list_objects_v2') + mock_paginator.paginate.assert_called_once_with(Bucket='test-bucket', MaxKeys=2) + mock_logger.info.assert_any_call('Listed 3 objects from bucket test-bucket') From 33480fa1083a57b1d56d3591b4df89fed8a5089a Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Tue, 25 Nov 2025 16:44:51 -0300 Subject: [PATCH 07/16] SIENTIAPDE-1350: Add unit tests for the CleanupFiles workflow, covering both input bucket and default bucket scenarios. --- tests/workflows/test_cleanup_files.py | 76 +++++++++++++++++++++++++++ 1 file changed, 76 insertions(+) create mode 100644 tests/workflows/test_cleanup_files.py diff --git a/tests/workflows/test_cleanup_files.py b/tests/workflows/test_cleanup_files.py new file mode 100644 index 0000000..28d02e5 --- /dev/null +++ b/tests/workflows/test_cleanup_files.py @@ -0,0 +1,76 @@ +"""Unit tests for the CleanupFiles workflow.""" + +from unittest.mock import AsyncMock, patch + +import pytest + + +@pytest.mark.asyncio +@patch('model_manager.workflows.cleanup_files.workflow') +@patch('model_manager.workflows.cleanup_files.POD_ID', 'temporal-pod') +async def test_cleanup_files_workflow_with_input_bucket(mock_workflow_module): + """Test the CleanupFiles workflow when bucket_name is provided in the input.""" + from model_manager.workflows.cleanup_files import CleanupFiles + + # Mock execute_activity_method + mock_workflow_module.execute_activity_method = AsyncMock() + + # Instantiate and run the workflow + workflow_instance = CleanupFiles() + await workflow_instance.run({'bucket_name': 'input-bucket'}) + + # Verify that the activities were called with the correct parameters + calls = mock_workflow_module.execute_activity_method.call_args_list + assert len(calls) == 2 + + # Check cleanup_minio_files call + minio_call_args = calls[0][0][1] + assert minio_call_args['bucket_name'] == 'input-bucket' + assert minio_call_args['metadata'] == { + 'pod_id': 'temporal-pod', + 'workflow_name': 'cleanup_files', + } + + # Check cleanup_temp_directories call + local_call_args = calls[1][0][1] + assert local_call_args['temp_path'] == 'model_manager/reports/temp' + assert local_call_args['metadata'] == { + 'pod_id': 'temporal-pod', + 'workflow_name': 'cleanup_files', + } + + +@pytest.mark.asyncio +@patch('model_manager.workflows.cleanup_files.workflow') +@patch('model_manager.workflows.cleanup_files.POD_ID', 'temporal-pod') +@patch('model_manager.workflows.cleanup_files.DEFAULT_CLEANUP_BUCKET', 'env-var-bucket') +async def test_cleanup_files_workflow_with_default_bucket(mock_workflow_module): + """Test the CleanupFiles workflow when using the default bucket from environment variables.""" + from model_manager.workflows.cleanup_files import CleanupFiles + + # Mock execute_activity_method + mock_workflow_module.execute_activity_method = AsyncMock() + + # Instantiate and run the workflow + workflow_instance = CleanupFiles() + await workflow_instance.run({}) # Empty input + + # Verify that the activities were called + calls = mock_workflow_module.execute_activity_method.call_args_list + assert len(calls) == 2 + + # Check cleanup_minio_files call + minio_call_args = calls[0][0][1] + assert minio_call_args['bucket_name'] == 'env-var-bucket' + assert minio_call_args['metadata'] == { + 'pod_id': 'temporal-pod', + 'workflow_name': 'cleanup_files', + } + + # Check cleanup_temp_directories call + local_call_args = calls[1][0][1] + assert local_call_args['temp_path'] == 'model_manager/reports/temp' + assert local_call_args['metadata'] == { + 'pod_id': 'temporal-pod', + 'workflow_name': 'cleanup_files', + } From 823df2d65dd70ef4e89a904fbece6d86a41c9bbb Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Tue, 25 Nov 2025 16:54:45 -0300 Subject: [PATCH 08/16] SIENTIAPDE-1350: Refactor: Replace AsyncMock with MagicMock in cleanup tests for synchronous calls --- tests/activities/test_cleanup.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/tests/activities/test_cleanup.py b/tests/activities/test_cleanup.py index 79379ee..b86f0d8 100644 --- a/tests/activities/test_cleanup.py +++ b/tests/activities/test_cleanup.py @@ -226,7 +226,7 @@ def test_cleanup_minio_files_delete_error( metrics_controller=mock_metrics_controller, ) cleanup._emit_metrics = AsyncMock() - cleanup.error = AsyncMock() + cleanup.error = MagicMock() old_ts = int((datetime.now(UTC) - timedelta(hours=48)).timestamp() * 1000) mock_storage_repository.list_bucket_objects.return_value = [f'{old_ts}-old-file.txt'] @@ -253,7 +253,7 @@ def test_cleanup_minio_files_exception_handling( notification_handler=mock_notification_handler, metrics_controller=mock_metrics_controller, ) - cleanup.send_notification = AsyncMock() + cleanup.send_notification = MagicMock() cleanup._emit_metrics = AsyncMock() mock_storage_repository.list_bucket_objects.side_effect = Exception('Connection Error') @@ -284,7 +284,7 @@ def test_cleanup_temp_directories_nonexistent_path( metrics_controller=mock_metrics_controller, ) cleanup._emit_metrics = AsyncMock() - cleanup.warning = AsyncMock() + cleanup.warning = MagicMock() asyncio.run( cleanup.cleanup_temp_directories({'temp_path': '/nonexistent/path', 'metadata': {}}) @@ -382,7 +382,7 @@ def test_cleanup_temp_directories_delete_error( metrics_controller=mock_metrics_controller, ) cleanup._emit_metrics = AsyncMock() - cleanup.error = AsyncMock() + cleanup.error = MagicMock() old_time = (datetime.now() - timedelta(hours=48)).strftime('%Y%m%d_%H%M%S_000000') old_dir = os.path.join(temp_dir, f'old_dir_{old_time}') @@ -447,7 +447,7 @@ def test_cleanup_temp_directories_with_files_and_unmatched_dirs( metrics_controller=mock_metrics_controller, ) cleanup._emit_metrics = AsyncMock() - cleanup.debug = AsyncMock() + cleanup.debug = MagicMock() # Create a file and a directory with a non-matching name with open(os.path.join(temp_dir, 'a_file.txt'), 'w') as f: @@ -483,7 +483,7 @@ def test_cleanup_temp_directories_invalid_timestamp_format( metrics_controller=mock_metrics_controller, ) cleanup._emit_metrics = AsyncMock() - cleanup.error = AsyncMock() + cleanup.error = MagicMock() # Create a directory with a malformed timestamp that matches the regex but fails parsing malformed_dir_name = 'dir_20239999_999999_999999' @@ -515,7 +515,7 @@ def test_cleanup_temp_directories_generic_exception( metrics_controller=mock_metrics_controller, ) cleanup._emit_metrics = AsyncMock() - cleanup.send_notification = AsyncMock() + cleanup.send_notification = MagicMock() with patch('os.listdir', side_effect=Exception('Unexpected OS Error')): with pytest.raises(Exception, match='Unexpected OS Error'): From 9ecac9c8fd7d1ba764c56b9678606c19e302f91e Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Tue, 25 Nov 2025 17:03:06 -0300 Subject: [PATCH 09/16] SIENTIAPDE-1350: Update values.yaml with new environment variables and remove outdated task from todo-list.txt --- todo-list.txt | 1 - values.yaml | 19 +++++++++++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/todo-list.txt b/todo-list.txt index d6338b8..2ff06bd 100644 --- a/todo-list.txt +++ b/todo-list.txt @@ -1,6 +1,5 @@ - Adcionar configuração do cron job ao helm chart - Refatorar o arquivo .dockerignore para só deixar copiar os arquivos que forem necessários para a execução do container, pois ele está copiando muitos arquivos desnecessários. -- atualizar as variáveis de ambiente no helm chart - Criar um gráfico no grafana para cada nova atividade. - Atualizar a documentação dos métodos alterados. - Atualizar a documentação do projeto. diff --git a/values.yaml b/values.yaml index 5dca7b5..eeee7ac 100644 --- a/values.yaml +++ b/values.yaml @@ -187,6 +187,12 @@ env: value: "temporal-frontend.temporal.svc.cluster.local:7233" - name: TEMPORAL_NAMESPACE value: "model-manager" + - name: TRAIN_TASK_QUEUE + value: "train_model_queue" + - name: CLEANUP_TASK_QUEUE + value: "cleanup_queue" + - name: TEMPORAL_USE_TLS + value: "false" - name: MONGODB_USERNAME value: "root" @@ -227,6 +233,19 @@ env: - name: TIMEOUT_UPDATE_DATABASE value: "30" + - name: CLEANUP_RETENTION_HOURS + value: "24" + - name: CLEANUP_DRY_RUN + value: "false" + - name: TIMEOUT_CLEANUP_MINIO + value: "300" + - name: TIMEOUT_CLEANUP_LOCAL + value: "120" + - name: MAX_KEYS_CLEANUP + value: "1000" + - name: DEFAULT_CLEANUP_BUCKET + value: "model-training" + - name: EXTRA_PIP_REQUIREMENTS value: "git+https://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@github.com/Aignosi/sientia-mlops-library.git" From b028dc1b7d78fe7b2d7e62b381348104840323d8 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Tue, 25 Nov 2025 18:12:53 -0300 Subject: [PATCH 10/16] SIENTIAPDE-1350: Implement scheduled cleanup workflow using Temporal schedules. Adds schedule creation to worker startup and configures environment variables. --- .env.example | 5 ++ model_manager/schedules/__init__.py | 0 model_manager/schedules/cleanup_schedule.py | 85 +++++++++++++++++++++ model_manager/worker/worker.py | 9 +++ run_local.sh | 4 +- tests/activities/test_cleanup.py | 15 ++++ tests/schedules/__init__.py | 0 tests/worker/test_worker.py | 18 +++-- todo-list.txt | 1 - values.yaml | 10 +++ 10 files changed, 140 insertions(+), 7 deletions(-) create mode 100644 model_manager/schedules/__init__.py create mode 100644 model_manager/schedules/cleanup_schedule.py create mode 100644 tests/schedules/__init__.py diff --git a/.env.example b/.env.example index 6efb9cb..22e8021 100644 --- a/.env.example +++ b/.env.example @@ -49,4 +49,9 @@ TIMEOUT_CLEANUP_LOCAL=120 MAX_KEYS_CLEANUP=1000 DEFAULT_CLEANUP_BUCKET=model-training +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/model_manager/schedules/__init__.py b/model_manager/schedules/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/model_manager/schedules/cleanup_schedule.py b/model_manager/schedules/cleanup_schedule.py new file mode 100644 index 0000000..084051b --- /dev/null +++ b/model_manager/schedules/cleanup_schedule.py @@ -0,0 +1,85 @@ +"""Schedule configuration for cleanup workflow.""" + +import os +from datetime import timedelta + +from sientia_do.observability.logger import Logger as SientiaLogger +from temporalio.client import ( + Client, + Schedule, + ScheduleActionStartWorkflow, + ScheduleSpec, +) + +# Schedule configuration from environment variables +SCHEDULE_ID = os.getenv('CLEANUP_SCHEDULE_ID', 'cleanup-files-daily') +CLEANUP_CRON = os.getenv('CLEANUP_CRON', '0 0 * * *') # Default: midnight UTC +CLEANUP_TIMEZONE = os.getenv('CLEANUP_TIMEZONE', 'UTC') +CLEANUP_TASK_QUEUE = os.getenv('CLEANUP_TASK_QUEUE', 'cleanup_queue') +CLEANUP_EXECUTION_TIMEOUT_HOURS = int(os.getenv('CLEANUP_EXECUTION_TIMEOUT_HOURS', '1')) + + +async def schedule_exists( + client: Client, schedule_id: str, logger: SientiaLogger, metadata: dict[str, str | None] +) -> bool: + """ + Check if a schedule already exists. + + Args: + client: Temporal client instance + schedule_id: ID of the schedule to check + + Returns: + True if schedule exists, False otherwise + """ + try: + async for schedule in await client.list_schedules(): + if schedule.id == schedule_id: + return True + return False + except Exception as e: # noqa: BLE001 + logger.custom_error(f'Error checking if schedule exists: {e}', metadata) + return False + + +async def create_cleanup_schedule( + client: Client, logger: SientiaLogger, metadata: dict[str, str | None] +) -> None: + """ + Create or update the cleanup files schedule. + + This function is idempotent and can be called multiple times safely. + It will only create the schedule if it doesn't already exist. + + Args: + client: Temporal client instance + """ + # Check if schedule already exists + if await schedule_exists(client, SCHEDULE_ID, logger, metadata): + logger.custom_info( + f"Schedule '{SCHEDULE_ID}' already configured, skipping creation", metadata + ) + return + + await client.create_schedule( + SCHEDULE_ID, + Schedule( + action=ScheduleActionStartWorkflow( + 'cleanup_files', + {}, # Empty input, will use default bucket from environment + id=f'cleanup-files-scheduled-{SCHEDULE_ID}', + task_queue=CLEANUP_TASK_QUEUE, + execution_timeout=timedelta(hours=CLEANUP_EXECUTION_TIMEOUT_HOURS), + ), + spec=ScheduleSpec( + cron_expressions=[CLEANUP_CRON], + time_zone_name=CLEANUP_TIMEZONE, + ), + ), + ) + + logger.custom_info( + f"Schedule '{SCHEDULE_ID}' created successfully. " + f'Cleanup will run at: {CLEANUP_CRON} ({CLEANUP_TIMEZONE})', + metadata, + ) diff --git a/model_manager/worker/worker.py b/model_manager/worker/worker.py index 29200aa..4cda4b1 100644 --- a/model_manager/worker/worker.py +++ b/model_manager/worker/worker.py @@ -37,6 +37,7 @@ with workflow.unsafe.imports_passed_through(): from model_manager import metrics from model_manager.activities.activities import Activities + from model_manager.schedules.cleanup_schedule import create_cleanup_schedule from model_manager.utils.connectors_config import ( build_minio_config, build_mlflow_config, @@ -117,6 +118,14 @@ async def main(): logger.custom_info(f'Temporal client initialized at {host}', metadata) + # Create cleanup schedule (idempotent - only creates if doesn't exist) + try: + await create_cleanup_schedule(temporal_client, logger, metadata) + except Exception as e: # noqa: BLE001 + logger.custom_error(f'Failed to configure cleanup schedule: {e}', metadata) + # Don't fail the worker startup if schedule creation fails + # The schedule can be created manually if needed + workers = [ Worker( temporal_client, diff --git a/run_local.sh b/run_local.sh index 4c88728..4fa3fba 100755 --- a/run_local.sh +++ b/run_local.sh @@ -4,7 +4,9 @@ set -e if [ -f .env ]; then - export $(cat .env | grep -v '^#' | xargs) + set -a + source <(cat .env | grep -v '^#' | grep -v '^$') + set +a echo "Environment variables loaded from .env" else echo "Warning: .env file not found. Continuing without environment variables." diff --git a/tests/activities/test_cleanup.py b/tests/activities/test_cleanup.py index b86f0d8..2d8f7b3 100644 --- a/tests/activities/test_cleanup.py +++ b/tests/activities/test_cleanup.py @@ -54,6 +54,14 @@ def temp_dir(): # --- Initialization Tests --- +@patch.dict( + 'model_manager.activities.cleanup.os.environ', + { + 'CLEANUP_RETENTION_HOURS': '24', + 'CLEANUP_DRY_RUN': 'false', + 'MAX_KEYS_CLEANUP': '1000', + }, +) def test_cleanup_init_default_values( mock_storage_repository, mock_logger, @@ -61,6 +69,9 @@ def test_cleanup_init_default_values( mock_metrics_controller, ): """Test Cleanup initialization uses default environment values.""" + import model_manager.activities.cleanup + + reload(model_manager.activities.cleanup) from model_manager.activities.cleanup import Cleanup cleanup = Cleanup( @@ -140,6 +151,7 @@ def test_cleanup_minio_files_missing_bucket_name( asyncio.run(cleanup.cleanup_minio_files({'metadata': {}})) +@patch.dict('model_manager.activities.cleanup.os.environ', {'CLEANUP_DRY_RUN': 'false'}) def test_cleanup_minio_files_success_with_deletions( mock_storage_repository, mock_logger, @@ -207,6 +219,7 @@ def test_cleanup_minio_files_dry_run( cleanup._emit_metrics.assert_called_once() +@patch.dict('model_manager.activities.cleanup.os.environ', {'CLEANUP_DRY_RUN': 'false'}) def test_cleanup_minio_files_delete_error( mock_storage_repository, mock_logger, @@ -294,6 +307,7 @@ def test_cleanup_temp_directories_nonexistent_path( cleanup._emit_metrics.assert_called_once() +@patch.dict('model_manager.activities.cleanup.os.environ', {'CLEANUP_DRY_RUN': 'false'}) def test_cleanup_temp_directories_success_with_deletions( temp_dir, mock_storage_repository, @@ -362,6 +376,7 @@ def test_cleanup_temp_directories_dry_run( cleanup._emit_metrics.assert_called_once() +@patch.dict('model_manager.activities.cleanup.os.environ', {'CLEANUP_DRY_RUN': 'false'}) def test_cleanup_temp_directories_delete_error( temp_dir, mock_storage_repository, diff --git a/tests/schedules/__init__.py b/tests/schedules/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/worker/test_worker.py b/tests/worker/test_worker.py index 1a3d5d0..1b81da1 100644 --- a/tests/worker/test_worker.py +++ b/tests/worker/test_worker.py @@ -335,6 +335,7 @@ async def test_main_handles_exception( @pytest.mark.asyncio +@patch('model_manager.worker.worker.create_cleanup_schedule') @patch('model_manager.worker.worker.Worker') @patch('model_manager.worker.worker.client.Client') @patch('model_manager.worker.worker.Runtime') @@ -360,11 +361,14 @@ async def test_main_temporal_client_configuration( mock_runtime_class, mock_client_class, mock_worker_class, + mock_create_cleanup_schedule, mock_logger, ): """Test that Temporal client is configured correctly.""" from model_manager.worker.worker import main + mock_create_cleanup_schedule.return_value = AsyncMock() + with patch.dict( os.environ, {'TEMPORAL_HOST': 'temporal.example.com:7233', 'TEMPORAL_NAMESPACE': 'production'}, @@ -410,11 +414,12 @@ async def test_main_temporal_client_configuration( target_host='temporal.example.com:7233', namespace='production', runtime=mock_runtime, - tls=False, + tls=True, ) @pytest.mark.asyncio +@patch('model_manager.worker.worker.create_cleanup_schedule') @patch('model_manager.worker.worker.Worker') @patch('model_manager.worker.worker.client.Client') @patch('model_manager.worker.worker.Runtime') @@ -440,12 +445,15 @@ async def test_main_worker_configuration( mock_runtime_class, mock_client_class, mock_worker_class, + mock_create_cleanup_schedule, mock_env_vars, mock_logger, ): """Test that Temporal worker is configured with correct parameters.""" from model_manager.worker.worker import main + mock_create_cleanup_schedule.return_value = AsyncMock() + # Setup mocks mock_get_logger.return_value = mock_logger mock_build_mongodb.return_value = { @@ -488,10 +496,10 @@ async def test_main_worker_configuration( # Verify Worker was created with correct configuration assert mock_worker_class.call_count == 2 - # Primeira chamada: worker de treinamento (train_model-queue) + # Primeira chamada: worker de treinamento (train_model-local_queue) train_call_args = mock_worker_class.call_args_list[0] assert train_call_args[0][0] == mock_client_instance # temporal_client - assert train_call_args[1]['task_queue'] == 'train_model-queue' + assert train_call_args[1]['task_queue'] == 'train_model-local_queue' assert train_call_args[1]['max_concurrent_workflow_tasks'] == 10 assert train_call_args[1]['max_concurrent_activities'] == 10 assert train_call_args[1]['max_concurrent_local_activities'] == 10 @@ -503,10 +511,10 @@ async def test_main_worker_configuration( assert mock_activities.train_model in train_activities_list assert mock_activities.cleanup_resources in train_activities_list - # Segunda chamada: worker de cleanup (cleanup-queue) + # Segunda chamada: worker de cleanup (cleanup-local_queue) cleanup_call_args = mock_worker_class.call_args_list[1] assert cleanup_call_args[0][0] == mock_client_instance # temporal_client - assert cleanup_call_args[1]['task_queue'] == 'cleanup-queue' + assert cleanup_call_args[1]['task_queue'] == 'cleanup-local_queue' assert cleanup_call_args[1]['max_concurrent_workflow_tasks'] == 20 assert cleanup_call_args[1]['max_concurrent_activities'] == 20 assert cleanup_call_args[1]['max_concurrent_local_activities'] == 20 diff --git a/todo-list.txt b/todo-list.txt index 2ff06bd..d8c18c2 100644 --- a/todo-list.txt +++ b/todo-list.txt @@ -1,4 +1,3 @@ -- Adcionar configuração do cron job ao helm chart - Refatorar o arquivo .dockerignore para só deixar copiar os arquivos que forem necessários para a execução do container, pois ele está copiando muitos arquivos desnecessários. - Criar um gráfico no grafana para cada nova atividade. - Atualizar a documentação dos métodos alterados. diff --git a/values.yaml b/values.yaml index eeee7ac..e488d00 100644 --- a/values.yaml +++ b/values.yaml @@ -246,6 +246,16 @@ env: - name: DEFAULT_CLEANUP_BUCKET value: "model-training" + # Cleanup Schedule Configuration + - name: CLEANUP_SCHEDULE_ID + value: "cleanup-files-daily" + - name: CLEANUP_CRON + value: "0 0 * * *" # Midnight UTC + - name: CLEANUP_TIMEZONE + value: "UTC" + - name: CLEANUP_EXECUTION_TIMEOUT_HOURS + value: "1" + - name: EXTRA_PIP_REQUIREMENTS value: "git+https://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@github.com/Aignosi/sientia-mlops-library.git" From bc917e97d84f61a81ab4aa4dedc7432969d12453 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Tue, 25 Nov 2025 19:21:55 -0300 Subject: [PATCH 11/16] SIENTIAPDE-1350: Add cleanup schedule tests and configure worker with task queues from env vars --- tests/schedules/test_cleanup_schedule.py | 362 +++++++++++++++++++++++ tests/worker/test_worker.py | 115 +++---- 2 files changed, 423 insertions(+), 54 deletions(-) create mode 100644 tests/schedules/test_cleanup_schedule.py diff --git a/tests/schedules/test_cleanup_schedule.py b/tests/schedules/test_cleanup_schedule.py new file mode 100644 index 0000000..72ba651 --- /dev/null +++ b/tests/schedules/test_cleanup_schedule.py @@ -0,0 +1,362 @@ +"""Tests for cleanup schedule management.""" + +import os +from datetime import timedelta +from importlib import reload +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + + +@pytest.fixture +def mock_temporal_client(): + """Fixture for a mock Temporal client.""" + client = AsyncMock() + client.list_schedules = AsyncMock() + client.create_schedule = AsyncMock() + return client + + +@pytest.fixture +def mock_logger(): + """Fixture for a mock Sientia logger.""" + logger = MagicMock() + logger.custom_info = MagicMock() + logger.custom_error = MagicMock() + return logger + + +@pytest.fixture +def metadata(): + """Fixture for metadata dict.""" + return {'pod_id': 'test-pod', 'project_name': 'test-project'} + + +# --- schedule_exists Tests --- + + +@pytest.mark.asyncio +async def test_schedule_exists_returns_true_when_schedule_found( + mock_temporal_client, mock_logger, metadata +): + """Test that schedule_exists returns True when schedule is found.""" + from model_manager.schedules.cleanup_schedule import schedule_exists + + # Mock schedule list with matching schedule + mock_schedule = MagicMock() + mock_schedule.id = 'test-schedule-id' + + async def mock_list_schedules(): + yield mock_schedule + + mock_temporal_client.list_schedules.return_value = mock_list_schedules() + + result = await schedule_exists(mock_temporal_client, 'test-schedule-id', mock_logger, metadata) + + assert result is True + mock_temporal_client.list_schedules.assert_called_once() + + +@pytest.mark.asyncio +async def test_schedule_exists_returns_false_when_schedule_not_found( + mock_temporal_client, mock_logger, metadata +): + """Test that schedule_exists returns False when schedule is not found.""" + from model_manager.schedules.cleanup_schedule import schedule_exists + + # Mock empty schedule list + async def mock_list_schedules(): + return + yield # Make it an async generator + + mock_temporal_client.list_schedules.return_value = mock_list_schedules() + + result = await schedule_exists( + mock_temporal_client, 'nonexistent-schedule', mock_logger, metadata + ) + + assert result is False + mock_temporal_client.list_schedules.assert_called_once() + + +@pytest.mark.asyncio +async def test_schedule_exists_returns_false_when_different_schedule_found( + mock_temporal_client, mock_logger, metadata +): + """Test that schedule_exists returns False when only different schedules exist.""" + from model_manager.schedules.cleanup_schedule import schedule_exists + + # Mock schedule list with non-matching schedule + mock_schedule = MagicMock() + mock_schedule.id = 'different-schedule-id' + + async def mock_list_schedules(): + yield mock_schedule + + mock_temporal_client.list_schedules.return_value = mock_list_schedules() + + result = await schedule_exists(mock_temporal_client, 'test-schedule-id', mock_logger, metadata) + + assert result is False + mock_temporal_client.list_schedules.assert_called_once() + + +@pytest.mark.asyncio +async def test_schedule_exists_handles_exception(mock_temporal_client, mock_logger, metadata): + """Test that schedule_exists handles exceptions gracefully.""" + from model_manager.schedules.cleanup_schedule import schedule_exists + + # Mock list_schedules to raise an exception + mock_temporal_client.list_schedules.side_effect = Exception('Connection error') + + result = await schedule_exists(mock_temporal_client, 'test-schedule-id', mock_logger, metadata) + + assert result is False + mock_logger.custom_error.assert_called_once() + assert 'Error checking if schedule exists' in mock_logger.custom_error.call_args[0][0] + + +# --- create_cleanup_schedule Tests --- + + +@pytest.mark.asyncio +async def test_create_cleanup_schedule_skips_when_exists( + mock_temporal_client, mock_logger, metadata +): + """Test that create_cleanup_schedule skips creation when schedule already exists.""" + import model_manager.schedules.cleanup_schedule + + reload(model_manager.schedules.cleanup_schedule) + from model_manager.schedules.cleanup_schedule import create_cleanup_schedule + + # Mock schedule already exists + mock_schedule = MagicMock() + mock_schedule.id = 'cleanup-files-daily' + + async def mock_list_schedules(): + yield mock_schedule + + mock_temporal_client.list_schedules.return_value = mock_list_schedules() + + await create_cleanup_schedule(mock_temporal_client, mock_logger, metadata) + + # Verify schedule creation was NOT called + mock_temporal_client.create_schedule.assert_not_called() + + # Verify info log was called + mock_logger.custom_info.assert_called_once() + assert 'already configured' in mock_logger.custom_info.call_args[0][0] + + +@pytest.mark.asyncio +@patch.dict( + 'model_manager.schedules.cleanup_schedule.os.environ', + { + 'CLEANUP_SCHEDULE_ID': 'test-cleanup-schedule', + 'CLEANUP_CRON': '0 2 * * *', + 'CLEANUP_TIMEZONE': 'America/Sao_Paulo', + 'CLEANUP_TASK_QUEUE': 'test-cleanup-queue', + 'CLEANUP_EXECUTION_TIMEOUT_HOURS': '2', + }, +) +async def test_create_cleanup_schedule_creates_with_custom_config( + mock_temporal_client, mock_logger, metadata +): + """Test that create_cleanup_schedule creates schedule with custom configuration.""" + import model_manager.schedules.cleanup_schedule + + reload(model_manager.schedules.cleanup_schedule) + from model_manager.schedules.cleanup_schedule import create_cleanup_schedule + + # Mock schedule does not exist (empty list) + async def mock_list_schedules(): + return + yield # Make it an async generator + + mock_temporal_client.list_schedules.return_value = mock_list_schedules() + + await create_cleanup_schedule(mock_temporal_client, mock_logger, metadata) + + # Verify schedule creation was called + mock_temporal_client.create_schedule.assert_called_once() + + # Verify schedule parameters + call_args = mock_temporal_client.create_schedule.call_args + schedule_id = call_args[0][0] + schedule_obj = call_args[0][1] + + assert schedule_id == 'test-cleanup-schedule' + assert schedule_obj.action.workflow == 'cleanup_files' + assert schedule_obj.action.task_queue == 'test-cleanup-queue' + assert schedule_obj.action.execution_timeout == timedelta(hours=2) + assert schedule_obj.spec.cron_expressions == ['0 2 * * *'] + assert schedule_obj.spec.time_zone_name == 'America/Sao_Paulo' + + # Verify success log was called + assert mock_logger.custom_info.call_count == 1 + assert 'created successfully' in mock_logger.custom_info.call_args[0][0] + + +@pytest.mark.asyncio +@patch.dict( + 'model_manager.schedules.cleanup_schedule.os.environ', + { + 'CLEANUP_SCHEDULE_ID': 'default-schedule', + }, +) +async def test_create_cleanup_schedule_uses_defaults(mock_temporal_client, mock_logger, metadata): + """Test that create_cleanup_schedule uses default values when env vars not set.""" + import model_manager.schedules.cleanup_schedule + + # Remove optional env vars to test defaults + for key in [ + 'CLEANUP_CRON', + 'CLEANUP_TIMEZONE', + 'CLEANUP_TASK_QUEUE', + 'CLEANUP_EXECUTION_TIMEOUT_HOURS', + ]: + os.environ.pop(key, None) + + reload(model_manager.schedules.cleanup_schedule) + from model_manager.schedules.cleanup_schedule import create_cleanup_schedule + + # Mock schedule does not exist (empty list) + async def mock_list_schedules(): + return + yield # Make it an async generator + + mock_temporal_client.list_schedules.return_value = mock_list_schedules() + + await create_cleanup_schedule(mock_temporal_client, mock_logger, metadata) + + # Verify schedule creation was called + mock_temporal_client.create_schedule.assert_called_once() + + # Verify default parameters + call_args = mock_temporal_client.create_schedule.call_args + schedule_obj = call_args[0][1] + + assert schedule_obj.spec.cron_expressions == ['0 0 * * *'] # Default midnight + assert schedule_obj.spec.time_zone_name == 'UTC' # Default UTC + assert schedule_obj.action.task_queue == 'cleanup_queue' # Default queue + assert schedule_obj.action.execution_timeout == timedelta(hours=1) # Default 1 hour + + +@pytest.mark.asyncio +async def test_create_cleanup_schedule_workflow_id_format( + mock_temporal_client, mock_logger, metadata +): + """Test that workflow ID is correctly formatted with schedule ID.""" + import model_manager.schedules.cleanup_schedule + + reload(model_manager.schedules.cleanup_schedule) + from model_manager.schedules.cleanup_schedule import ( + SCHEDULE_ID, + create_cleanup_schedule, + ) + + # Mock schedule does not exist (empty list) + async def mock_list_schedules(): + return + yield # Make it an async generator + + mock_temporal_client.list_schedules.return_value = mock_list_schedules() + + await create_cleanup_schedule(mock_temporal_client, mock_logger, metadata) + + # Verify workflow ID format + call_args = mock_temporal_client.create_schedule.call_args + schedule_obj = call_args[0][1] + + expected_workflow_id = f'cleanup-files-scheduled-{SCHEDULE_ID}' + assert schedule_obj.action.id == expected_workflow_id + + +@pytest.mark.asyncio +async def test_create_cleanup_schedule_empty_workflow_args( + mock_temporal_client, mock_logger, metadata +): + """Test that workflow is created with empty args (uses env defaults).""" + import model_manager.schedules.cleanup_schedule + + reload(model_manager.schedules.cleanup_schedule) + from model_manager.schedules.cleanup_schedule import create_cleanup_schedule + + # Mock schedule does not exist (empty list) + async def mock_list_schedules(): + return + yield # Make it an async generator + + mock_temporal_client.list_schedules.return_value = mock_list_schedules() + + await create_cleanup_schedule(mock_temporal_client, mock_logger, metadata) + + # Verify workflow args are empty (it's a list with one empty dict) + call_args = mock_temporal_client.create_schedule.call_args + schedule_obj = call_args[0][1] + + # The args are passed as positional args, so it's a list with one element + assert schedule_obj.action.args == [{}] + + +# --- Environment Variable Configuration Tests --- + + +@patch.dict( + 'model_manager.schedules.cleanup_schedule.os.environ', + { + 'CLEANUP_SCHEDULE_ID': 'custom-id', + 'CLEANUP_CRON': '30 3 * * 1', + 'CLEANUP_TIMEZONE': 'Europe/London', + 'CLEANUP_TASK_QUEUE': 'custom-queue', + 'CLEANUP_EXECUTION_TIMEOUT_HOURS': '3', + }, +) +def test_environment_variables_loaded_correctly(): + """Test that environment variables are loaded correctly.""" + import model_manager.schedules.cleanup_schedule + + reload(model_manager.schedules.cleanup_schedule) + from model_manager.schedules.cleanup_schedule import ( + CLEANUP_CRON, + CLEANUP_EXECUTION_TIMEOUT_HOURS, + CLEANUP_TASK_QUEUE, + CLEANUP_TIMEZONE, + SCHEDULE_ID, + ) + + assert SCHEDULE_ID == 'custom-id' + assert CLEANUP_CRON == '30 3 * * 1' + assert CLEANUP_TIMEZONE == 'Europe/London' + assert CLEANUP_TASK_QUEUE == 'custom-queue' + assert CLEANUP_EXECUTION_TIMEOUT_HOURS == 3 + + +def test_environment_variables_use_defaults_when_not_set(): + """Test that default values are used when environment variables are not set.""" + import model_manager.schedules.cleanup_schedule + + # Remove all env vars + for key in [ + 'CLEANUP_SCHEDULE_ID', + 'CLEANUP_CRON', + 'CLEANUP_TIMEZONE', + 'CLEANUP_TASK_QUEUE', + 'CLEANUP_EXECUTION_TIMEOUT_HOURS', + ]: + os.environ.pop(key, None) + + reload(model_manager.schedules.cleanup_schedule) + from model_manager.schedules.cleanup_schedule import ( + CLEANUP_CRON, + CLEANUP_EXECUTION_TIMEOUT_HOURS, + CLEANUP_TASK_QUEUE, + CLEANUP_TIMEZONE, + SCHEDULE_ID, + ) + + assert SCHEDULE_ID == 'cleanup-files-daily' + assert CLEANUP_CRON == '0 0 * * *' + assert CLEANUP_TIMEZONE == 'UTC' + assert CLEANUP_TASK_QUEUE == 'cleanup_queue' + assert CLEANUP_EXECUTION_TIMEOUT_HOURS == 1 diff --git a/tests/worker/test_worker.py b/tests/worker/test_worker.py index 1b81da1..4980b0d 100644 --- a/tests/worker/test_worker.py +++ b/tests/worker/test_worker.py @@ -18,6 +18,8 @@ def mock_env_vars(): 'TEMPORAL_HOST': 'localhost:7233', 'TEMPORAL_NAMESPACE': 'test-namespace', 'PROJECT_NAME': 'test-project', + 'TRAIN_TASK_QUEUE': 'train_model-local_queue', + 'CLEANUP_TASK_QUEUE': 'cleanup-local_queue', } with patch.dict(os.environ, env_vars, clear=False): @@ -454,71 +456,76 @@ async def test_main_worker_configuration( mock_create_cleanup_schedule.return_value = AsyncMock() - # Setup mocks - mock_get_logger.return_value = mock_logger - mock_build_mongodb.return_value = { - 'connection_string': 'mongodb://test', - 'database_name': 'test_db', - 'uri': 'localhost:27018', - } - mock_build_postgres.return_value = {} - mock_build_mlflow.return_value = {} - mock_build_minio.return_value = {} + # Patch the task queue constants directly + with ( + patch('model_manager.worker.worker.TRAIN_TASK_QUEUE', 'train_model-local_queue'), + patch('model_manager.worker.worker.CLEANUP_TASK_QUEUE', 'cleanup-local_queue'), + ): + # Setup mocks + mock_get_logger.return_value = mock_logger + mock_build_mongodb.return_value = { + 'connection_string': 'mongodb://test', + 'database_name': 'test_db', + 'uri': 'localhost:27018', + } + mock_build_postgres.return_value = {} + mock_build_mlflow.return_value = {} + mock_build_minio.return_value = {} - mock_notification_handler = Mock() - mock_notification_handler_class.return_value = mock_notification_handler + mock_notification_handler = Mock() + mock_notification_handler_class.return_value = mock_notification_handler - mock_activities = AsyncMock() - mock_activities.update_experiment_run = Mock() - mock_activities.validate_train_params = Mock() - mock_activities.train_model = Mock() - mock_activities.cleanup_resources = Mock() - mock_activities.shutdown = AsyncMock() - mock_activities_class.return_value = mock_activities + mock_activities = AsyncMock() + mock_activities.update_experiment_run = Mock() + mock_activities.validate_train_params = Mock() + mock_activities.train_model = Mock() + mock_activities.cleanup_resources = Mock() + mock_activities.shutdown = AsyncMock() + mock_activities_class.return_value = mock_activities - mock_runtime = Mock() - mock_runtime_class.return_value = mock_runtime + mock_runtime = Mock() + mock_runtime_class.return_value = mock_runtime - mock_client_instance = AsyncMock() - mock_client_class.connect = AsyncMock(return_value=mock_client_instance) + mock_client_instance = AsyncMock() + mock_client_class.connect = AsyncMock(return_value=mock_client_instance) - mock_worker_instance = Mock() - mock_worker_instance.run = AsyncMock(side_effect=asyncio.CancelledError()) - mock_worker_class.return_value = mock_worker_instance + mock_worker_instance = Mock() + mock_worker_instance.run = AsyncMock(side_effect=asyncio.CancelledError()) + mock_worker_class.return_value = mock_worker_instance - mock_app_up = Mock() - mock_metrics.APP_UP.labels.return_value = mock_app_up + mock_app_up = Mock() + mock_metrics.APP_UP.labels.return_value = mock_app_up - # Run main() - with pytest.raises(SystemExit): - await main() + # Run main() + with pytest.raises(SystemExit): + await main() - # Verify Worker was created with correct configuration - assert mock_worker_class.call_count == 2 + # Verify Worker was created with correct configuration + assert mock_worker_class.call_count == 2 - # Primeira chamada: worker de treinamento (train_model-local_queue) - train_call_args = mock_worker_class.call_args_list[0] - assert train_call_args[0][0] == mock_client_instance # temporal_client - assert train_call_args[1]['task_queue'] == 'train_model-local_queue' - assert train_call_args[1]['max_concurrent_workflow_tasks'] == 10 - assert train_call_args[1]['max_concurrent_activities'] == 10 - assert train_call_args[1]['max_concurrent_local_activities'] == 10 - assert train_call_args[1]['max_cached_workflows'] == 100 + # Primeira chamada: worker de treinamento (train_model-local_queue) + train_call_args = mock_worker_class.call_args_list[0] + assert train_call_args[0][0] == mock_client_instance # temporal_client + assert train_call_args[1]['task_queue'] == 'train_model-local_queue' + assert train_call_args[1]['max_concurrent_workflow_tasks'] == 10 + assert train_call_args[1]['max_concurrent_activities'] == 10 + assert train_call_args[1]['max_concurrent_local_activities'] == 10 + assert train_call_args[1]['max_cached_workflows'] == 100 - train_activities_list = train_call_args[1]['activities'] - assert mock_activities.update_experiment_run in train_activities_list - assert mock_activities.validate_train_params in train_activities_list - assert mock_activities.train_model in train_activities_list - assert mock_activities.cleanup_resources in train_activities_list + train_activities_list = train_call_args[1]['activities'] + assert mock_activities.update_experiment_run in train_activities_list + assert mock_activities.validate_train_params in train_activities_list + assert mock_activities.train_model in train_activities_list + assert mock_activities.cleanup_resources in train_activities_list - # Segunda chamada: worker de cleanup (cleanup-local_queue) - cleanup_call_args = mock_worker_class.call_args_list[1] - assert cleanup_call_args[0][0] == mock_client_instance # temporal_client - assert cleanup_call_args[1]['task_queue'] == 'cleanup-local_queue' - assert cleanup_call_args[1]['max_concurrent_workflow_tasks'] == 20 - assert cleanup_call_args[1]['max_concurrent_activities'] == 20 - assert cleanup_call_args[1]['max_concurrent_local_activities'] == 20 - assert cleanup_call_args[1]['max_cached_workflows'] == 100 + # Segunda chamada: worker de cleanup (cleanup-local_queue) + cleanup_call_args = mock_worker_class.call_args_list[1] + assert cleanup_call_args[0][0] == mock_client_instance # temporal_client + assert cleanup_call_args[1]['task_queue'] == 'cleanup-local_queue' + assert cleanup_call_args[1]['max_concurrent_workflow_tasks'] == 20 + assert cleanup_call_args[1]['max_concurrent_activities'] == 20 + assert cleanup_call_args[1]['max_concurrent_local_activities'] == 20 + assert cleanup_call_args[1]['max_cached_workflows'] == 100 @patch('model_manager.worker.worker.asyncio.run') From c2819693816f58e440ac4b1126b1bd174c90d689 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Wed, 26 Nov 2025 11:30:22 -0300 Subject: [PATCH 12/16] SIENTIAPDE-1350: Add test to ensure worker starts despite cleanup schedule creation failure. --- tests/worker/test_worker.py | 94 +++++++++++++++++++++++++++++++++++++ 1 file changed, 94 insertions(+) diff --git a/tests/worker/test_worker.py b/tests/worker/test_worker.py index 4980b0d..7a45d7f 100644 --- a/tests/worker/test_worker.py +++ b/tests/worker/test_worker.py @@ -528,6 +528,100 @@ async def test_main_worker_configuration( assert cleanup_call_args[1]['max_cached_workflows'] == 100 +@pytest.mark.asyncio +@patch('model_manager.worker.worker.create_cleanup_schedule') +@patch('model_manager.worker.worker.Worker') +@patch('model_manager.worker.worker.client.Client') +@patch('model_manager.worker.worker.Runtime') +@patch('model_manager.worker.worker.Activities') +@patch('model_manager.worker.worker.NotificationHandler') +@patch('model_manager.worker.worker.build_mongodb_config') +@patch('model_manager.worker.worker.build_postgres_config') +@patch('model_manager.worker.worker.build_mlflow_config') +@patch('model_manager.worker.worker.build_minio_config') +@patch('model_manager.worker.worker.get_logger') +@patch('model_manager.worker.worker.start_prometheus_server') +@patch('model_manager.worker.worker.metrics') +async def test_main_schedule_creation_failure_does_not_stop_worker( + mock_metrics, + mock_start_prometheus, + mock_get_logger, + mock_build_minio, + mock_build_mlflow, + mock_build_postgres, + mock_build_mongodb, + mock_notification_handler_class, + mock_activities_class, + mock_runtime_class, + mock_client_class, + mock_worker_class, + mock_create_cleanup_schedule, + mock_logger, +): + """Test that schedule creation failure does not prevent worker startup.""" + from model_manager.worker.worker import main + + # Mock schedule creation to raise an exception (as coroutine) + async def mock_schedule_error(*args, **kwargs): + raise Exception('Schedule creation failed') + + mock_create_cleanup_schedule.side_effect = mock_schedule_error + + # Setup mocks + mock_get_logger.return_value = mock_logger + mock_build_mongodb.return_value = { + 'connection_string': 'mongodb://test', + 'database_name': 'test_db', + 'uri': 'localhost:27018', + } + mock_build_postgres.return_value = {} + mock_build_mlflow.return_value = {} + mock_build_minio.return_value = {} + + mock_notification_handler = Mock() + mock_notification_handler.shutdown = Mock() + mock_notification_handler_class.return_value = mock_notification_handler + + mock_activities = AsyncMock() + mock_activities.shutdown = AsyncMock() + mock_activities_class.return_value = mock_activities + + mock_runtime = Mock() + mock_runtime_class.return_value = mock_runtime + + mock_client_instance = AsyncMock() + mock_client_class.connect = AsyncMock(return_value=mock_client_instance) + + mock_worker_instance = Mock() + mock_worker_instance.run = AsyncMock(side_effect=asyncio.CancelledError()) + mock_worker_class.return_value = mock_worker_instance + + mock_app_up = Mock() + mock_metrics.APP_UP.labels.return_value = mock_app_up + + # Run main() - should not fail despite schedule creation error + with pytest.raises(SystemExit): + await main() + + # Verify schedule creation was attempted + mock_create_cleanup_schedule.assert_called_once() + + # Verify error was logged - check all custom_error calls + assert mock_logger.custom_error.call_count >= 1 + + # Find the call that contains the schedule error message + schedule_error_logged = False + for call in mock_logger.custom_error.call_args_list: + if 'Failed to configure cleanup schedule' in call[0][0]: + schedule_error_logged = True + break + + assert schedule_error_logged, 'Schedule creation error should be logged' + + # Verify workers were still created (startup continued) + assert mock_worker_class.call_count == 2 + + @patch('model_manager.worker.worker.asyncio.run') def test_main_entrypoint(mock_asyncio_run): """Test the __main__ entrypoint.""" From 11aaa0780fcb8f322241c930527b39da11c3c558 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Wed, 26 Nov 2025 17:52:44 -0300 Subject: [PATCH 13/16] SIENTIAPDE-1350: Improve Docker image build context and fix queue names This commit refactors the .dockerignore file to use a whitelist approach, significantly reducing the Docker image size by excluding unnecessary files. It also fixes the queue names in values.yaml and adds TEMPORAL_USE_TLS to the test environment. --- .dockerignore | 290 ++++++++++-------------------------- tests/worker/test_worker.py | 6 +- todo-list.txt | 1 - values.yaml | 4 +- 4 files changed, 86 insertions(+), 215 deletions(-) diff --git a/.dockerignore b/.dockerignore index 14aa459..7f786a5 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1,225 +1,93 @@ -# Git -.git -.gitignore -.gitattributes +# ============================================================================ +# WHITELIST APPROACH: Block everything by default, then allow only what's needed +# ============================================================================ -# Documentation -*.md -docs/ -README* +# Block everything first +* -# Tests and development -tests/ -.pytest_cache/ -.coverage -htmlcov/ -.tox/ -.nox/ -.mypy_cache/ -.pyre/ -coverage.xml -*.cover -.hypothesis/ +# ============================================================================ +# ALLOW: Application source code (model_manager package) +# ============================================================================ + +# Allow the main package directory and all Python files +!model_manager/ +!model_manager/**/*.py +!model_manager/**/__init__.py + +# Allow subdirectories structure +!model_manager/activities/ +!model_manager/activities/** +!model_manager/schedules/ +!model_manager/schedules/** +!model_manager/sientia/ +!model_manager/sientia/** +!model_manager/utils/ +!model_manager/utils/** +!model_manager/utils/models/ +!model_manager/utils/models/** +!model_manager/utils/repository/ +!model_manager/utils/repository/** +!model_manager/worker/ +!model_manager/worker/** +!model_manager/workflows/ +!model_manager/workflows/** + +# Allow reports directory with header.html +!model_manager/reports/ +!model_manager/reports/header.html + +# Allow temp directory structure (but not its contents) +!model_manager/reports/temp/ + +# ============================================================================ +# ALLOW: Dependencies file (needed for pip install in Dockerfile) +# ============================================================================ +!requirements.txt + +# ============================================================================ +# BLOCK: Explicitly block unwanted files even if they match above patterns +# ============================================================================ # Python cache and compiled files -__pycache__/ -*.py[cod] -*$py.class -*.so -.Python -build/ -develop-eggs/ -dist/ -downloads/ -eggs/ -.eggs/ -lib/ -lib64/ -parts/ -sdist/ -var/ -wheels/ -*.egg-info/ -.installed.cfg -*.egg -MANIFEST +**/__pycache__/ +**/*.pyc +**/*.pyo +**/*.pyd +**/.Python +**/*.so +**/*.egg +**/*.egg-info/ -# Virtual environments -venv/ -env/ -ENV/ -.venv/ -.env/ +# Tests (not needed in production) +model_manager/**/test_*.py +model_manager/**/*_test.py -# IDE and editors -.vscode/ -.idea/ -*.swp -*.swo -*~ -.DS_Store -Thumbs.db +# IDE and editor files +**/.vscode/ +**/.idea/ +**/*.swp +**/*.swo +**/*~ # OS files -.dockerignore -.dockerignore.dockerignore +**/.DS_Store +**/Thumbs.db -# CI/CD -.github/ -.gitlab-ci.yml -.travis.yml -.circleci/ -Jenkinsfile +# Logs and temporary files +**/*.log +**/*.tmp +**/*.temp # Local configuration -.env -.env.local -.env.*.local -config/local/ -*.local +**/.env +**/.env.local +**/*.local -# Logs -*.log -logs/ -npm-debug.log* -yarn-debug.log* -yarn-error.log* - -# Runtime data -pids/ -*.pid -*.seed -*.pid.lock - -# Temporary files -tmp/ -temp/ -*.tmp -*.temp - -# Node.js (if any frontend tools) -node_modules/ -npm-debug.log* - -# Database -*.db -*.sqlite -*.sqlite3 - -# Jupyter Notebook -.ipynb_checkpoints - -# pyenv -.python-version - -# pipenv -Pipfile.lock - -# PEP 582 -__pypackages__/ - -# Celery -celerybeat-schedule -celerybeat.pid - -# SageMath parsed files -*.sage.py - -# Spyder project settings -.spyderproject -.spyproject - -# Rope project settings -.ropeproject - -# mkdocs documentation -/site +# Documentation inside code +**/*.md +**/README* # Backup files -*.bak -*.backup -*.old - -# Local development scripts -scripts/local/ -dev-* - -# Docker files (excluding the main ones) -docker-compose*.yml -docker-compose*.yaml -Dockerfile.* -!Dockerfile - -# Helm charts (already in .gitignore but reinforcing) -charts/ - -# Kubernetes manifests -k8s/ -kube-* - -# Terraform -*.tfstate -*.tfstate.* -.terraform/ - -# Monitoring and profiling -*.prof -*.profile -.perf - -# Security -*.pem -*.key -*.crt -*.p12 -secrets/ -*.secret - -# Large binaries and datasets -*.bin -*.pkl -*.pickle -*.joblib -data/ -datasets/ -models/pre-trained/ - -# Build artifacts -build/ -dist/ -target/ -out/ - -# Package manager lock files (keeping requirements.txt) -package-lock.json -yarn.lock -Pipfile.lock - -# Local tools -tools/local/ -bin/local/ - -# Cache directories -.cache/ -cache/ - -# Development and configuration files -.env.example -requirements-dev.txt -pyproject.toml -sonar-project.properties -todo-list.txt -validate.sh -run_local.sh -LICENSE - -# Helm charts (development only) -sientia-module/ -*.yaml - -# Local directories -data/ -logs/ -models/ -temp/ -scripts/ +**/*.bak +**/*.backup +**/*.old diff --git a/tests/worker/test_worker.py b/tests/worker/test_worker.py index 7a45d7f..89e373e 100644 --- a/tests/worker/test_worker.py +++ b/tests/worker/test_worker.py @@ -373,7 +373,11 @@ async def test_main_temporal_client_configuration( with patch.dict( os.environ, - {'TEMPORAL_HOST': 'temporal.example.com:7233', 'TEMPORAL_NAMESPACE': 'production'}, + { + 'TEMPORAL_HOST': 'temporal.example.com:7233', + 'TEMPORAL_NAMESPACE': 'production', + 'TEMPORAL_USE_TLS': 'true', + }, ): # Setup mocks mock_get_logger.return_value = mock_logger diff --git a/todo-list.txt b/todo-list.txt index d8c18c2..37d6654 100644 --- a/todo-list.txt +++ b/todo-list.txt @@ -1,4 +1,3 @@ -- Refatorar o arquivo .dockerignore para só deixar copiar os arquivos que forem necessários para a execução do container, pois ele está copiando muitos arquivos desnecessários. - Criar um gráfico no grafana para cada nova atividade. - Atualizar a documentação dos métodos alterados. - Atualizar a documentação do projeto. diff --git a/values.yaml b/values.yaml index e488d00..0d61e4a 100644 --- a/values.yaml +++ b/values.yaml @@ -188,9 +188,9 @@ env: - name: TEMPORAL_NAMESPACE value: "model-manager" - name: TRAIN_TASK_QUEUE - value: "train_model_queue" + value: "train_model-queue" - name: CLEANUP_TASK_QUEUE - value: "cleanup_queue" + value: "cleanup-queue" - name: TEMPORAL_USE_TLS value: "false" From 5dcb62ef4b0f97ede7d7495440365bf749f851de Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Wed, 26 Nov 2025 22:11:20 -0300 Subject: [PATCH 14/16] SIENTIAPDE-1350: Integrate cleanup workflow and update default task queue names. --- model_manager/activities/activities.py | 9 ++++----- model_manager/schedules/cleanup_schedule.py | 6 +++++- model_manager/worker/worker.py | 14 ++++++++++---- tests/schedules/test_cleanup_schedule.py | 4 ++-- 4 files changed, 21 insertions(+), 12 deletions(-) diff --git a/model_manager/activities/activities.py b/model_manager/activities/activities.py index 972c27e..27ad6da 100644 --- a/model_manager/activities/activities.py +++ b/model_manager/activities/activities.py @@ -20,13 +20,12 @@ class Activities(ExperimentTracking, Training, Cleanup): This class combines functionality from multiple activity classes to provide a unified interface for all workflow operations. It manages database connections, - MLFlow model interactions, MinIO storage operations, and data quality validation. + MLFlow model interactions, MinIO storage operations, and cleanup operations. The class implements multiple inheritance to combine specialized functionality: - - ExperimentTracking: ML experiment lifecycle tracking and database operations (extends Postgres) - - MLFlow: Model saving and artifact management operations - - MinIO: Object storage operations (file upload/download/delete) - - Training: ML model training operations (extends BaseActivity) + - ExperimentTracking: ML experiment lifecycle tracking and database operations + - Training: ML model training operations with MLFlow and MinIO integration + - Cleanup: File and directory cleanup operations for MinIO and local filesystem Attributes: postgres_config (dict): PostgreSQL connection configuration diff --git a/model_manager/schedules/cleanup_schedule.py b/model_manager/schedules/cleanup_schedule.py index 084051b..4a54d1f 100644 --- a/model_manager/schedules/cleanup_schedule.py +++ b/model_manager/schedules/cleanup_schedule.py @@ -15,7 +15,7 @@ from temporalio.client import ( SCHEDULE_ID = os.getenv('CLEANUP_SCHEDULE_ID', 'cleanup-files-daily') CLEANUP_CRON = os.getenv('CLEANUP_CRON', '0 0 * * *') # Default: midnight UTC CLEANUP_TIMEZONE = os.getenv('CLEANUP_TIMEZONE', 'UTC') -CLEANUP_TASK_QUEUE = os.getenv('CLEANUP_TASK_QUEUE', 'cleanup_queue') +CLEANUP_TASK_QUEUE = os.getenv('CLEANUP_TASK_QUEUE', 'cleanup-queue') CLEANUP_EXECUTION_TIMEOUT_HOURS = int(os.getenv('CLEANUP_EXECUTION_TIMEOUT_HOURS', '1')) @@ -28,6 +28,8 @@ async def schedule_exists( Args: client: Temporal client instance schedule_id: ID of the schedule to check + logger: Logger instance for error logging + metadata: Metadata dictionary for logging context Returns: True if schedule exists, False otherwise @@ -53,6 +55,8 @@ async def create_cleanup_schedule( Args: client: Temporal client instance + logger: Logger instance for logging schedule operations + metadata: Metadata dictionary for logging context """ # Check if schedule already exists if await schedule_exists(client, SCHEDULE_ID, logger, metadata): diff --git a/model_manager/worker/worker.py b/model_manager/worker/worker.py index 4cda4b1..f5b03b4 100644 --- a/model_manager/worker/worker.py +++ b/model_manager/worker/worker.py @@ -2,9 +2,11 @@ This module provides the main worker implementation for the Sientia DataOps Model Manager system. It orchestrates Temporal workers, manages task queues, and handles the lifecycle of -model training workflows. +model training and cleanup workflows. -The worker supports the train_model-queue task queue for ML model training workflows. +The worker supports two task queues: +- train_model-queue: For ML model training workflows +- cleanup-queue: For file cleanup workflows Key Features: - Automatic scaling with PollerBehaviorAutoscaling @@ -12,14 +14,18 @@ Key Features: - Comprehensive error handling and logging - Graceful shutdown with cleanup - ML model training pipeline orchestration +- Automated cleanup schedule management Environment Variables: - TEMPORAL_HOST: Temporal server address (default: localhost:7233) -- TEMPORAL_NAMESPACE: Temporal namespace (default: model_manager) +- TEMPORAL_NAMESPACE: Temporal namespace (default: model-manager) +- TEMPORAL_USE_TLS: Enable TLS for Temporal connection (default: false) +- TRAIN_TASK_QUEUE: Task queue for training workflows (default: train_model-queue) +- CLEANUP_TASK_QUEUE: Task queue for cleanup workflows (default: cleanup-queue) - POD_ID: Kubernetes pod identifier for metrics - HTTP_METRICS_PORT: Prometheus metrics server port (default: 9090) - HTTP_SDK_METRICS_PORT: Temporal SDK metrics port (default: 9091) -- PROJECT_NAME: Project name for notifications (default: model_manager) +- PROJECT_NAME: Project name for notifications (default: model-manager) """ from temporalio import client, workflow diff --git a/tests/schedules/test_cleanup_schedule.py b/tests/schedules/test_cleanup_schedule.py index 72ba651..3f4277c 100644 --- a/tests/schedules/test_cleanup_schedule.py +++ b/tests/schedules/test_cleanup_schedule.py @@ -238,7 +238,7 @@ async def test_create_cleanup_schedule_uses_defaults(mock_temporal_client, mock_ assert schedule_obj.spec.cron_expressions == ['0 0 * * *'] # Default midnight assert schedule_obj.spec.time_zone_name == 'UTC' # Default UTC - assert schedule_obj.action.task_queue == 'cleanup_queue' # Default queue + assert schedule_obj.action.task_queue == 'cleanup-queue' # Default queue assert schedule_obj.action.execution_timeout == timedelta(hours=1) # Default 1 hour @@ -358,5 +358,5 @@ def test_environment_variables_use_defaults_when_not_set(): assert SCHEDULE_ID == 'cleanup-files-daily' assert CLEANUP_CRON == '0 0 * * *' assert CLEANUP_TIMEZONE == 'UTC' - assert CLEANUP_TASK_QUEUE == 'cleanup_queue' + assert CLEANUP_TASK_QUEUE == 'cleanup-queue' assert CLEANUP_EXECUTION_TIMEOUT_HOURS == 1 From 486e1756c957cbf6a47da5ac4811bbaca6f13766 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Thu, 27 Nov 2025 10:22:38 -0300 Subject: [PATCH 15/16] SIENTIAPDE-1350: Implement automated file cleanup workflow and update documentation. Added a new workflow for automated cleanup of stale files from MinIO and local filesystem, including scheduled execution and configurable retention. Updated the README with detailed information about the new workflow, its configuration, and related metrics. (118 additions, 5 deletions) --- README.md | 175 ++++++++++++++++++++++++++++++++++++++++++++------ todo-list.txt | 2 - 2 files changed, 157 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index d78a3f2..65884e4 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) + - [Cleanup Files Workflow](#cleanup-files-workflow-cleanup_filespy) - [Installation & Setup](#installation--setup) - [Prerequisites](#prerequisites) - [Environment Setup](#environment-setup) @@ -66,11 +67,13 @@ An enterprise-grade ML model training orchestration platform built on Temporal. ### Core Functionality - **ML Model Training Pipeline**: Complete training workflow from validation to deployment using MLFlow +- **Automated File Cleanup**: Scheduled cleanup of stale files from MinIO and local filesystem - **Temporal Workflow Orchestration**: Robust workflow management with granular retry policies and fault tolerance - **Parameter Validation**: Defense-in-depth validation with business rules and type checking - **Experiment Tracking**: Comprehensive status tracking in PostgreSQL database - **Resource Management**: Automatic cleanup of temporary files and storage - **Comprehensive Monitoring**: Prometheus metrics and detailed logging for operational visibility +- **Scheduled Jobs**: Automated daily cleanup with configurable cron schedules ### Advanced Capabilities - **Granular Retry Policies**: Different strategies for network, training, MLFlow, database, and filesystem operations @@ -133,38 +136,49 @@ 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 - - Dedicated task queue: `train_model-queue` for ML model training workflows + - Two dedicated task queues: + - `train_model-queue`: ML model training workflows + - `cleanup-queue`: File cleanup workflows + - Automated cleanup schedule management #### **Workflows (`model_manager/workflows/`)** - **TrainModel**: Complete ML model training pipeline from validation to deployment +- **CleanupFiles**: Automated cleanup of stale files from MinIO and local filesystem - **Key Features**: - Temporal workflow definitions with granular retry policies - Parameter validation with business rules - Comprehensive error handling and status tracking - Configurable timeouts for different operation types - Automatic resource cleanup and management + - Scheduled cleanup jobs with cron expressions #### **Activities (`model_manager/activities/`)** - **Activities**: Main activity orchestrator combining all functionality through multiple inheritance -- **ExperimentTracking**: ML experiment lifecycle tracking and database operations (extends Postgres) +- **ExperimentTracking**: ML experiment lifecycle tracking and database operations - Unified `update_experiment_run()` method for all experiment status updates - Support for three update types: STATUS, STATUS_WITH_ERROR, MODEL_SAVED - Automatic error message truncation (1024 chars) - - Connection pooling and retry logic via Postgres base class -- **Training**: ML model training operations (standalone activity, composition pattern) + - Connection pooling and retry logic +- **Training**: ML model training operations with MLFlow and MinIO integration - Unified `train_model()` method for complete training pipeline - Receives pre-downloaded files (BytesIO) to avoid memory leaks - Returns success/failure status with TrainModelResult or error message - No exception raising on failure - allows workflow to handle errors gracefully - Integration with TrainingRepository for business logic separation -- **MLFlow**: Model saving and artifact management operations -- **MinIO**: Object storage operations for training data management + - MLFlow model saving and artifact management + - MinIO object storage operations +- **Cleanup**: File and directory cleanup operations + - `cleanup_minio_files()`: Removes stale files from MinIO based on timestamp prefixes + - `cleanup_temp_directories()`: Cleans local temporary directories + - Configurable retention period (default: 24 hours) + - Dry-run mode for testing - **Key Features**: - Multiple inheritance pattern for unified activity interface - Parameter validation with business rules - MLFlow integration for model persistence - Comprehensive error handling and notification integration - Experiment tracking with automatic status management + - Timestamp-based file cleanup with regex pattern matching #### **Data Services (`model_manager/utils/`)** - **Connectors Config**: Environment variable-based configuration management @@ -311,6 +325,79 @@ The workflow validates 10 business rules beyond type checking: 5. **target_variable**: Must be in variable_columns 6. **bucket_name, file_name, experiment_name**: Cannot be empty or whitespace +### Cleanup Files Workflow (`cleanup_files.py`) + +The **CleanupFiles** workflow provides automated cleanup of stale files from MinIO storage and local temporary directories. It runs on a scheduled basis (default: daily at midnight UTC) to maintain storage hygiene. + +#### Purpose +- **Storage Management**: Automatic removal of old files from MinIO and local filesystem +- **Retention Policy**: Configurable retention period (default: 24 hours) +- **Scheduled Execution**: Cron-based scheduling for automated cleanup +- **Resource Optimization**: Prevents storage bloat and reduces costs + +#### Execution Flow +1. **Cleanup MinIO Files**: Scan and delete files older than retention period from MinIO bucket +2. **Cleanup Local Directories**: Remove temporary directories older than retention period + +#### Key Features +- **Timestamp-Based Cleanup**: Uses filename/directory timestamps for age determination +- **Pattern Matching**: Regex patterns for MinIO (`timestamp-filename`) and directories (`name_YYYYMMDD_HHMMSS_microseconds`) +- **Configurable Retention**: Environment variable-based retention period +- **Dry-Run Mode**: Test cleanup operations without actual deletion +- **Idempotent**: Safe to run multiple times +- **Error Handling**: Continues cleanup even if individual operations fail + +#### Input Parameters +```json +{ + "bucket_name": "model-training" // Optional, defaults to DEFAULT_CLEANUP_BUCKET env var +} +``` + +#### Schedule Configuration + +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 | +| **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 | +| **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 | +| **Max Keys** | `MAX_KEYS_CLEANUP` | `1000` | MinIO list operation page size | + +#### Architecture Diagram +```mermaid +flowchart TD + A[Scheduled Trigger] --> B[cleanup_minio_files] + B --> C[cleanup_temp_directories] + + B -.-> MinIO[MinIO Storage] + C -.-> FS[Local Filesystem] +``` + +#### Retry Strategies + +| Operation Type | Initial Interval | Max Interval | Backoff | Max Attempts | Use Case | +|---------------|------------------|--------------|---------|--------------|----------| +| **Network** | 1s | 10s | 2.0x | 5 | MinIO operations (transient network errors) | +| **No Retry** | - | - | - | 1 | Local filesystem operations (permanent errors) | + +#### Cleanup Patterns + +**MinIO Files:** +- Pattern: `{timestamp}-{filename}` where timestamp is milliseconds since epoch +- Example: `1638360000000-training_data.csv` +- Retention: Files older than `CLEANUP_RETENTION_HOURS` are deleted + +**Local Directories:** +- Pattern: `{name}_{YYYYMMDD}_{HHMMSS}_{microseconds}` +- Example: `temp_20231201_143052_123456` +- Retention: Directories older than `CLEANUP_RETENTION_HOURS` are deleted + ## Installation & Setup ### Prerequisites @@ -773,11 +860,23 @@ The Model Manager system exposes comprehensive Prometheus metrics for operationa - `app_up`: Application health status (1=healthy, 0=unhealthy) - Labels: `pod_id` +### Workflow Execution Metrics +- `workflow_execution_total`: Total workflow executions + - Labels: `workflow_name`, `status` (success/failure) +- `activity_execution_total`: Total activity executions + - Labels: `activity_name`, `status` (success/failure) + ### Training Metrics - Training success/failure rates through notification system - Model save performance metrics - Experiment status tracking +### Cleanup Metrics +- Cleanup execution success/failure rates +- Number of files deleted from MinIO +- Number of directories cleaned from local filesystem +- Cleanup duration and performance + ## Configuration ### Environment Variables @@ -786,6 +885,9 @@ The Model Manager system exposes comprehensive Prometheus metrics for operationa |----------|-------------|---------|----------| | `TEMPORAL_HOST` | Temporal server address | `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 | | `POSTGRES_HOST` | PostgreSQL hostname | `localhost` | Yes | | `POSTGRES_PORT` | PostgreSQL port | `5432` | Yes | | `POSTGRES_USER` | PostgreSQL username | `sientia` | Yes | @@ -811,6 +913,14 @@ The Model Manager system exposes comprehensive Prometheus metrics for operationa | `MONGODB_PASSWORD` | MongoDB password | `wKZDbMNU1c` | Yes | | `MONGODB_DATABASE_NAME` | 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 | +| `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 | +| `MAX_KEYS_CLEANUP` | MinIO list operation page size | `1000` | No | +| `DEFAULT_CLEANUP_BUCKET` | Default bucket for cleanup | `model-training` | No | | `LOG_LEVEL` | Application log level | `INFO` | No | | `PROJECT_NAME` | Project name for metrics | `model-manager` | No | | `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No | @@ -819,18 +929,24 @@ The Model Manager system exposes comprehensive Prometheus metrics for operationa #### Workflow Activity Timeouts -These timeouts control how long each activity in the training workflow can run before timing out. All values are in seconds and are designed to handle large files (up to 200MB). +These timeouts control how long each activity in workflows can run before timing out. All values are in seconds. + +**Training Workflow Timeouts:** | Variable | Description | Default | Calculation Basis | |----------|-------------|---------|-------------------| | `TIMEOUT_VALIDATE_PARAMS` | Parameter validation timeout | `30` | Fast operation, no I/O | -| `TIMEOUT_DOWNLOAD_FILE` | File download from MinIO timeout | `600` | 200MB @ 1MB/s with 3x buffer (10 min) | -| `TIMEOUT_TRAIN_MODEL` | Model training timeout | `1800` | Large dataset processing (30 min) | -| `TIMEOUT_SAVE_MODEL` | Save model to MLFlow timeout | `300` | Artifact upload and logging (5 min) | -| `TIMEOUT_CLEANUP_DIRECTORY` | Cleanup temporary directory timeout | `60` | Local filesystem operation (1 min) | -| `TIMEOUT_DELETE_FILE` | Delete file from MinIO timeout | `60` | MinIO delete operation (1 min) | +| `TIMEOUT_TRAIN_MODEL` | Model training timeout | `2700` | Large dataset processing (45 min) | +| `TIMEOUT_DELETE_FILE` | Delete file from MinIO timeout | `120` | MinIO delete operation (2 min) | | `TIMEOUT_UPDATE_DATABASE` | Database update timeout | `30` | PostgreSQL update query (30 sec) | +**Cleanup Workflow Timeouts:** + +| Variable | Description | Default | Calculation Basis | +|----------|-------------|---------|-------------------| +| `TIMEOUT_CLEANUP_MINIO` | MinIO cleanup timeout | `300` | Scan and delete multiple files (5 min) | +| `TIMEOUT_CLEANUP_LOCAL` | Local cleanup timeout | `120` | Scan and delete directories (2 min) | + **Note**: These timeouts can be adjusted based on your infrastructure performance and file sizes. If you're processing files larger than 200MB or have slower network/compute resources, increase these values accordingly. ## Development @@ -908,18 +1024,23 @@ model_manager/ │ ├── __init__.py │ ├── activities.py # Main activities orchestrator (combines all activities) │ ├── experiment_tracking.py # Experiment status tracking and database operations -│ ├── training.py # ML model training operations -│ ├── minio.py # MinIO object storage operations -│ └── mlflow.py # MLFlow model saving and artifact management +│ ├── training.py # ML model training operations (includes MLFlow & MinIO) +│ └── cleanup.py # File and directory cleanup operations ├── workflows/ # Temporal workflow definitions │ ├── __init__.py -│ └── train_model.py # Complete ML model training workflow +│ ├── train_model.py # Complete ML model training workflow +│ └── cleanup_files.py # Automated file cleanup workflow +├── schedules/ # Temporal schedule configurations +│ ├── __init__.py +│ └── cleanup_schedule.py # Cleanup schedule creation and management ├── worker/ # Worker implementation │ ├── __init__.py │ └── worker.py # Main worker orchestrator (Temporal client setup) ├── utils/ # Utility functions and helpers │ ├── __init__.py │ ├── connectors_config.py # Environment-based configuration builders +│ ├── exceptions.py # Custom exception definitions +│ ├── logger_helper.py # Logger initialization utilities │ ├── models/ # Data models and schemas │ │ ├── __init__.py │ │ ├── train_model_params.py # Training parameters model @@ -928,7 +1049,19 @@ model_manager/ │ └── repository/ # Data access layer │ ├── __init__.py │ ├── training_repository.py # Training business logic -│ └── model_repository.py # MLFlow artifact management +│ ├── model_repository.py # MLFlow artifact management +│ └── storage_repository.py # MinIO storage operations +├── sientia/ # Sientia-specific implementations +│ ├── __init__.py +│ ├── exceptions.py # Custom exceptions +│ ├── metrics.py # Business metrics +│ ├── models.py # ML model implementations +│ ├── model_serving.py # Model serving utilities +│ ├── reports.py # Report generation +│ └── utils.py # Utility functions +├── reports/ # Report templates and temporary files +│ ├── header.html # HTML report header template +│ └── temp/ # Temporary report files (cleaned up automatically) ├── metrics.py # Prometheus metrics definitions └── __init__.py ``` @@ -994,16 +1127,22 @@ export LOG_LEVEL=DEBUG ### Key Parameters - **Worker Concurrency**: Adjust `max_concurrent_workflow_tasks` and `max_concurrent_activities` + - Training workflows: 10 concurrent tasks/activities + - Cleanup workflows: 20 concurrent tasks/activities - **Connection Pools**: Optimize database connection pool sizes - **Model Retention**: Configure MLFlow model retention based on requirements - **Batch Sizes**: Adjust data processing batch sizes for optimal throughput +- **Cleanup Performance**: Tune `MAX_KEYS_CLEANUP` for MinIO list operation page size ### Scaling Considerations - **Horizontal Scaling**: Deploy multiple worker instances -- **Task Queue Distribution**: Use multiple task queues for different workflow types +- **Task Queue Distribution**: Two dedicated task queues for workflow isolation + - `train_model-queue`: Training workflows + - `cleanup-queue`: Cleanup workflows - **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 ## Contributing diff --git a/todo-list.txt b/todo-list.txt index 37d6654..28d31bc 100644 --- a/todo-list.txt +++ b/todo-list.txt @@ -1,6 +1,4 @@ - Criar um gráfico no grafana para cada nova atividade. -- Atualizar a documentação dos métodos alterados. -- Atualizar a documentação do projeto. - Atualizar o .github/workflows/quality-gate.yml para usar os pipelines genéricos do github; Criar um workflow para fazer o deploy no suse. From 9d3ecc1c2f1b63a07fccde1c7743ce27242539f4 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Thu, 27 Nov 2025 11:00:16 -0300 Subject: [PATCH 16/16] SIENTIAPDE-1350: Simplify exception handling for file and directory deletion in cleanup activity --- model_manager/activities/cleanup.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/model_manager/activities/cleanup.py b/model_manager/activities/cleanup.py index bd1e1ec..1f9654e 100644 --- a/model_manager/activities/cleanup.py +++ b/model_manager/activities/cleanup.py @@ -140,7 +140,7 @@ class Cleanup(SientiaMonitoring): self.storage_repository.delete_file(bucket_name, obj_key) self.info(f'Deleted stale file: {obj_key}', metadata) files_deleted += 1 - except (OSError, PermissionError, ConnectionError) as e: + except OSError as e: error_msg = f'Failed to delete {obj_key}: {str(e)}' errors.append(error_msg) self.error(error_msg, metadata) @@ -258,7 +258,7 @@ class Cleanup(SientiaMonitoring): metadata, ) directories_deleted += 1 - except (OSError, PermissionError) as e: + except OSError as e: error_msg = f'Failed to delete directory {item_name}: {str(e)}' errors.append(error_msg) self.error(error_msg, metadata)