04 · 演示应用与 APM 采集：中间件是平台的心脏

"""
forum/models.py — 演示论坛的数据模型

注意：这里刻意保留了几个"性能陷阱"设计（与真实 V2EX 求助场景一致）：
1. Topic.category 没有加 db_index —— 列表页按分类过滤时会全表扫描；
2. Topic.author / Reply.topic / Reply.author 都是普通外键 ——
   不使用 select_related 时每访问一次都会触发一条额外 SQL。
"""
from django.conf import settings
from django.db import models


class Topic(models.Model):
    """论坛帖子"""

    title = models.CharField('标题', max_length=200)
    content = models.TextField('正文')
    # 陷阱：未加索引的过滤字段
    category = models.CharField('版块', max_length=32, default='闲聊')
    author = models.ForeignKey(
        settings.AUTH_USER_MODEL, on_delete=models.CASCADE,
        related_name='topics', verbose_name='作者',
    )
    views = models.PositiveIntegerField('浏览量', default=0)
    created_at = models.DateTimeField('发布时间', auto_now_add=True)

    class Meta:
        verbose_name = '帖子'
        verbose_name_plural = verbose_name
        ordering = ['-created_at']

    def __str__(self):
        return self.title


class Reply(models.Model):
    """帖子回复"""

    topic = models.ForeignKey(
        Topic, on_delete=models.CASCADE, related_name='replies', verbose_name='所属帖子',
    )
    author = models.ForeignKey(
        settings.AUTH_USER_MODEL, on_delete=models.CASCADE,
        related_name='replies', verbose_name='回复人',
    )
    content = models.TextField('回复内容')
    created_at = models.DateTimeField('回复时间', auto_now_add=True)

    class Meta:
        verbose_name = '回复'
        verbose_name_plural = verbose_name
        ordering = ['created_at']

    def __str__(self):
        return f'回复@{self.topic_id}'

"""
forum/views.py — 演示目标应用的视图

提供"问题版 / 优化版"两个路由做同功能对比：
- /forum/problem/   故意埋了典型性能问题
- /forum/optimized/ 用 Django 标准手段修复后的问题

两个视图渲染同一个模板，页面底部会显示本次请求的 SQL 查询数与耗时，
方便直观对比。
"""
import time

from django.db import connection
from django.db.models import Count
from django.shortcuts import get_object_or_404, render

from .models import Reply, Topic


def _request_debug(start):
    """返回本次请求的 SQL 统计信息（给模板底部展示用）"""
    return {
        'sql_count': len(connection.queries),
        'duration_ms': round((time.perf_counter() - start) * 1000, 1),
    }


# ---------------------------------------------------------------
# 问题版：V2EX 上那位楼主写的代码（还原度 100%）
# ---------------------------------------------------------------
def topic_list_problem(request):
    """帖子列表——问题版

    埋了 4 类典型问题：
    1. 无索引过滤：category 字段没有 db_index，且逐条在 Python 里过滤；
    2. N+1 查询：循环里逐条取作者、逐条 count 回复数；
    3. 重复计算：每条帖子都重新统计一遍"各版块帖子数"，结果却从未缓存；
    4. 无 limit 的 all()：把全表捞到内存后再切片。
    """
    start = time.perf_counter()

    topics = Topic.objects.all()          # 问题 4：全表加载，无 limit
    rows = []
    for t in topics:                      # 问题 2：循环体里都是查询
        if t.category != '闲聊':           # 问题 1：无索引字段过滤（还是 Python 层过滤）
            continue
        author_name = t.author.username   # 问题 2：每条触发一次作者查询（N+1）
        reply_count = Reply.objects.filter(topic=t).count()  # 问题 2：每条再 count 一次
        # 问题 3：重复计算——每条帖子都统计各版块帖子数，循环 100 次就算 100 遍
        stats = {
            c: Topic.objects.filter(category=c).count()
            for c in ('闲聊', '技术', '求助')
        }
        rows.append({
            'id': t.id,
            'title': t.title,
            'author': author_name,
            'category': t.category,
            'reply_count': reply_count,
            'views': t.views,
            'created_at': t.created_at,
            'stats': stats,               # 顺手把重复计算的结果塞进行数据
        })
    rows = rows[:20]                      # 捞了全表却只用前 20 条

    context = {
        'version': 'problem',
        'version_label': '问题版',
        'topics': rows,
        'debug': _request_debug(start),
    }
    return render(request, 'forum/topic_list.html', context)


# ---------------------------------------------------------------
# 优化版：同样的功能，标准 Django 优化手段
# ---------------------------------------------------------------
def topic_list_optimized(request):
    """帖子列表——优化版

    对应修复：
    1. 数据库层过滤（建议配合 Meta.indexes 为 category 加索引）；
    2. select_related 预取作者外键 + annotate 一次性统计回复数，消灭 N+1；
    3. 版块统计只算一次，并放在数据库聚合里完成；
    4. QuerySet 直接 [:20] 切片，SQL 带 LIMIT，不捞全表。
    """
    start = time.perf_counter()

    rows = list(
        Topic.objects
        .filter(category='闲聊')                     # 修复 1/4：数据库过滤
        .select_related('author')                    # 修复 2：JOIN 预取作者
        .annotate(reply_count=Count('replies'))      # 修复 2：聚合代替循环 count
        .values('id', 'title', 'category', 'views',
                'created_at', 'author__username', 'reply_count')[:20]  # 修复 4：LIMIT
    )
    # 修复 3：版块统计只算一次
    category_counts = dict(
        Topic.objects.values_list('category')
        .annotate(n=Count('id'))
        .values_list('category', 'n')
    )

    topics = [
        {
            'id': r['id'],
            'title': r['title'],
            'author': r['author__username'],
            'category': r['category'],
            'reply_count': r['reply_count'],
            'views': r['views'],
            'created_at': r['created_at'],
            'stats': category_counts,
        }
        for r in rows
    ]

    context = {
        'version': 'optimized',
        'version_label': '优化版',
        'topics': topics,
        'debug': _request_debug(start),
    }
    return render(request, 'forum/topic_list.html', context)


def topic_detail(request, pk):
    """帖子详情页：展示回复列表（顺手演示 reply 预取）"""
    start = time.perf_counter()
    topic = get_object_or_404(Topic.objects.select_related('author'), pk=pk)
    replies = topic.replies.select_related('author').all()
    context = {
        'topic': topic,
        'replies': replies,
        'debug': _request_debug(start),
    }
    return render(request, 'forum/topic_detail.html', context)

{% extends "base.html" %}
{% block title %}{{ version_label }}{% endblock %}
{% block content %}
<h1>论坛 · {{ version_label }}</h1>
<table border="1" cellpadding="6" style="border-collapse:collapse; background:#fff;">
  <tr><th>标题</th><th>作者</th><th>版块</th><th>回复</th><th>浏览</th></tr>
  {% for t in topics %}
  <tr>
    <td><a href="/forum/topic/{{ t.id }}/">{{ t.title }}</a></td>
    <td>{{ t.author }}</td>
    <td>{{ t.category }}</td>
    <td>{{ t.reply_count }}</td>
    <td>{{ t.views }}</td>
  </tr>
  {% endfor %}
</table>
<p><a href="/forum/problem/">问题版</a> ｜ <a href="/forum/optimized/">优化版</a></p>
<p><b>本次请求：</b>{{ debug.sql_count }} 条 SQL，{{ debug.duration_ms }} ms</p>
{% endblock %}

{% extends "base.html" %}
{% block title %}{{ topic.title }}{% endblock %}
{% block content %}
<h1>{{ topic.title }}</h1>
<p>{{ topic.content }}</p>
<hr>
<h2>回复（{{ replies.count }}）</h2>
{% for r in replies %}
  <p><b>{{ r.author.username }}</b>：{{ r.content }}</p>
{% endfor %}
<p><a href="/forum/problem/">返回问题版列表</a></p>
{% endblock %}

from django.urls import path

from . import views

app_name = 'forum'

urlpatterns = [
    path('problem/', views.topic_list_problem, name='problem'),
    path('optimized/', views.topic_list_optimized, name='optimized'),
    path('topic/<int:pk>/', views.topic_detail, name='detail'),
]

    # 演示目标应用（含性能问题的论坛 + 故障演练）
    path('forum/', include('forum.urls')),

from django.contrib.auth.models import User
from forum.models import Topic, Reply
import random

u, _ = User.objects.get_or_create(username='demo', defaults={'is_staff': False})
cats = ['闲聊', '技术', '求助']
for i in range(60):
    t = Topic.objects.create(
        title=f'第{i}帖：为什么我的 Django 这么慢？',
        content='循环里查数据库，越跑越卡，求各位指点。',
        category=random.choice(cats), author=u, views=random.randint(10, 2000))
    for j in range(random.randint(0, 5)):
        Reply.objects.create(topic=t, author=u, content=f'回复{j}：加个缓存试试')
print('done:', Topic.objects.count())

    # 自研请求计时中间件：采集耗时 / SQL / 慢查询 / 调用链 span，写入 RequestMetric
    'monitor.middleware.RequestTimingMiddleware',

"""
monitor/middleware.py — 自研请求计时中间件（APM 数据采集核心）

职责：
1. 统计每个请求的总耗时，生成本次请求的 TraceID；
2. 通过 force_debug_cursor 让 Django 记录本请求内的全部 SQL（不依赖 DEBUG 开关），
   并把每条 SQL 的时间切片组装成"调用链 span"（请求 span + SQL span + 视图/渲染 span）；
3. 筛出超过阈值的慢查询，连同耗时、查询数一起交给批量缓冲（monitor/buffer.py）落库；
4. 捕获错误请求（状态码 >= 400 / 视图抛异常），供错误率统计与告警；
5. 生成一个"CPU 占用"推算指标，供面板折线图演示（真实主机 CPU 见主机监控）。

注意：跳过平台自身路径（settings.OBSERVABILITY.SKIP_PATHS），避免面板请求污染采集数据。
"""
import json
import logging
import random
import re
import time

from django.conf import settings
from django.core.exceptions import PermissionDenied, SuspiciousOperation
from django.db import connection
from django.http import Http404

from . import buffer, tracing

MON = settings.OBSERVABILITY
SLOW_QUERY_MS = MON['SLOW_QUERY_MS']
SKIP_PATHS = MON['SKIP_PATHS']
# 只有确认部署在可信反向代理之后才设 OBS_TRUST_XFORWARDED_FOR=1，
# 否则客户端可伪造 XFF 头污染 IP 统计与审计
TRUST_XFF = __import__('os').environ.get('OBS_TRUST_XFORWARDED_FOR') == '1'

logger = logging.getLogger(__name__)

# SQL 文本中的敏感字面量脱敏：password='xxx' / secret_key: xxx / token LIKE 'sk-%' 等。
# 关键词允许带常见后缀（password_hash / api_key_id）并原样保留后缀（避免 author_id
# 这类普通列名被改写——不收录裸 auth/pass 等易误伤词，改用 auth_token/authorization）。
_SENSITIVE_SQL_RE = re.compile(
    r"(?i)\b(password|passwd|pwd|secret|token|credential|api[_-]?key|authorization|auth[_-]?token)"
    r"([a-z0-9_-]*)"
    r"(\s*(?:=|!=|<>|like|ilike|in)\s*|\s*:\s*)"
    r"('[^']*'|\"[^\"]*\"|\([^)]*\)|[^\s,)]+)"
)


def _sanitize_sql(sql):
    """脱敏 SQL 中的敏感字面量（落库展示前调用，避免凭据进入监控数据）；
    列名后缀原样保留：password_hash='x' -> password_hash='***'"""
    return _SENSITIVE_SQL_RE.sub(r"\1\2\3'***'", sql)


def _client_ip(request):
    """取客户端 IP：默认只信 REMOTE_ADDR；仅在可信反代后开启 OBS_TRUST_XFORWARDED_FOR"""
    if TRUST_XFF:
        fwd = request.META.get('HTTP_X_FORWARDED_FOR')
        if fwd:
            return fwd.split(',')[0].strip()[:60]
    return (request.META.get('REMOTE_ADDR') or '')[:60]


def _status_for_exception(exc):
    """把 Django 信号异常映射为真实状态码，避免 404/403 被计成 500 污染错误率"""
    if isinstance(exc, Http404):
        return 404
    if isinstance(exc, PermissionDenied):
        return 403
    if isinstance(exc, SuspiciousOperation):
        return 400
    return 500


class RequestTimingMiddleware:
    """记录每个请求的耗时 / SQL 查询数 / 慢查询 / 调用链 span / TraceID"""

    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # 平台自身路径不采集
        if request.path.startswith(SKIP_PATHS):
            return self.get_response(request)

        start = time.perf_counter()
        # W3C Trace Context：采纳上游 trace_id（跨服务串联）或生成新的；
        # 同时绑定到 contextvars，请求内的日志会自动携带 trace_id
        trace_id = tracing.bind_request(request)
        # 打开 SQL 调试游标：无论 DEBUG 与否都会记录 SQL 到 connection.queries
        connection.force_debug_cursor = True
        try:
            response = self.get_response(request)
            status = getattr(response, 'status_code', 0)
        except Exception as exc:  # 视图抛异常：记录一条指标后原样抛出
            elapsed_ms = (time.perf_counter() - start) * 1000
            self._save(request, _status_for_exception(exc), elapsed_ms,
                       error=type(exc).__name__,  # 只留异常类名，不落异常消息（可能含敏感上下文）
                       queries=list(connection.queries),
                       trace_id=trace_id)
            raise
        finally:
            queries = list(connection.queries)
            # 关闭调试游标，避免后续内部请求（如本次落库）也被记录
            connection.force_debug_cursor = False
            tracing.unbind_request()

        elapsed_ms = (time.perf_counter() - start) * 1000
        # 调用方可凭响应头拿到本次链路 ID：X-Trace-Id 便于人工排障，
        # traceresponse 为 W3C Trace Context 规范草案的响应头形式
        response['X-Trace-Id'] = trace_id
        response['traceresponse'] = tracing.format_traceparent(
            trace_id, tracing.new_span_id())
        self._save(request, status, elapsed_ms, queries=queries,
                   trace_id=trace_id)
        return response

    def _save(self, request, status, elapsed_ms, error=None, queries=None,
              trace_id=''):
        queries = queries or []
        sql_total_ms = 0.0
        slow = []
        spans = []
        offset = 0.0
        for q in queries:
            t_ms = float(q.get('time') or 0) * 1000
            sql_total_ms += t_ms
            sql_text = ' '.join((q.get('sql') or '').split())[:120]
            spans.append({'kind': 'sql', 'name': _sanitize_sql(sql_text),
                          'off': round(offset, 1), 'dur': round(t_ms, 2)})
            offset += t_ms
            if t_ms > SLOW_QUERY_MS:
                slow.append({'sql': _sanitize_sql(q['sql'])[:500], 'time_ms': round(t_ms, 1)})

        # 视图与渲染 span：总耗时减去 SQL 时间（近似切分，页面已有标注）
        view_ms = max(0.0, elapsed_ms - sql_total_ms)
        spans.insert(0, {'kind': 'request', 'name': f'{request.method} {request.path}',
                         'off': 0.0, 'dur': round(elapsed_ms, 2)})
        spans.append({'kind': 'view', 'name': (getattr(request.resolver_match, 'view_name', '') or '')[:80],
                      'off': round(sql_total_ms, 1), 'dur': round(view_ms, 2)})

        # 模拟 CPU 占用：由耗时与查询数推算（仅演示用，真实主机 CPU 见主机监控）
        cpu = min(95.0, elapsed_ms / 25 + len(queries) * 0.6 + random.uniform(2, 12))

        # TraceID：上游有 traceparent 时采纳上游 trace_id（跨服务同链路），
        # 否则为本次请求新生成的 W3C 128bit id
        trace_id = trace_id or getattr(request, 'obs_trace_id', '') or tracing.new_trace_id()
        try:
            # 交给批量缓冲（monitor/buffer.py，05 篇实现）：请求线程不碰数据库写锁。
            # 缓冲关闭时 submit 内部退化为同步写库——所以本篇先写"占位缓冲"也能跑
            buffer.submit({
                'path': request.path[:200],
                'method': request.method,
                'status_code': status,
                'duration_ms': round(elapsed_ms, 2),
                'sql_count': len(queries),
                'sql_time_ms': round(sql_total_ms, 2),
                'slow_queries': json.dumps(slow, ensure_ascii=False),
                'slow_query_count': len(slow),
                'cpu_percent': round(cpu, 1),
                'trace_id': trace_id,
                'view_name': (getattr(request.resolver_match, 'view_name', '') or '')[:100],
                'is_error': (status >= 400) or (error is not None),
                'client_ip': ip,   # ← 见下方 geoip 说明：本篇先用 ip 变量，05 篇前补 geoip.resolve
                'geo_province': '',
                'geo_city': '',
                'spans': json.dumps(spans, ensure_ascii=False),
            })
        except Exception:
            # 采集失败不影响正常请求，但不能静默——平台自己就是监控工具
            logger.exception('RequestMetric 采集提交失败 path=%s', request.path)

"""
monitor/tracing.py — W3C Trace Context 支持（跨服务链路关联）

1. 上游服务按 W3C 规范携带 `traceparent: 00-<trace-id>-<parent-span-id>-<flags>`
   请求头访问平台时，中间件**采纳上游 trace_id**（同一调用链跨服务可串联）；
2. 无上游链路时生成符合规范的 128bit trace_id；
3. 响应回写 `traceresponse`（W3C 规范草案）与 `X-Trace-Id`；
4. 请求处理期间通过 contextvars 暴露当前 trace_id：
   - 日志 Handler 把 trace_id 写入 LogEntry，实现"日志 ↔ 调用链"按 ID 互查；
   - 后台线程 / shell 中 context 为空，日志不携带 trace 前缀，互不污染。

纯标准库实现；协议细节见 https://www.w3.org/TR/trace-context/
"""
import contextvars
import logging
import secrets

# W3C traceparent：version(2hex)-trace-id(32hex)-parent-span-id(16hex)-flags(2hex)
# version 0xff 非法；trace-id / span-id 全零非法（规范 §3.2 / §3.3）
_TRACEPARENT_LEN = 55
_HEX = set('0123456789abcdef')

_current_trace_id: contextvars.ContextVar[str] = contextvars.ContextVar('obs_trace_id', default='')


def _is_hex(value, n):
    return len(value) == n and all(c in _HEX for c in value)


def new_trace_id():
    """生成符合 W3C 规范的 128bit trace_id（32 个小写 hex，非全零）"""
    tid = secrets.token_hex(16)
    return tid if tid.strip('0') else new_trace_id()


def new_span_id():
    """生成 64bit span_id（16 个小写 hex，非全零）"""
    sid = secrets.token_hex(8)
    return sid if sid.strip('0') else new_span_id()


def parse_traceparent(value):
    """解析并校验 traceparent 头。

    合法返回 {'trace_id':…, 'span_id':…, 'flags':…}；任何不规范输入返回 None
    （规范要求：无法解析时不得采信，也不得因此拒绝请求）。
    """
    if not value or len(value) != _TRACEPARENT_LEN:
        return None
    parts = value.split('-')
    if len(parts) != 4:
        return None
    version, trace_id, span_id, flags = parts
    if not _is_hex(version, 2) or version == 'ff':
        return None
    if not _is_hex(trace_id, 32) or trace_id.strip('0') == '':
        return None
    if not _is_hex(span_id, 16) or span_id.strip('0') == '':
        return None
    if not _is_hex(flags, 2):
        return None
    return {'trace_id': trace_id, 'span_id': span_id, 'flags': flags}


def format_traceparent(trace_id, span_id, sampled=True):
    return f'00-{trace_id}-{span_id}-{"01" if sampled else "00"}'


def bind_request(request):
    """从请求中提取/生成 trace 上下文，并绑定到当前执行上下文。

    返回 trace_id（32 hex）。写的是 contextvars 而不是 threading.local——
    async 视图（ASGI）下同样正确。
    """
    ctx = parse_traceparent(request.headers.get('traceparent', ''))
    trace_id = ctx['trace_id'] if ctx else new_trace_id()
    request.obs_trace_id = trace_id
    request.obs_parent_span_id = ctx['span_id'] if ctx else ''
    _current_trace_id.set(trace_id)
    return trace_id


def unbind_request():
    """请求结束后清除绑定，避免同线程的下一个请求读到过期 trace"""
    _current_trace_id.set('')


def current_trace_id():
    """当前执行上下文关联的 trace_id（无则空串）"""
    return _current_trace_id.get()


class TraceLogFilter(logging.Filter):
    """给日志记录附加 obs_trace_id 属性（当前请求有链路时才有值）"""

    def filter(self, record):
        record.obs_trace_id = current_trace_id()
        return True


class TraceFormatter(logging.Formatter):
    """控制台日志格式：请求上下文中的日志追加 [trace=...] 后缀，便于本地排障"""

    def format(self, record):
        text = super().format(record)
        trace_id = getattr(record, 'obs_trace_id', '')
        if trace_id:
            text = f'{text} [trace={trace_id}]'
        return text

LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'filters': {
        'trace': {'()': 'monitor.tracing.TraceLogFilter'},
    },
    'formatters': {
        'obs': {'()': 'monitor.tracing.TraceFormatter',
                'format': '%(levelname)s %(name)s %(message)s'},
    },
    'handlers': {
        'console': {'class': 'logging.StreamHandler', 'filters': ['trace'],
                    'formatter': 'obs'},
    },
    'root': {'handlers': ['console'], 'level': 'INFO'},
}

"""
monitor/buffer.py — 请求指标缓冲（本篇占位版：直接同步写库）
05 篇会换成"队列 + 后台批量落库"的完整版，接口保持不变。
"""
from django.db import transaction


def write_rows(rows, retries=0):
    """把一批字段字典写入库，返回落库条数"""
    from .models import RequestMetric
    if not rows:
        return 0
    objs = [RequestMetric(**r) for r in rows]
    with transaction.atomic():
        RequestMetric.objects.bulk_create(objs)
    return len(objs)


def submit(payload):
    """提交一条请求指标（本篇占位：同步写库）"""
    from django.utils import timezone
    payload.setdefault('created_at', timezone.now())
    write_rows([payload])
    return 'written'

python manage.py makemigrations forum
python manage.py migrate
python manage.py runserver 8014

from monitor.models import RequestMetric
for m in RequestMetric.objects.all()[:5]:
    print(m.method, m.path, m.status_code, m.duration_ms, 'ms,',
          m.sql_count, '条SQL, trace=', m.trace_id[:12],
          ', spans=', len(__import__('json').loads(m.spans)))

GET /forum/problem/ 200 312.5 ms, 243 条SQL, trace= 9f3c…, spans= 245
GET /forum/optimized/ 200 18.2 ms, 2 条SQL, trace= 4a1e…, spans= 4

# Windows
curl.exe -s -D - -o NUL -H "traceparent: 00-11111111111111111111111111111111-2222222222222222-01" http://127.0.0.1:8014/api/health/

# macOS / Linux
curl -s -D - -o /dev/null -H "traceparent: 00-11111111111111111111111111111111-2222222222222222-01" http://127.0.0.1:8014/api/health/