drgn：用 Python 脚本化调试运行中的 Linux 内核深度实战

引言：内核调试的范式转变

当 Linux 内核在生产环境中出现诡异问题——一个难以复现的 race condition、一处隐秘的内存泄漏、一次突如其来的 OOM kill——传统工具往往显得力不从心。crash 是经典的内核调试器，但它的类 REPL 交互方式和专有命令语言在探索复杂数据结构时效率低下；GDB 通过 vmlinux 也能调试内核，但缺乏对内核语义的深层理解。Meta 开源的 drgn（发音 "dragon"）彻底改变了这一局面：它用 Python 作为脚本语言，将内核数据结构变成可编程对象，让你能以任意精度遍历链表、解析 slab cache、甚至复现 eBPF map 的键值对。

自从 2020 年开源后，drgn 已经在 Meta 大规模生产运维中验证了自己的价值。它是纯 C 实现的核心库绑定到 Python3，无需重新编译内核（只需 vmlinux 和 debug info），支持活机和 coredump 两种模式。本文将从 drgn 架构原理切入，通过四个生产级实战场景，掌握这套"脚本化内核手术刀"。

一、drgn 核心原理与数据模型

1.1 架构分层

drgn 的设计哲学是"一切都是对象"。它的架构分为三层：

┌─────────────────────────────────────────┐
│         Python 脚本（用户层）            │
│   prog["init_task"].tasks.next          │
├─────────────────────────────────────────┤
│      drgn API 层（drgn/__init__.py）     │
│   Program / Object / Type / Symbol      │
├─────────────────────────────────────────┤
│      libdrgn 核心库（C 实现）             │
│   DWARF 解析 / ELF 加载 / 内存访问       │
└─────────────────────────────────────────┘

核心抽象 Program 代表一个内核实例（活机或 coredump），Object 是对任意内核数据结构的引用。每个 Object 包含 Type、地址、符号名，drgn 通过 DWARF debug info 自动还原完整的类型信息——你不需要手动读头文件就能知道 struct mm_struct 的每个字段含义。

1.2 内存访问机制

drgn 读取物理内存和虚拟内存，遵循 x86_64/ARM64 的分页表结构。关键在于：drgn 不会引入被调试内核的代码执行——它通过 /dev/mem 或 coredump 文件直接读取物理页，所有类型解析都在用户态完成。这意味着即使内核部分崩溃状态不稳定，drgn 依然可以正常工作。

1.3 与 crash 的关键差异

维度	crash	drgn
脚本语言	受限命令语言	完整 Python3 生态
类型系统	手动指定字段	DWARF 自动解析
容器支持	基础 namespace 感知	完整 user/pid/mnt ns 遍历
扩展性	加载 .o 模块	Python 标准库任意使用
API 设计	过程式命令	面向对象 + 迭代器

二、环境搭建与基础操作

2.1 安装

# Ubuntu/Debian
sudo apt install drgn

# 或从源码编译（推荐获取最新功能）
git clone https://github.com/osandov/drgn.git
cd drgn
python3 setup.py build
sudo python3 setup.py install

2.2 准备 Debug Symbols

drgn 需要带 DWARF 信息的 vmlinux：

# Ubuntu 通常已安装
sudo apt install linux-image-$(uname -r)-dbgsym

# vmlinux 路径
ls /usr/lib/debug/boot/vmlinux-$(uname -r)

2.3 启动交互式调试

# 调试运行中的内核（需要 root）
sudo drgn

# 调试 vmcore（coredump 模式）
sudo drgn -k /path/to/vmcore -c /path/to/vmlinux

# 直接运行脚本
sudo drgn -s script.py

2.4 基础 API 速览

import drgn

prog = drgn.program  # 当前绑定的 Program 实例

# 通过名字获取全局变量/函数
init_task = prog["init_task"]
jiffies = prog["jiffies_64"]

# 通过地址获取对象
task = prog.object_struct("task_struct", addr=0xffff88807c001c00)

# 遍历 init_task 的所有子进程
for task in init_task.tasks:
    print(f"PID={task.pid.value_()} COMM={task.comm.string_().decode()}")

# 获取类型信息
mm_type = prog.type("struct mm_struct")

三、实战一：OOM Killer 事后尸检

3.1 问题背景

凌晨 3 点，监控报警 nginx_worker 被 OOM kill，/var/log/syslog 只有寥寥一行：

Out of memory: Killed process 2843721 (nginx_worker) total-vm:8598156kB

需要回答：它被 kill 前，哪几个进程吃了最多内存？物理页分配趋势如何？NPU/GPU 驱动是否占用了不可回收页？

3.2 drgn 调查脚本

import drgn
from drgn import container_of
from collections import defaultdict

prog = drgn.prog

def get_rss_kb(task):
    """获取进程的 RSS（Resident Set Size）"""
    mm = task.mm
    if not mm:
        return 0
    # rss_stat 自 5.5 内核起可用
    try:
        return (mm.rss_stat[drgn.Object(prog, "enum vm_zone_item", NR_FILE_PAGES)].value_() +
                mm.rss_stat[drgn.Object(prog, "enum vm_zone_item", NR_ANON_MAPPED)].value_() +
                mm.rss_stat[drgn.Object(prog, "enum vm_zone_item", NR_SHMEM)].value_()) * 4  # PAGE_SIZE
    except:
        return mm.vm_file.value_() * 4  # fallback

# 收集所有进程的内存占用
proc_mem = []
for task in prog["init_task"].tasks:
    pid = task.pid.value_()
    comm = task.comm.string_().decode()
    rss = get_rss_kb(task)
    oom_score = task.signal.oom_score_adj.value_() if task.signal else 0
    proc_mem.append((pid, comm, rss, oom_score))

# 按 RSS 降序排列
proc_mem.sort(key=lambda x: x[2], reverse=True)

print(f"{'PID':>10} {'COMM':<25} {'RSS(KB)':>12} {'OOM_SCORE':>10}")
print("-" * 60)
for pid, comm, rss, score in proc_mem[:20]:
    print(f"{pid:>10} {comm:<25} {rss:>12} {score:>10}")

# 检查不可回收内存
print(f"\n=== Slab 内存统计 ===")
slab_caches = prog["slab_caches"]
total_slab = 0
for cachep in slab_caches:
    if cachep.value_() == 0:
        continue
    name = cachep.name.string_().decode()
    num = cachep.num.value_()  # 每 cache 的对象数
    size = cachep.size.value_()
    inuse = cachep.inuse.value_()
    total = num * size
    if total > 1024 * 512:  # > 512KB
        print(f"  {name:<32} size={size:>8} num={num:>6} total={total/1024:.1f}KB")
print(f"\nNUMA 分布:")
for node in range(prog["nr_node_ids"].value_()):
    zone = prog["node_data"].array(node).node_zones
    free = zone[PRESENT_PAGES].value_()
    managed = zone[MANAGED_PAGES].value_()
    print(f"  Node {node}: managed_pages={managed}, free_approx={free}")

3.3 输出解读与分析路径

运行脚本后，你会看到各进程的 RSS 排行榜。关键问题转换为：

• 是否有 page cache 异常膨胀？检查 NR_FILE_PAGES 占比
• 是否有内核 slab 泄漏？对比 /proc/slabinfo，drgn 能看到 slab cache 的完整对象布局
• NUMA 是否不均衡？查看 node_data[].node_zones[] 的 present_pages

drgn 在此场景的核心价值：从 coredump 获取"被杀瞬间"的全局内存快照，包括每个 slab cache 的活跃对象数——这些数据在 /proc 中要么分散，要么不存在。

四、实战二：追踪 kmemleak 风格的 Slab 泄漏

4.1 问题背景

一个自定义内核模块在 72 小时运行后，available_memory 从 128GB 下降到 2GB，slabtop 显示 kmalloc-128 以每小时 5MB 的速度增长——但无法确定是哪个调用路径在泄漏。

4.2 drgn 泄漏定位脚本

import drgn
from drgn import container_of
from collections import Counter

prog = drgn.prog

SLAB_CACHES_TO_WATCH = ["kmalloc-128", "kmalloc-256"]

def walk_slab_objects(cache):
    """遍历 slab cache 中的所有已分配对象"""
    prog = cache.prog_
    # 从 active_slab 开始遍历
    page = cache.cpu_slab.page
    seen = set()
    
    while page.value_() != 0 and page.value_() not in seen:
        seen.add(page.value_)
        # fault-in 物理页获取 freelist
        try:
            s_mem = page.s_mem
            # 通过 page 结构定位 slab
            slab = container_of(page.value_(), "struct slab", "page")
            freelist = slab.freelist
            
            # 检查 s_mem 区域的已分配对象
            # drgn 可以解析每个对象的内存内容
            addr = s_mem.value_()
            end = s_mem.value_() + (1 << slab.slab_cache.gfporder.value_()) * prog["PAGE_SIZE"].value_()
            
            # 只采样前 N 个对象看模式
            sample = []
            obj_addr = slab.active  # 如果内核跟踪了 active 指针
        except Exception as e:
            pass  # fault-in 失败时静默跳过
        page = page.next

# 更实用的方法：通过 page owner 追踪分配栈
print("=== 检查 page owner 记录 (需开启 CONFIG_PAGE_OWNER) ===")

# Page owner 需要内核编译时开启
try:
    pgd_data = prog["mem_map"]
    # 遍历所有 page 结构，找到属于目标 cache 的
    for pfn in range(prog["min_low_pfn"].value_, prog["max_low_pfn"].value_, 1000):
        page = prog["mem_map"] + pfn
        try:
            if page.value_() == 0:
                continue
            # 检查 page 的 slab cache 指针
            slab_page = prog.type("struct page")
            slub = page.mapping  # 简化示意
        except:
            continue
except Exception as e:
    print(f"Page owner not available: {e}")

# 替代策略：通过 kmemleak 仿真——监控 slab cache inuse 趋势
# 配合 systemd timer 每 5 分钟运行一次，记录快照
for cachep in prog["slab_caches"]:
    if cachep.value_()() == 0:
        continue
    name = cachep.name.string_().decode()
    if any(w in name for w in SLAB_CACHES_TO_WATCH):
        total_objs = cachep.num.value_()
        obj_size = cachep.size.value_()
        inuse_objs = cachep.inuse.value_()
        # 查看 batchcount 和 limit
        print(f"{name}  objsize={obj_size} total={total_objs} inuse={inuse_objs} "
              f"utilization={inuse_objs/total_objs*100:.1f}%")

4.3 进阶：通过 kprobe 串联分配栈

drgn 结合 eBPF 可以实现动态追踪。编写一个 BPF 程序记录每次 kmem_cache_alloc 的调用栈，然后在 drgn 中读取环形缓冲区：

// trace_kmem_alloc.bpf.c
#include "vmlinux.h"
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_tracing.h>

struct event {
    u32 pid;
    u64 timestamp;
    u64 stack_id;
    u32 slab_type;  // 通过参数识别 cache name
    u64 size;
};

struct {
    __uint(type, BPF_MAP_TYPE_RINGBUF);
    __uint(max_entries, 1 << 24);
} rb SEC(".maps");

SEC("kprobe/kmem_cache_alloc")
int BPF_KPROBE(trace_alloc, struct kmem_cache *s, gfp_t flags) {
    struct event *e = bpf_ringbuf_reserve(&rb, sizeof(*e), 0);
    if (!e) return 0;
    e->pid = bpf_get_current_pid_tgid() >> 32;
    e->timestamp = bpf_ktime_get_ns();
    e->stack_id = bpf_get_stackid(ctx, &stackmap, BPF_F_USER_STACK);
    bpf_ringbuf_submit(e, 0);
    return 0;
}

然后在 drgn 中分析 BPF map 的内容——这就是 eBPF + drgn 联合分析范式：eBPF 负责高速采集事件，drgn 负责离线深度解析。

五、实战三：网络协议栈深度排查

5.1 问题背景

某 UDP 服务 P99 延迟突然从 1ms 上升到 500ms，netstat -sus 显示 packet receive errors 增长，但无法确定是应用层 backlog 满还是驱动层 drop。

5.2 drgn 网络协议栈探测

import drgn
prog = drgn.prog

def diag_udp_recv_q(pnum=5):
    """诊断 UDP 接收队列积压"""
    # 遍历所有进程，找到目标 UDP 服务
    for task in prog["init_task"].tasks:
        comm = task.comm.string_().decode()
        if "my_udp_server" not in str(comm):
            continue
        
        # 遍历 fd 表，找到 UDP socket
        for fd, file in enumerate(task.files.fdt.table):  # 简化示意
            if not file:
                continue
            ops = file.f_op
            # 检查是否是 socket 操作
            try:
                if prog.symbol("socket_file_ops").address == ops.value_():
                    sock = file.private_data  # struct socket -> struct sock
                    sk = prog.type("struct sock_common")
                    sk_wmem_alloc = sock.sk.sk_wmem_alloc.counter.value_() if sock.sk else 0
                    sk_rcvbuf = sock.sk.sk_rmem_alloc.counter.value_() if sock.sk else 0
                    backlog = sock.sk.sk_backlog.len.value_() if sock.sk else 0
                    
                    print(f"PID {task.pid.value_()}: "
                          f"rmem_alloc={sk_rcvbuf/1024:.1f}KB "
                          f"wmem_alloc={sk_wmem_alloc/1024:.1f}KB "
                          f"backlog={backlog}")
                    
                    # 检查 socket 缓冲区上限
                    rcvbuf = sock.sk.sk_rcvbuf.value_()
                    sndbuf = sock.sk.sk_sndbuf.value_()
                    print(f"  socket buf: rcvbuf={rcvbuf} sndbuf={sndbuf}")
            except:
                continue

def check_netif_rx_drops():
    """检查网卡驱动层的丢包统计"""
    for i in range(prog["nr_cpu_ids"].value_()):
        # softnet_stats 记录每个 CPU 的网络处理统计
        stats = prog["per_cpu__softnet_stats"]._cpu(i)
        # 注意：__percpu 宏的处理
        received = stats.received.value_()
        dropped = stats.dropped.value_()
        time_squeeze = stats.time_squeeze.value_()
        if dropped > 0:
            print(f"CPU{i}: received={received} dropped={dropped} time_squeeze={time_squeeze}")

def trace_napi_device_queue():
    """检查 NAPI poll 队列状态"""
    for dev in prog["dev_base_head"]:
        name = dev.name.string_().decode()
        try:
            # napi 结构在 driver 私有区域
            # 通用方法：检查 net_device 的状态标志
            flags = dev.flags.value_()
            state = dev.state.value_()
            print(f"iface {name}: flags=0x{flags:x} state=0x{state:x}")
            if dev.value_() != 0:
                # qdisc 队列状态
                tx_queues = dev.num_tx_queues.value_()
                print(f"  num_tx_queues={tx_queues}")
        except:
            continue

diag_udp_recv_q()
check_netif_rx_drops()
trace_napi_device_queue()

5.3 drgn 在网络排查中的独特优势

相比 ss -ti 只能看到 socket 统计，drgn 可以：

• 直接读取 struct inet_sock 中的 inet_dport/inet_sport 而无需解析 /proc/net
• 查看 TCP control block 的 rcv_nxt/snd_nxt 序列号
• 遍历 struct net namespace 中的协议栈状态
• 检查 eBPF XDP 程序挂载状态和 map 内容

六、实战四：CPU 飙高的根因定位

6.1 问题背景

40 核机器 CPU 使用率从 5% 突然飙升至 100%，perf top 显示 __lock_text_start 占 60% CPU，但无法确定是哪个内核锁。

6.2 drgn 锁争用分析

import drgn
prog = drgn.prog

def diagnose_lock_contention():
    """诊断内核锁争用"""
    print("=== Spinlock 分析 ===")
    
    # 方法1：检查每个 CPU 的 runqueue 和 hrtick
    for cpu in range(prog["nr_cpu_ids"].value_()):
        rq = prog["runqueues"].address_of_()._cpu(cpu)  # per-cpu runqueue
        
        curr = rq.curr
        if curr.value_() == 0:
            continue
        comm = curr.comm.string_decode()
        pid = curr.pid.value_()
        
        # 检查 hrtimer 状态
        hrtick = rq.hrtick_timer
        if hrtick.enabled.value_():
            print(f"CPU{cpu}: running PID={pid} {comm} [hrtick pending]")
    
    print("\n=== Workqueue 积压检查 ===")
    # 全局 workqueue 列表
    for wq in prog["workqueues"]:
        if wq.value_()() == 0:
            continue
        name = wq.name.string_().decode()
        pwq = wq.cpu_pwqs  # per-cpu pool_workqueue
        # 检查工作项积压
        try:
            nr_active = wq.nr_active.value_()
            max_active = wq.max_active.value_()
            if nr_active > 0:
                print(f"  workqueue {name}: active={nr_active}/{max_active}")
        except:
            pass
    
    print("\n=== 中断风暴检查 ===")
    # 检查中断（IRQ）状态
    for irq_num in range(prov["nr_irqs"].value_()):
        desc = prog["irq_desc_array"]._index_array(irq_num)
        if desc.value_() == 0:
            continue
        action = desc.action
        if action.value_() == 0:
            continue
        count = desc.kstat_irqs  # per-cpu 中断计数
        total = 0
        for cpu in range(prog["nr_cpu_ids"].value_()):
            total += count._cpu(cpu).value_()
        if total > 100000:  # 异常频繁
            name = action.name.string_().decode()
            handler = action.handler
            # 解析 handler 符号
            sym = prog.symbol(handler.value_())
            print(f"  IRQ{irq_num}: {name} (handler={sym.name}) total_count={total}")

diagnose_lock_contention()

6.3 结合 perf_event_open 进行采样分析

drgn 还可以读取内核的 perf event 采样数据：

def read_perf_event_samples():
    """从 perf ring buffer 读取采样数据"""
    for cpu in range(prog["nr_cpu_ids"].value_()):
        # 遍历活跃的 perf event
        # 检查 per-cpu perf event 列表
        pass  # 需要深入了解 perf_event 内部结构
    
    # 更实用的方式：通过 drgn 直接解析 perf_event mmapped 区域
    pass

七、drgn 高级编程模式

7.1 DSL 模式：编写领域特定命令

# 将常用分析逻辑封装为 DSL
commands = {
    "memhogs": "列出 TOP 10 内存占用进程",
    "nginx_mem": "分析 nginx worker 共享内存",
    "inode_cache": "检查 dentry/inode cache 占用",
    "tcp_retrans": "追踪 TCP 重传率",
}

def memhogs(n=10):
    """内存占用 TOP N 进程"""
    import heapq
    tasks = []
    for task in prog["init_task"].tasks:
        rss = task.mm.rss if task.mm else 0
        tasks.append((rss, task))
    
    for rss, task in heapq.nlargest(n, tasks, key=lambda x: x[0].value_()):
        print(f"  PID {task.pid.value_():>8} {task.comm.string_().decode():<25} RSS={rss.value_() >> 10}MB")

def inode_cache_stats():
    """分析 inode/dentry cache"""
    print(f"nr_dentry_unused = {prog['nr_dentry_unused'].value_()}")
    # 遍历 super_blocks
    sb_list = prog["super_blocks"]
    total_dentry = 0
    for sb in sb_list:
        if sb.value_() == 0:
            continue
        # 遍历 sb 上的 dentry
        try:
            for dentry in sb.s_dentry:
                total_dentry += 1
        except:
            pass
    print(f"Total dentries scanned: {total_dentry}")

7.2 跨 Namespace 操作

drgn 完整支持 user/pid/mnt/network namespace 的遍历：

def traverse_netns():
    """遍历所有 network namespace"""
    init_net = prog["init_net"]
    for net in init_net.list:
        if net.value_() == 0:
            continue
        # 每个 net 中有独立的协议栈状态
        print(f"net ns {net.value_()}: ifindex_count={net.ifindex_count.value_()}")
        
        # 遍历该 netns 中的网络设备
        for dev in net.dev_name_head:
            name = dev.name.string_().decode()
            print(f"  dev: {name}")

7.3 drgn 脚本与 CI 集成

#!/usr/bin/env python3
# kernel_health_check.py — 与 Prometheus Exporter 集成
import drgn
import json

prog = drgn.prog

metrics = {}

# 收集内存指标
for node in range(prog["nr_node_ids"].value_()):
    zone = prog["node_data"].array(node).node_zones
    metrics[f"node_{node}_managed_pages"] = zone[MANAGED_PAGES].value_()
    metrics[f"node_{node}_free_pages"] = zone[FREE_PAGES].value_()

# 收集 slab 泄漏检测
for cachep in prog["slab_caches"]:
    if cachep.value_() == 0:
        continue
    name = cachep.name.string_().decode()
    util = cachep.inuse.value_() / cachep.num.value_() if cachep.num.value_() else 0
    if util > 0.9:
        metrics[f"slab_high_util_{name}"] = util

八、生产环境部署建议

8.1 性能与安全考量

drgn 读取物理内存会带来可测量的开销——每次内存访问都需要 fault-in 页面。在生产环境使用时建议：

• 使用 coredump 而非活机：coredump 期间系统短暂停顿，但后续分析不干扰运行
• 限制采样范围：不要遍历全量 slabs，只关注高使用率的 cache
• 控制执行时间：长脚本放在 coredump 环境执行，避免 /dev/mem 长时间占用
• 内存访问故障处理：确保有 try/except 包裹每个 fault-in 操作

8.2 推荐部署模式

┌─────────────┐     VMCORE采集     ┌──────────────────┐
│  生产服务器  │ ─────────────────→ │ Analysis Server   │
│ kdump触发   │    (kdump.conf)    │ drgn + vmlinux   │
└─────────────┘                    └──────────────────┘
         ↓                                ↓
   监控异常触发                     自动化脚本分析
   (OOM/softlockup)               (Python/Prometheus)

8.3 与现有工具的协同

场景	首选工具	drgn 的角色
快速 CPU profiling	perf	离线深度解析 callchain
内存泄漏检测	kmemleak	静态 dump 交叉验证
网络问题定位	tcpdump/ss	协议栈内部状态查看
块层问题	blktrace	request_queue 内部诊断
容器问题	crictl/nsenter	跨 namespace 对象遍历

总结

drgn 将内核调试从"盲人摸象"升级为"外科手术"。它不是要取代 perf 或 crash，而是填补了"可编程深度内核分析"的生态空白。四个实战场景展示了它的核心价值：

1. OOM 尸检：获取杀前全局内存快照
2. Slab 泄漏追踪：跨时间维度对比 cache utilization
3. 网络协议栈探测：从 socket 到 driver 的全链路洞察
4. 锁争用诊断：直接读取内核锁的内部状态

掌握 drgn 的关键不在于记住所有 API，而在于理解"一切皆 Object"的思维——当你能想象内核中每个对象的关系图并通过 Python 编程探索它时，任何内核问题都将变得可解。

---

**安装提示**：`pip install drgn` 在多数 Linux 发行版可直接获取。首次运行需要 root 或 `CAP_SYS_ADMIN` 权限。建议搭配 Kernel 5.10+ 使用以获得最完整的 type 支持。