gitea_admin/sientia-dataops-model-manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1717: Remove MinIO cleanup functionality and associated components. This change streamlines the cleanup workflow to focus solely on local temporary directories, removes the ModelTrainingError exception, and updates related configurations, documentation, and tests.

Browse Source
This commit is contained in:
Bruno Domingues
2026-03-30 14:14:10 -03:00
parent 63a94dae0a
commit 7a0961f29d
20 changed files with 56 additions and 778 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
model_manager/activities/activities.py
View File

@@ -111,7 +111,6 @@ class Activities(ExperimentTracking, Training, Cleanup):
Cleanup.__init__(
self,
storage_repository=self.storage_repository,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,

122
model_manager/activities/cleanup.py
View File

@@ -1,5 +1,5 @@
"""
Cleanup activities for removing stale files from MinIO and local filesystem.
Cleanup activities for removing stale files from local filesystem.
This module provides activities for cleaning up temporary files and directories
that are older than the configured retention period. It operates independently
@@ -13,7 +13,7 @@ with workflow.unsafe.imports_passed_through():
import re
import shutil
import traceback
from datetime import UTC, datetime, timedelta
from datetime import datetime, timedelta
from typing import Any
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
@@ -23,11 +23,9 @@ with workflow.unsafe.imports_passed_through():
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):
@@ -35,13 +33,11 @@ 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,
@@ -50,135 +46,21 @@ class Cleanup(SientiaMonitoring):
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 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:
"""

26
model_manager/activities/training.py
View File

@@ -19,7 +19,6 @@ with workflow.unsafe.imports_passed_through():
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
from model_manager.metrics import ACTIVITY_EXECUTION_TOTAL, WORKFLOW_EXECUTION_TOTAL
from model_manager.utils.exceptions import ModelTrainingError
from model_manager.utils.models.train_model_params import TrainModelParams
from model_manager.utils.repository.model_repository import ModelRepository
from model_manager.utils.repository.storage_repository import StorageRepository
@@ -143,8 +142,6 @@ class Training(SientiaMonitoring):
train_params = TrainModelParams.from_dict(train_params)
# type: ignore[assignment]
model_trained = False
model_saved = False
metrics_status = 'success'
try:
@@ -157,9 +154,7 @@ class Training(SientiaMonitoring):
train_params, train_result
)
model_trained = True
train_result = self.model_repository.save_model(train_result)
model_saved = True
return {
'run_name': train_result.run_name,
@@ -168,11 +163,7 @@ class Training(SientiaMonitoring):
except Exception as e: # noqa: BLE001
metrics_status = 'error'
error_msg = (
'Error training model - '
f'model_trained={model_trained}, model_saved={model_saved}, '
f'error: {str(e)}'
)
error_msg = f'Error training model - error: {str(e)}'
trace = traceback.format_exc()
@@ -185,10 +176,7 @@ class Training(SientiaMonitoring):
attachment_content=trace,
)
raise ModelTrainingError(
model_trained=model_trained,
model_saved=model_saved,
) from e
raise e
finally:
await self._emit_metrics(
metadata=metadata,
@@ -206,28 +194,20 @@ class Training(SientiaMonitoring):
input_data: Cleanup configuration containing:
- metadata (dict): Workflow execution metadata.
- run_dir (str): Temporary directory to remove.
- bucket_name (str): MinIO bucket of the uploaded file.
- file_name (str): MinIO object key to delete.
Raises:
Exception: If cleanup fails (after sending notification).
"""
metadata = input_data.get('metadata', {})
run_dir = input_data.get('run_dir', '')
bucket_name = input_data.get('bucket_name', '')
file_name = input_data.get('file_name', '')
metrics_status = 'success'
try:
self.model_repository.cleanup_run_directory(run_dir)
self.storage_repository.delete_file(bucket_name, file_name)
except Exception as e: # noqa: BLE001
metrics_status = 'error'
error_msg = (
f'Error cleaning up resources - Run directory: {run_dir}, '
f'File: {bucket_name}/{file_name}, Error: {str(e)}'
)
error_msg = f'Error cleaning up resources - Run directory: {run_dir}, Error: {str(e)}'
trace = traceback.format_exc()