SIENTIAPDE-1350: Integrate cleanup workflow and update default task queue names.

This commit is contained in:
Bruno Domingues
2025-11-26 22:11:20 -03:00
parent 11aaa0780f
commit 5dcb62ef4b
4 changed files with 21 additions and 12 deletions

View File

@@ -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

View File

@@ -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):

View File

@@ -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

View File

@@ -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