Trace and Monitor Industrial LLM Inference with OpenTelemetry and KServe

                      
                     
"""
Production implementation for tracing and monitoring Industrial LLM Inference using OpenTelemetry and KServe.
Provides secure, scalable operations with robust logging and error handling.
"""
from typing import Dict, Any, List
import os
import logging
import requests
import time
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, ValidationError
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.exporter.otlp.proto.grpc.exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

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

# OpenTelemetry Configuration
tracer_provider = TracerProvider()
span_exporter = OTLPSpanExporter(endpoint=os.getenv('OTLP_ENDPOINT'), insecure=True)
span_processor = BatchSpanProcessor(span_exporter)
tracer_provider.add_span_processor(span_processor)
trace.set_tracer_provider(tracer_provider)

# FastAPI app initialization
app = FastAPI()
FastAPIInstrumentor.instrument_app(app)

class InferenceRequest(BaseModel):
    model_id: str
    input_data: Dict[str, Any]

class Config:
    database_url: str = os.getenv('DATABASE_URL')

async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate request data.
    
    Args:
        data: Input to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'model_id' not in data or 'input_data' not in data:
        raise ValueError('Missing model_id or input_data')
    return True

async def fetch_data(url: str) -> Dict[str, Any]:
    """Fetch data from the specified URL.
    
    Args:
        url: The URL to fetch data from.
    Returns:
        JSON response as a dictionary.
    Raises:
        HTTPException: If the request fails.
    """
    try:
        response = requests.get(url)
        response.raise_for_status()  # Raise an error for bad responses
        return response.json()  # Return parsed JSON
    except requests.RequestException as e:
        logger.error(f'Error fetching data: {e}')
        raise HTTPException(status_code=500, detail='Data fetch error')

async def save_to_db(data: Dict[str, Any]) -> None:
    """Save inference result to the database.
    
    Args:
        data: Data to save.
    Raises:
        Exception: If saving fails.
    """
    # Simulated save operation
    logger.info('Saving data to DB...')
    # Use a real DB call in production

async def normalize_data(data: Dict[str, Any]) -> Dict[str, Any]:
    """Normalize input data for processing.
    
    Args:
        data: Raw input data.
    Returns:
        Normalized data.
    """
    # Example normalization logic
    return {k: v for k, v in data.items() if v is not None}

async def process_batch(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Process a batch of inference requests.
    
    Args:
        data: List of input data dictionaries.
    Returns:
        List of processed results.
    """
    results = []
    for item in data:
        normalized = await normalize_data(item)
        # Simulate processing
        results.append(normalized)
    return results

@app.post('/inference')
async def run_inference(request: InferenceRequest) -> Dict[str, Any]:
    """Run inference based on the request.
    
    Args:
        request: InferenceRequest data model.
    Returns:
        Inference result as a dictionary.
    Raises:
        HTTPException: If any error occurs.
    """
    try:
        await validate_input(request.dict())  # Validate input
        logger.info('Input validated successfully.')  
        inference_results = await process_batch([request.input_data])  # Process data
        await save_to_db(inference_results)  # Save results
        return {'results': inference_results}
    except ValueError as ve:
        logger.warning(f'Validation error: {ve}')
        raise HTTPException(status_code=400, detail=str(ve))
    except Exception as e:
        logger.error(f'Inference error: {e}')
        raise HTTPException(status_code=500, detail='Inference processing failed')

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)
    # Example usage: curl -X POST http://localhost:8000/inference -H 'Content-Type: application/json' -d '{"model_id":"my_model", "input_data":{"key":"value"}}'