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

Merge branch 'main' into release/SIENTIAPDE-1645

Browse Source
This commit is contained in:
vitor-aignosi
2026-04-06 08:34:15 -03:00
parent bf2b6b3888 dfe05a063a
commit d2bb588739
18 changed files with 85 additions and 676 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
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
@@ -27,7 +27,6 @@ with workflow.unsafe.imports_passed_through():
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 +34,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,
minio_repository: MinioRepository,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
@@ -50,145 +47,21 @@ class Cleanup(SientiaMonitoring):
Initialize Cleanup activity.
Args:
minio_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.minio_repository = minio_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. MinioRepository applies BASE_PREFIX
# internally; we request all objects under that prefix for this bucket.
objects = await self.minio_repository.list_objects(
prefix='',
bucket=bucket_name,
recursive=True,
metadata=metadata,
)
for obj_key in objects:
files_scanned += 1
# Extract timestamp from the filename portion of the object key
filename = obj_key.split('/')[-1]
match = self.minio_timestamp_pattern.match(filename)
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:
await self.minio_repository.delete_file(
object_name=obj_key,
bucket=None,
metadata=metadata,
)
self.info(f'Deleted stale file: {obj_key}', metadata)
files_deleted += 1
except Exception as e: # noqa: BLE001
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:
"""

7
model_manager/activities/training.py
View File

@@ -24,7 +24,6 @@ with workflow.unsafe.imports_passed_through():
from sientia_model.wrappers.sientia_model import SientiaModel
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.data_manager_repository import DataManagerRepository
@@ -261,11 +260,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()

38
model_manager/utils/exceptions.py
View File

@@ -1,38 +0,0 @@
"""
Custom exception types for the Model Manager.
This module defines domain-specific exceptions used across the training
workflow to convey additional context (e.g., flags indicating which steps
completed successfully) without altering control flow semantics.
"""
class ModelTrainingError(Exception):
"""
Exception raised when the model training workflow fails.
This exception carries flags indicating whether the model was trained
and/or saved successfully, enabling the workflow to map errors to
appropriate experiment statuses.
"""
def __init__(self, model_trained: bool, model_saved: bool, message: str | None = None):
"""
Initialize ModelTrainingError with training state flags.
Args:
model_trained: True if the training step completed successfully.
model_saved: True if the model saving step completed successfully.
message: Optional custom error message. If None, a default message
including the state flags is generated.
"""
self.model_trained = model_trained
self.model_saved = model_saved
if message is None:
message = (
'Model training workflow failed '
f'(model_trained={model_trained}, model_saved={model_saved})'
)
super().__init__(message)

10
model_manager/utils/models/experiment_status.py
View File

@@ -14,21 +14,13 @@ class ExperimentStatus(StrEnum):
to maintain compatibility with existing database records and monitoring systems.
Attributes:
ORCHESTRATOR_VALIDATION_ERROR: Error in the parameters validation.
ORCHESTRATOR_WAITING_PROC: Initial status indicating experiment is registered and waiting for processing.
ORCHESTRATOR_VALIDATION_ERROR: Error in the parameters validation.
TRAINING_SUCCESS: Training completed successfully with model and metrics calculated.
TRAINING_ERROR: Training failed due to data issues, model errors, or other exceptions.
TRACKING_SENT: Model successfully saved to MLFlow.
TRACKING_SEND_ERROR: Model saving to MLFlow failed due to connection or serialization errors.
FILE_DELETED: Cleanup completed successfully with all artifacts removed.
FILE_DELETE_ERROR: Cleanup failed due to file system or MinIO errors.
"""
ORCHESTRATOR_VALIDATION_ERROR = 'ORCHESTRATOR_VALIDATION_ERROR'
ORCHESTRATOR_WAITING_PROC = 'ORCHESTRATOR_WAITING_PROC'
TRAINING_SUCCESS = 'TRAINING_SUCCESS'
TRAINING_ERROR = 'TRAINING_ERROR'
TRACKING_SENT = 'TRACKING_SENT'
TRACKING_SEND_ERROR = 'TRACKING_SEND_ERROR'
FILE_DELETED = 'FILE_DELETED'
FILE_DELETE_ERROR = 'FILE_DELETE_ERROR'

1
model_manager/worker/worker.py
View File

@@ -190,7 +190,6 @@ async def main():
main_workflow=CleanupFiles,
other_workflows=[],
activities=[
activities.cleanup_minio_files,
activities.cleanup_temp_directories,
],
temporal_client=temporal_client,

27
model_manager/workflows/cleanup_files.py
View File

@@ -1,5 +1,5 @@
"""
Cleanup workflow for removing stale files from MinIO and local filesystem.
Cleanup workflow for removing 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.
@@ -13,11 +13,9 @@ with workflow.unsafe.imports_passed_through():
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
from model_manager.workflows.train_model import POD_ID, 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')
@@ -26,7 +24,6 @@ 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
@@ -38,17 +35,10 @@ class CleanupFiles:
"""
Execute the cleanup workflow.
This method orchestrates the cleanup of MinIO files and local directories
This method orchestrates the cleanup of 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'
@@ -60,17 +50,6 @@ class CleanupFiles:
}
}
# 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,

8
model_manager/workflows/train_model.py
View File

@@ -20,7 +20,6 @@ with workflow.unsafe.imports_passed_through():
from model_manager.activities.activities import Activities
from model_manager.activities.experiment_tracking import UpdateType
from model_manager.utils.exceptions import ModelTrainingError
from model_manager.utils.models.experiment_status import ExperimentStatus
from model_manager.utils.models.train_model_params import TrainModelParams
@@ -273,7 +272,7 @@ class TrainModel:
metadata=metadata,
experiment_run_id=experiment_run_id,
update_type=UpdateType.MODEL_SAVED,
status=ExperimentStatus.TRACKING_SENT,
status=ExperimentStatus.TRAINING_SUCCESS,
run_name=train_result.get('run_name'),
)
@@ -310,7 +309,7 @@ class TrainModel:
metadata: dict[str, Any],
) -> None:
"""
Cleanup resources and delete file from MinIO.
Cleanup resources.
This method deletes the training file from MinIO. On success, updates
DB status to FILE_DELETED.
@@ -319,9 +318,6 @@ class TrainModel:
Args:
experiment_run_id: Validated experiment run ID
metadata: Workflow execution metadata
Raises:
Exception: If cleanup fails (after updating DB status)
"""
try:
await workflow.execute_activity_method(