描述:

适用于生产环境下的BPF应用的通用基础模板

bootstrap 是一个简单但适用于生产环境下的 BPF 应用。它依赖 BPF CO-RE，同时要求内核编译时配置 CONFIG_DEBUG_INFO_BTF=y。

它会跟踪进程的启动和退出事件，并输出相关数据，包括文件名、进程 ID（PID）、父进程 ID（PPID），以及进程的退出状态和生命周期。

它展示了几个典型的 BPF 功能的使用方式:

  • 协调工作的 BPF 程序
  • 用于维护状态的 BPF map
  • 用于向用户空间发送数据的 BPF ring buffer
  • 用于实现应用程序行为参数化的全局变量
  • 利用 BPF CO-RE 和 vmlinux.h 头文件，从内核的 task_struct 结构体中读取额外的进程信息

运行

  • 显示存活时间至少 50ms 的进程
1
2
3
4
5
6
7
$ ./bootstrap -d 50
TIME     EVENT COMM             PID     PPID    FILENAME/EXIT CODE
10:38:01 EXIT  tail             48249   48247   [0] (133ms)
10:38:01 EXIT  tail             48327   48325   [0] (52ms)
10:38:01 EXIT  tail             48365   48363   [0] (52ms)
10:38:01 EXIT  manpath          48416   48193   [0] (53ms)
10:38:38 EXIT  ping             48530   48193   [0] (5654ms)
  • 开启 libbpf 的 debug 日志
1
$ ./bootstrap -v

The BPF side

头文件

1
2
3
4
5
#include "vmlinux.h"
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_tracing.h>
#include <bpf/bpf_core_read.h>
#include "bootstrap.h"

vmlinux.h 头文件中包含了 BPF 程序可能需要的、与某个内核相关的所有数据结构信息。它是在 libbpf-bootstrap 项目里预先生成的，开发者也可以自动使用 bpftool 生成，具体可参考 gen_vmlinux_h.sh。

TIPS:

只使用 libbpf 提供的 vmlinux.h 头文件就可以，如果同时包含其它内核头文件，会出现重复定义的问题。

Read-only global variable

1
const volatile unsigned long long min_duration_ns = 0;

const volatile 表示该变量不论是在 BPF side，还是在 user-space side，都是只读的。如果不使用 volatile，Clang 就可能假设该变量的值为 0，并将其彻底移除。

BPF maps

BPF_MAP_TYPE_HASH

1
2
3
4
5
6
struct {
	__uint(type, BPF_MAP_TYPE_HASH);
	__uint(max_entries, 8192);
	__type(key, pid_t);
	__type(value, u64);
} exec_start SEC(".maps");

定义一个名为 exec_start，类型为 BPF_MAP_TYPE_HASH 的哈希表。最大容纳 8192 个元素，key 是 pid_t 类型，value 是 u64 类型，用于存储进程运行事件的纳秒粒度的时间戳。

SEC(“tp/sched/sched_process_exec”)

1
2
SEC("tp/sched/sched_process_exec")
int handle_exec(struct trace_event_raw_sched_process_exec *ctx)

bpf_map_update_elem

1
2
3
4
5
6
7
pid_t pid;
u64 ts;

/* remember time exec() was executed for this PID */
pid = bpf_get_current_pid_tgid() >> 32;
ts = bpf_ktime_get_ns();
bpf_map_update_elem(&exec_start, &pid, &ts, BPF_ANY);

在哈希表中添加，更新元素。

  • BPF_ANY: 添加一个新的键（key），或者更新已有的键值对
  • BPF_NOEXIST：仅当键（key）不存在时才会执行更新操作，以防止覆盖已有的数据。
  • BPF_EXIST：仅当键（key）已存在时才会执行更新操作，以确保仅对已有键进行更新，而不创建新的键

SEC(“tp/sched/sched_process_exit”)

1
2
SEC("tp/sched/sched_process_exit")
int handle_exit(struct trace_event_raw_sched_process_template *ctx)

bpf_map_lookup_elem

在另一个 BPF 程序 (handle_exit) 中，从同一个 BPF map 中查询元素，然后删除它。

1
2
3
4
5
6
start_ts = bpf_map_lookup_elem(&exec_start, &pid);
if (start_ts)
    duration_ns = bpf_ktime_get_ns() - *start_ts;
else if (min_duration_ns)
    return 0;
bpf_map_delete_elem(&exec_start, &pid);

BPF ring buffer

BPF_MAP_TYPE_RINGBUF

1
2
3
4
struct {
	__uint(type, BPF_MAP_TYPE_RINGBUF);
	__uint(max_entries, 256 * 1024);
} rb SEC(".maps");

定义一个名为 rb，类型为 BPF_MAP_TYPE_RINGBUF 的环形缓冲区。

bpf_ringbuf_reserve

1
2
3
4
/* reserve sample from BPF ringbuf */
e = bpf_ringbuf_reserve(&rb, sizeof(*e), 0);
if (!e)
    return 0;

bpf_ringbuf_reserve 用于在环形缓冲区申请连续的内存块，第一个参数为指向环形缓冲区的指针，第二个参数为要在环形缓冲区中预留的字节数，第三个参数必须为 0。

BPF_CORE_READ

1
2
3
4
5
6
7
/* fill out the sample with data */
task = (struct task_struct *)bpf_get_current_task();

e->exit_event = false;
e->pid = pid;
e->ppid = BPF_CORE_READ(task, real_parent, tgid);
bpf_get_current_comm(&e->comm, sizeof(e->comm));

BPF_CORE_READ 用于简化 BPF CO-RE 可重定位读取操作，尤其适用于指针追踪步骤较少的场景。BPF_CORE_READ(task, real_parent, tgid) 相当于 task->real_parent->tgid。

ctx->__data_loc_filename

1
2
fname_off = ctx->__data_loc_filename & 0xFFFF;
bpf_probe_read_str(&e->filename, sizeof(e->filename), (void *)ctx + fname_off);

libbpf 访问 tracepoint 上下文字段的格式： __data_loc_<some_field>。（低 16 位偏移 + 高 16 位长度）

& 0xFFFF 用于提取低 16 位的偏移量。

参考：

Learning eBPF: 7.4.3 Tracepointshttps://gaoyangu.github.io/wiki/ebpf/learning-ebpf-07/#7-4-3-Tracepoints

每个跟踪点（tracepoint）都有一个格式（format），该格式用于描述从该跟踪点中会被跟踪输出的字段：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
$ cat /sys/kernel/tracing/events/sched/sched_process_exec/format
name: sched_process_exec
ID: 317
format:
        field:unsigned short common_type;               offset:0;       size:2; signed:0;
        field:unsigned char common_flags;               offset:2;       size:1; signed:0;
        field:unsigned char common_preempt_count;       offset:3;       size:1;signed:0;
        field:int common_pid;                           offset:4;       size:4; signed:1;

        field:__data_loc char[] filename;               offset:8;       size:4; signed:0;
        field:pid_t pid;                                offset:12;      size:4; signed:1;
        field:pid_t old_pid;                            offset:16;      size:4; signed:1;

print fmt: "filename=%s pid=%d old_pid=%d", __get_str(filename), REC->pid, REC->old_pid

bpf_ringbuf_submit

1
2
/* successfully submit it to user-space for post-processing */
bpf_ringbuf_submit(e, 0);

bpf_ringbuf_submit 会使环形缓冲区中预留的数据变为可读取状态。

  • 0：发送自适应的新数据可用通知
  • BPF_RB_NO_WAKEUP：不发送新数据可用通知
  • BPF_RB_FORCE_WAKEUP：无条件发送新数据可用通知

The user-space side

open & load

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
/* Load and verify BPF application */
skel = bootstrap_bpf__open();
if (!skel) {
    fprintf(stderr, "Failed to open and load BPF skeleton\n");
    return 1;
}

/* Parameterize BPF code with minimum duration parameter */
skel->rodata->min_duration_ns = env.min_duration_ms * 1000000ULL;

/* Load & verify BPF programs */
err = bootstrap_bpf__load(skel);
if (err) {
    fprintf(stderr, "Failed to load and verify BPF skeleton\n");
    goto cleanup;
}

只读全局变量需要在 BPF skeleton 加载到内核之前完成设置，因此需要分 __open 和 __load 两步进行。当 BPF skeleton 完成 __load 之后，不论是在 BPF side，还是在 user-space side 都只能读该变量。

attach

1
2
3
4
5
6
/* Attach tracepoints */
err = bootstrap_bpf__attach(skel);
if (err) {
    fprintf(stderr, "Failed to attach BPF skeleton\n");
    goto cleanup;
}

ring_buffer__new

1
2
3
4
5
6
7
8
struct ring_buffer *rb = NULL;

rb = ring_buffer__new(bpf_map__fd(skel->maps.rb), handle_event, NULL, NULL);
if (!rb) {
    err = -1;
    fprintf(stderr, "Failed to create ring buffer\n");
    goto cleanup;
}

环形缓冲区（ring buffer） 指的是一种循环缓冲区，其中 eBPF 程序作为生产者，用户空间（userspace）作为消费者。尽管返回值的数据类型是 struct ring_buffer *，但它实际上是用于管理多个环形缓冲区的管理器。

ring_buffer__poll

1
2
3
4
5
6
7
8
9
10
11
12
13
/* Process events */
while (!exiting) {
    err = ring_buffer__poll(rb, 100 /* timeout, ms */);
    /* Ctrl-C will cause -EINTR */
    if (err == -EINTR) {
        err = 0;
        break;
    }
    if (err < 0) {
        printf("Error polling perf buffer: %d\n", err);
        break;
    }
}

轮询属于环形缓冲区管理器（ring buffer manager）的任一环形缓冲区（ring buffer）上的可用数据。

  • 若存在可用数据，已注册的回调函数（callback functions）将被调用；
  • 若不存在可用数据，该函数会等待 timeout_ms（毫秒）以等待数据到达，并在此期间处于阻塞（block）状态。

clean_up

1
2
3
4
cleanup:
    /* Clean up */
    ring_buffer__free(rb);
    bootstrap_bpf__destroy(skel);

todo

using argp API (part of libc) for command-line argument parsing
参考资料

快来参与讨论吧~