""" 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 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 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, }, )