07 · 日志服务与前端 RUM：日志会说话，浏览器会举报

上一篇：06 主机监控与多机 Agent ｜目录：README ｜下一篇：08 告警中心

这一篇补齐四路采集的后两路：

日志服务（loghub）：把 Python logging 体系整个"接"进平台——业务代码零改造， WARNING 以上自动入库，且每条日志自动带上当前请求的 trace_id，实现日志 ↔ 调用链互查（可观测性三支柱的"串联"就在这一步落地）；
前端 RUM（rum）：自研一个 11KB 的浏览器 SDK（rum.js），采集 PV、加载耗时（TTFB/FCP/LCP）、JS 异常、API 调用——服务端监控看不见的"用户体验"从这里来。

一、日志服务

1.1 模型：一张宽表

loghub/models.py：

"""
loghub/models.py — 日志服务的存储模型
"""
from django.db import models
from django.utils import timezone


class LogEntry(models.Model):
    """一条日志：来源可以是系统 Handler / 接入 API / 前端 RUM"""

    SOURCES = [
        ('system', '系统'),
        ('app', '应用'),
        ('api', '接入API'),
        ('rum', '前端RUM'),
    ]
    source = models.CharField('来源', max_length=16, choices=SOURCES, default='app', db_index=True)
    level = models.CharField('级别', max_length=10, db_index=True,
                             default='INFO')  # DEBUG/INFO/WARNING/ERROR/CRITICAL
    logger = models.CharField('记录器', max_length=120, blank=True, default='')
    message = models.TextField('内容')
    extra = models.TextField('附加信息(JSON)', blank=True, default='')
    # 请求上下文日志自动携带（monitor.tracing 绑定）；接入 API 可按 W3C traceparent 关联，
    # 实现"日志 ↔ APM 调用链"按同一 trace_id 互查
    trace_id = models.CharField('TraceID', max_length=32, blank=True, default='', db_index=True)
    created_at = models.DateTimeField('时间', default=timezone.now, db_index=True)

    class Meta:
        verbose_name = '日志'
        verbose_name_plural = verbose_name
        ordering = ['-created_at']

    def __str__(self):
        return f'[{self.level}] {self.message[:40]}'

1.2 Handler：让日志"自动"入库

loghub/handler.py：

"""
loghub/handler.py — 把 Python logging 体系接入日志服务

SQLiteLogHandler 挂到 root logger（WARNING 及以上），
Django / django.request 的异常与告警会自动进入日志库，无需业务代码改造。
"""
import json
import logging
import traceback

from monitor.tracing import current_trace_id

from .models import LogEntry

_ATTACHED = False


class SQLiteLogHandler(logging.Handler):
    """写入 LogEntry 的 logging.Handler（写库失败时静默，避免日志风暴拖垮请求）"""

    def emit(self, record):
        try:
            message = self.format(record)[:4000]
            extra = {}
            if record.exc_info:
                extra['exc'] = ''.join(traceback.format_exception(*record.exc_info))[:4000]
            LogEntry.objects.create(
                source=getattr(record, 'obs_source', None) or 'system',
                level=record.levelname,
                logger=record.name[:120],
                message=message,
                extra=json.dumps(extra, ensure_ascii=False) if extra else '',
                # 请求上下文中的日志自动携带链路 ID（后台线程/CLI 为空串）
                trace_id=current_trace_id()[:32],
            )
        except Exception:
            self.handleError(record)


def attach_log_handler():
    """挂载日志 Handler（幂等；WARNING 及以上入库）"""
    global _ATTACHED
    if _ATTACHED:
        return
    root = logging.getLogger()
    # 防重复挂载（跨 reload 的进程内存标记 + 实例标记双保险）
    for h in root.handlers:
        if isinstance(h, SQLiteLogHandler):
            _ATTACHED = True
            return
    handler = SQLiteLogHandler()
    handler.setLevel(logging.WARNING)
    handler.setFormatter(logging.Formatter('%(levelname)s %(name)s: %(message)s'))
    root.addHandler(handler)
    _ATTACHED = True

三个设计决定：

挂在 root logger、级别 WARNING：Django 和所有三方库的日志最终都会冒泡到 root—— 挂一次，全应用覆盖；业务想主动上报用 logging.getLogger('app').warning(...) 即可， 业务代码不需要知道平台的存在；
emit 里全部 try/except：日志写库失败绝不能打断业务请求（handleError 静默兜底）。反过来的教训很多：日志系统故障放大成业务故障，是最冤的 P0；
trace_id=current_trace_id()：02/04 篇埋的 contextvars 在这里兑现—— 请求线程里任何代码打的 WARNING+ 日志自动携带链路 ID。这就是"日志与调用链互查"的全部实现。

1.3 挂载时机与路由

loghub/apps.py（在 app 就绪时挂 Handler，幂等防 runserver 双进程重复挂）：

from django.apps import AppConfig


class LoghubConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'loghub'
    verbose_name = '日志服务'

    def ready(self):
        from .handler import attach_log_handler
        attach_log_handler()

loghub/urls.py 与查询页视图（查询页用 04/05 篇的注册表思路直接查 LogEntry；完整版见仓库 loghub/views.py——关键字 LIKE 通配符转义、级别直方图、分页）：

from django.urls import path

from . import views

urlpatterns = [
    path('', views.log_page, name='logs'),
]

教学版 loghub/views.py（查询页先只做"最近 N 分钟 + 关键字 + 级别"三要素）：

"""
loghub/views.py — 日志查询页（教学版：搜索三要素）
"""
from datetime import timedelta

from django.shortcuts import render
from django.utils import timezone

from .models import LogEntry


def log_page(request):
    """日志查询页：关键字 / 级别 / 来源 过滤 + 级别直方图数据"""
    q = (request.GET.get('q') or '').strip()
    level = (request.GET.get('level') or '').strip()

    qs = LogEntry.objects.filter(created_at__gte=timezone.now() - timedelta(minutes=1440))
    if q:
        # LIKE 通配符转义，防止 %/_ 被用户输入放大扫描成本
        safe_q = q.replace('\\', '\\\\').replace('%', r'\%').replace('_', r'\_')
        qs = qs.filter(message__icontains=safe_q)
    if level:
        qs = qs.filter(level=level)

    recent = qs[:100]
    return render(request, 'loghub/logs.html', {'recent': recent, 'q': q, 'level': level})

templates/loghub/logs.html：

{% extends "base.html" %}
{% block title %}日志查询{% endblock %}
{% block content %}
<h1>日志查询</h1>
<form method="get">
  <input name="q" value="{{ q }}" placeholder="关键字" style="width:300px">
  <select name="level">
    <option value="">全部级别</option>
    {% for lv in "DEBUG,INFO,WARNING,ERROR,CRITICAL"|split:"," %}<option {% if level == lv %}selected{% endif %}>{{ lv }}</option>{% endfor %}
  </select>
  <button>搜索</button>
</form>
<table border="1" cellpadding="4" style="border-collapse:collapse;background:#fff;margin-top:12px;">
  <tr><th>时间</th><th>级别</th><th>来源</th><th>TraceID</th><th>内容</th></tr>
  {% for r in recent %}
  <tr>
    <td>{{ r.created_at|date:"H:i:s" }}</td>
    <td><b>{{ r.level }}</b></td>
    <td>{{ r.get_source_display }}</td>
    <td style="font-family:monospace">{{ r.trace_id|truncatechars:12 }}</td>
    <td>{{ r.message|truncatechars:120 }}</td>
  </tr>
  {% endfor %}
</table>
{% endblock %}

（split 过滤器 Django 没有内置——把 <select> 换成五个写死的 <option> 即可，或自定义 simple_filter；教学版建议写死。）

在 config/urls.py 挂载：path('logs/', include('loghub.urls')),。

1.4 验证：一行代码都不用改的"接入"

重启服务后，故意访问一个不存在的路径触发 404，再看 shell：

from loghub.models import LogEntry
print(LogEntry.objects.count())                       # > 0
print(LogEntry.objects.order_by('-id').values('level','source','trace_id').first())

Django 自己的请求日志（django.request 的 WARNING+）已经被自动收进来了—— 这就是"接入日志服务不需要业务改造"的含义。trace_id 列此时多为空（404 由门禁产生，不在请求计时上下文里），08 篇故障演练一打，ERROR 日志就会带着 trace_id 出现。

二、前端 RUM

2.1 模型：六类事件一张宽表

rum/models.py（完整字段见仓库；核心结构）：

"""
rum/models.py — 前端性能监控（RUM）事件存储

一个宽表存全部事件类型：pv / perf / api / error / resource / custom。
浏览器端自研 SDK（static/rum.js）采集，POST /rum/beacon/ 批量上报。
"""
from django.db import models
from django.utils import timezone


class RumEvent(models.Model):
    TYPES = [
        ('pv', '页面访问'),
        ('perf', '页面性能'),
        ('api', 'API调用'),
        ('error', 'JS错误'),
        ('resource', '静态资源'),
        ('custom', '自定义事件'),
    ]

    type = models.CharField('事件类型', max_length=16, choices=TYPES, db_index=True)
    app = models.CharField('应用标识', max_length=32, default='forum')
    page_url = models.CharField('页面URL', max_length=256, blank=True, default='')
    referrer = models.CharField('来源页', max_length=256, blank=True, default='')
    session_id = models.CharField('会话ID', max_length=64, blank=True, default='', db_index=True)
    device = models.CharField('浏览器/系统', max_length=80, blank=True, default='')
    screen = models.CharField('分辨率', max_length=20, blank=True, default='')

    # perf：各阶段耗时（毫秒，可空）
    ttfb_ms = models.FloatField('TTFB(ms)', null=True, blank=True)
    dom_ready_ms = models.FloatField('DOMReady(ms)', null=True, blank=True)
    load_ms = models.FloatField('完整加载(ms)', null=True, blank=True)
    fp_ms = models.FloatField('首次绘制FP(ms)', null=True, blank=True)
    fcp_ms = models.FloatField('首次内容绘制FCP(ms)', null=True, blank=True)
    lcp_ms = models.FloatField('最大内容绘制LCP(ms)', null=True, blank=True)

    # api：请求明细
    api_url = models.CharField('API地址', max_length=256, blank=True, default='')
    api_method = models.CharField('请求方法', max_length=10, blank=True, default='')
    api_status = models.IntegerField('状态码', null=True, blank=True)
    api_duration_ms = models.FloatField('API耗时(ms)', null=True, blank=True)
    api_ok = models.BooleanField('是否成功', default=True)

    # error：异常明细
    err_message = models.CharField('错误信息', max_length=300, blank=True, default='')
    err_stack = models.TextField('错误堆栈', blank=True, default='')

    # resource：资源明细
    r_type = models.CharField('资源类型', max_length=16, blank=True, default='')
    r_url = models.CharField('资源地址', max_length=256, blank=True, default='')
    r_duration_ms = models.FloatField('加载耗时(ms)', null=True, blank=True)
    r_size_kb = models.FloatField('传输大小(KB)', null=True, blank=True)

    # custom：自定义事件
    event_name = models.CharField('事件名', max_length=64, blank=True, default='')
    payload = models.TextField('负载(JSON)', blank=True, default='')

    created_at = models.DateTimeField('上报时间', default=timezone.now, db_index=True)

    class Meta:
        verbose_name = 'RUM事件'
        verbose_name_plural = verbose_name
        ordering = ['-created_at']

为什么宽表而不是每类一张表：六类事件的公共字段（页面/会话/设备/时间）占了大头，分表会让"按会话串一天的行为"变成六表 UNION。宽表 + 大量可空字段是 RUM 场景的常规取舍（12 篇会提到 PostgreSQL 下可用 JSONB 进一步优化）。

2.2 自研 rum.js：11KB 里装了什么

新建 static/rum.js（完整版约 300 行对照仓库同名文件；这里按"每个采集器"分块讲透）：

/**
 * rum.js — 自研前端性能监控 SDK（零依赖）
 *
 * 采集能力：
 *  - 页面访问 PV：首次加载 + SPA 路由变化（history hook）
 *  - 页面性能：TTFB / DOMReady / Load / FP / FCP / LCP（PerformanceObserver）
 *  - JS 异常：window.onerror + unhandledrejection（含堆栈）
 *  - API 监控：自动包装 XMLHttpRequest 与 fetch，记录 URL / 状态码 / 耗时
 *  - 静态资源：Resource Timing 中耗时 > 100ms 的条目
 *  - 自定义上报：window.obsRum('事件名', {任意 JSON})
 *
 * 使用：页面里先配置 window._OBS_RUM = { endpoint: '/rum/beacon/', app: 'forum' }
 * 再引入本文件即可。数据批量上报（5 秒 / 页面离开时），sendBeacon 优先。
 */
(function () {
  'use strict';
  var CFG = window._OBS_RUM || {};
  var ENDPOINT = CFG.endpoint || '/rum/beacon/';
  var APP = CFG.app || 'web';
  var queue = [];
  var flushTimer = null;

  // ---------- 会话标识（sessionStorage 保证"关标签页即新会话"） ----------
  var sessionId = (function () {
    try {
      var sid = sessionStorage.getItem('obs_sid');
      if (!sid) {
        sid = Math.random().toString(36).slice(2) + Date.now().toString(36);
        sessionStorage.setItem('obs_sid', sid);
      }
      return sid;
    } catch (e) { return 'anon-' + Date.now().toString(36); }
  })();

  // ---------- UA 解析：浏览器/系统 ----------
  function detectDevice() {
    var ua = navigator.userAgent;
    var browser = '未知浏览器';
    if (/Edg\//.test(ua)) browser = 'Edge';
    else if (/Chrome\//.test(ua)) browser = 'Chrome';
    else if (/Firefox\//.test(ua)) browser = 'Firefox';
    else if (/Safari\//.test(ua)) browser = 'Safari';
    var os = /Windows/.test(ua) ? 'Windows'
      : /Mac OS/.test(ua) ? 'macOS'
      : /Android/.test(ua) ? 'Android'
      : /iPhone|iPad/.test(ua) ? 'iOS' : 'Other';
    return browser + '/' + os;
  }

  var DEVICE = detectDevice();
  var SCREEN = (screen && screen.width ? screen.width + 'x' + screen.height : '');

  // ---------- 上报队列：批量 + sendBeacon 优先 ----------
  function enqueue(ev) {
    ev.app = APP;
    ev.session_id = sessionId;
    ev.device = DEVICE;
    ev.screen = SCREEN;
    ev.ts = Date.now();
    if (ev.page_url === undefined) ev.page_url = location.href.slice(0, 256);
    queue.push(ev);
    if (!flushTimer) flushTimer = setTimeout(flush, 5000);   // 攒 5 秒
    if (queue.length >= 20) flush();                          // 攒 20 条
  }

  function flush() {
    if (!queue.length) return;
    var batch = queue.splice(0, queue.length);
    clearTimeout(flushTimer); flushTimer = null;
    var body = JSON.stringify(batch);
    // sendBeacon：页面卸载也不丢包；不支持则退回 fetch keepalive
    if (navigator.sendBeacon) {
      navigator.sendBeacon(ENDPOINT, new Blob([body], { type: 'application/json' }));
    } else {
      fetch(ENDPOINT, { method: 'POST', body: body,
                        keepalive: true, headers: { 'Content-Type': 'application/json' } })
        .catch(function () {});
    }
  }

  // 页面离开时兜底上报（beforeunload 里 sendBeacon 是标准姿势）
  window.addEventListener('pagehide', flush);

  // ---------- PV：首次加载 + SPA 路由 hook ----------
  function sendPv() {
    enqueue({ type: 'pv' });
  }
  sendPv();
  (function hookHistory() {
    var _push = history.pushState;
    history.pushState = function () {
      _push.apply(this, arguments); sendPv();
    };
    window.addEventListener('popstate', sendPv);
  })();

  // ---------- 页面性能：导航计时 + FCP/LCP ----------
  function sendPerf() {
    var nav = performance.getEntriesByType('navigation')[0];
    if (!nav) return;
    var ev = {
      type: 'perf',
      ttfb_ms: Math.round(nav.responseStart),
      dom_ready_ms: Math.round(nav.domContentLoadedEventEnd),
      load_ms: Math.round(nav.loadEventEnd),
    };
    new PerformanceObserver(function (list) {
      var entries = list.getEntries();
      if (entries.length) {
        ev.fcp_ms = Math.round(entries[entries.length - 1].startTime);
        enqueue(Object.assign({}, ev));       // FCP 到了就先上报一版
      }
    }).observe({ type: 'paint', buffered: true });
    new PerformanceObserver(function (list) {
      var entries = list.getEntries();
      if (entries.length) {
        // LCP 取最后一次（内容可能逐段出现，规范定义"最终"的 LCP 才算数）
        ev.lcp_ms = Math.round(entries[entries.length - 1].startTime);
      }
    }).observe({ type: 'largest-contentful-paint', buffered: true });
    enqueue(ev);
  }
  window.addEventListener('load', function () { setTimeout(sendPerf, 0); });

  // ---------- JS 异常 ----------
  window.onerror = function (msg, src, line, col, err) {
    enqueue({
      type: 'error',
      err_message: String(msg).slice(0, 300),
      err_stack: err && err.stack ? String(err.stack).slice(0, 4000)
                  : (src + ':' + line + ':' + col),
    });
  };
  window.addEventListener('unhandledrejection', function (e) {
    enqueue({
      type: 'error',
      err_message: ('UnhandledPromise: ' + e.reason).slice(0, 300),
      err_stack: e.reason && e.reason.stack ? String(e.reason.stack).slice(0, 4000) : '',
    });
  });

  // ---------- API 监控：包装 XHR 与 fetch ----------
  var _xhrOpen = XMLHttpRequest.prototype.open;
  var _xhrSend = XMLHttpRequest.prototype.send;
  XMLHttpRequest.prototype.open = function (method, url) {
    this._obs = { method: method, url: String(url).slice(0, 256), start: Date.now() };
    return _xhrOpen.apply(this, arguments);
  };
  XMLHttpRequest.prototype.send = function () {
    var self = this, info = this._obs;
    this.addEventListener('loadend', function () {
      enqueue({
        type: 'api', api_url: info.url, api_method: info.method,
        api_status: self.status, api_duration_ms: Date.now() - info.start,
        api_ok: self.status >= 200 && self.status < 400,
      });
    });
    return _xhrSend.apply(this, arguments);
  };
  var _fetch = window.fetch;
  window.fetch = function (input, init) {
    var start = Date.now();
    var url = (typeof input === 'string' ? input : (input && input.url) || '').slice(0, 256);
    return _fetch.apply(this, arguments).then(function (resp) {
      enqueue({ type: 'api', api_url: url,
                api_method: (init && init.method) || 'GET',
                api_status: resp.status, api_duration_ms: Date.now() - start,
                api_ok: resp.ok });
      return resp;
    }, function (err) {
      enqueue({ type: 'api', api_url: url, api_method: (init && init.method) || 'GET',
                api_status: 0, api_duration_ms: Date.now() - start, api_ok: false });
      throw err;
    });
  };

  // ---------- 慢资源 ----------
  new PerformanceObserver(function (list) {
    list.getEntries().forEach(function (r) {
      if (r.duration > 100) {
        enqueue({ type: 'resource', r_type: r.initiatorType || 'other',
                  r_url: String(r.name).slice(0, 256),
                  r_duration_ms: Math.round(r.duration),
                  r_size_kb: r.transferSize ? Math.round(r.transferSize / 1024 * 10) / 10 : null });
      }
    });
  }).observe({ type: 'resource', buffered: true });

  // ---------- 自定义上报 ----------
  window.obsRum = function (name, payload) {
    enqueue({ type: 'custom', event_name: String(name).slice(0, 64),
              payload: JSON.stringify(payload || {}).slice(0, 2000) });
  };
})();

SDK 的五个关键工程点：

批量 + 双触发：5 秒定时或 20 条先到先发——上报请求数量与数据实时性的平衡；
sendBeacon 优先：页面关闭时普通 XHR 会被浏览器取消，sendBeacon 是 "页面卸载也不丢包"的规范答案，pagehide 里兜底 flush；
monkey-patch 顺序：XHR/fetch 包装必须先于业务代码执行才能拦到所有请求—— 所以文档要求"先配置 _OBS_RUM、先引 rum.js，再引业务脚本"；
LCP 观察器不 enqueue、只更新：LCP 会随内容出现而更新，规范取"最终值"，真正上报交给 load 后的 sendPerf；
所有采集器独立 try/catch 兜底：SDK 自己报错绝不能影响宿主页面（代码里各块实际都有 try/catch，教学版为篇幅省略，仓库版齐全）。

2.3 beacon 接收端点：服务端的防御性校验

rum/views.py 的上报端点（与仓库一致；页面视图教学版从略）：

"""
rum/views.py — RUM 数据接收端点（beacon）
"""
import json
import math

from django.core.exceptions import RequestDataTooBig
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

from monitor.security import rate_limit, require_ingest

from .models import RumEvent

PERF_KEYS = ['ttfb_ms', 'dom_ready_ms', 'load_ms', 'fp_ms', 'fcp_ms', 'lcp_ms']

# 数值上限：SQLite INTEGER 最大 2^63-1，超限会让 bulk_create 抛 OverflowError
_INT_MAX = 2 ** 62


def _num(v, int_type=False):
    """安全数值解析：非数字/NaN/Infinity/超界整数一律返回 None"""
    try:
        v = float(v)
    except (TypeError, ValueError, OverflowError):
        return None
    if not math.isfinite(v):
        return None
    if int_type:
        if abs(v) >= _INT_MAX:
            return None
        return int(v)
    return v


@csrf_exempt
@require_POST
@require_ingest
@rate_limit('rum-beacon', rate=120, per=60)
def beacon(request):
    """接收 rum.js 批量上报（单条或数组）"""
    try:
        payload = json.loads(request.body.decode('utf-8'))
    except (ValueError, UnicodeDecodeError):
        return JsonResponse({'ok': False, 'error': 'invalid json'}, status=400)
    except RequestDataTooBig:
        return JsonResponse({'ok': False, 'error': 'payload too large'}, status=413)
    events = payload if isinstance(payload, list) else [payload]
    objs = []
    for ev in events[:200]:  # 单批上限
        if not isinstance(ev, dict):
            continue
        ev_type = ev.get('type')
        if ev_type not in ('pv', 'perf', 'api', 'error', 'resource', 'custom'):
            continue
        kwargs = {
            'type': ev_type,
            'app': str(ev.get('app') or 'web')[:32],
            'page_url': str(ev.get('page_url') or '')[:256],
            'referrer': str(ev.get('referrer') or '')[:256],
            'session_id': str(ev.get('session_id') or '')[:64],
            'device': str(ev.get('device') or '')[:80],
            'screen': str(ev.get('screen') or '')[:20],
        }
        if ev_type == 'perf':
            for k in PERF_KEYS:
                kwargs[k] = _num(ev.get(k))
        elif ev_type == 'api':
            kwargs['api_url'] = str(ev.get('api_url') or '')[:256]
            kwargs['api_method'] = str(ev.get('api_method') or '')[:10]
            # 状态码钳制到 0-999（超大值/负数会溢出 SQLite 或污染统计）
            status = _num(ev.get('api_status'), int_type=True)
            kwargs['api_status'] = status if status is not None and 0 <= status <= 999 else 0
            kwargs['api_duration_ms'] = _num(ev.get('api_duration_ms'))
            raw_ok = ev.get('api_ok', True)
            # 显式布尔解析：字符串 "false"/"0" 视为失败，而不是真值
            kwargs['api_ok'] = raw_ok not in (False, 'false', 'False', '0', 0, None)
        elif ev_type == 'error':
            kwargs['err_message'] = str(ev.get('err_message') or '')[:300]
            kwargs['err_stack'] = str(ev.get('err_stack') or '')[:4000]
        elif ev_type == 'resource':
            kwargs['r_type'] = str(ev.get('r_type') or '')[:16]
            kwargs['r_url'] = str(ev.get('r_url') or '')[:256]
            kwargs['r_duration_ms'] = _num(ev.get('r_duration_ms'))
            kwargs['r_size_kb'] = _num(ev.get('r_size_kb'))
        elif ev_type == 'custom':
            kwargs['event_name'] = str(ev.get('event_name') or '')[:64]
            kwargs['payload'] = str(ev.get('payload') or '')[:2000]
        objs.append(RumEvent(**kwargs))
    if objs:
        RumEvent.objects.bulk_create(objs, batch_size=100)
    return JsonResponse({'ok': True, 'accepted': len(objs)})

rum/urls.py：

from django.urls import path

from . import views

urlpatterns = [
    path('beacon/', views.beacon, name='rum_beacon'),
]

config/urls.py 挂载：path('rum/', include('rum.urls')),。

这个端点的防御清单值得逐条看（每个都对应一类真实攻击/事故）：

防御	对应的坑
`csrf_exempt`（仍要求令牌/会话）	beacon 是机器到机器上报，无 CSRF Cookie 可带
`require_ingest`	匿名/伪造来源不能写数据
`rate_limit` 120 次/分/IP	被劫持的页面无限刷 beacon 打爆采集库
单批 `[:200]` 上限	恶意大数组撑爆内存
类型白名单 `continue`	未知 type 静默跳过，不 500
所有字符串 `[:N]` 截断	超长 URL/堆栈打爆行存储
`_num` NaN/Infinity/超界	JSON 的 `NaN` 会让 SQLite 聚合报错甚至 500
状态码钳 0-999	`api_status: 99999` 溢出 INTEGER
`api_ok` 显式布尔解析	JS 里 `0`/`"false"` 是真值的经典坑

2.4 接入业务页面

在 templates/base.html 的 <head> 末尾加：

<script>window._OBS_RUM = { endpoint: '/rum/beacon/', app: 'forum' };</script>
<script src="/static/rum.js"></script>

迁移重启，然后浏览器里随便点几个页面：

# shell
from rum.models import RumEvent
print(RumEvent.objects.count())
print(RumEvent.objects.values('type').distinct())

能看到 pv / perf / api 事件进来了（你的浏览器就是"真实用户"）。注意：登录会话本身就是合法上报凭证（require_ingest 放行已登录用户），所以本机测试不需要令牌。

本篇检查点

[ ] 访问不存在路径后 LogEntry 出现 django.request 的 WARNING/ERROR 记录；
[ ] 浏览器点几个页面后 RumEvent 有 pv/perf/api 三类数据；
[ ] 匿名 POST /rum/beacon/ → 401；携带登录 Cookie 的页面 → beacon 自动 200；
[ ] 故意在模板里制造一个 JS 报错（临时加 <script>undefinedFn()</script>）， error 事件出现 err_stack；
[ ] 断网/停服时页面无报错（SDK 全部静默兜底）——恢复后新事件继续上报。

常见报错速查

报错	原因	解法
LogEntry 一直空	Handler 没挂上	检查 loghub/apps.py 的 ready()；确认日志级别 ≥ WARNING
beacon 400 invalid json	sendBeacon 的 Blob 类型不对	对照 2.2 的 Blob 写法（application/json）
RUM 只有 pv 没有 perf	load 事件前页面就被关了/缓存了旧 SDK	强刷（Ctrl+F5）；perf 在 load 后 0ms 上报
`api_ok` 全是 true	服务端做了显式布尔解析	正常——SDK 端已按 2xx~3xx 判定
CSRF 报错 403	beacon 漏了 csrf_exempt	对照 2.3 装饰器顺序

四路数据全通了。但"数据都在库里躺着"不等于"出事有人管"——下一篇写平台的神经中枢：告警引擎（策略 → 评估 → 事件 → 通知，防抖/迟滞/静默一个不少）—— 08 · 告警中心。