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.