MCP Server 的日志与可观测性

import logging

# 设置日志级别
logging.basicConfig(level=logging.INFO)

# 获取日志记录器
logger = logging.getLogger(__name__)

# 记录不同级别的日志
logger.debug("这是一条 DEBUG 级别的日志")
logger.info("这是一条 INFO 级别的日志")
logger.warning("这是一条 WARNING 级别的日志")
logger.error("这是一条 ERROR 级别的日志")
logger.critical("这是一条 CRITICAL 级别的日志")

import logging

# 设置文本日志格式
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

logger = logging.getLogger(__name__)
logger.info("这是一条带格式的 INFO 级别的日志")

import logging
import json
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_record = {
            "timestamp": datetime.now().isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "filename": record.filename,
            "lineno": record.lineno,
            "message": record.getMessage(),
            "exc_info": record.exc_info,
            "stack_info": record.stack_info
        }
        return json.dumps(log_record)

# 设置 JSON 日志格式
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
logger.addHandler(handler)

logger.info("这是一条 JSON 格式的 INFO 级别的日志")

import logging

# 设置文件日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S',
    filename='mcp_server.log',
    filemode='a'
)

logger = logging.getLogger(__name__)
logger.info("这是一条写入文件的 INFO 级别的日志")

import logging
from elasticsearch import Elasticsearch
from elasticsearch.handlers import ElasticsearchHandler

# 配置 Elasticsearch 连接
es = Elasticsearch(['http://localhost:9200'])

# 创建 Elasticsearch 日志处理器
es_handler = ElasticsearchHandler(
    hosts=[{'host': 'localhost', 'port': 9200}],
    index_name='mcp-server-logs'
)

# 设置日志级别和格式
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
logger.addHandler(es_handler)

logger.info("这是一条发送到 ELK Stack 的 INFO 级别的日志")

import logging
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler

# 按大小轮转的日志处理器
rotating_handler = RotatingFileHandler(
    filename='mcp_server_rotating.log',
    maxBytes=10 * 1024 * 1024,  # 10MB
    backupCount=5  # 保留 5 个备份文件
)

# 按时间轮转的日志处理器
timed_rotating_handler = TimedRotatingFileHandler(
    filename='mcp_server_timed.log',
    when='midnight',  # 每天 midnight 轮转
    interval=1,  # 每 1 天轮转一次
    backupCount=7  # 保留 7 个备份文件
)

# 设置日志格式
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s')
rotating_handler.setFormatter(formatter)
timed_rotating_handler.setFormatter(formatter)

# 配置日志记录器
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
logger.addHandler(rotating_handler)
logger.addHandler(timed_rotating_handler)

logger.info("这是一条带轮转的 INFO 级别的日志")

import psutil
import time
import logging

logger = logging.getLogger(__name__)

# 监控系统指标
def monitor_system_metrics(interval: int = 5):
    while True:
        # 获取 CPU 使用率
        cpu_percent = psutil.cpu_percent(interval=1)
        
        # 获取内存使用率
        memory = psutil.virtual_memory()
        memory_percent = memory.percent
        memory_used = memory.used / (1024 * 1024)  # 转换为 MB
        memory_total = memory.total / (1024 * 1024)  # 转换为 MB
        
        # 获取磁盘使用率
        disk = psutil.disk_usage('/')
        disk_percent = disk.percent
        disk_used = disk.used / (1024 * 1024 * 1024)  # 转换为 GB
        disk_total = disk.total / (1024 * 1024 * 1024)  # 转换为 GB
        
        # 获取网络流量
        network = psutil.net_io_counters()
        bytes_sent = network.bytes_sent / (1024 * 1024)  # 转换为 MB
        bytes_recv = network.bytes_recv / (1024 * 1024)  # 转换为 MB
        
        # 记录监控指标
        logger.info(
            f"系统监控: CPU: {cpu_percent}% | 内存: {memory_percent}% ({memory_used:.2f}/{memory_total:.2f} MB) | "
            f"磁盘: {disk_percent}% ({disk_used:.2f}/{disk_total:.2f} GB) | "
            f"网络发送: {bytes_sent:.2f} MB | 网络接收: {bytes_recv:.2f} MB"
        )
        
        # 等待指定时间间隔
        time.sleep(interval)

# 启动系统指标监控
if __name__ == "__main__":
    monitor_system_metrics(interval=5)

from fastapi import FastAPI
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

app = FastAPI(title="MCP Server", version="2.0")

# 定义 Prometheus 指标
REQUEST_COUNT = Counter('mcp_server_requests_total', 'Total number of requests', ['method', 'endpoint', 'status'])
REQUEST_LATENCY = Histogram('mcp_server_request_latency_seconds', 'Request latency in seconds', ['method', 'endpoint'])
ACTIVE_CONNECTIONS = Gauge('mcp_server_active_connections', 'Number of active connections')
TOOLS_EXECUTED = Counter('mcp_server_tools_executed_total', 'Total number of tools executed', ['tool_name'])

# 启动 Prometheus 服务器
start_http_server(8001)

# 中间件：记录请求指标
@app.middleware("http")
async def prometheus_middleware(request, call_next):
    start_time = time.time()
    
    # 增加活跃连接数
    ACTIVE_CONNECTIONS.inc()
    
    # 处理请求
    response = await call_next(request)
    
    # 计算请求延迟
    latency = time.time() - start_time
    
    # 记录请求指标
    REQUEST_COUNT.labels(
        method=request.method,
        endpoint=request.url.path,
        status=response.status_code
    ).inc()
    
    REQUEST_LATENCY.labels(
        method=request.method,
        endpoint=request.url.path
    ).observe(latency)
    
    # 减少活跃连接数
    ACTIVE_CONNECTIONS.dec()
    
    return response

# 示例路由
@app.get("/api/v1/health")
async def health_check():
    return {"status": "healthy"}

# 示例工具执行路由
@app.post("/api/v1/tools/{tool_name}/execute")
async def execute_tool(tool_name: str):
    # 记录工具执行指标
    TOOLS_EXECUTED.labels(tool_name=tool_name).inc()
    return {"tool_name": tool_name, "result": "success"}

# prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

rule_files:
  - "rules.yml"

scrape_configs:
  - job_name: "mcp-server"
    static_configs:
      - targets: ["localhost:8000"]

# rules.yml
groups:
- name: mcp-server-alerts
  rules:
  # CPU 使用率告警
  - alert: HighCPUUsage
    expr: cpu_usage_percent > 80
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "High CPU Usage"
      description: "CPU usage is above 80% for more than 5 minutes."
  
  # 内存使用率告警
  - alert: HighMemoryUsage
    expr: memory_usage_percent > 85
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "High Memory Usage"
      description: "Memory usage is above 85% for more than 5 minutes."
  
  # 请求错误率告警
  - alert: HighErrorRate
    expr: rate(http_requests_total{status=~"5.."}[5m]) / rate(http_requests_total[5m]) > 0.05
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "High Error Rate"
      description: "Error rate is above 5% for more than 5 minutes."

# alertmanager.yml
global:
  resolve_timeout: 5m

route:
  group_by: ['alertname', 'severity']
  group_wait: 30s
  group_interval: 5m
  repeat_interval: 1h
  receiver: 'email'

receivers:
- name: 'email'
  email_configs:
  - to: 'admin@example.com'
    from: 'alerts@example.com'
    smarthost: 'smtp.example.com:587'
    auth_username: 'alerts@example.com'
    auth_password: 'password'
    require_tls: true

from fastapi import FastAPI
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
import requests

# 初始化追踪提供者
trace.set_tracer_provider(TracerProvider())

# 配置 Jaeger 导出器
jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)

# 配置批量 Span 处理器
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(jaeger_exporter)
)

app = FastAPI(title="MCP Server", version="2.0")

# 初始化 FastAPI 追踪
FastAPIInstrumentor.instrument_app(app)

# 获取追踪器
tracer = trace.get_tracer(__name__)

# 示例路由：调用外部服务
@app.get("/api/v1/external-call")
async def external_call():
    with tracer.start_as_current_span("external_call") as span:
        # 设置 Span 标签
        span.set_attribute("service.name", "mcp-server")
        span.set_attribute("endpoint", "/api/v1/external-call")
        
        # 添加 Span 事件
        span.add_event("Calling external service", {
            "external_service": "example-service",
            "url": "http://localhost:9000/api/v1/data"
        })
        
        try:
            # 调用外部服务
            response = requests.get("http://localhost:9000/api/v1/data")
            response.raise_for_status()
            
            # 设置 Span 状态
            span.set_status(trace.Status(trace.StatusCode.OK))
            
            return {"result": response.json()}
        except Exception as e:
            # 设置 Span 错误状态
            span.set_status(trace.Status(trace.StatusCode.ERROR))
            span.record_exception(e)
            raise

# 示例路由：工具调用
tool_execution_span_counter = 0

@app.post("/api/v1/tools/{tool_name}/execute")
async def execute_tool(tool_name: str):
    global tool_execution_span_counter
    tool_execution_span_counter += 1
    
    # 创建工具执行 Span
    with tracer.start_as_current_span(f"execute_tool:{tool_name}") as span:
        span.set_attribute("tool.name", tool_name)
        span.set_attribute("tool.execution.id", tool_execution_span_counter)
        
        # 模拟工具执行延迟
        import time
        time.sleep(0.1)
        
        # 记录工具执行结果
        span.add_event("Tool execution completed", {
            "tool.name": tool_name,
            "result": "success",
            "execution.time": 0.1
        })
        
        return {"tool_name": tool_name, "result": "success", "execution_id": tool_execution_span_counter}

from opentelemetry import trace
from opentelemetry.propagators import inject, extract
import requests

# 获取追踪器
tracer = trace.get_tracer(__name__)

# 发送请求时注入上下文
def send_request(url):
    with tracer.start_as_current_span("send_request") as span:
        # 创建 HTTP 请求头
        headers = {}
        
        # 注入追踪上下文到 HTTP 头
        inject(headers)
        
        # 发送 HTTP 请求
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        
        return response.json()

# 接收请求时提取上下文
@app.get("/api/v1/resource")
async def get_resource(request):
    # 从请求头提取追踪上下文
    carrier = {
        key.lower(): value for key, value in request.headers.items()
    }
    ctx = extract(carrier)
    
    # 使用提取的上下文创建 Span
    with tracer.start_as_current_span("get_resource", context=ctx) as span:
        span.set_attribute("resource.name", "test-resource")
        
        return {"resource": "test-data"}

# docker-compose.yml
grafana:
  image: grafana/grafana:latest
  ports:
    - "3000:3000"
  volumes:
    - grafana_data:/var/lib/grafana
  environment:
    - GF_SECURITY_ADMIN_PASSWORD=admin
  depends_on:
    - prometheus
    - jaeger
    - elasticsearch

prometheus:
  image: prom/prometheus:latest
  ports:
    - "9090:9090"
  volumes:
    - ./prometheus.yml:/etc/prometheus/prometheus.yml
    - prometheus_data:/prometheus

jaeger:
  image: jaegertracing/all-in-one:latest
  ports:
    - "6831:6831/udp"
    - "16686:16686"
    - "14268:14268"
  volumes:
    - jaeger_data:/var/lib/jaeger

elasticsearch:
  image: docker.elastic.co/elasticsearch/elasticsearch:7.17.0
  ports:
    - "9200:9200"
  environment:
    - discovery.type=single-node
    - ES_JAVA_OPTS=-Xms512m -Xmx512m
  volumes:
    - elasticsearch_data:/usr/share/elasticsearch/data

volumes:
  grafana_data:
  prometheus_data:
  jaeger_data:
  elasticsearch_data:

# otel-collector-config.yml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  batch:
    timeout: 1s
    send_batch_size: 1024

  memory_limiter:
    check_interval: 1s
    limit_mib: 4000

  attributes:
    actions:
      - key: service.name
        value: mcp-server
        action: insert

  resource:
    attributes:
      - key: deployment.environment
        value: production
        action: insert

exporters:
  otlp/jaeger:
    endpoint: jaeger:4317
    tls:
      insecure: true

  prometheus:
    endpoint: 0.0.0.0:8889
    resource_to_telemetry_conversion:
      enabled: true

  logging:
    loglevel: debug

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, batch, attributes, resource]
      exporters: [otlp/jaeger, logging]
    metrics:
      receivers: [otlp]
      processors: [memory_limiter, batch, attributes, resource]
      exporters: [prometheus, logging]
    logs:
      receivers: [otlp]
      processors: [memory_limiter, batch, attributes, resource]
      exporters: [logging]

GET /api/v1/logs
Authorization: Bearer <api-key>
Content-Type: application/json

{
  "level": "info",
  "start_time": "2026-01-01T00:00:00Z",
  "end_time": "2026-01-01T23:59:59Z",
  "limit": 100,
  "offset": 0
}

200 OK
Content-Type: application/json

{
  "total": 1500,
  "logs": [
    {
      "timestamp": "2026-01-01T12:00:00Z",
      "level": "info",
      "logger": "mcp_server",
      "message": "MCP Server started successfully",
      "metadata": {
        "version": "2.0",
        "port": 8000
      }
    },
    {
      "timestamp": "2026-01-01T12:01:00Z",
      "level": "info",
      "logger": "mcp_server.tools",
      "message": "Tool executed successfully",
      "metadata": {
        "tool_name": "hello_world",
        "execution_time": 0.1,
        "result": "Hello, World!"
      }
    }
  ]
}

GET /api/v1/metrics
Authorization: Bearer <api-key>
Content-Type: application/json

{
  "metrics": ["cpu_usage", "memory_usage", "request_count"],
  "start_time": "2026-01-01T00:00:00Z",
  "end_time": "2026-01-01T23:59:59Z",
  "interval": "5m"
}

200 OK
Content-Type: application/json

{
  "metrics": [
    {
      "name": "cpu_usage",
      "data": [
        {"timestamp": "2026-01-01T00:00:00Z", "value": 45.5},
        {"timestamp": "2026-01-01T00:05:00Z", "value": 47.2},
        {"timestamp": "2026-01-01T00:10:00Z", "value": 43.8}
      ]
    },
    {
      "name": "memory_usage",
      "data": [
        {"timestamp": "2026-01-01T00:00:00Z", "value": 65.3},
        {"timestamp": "2026-01-01T00:05:00Z", "value": 66.7},
        {"timestamp": "2026-01-01T00:10:00Z", "value": 67.2}
      ]
    },
    {
      "name": "request_count",
      "data": [
        {"timestamp": "2026-01-01T00:00:00Z", "value": 120},
        {"timestamp": "2026-01-01T00:05:00Z", "value": 135},
        {"timestamp": "2026-01-01T00:10:00Z", "value": 142}
      ]
    }
  ]
}

GET /api/v1/traces/{trace_id}
Authorization: Bearer <api-key>
Content-Type: application/json

200 OK
Content-Type: application/json

{
  "trace_id": "1234567890abcdef",
  "spans": [
    {
      "span_id": "span1",
      "parent_span_id": null,
      "name": "mcp_server_request",
      "start_time": "2026-01-01T12:00:00Z",
      "end_time": "2026-01-01T12:00:01Z",
      "duration": 1000,
      "status": "ok",
      "attributes": {
        "service.name": "mcp-server",
        "endpoint": "/api/v1/tools/hello_world/execute",
        "method": "POST",
        "status_code": 200
      },
      "events": [
        {
          "timestamp": "2026-01-01T12:00:00Z",
          "name": "request_received",
          "attributes": {
            "client_ip": "192.168.1.1",
            "user_agent": "MCP Client/2.0"
          }
        },
        {
          "timestamp": "2026-01-01T12:00:01Z",
          "name": "request_completed",
          "attributes": {
            "response_time": 1000
          }
        }
      ]
    },
    {
      "span_id": "span2",
      "parent_span_id": "span1",
      "name": "execute_tool:hello_world",
      "start_time": "2026-01-01T12:00:00Z",
      "end_time": "2026-01-01T12:00:01Z",
      "duration": 800,
      "status": "ok",
      "attributes": {
        "tool.name": "hello_world",
        "tool.version": "1.0.0"
      },
      "events": [
        {
          "timestamp": "2026-01-01T12:00:00Z",
          "name": "tool_execution_started",
          "attributes": {
            "tool.name": "hello_world"
          }
        },
        {
          "timestamp": "2026-01-01T12:00:01Z",
          "name": "tool_execution_completed",
          "attributes": {
            "tool.name": "hello_world",
            "result": "Hello, World!",
            "execution_time": 800
          }
        }
      ]
    }
  ]
}