From 5dcb62ef4b0f97ede7d7495440365bf749f851de Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Wed, 26 Nov 2025 22:11:20 -0300 Subject: [PATCH] SIENTIAPDE-1350: Integrate cleanup workflow and update default task queue names. --- model_manager/activities/activities.py | 9 ++++----- model_manager/schedules/cleanup_schedule.py | 6 +++++- model_manager/worker/worker.py | 14 ++++++++++---- tests/schedules/test_cleanup_schedule.py | 4 ++-- 4 files changed, 21 insertions(+), 12 deletions(-) diff --git a/model_manager/activities/activities.py b/model_manager/activities/activities.py index 972c27e..27ad6da 100644 --- a/model_manager/activities/activities.py +++ b/model_manager/activities/activities.py @@ -20,13 +20,12 @@ class Activities(ExperimentTracking, Training, Cleanup): This class combines functionality from multiple activity classes to provide a unified interface for all workflow operations. It manages database connections, - MLFlow model interactions, MinIO storage operations, and data quality validation. + MLFlow model interactions, MinIO storage operations, and cleanup operations. The class implements multiple inheritance to combine specialized functionality: - - ExperimentTracking: ML experiment lifecycle tracking and database operations (extends Postgres) - - MLFlow: Model saving and artifact management operations - - MinIO: Object storage operations (file upload/download/delete) - - Training: ML model training operations (extends BaseActivity) + - ExperimentTracking: ML experiment lifecycle tracking and database operations + - Training: ML model training operations with MLFlow and MinIO integration + - Cleanup: File and directory cleanup operations for MinIO and local filesystem Attributes: postgres_config (dict): PostgreSQL connection configuration diff --git a/model_manager/schedules/cleanup_schedule.py b/model_manager/schedules/cleanup_schedule.py index 084051b..4a54d1f 100644 --- a/model_manager/schedules/cleanup_schedule.py +++ b/model_manager/schedules/cleanup_schedule.py @@ -15,7 +15,7 @@ from temporalio.client import ( SCHEDULE_ID = os.getenv('CLEANUP_SCHEDULE_ID', 'cleanup-files-daily') CLEANUP_CRON = os.getenv('CLEANUP_CRON', '0 0 * * *') # Default: midnight UTC CLEANUP_TIMEZONE = os.getenv('CLEANUP_TIMEZONE', 'UTC') -CLEANUP_TASK_QUEUE = os.getenv('CLEANUP_TASK_QUEUE', 'cleanup_queue') +CLEANUP_TASK_QUEUE = os.getenv('CLEANUP_TASK_QUEUE', 'cleanup-queue') CLEANUP_EXECUTION_TIMEOUT_HOURS = int(os.getenv('CLEANUP_EXECUTION_TIMEOUT_HOURS', '1')) @@ -28,6 +28,8 @@ async def schedule_exists( Args: client: Temporal client instance schedule_id: ID of the schedule to check + logger: Logger instance for error logging + metadata: Metadata dictionary for logging context Returns: True if schedule exists, False otherwise @@ -53,6 +55,8 @@ async def create_cleanup_schedule( Args: client: Temporal client instance + logger: Logger instance for logging schedule operations + metadata: Metadata dictionary for logging context """ # Check if schedule already exists if await schedule_exists(client, SCHEDULE_ID, logger, metadata): diff --git a/model_manager/worker/worker.py b/model_manager/worker/worker.py index 4cda4b1..f5b03b4 100644 --- a/model_manager/worker/worker.py +++ b/model_manager/worker/worker.py @@ -2,9 +2,11 @@ This module provides the main worker implementation for the Sientia DataOps Model Manager system. It orchestrates Temporal workers, manages task queues, and handles the lifecycle of -model training workflows. +model training and cleanup workflows. -The worker supports the train_model-queue task queue for ML model training workflows. +The worker supports two task queues: +- train_model-queue: For ML model training workflows +- cleanup-queue: For file cleanup workflows Key Features: - Automatic scaling with PollerBehaviorAutoscaling @@ -12,14 +14,18 @@ Key Features: - Comprehensive error handling and logging - Graceful shutdown with cleanup - ML model training pipeline orchestration +- Automated cleanup schedule management Environment Variables: - TEMPORAL_HOST: Temporal server address (default: localhost:7233) -- TEMPORAL_NAMESPACE: Temporal namespace (default: model_manager) +- TEMPORAL_NAMESPACE: Temporal namespace (default: model-manager) +- TEMPORAL_USE_TLS: Enable TLS for Temporal connection (default: false) +- TRAIN_TASK_QUEUE: Task queue for training workflows (default: train_model-queue) +- CLEANUP_TASK_QUEUE: Task queue for cleanup workflows (default: cleanup-queue) - POD_ID: Kubernetes pod identifier for metrics - HTTP_METRICS_PORT: Prometheus metrics server port (default: 9090) - HTTP_SDK_METRICS_PORT: Temporal SDK metrics port (default: 9091) -- PROJECT_NAME: Project name for notifications (default: model_manager) +- PROJECT_NAME: Project name for notifications (default: model-manager) """ from temporalio import client, workflow diff --git a/tests/schedules/test_cleanup_schedule.py b/tests/schedules/test_cleanup_schedule.py index 72ba651..3f4277c 100644 --- a/tests/schedules/test_cleanup_schedule.py +++ b/tests/schedules/test_cleanup_schedule.py @@ -238,7 +238,7 @@ async def test_create_cleanup_schedule_uses_defaults(mock_temporal_client, mock_ assert schedule_obj.spec.cron_expressions == ['0 0 * * *'] # Default midnight assert schedule_obj.spec.time_zone_name == 'UTC' # Default UTC - assert schedule_obj.action.task_queue == 'cleanup_queue' # Default queue + assert schedule_obj.action.task_queue == 'cleanup-queue' # Default queue assert schedule_obj.action.execution_timeout == timedelta(hours=1) # Default 1 hour @@ -358,5 +358,5 @@ def test_environment_variables_use_defaults_when_not_set(): assert SCHEDULE_ID == 'cleanup-files-daily' assert CLEANUP_CRON == '0 0 * * *' assert CLEANUP_TIMEZONE == 'UTC' - assert CLEANUP_TASK_QUEUE == 'cleanup_queue' + assert CLEANUP_TASK_QUEUE == 'cleanup-queue' assert CLEANUP_EXECUTION_TIMEOUT_HOURS == 1