#FastAPI Nginx与Gunicorn生产部署完全指南

# gunicorn.conf.py - Gunicorn生产配置
import multiprocessing
import os

# 服务器配置
bind = "0.0.0.0:8000"  # 绑定地址和端口
backlog = 2048  # 监听队列大小

# 进程配置
workers = multiprocessing.cpu_count() * 2 + 1  # 工作进程数 (推荐: CPU核心数 * 2 + 1)
worker_class = "uvicorn.workers.UvicornWorker"  # 使用Uvicorn作为worker
worker_connections = 1000  # 每个worker的最大连接数
max_requests = 1000  # 每个worker处理的最大请求数后重启
max_requests_jitter = 100  # 随机抖动，避免同时重启
timeout = 300  # 请求超时时间
graceful_timeout = 30  # 优雅关闭超时时间
keepalive = 5  # HTTP长连接超时时间

# 进程命名
proc_name = "daoman_api"

# 服务器软件标识
server_software = "DaomanAPI/1.0"

# 重启配置
max_requests = 1000  # 每个worker处理1000个请求后重启
max_requests_jitter = 100  # 随机抖动避免同时重启

# 日志配置
accesslog = "/var/log/gunicorn/access.log"  # 访问日志
errorlog = "/var/log/gunicorn/error.log"    # 错误日志
loglevel = "info"  # 日志级别
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'

# 进程管理
preload_app = True  # 预加载应用
daemon = False  # 不作为守护进程运行
enable_stdio_inheritance = True  # 继承stdio

# 资源限制
limit_request_line = 4094  # HTTP请求行最大长度
limit_request_fields = 100  # HTTP头部字段最大数量
limit_request_field_size = 8190  # HTTP头部字段最大长度

# SSL配置（如果需要）
# keyfile = "/path/to/keyfile"
# certfile = "/path/to/certfile"

# multi_port_gunicorn.py - 多端口配置示例
import multiprocessing
import socket
from contextlib import closing

def find_free_port():
    """查找可用端口"""
    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
        s.bind(('', 0))
        s.listen(1)
        port = s.getsockname()[1]
    return port

# 生成多个Gunicorn配置
def generate_multi_port_config():
    """生成多端口配置"""
    cpu_cores = multiprocessing.cpu_count()
    ports = [find_free_port() for _ in range(min(4, cpu_cores))]
    
    configs = []
    for i, port in enumerate(ports):
        config = {
            'bind': f'0.0.0.0:{port}',
            'workers': 1,
            'worker_class': 'uvicorn.workers.UvicornWorker',
            'worker_connections': 1000,
            'max_requests': 1000,
            'max_requests_jitter': 100,
            'timeout': 300,
            'graceful_timeout': 30,
            'keepalive': 5,
            'preload_app': True,
        }
        configs.append(config)
    
    return configs, ports

# 使用supervisor管理多个Gunicorn实例
"""
[group:daoman_api]
programs:daoman_api_8000,daoman_api_8001,daoman_api_8002,daoman_api_8003

[program:daoman_api_8000]
command=gunicorn main:app --bind 0.0.0.0:8000 --workers 1 --worker-class uvicorn.workers.UvicornWorker
directory=/opt/daoman
user=www-data
autostart=true
autorestart=true
redirect_stderr=true
stdout_logfile=/var/log/gunicorn/daoman_api_8000.log

[program:daoman_api_8001]
command=gunicorn main:app --bind 0.0.0.0:8001 --workers 1 --worker-class uvicorn.workers.UvicornWorker
directory=/opt/daoman
user=www-data
autostart=true
autorestart=true
redirect_stderr=true
stdout_logfile=/var/log/gunicorn/daoman_api_8001.log
"""

# ssl_monitor.py - SSL证书监控
import ssl
import socket
from datetime import datetime, timedelta
import smtplib
from email.mime.text import MIMEText
import subprocess

def check_ssl_expiry(domain, port=443, warning_days=30):
    """检查SSL证书到期时间"""
    context = ssl.create_default_context()
    
    with socket.create_connection((domain, port), timeout=10) as sock:
        with context.wrap_socket(sock, server_hostname=domain) as ssock:
            cert = ssock.getpeercert()
            
    # 解析证书到期时间
    expiry_date = datetime.strptime(cert['notAfter'], '%b %d %H:%M:%S %Y %Z')
    days_until_expiry = (expiry_date - datetime.now()).days
    
    return {
        'domain': domain,
        'expiry_date': expiry_date.strftime('%Y-%m-%d'),
        'days_until_expiry': days_until_expiry,
        'is_expired': days_until_expiry <= 0,
        'needs_renewal': days_until_expiry <= warning_days
    }

def renew_certificates(domains):
    """续订证书"""
    for domain in domains:
        result = subprocess.run([
            'certbot', 'renew', '--cert-name', domain, '--quiet'
        ], capture_output=True, text=True)
        
        if result.returncode == 0:
            print(f"Successfully renewed certificate for {domain}")
        else:
            print(f"Failed to renew certificate for {domain}: {result.stderr}")

def send_alert_email(recipients, subject, body):
    """发送告警邮件"""
    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = 'monitor@your-domain.com'
    msg['To'] = ', '.join(recipients)
    
    # 发送邮件（需要配置SMTP服务器）
    # smtp_server = smtplib.SMTP('smtp.your-server.com', 587)
    # smtp_server.starttls()
    # smtp_server.login('username', 'password')
    # smtp_server.send_message(msg)
    # smtp_server.quit()

# 检查证书到期时间
domains_to_check = ['your-domain.com', 'www.your-domain.com']
for domain in domains_to_check:
    cert_info = check_ssl_expiry(domain)
    print(f"Domain: {cert_info['domain']}")
    print(f"Expiry: {cert_info['expiry_date']}")
    print(f"Days until expiry: {cert_info['days_until_expiry']}")
    print(f"Needs renewal: {cert_info['needs_renewal']}")
    print("-" * 50)

# health_check.py - 应用层健康检查
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from datetime import datetime
import asyncio
import logging
import psutil
import platform

app = FastAPI()

class HealthStatus(BaseModel):
    status: str
    timestamp: str
    uptime: float
    version: str
    services: dict
    system: dict

@app.get("/health", response_model=HealthStatus)
async def health_check():
    """健康检查端点"""
    import time
    start_time = time.time()
    
    # 检查各个服务状态
    services_status = {
        "database": await check_database_connection(),
        "redis": await check_redis_connection(),
        "external_api": await check_external_api(),
        "disk_space": await check_disk_space(),
        "memory": await check_memory_usage(),
    }
    
    # 获取系统信息
    system_info = {
        "cpu_percent": psutil.cpu_percent(interval=1),
        "memory_percent": psutil.virtual_memory().percent,
        "disk_percent": psutil.disk_usage('/').percent,
        "platform": platform.platform(),
        "python_version": platform.python_version(),
    }
    
    # 计算总状态
    service_statuses = list(services_status.values())
    overall_status = "healthy" if all(service_statuses) else "degraded"
    
    return HealthStatus(
        status=overall_status,
        timestamp=datetime.now().isoformat(),
        uptime=time.time() - start_time,
        version="1.0.0",
        services=services_status,
        system=system_info
    )

async def check_database_connection():
    """检查数据库连接"""
    try:
        # 实现数据库连接检查逻辑
        await asyncio.sleep(0.1)  # 模拟数据库调用
        return True
    except Exception as e:
        logging.error(f"Database connection check failed: {e}")
        return False

async def check_redis_connection():
    """检查Redis连接"""
    try:
        # 实现Redis连接检查逻辑
        await asyncio.sleep(0.05)  # 模拟Redis调用
        return True
    except Exception as e:
        logging.error(f"Redis connection check failed: {e}")
        return False

async def check_external_api():
    """检查外部API"""
    try:
        # 实现外部API检查逻辑
        await asyncio.sleep(0.1)  # 模拟外部调用
        return True
    except Exception as e:
        logging.error(f"External API check failed: {e}")
        return False

async def check_disk_space():
    """检查磁盘空间"""
    try:
        disk_usage = psutil.disk_usage('/')
        # 如果磁盘使用率超过90%，返回False
        return disk_usage.percent < 90
    except Exception as e:
        logging.error(f"Disk space check failed: {e}")
        return False

async def check_memory_usage():
    """检查内存使用"""
    try:
        memory_percent = psutil.virtual_memory().percent
        # 如果内存使用率超过90%，返回False
        return memory_percent < 90
    except Exception as e:
        logging.error(f"Memory usage check failed: {e}")
        return False

# 就绪探针（用于Kubernetes等编排系统）
@app.get("/ready")
async def readiness_check():
    """就绪检查端点"""
    # 实现就绪检查逻辑
    return {"status": "ready"}

# 存活探针
@app.get("/live")
async def liveness_check():
    """存活检查端点"""
    return {"status": "alive"}

# security_middleware.py - 安全中间件
from fastapi import FastAPI, Request, Response
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
import time
import hashlib
import hmac

app = FastAPI()

class SecurityMiddleware(BaseHTTPMiddleware):
    """安全中间件"""
    
    async def dispatch(self, request: Request, call_next):
        # 添加安全头部
        response = await call_next(request)
        
        # 添加安全头部
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        
        # 防止暴力破解
        if request.method == "POST" and "/auth/" in request.url.path:
            if await self.check_rate_limit(request):
                return JSONResponse(
                    status_code=429,
                    content={"detail": "Too many requests"}
                )
        
        return response
    
    async def check_rate_limit(self, request: Request):
        """检查请求频率限制"""
        # 实现请求频率限制逻辑
        # 这里只是示例
        return False

# 添加安全中间件
app.add_middleware(SecurityMiddleware)

# 添加可信主机中间件
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["your-domain.com", "www.your-domain.com", "*.your-domain.com"]
)

# CSRF保护（如果需要）
class CSRFProtection:
    """CSRF保护"""
    
    def __init__(self, secret_key: str):
        self.secret_key = secret_key
    
    def generate_token(self, request: Request) -> str:
        """生成CSRF令牌"""
        timestamp = str(int(time.time()))
        data = f"{request.client.host}:{timestamp}"
        signature = hmac.new(
            self.secret_key.encode(),
            data.encode(),
            hashlib.sha256
        ).hexdigest()
        return f"{timestamp}.{signature}"
    
    def verify_token(self, request: Request, token: str) -> bool:
        """验证CSRF令牌"""
        try:
            timestamp, signature = token.split(".", 1)
            expected_timestamp = str(int(time.time()))
            
            # 检查时间戳（令牌有效期5分钟）
            if abs(int(expected_timestamp) - int(timestamp)) > 300:
                return False
            
            data = f"{request.client.host}:{timestamp}"
            expected_signature = hmac.new(
                self.secret_key.encode(),
                data.encode(),
                hashlib.sha256
            ).hexdigest()
            
            return hmac.compare_digest(signature, expected_signature)
        except:
            return False

# performance_config.py - Gunicorn性能配置
import multiprocessing
import os

def get_optimal_workers():
    """获取最优工作进程数"""
    cpu_count = multiprocessing.cpu_count()
    # 对于I/O密集型应用，通常设置为CPU核心数的2倍
    # 对于CPU密集型应用，通常设置为CPU核心数
    return min(cpu_count * 2, 8)  # 最大不超过8个进程

def get_optimal_connections():
    """获取最优连接数"""
    # 每个worker的连接数
    return 1000

class PerformanceConfig:
    """性能配置类"""
    
    @staticmethod
    def get_gunicorn_config():
        """获取Gunicorn性能配置"""
        cpu_count = multiprocessing.cpu_count()
        
        return {
            # 服务器配置
            'bind': '0.0.0.0:8000',
            'backlog': 2048,
            
            # 进程配置
            'workers': get_optimal_workers(),
            'worker_class': 'uvicorn.workers.UvicornWorker',
            'worker_connections': get_optimal_connections(),
            'max_requests': 1000,
            'max_requests_jitter': 100,
            'timeout': 300,
            'graceful_timeout': 30,
            'keepalive': 5,
            
            # 性能优化
            'preload_app': True,
            'worker_tmp_dir': '/dev/shm',  # 使用内存临时目录
            'max_keepalive_requests': 100,  # 最大长连接请求数
            'keepalive_timeout': 2,        # 长连接超时时间
            
            # 日志配置
            'accesslog': '/var/log/gunicorn/access.log',
            'errorlog': '/var/log/gunicorn/error.log',
            'loglevel': 'info',
        }

# 内存优化配置
"""
# 在gunicorn.conf.py中添加
worker_tmp_dir = '/dev/shm'  # 使用内存临时目录，提高I/O性能
max_keepalive_requests = 100  # 限制长连接请求数，防止内存泄漏
keepalive = 5  # 合理的长连接超时时间
"""

# 预加载优化
"""
preload_app = True  # 预加载应用，减少worker启动时间
worker_class = 'uvicorn.workers.UvicornWorker'  # 使用高性能worker
"""

# performance_optimization.py - 应用层性能优化
from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
import asyncio
import time
from functools import wraps
import aiocache
from aiocache import cached, Cache
from aiocache.serializers import JsonSerializer

app = FastAPI()

class PerformanceMiddleware(BaseHTTPMiddleware):
    """性能监控中间件"""
    
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        
        response = await call_next(request)
        
        process_time = time.time() - start_time
        response.headers["X-Process-Time"] = str(process_time)
        
        # 记录慢查询
        if process_time > 1.0:  # 超过1秒的请求
            print(f"Slow request: {request.url} took {process_time:.2f}s")
        
        return response

app.add_middleware(PerformanceMiddleware)

# 缓存配置
cache = Cache(
    aiocache.SimpleMemoryCache,
    serializer=JsonSerializer(),
    ttl=300  # 5分钟缓存
)

# 缓存装饰器
@cached(cache=cache, ttl=300)
async def get_expensive_data(param: str):
    """带缓存的昂贵操作"""
    # 模拟昂贵的数据库查询或外部API调用
    await asyncio.sleep(0.1)
    return {"data": f"expensive result for {param}"}

# 数据库连接池优化
"""
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,  # 连接池大小
    max_overflow=30,  # 最大溢出连接数
    pool_pre_ping=True,  # 连接前ping检查
    pool_recycle=300,  # 连接回收时间
    echo=False  # 是否打印SQL语句
)

AsyncSessionLocal = sessionmaker(
    engine, 
    class_=AsyncSession, 
    expire_on_commit=False
)
"""

# 响应压缩中间件
"""
from starlette.middleware.gzip import GZipMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1000)
"""

# logging_config.py - 日志配置
import logging
import logging.handlers
import json
from datetime import datetime
import sys
from typing import Dict, Any

class JSONFormatter(logging.Formatter):
    """JSON格式化器"""
    
    def format(self, record):
        log_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno,
        }
        
        # 添加异常信息
        if record.exc_info:
            log_entry['exception'] = self.formatException(record.exc_info)
        
        # 添加额外信息
        if hasattr(record, 'extra'):
            log_entry.update(record.extra)
        
        return json.dumps(log_entry)

def setup_logging(level=logging.INFO):
    """设置日志配置"""
    # 创建logger
    logger = logging.getLogger()
    logger.setLevel(level)
    
    # 清除现有处理器
    logger.handlers.clear()
    
    # 控制台处理器
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setLevel(level)
    console_formatter = JSONFormatter()
    console_handler.setFormatter(console_formatter)
    logger.addHandler(console_handler)
    
    # 文件处理器
    file_handler = logging.handlers.RotatingFileHandler(
        '/var/log/daoman/app.log',
        maxBytes=100 * 1024 * 1024,  # 100MB
        backupCount=10
    )
    file_handler.setLevel(level)
    file_formatter = JSONFormatter()
    file_handler.setFormatter(file_formatter)
    logger.addHandler(file_handler)
    
    # 错误日志处理器
    error_handler = logging.handlers.RotatingFileHandler(
        '/var/log/daoman/error.log',
        maxBytes=100 * 1024 * 1024,
        backupCount=10
    )
    error_handler.setLevel(logging.ERROR)
    error_formatter = JSONFormatter()
    error_handler.setFormatter(error_formatter)
    logger.addHandler(error_handler)

# 使用日志配置
setup_logging(logging.INFO)

# 结构化日志记录
def log_request(request, response_time, status_code):
    """记录请求日志"""
    logger = logging.getLogger(__name__)
    logger.info(
        "Request processed",
        extra={
            'method': request.method,
            'path': str(request.url),
            'response_time': response_time,
            'status_code': status_code,
            'user_agent': request.headers.get('user-agent'),
            'ip_address': request.client.host
        }
    )

# system_monitor.py - 系统监控
import psutil
import time
import requests
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
from typing import Dict, Any

class SystemMonitor:
    """系统监控类"""
    
    def __init__(self, api_url: str, thresholds: Dict[str, float] = None):
        self.api_url = api_url
        self.thresholds = thresholds or {
            'cpu_percent': 80.0,
            'memory_percent': 85.0,
            'disk_percent': 90.0,
            'response_time': 2.0
        }
    
    def get_system_metrics(self) -> Dict[str, Any]:
        """获取系统指标"""
        return {
            'timestamp': datetime.now().isoformat(),
            'cpu_percent': psutil.cpu_percent(interval=1),
            'memory_percent': psutil.virtual_memory().percent,
            'disk_percent': psutil.disk_usage('/').percent,
            'load_average': psutil.getloadavg(),
            'network_io': psutil.net_io_counters()._asdict(),
            'process_count': len(psutil.pids()),
        }
    
    def check_api_health(self) -> Dict[str, Any]:
        """检查API健康状态"""
        start_time = time.time()
        try:
            response = requests.get(f"{self.api_url}/health", timeout=10)
            response_time = time.time() - start_time
            
            return {
                'status_code': response.status_code,
                'response_time': response_time,
                'is_healthy': response.status_code == 200,
                'content_length': len(response.content)
            }
        except Exception as e:
            return {
                'status_code': 0,
                'response_time': time.time() - start_time,
                'is_healthy': False,
                'error': str(e)
            }
    
    def check_alerts(self, metrics: Dict[str, Any]) -> Dict[str, bool]:
        """检查告警条件"""
        alerts = {}
        
        for metric, threshold in self.thresholds.items():
            if metric in metrics:
                alerts[metric] = metrics[metric] > threshold
        
        return alerts
    
    def send_alert(self, alert_message: str, recipients: list):
        """发送告警"""
        print(f"ALERT: {alert_message}")  # 实际部署时替换为邮件或短信通知
        
    def run_monitoring_cycle(self, recipients: list):
        """运行监控周期"""
        # 获取系统指标
        metrics = self.get_system_metrics()
        
        # 检查API健康
        api_health = self.check_api_health()
        metrics.update(api_health)
        
        # 检查告警
        alerts = self.check_alerts(metrics)
        
        # 处理告警
        for metric, is_alert in alerts.items():
            if is_alert:
                alert_msg = f"ALERT: {metric} exceeded threshold. Current value: {metrics[metric]}"
                self.send_alert(alert_msg, recipients)
        
        return metrics

# 使用示例
if __name__ == "__main__":
    monitor = SystemMonitor(
        api_url="https://your-domain.com",
        thresholds={
            'cpu_percent': 80.0,
            'memory_percent': 85.0,
            'disk_percent': 90.0,
            'response_time': 2.0
        }
    )
    
    recipients = ["admin@your-domain.com"]
    
    while True:
        try:
            metrics = monitor.run_monitoring_cycle(recipients)
            print(f"System metrics: {metrics}")
            time.sleep(60)  # 每分钟检查一次
        except KeyboardInterrupt:
            print("Monitoring stopped.")
            break
        except Exception as e:
            print(f"Monitoring error: {e}")
            time.sleep(60)

# performance_analysis.py - 性能分析
import cProfile
import pstats
import io
from contextlib import redirect_stdout
import asyncio
import time
from functools import wraps
import tracemalloc

def profile_endpoint(func):
    """性能分析装饰器"""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        # 内存追踪
        tracemalloc.start()
        
        # CPU性能分析
        pr = cProfile.Profile()
        pr.enable()
        
        start_time = time.time()
        
        try:
            result = await func(*args, **kwargs)
        finally:
            end_time = time.time()
            
            pr.disable()
            
            # 保存性能数据
            s = io.StringIO()
            ps = pstats.Stats(pr, stream=s)
            ps.sort_stats('cumulative')
            ps.print_stats(20)  # 显示前20个最耗时的函数
            
            # 内存使用情况
            current, peak = tracemalloc.get_traced_memory()
            tracemalloc.stop()
            
            print(f"\n=== 性能分析报告 ===")
            print(f"函数: {func.__name__}")
            print(f"执行时间: {end_time - start_time:.4f}秒")
            print(f"当前内存使用: {current / 1024 / 1024:.2f} MB")
            print(f"峰值内存使用: {peak / 1024 / 1024:.2f} MB")
            print(f"\n最耗时的函数:")
            print(s.getvalue())
    
    return wrapper

# 使用示例
"""
@app.get("/profiled-endpoint")
@profile_endpoint
async def profiled_endpoint():
    # 模拟一些处理
    await asyncio.sleep(0.1)
    return {"message": "Profiled response"}
"""

# 请求追踪
class RequestTracer:
    """请求追踪器"""
    
    def __init__(self):
        self.requests = {}
    
    async def trace_request(self, request_id: str, operation: str, duration: float):
        """追踪请求"""
        if request_id not in self.requests:
            self.requests[request_id] = []
        
        self.requests[request_id].append({
            'operation': operation,
            'duration': duration,
            'timestamp': time.time()
        })
    
    def get_request_trace(self, request_id: str):
        """获取请求追踪信息"""
        return self.requests.get(request_id, [])

# 异步性能监控
async def async_performance_monitor():
    """异步性能监控"""
    import aiohttp
    import asyncio
    
    urls = [
        "https://your-domain.com/health",
        "https://your-domain.com/api/users",
        "https://your-domain.com/api/items"
    ]
    
    async def fetch(session, url):
        start_time = time.time()
        try:
            async with session.get(url) as response:
                response_time = time.time() - start_time
                return {
                    'url': url,
                    'status': response.status,
                    'response_time': response_time,
                    'success': response.status == 200
                }
        except Exception as e:
            return {
                'url': url,
                'status': 0,
                'response_time': time.time() - start_time,
                'success': False,
                'error': str(e)
            }
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        
        print("=== 异步性能测试结果 ===")
        for result in results:
            status = "✓" if result['success'] else "✗"
            print(f"{status} {result['url']} - {result['response_time']:.3f}s (Status: {result['status']})")

# 运行异步性能测试
# asyncio.run(async_performance_monitor())