【eBPF 内核实现深度拆解】实战：构建微型 eBPF 可观测 Agent

你已经读完了 BPF 指令集、verifier 算法、JIT 后端、map 实现、helper 子系统、BTF 格式、CO-RE 引擎——现在干什么？孤立的知识点不等于能写出生产级 BPF 程序。这篇文章把前面 17 篇的内核知识串成一条从零到一的实践线：写一个完整的、可编译、可运行的 eBPF 可观测 Agent。

这个 Agent 追踪系统中所有 execve() 调用，将事件通过 ring buffer 回传到用户态，支持跨内核版本运行（CO-RE），支持热升级（map pinning），附带 verifier 错误排障的完整示例。

一、目标：一个麻雀虽小五脏俱全的 Agent

Agent 的功能需求：

1. 追踪全系统所有 execve() 系统调用（进程创建）
2. 捕获 pid、进程名（comm）、execve 的文件名、UID
3. 通过 ring buffer 将事件推送到用户态
4. 用户态程序打印事件并计数
5. 支持跨内核版本（CO-RE）——同一个字节码在 5.15 到 6.8 上都运行
6. 支持热升级——重启 Agent 时计数器不归零（map pinning）
7. 干净的 Ctrl-C 退出（自动 detach）

项目结构（完整源码见同目录 mini-agent/）：

mini-agent/
├── mini-agent.bpf.c   # BPF 程序（内核态）
├── mini-agent.c        # 用户态加载器
├── mini-agent.h        # 共享数据结构
├── Makefile            # 构建（pkg-config libbpf）
├── README.md           # 构建与运行说明
└── vmlinux.h           # bpftool 生成的 BTF 类型头文件（make vmlinux）

Agent 的完整数据流如下：

flowchart TD
    A["应用调用 execve()"] --> B["内核: sys_enter_execve tracepoint"]
    B --> C["BPF 程序 handle_execve()"]
    C --> D["bpf_get_current_pid_tgid()"]
    C --> E["BPF_CORE_READ(task, cred, uid)"]
    C --> F["bpf_probe_read_user_str(filename)"]
    D --> G["构造 exec_event"]
    E --> G
    F --> G
    G --> H["bpf_ringbuf_reserve()"]
    H --> I["bpf_ringbuf_submit()"]
    I --> J["ring buffer (BPF map)"]
    J --> K["用户态: ring_buffer__poll()"]
    K --> L["handle_event() 回调"]
    L --> M["printf 输出事件"]

    subgraph KERNEL["内核态"]
        B
        C
        D
        E
        F
        G
        H
        I
    end

    subgraph USER["用户态"]
        K
        L
        M
    end

    J

二、Step 1：Makefile 与构建基础设施

# Makefile（节选，完整版见 mini-agent/Makefile）
CLANG ?= clang
BPFTOOL ?= bpftool
CC ?= gcc

LIBBPF_CFLAGS := $(shell pkg-config --cflags libbpf 2>/dev/null)
LIBBPF_LDLIBS := $(shell pkg-config --libs libbpf 2>/dev/null)

CFLAGS := -g -O2 -Wall $(LIBBPF_CFLAGS)
BPF_CFLAGS := -g -O2 -target bpf -D__TARGET_ARCH_x86

.PHONY: all clean vmlinux

all: mini-agent

vmlinux:
    $(BPFTOOL) btf dump file /sys/kernel/btf/vmlinux format c > vmlinux.h

mini-agent.bpf.o: mini-agent.bpf.c vmlinux.h mini-agent.h
    $(CLANG) $(BPF_CFLAGS) -c $< -o $@

mini-agent.skel.h: mini-agent.bpf.o
    $(BPFTOOL) gen skeleton $< > $@

mini-agent: mini-agent.c mini-agent.skel.h
    $(CC) $(CFLAGS) -o $@ mini-agent.c $(LIBBPF_LDLIBS)

关键步骤说明：

1. 生成 vmlinux.h：bpftool btf dump 从运行中内核的 BTF 信息生成包含所有内核类型的 C 头文件。这个文件替代了 #include <linux/sched.h> 等内核头文件，是 CO-RE 的基础。
2. 编译 BPF 对象：clang -target bpf 生成 BPF 字节码（ELF 目标文件）
3. 生成 skeleton：bpftool gen skeleton 将 BPF 目标文件的 ELF 结构解析为 C 代码——map 定义、程序加载、auto-attach 逻辑都自动生成
4. 编译用户态：通过 pkg-config libbpf 获取头文件与链接参数（无 pkg-config 时可回退 -I/usr/include/bpf -lbpf -lelf -lz）

三、Step 2：共享数据结构

/* mini-agent.h */
#ifndef MINI_AGENT_H
#define MINI_AGENT_H

#define MAX_FILENAME 256
#define MAX_COMM     16
#define TASK_COMM_LEN 16

/* 事件：通过 ring buffer 从内核态传到用户态 */
struct exec_event {
    u32 pid;                    /* 进程 PID (tgid) */
    u32 uid;                    /* 用户 UID */
    char comm[TASK_COMM_LEN];   /* 进程名 */
    char filename[MAX_FILENAME];/* execve 的文件名 */
};

#endif

四、Step 3：BPF 内核态程序

/* mini-agent.bpf.c */
#include "vmlinux.h"
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_tracing.h>
#include <bpf/bpf_core_read.h>
#include "mini-agent.h"

char LICENSE[] SEC("license") = "GPL";

/* ring buffer：向用户态推送事件 */
struct {
    __uint(type, BPF_MAP_TYPE_RINGBUF);
    __uint(max_entries, 256 * 1024);
} events SEC(".maps");

/* 全局 execve 计数——pin 此 map 实现热升级后计数不归零 */
struct {
    __uint(type, BPF_MAP_TYPE_ARRAY);
    __uint(max_entries, 1);
    __type(key, u32);
    __type(value, u64);
} exec_count SEC(".maps");

SEC("tp/syscalls/sys_enter_execve")
int handle_execve(struct trace_event_raw_sys_enter *ctx)
{
    struct exec_event *event;
    struct task_struct *task;
    u32 key = 0;
    u64 *count;
    u64 pid_tgid;
    const char *filename;

    pid_tgid = bpf_get_current_pid_tgid();
    task = (struct task_struct *)bpf_get_current_task();

    count = bpf_map_lookup_elem(&exec_count, &key);
    if (count)
        __sync_fetch_and_add(count, 1);

    event = bpf_ringbuf_reserve(&events, sizeof(*event), 0);
    if (!event)
        return 0;

    event->pid = pid_tgid >> 32;
    event->uid = BPF_CORE_READ(task, cred, uid.val);
    bpf_get_current_comm(&event->comm, sizeof(event->comm));

    filename = (const char *)BPF_CORE_READ(ctx, args[0]);
    bpf_probe_read_user_str(&event->filename, sizeof(event->filename),
                             filename);

    bpf_ringbuf_submit(event, 0);
    return 0;
}

4.1 代码逐点解析

#include "vmlinux.h"：使用 bpftool 从 BTF 生成的内核类型定义。替代所有 #include <linux/*.h>，包含所有内核结构体的完整定义。这是 CO-RE 的第一步——不依赖内核头文件。

BPF_CORE_READ(task, cred, uid.val)：CO-RE 的核心宏。编译时生成 field access 指令但使用相对偏移；加载时 libbpf 根据目标内核的 BTF 计算出精确的字段偏移并修补指令。这是 “一次编译，到处运行” 的关键。

bpf_probe_read_user_str()：读取用户态字符串。verifier 要求所有用户态内存访问必须通过 bpf_probe_read_user 系列 helper——直接解引用用户态指针会被 verifier 拒绝。

bpf_ringbuf_reserve()/submit()：ring buffer 的两步提交语义。先 reserve 一块空间，填充后再 submit。如果 ring buffer 满，reserve 返回 NULL——事件被丢弃。这比 perf buffer 的单步提交更灵活，也避免了内部锁竞争。

五、Step 4：用户态加载器

/* mini-agent.c */
#include <stdio.h>
#include <stdlib.h>
#include <signal.h>
#include <unistd.h>
#include <bpf/libbpf.h>
#include <bpf/bpf.h>
#include "mini-agent.skel.h"
#include "mini-agent.h"

static volatile bool exiting = false;

static void sig_handler(int sig)
{
    exiting = true;
}

/* ring buffer 事件回调 */
static int handle_event(void *ctx, void *data, size_t data_sz)
{
    const struct exec_event *e = data;

    printf("%-16s %-6u %-6u %s\n",
           e->comm, e->pid, e->uid, e->filename);
    return 0;
}

int main(int argc, char **argv)
{
    struct mini_agent_bpf *skel;
    struct ring_buffer *rb = NULL;
    int err;

    /* 1. 设置信号处理（Ctrl-C 干净退出）*/
    signal(SIGINT, sig_handler);
    signal(SIGTERM, sig_handler);

    /* 2. 打开 BPF 骨架 */
    skel = mini_agent_bpf__open();
    if (!skel) {
        fprintf(stderr, "Failed to open BPF skeleton\n");
        return 1;
    }

    /* 3. 尝试复用已 pin 的 exec_count map（热升级保留计数） */
    setup_map_pinning(skel);

    /* 4. 加载 BPF 程序到内核（触发 verifier + JIT）*/
    err = mini_agent_bpf__load(skel);
    if (err) {
        fprintf(stderr, "Failed to load BPF skeleton: %d\n", err);
        goto cleanup;
    }

    /* 5. 挂载（attach）BPF 程序到 tracepoint */
    err = mini_agent_bpf__attach(skel);
    if (err) {
        fprintf(stderr, "Failed to attach BPF skeleton: %d\n", err);
        goto cleanup;
    }

    /* 6. 设置 ring buffer 消费者 */
    rb = ring_buffer__new(bpf_map__fd(skel->maps.events),
                          handle_event, NULL, NULL);
    if (!rb) {
        fprintf(stderr, "Failed to create ring buffer\n");
        goto cleanup;
    }

    printf("%-16s %-6s %-6s %s\n", "COMM", "PID", "UID", "FILENAME");
    printf("----------------------------------------\n");

    /* 7. 事件循环——轮询 ring buffer */
    while (!exiting) {
        err = ring_buffer__poll(rb, 100 /* timeout ms */);
        if (err == -EINTR) {
            err = 0;
            break;
        }
        if (err < 0) {
            fprintf(stderr, "Error polling ring buffer: %d\n", err);
            break;
        }
    }

cleanup:
    ring_buffer__free(rb);
    mini_agent_bpf__destroy(skel);
    return err < 0 ? -err : 0;
}

5.1 Skeleton 的工作流

mini_agent_bpf__open() → __load() → __attach() 是最常见的 libbpf skeleton 使用模式：

• open()：解析 BPF ELF 对象文件，创建 map 和程序的结构体，但不触发任何内核交互
• load()：将 BPF 程序加载到内核——触发 verifier 和 JIT 编译。如果 verifier 拒绝，这一步返回错误
• attach()：将 BPF 程序挂载到 hook 点（tracepoint、kprobe 等）。Skeleton 自动识别 SEC("tp/...") 注解并选择正确的 attach 类型

skeleton 文件由 bpftool gen skeleton mini-agent.bpf.o 自动生成——它解析 BPF ELF 文件，生成包含所有 map 和 prog 引用的结构体。

六、Step 5：CO-RE 可移植性

6.1 CO-RE 的工作原理在 Agent 中的体现

CO-RE 的使用在本 Agent 中有两处体现：

1. vmlinux.h 替代内核头文件：BPF 程序不 #include <linux/*.h>，而是 #include "vmlinux.h"。vmlinux.h 是目标内核 BTF 的 C 语言投影，包含了该内核版本中所有类型的精确定义。

2. BPF_CORE_READ() 宏：替代直接的结构体字段访问。编译后生成 BTF.ext 重定位记录，libbpf 在加载时根据目标内核的 BTF 计算实际字段偏移并修补 BPF 指令：

/* 直接访问（不可移植）：*/
u32 uid = task->cred->uid.val;

/* CO-RE 访问（可移植）：*/
u32 uid = BPF_CORE_READ(task, cred, uid.val);
/* 如果内核 5.x 中 cred 偏移是 0x3b8，
 *   但 6.x 中变成了 0x3c0，
 *   CO-RE 在加载时自动修正 */

6.2 生成 vmlinux.h

# 方法 1：从当前运行的内核生成（推荐）
bpftool btf dump file /sys/kernel/btf/vmlinux format c > vmlinux.h

# 方法 2：从内核编译输出生成（跨内核版本）
bpftool btf dump file /path/to/other-kernel/vmlinux format c > vmlinux_6.8.h

# 方法 3：从 BTF 原始文件生成
bpftool btf dump file /boot/vmlinux-$(uname -r) format c > vmlinux.h

七、Step 6：Map Pinning 热升级

7.1 为什么需要 Map Pinning

当 Agent 重启时，BPF 程序与 ring buffer map 会随进程销毁，但 exec_count array map 可通过 bpffs pin 持久化——重启后复用同一 map，累计 execve 次数不归零。ring buffer 仅负责事件流，不适合作为计数器持久化载体。

# 确保 bpffs 已挂载
mount | grep bpf
# 如果没有，挂载它
mount -t bpf /sys/fs/bpf

# pin map 后，它独立于进程存在

7.2 在 Agent 中实现 Map Pinning

修改用户态代码以支持 map pinning：

/* 添加到 mini-agent.c 的加载流程中 */

#define PIN_DIR "/sys/fs/bpf/mini-agent"

/* 在 __open() 之后、__load() 之前：尝试复用已 pin 的 exec_count */
static int setup_map_pinning(struct mini_agent_bpf *skel)
{
    int fd;

    mkdir(PIN_DIR, 0700);
    fd = bpf_obj_get(PIN_DIR "/exec_count");
    if (fd < 0)
        return 0;

    bpf_map__reuse_fd(skel->maps.exec_count, fd);
    printf("Reusing pinned exec_count map\n");
    return 0;
}

/* 在 __load() 之后：pin 新创建的 exec_count map */
static int pin_maps(struct mini_agent_bpf *skel)
{
    return bpf_map__pin(skel->maps.exec_count, PIN_DIR "/exec_count");
}

修改主函数的加载流程：

/* mini-agent.c main() 中修改的部分 */
skel = mini_agent_bpf__open();
setup_map_pinning(skel);       /* 在 load 前尝试复用 */

err = mini_agent_bpf__load(skel);  /* 加载 */
pin_maps(skel);                /* 在 load 后 pin（如果新建）*/

7.3 Map Pinning 的限制

• Map pin 只持久化 map 数据，不持久化 BPF 程序——重启后程序需要重新加载和 JIT
• 本 Agent 只 pin exec_count；events ring buffer 每次启动新建即可
• 复用已 pin 的 map 时，新程序的 map 属性（大小、类型）必须与已经 pin 的 map 完全一致——否则 bpf_map__reuse_fd() 失败
• bpffs 在系统重启后自动清空（tmpfs），需要 init 脚本重新挂载

八、Step 7：Verifier 排障实战

8.1 有意引入的 Verifier 错误

将 mini-agent 的 handle_execve() 中的 ring buffer NULL 检查去掉：

/* 错误的代码——去掉 NULL 检查 */
SEC("tp/syscalls/sys_enter_execve")
int handle_execve(struct trace_event_raw_sys_enter *ctx)
{
    struct exec_event *event;

    event = bpf_ringbuf_reserve(&events, sizeof(*event), 0);
    /* 如果 reserve 返回 NULL，下面访问 event 字段会触发 verifier 拒绝 */

    event->pid = bpf_get_current_pid_tgid() >> 32;  /* VERIFIER 拒绝在此 */
    ...
}

Verifier 输出（Linux 6.6 典型场景，经删减）：

15: (18) r1 = 0xffffc9000000b000    ; r1 = ringbuf reserve result
17: (7b) *(u64 *)(r10 - 8) = r1     ; store event pointer

18: (85) call bpf_get_current_pid_tgid#14
19: (bf) r2 = r0

; event->pid = pid_tgid >> 32;
20: (77) r2 >>= 32
21: (63) *(u32 *)(r1 + 0) = r2
R1 invalid mem access 'scalar'
processed 20 insns (limit 1000000)
verification time 12 usec
stack depth 48+256

Error: R1 is not pointing to a valid memory object

Verifier 报告 R1（event 指针）是 scalar——不是类型化的指针。这是因为 bpf_ringbuf_reserve() 返回 NULL 的可能性没有被检查。修复：添加 if 保护。

8.2 Verifier 日志级别

/* 在用户态代码中设置 verifier 日志级别 */
struct bpf_object_open_opts opts = {
    .kernel_log_level = 1,    /* 1: 基本日志     */
                               /* 2: 详细日志     */
                               /* 4: 寄存器状态   */
    .kernel_log_buf = log_buf,
    .kernel_log_size = sizeof(log_buf),
};

obj = bpf_object__open_file("mini-agent.bpf.o", &opts);

kernel_log_level 的级别：

8.3 常见 Verifier 错误速查

九、运行与验证

9.1 编译和运行

# 生成 vmlinux.h（只需要运行一次）
bpftool btf dump file /sys/kernel/btf/vmlinux format c > vmlinux.h

# 编译
make

# 运行（需要 CAP_BPF 或 root）
sudo ./mini-agent

# 输出示例：
# COMM             PID    UID    FILENAME
# ----------------------------------------
# bash             12345  1000   /usr/bin/ls
# bash             12345  1000   /usr/bin/cat
# systemd          1      0      /usr/lib/systemd/systemd-logind

9.2 验证 Agent 的正确性

# 检查 BPF 程序是否已加载
sudo bpftool prog list | grep handle_execve

# 查看程序的运行统计
sudo bpftool prog show id <ID>
# 输出包含 run_cnt（执行次数）和 run_time_ns（累计运行时间）

# 查看 ring buffer map
sudo bpftool map list | grep events
sudo bpftool map dump id <MAP_ID>

# 并发测试——触发大量 execve 事件
for i in $(seq 1 1000); do /bin/true; done

9.3 性能评估

# 用 bpftool 查看每次调用的平均开销
sudo bpftool prog show id <ID> --json | jq '{
    run_cnt: .run_cnt,
    avg_ns: (.run_time_ns / .run_cnt)
}'

tracepoint handler 的单次开销取决于 ring buffer 状态与系统负载，通常远低于 kprobe 路径；在 execve 频率不高的服务器上，Agent 的 CPU 占用通常可忽略。具体数值需在本机用 bpftool prog show 的 run_time_ns/run_cnt 或 perf record 实测。

本系列中，本文将前面 17 篇的理论知识落地为可运行的代码——verifier 模型（第 02–04 篇）、map 实现（第 06–07 篇）、libbpf 加载器（第 10 篇）、BTF 与 CO-RE（第 11–12 篇）、ring buffer（第 07 篇）的知识协同运作构成了这个完整 Agent。本文也是第 15 篇（fentry/fexit）的实践延伸——tracepoint 是本次使用的 hook 类型，在高频场景下可以替换为 fentry 以降低开销。ring buffer 的生产-消费模型与第 16 篇的并发语义也有直接关联。

十、从 Mini-Agent 到生产级 Agent

这个 mini-agent 只做了一件事——追踪 execve 并打印。但它展示了生产级 eBPF Agent 所需的全部基础设施。真实的 eBPF 观测项目在这些基础上扩展：

延伸阅读和参考项目： - Falco：运行时安全检测，基于内核模块 + eBPF - Tetragon：Cilium 的 eBPF 安全可观测性框架 - Pixie：Kubernetes 观测（Go + eBPF） - DeepFlow：eBPF 分布式追踪

十一、总结

构建这个 mini-agent 的过程验证了前面 17 篇的所有核心知识点：

1. BPF 程序结构（第 01 篇）：SEC() 注解决定程序类型和挂载点
2. Verifier 要求（第 02–04 篇）：NULL 检查、边界检查、类型验证
3. Map 实现（第 06–07 篇）：ring buffer 的两步提交与 RCU 保护
4. Helper 子系统（第 08 篇）：bpf_probe_read_user_str() 的类型约束
5. 程序生命周期（第 09 篇）：skeleton 的 open/load/attach/destroy
6. libbpf 加载器（第 10 篇）：skeleton 自动生成与 map pinning
7. BTF 格式（第 11 篇）：vmlinux.h 的生成
8. CO-RE 重定位（第 12 篇）：BPF_CORE_READ() 跨版本兼容
9. 调试工具（第 14 篇）：verifier log、bpftool prog dump
10. 并发模型（第 16 篇）：per-CPU ring buffer 的免锁生产

这 10 个知识点的协同运作，构成了一个完整的 eBPF 可观测 Agent。每个知识点独立理解不难，真正的挑战在于让它们协同工作。

参考资料

• Linux 内核源码 tools/lib/bpf/：libbpf 实现
• Linux 内核源码 tools/bpf/bpftool/：bpftool 实现
• Linux 内核源码 tools/testing/selftests/bpf/：BPF 自测用例
• libbpf 文档：libbpf API 概述
• CO-RE 指南：Andrii Nakryiko 的 CO-RE 详解