Monitoring & Analytics¶

# health_check.py
import httpx
from sqlalchemy import create_engine, text

async def check_system_health():
    """
    Comprehensive health check for all system components
    """
    health_status = {
        "saas_api": False,
        "litellm_proxy": False,
        "database": False,
        "overall": "unhealthy"
    }

    # 1. Check SaaS API
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.get("http://localhost:8003/health")
            health_status["saas_api"] = response.status_code == 200
    except Exception as e:
        print(f"SaaS API check failed: {e}")

    # 2. Check LiteLLM Proxy
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.get("http://localhost:8002/health")
            health_status["litellm_proxy"] = response.status_code == 200
    except Exception as e:
        print(f"LiteLLM Proxy check failed: {e}")

    # 3. Check Database
    try:
        engine = create_engine("postgresql://...")
        with engine.connect() as conn:
            result = conn.execute(text("SELECT 1"))
            health_status["database"] = result.fetchone()[0] == 1
    except Exception as e:
        print(f"Database check failed: {e}")

    # Overall health
    if all([health_status["saas_api"], health_status["litellm_proxy"], health_status["database"]]):
        health_status["overall"] = "healthy"
    elif any([health_status["saas_api"], health_status["litellm_proxy"], health_status["database"]]):
        health_status["overall"] = "degraded"

    return health_status

# Usage
health = await check_system_health()
print(health)

# services/credit_alerts.py
from sqlalchemy import and_
from models.credits import TeamCredits
from models.organizations import Organization

def check_low_credit_teams(db: Session, threshold_percent: float = 20.0):
    """
    Find teams with credits below threshold percentage
    """
    teams = db.query(TeamCredits).filter(
        and_(
            TeamCredits.credits_allocated > 0,
            (TeamCredits.credits_remaining / TeamCredits.credits_allocated * 100) <= threshold_percent
        )
    ).all()

    alerts = []
    for team in teams:
        org = db.query(Organization).filter(
            Organization.organization_id == team.organization_id
        ).first()

        alerts.append({
            "team_id": team.team_id,
            "organization_id": team.organization_id,
            "organization_name": org.name if org else "Unknown",
            "credits_remaining": team.credits_remaining,
            "credits_allocated": team.credits_allocated,
            "usage_percent": round((team.credits_used / team.credits_allocated * 100), 2)
        })

    return alerts

# Usage
low_credit_teams = check_low_credit_teams(db, threshold_percent=10.0)
for team in low_credit_teams:
    send_alert(team)

# config/alerts.py
ALERT_THRESHOLDS = {
    "credit_low": 10,  # Percentage
    "credit_critical": 0,  # Credits remaining
    "error_rate_high": 5.0,  # Percentage
    "latency_high": 5000,  # Milliseconds
    "cost_spike": 200.0,  # Percent increase
}

ALERT_CHANNELS = {
    "email": ["admin@company.com", "ops@company.com"],
    "slack": "https://hooks.slack.com/services/...",
    "webhook": "https://your-monitoring-system.com/alerts"
}

# services/alert_manager.py
import httpx
from typing import Dict, Any

class AlertManager:
    def __init__(self, config: Dict[str, Any]):
        self.config = config

    async def send_alert(self, alert_type: str, message: str, severity: str = "warning"):
        """
        Send alert to configured channels
        """
        alert_data = {
            "type": alert_type,
            "message": message,
            "severity": severity,
            "timestamp": datetime.utcnow().isoformat()
        }

        # Send to email
        if "email" in self.config:
            await self._send_email(alert_data)

        # Send to Slack
        if "slack" in self.config:
            await self._send_slack(alert_data)

        # Send to webhook
        if "webhook" in self.config:
            await self._send_webhook(alert_data)

    async def _send_slack(self, alert_data: Dict[str, Any]):
        """Send alert to Slack"""
        webhook_url = self.config["slack"]

        payload = {
            "text": f"🚨 {alert_data['severity'].upper()}: {alert_data['message']}",
            "attachments": [
                {
                    "color": self._get_color(alert_data['severity']),
                    "fields": [
                        {"title": "Type", "value": alert_data['type'], "short": True},
                        {"title": "Time", "value": alert_data['timestamp'], "short": True}
                    ]
                }
            ]
        }

        async with httpx.AsyncClient() as client:
            await client.post(webhook_url, json=payload)

    def _get_color(self, severity: str) -> str:
        colors = {
            "info": "#36a64f",
            "warning": "#ff9800",
            "error": "#f44336",
            "critical": "#9c27b0"
        }
        return colors.get(severity, "#808080")

# Usage
alert_manager = AlertManager(ALERT_CHANNELS)

# Low credit alert
if team.credits_remaining <= ALERT_THRESHOLDS["credit_critical"]:
    await alert_manager.send_alert(
        alert_type="credit_critical",
        message=f"Team {team.team_id} has {team.credits_remaining} credits remaining",
        severity="critical"
    )

# scripts/check_alerts.py
import asyncio
from services.alert_manager import AlertManager, ALERT_THRESHOLDS

async def check_all_alerts():
    """
    Run all alert checks
    """
    alert_manager = AlertManager(ALERT_CHANNELS)

    # Check low credits
    low_credit_teams = check_low_credit_teams(db, ALERT_THRESHOLDS["credit_low"])
    for team in low_credit_teams:
        await alert_manager.send_alert(
            "credit_low",
            f"Team {team['team_id']} has {team['credits_remaining']} credits ({team['usage_percent']}% used)",
            "warning"
        )

    # Check high error rates
    error_stats = check_error_rates(db)
    for stat in error_stats:
        if stat['error_rate'] > ALERT_THRESHOLDS["error_rate_high"]:
            await alert_manager.send_alert(
                "error_rate_high",
                f"Job type {stat['job_type']} has {stat['error_rate']}% error rate",
                "error"
            )

    # Check high latency
    latency_stats = check_latency(db)
    for stat in latency_stats:
        if stat['p95_latency'] > ALERT_THRESHOLDS["latency_high"]:
            await alert_manager.send_alert(
                "latency_high",
                f"Model group {stat['model_group']} has {stat['p95_latency']}ms P95 latency",
                "warning"
            )

if __name__ == "__main__":
    asyncio.run(check_all_alerts())

# config/logging.py
import logging
from logging.handlers import RotatingFileHandler

def setup_logging():
    logger = logging.getLogger('saas_llm')
    logger.setLevel(logging.INFO)

    # File handler with rotation
    handler = RotatingFileHandler(
        'logs/saas_llm.log',
        maxBytes=10485760,  # 10MB
        backupCount=10
    )

    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    handler.setFormatter(formatter)
    logger.addHandler(handler)

    return logger

# scripts/recalculate_usage.py
from services.usage_calculator import calculate_team_usage

def recalculate_all_usage(period: str):
    teams = db.query(TeamCredits).all()

    for team in teams:
        summary = calculate_team_usage(
            db,
            team_id=team.team_id,
            period=period
        )

        # Store summary
        db.merge(summary)

    db.commit()

# Usage
recalculate_all_usage("2025-10")