Uvicorn与Azure Log Analytics：Python ASGI服务器日志查询与分析完整指南

【免费下载链接】uvicorn An ASGI web server, for Python. 🦄 项目地址: https://gitcode.com/GitHub_Trending/uv/uvicorn

Uvicorn作为Python生态中高性能的ASGI服务器，为FastAPI、Starlette等现代Web框架提供强大的异步支持。在实际生产环境中，如何有效收集、存储和分析Uvicorn的日志数据是确保应用稳定运行的关键。本文将详细介绍如何将Uvicorn的日志系统与Azure Log Analytics集成，实现企业级的日志监控和分析解决方案。

📊 Uvicorn日志系统架构解析

Uvicorn内置了完善的日志系统，通过Python标准库的logging模块实现。在uvicorn/logging.py中，我们可以看到两个核心的日志格式化器：

• DefaultFormatter：处理通用日志，支持彩色输出
• AccessFormatter：专门处理HTTP访问日志，包含客户端地址、请求方法、路径和状态码

默认的日志配置位于uvicorn/config.py的LOGGING_CONFIG字典中，定义了三个主要的日志记录器：

• uvicorn：服务器核心日志
• uvicorn.error：错误日志
• uvicorn.access：访问日志，输出到标准输出

Uvicorn的星空独角兽标识，代表高性能与创新的Python ASGI服务器

🔧 Uvicorn日志配置详解

基础配置方法

Uvicorn支持多种日志配置方式，可以通过命令行参数或Python代码进行配置：

# 通过代码配置
import uvicorn
from uvicorn.config import Config

app = "your_app:app"

config = Config(
    app=app,
    host="0.0.0.0",
    port=8000,
    log_level="info",
    log_config={
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "default": {
                "()": "uvicorn.logging.DefaultFormatter",
                "fmt": "%(levelprefix)s %(message)s",
            },
            "access": {
                "()": "uvicorn.logging.AccessFormatter",
                "fmt": '%(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s',
            },
        },
        "handlers": {
            "default": {
                "formatter": "default",
                "class": "logging.StreamHandler",
                "stream": "ext://sys.stderr",
            },
            "access": {
                "formatter": "access",
                "class": "logging.StreamHandler",
                "stream": "ext://sys.stdout",
            },
        },
        "loggers": {
            "uvicorn": {"handlers": ["default"], "level": "INFO"},
            "uvicorn.error": {"level": "INFO"},
            "uvicorn.access": {"handlers": ["access"], "level": "INFO"},
        },
    }
)

server = uvicorn.Server(config)
await server.serve()

使用外部配置文件

Uvicorn支持JSON和YAML格式的日志配置文件：

# 使用JSON配置文件
uvicorn your_app:app --log-config logging.json

# 使用YAML配置文件（需要安装PyYAML）
uvicorn your_app:app --log-config logging.yaml

🚀 集成Azure Log Analytics的完整方案

1. 安装必要的Azure SDK

pip install azure-monitor-opentelemetry azure-identity

2. 创建Azure Log Analytics自定义日志处理器

# azure_log_handler.py
import logging
import json
from datetime import datetime
from azure.monitor.opentelemetry import configure_azure_monitor
from opentelemetry import trace
from opentelemetry.sdk.resources import Resource

class AzureLogAnalyticsHandler(logging.Handler):
    """自定义处理器，将Uvicorn日志发送到Azure Log Analytics"""
    
    def __init__(self, workspace_id, shared_key, log_type="UvicornLogs"):
        super().__init__()
        self.workspace_id = workspace_id
        self.shared_key = shared_key
        self.log_type = log_type
        self._setup_azure_monitor()
    
    def _setup_azure_monitor(self):
        """配置Azure Monitor"""
        configure_azure_monitor(
            connection_string=f"InstrumentationKey={self.shared_key}",
            resource=Resource.create({
                "service.name": "uvicorn-server",
                "service.instance.id": "production-001"
            })
        )
    
    def emit(self, record):
        """发送日志记录到Azure Log Analytics"""
        try:
            log_data = {
                "TimeGenerated": datetime.utcnow().isoformat(),
                "LogLevel": record.levelname,
                "Message": self.format(record),
                "LoggerName": record.name,
                "ProcessID": record.process,
                "ThreadID": record.thread,
                "FunctionName": record.funcName,
                "LineNumber": record.lineno,
                "FileName": record.filename,
            }
            
            # 添加Uvicorn特定的字段
            if hasattr(record, 'client_addr'):
                log_data['ClientAddress'] = record.client_addr
            if hasattr(record, 'request_line'):
                log_data['RequestLine'] = record.request_line
            if hasattr(record, 'status_code'):
                log_data['StatusCode'] = record.status_code
            
            # 这里应该实现实际的Azure Log Analytics API调用
            # 为了示例简化，实际使用时需要实现HTTP数据收集API
            self._send_to_azure(log_data)
            
        except Exception as e:
            print(f"Failed to send log to Azure: {e}")
    
    def _send_to_azure(self, log_data):
        """实际发送数据到Azure Log Analytics的HTTP数据收集API"""
        # 实现Azure Data Collector API调用
        pass

3. 配置Uvicorn使用Azure日志处理器

# main_with_azure.py
import uvicorn
import logging
from azure_log_handler import AzureLogAnalyticsHandler

# Azure Log Analytics配置
AZURE_WORKSPACE_ID = "your-workspace-id"
AZURE_SHARED_KEY = "your-shared-key"

def configure_logging():
    """配置包含Azure Log Analytics的日志系统"""
    
    # 创建Azure处理器
    azure_handler = AzureLogAnalyticsHandler(
        workspace_id=AZURE_WORKSPACE_ID,
        shared_key=AZURE_SHARED_KEY
    )
    azure_handler.setLevel(logging.INFO)
    
    # 创建标准格式化器
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    azure_handler.setFormatter(formatter)
    
    # 配置Uvicorn日志记录器
    uvicorn_logger = logging.getLogger("uvicorn")
    uvicorn_logger.addHandler(azure_handler)
    
    uvicorn_access_logger = logging.getLogger("uvicorn.access")
    uvicorn_access_logger.addHandler(azure_handler)
    
    uvicorn_error_logger = logging.getLogger("uvicorn.error")
    uvicorn_error_logger.addHandler(azure_handler)

if __name__ == "__main__":
    # 配置日志
    configure_logging()
    
    # 启动Uvicorn服务器
    uvicorn.run(
        "your_app:app",
        host="0.0.0.0",
        port=8000,
        log_level="info",
        # 禁用默认的日志配置，使用我们的自定义配置
        log_config=None
    )

📈 Azure Log Analytics中的查询与分析

1. 基础查询示例

// 查询所有Uvicorn日志
UvicornLogs
| where TimeGenerated > ago(24h)
| project TimeGenerated, LogLevel, Message, ClientAddress, StatusCode

// 按日志级别统计
UvicornLogs
| where TimeGenerated > ago(7d)
| summarize count() by LogLevel
| render piechart

// 查找错误日志
UvicornLogs
| where LogLevel == "ERROR" or LogLevel == "CRITICAL"
| where TimeGenerated > ago(1h)
| order by TimeGenerated desc

// 分析HTTP状态码分布
UvicornLogs
| where isnotempty(StatusCode)
| summarize count() by StatusCode
| order by count_ desc

2. 性能监控查询

// 监控请求延迟
UvicornLogs
| where Message contains "GET" or Message contains "POST"
| extend RequestTime = extract(@"(\d+\.\d+)ms", 1, Message)
| where isnotempty(RequestTime)
| summarize 
    avg(RequestTime), 
    p95(RequestTime), 
    p99(RequestTime), 
    max(RequestTime) 
    by bin(TimeGenerated, 5m)

// 识别慢请求
UvicornLogs
| where Message contains "ms"
| extend RequestTime = extract(@"(\d+\.\d+)ms", 1, Message)
| where RequestTime > 1000  // 超过1秒的请求
| project TimeGenerated, ClientAddress, Message, RequestTime
| order by RequestTime desc

3. 安全分析查询

// 检测异常访问模式
UvicornLogs
| where StatusCode == "404"
| summarize count() by ClientAddress
| where count_ > 100  // 同一客户端大量404错误
| order by count_ desc

// 监控认证失败
UvicornLogs
| where StatusCode == "401" or StatusCode == "403"
| summarize 
    FailedAttempts = count(), 
    DistinctUsers = dcount(extract(@"user=([^,]+)", 1, Message))
    by ClientAddress, bin(TimeGenerated, 1h)
| where FailedAttempts > 10

🛠️ 实战配置示例

完整的Docker部署配置

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 配置环境变量
ENV AZURE_WORKSPACE_ID=${AZURE_WORKSPACE_ID}
ENV AZURE_SHARED_KEY=${AZURE_SHARED_KEY}
ENV LOG_LEVEL=info
ENV PORT=8000

# 启动命令
CMD ["python", "main_with_azure.py"]

Kubernetes部署配置

# k8s-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: uvicorn-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: uvicorn-app
  template:
    metadata:
      labels:
        app: uvicorn-app
    spec:
      containers:
      - name: uvicorn
        image: your-registry/uvicorn-app:latest
        ports:
        - containerPort: 8000
        env:
        - name: AZURE_WORKSPACE_ID
          valueFrom:
            secretKeyRef:
              name: azure-secrets
              key: workspace-id
        - name: AZURE_SHARED_KEY
          valueFrom:
            secretKeyRef:
              name: azure-secrets
              key: shared-key
        - name: LOG_LEVEL
          value: "info"
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"
          limits:
            memory: "512Mi"
            cpu: "500m"

🔍 高级监控与告警配置

1. Azure Monitor告警规则

{
  "location": "global",
  "properties": {
    "description": "Uvicorn错误率超过阈值",
    "severity": 2,
    "enabled": true,
    "scopes": [
      "/subscriptions/{subscription-id}/resourceGroups/{resource-group}/providers/Microsoft.OperationalInsights/workspaces/{workspace-name}"
    ],
    "evaluationFrequency": "PT5M",
    "windowSize": "PT5M",
    "criteria": {
      "allOf": [
        {
          "query": "UvicornLogs\n| where LogLevel == 'ERROR' or LogLevel == 'CRITICAL'\n| summarize ErrorCount = count() by bin(TimeGenerated, 5m)\n| join kind=inner (UvicornLogs\n| summarize TotalCount = count() by bin(TimeGenerated, 5m)) on TimeGenerated\n| extend ErrorRate = ErrorCount * 100.0 / TotalCount\n| where ErrorRate > 5",
          "timeAggregation": "Average",
          "threshold": 5,
          "operator": "GreaterThan",
          "metricMeasureColumn": "ErrorRate"
        }
      ],
      "odata.type": "Microsoft.Azure.Monitor.SingleResourceMultipleMetricCriteria"
    },
    "actions": [
      {
        "actionGroupId": "/subscriptions/{subscription-id}/resourceGroups/{resource-group}/providers/microsoft.insights/actionGroups/{action-group}"
      }
    ]
  }
}

2. 性能指标监控

GitHub Actions中的CI/CD流程监控，类似Azure Log Analytics的日志分析面板

💡 最佳实践与优化建议

1. 日志结构化

确保日志消息包含结构化数据，便于Azure Log Analytics解析：

import json
import logging

class StructuredLogger:
    def __init__(self, name):
        self.logger = logging.getLogger(name)
    
    def log_request(self, client_addr, method, path, status_code, duration_ms):
        log_data = {
            "event": "http_request",
            "client_addr": client_addr,
            "method": method,
            "path": path,
            "status_code": status_code,
            "duration_ms": duration_ms,
            "timestamp": datetime.utcnow().isoformat()
        }
        self.logger.info(json.dumps(log_data))

2. 采样与聚合

对于高流量应用，实现日志采样：

import random

class SampledAzureHandler(AzureLogAnalyticsHandler):
    def __init__(self, workspace_id, shared_key, sample_rate=0.1):
        super().__init__(workspace_id, shared_key)
        self.sample_rate = sample_rate
    
    def emit(self, record):
        # 只发送部分日志以减少成本
        if random.random() < self.sample_rate:
            super().emit(record)

3. 本地开发与调试

创建本地开发配置：

# local_config.py
import logging

def get_log_config(environment="development"):
    if environment == "production":
        # 生产环境使用Azure Log Analytics
        return {
            "version": 1,
            "disable_existing_loggers": False,
            "handlers": {
                "azure": {
                    "class": "azure_log_handler.AzureLogAnalyticsHandler",
                    "workspace_id": "your-workspace-id",
                    "shared_key": "your-shared-key",
                    "level": "INFO"
                }
            },
            "loggers": {
                "uvicorn": {"handlers": ["azure"], "level": "INFO"},
                "uvicorn.access": {"handlers": ["azure"], "level": "INFO"},
                "uvicorn.error": {"level": "INFO"}
            }
        }
    else:
        # 开发环境使用控制台输出
        return {
            "version": 1,
            "disable_existing_loggers": False,
            "formatters": {
                "default": {
                    "()": "uvicorn.logging.DefaultFormatter",
                    "fmt": "%(levelprefix)s %(message)s",
                    "use_colors": True,
                },
                "access": {
                    "()": "uvicorn.logging.AccessFormatter",
                    "fmt": '%(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s',
                },
            },
            "handlers": {
                "default": {
                    "formatter": "default",
                    "class": "logging.StreamHandler",
                    "stream": "ext://sys.stderr",
                },
                "access": {
                    "formatter": "access",
                    "class": "logging.StreamHandler",
                    "stream": "ext://sys.stdout",
                },
            },
            "loggers": {
                "uvicorn": {"handlers": ["default"], "level": "DEBUG"},
                "uvicorn.error": {"level": "DEBUG"},
                "uvicorn.access": {"handlers": ["access"], "level": "INFO"},
            },
        }

📊 监控仪表板配置

在Azure Portal中创建监控仪表板，包含以下关键指标：

1. 请求吞吐量：每分钟请求数
2. 错误率：HTTP错误状态码比例
3. 响应时间：P50、P95、P99延迟
4. 资源使用：CPU、内存消耗
5. 用户分布：按地理位置的访问量

🎯 总结

通过将Uvicorn与Azure Log Analytics集成，您可以获得：

1. 实时监控：及时发现并响应生产环境问题
2. 历史分析：基于历史数据进行趋势分析和容量规划
3. 智能告警：基于自定义规则的自动告警
4. 成本优化：通过采样和聚合控制日志存储成本
5. 合规审计：满足安全和合规要求的日志记录

这种集成方案不仅适用于Uvicorn，也可以扩展到其他Python ASGI服务器和Web框架。通过合理的日志结构化和Azure Log Analytics的强大查询能力，您可以构建一个完整、可扩展的应用程序监控解决方案。

记住，良好的日志实践是生产环境稳定运行的基础。从项目初期就规划好日志策略，将为后续的运维和故障排查节省大量时间和精力。

【免费下载链接】uvicorn An ASGI web server, for Python. 🦄 项目地址: https://gitcode.com/GitHub_Trending/uv/uvicorn