SIENTIAPDE-1248: Integrate MinIO for object storage and add related configurations
This commit introduces MinIO integration for object storage within the Model Manager system. It includes: - Added MinIO activity class for file operations (fetch, delete). - Updated Activities orchestrator to include MinIO activities. - Added MinIO configuration builder to utils/connectors_config.py. - Added environment variables for MinIO configuration in .env.example. - Added boto3 and botocore dependencies to requirements.txt. - Added unit tests for MinIO activities.
This commit is contained in:
Bruno Domingues
@@ -8,25 +8,28 @@ with workflow.unsafe.imports_passed_through():
|
||||
from sientia_do.temporal.activities.postgres import Postgres
|
||||
|
||||
from model_manager.activities.gates import Gates
|
||||
from model_manager.activities.minio import MinIO
|
||||
from model_manager.activities.mlflow import MLFlow
|
||||
|
||||
|
||||
class Activities(Postgres, MLFlow, Gates):
|
||||
class Activities(Postgres, MLFlow, MinIO, Gates):
|
||||
"""
|
||||
Main activities orchestrator for the Model Manager system.
|
||||
|
||||
This class combines functionality from multiple activity classes to provide
|
||||
a unified interface for all workflow operations. It manages database connections,
|
||||
MLFlow model interactions, and data quality validation.
|
||||
MLFlow model interactions, MinIO storage operations, and data quality validation.
|
||||
|
||||
The class implements multiple inheritance to combine specialized functionality:
|
||||
- Postgres: Database operations and data persistence
|
||||
- MLFlow: Model inference and transformation operations
|
||||
- MinIO: Object storage operations (file upload/download/delete)
|
||||
- Gates: Data quality validation and filtering mechanisms
|
||||
|
||||
Attributes:
|
||||
postgres_config (dict): PostgreSQL connection configuration
|
||||
mlflow_config (dict): MLFlow server configuration
|
||||
minio_config (dict): MinIO storage configuration
|
||||
logger (Logger): Logging and observability instance
|
||||
notification_handler (NotificationHandler): Notification management instance
|
||||
"""
|
||||
@@ -35,6 +38,7 @@ class Activities(Postgres, MLFlow, Gates):
|
||||
self,
|
||||
postgres_config: dict[str, Any],
|
||||
mlflow_config: dict[str, Any],
|
||||
minio_config: dict[str, Any],
|
||||
logger: Logger,
|
||||
notification_handler: NotificationHandler,
|
||||
):
|
||||
@@ -49,6 +53,8 @@ class Activities(Postgres, MLFlow, Gates):
|
||||
Required keys: host, port, user, password, dbname, min_connections, max_connections
|
||||
mlflow_config: MLFlow server configuration dictionary
|
||||
Required keys: host, port, username, password
|
||||
minio_config: MinIO storage configuration dictionary
|
||||
Required keys: endpoint_url, access_key, secret_key, region, use_ssl
|
||||
logger: Logger instance for observability and debugging
|
||||
notification_handler: Notification handler for alerts and monitoring
|
||||
|
||||
@@ -79,6 +85,21 @@ class Activities(Postgres, MLFlow, Gates):
|
||||
notification_handler=notification_handler,
|
||||
)
|
||||
|
||||
MinIO.__init__(
|
||||
self,
|
||||
endpoint_url=minio_config['endpoint_url'],
|
||||
access_key=minio_config['access_key'],
|
||||
secret_key=minio_config['secret_key'],
|
||||
region=minio_config['region'],
|
||||
use_ssl=minio_config['use_ssl'],
|
||||
max_retry_attempts=minio_config['max_retry_attempts'],
|
||||
retry_mode=minio_config['retry_mode'],
|
||||
connect_timeout=minio_config['connect_timeout'],
|
||||
read_timeout=minio_config['read_timeout'],
|
||||
logger=logger,
|
||||
notification_handler=notification_handler,
|
||||
)
|
||||
|
||||
Gates.__init__(self, logger=logger, notification_handler=notification_handler)
|
||||
|
||||
async def shutdown(self):
|
||||
|
||||
229
model_manager/activities/minio.py
Normal file
229
model_manager/activities/minio.py
Normal file
@@ -0,0 +1,229 @@
|
||||
from temporalio import activity, workflow
|
||||
|
||||
with workflow.unsafe.imports_passed_through():
|
||||
import traceback
|
||||
from io import BytesIO
|
||||
from typing import Any
|
||||
|
||||
import boto3 # type: ignore[import-untyped]
|
||||
from botocore.config import Config # type: ignore[import-untyped]
|
||||
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.temporal.activities.base import BaseActivity
|
||||
|
||||
|
||||
class MinIO(BaseActivity):
|
||||
"""
|
||||
MinIO (S3-compatible) storage activities for file operations.
|
||||
|
||||
This class provides activities for interacting with MinIO object storage,
|
||||
including file download and deletion operations. It handles authentication,
|
||||
connection management, and comprehensive error handling.
|
||||
|
||||
The class implements best practices for S3/MinIO operations:
|
||||
- Connection reuse (boto3 client is thread-safe)
|
||||
- Automatic retry with exponential backoff
|
||||
- Comprehensive error handling and logging
|
||||
- Notification integration for critical errors
|
||||
|
||||
Attributes:
|
||||
endpoint_url (str): MinIO server endpoint URL
|
||||
access_key (str): MinIO access key ID
|
||||
secret_key (str): MinIO secret access key
|
||||
region (str): MinIO region name
|
||||
use_ssl (bool): Whether to use SSL/TLS for connections
|
||||
minio_client: Boto3 S3 client configured for MinIO
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
endpoint_url: str,
|
||||
access_key: str,
|
||||
secret_key: str,
|
||||
region: str,
|
||||
use_ssl: bool,
|
||||
max_retry_attempts: int,
|
||||
retry_mode: str,
|
||||
connect_timeout: int,
|
||||
read_timeout: int,
|
||||
logger: Logger,
|
||||
notification_handler: NotificationHandler,
|
||||
):
|
||||
"""
|
||||
Initialize MinIO activities with server configuration.
|
||||
|
||||
This constructor creates a persistent boto3 S3 client that will be
|
||||
reused across all activity calls. The client is thread-safe and
|
||||
includes automatic retry configuration.
|
||||
|
||||
Args:
|
||||
endpoint_url: MinIO server endpoint URL (e.g., http://localhost:9000)
|
||||
access_key: MinIO access key ID for authentication
|
||||
secret_key: MinIO secret access key for authentication
|
||||
region: MinIO region name (e.g., us-east-1)
|
||||
use_ssl: Whether to use SSL/TLS for connections
|
||||
max_retry_attempts: Maximum number of retry attempts (e.g., 3)
|
||||
retry_mode: Retry mode - standard, legacy, or adaptive (e.g., adaptive)
|
||||
connect_timeout: Connection timeout in seconds (e.g., 10)
|
||||
read_timeout: Read timeout in seconds (e.g., 60)
|
||||
logger: Logger instance for observability and debugging
|
||||
notification_handler: Notification handler for alerts and monitoring
|
||||
|
||||
Raises:
|
||||
Exception: If boto3 client initialization fails
|
||||
"""
|
||||
BaseActivity.__init__(self, logger, notification_handler, set_error_counter=True)
|
||||
self.endpoint_url = endpoint_url
|
||||
self.access_key = access_key
|
||||
self.secret_key = secret_key
|
||||
self.region = region
|
||||
self.use_ssl = use_ssl
|
||||
self.max_retry_attempts = max_retry_attempts
|
||||
self.retry_mode = retry_mode
|
||||
self.connect_timeout = connect_timeout
|
||||
self.read_timeout = read_timeout
|
||||
|
||||
# Configure boto3 with retry strategy
|
||||
# This handles transient network errors and connection issues automatically
|
||||
boto_config = Config(
|
||||
region_name=region,
|
||||
retries={
|
||||
'max_attempts': max_retry_attempts,
|
||||
'mode': retry_mode,
|
||||
},
|
||||
connect_timeout=connect_timeout,
|
||||
read_timeout=read_timeout,
|
||||
)
|
||||
|
||||
try:
|
||||
self.minio_client = boto3.client(
|
||||
's3',
|
||||
endpoint_url=endpoint_url,
|
||||
aws_access_key_id=access_key,
|
||||
aws_secret_access_key=secret_key,
|
||||
config=boto_config,
|
||||
use_ssl=use_ssl,
|
||||
)
|
||||
self.info(f'MinIO client initialized successfully: {endpoint_url}')
|
||||
except Exception as e:
|
||||
error_msg = f'Failed to initialize MinIO client: {str(e)}'
|
||||
self.error(error_msg)
|
||||
raise Exception(error_msg) from e
|
||||
|
||||
@activity.defn(name='fetch_file_from_minio')
|
||||
async def fetch_file_from_minio(self, input_data: dict[str, Any]) -> BytesIO:
|
||||
"""
|
||||
Fetch a file from MinIO and return its content as a BytesIO object.
|
||||
|
||||
This activity downloads a file from a MinIO bucket and returns the
|
||||
content as a BytesIO object, which is a file-like object that can be
|
||||
used directly with many Python libraries (pandas, PIL, etc.).
|
||||
|
||||
The operation includes:
|
||||
1. Input validation
|
||||
2. File download from MinIO
|
||||
3. Content reading and wrapping in BytesIO
|
||||
4. Comprehensive error handling and logging
|
||||
|
||||
Args:
|
||||
input_data: Configuration for file fetch operation
|
||||
Required keys:
|
||||
- metadata (dict): Workflow execution metadata
|
||||
- bucket_name (str): MinIO bucket name
|
||||
- file_name (str): File path/key in the bucket
|
||||
|
||||
Returns:
|
||||
BytesIO: File content as a file-like object
|
||||
|
||||
Raises:
|
||||
Exception: If file fetch fails due to network, permission, or other errors
|
||||
"""
|
||||
metadata = input_data.get('metadata', {})
|
||||
bucket_name = input_data['bucket_name']
|
||||
file_name = input_data['file_name']
|
||||
|
||||
self.info(f'Fetching file from MinIO: {bucket_name}/{file_name}', metadata)
|
||||
|
||||
try:
|
||||
# Download file from MinIO
|
||||
response = self.minio_client.get_object(Bucket=bucket_name, Key=file_name)
|
||||
|
||||
# Read file content
|
||||
with response['Body'] as body:
|
||||
file_content = body.read()
|
||||
|
||||
file_size = len(file_content)
|
||||
self.info(
|
||||
f'File fetched successfully: {bucket_name}/{file_name} ({file_size} bytes)',
|
||||
metadata,
|
||||
)
|
||||
|
||||
return BytesIO(file_content)
|
||||
|
||||
except Exception as e: # noqa: BLE001
|
||||
error_msg = f'Error fetching file from MinIO - Bucket: {bucket_name}, File: {file_name}, Error: {str(e)}'
|
||||
trace = traceback.format_exc()
|
||||
self.send_notification(
|
||||
metadata=metadata,
|
||||
notification_id='FETCH_FILE_FROM_MINIO_ERROR',
|
||||
message=error_msg,
|
||||
block='fetch_file_from_minio',
|
||||
level=NotificationLevel.ERROR,
|
||||
attachment_content=trace,
|
||||
)
|
||||
self.error(trace, metadata=metadata)
|
||||
raise Exception(error_msg) from e
|
||||
|
||||
@activity.defn(name='delete_file_from_minio')
|
||||
async def delete_file_from_minio(self, input_data: dict[str, Any]) -> None:
|
||||
"""
|
||||
Delete a file from MinIO storage.
|
||||
|
||||
This activity removes a file from a MinIO bucket. The operation is
|
||||
idempotent - deleting a non-existent file is considered successful.
|
||||
|
||||
The operation includes:
|
||||
1. Input validation
|
||||
2. File deletion from MinIO
|
||||
3. Comprehensive error handling and logging
|
||||
|
||||
Args:
|
||||
input_data: Configuration for file deletion operation
|
||||
Required keys:
|
||||
- metadata (dict): Workflow execution metadata
|
||||
- bucket_name (str): MinIO bucket name
|
||||
- file_name (str): File path/key to delete
|
||||
|
||||
Returns:
|
||||
None
|
||||
|
||||
Raises:
|
||||
Exception: If file deletion fails due to permission or other errors
|
||||
"""
|
||||
metadata = input_data.get('metadata', {})
|
||||
bucket_name = input_data['bucket_name']
|
||||
file_name = input_data['file_name']
|
||||
|
||||
self.info(f'Deleting file from MinIO: {bucket_name}/{file_name}', metadata)
|
||||
|
||||
try:
|
||||
# Delete file from MinIO
|
||||
# Note: delete_object is idempotent - no error if file doesn't exist
|
||||
self.minio_client.delete_object(Bucket=bucket_name, Key=file_name)
|
||||
|
||||
self.info(f'File deleted successfully: {bucket_name}/{file_name}', metadata)
|
||||
|
||||
except Exception as e: # noqa: BLE001
|
||||
error_msg = f'Error deleting file from MinIO - Bucket: {bucket_name}, File: {file_name}, Error: {str(e)}'
|
||||
trace = traceback.format_exc()
|
||||
self.send_notification(
|
||||
metadata=metadata,
|
||||
notification_id='DELETE_FILE_FROM_MINIO_ERROR',
|
||||
message=error_msg,
|
||||
block='delete_file_from_minio',
|
||||
level=NotificationLevel.ERROR,
|
||||
attachment_content=trace,
|
||||
)
|
||||
self.error(trace, metadata=metadata)
|
||||
raise Exception(error_msg) from e
|
||||
Reference in New Issue
Block a user