SIENTIAPDE-1350: Implement file cleanup workflow and activities, including MinIO and local directory cleanup, configuration, and metrics.
This commit is contained in:
Bruno Domingues
@@ -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.
|
||||
|
||||
338
model_manager/activities/cleanup.py
Normal file
338
model_manager/activities/cleanup.py
Normal file
@@ -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,
|
||||
},
|
||||
)
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(),
|
||||
),
|
||||
|
||||
83
model_manager/workflows/cleanup_files.py
Normal file
83
model_manager/workflows/cleanup_files.py
Normal file
@@ -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),
|
||||
)
|
||||
Reference in New Issue
Block a user