Serve Production Models at Scale with Seldon Core and Prometheus Client

                      
                     
"""
Production implementation for serving models at scale using Seldon Core and Prometheus Client.
Provides secure, scalable operations with robust monitoring.
"""
from typing import Dict, Any, List, Tuple
import os
import logging
import requests
import json
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from contextlib import contextmanager
import time

# Logging setup
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration class for environment variables
class Config:
    prometheus_url: str = os.getenv('PROMETHEUS_URL', 'http://localhost:9090')
    model_service_url: str = os.getenv('MODEL_SERVICE_URL', 'http://localhost:8000')

# Input model for requests
class InputData(BaseModel):
    data: List[Dict[str, Any]] = Field(..., description='Input data for model prediction')

# Function to validate input data
async def validate_input(data: InputData) -> None:
    """Validate request data.
    
    Args:
        data: Input to validate
    Raises:
        ValueError: If validation fails
    """
    if not data.data:
        raise ValueError('Data field cannot be empty')
    logger.info("Input data validated successfully.")

# Function to fetch metrics from Prometheus
async def fetch_metrics(query: str) -> Dict[str, Any]:
    """Fetch metrics from Prometheus.
    
    Args:
        query: Prometheus query string
    Returns:
        Parsed metrics as dictionary
    Raises:
        Exception: If request fails
    """
    try:
        response = requests.get(f'{Config.prometheus_url}/api/v1/query', params={'query': query})
        response.raise_for_status()
        return response.json()
    except Exception as e:
        logger.error(f'Error fetching metrics: {e}')
        raise

# Function to call the model service for predictions
async def call_model_service(data: InputData) -> Dict[str, Any]:
    """Call the model service to get predictions.
    
    Args:
        data: Input data for prediction
    Returns:
        Model predictions
    Raises:
        HTTPException: If service call fails
    """
    try:
        response = requests.post(f'{Config.model_service_url}/predict', json=data.dict())
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        logger.error(f'Error calling model service: {e}')
        raise HTTPException(status_code=500, detail='Model service error')

# Function to aggregate metrics for logging
async def aggregate_metrics(metrics: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Aggregate metrics for logging.
    
    Args:
        metrics: List of metrics to aggregate
    Returns:
        Aggregated metrics
    """
    # Placeholder for aggregation logic
    aggregated = {"count": len(metrics)}
    logger.info(f'Aggregated metrics: {aggregated}')
    return aggregated

# Main FastAPI app setup
app = FastAPI()

@app.post('/predict', response_model=Dict[str, Any])
async def predict(data: InputData) -> Dict[str, Any]:
    """Endpoint for model predictions.
    
    Args:
        data: Input data for prediction
    Returns:
        Model predictions
    Raises:
        HTTPException: If an error occurs
    """
    # Validate input data
    await validate_input(data)
    # Call the model service and fetch metrics
    predictions = await call_model_service(data)
    metrics = await fetch_metrics('model_predictions_total')
    # Aggregate metrics
    await aggregate_metrics(metrics['data'])
    logger.info('Prediction successful')
    return predictions

# Context manager for resource cleanup
@contextmanager
def resource_cleanup():
    try:
        yield
    finally:
        logger.info('Cleaning up resources...')

if __name__ == '__main__':
    # Example usage
    logger.info('Starting the service...')
    with resource_cleanup():
        # Simulated server start
        app.run(host='0.0.0.0', port=8000)
        logger.info('Service is running.')