nettrace rtt分析器

开源工具学习记录之流程梳理

近期对腾讯的的开源项目: nettrace(网络故障分析工具) ,进行源码学习。
开源仓库：Nettrace开源仓库
开源工具实现注释：nettrace学习记录

• Nettrace学习记录之流程梳理
• Nettrace eBPF程序自动挂载方式探究

nettrace rtt分析器

DEFINE_ANALYZER_ENTRY(rtt, TRACE_MODE_ALL_MASK)
{
	/*1.通过define_pure_event提取rtt事件数据*/
	define_pure_event(rtt_event_t, event, e->event);
	char *msg = malloc(1024);

	msg[0] = '\0';
	/*2.将first_rtt和last_rtt格式化成字符串，用于输出和日志记录*/
	sprintf(msg, PFMT_EMPH_STR(" *rtt:%ums, rtt_min:%ums*"),
		event->first_rtt, event->last_rtt);
	/*3.entry_set_msg将聚合后的消息绑定到当前分析条目analy_entry_t*/
	entry_set_msg(e, msg);

	return RESULT_CONT;
}
DEFINE_ANALYZER_EXIT_FUNC_DEFAULT(rtt)

1. rtt分析器逻辑

1.1 逻辑梳理

首先看一下哪些跟踪点会触发该分析器:

trace_t trace_tcp_ack_update_rtt = {
	.desc = "",
	.type = TRACE_FUNCTION,
	.analyzer = &ANALYZER(rtt),
	.is_backup = false,
	.probe = false,
	.name = "tcp_ack_update_rtt",
	.sk = 1,
	.custom = true,
	.def = true,
	.index = INDEX_tcp_ack_update_rtt,
	.prog = "__trace_tcp_ack_update_rtt",
	.parent = &group_tcp_state,
	.rules = LIST_HEAD_INIT(trace_tcp_ack_update_rtt.rules),
};
trace_list_t trace_tcp_ack_update_rtt_list = {
	.trace = &trace_tcp_ack_update_rtt,
	.list = LIST_HEAD_INIT(trace_tcp_ack_update_rtt_list.list)
};

在跟踪点的定义中,发现tcp_ack_update_rtt触发该分析器;

看一下该分析器的具体逻辑:

analyzer_result_t analyzer_rtt_exit(trace_t *trace, analy_exit_t *e) __attribute__((weak));
analyzer_result_t analyzer_rtt_entry(trace_t *trace, analy_entry_t *e) __attribute__((weak));
/*analyzer_rtt结构体，trace_t中的.analyzer指向的就是这个结构体*/
analyzer_t analyzer_rtt = {
    .analy_entry = analyzer_rtt_entry,
    .analy_exit = analyzer_rtt_exit,
    .mode = TRACE_MODE_ALL_MASK,
};
analyzer_result_t analyzer_rtt_entry(trace_t *trace, analy_entry_t *e)
{
  	/*1.通过define_pure_event提取rtt事件数据*/
	define_pure_event(rtt_event_t, event, e->event);
	char *msg = malloc(1024);

	msg[0] = '\0';
	/*2.将first_rtt和last_rtt格式化成字符串，用于输出和日志记录*/
	sprintf(msg, PFMT_EMPH_STR(" *rtt:%ums, rtt_min:%ums*"),
		event->first_rtt, event->last_rtt);
	/*3.entry_set_msg将聚合后的消息绑定到当前分析条目analy_entry_t*/
	entry_set_msg(e, msg);

	return RESULT_CONT;
}

analyzer_result_t analyzer_rtt_exit(trace_t *trace, analy_exit_t *e)
{
    rule_run_ret(e->entry, trace, e->event.val);
    return RESULT_CONT;
}

该分析器首先先通过define_pure_event定义pure_rtt_event_t类型的指针event、并指向采集到的原始数据；接着将数据first_rtt, last_rtt等写入msg中；最终调用entry_set_msg将数据放入对应分析器msg中；

1.2 详细分析

上面简单介绍了rtt分析器的实现过程，重点可以放在define_pure_event ，以及entry_set_msg上，前者获取到原始数据， 后者将有用的数据存入分析器analy_entry_t中的msg中；

1.2.1 获取原始数据

在shared.h文件中定义了三个变量：rtt_event_t，detail_rtt_event_t，pure_rtt_event_t,

这三个变量都包含与rtt分析器相关的信息：state, flags, last_update, qlen

#define DEFINE_EVENT(name, fields...)		\
typedef struct {				\
	event_t event;				\
	int __event_filed[0];			\
	fields					\
} name;						\
typedef struct {				\
	detail_event_t event;			\
	int __event_filed[0];			\
	fields					\
} detail_##name;				\
typedef struct {				\
	fields					\
} pure_##name;
#define event_field(type, name) type name;

DEFINE_EVENT(rtt_event_t,
	event_field(u32, first_rtt)
	event_field(u32, last_rtt)
)

define_pure_event函数根据是否需要detail信息，来选择所定义的变量event指向rtt_event_t或detail_rtt_event_t对应的__event_filed，从而获取到原始数据；

#define define_pure_event(type, name, data)			\
	pure_##type *name =					\
		(!trace_ctx.detail ? (void *)(data) +		\
			offsetof(type, __event_filed) :		\
			(void *)(data) +			\
			offsetof(detail_##type, __event_filed))

1.2.2数据处理与记录

通过entry_set_msg将有用数据记录到analy_entry_t中的msg字段；

static inline void entry_set_msg(analy_entry_t *e, char *info)
{
	e->msg = info;
	e->status |= ANALY_ENTRY_MSG;
}

2. rtt全流程分析案例

nettrace提供了 基于RTT的性能分析 功能，rtt模式下可以去分析连接的RTT变化，支持根据rtt和srtt来进行过滤。这里是通过跟踪tcp_ack_update_rtt内核函数的调用来获取套接口的rtt更新事件的，默认情况下是统计RTT的分布情况的，使用方式如下：

./nettrace --rtt

可以通过-detail来输出详细的数据：

该功能通过tcp_ack_update_rtt挂载点获取rtt‘更新数据的时间， 当获取到的数据进入到perf_buffer缓冲区后，会触发数据处理函数stats_poll_handler执行相应的数据处理逻辑，在该逻辑中便会触发rtt分析器，去分析数据的时延以及时延分布直方图；

详细流程如下：

2.1 数据采集

用户通过./nettrace -rtt来开启rtt 功能, nettrace识别到–rtt ,将模式设定为RTT跟踪模式, 并根据RTT跟踪模式初始化其对应的操作函数stats_poll_handler,在加载和挂载bpf程序时将tcp_ack_update_rtt挂载到内核上；

在成功将ebpf程序挂载之后，便初始化perfbuffer缓冲区持续等待内核中数据，当缓冲区中收到了关于rtt的数据之后，便根据trace_poll() ->poll_handler_wrap->trace_ctx.ops->trace_poll去执行与RTT模式绑定的数据处理回调函数stats_poll_handler;

stats_poll_handler是如何进行数据处理的，将会在后面介绍

int main(int argc, char *argv[])
{
	/*1.初始化跟踪组*/
	init_trace_group();
	do_parse_args(argc, argv);
	/*2.跟踪点有效性标记*/
	if (trace_prepare())
		goto err;
	/*3.加载并附加eBPF程序*/
	if (trace_bpf_load_and_attach()) {
		pr_err("failed to load bpf\n");
		goto err;
	}
	/*4.启动网络追踪功能*/
	trace_poll(trace_ctx);

}

在trace_prepare中进行跟踪模式设定、操作集初始化、挂载点预选；

/*遍历所有挂载点,通过比对内核符号表对每个跟踪点进行有效性标记*/
int trace_prepare()
{
	/*1.BTF支持检查*/
	/*2.内核版本兼容检查*/
	
	/*3.跟踪模式设定*/
	err = trace_prepare_args();
	
	/*4.初始化操作集
	 *  遍历trace_ops_all选择支持当前系统的操作集
	 */
	for (; i < ARRAY_SIZE(trace_ops_all); i++) {
		if (trace_ops_all[i]->trace_supported()) {
			set_trace_ops(trace_ops_all[i]);
			break;
		}
	}
	
	/*5.挂载点的预处理*/
	err = trace_prepare_traces();
	
	/*6.确保以root权限运行*/
	/*7.检查内核特性*/
	/*8.只启用第一个有校挂载点,其他的备用挂载点标记为无效*/
	/*9.打印所有有效挂载点*/
}

在trace_bpf_load_and_attach 中将bpf程序加载并挂载在预选的跟踪点上，并根据RTT跟踪模式将数据处理回调函数设定为stats_poll_handler；

/*加载并附加eBPF程序*/
int trace_bpf_load_and_attach()
{
	/*1.load_bpf*/
	trace_bpf_load();
	/*2.attach bpf*/

	/*3.准备ops操作函数*/
	trace_prepare_ops();
}

2.1.1 RTT跟踪模式设定

main->trace_prepare->trace_prepare_args 设定为RTT跟踪模式:

static int trace_prepare_args()
{
	...
	ASSIGN_MODE(rtt, RTT);
	...
	if (bpf_args->first_rtt || bpf_args->last_rtt)
		trace_tcp_ack_update_rtt.monitor = 2;
	/*根据当前的跟踪模式，对相应的追踪点进行标记*/
	if (trace_prepare_mode(args))
		goto err;
     ...
 }

2.1.2 RTT数据处理

main->trace_bpf_load_and_attach->trace_prepare_ops设定RTT模式下的数据处理函数

static void trace_prepare_ops()
{
	...
	switch (trace_ctx.mode) {
	...
	case TRACE_MODE_RTT:
		trace_ctx.ops->raw_poll = stats_poll_handler;
	...
	}	

2.1.3 数据监听

点那个ebpf程序加载并挂载之后，便开始在内核中采集相应的数据，并将数据发送至perfbuffer, 并触发回调函数stats_poll_handler进行数据处理；

int trace_poll()
{
	/*1.获取map描述符*/
	
	/*2.创建perf_buffer 用于监听eBPF程序的输出；
	 *  poll_handler_wrap函数根据指定的模式进行数据的聚合与分析；
	 *  trace_on_lost 用于处理丢失事件的回调函数；
	 */
	struct perf_buffer_opts pb_opts = {
		.sample_cb = poll_handler_wrap,
		.lost_cb = trace_on_lost,
	};
	
	/*3. 检查perf buffer是否创建成功*/
	 
	/*4.开始监听数据
	 *  调用perf_buffer__poll 监听数据,每次有数据进来，都会触发回调函数poll_handler_wrap；
	*/
	while ((err = perf_buffer__poll(pb, 1000)) >= 0) {
		if (poll_timeout(err))
			break;
	}
}

2.2 数据处理

RTT跟踪模式下对应的数据处理回调函数是stats_poll_handler；该函数持续接收来自bpf_map中的数据，并将其

/**
 * stats_poll_handler - 处理统计数据并以指定格式输出的处理器函数
 * 
 * 功能：
 * - 读取 BPF map `m_stats` 中的统计数据。
 * - 根据模式选择不同的统计类型（RTT 或延迟）。
 * - 输出统计分布信息。
 *
 * 返回值：
 * - 成功时返回 0，失败时返回错误码。
 */
int stats_poll_handler()
{
    /*1. 获取名为 "m_stats" 的 BPF map 文件描述符*/
    int map_fd = bpf_object__find_map_fd_by_name(trace_ctx.obj, "m_stats");


    /*2. 持续循环处理统计数据，直到追踪停止*/
    while (!trace_stopped()) {
        int start = 0, j;
        __u64 total = 0; // 统计总数

        // 从 BPF map 中读取统计数据，并计算总数
        for (i = 0; i < 16; i++) {
            bpf_map_lookup_elem(map_fd, &i, count + i);
            total += count[i];
        }

        // 打印统计分布的标题和总数
        pr_info("%-34s%llu\n", header, total);

        // 遍历每个桶并输出统计分布
        for (i = 0; i < 16; i++) {
            bool has_count = false; // 标志当前桶及其后续桶是否有数据
            int p = 0, t = 0;       // 百分比和精确度的临时变量

            // 检查当前桶及其后续桶是否有数据
            for (j = i; j < 16; j++) {
                if (count[j])
                    has_count = true;
            }

            // 如果当前桶及后续桶都没有数据，且索引已超过 8，则提前结束循环
            if (!has_count && i > 8)
                break;

            // 计算当前桶的范围
            start = 1 << i; // 当前桶的起始值（指数）
            sprintf(buf, "%d - %5d%s", start == 1 ? 0 : start,
                    (start << 1) - 1, unit); // 生成桶范围字符串

            // 如果总数不为零，则计算百分比
            if (total) {
                p = count[i] / total;              // 整数部分
                t = (count[i] % total) * 10000 / total; // 小数部分（四位精度）
            }

            // 打印当前桶的统计信息
            pr_info("%32s: %-8llu %d.%04d\n", buf, count[i], p, t);
        }

        // 休眠 1 秒后继续处理下一轮统计数据
        sleep(1);
    }

    return 0; // 处理成功返回 0
}

2.3 数据分析

见1.rtt分析逻辑

            p = count[i] / total;              // 整数部分
            t = (count[i] % total) * 10000 / total; // 小数部分（四位精度）
        }

        // 打印当前桶的统计信息
        pr_info("%32s: %-8llu %d.%04d\n", buf, count[i], p, t);
    }

    // 休眠 1 秒后继续处理下一轮统计数据
    sleep(1);
}

return 0; // 处理成功返回 0

}