05 · 批量缓冲与指标注册表：一次 p99 2190ms → 251ms 的改造

请求线程: [视图 8ms] ──[排队等写锁 ~200ms]──[INSERT + fsync ~3ms]──[返回]
                     ↑ 所有并发请求在这里挤独木桥

"""
monitor/buffer.py — 请求指标批量缓冲（APM 热路径与写库解耦）

改造前：中间件在响应返回前同步 INSERT 一行 RequestMetric。单写者数据库
（SQLite）下所有请求线程排队抢同一把写锁、每条 INSERT 各自 fsync，
压测里表现为 p50 5ms 而 p99 800ms 的长尾。

改造后：请求线程只做一次 put_nowait（微秒级、不碰数据库），
后台线程按"攒满 N 条 或 到期 T 秒"用一条 bulk_create 批量落库——
写库次数从"每请求一次"降到"每批一次"，事务数（fsync 数）同样被摊薄。

代价与边界（都是有意选择，且可观测）：
- 指标可见延迟最多 FLUSH_SEC 秒（大盘/告警读的是库，不是实时总线）；
- 队列满时丢新点并计数：观测数据可以采样，采集不允许反压业务；
  丢弃数经 /metrics 的 obs_metric_buffer_dropped_total 暴露；
- 后台落库遇瞬时故障（单写者锁、连接抖动）先重试 2 次，仍失败才整批丢弃并计数；
- 进程被 SIGKILL 时缓冲区里未落库的点会丢（atexit 只覆盖正常退出）。

`OBSERVABILITY['METRIC_BUFFER_ENABLED']` 为假时 submit() 直接同步写库，
管理命令与测试因此仍看到"请求结束即落库"的语义。
"""
import atexit
import logging
import os
import queue
import threading
import time

from django.conf import settings
from django.db import connection, transaction
from django.utils import timezone

logger = logging.getLogger(__name__)

DEFAULT_BATCH_SIZE = 200
DEFAULT_FLUSH_SEC = 1.0
DEFAULT_QUEUE_SIZE = 20000

# 后台线程落库遇到瞬时故障（单写者锁、连接抖动）时的重试预算：最多 1+2 次尝试
WRITE_RETRIES = 2
WRITE_RETRY_SEC = 0.05

# 队列里放的是构造 RequestMetric 的字段字典（请求线程不实例化模型，省一次开销）
_q = queue.Queue()
_q_sized = False

_stats = {'written': 0, 'dropped': 0, 'batches': 0, 'errors': 0}

# 可重入：start() 持锁期间会调用 _ensure_queue_size()
_lock = threading.RLock()
_flusher = None
_stop = threading.Event()
_last_drop_log = 0.0


def _cfg(key, default):
    return settings.OBSERVABILITY.get(key, default)


def _ensure_queue_size():
    """按配置确定队列容量（Queue 容量建好后不可改，故延迟到首次使用）"""
    global _q, _q_sized
    if _q_sized:
        return
    with _lock:
        if _q_sized:
            return
        size = int(_cfg('METRIC_BUFFER_QUEUE_SIZE', DEFAULT_QUEUE_SIZE))
        if size > 0:
            _q = queue.Queue(maxsize=size)
        _q_sized = True


def enabled():
    """缓冲区是否启用（关闭时 submit 退化为同步写库）"""
    return bool(_cfg('METRIC_BUFFER_ENABLED', False))


def queued():
    return _q.qsize()


def stats():
    """缓冲区自观测：落库/丢弃条数、批次数、失败批次数、当前积压"""
    return dict(_stats, pending=queued(), enabled=enabled(),
                batch_size=int(_cfg('METRIC_BUFFER_BATCH_SIZE', DEFAULT_BATCH_SIZE)),
                flush_sec=float(_cfg('METRIC_BUFFER_FLUSH_SEC', DEFAULT_FLUSH_SEC)))


def write_rows(rows, retries=0):
    """把一批字段字典写入库，返回落库条数。

    `retries` > 0 时，失败先重试（间隔逐次翻倍）再放弃——SQLite 的单写者锁
    "database table is locked" 属于典型瞬时故障，让整批点因为一次锁竞争蒸发
    说不过去。最终仍写不进去才整批丢弃并计数：采集组件不能因为自己写不进去
    而把调用方拖垮。失败后关闭本线程连接，下一批自动重连（数据库重启/连接失效）。
    """
    from .models import RequestMetric

    if not rows:
        return 0
    objs = [RequestMetric(**r) for r in rows]
    batch_size = max(1, int(_cfg('METRIC_BUFFER_BATCH_SIZE', DEFAULT_BATCH_SIZE)))
    for attempt in range(retries + 1):
        try:
            with transaction.atomic():
                RequestMetric.objects.bulk_create(objs, batch_size=batch_size)
            _stats['written'] += len(objs)
            _stats['batches'] += 1
            return len(objs)
        except Exception as exc:
            if attempt < retries:
                logger.warning('RequestMetric 批量落库失败（第 %d/%d 次尝试）：%s',
                               attempt + 1, retries + 1, exc)
                time.sleep(WRITE_RETRY_SEC * (2 ** attempt))
                continue
            _stats['errors'] += 1
            _stats['dropped'] += len(objs)
            logger.exception('RequestMetric 批量落库失败，本批 %d 条已丢弃', len(objs))
    try:
        connection.close()
    except Exception:
        pass
    return 0


def submit(payload):
    """提交一条请求指标。返回 'buffered' / 'written' / 'dropped'，永不抛异常。"""
    global _last_drop_log
    payload.setdefault('created_at', timezone.now())
    if not enabled():
        write_rows([payload])
        return 'written'
    _ensure_queue_size()
    try:
        _q.put_nowait(payload)
        return 'buffered'
    except queue.Full:
        _stats['dropped'] += 1
        now = time.monotonic()
        if now - _last_drop_log > 60:  # 限流日志：打满时每秒刷一条会淹没真错误
            _last_drop_log = now
            logger.error('请求指标缓冲区已满，累计丢弃 %d 条（写库跟不上采集）',
                         _stats['dropped'])
        return 'dropped'


def flush():
    """立即排空队列并落库（压测/管理命令/测试同步取数用），返回落库条数"""
    written = 0
    while True:
        rows, drained = [], False
        try:
            while True:
                rows.append(_q.get_nowait())
        except queue.Empty:
            drained = True
        written += write_rows(rows)
        if drained:
            return written


def _flusher_loop():
    batch_size = max(1, int(_cfg('METRIC_BUFFER_BATCH_SIZE', DEFAULT_BATCH_SIZE)))
    flush_sec = float(_cfg('METRIC_BUFFER_FLUSH_SEC', DEFAULT_FLUSH_SEC))
    batch = []
    deadline = time.monotonic() + flush_sec
    while not _stop.is_set():
        if batch and (len(batch) >= batch_size or time.monotonic() >= deadline):
            write_rows(batch, retries=WRITE_RETRIES)
            batch = []
            deadline = time.monotonic() + flush_sec
        try:
            batch.append(_q.get(timeout=0.2))  # 短等待，停服时能尽快退出
        except queue.Empty:
            continue
    write_rows(batch, retries=WRITE_RETRIES)


def start():
    """启动 flusher 线程（幂等）。缓冲区未启用时不启动，保持同步写库语义。"""
    global _flusher
    if not enabled():
        return False
    with _lock:
        if _flusher is not None:
            return True
        _ensure_queue_size()
        _stop.clear()
        _flusher = threading.Thread(target=_flusher_loop, name='obs-metric-buffer',
                                    daemon=True)
        _flusher.start()
        atexit.register(shutdown)
        logger.info('请求指标缓冲线程已启动：batch=%s flush=%ss',
                    _cfg('METRIC_BUFFER_BATCH_SIZE', DEFAULT_BATCH_SIZE),
                    _cfg('METRIC_BUFFER_FLUSH_SEC', DEFAULT_FLUSH_SEC))
        return True


def shutdown():
    """停止 flusher 并把剩余点排空（正常退出路径不丢数据）"""
    global _flusher
    thread = _flusher
    if thread is None:
        flush()
        return
    _flusher = None
    _stop.set()
    thread.join(timeout=5)
    flush()


def _reset_after_fork():
    """gunicorn/uwsgi 预加载（--preload）场景：master 进程建好缓冲区后 fork，
    worker 只继承内存、不继承线程——句柄失效，重建。"""
    global _lock, _flusher, _q, _q_sized
    _lock = threading.RLock()
    _flusher = None
    _q = queue.Queue()
    _q_sized = False
    _stop.clear()
    try:
        start()
    except Exception:  # pragma: no cover - fork 处理路径尽力而为
        logger.exception('fork 后重启请求指标缓冲失败')


if hasattr(os, 'register_at_fork'):
    os.register_at_fork(after_in_child=_reset_after_fork)

import os
import sys
from pathlib import Path

    # ---- 请求指标批量缓冲（monitor/buffer.py）----
    # 请求线程只入队，后台线程按"攒满 N 条或到期 T 秒"bulk_create 批量落库。
    # 关闭（或管理命令/测试进程）时退化为同步写库，保持"请求结束即落库"的语义。
    'METRIC_BUFFER_BATCH_SIZE': max(1, int(os.environ.get('OBS_METRIC_BUFFER_BATCH', '200'))),
    'METRIC_BUFFER_FLUSH_SEC': max(0.1, float(os.environ.get('OBS_METRIC_BUFFER_FLUSH_SEC', '1'))),
    'METRIC_BUFFER_QUEUE_SIZE': max(1, int(os.environ.get('OBS_METRIC_BUFFER_QUEUE_SIZE', '20000'))),
    'METRIC_BUFFER_ENABLED': os.environ.get('OBS_METRIC_BUFFER', '1') != '0',

# ---------------- 管理命令与测试：缓冲关闭，退回同步写库 ----------------
# 这些场景必须"写完立刻读得到"，且 migrate 期间表可能还不存在，缓冲线程反而添乱。
_current_cmd = sys.argv[1] if len(sys.argv) > 1 else ''
if _current_cmd in WORKER_START_EXCLUDE_CMDS:
    OBSERVABILITY['METRIC_BUFFER_ENABLED'] = False

"""
monitor/workers.py — 后台线程的启动器
runserver 下只在自动重载的子进程里启动一次（RUN_MAIN=1），
migrate / shell / init_data 等管理命令不启动。线程全部为 daemon，随进程退出。
"""
import logging
import os
import sys
import threading

from django.conf import settings

logger = logging.getLogger(__name__)

_started = False
_start_lock = threading.Lock()


def _should_start():
    if os.environ.get('OBS_DISABLE_WORKERS') == '1':
        return False
    if 'runserver' in sys.argv:
        # 自动重载模式：仅子进程（RUN_MAIN=1）启动；--noreload 模式：本进程即服务进程
        return os.environ.get('RUN_MAIN') == '1' or '--noreload' in sys.argv
    command = (sys.argv[1] if len(sys.argv) > 1 else '')
    if command in settings.WORKER_START_EXCLUDE_CMDS:
        return False
    return True


def maybe_start_workers():
    # 请求指标缓冲线程在 Web 进程里也必须起：指标由 Web 进程产生
    from . import buffer
    try:
        buffer.start()
    except Exception:
        logger.exception('请求指标缓冲线程启动失败')
    # （06 篇在这里加：if _should_start(): start_workers() —— 采集/告警等四条线程）

from django.apps import AppConfig
from django.db.backends.signals import connection_created


def _set_sqlite_pragma(sender, connection, **kwargs):
    """SQLite 开启 WAL：后台采集线程与请求线程并发读写更稳
    （config/__init__.py 已全局注册一份；这里保留 app 级注册，两处幂等）"""
    if connection.vendor != 'sqlite':
        return
    try:
        cursor = connection.cursor()
        cursor.execute('PRAGMA journal_mode=WAL;')
        cursor.execute('PRAGMA synchronous=NORMAL;')
        cursor.close()
    except Exception:
        pass


class MonitorConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'monitor'
    verbose_name = '全栈可观测运维平台'

    def ready(self):
        connection_created.connect(_set_sqlite_pragma, dispatch_uid='obs-sqlite-wal')
        from .workers import maybe_start_workers
        maybe_start_workers()

# 例：16 并发 2000 请求（ab 是 Apache Benchmark）
ab -n 2000 -c 16 -C "sessionid=x" http://127.0.0.1:8014/forum/problem/

"""
monitor/registry.py — 指标注册表（全平台统一取数入口）

总览页、自定义大盘、告警策略下拉、智能分析共用：
每个指标 key 对应 {标签, 单位, series(minutes)->按分钟聚合的点列}。

新增一个可观测指标只需在这里注册一项，大盘/告警/分析立即生效。
"""
from django.db.models import Avg, Count, Sum
from django.db.models.functions import TruncMinute
from django.utils import timezone

# ---------------------------------------------------------------------------
# 通用工具
# ---------------------------------------------------------------------------


def _since(minutes):
    return timezone.now() - timezone.timedelta(minutes=minutes)


def _bucket(qs, annotate_map, ts_field='created_at'):
    """按分钟聚合，返回 [(时间字符串, dict(...))] 升序"""
    rows = (
        qs.annotate(bucket=TruncMinute(ts_field))
        .values('bucket')
        .annotate(**annotate_map)
        .order_by('bucket')
    )
    return [(r['bucket'], r) for r in rows]


def _fmt(dt):
    return timezone.localtime(dt).strftime('%H:%M')


def _p95(values):
    if not values:
        return 0.0
    vs = sorted(values)
    k = max(0, min(len(vs) - 1, int(len(vs) * 0.95)))
    return float(vs[k])


# ---------------------------------------------------------------------------
# HTTP / APM 指标（RequestMetric）
# ---------------------------------------------------------------------------

def http_request_count(minutes):
    from .models import RequestMetric
    pts = _bucket(
        RequestMetric.objects.filter(created_at__gte=_since(minutes)),
        {'v': Count('id')},
    )
    return [{'t': _fmt(t), 'v': float(r['v'] or 0)} for t, r in pts]


def http_avg_duration(minutes):
    from .models import RequestMetric
    pts = _bucket(
        RequestMetric.objects.filter(created_at__gte=_since(minutes)),
        {'v': Avg('duration_ms')},
    )
    return [{'t': _fmt(t), 'v': round(float(r['v'] or 0), 1)} for t, r in pts]


def http_p95_duration(minutes):
    from .models import RequestMetric
    qs = RequestMetric.objects.filter(created_at__gte=_since(minutes))
    per_min = {}
    for ts, dur in qs.values_list('created_at', 'duration_ms'):
        key = ts.replace(second=0, microsecond=0)
        per_min.setdefault(key, []).append(dur)
    out = []
    for t in sorted(per_min):
        out.append({'t': _fmt(t), 'v': round(_p95(per_min[t]), 1)})
    return out


def http_error_rate(minutes):
    from django.db.models import Case, When
    from .models import RequestMetric
    pts = _bucket(
        RequestMetric.objects.filter(created_at__gte=_since(minutes)),
        {'total': Count('id'),
         'errs': Sum(Case(When(is_error=True, then=1), default=0))},
    )
    return [
        {'t': _fmt(t), 'v': round((r['errs'] or 0) * 100.0 / r['total'], 2) if r['total'] else 0.0}
        for t, r in pts
    ]


def http_sql_avg(minutes):
    from .models import RequestMetric
    pts = _bucket(
        RequestMetric.objects.filter(created_at__gte=_since(minutes)),
        {'v': Avg('sql_count')},
    )
    return [{'t': _fmt(t), 'v': round(float(r['v'] or 0), 1)} for t, r in pts]


def http_slow_query_count(minutes):
    from .models import RequestMetric
    pts = _bucket(
        RequestMetric.objects.filter(created_at__gte=_since(minutes)),
        {'v': Sum('slow_query_count')},
    )
    return [{'t': _fmt(t), 'v': float(r['v'] or 0)} for t, r in pts]


# ---------------------------------------------------------------------------
# 注册表本体（本篇先注册 HTTP 域；主机/RUM/日志域在对应篇目加入）
# ---------------------------------------------------------------------------

REGISTRY = {
    'http.request_count': ('请求数/分钟', '次/分', http_request_count),
    'http.avg_duration': ('接口平均耗时', 'ms', http_avg_duration),
    'http.p95_duration': ('接口 P95 耗时', 'ms', http_p95_duration),
    'http.error_rate': ('接口错误率', '%', http_error_rate),
    'http.sql_avg': ('平均 SQL 次数', '条/请求', http_sql_avg),
    'http.slow_query_count': ('慢查询数', '条/分', http_slow_query_count),
}


def series(key, minutes):
    """按 key 取指标序列；custom.<name> 动态分发。
    返回 None 表示 key 不存在。"""
    if key.startswith('custom.'):
        name = key[len('custom.'):]
        from .models import CustomMetric
        if not CustomMetric.objects.filter(name=name).exists():
            return None  # 自定义指标不存在：与未知 registry key 一致返回 None
        pts = _bucket(
            CustomMetric.objects.filter(name=name, created_at__gte=_since(minutes)),
            {'v': Avg('value')},
        )
        return {'key': key, 'label': f'自定义指标 {name}', 'unit': '',
                'points': [{'t': _fmt(t), 'v': round(float(r['v'] or 0), 3)} for t, r in pts]}
    item = REGISTRY.get(key)
    if not item:
        return None
    label, unit, fn = item
    return {'key': key, 'label': label, 'unit': unit, 'points': fn(minutes)}


def metric_value(key, minutes=5):
    """取最近 N 分钟的均值（告警引擎用）；无数据返回 None"""
    s = series(key, minutes)
    if not s or not s['points']:
        return None
    vals = [p['v'] for p in s['points'] if p['v'] is not None]
    return round(sum(vals) / len(vals), 2) if vals else None


def catalog():
    """给下拉框用：[{key,label,unit}]，含 custom.* 动态指标"""
    items = [
        {'key': k, 'label': f'{v[0]} ({v[1]})', 'unit': v[1]}
        for k, v in REGISTRY.items()
    ]
    from .models import CustomMetric
    names = set(CustomMetric.objects.values_list('name', flat=True).distinct()[:50])
    items += [{'key': f'custom.{n}', 'label': f'自定义指标 {n}', 'unit': ''} for n in sorted(names)]
    return items

def overview(request):
    """监控总览：KPI 卡片 + 注册表指标曲线"""
    from .registry import series

    kpi = {
        'req_5m': metric_value('http.request_count', 5),
        'avg_5m': metric_value('http.avg_duration', 5),
        'err_5m': metric_value('http.error_rate', 5),
    }
    charts = {
        'request_count': series('http.request_count', 60),
        'avg_duration': series('http.avg_duration', 60),
    }
    return render(request, 'monitor/overview.html', {'kpi': kpi, 'charts': charts})

{% extends "base.html" %}
{% block title %}监控总览{% endblock %}
{% block content %}
<h1>监控总览</h1>
<p>
  近5分钟请求：<b>{{ kpi.req_5m|default:"-" }}</b> ｜
  平均耗时：<b>{{ kpi.avg_5m|default:"-" }}</b> ms ｜
  错误率：<b>{{ kpi.err_5m|default:"-" }}</b> %
</p>
<div id="c1" style="width:100%;height:320px;background:#fff;"></div>
<h3>接口平均耗时（60 分钟）</h3>
<div id="c2" style="width:100%;height:320px;background:#fff;"></div>

<script src="/static/echarts.min.js"></script>
<script>
  const c1 = echarts.init(document.getElementById('c1'));
  c1.setOption({
    title: { text: '请求数/分钟' },
    xAxis: { type: 'category', data: {{ charts.request_count.points|safe }}.map(p => p.t) },
    yAxis: { type: 'value' },
    series: [{ type: 'line', data: {{ charts.request_count.points|safe }}.map(p => p.v),
               smooth: true, areaStyle: {} }],
    tooltip: { trigger: 'axis' },
  });
  const c2 = echarts.init(document.getElementById('c2'));
  c2.setOption({
    title: { text: '接口平均耗时 (ms)' },
    xAxis: { type: 'category', data: {{ charts.avg_duration.points|safe }}.map(p => p.t) },
    yAxis: { type: 'value' },
    series: [{ type: 'line', data: {{ charts.avg_duration.points|safe }}.map(p => p.v),
               smooth: true, areaStyle: {} }],
    tooltip: { trigger: 'axis' },
  });
</script>
{% endblock %}

{{ charts.request_count.points|json_script:"d1" }}
{{ charts.avg_duration.points|json_script:"d2" }}
<script>
  const d1 = JSON.parse(document.getElementById('d1').textContent);
  const d2 = JSON.parse(document.getElementById('d2').textContent);
  // 之后 c1 用 d1.map(p => p.t) / d1.map(p => p.v)，c2 同理
</script>