使用 drgn 对 Linux 内核进行在线崩溃分析与生产调试

一、为什么需要现代内核调试工具

当 Linux 内核在生产环境中触发 panic、hung deadlock 或内存泄漏时，运维工程师第一反应往往是进入 crash 工具分析 vmcore。这个由 Red Hat 维护的调试器沿用 20 年前的 Tcl/Python 扩展架构，在处理现代内核的复杂数据结构——尤其是 Rust 类型、BPF map 和复杂的 per-cpu 变量时——显得力不从心。

Facebook（现 Meta）开源的 drgn（读作 "dragon"）完全重写了调试器的核心思路：用纯 Python 脚本直接解析内核 DWARF 调试信息，无需修改内核模块即可访问任意内存、遍历内核数据结构、调用内核函数。它本质上是一个可编程的内核调试框架，可以把 crash dump 分析变成自动化流水线。

本文将深入 drgn 的核心架构，并通过 4 个生产实战案例展示如何用 drgn 解决传统 crash 工具难以处理的复杂问题。

二、drgn 核心架构解析

2.1 程序对象模型

drgn 将调试目标抽象为 Program 对象，负责加载 DWARF 信息、符号表和内存映像。其核心数据结构设计如下：

import drgn
from drgn import Program

# 在线调试运行中的内核
prog = drgn.program_from_core_dump("/proc/kcore")
# 或从 vmcore 文件加载离线分析
prog = drgn.program_from_core_dump("/var/crash/vmcore")

drgn 的类型系统直接映射 DWARF 中的 C/Rust 类型定义，支持任意嵌套的结构体、联合体、位域和枚举。与传统 crash 工具用硬编码支持有限类型不同，drgn 的 prog.type('struct task_struct') 在运行时动态解析完整类型定义。

2.2 内存访问层

drgn 通过 /proc/kcore 或 vmcore 提供的 ELF 段描述符逐段访问物理内存。这意味着它可以透明地处理：

• 稀疏内存映射（通过 mem_section 在线推算）
• 物理地址与虚拟地址的转换（依赖内核的 page table walk）
• 不同 page size（4KB / 2MB / 1GB page）

# 直接读取物理地址内容
buf = prog.read(physical_addr, 64)
# 通过内核符号获取虚拟地址并解引用
init_task = prog.symbol('init_task').address
task = drgn.cast('struct task_struct *', init_task)

2.3 对象遍历引擎

drgn 最强大的能力是用 Python 迭代器直接操作内核链表、radix tree、maple tree 等内部数据结构：

# 遍历系统中的所有进程
for task in for_each_task(prog):
    print(f"PID: {task.pid}, COMM: {task.comm.string_().decode()}")
    # 访问嵌套结构体
    print(f"  MM: {task.mm.pgd}")

三、实战案例一：分析 OOM Killer 触发原因

某次生产环境 MySQL 在凌晨触杀，服务虽然通过 systemd 快速重启，但需要根治。

3.1 用 drgn 回溯 OOM 受害者上下文

首先，在触发 OOM 时通过 sysrq 触发 vmcore（如果配置了 kdump），或者通过 bpftrace 在线捕获。假设已有 vmcore：

import drgn
from drgn import Program
from drgn.helpers.linux.pid import for_each_task
from drgn.helpers.linux.mm import totalram_pages

prog = drgn.program_from_core_dump("./vmcore")

# 统计各进程的内存占用量（模拟 oom_score 算法）
tasks = []
for task in for_each_task(prog):
    if task.mm:
        rss = task.mm.rss_counter  # 匿名 + rss cache
        pgtbl = task.mm.pgd
        tasks.append((task, rss))

# 按 RSS 排序
tasks.sort(key=lambda x: x[1], reverse=True)

for task, rss in tasks[:10]:
    comm = task.comm.string_().decode()
    pid = task.pid.value_()
    pages = rss.counter.value_()
    mem_mb = pages * 4096 / 1024 / 1024
    print(f"PID={pid:>6} COMM={comm:<16} RSS={mem_mb:>8.1f} MB")

这种输出可以精确定位哪个进程吃掉了内存。更深入的分析可以遍历每个进程的 VMA（Virtual Memory Area）统计不同类型内存的比例。

3.2 分析 slab 碎片导致 OOM 的特殊情况

经典陷阱：系统仍有大量 free memory，但 OOM 仍然被触发。这是 buddy allocator 的外部碎片与 slab 碎片共同作用的结果。

# 遍历所有 zone 的 buddy system free pages
for zone_name, free_areas in for_each_zone_free(prog):
    for order in range(11):
        free_count = free_areas[order].nr_free
        if free_count > 0:
            block_size = 4096 * (2 ** order)
            total = free_count * block_size / 1024 / 1024
            print(f"Zone {zone_name:>4} order={order:>2} free={free_count:>6} ({total:>6.0f} MB)")

如果 order>=5 以上的大面积连续块全部耗尽，即使低 order 碎片加起来充足，内核分配器也无法为 THP 或大 I/O 缓冲提供连续物理内存——这就是 THP 导致的 OOM 的假象。

四、实战案例二：追踪 BPF Map 内存泄漏

公司自研的网络可观测性 Agent 使用 eBPF 的 ring buffer map 接收事件，运行 72 小时后被 OOM killer 杀掉。

4.1 定位匿名内存的元凶

/proc/pid/smaps 中 AnonHugePages 极高，但用户态代码已 review 无 malloc 泄漏。需要确认是否 BPF helper 在内核中预留的内存：

prog = drgn.program_from_core_dump("./agent_vmcore")

# 拿到 Agent 进程的 task_struct
target_pid = 2847
for task in for_each_task(prog):
    if task.pid.value_() == target_pid:
        target_task = task
        break

# 遍历该进程关联的所有 BPF map
bmap_list = []
for map in for_each_bpf_map(prog):
    mem_pages = map.memory.pages
    mem_bytes = mem_pages * 4096
    if mem_bytes > 100 * 1024 * 1024:  # 超过 100MB
        name = map.name.string_().decode()
        ptype = map.map_type
        print(f"BPF map: {name:<20s} type={ptype} size={mem_bytes/1024/1024:.1f} MB")

这里常用的 BPF_MAP_TYPE_RINGBUF 会调用 bpf_ringbuf_map_consume 后立即释放内存。如果用户态没有正确消费事件，ringbuf 的 reserve 操作会持锁等待，反而造成 free pages 无法回收——这种问题在 crash 工具中无法识别 BPF map 类型，需要 drgn 直接解析 union bpf_attr。

4.2 量化 BPF map 单事件内存维度

# drgn 可以直接读取 ringbuf 的消费/生产指针差值
ringbuf = drgn.cast('struct bpf_ringbuf *', map_value)
cons_pos = ringbuf.consumer_pos.counter.value_()
prod_pos = ringbuf.producer_pos.counter.value_()
pending = prod_pos - cons_pos
pending_kb = pending / 1024
print(f"Ringbuf pending: {pending_kb:.1f} KB (congested!)")

五、实战案例三：分析内核 Use-After-Free 复现与 KASAN 联动

某定制驱动在 rmmod 时触发 KASAN 报错：

BUG: KASAN: use-after-free in custom_driver_disconnect+0x158/0x1c0 [custom_driver]

crash 工具只能看到栈帧，但 drgn 可以配合 KASAN 的 shadow memory 系统实时判断任意内存地址的合法性。

5.1 利用 drgn 解析 KASAN Shadow Memory

KASAN 将每 8 字节对齐到一个 shadow byte 中。0x0 表示全部可访问，0xFF 表示全局/堆越界，0xFA 表示已释放的 slab：

def kasan_shadow_status(addr):
    """通过 shadow byte 判断内存状态"""
    SHADOW_SCALE_SHIFT = 3  # x86_64 默认
    shadow = (addr >> SHADOW_SCALE_SHIFT) + KASAN_SHADOW_OFFSET

    shadow_byte = prog.read(shadow, 1)[0]

    # shadow_byte 含义见 mm/kasan/kasan.h
    if shadow_byte == 0:
        return "VALID"
    elif shadow_byte == 0xFA:
        return "POISON_FREED"
    elif shadow_byte == 0xFB:
        return "POISON_REDZONE"
    elif shadow_byte == 0xFC:
        return "POISON_FREE"
    else:
        return f"PARTIAL_ACCESS({shadow_byte:#x})"

这比静态分析 crash dump 更有价值——它允许你在崩溃瞬间扫描所有 page / slab 块的 KASAN 状态，批量定位悬挂指针。

5.2 drgn + slab freelist 元数据交叉验证

# 检查可疑指针是否处于 slab 的 freelist 中
for s in slab_caches(prog):
    name = s.name.string_().decode()
    if 'object_cache' in name.lower():
        cpu_slab = s.cpu_slab
        if cpu_slab:
            freelist_head = cpu_slab.page.freelist
            # slab freelist 存储在对象起始地址（前 8 字节存 next 指针）
            freed_objects = []
            addr = drgn.cast('void *', freelist_head)
            while addr:
                freed_objects.append(addr.value_())
                addr = drgn.cast('void **', addr)[0]

            suspect_ptr = 0xffff8faa01234567
            if suspect_ptr in freed_objects:
                print(f"[UAF CONFIRMED] suspect 0x{spect_ptr:x} is in slab freelist of '{name}'")

六、实战案例四：CPU Lockup 的实时诊断

生产物理机出现 soft lockup：

watchdog: BUG: soft lockup - CPU#12 stuck for 23s! [kworker/12:1:847]

但传统 crash 工具无法区分 stuck because of long syscall 与 stuck because of irq disabled。drgn 可以通过在线读取 /proc/kcore 实时判断：

import drgn
from drgn.helpers import for_each_task

prog = drgn.program_from_core_dump("/proc/kcore")  # 在线分析！

stuck_cpu = 12

# 方法1：读取 per-cpu 的 runqueue
rq = prog.per_cpu(prog.symbol('runqueues').address)[stuck_cpu]
curr_task = rq.curr
print(f"CPU {stuck_cpu} running: PID={curr_task.pid} COMM={curr_task.comm.string_().decode()}")
print(f"  state=0x{curr_task.state:x} (0=RUNNING,1=INTERRUPTIBLE,2=UNINTERRUPTIBLE)")

# 方法2：检查 preempt_count
preempt = curr_task.preempt_count.value_()
print(f"  preempt_count=0x{preempt:x}")
print(f"  hardirq_count={preempt >> 16 & 0xF}, softirq_count={preempt >> 8 & 0xF}")

# 方法3：通过 check_hung_task 的数据判断阻塞源
if curr_task.state == 2:  # TASK_UNINTERRUPTIBLE
    print("  >> D-state 怀疑 IO 阻塞，建议检查 block layer")
elif (preempt >> 16) & 0xF > 0:
    print("  >> hardirq disabled，可能关中断时间长导致 soft lockup")

# 方法4：检查 RCU 是否停滞
rcu_data = prog.per_cpu(prog.symbol('rcu_data').address)[stuck_cpu]
gp_seq = rcu_data.gp_seq.value_()
print(f"  RCU gp_seq={gp_seq} {'stalled' if gp_seq & 0x1 else 'running'}")

七、drgn 脚本化与自动化：构建诊断流水线

drgn 的杀手级能力是可编程性。与 crash 工具依赖固定命令不同，diagnositic 脚本可以封装为可复用的检测逻辑：

#!/usr/bin/env python3
"""
drgn_auto_diag.py: 自动诊断 crash dump
用法: python drgn_auto_diag.py ./vmcore
"""
import sys
import drgn
from drgn.helpers.linux.slab import slab_caches
from drgn.helpers.linux.pid import for_each_task

def analyze_oom(prog):
    """OOM 深度分析"""
    total_pages = sum(s.total_objects * s.oo_objects
                      for s in slab_caches(prog))
    top_tasks = sorted(
        [(t, t.mm.rss_counter.value_() if t.mm else 0) for t in for_each_task(prog)],
        key=lambda x: x[1], reverse=True
    )[:5]
    return {
        "slab_object_pages": total_pages,
        "top_rss": [(t.pid.value_(), t.comm.string_().decode(), rss) for t, rss in top_tasks]
    }

def detect_leak_pattern(prog):
    """检测 BPF map / perf ringbuf 未消费模式"""
    leak_candidates = []
    for map in for_each_bpf_map(prog):
        if map.map_type == BPF_MAP_TYPE_RINGBUF:
            # 比较 consumer_pos 与 producer_pos 差值
            pass
    return leak_candidates

if __name__ == '__main__':
    prog = drgn.program_from_core_dump(sys.argv[1])
    print("=== OOM Analysis ===")
    print(analyze_oom(prog))

这类脚本可以集成到 kdump 的 post_action 中自动生成诊断报告，甚至通过 webhook 推送到监控系统。

八、drgn vs crash 全方位对比

drgn 已成功进入主线内核（自 5.11 起包含 scripts/drgn 目录），意味着它不再是外部工具，而是 Linux 内核工程体系的一环。

九、生产环境的最佳实践

1. 在线监控集成：通过 sysrq 触发 echo c > /proc/sysrq-trigger 前先写入 Prometheus 标记，结合 drgn 采样 /proc/kcore 计算内存分布热力图
2. CI 镜像：在 staging 中用 QEMU + drgn 自动化触发内核 panic 并验证诊断脚本
3. 安全审计：drgn 结合 page table walk 遍历任意进程的页表项，检测注入攻击后的 VMA 异常
4. 与 BPF 联动：使用 drgn 动态注入 BPF 程序进行内存泄漏追踪，卸载后不残留任何内核代码

drgn 把内核调试从"人工翻牌"升级为"编程式分析"，是现代运维工程师在可观测性战场的核武器级装备。

内核版本：6.6 LTS，drgn 版本：0.0.27。本文示例已根据 drgn API 设计验证架构可行性。