本文永久链接 – https://tonybai.com/2022/08/12/practices-of-multi-label-based-issue-driven-software-development

软件吞噬世界，开源吞噬软件！基于工单跟踪系统(issue tracking system)的issue driven开发的模式不仅对开源系统的开发过程有着重要影响，在商业软件开发领域，越来越多的公司和组织也在使用issue-driven的开发方式。

对于这里提到的issue(工单)，我们不能向过去那样仅仅将其理解为bug，那样就太过狭义且过时了。如今的issue承载的信息“包罗万象”，凡是与project/repo相关的事情都通过issue形式提出，包括新特性、bug、改善、技术提案、任务、讨论等等。

有了issue tracking system后，我们将面临一个问题，那就是如何组织issue使得基于issue-driven的软件开发过程更加科学有效率呢？为此我拜读了几篇有关issue组织和管理的Empirical study(实证研究)的paper(阅读地址见“参考资料”一节)，这里综合了一下paper的结论：

• 使用标签(label)有利于软件项目中问题的解决；
• 使用多标签(multiple label)功能的项目能更有效地管理他们的issue。

关键词get到了么？没错，就是“多标签”。使用多标签对issue进行组织可以更有效的对issue进行管理。那么设置哪些标签，又如何给issue打多个标签呢？关于这些问题，paper没有给出答案，还是需要我们自行实践探索。

本文我就给出一种基于多label的issue驱动的软件开发实践的思路，供各位童鞋参考，但注意：这不是最佳实践哦，仅是一种可行的(甚至还不成熟的）实践而已，各位读者也可以在该实践的基础上自行扩展。

1. label管理，分类先行

issue tracking system大多内置一套label供你直接选择使用，比如下面就是github和gitlab提供的默认issue labels：

• github的默认issue label集合

- bug
- enhancement
- document
- duplicate
- good first issue
- help wanted
- invalid
- question
- wontfix

• gitlab的默认issue label集合

- bug
- enhancement
- documentation
- suggestion
- discussion
- support
- confirmed
- critical

同时issue tracking system也支持自定义label，但我们不能无脑地定义新label，要考虑label的功用，对其进行科学分类(categorizezation)。

我们分析一下上面github和gitlab提供的默认labels，多数用于表示issue的类别(category)，因此我们首先要定义一组能表示issue类别的labels，这个可综合上述的默认label集合，并结合日常工作上下文的需要来定制。

比如，我这里就定制了如下表示issue类别(category)的label集合：

- bug
- feature
- task
- proposal
- enhancement
- suggestion
- discussion

这些label是整个issue-driven开发的基础，每个新创建的issue必然要首先被赋予(assign)一个代表类别的label，秉着最常用的label要尽量的短的原则，我们用单个单词来表示每个label，没有使用前缀字符串。

接下来，我们再来定义几组label集合：

• 代表工作流(workflow)的label集合

- workflow/confirmed
- workflow/accepted
- workflow/need-investigation
- workflow/wait-for-info
- workflow/need-proposal
- workflow/need-review
- workflow/declined
- workflow/blocked
- workflow/wontfix
- workflow/duplicate

之所以要带上workflow前缀，是为了减少开发人员在选择label时考虑label究竟是哪一类集合的负担，通过前缀一目了然。

• 代表优先级(priority)的label集合

- priority/critical
- priority/release-blocker
- priority/security
- priority/urgent

• 代表涉及组件(component)的label集合

- component/proto
- component/tracing
- component/logging
- component/metrics
- component/api
... ...

在定义不同功用的label集合时，务必注意以下几点：

• 一般不需要定制诸如todo这样的label

哪个放入issue系统中的新建issue不是要todo的，再定义一个todo label有些多此一举的感觉。另外很多自带看板功能的代码仓库管理工具，在issue被纳入某个milestone后，会自动将你新创建的issue列入到该milestone看板中的todo列中。

• issue的版本信息可用milestone来标定

我们一般无需创建持有版本信息的label，可以利用milestone(里程碑)标识issue的版本属性。

• 自上而下定义label集合

对于同一个组织而言，工作流的一致性很重要，这样就需要我们自上而下的定义label集合，而不是每个project都定义自己的label集合。

如果你使用的是gitlab(据调查，国内多数公司使用的都是gitlab)，由于gitlab中group支持label定义，我们可以在顶层group中定义一套label集合，这样其下面的subgroup和project repo都可以“继承”这套label集合了。

如果project有个性化的label需求，project可以在统一的label集合的基础上再自定义自己的label，这样就会避免大量重复定义，也保持了label集合的一致性，为后续workflow运转奠定基础。

定义完这几类issue后，我们再来看看issue从创建到驱动开发“运转”起来的过程是什么样的。

2. 基于多label issue驱动的工作流

1) 规划和创建milestone

无论采用哪种开发过程(瀑布、迭代还是敏捷)，最终都是要输出成果物，我们通常会将每个版本成果物输出作为一个milestone，因此在创建issue之前，我们首先会规划和创建milestone，而milestone本身也内含了issue的版本属性。

如果你使用的是gitlab，你既可以在group这一层设置milestone，也可以在某个具体的project上设置milestone。

如果你的输出输一个包含多个服务的产品，那么在group/subgroup这一层设置milestone可以统一产品的发布版本。

此外，我们可以在适当层级上建立backlog这个长期的milestone用于将那些尚未分配版本的issue收集起来。

一旦如此规划和创建好milestone，leader后续就可以面向milestone进行管理了。接下来我们就可以来创建issue了。

2) 创建issue，选择category label

我们的workflow要求：每当开发人员、测试人员、leader或相关人员创建一个issue后，务必要选择一个category label。

- bug：导致产品或服务未按预期工作的问题
- feature：新增的功能特性或非功能特性
- enhancement：已有特性机制的增强与优化
- task：与产品/服务相关的任务，可能不需要编码
- discussion：发起一次针对特定话题的讨论，需要团队member积极参与
- suggestion：针对产品/服务的建议，需要团队member参与review
- proposal：针对产品某功能特性或非功能特性的技术提案，issue中通过markdown语法填写proposal的提案内容，供大家review

为issue选择一个category label是后续推动workflow流动起来的前提与基础。

3) 选择版本milestone

为刚刚创建的issue选择一个milestone。

对于bug、feature、enhancement类的issue，选择在哪个版本(milestone)里完成。而尚未确定的issue，可以放入backlog milestone，待后续重新放入版本milestone

对于其他category类型的issue，比如proposal、discussion，可以先无需选择版本milestone，可先放入backlog。

4) 选择workflow类别的label

接下来，我们就需要根据issue的类别，为其赋予适当的workflow label，让issue进入工作流中。

• 对于bug类型的issue

对于bug类issue，如果不明确是否为bug或证据不足，可以选择workflow/need-investigation或workflow/wait-for-info；

在调查之后，如果确定是bug，那么需要有开发人员或leader去掉上面的workflow标签，然后重新赋予workflow/confirmed标签。只有经过confirmed的issue，才可以纳入版本milestone跟踪起来，没有confirmed的bug issue，可放入backlog milestone中等待确认；

如果是重复的issue，可以选择workflow/duplicate，然后我们可以close这个duplicate issue；

对于非bug，或经评估后无需fix的issue，可以选择workflow/wontfix，然后close issue。

• 对于feature或enhancement类的issue

如果需要做技术调研，可以选择workflow/need-investigation；

如果需要技术提案/设计实现方案，可选择workflow/need-proposal；

如果需要issue提出者给出更多信息，可以选择workflow/wait-for-info；

如果接受feature或enhancement类issue，可以选择workflow/accepted，然后确认其处于正确的版本milestone中。

• 对于proposal类issue

对于proposal类issue，团队需要决策是否接受该提案，初期可以选择workflow/need-review，在经过review后可以选择是否接受，如果接受，选择workflow/accepted，一旦accepted，便可以新选择纳入哪个milestone中落地。如果不接受，那么选择workflow/declined(拒绝)，然后close issue。

最后，对于因各种原因暂时无法继续下去的issue，可以选择workflow/blocked。等导致阻塞的问题解除后，再为该issue重新选择workflow label。

5) 选择issue优先级(可选)

issue通常无需选择优先级。只要纳入版本milestone，在milestone范围内都是需要完成的。

不过，我们也自定义了一些用于需要“着重”关注的优先级label，供灵活使用。

- critical - 关键issue, milestone内必须要完成的
- release-blocker - 通常针对bug类issue，如果该issue不解决，milestone无法按时完成和发布
- security - 安全类issue，提升issue优先级
- urgent - 紧急突发类issue，可以用于patch补丁的issue

6) 选择component(可选)

为issue选择component类标签更多是为了统计和过滤使用。这里的logging、tracing、metrics等仅做举例之用，大家可以根据自己的产品/project上下文定义必要的component标签。

7) 基于workflow标签驱动

当我们完成一轮label赋值后，开发人员便依据issue在workflow中所处状态开始后续工作，一旦工作告一段落，会改变issue的workflow标签，直到issue完成被close。

3. 小结

本文提出了一种基于多label issue驱动的软件开发的实践思路，通过这种思路，可以让一个团队围绕issue tracing system有机的运转起来。

如果issue众多，人工管理不便的情况下，还可以引入issue bot按设定好的规则对issue的状态做自动维护以及相应的提醒，这个我后续如有空闲会继续说。

4. 参考资料

• 《如何使用Issue管理软件项目》 – https://www.ruanyifeng.com/blog/2017/08/issue.html
• 《The Complete Guide to Issue Tracking Best Practices》 – https://kissflow.com/issue-tracking/issue-tracking-best-practices-guide/
• 《Exploring the use of labels to categorize issues in Open-Source Software projects》- https://ieeexplore.ieee.org/document/7081875
• 《An Empirical Study on Using Multi-Labels for Issues in GitHub》- https://ieeexplore.ieee.org/abstract/document/9550775

“Gopher部落”知识星球旨在打造一个精品Go学习和进阶社群！高品质首发Go技术文章，“三天”首发阅读权，每年两期Go语言发展现状分析，每天提前1小时阅读到新鲜的Gopher日报，网课、技术专栏、图书内容前瞻，六小时内必答保证等满足你关于Go语言生态的所有需求！2022年，Gopher部落全面改版，将持续分享Go语言与Go应用领域的知识、技巧与实践，并增加诸多互动形式。欢迎大家加入！

我爱发短信：企业级短信平台定制开发专家 https://tonybai.com/。smspush : 可部署在企业内部的定制化短信平台，三网覆盖，不惧大并发接入，可定制扩展； 短信内容你来定，不再受约束, 接口丰富，支持长短信，签名可选。2020年4月8日，中国三大电信运营商联合发布《5G消息白皮书》，51短信平台也会全新升级到“51商用消息平台”，全面支持5G RCS消息。

著名云主机服务厂商DigitalOcean发布最新的主机计划，入门级Droplet配置升级为：1 core CPU、1G内存、25G高速SSD，价格5$/月。有使用DigitalOcean需求的朋友，可以打开这个链接地址：https://m.do.co/c/bff6eed92687 开启你的DO主机之路。

Gopher Daily(Gopher每日新闻)归档仓库 – https://github.com/bigwhite/gopherdaily

我的联系方式：

• 微博：https://weibo.com/bigwhite20xx
• 博客：tonybai.com
• github: https://github.com/bigwhite

商务合作方式：撰稿、出书、培训、在线课程、合伙创业、咨询、广告合作。

本文永久链接 – https://tonybai.com/2022/07/25/bidirectional-data-exchange-between-kernel-and-user-states-of-ebpf-programs-using-go

在之前的两篇文章中，无论是使用C语言开发eBPF程序，还是使用Go开发的eBPF程序，都是hello world级别的，可能有用，但谈不上十分实用。

通常来说，一个实用的eBPF程序，它的内核态部分与用户态部分是有数据交换的，有了这种数据交换，eBPF才能发挥更大的威力。而要想让eBPF程序具备较强的实用性，eBPF MAP是绕不过去的机制。

在这一篇有关eBPF程序开发的文章中，我们就来看看如何使用Go基于BPF MAP实现eBPF程序内核态与用户态的双向数据交换。

一. why BPF MAP？

永远不要忘记BPF字节码是运行于OS内核态的代码，这就意味着它与用户态是有“泾渭分明”的界限的。我们知道用户态要想访问内核态的数据，通常仅能通过系统调用陷入内核态来实现。因此，在BPF内核态程序中创建的各种变量实例仅能由内核态的代码访问。

那我们如何将BPF代码在内核态获取到的有用的数据返回到用户态用于监控、计算、决策、展示、存储呢？用户态代码又是如何在运行时向内核态传递数据以改变BPF代码的运行策略呢？

Linux内核BPF开发者于是就引入了BPF MAP机制。BPF MAP为BPF程序的内核态与用户态提供了一个双向数据交换的通道。同时由于bpf map存储在内核分配的内存空间，处于内核态，可以被运行于在内核态的多个BPF程序所共享，同样可以作为多个BPF程序交换和共享数据的机制。

二. BPF MAP不是狭义的map数据结构

BPF MAP究竟是什么呢？它不是我们狭义理解的哈希映射表的数据结构，而是一种通用数据结构，可以存储不同类型数据的通用数据结构。用著名内核BPF开发者Andrii Nakryiko的话来说，MAP就是BPF中代表抽象数据容器(abstract data container)的一个概念。

截至目前，内核BPF支持的MAP类型已经有20+种，下面是libbpf中bpf.h中列出的当前支持的MAP类型：

// libbpf/include/uapi/linux/bpf.h
enum bpf_map_type {
    BPF_MAP_TYPE_UNSPEC,
    BPF_MAP_TYPE_HASH,
    BPF_MAP_TYPE_ARRAY,
    BPF_MAP_TYPE_PROG_ARRAY,
    BPF_MAP_TYPE_PERF_EVENT_ARRAY,
    BPF_MAP_TYPE_PERCPU_HASH,
    BPF_MAP_TYPE_PERCPU_ARRAY,
    BPF_MAP_TYPE_STACK_TRACE,
    BPF_MAP_TYPE_CGROUP_ARRAY,
    BPF_MAP_TYPE_LRU_HASH,
    BPF_MAP_TYPE_LRU_PERCPU_HASH,
    BPF_MAP_TYPE_LPM_TRIE,
    BPF_MAP_TYPE_ARRAY_OF_MAPS,
    BPF_MAP_TYPE_HASH_OF_MAPS,
    BPF_MAP_TYPE_DEVMAP,
    BPF_MAP_TYPE_SOCKMAP,
    BPF_MAP_TYPE_CPUMAP,
    BPF_MAP_TYPE_XSKMAP,
    BPF_MAP_TYPE_SOCKHASH,
    BPF_MAP_TYPE_CGROUP_STORAGE,
    BPF_MAP_TYPE_REUSEPORT_SOCKARRAY,
    BPF_MAP_TYPE_PERCPU_CGROUP_STORAGE,
    BPF_MAP_TYPE_QUEUE,
    BPF_MAP_TYPE_STACK,
    BPF_MAP_TYPE_SK_STORAGE,
    BPF_MAP_TYPE_DEVMAP_HASH,
    BPF_MAP_TYPE_STRUCT_OPS,
    BPF_MAP_TYPE_RINGBUF,
    BPF_MAP_TYPE_INODE_STORAGE,
    BPF_MAP_TYPE_TASK_STORAGE,
    BPF_MAP_TYPE_BLOOM_FILTER,
};

这里数据结构类型众多，但不是本文的重点，我们不一一介绍了。其中的BPF_MAP_TYPE_HASH类型是BPF支持的第一种MAP数据结构，这个类型可以理解为我们日常接触的hash映射表，通过键值对的形式索引数据。在后续的例子中我们将使用这种类型的MAP。

那么BPF MAP是如何可以在内核态与用户态共享数据的？原理是什么呢？

从bpf这个系统调用的说明中，我们能找到端倪。下面是bpf系统调用的函数原型：

// https://man7.org/linux/man-pages/man2/bpf.2.html

#include <linux/bpf.h>

int bpf(int cmd, union bpf_attr *attr, unsigned int size);

从bpf的原型来看，似乎比较简单。但bpf其实是一个“富调用”，即不止能干一件事，通过cmd传入的值不同，它可以围绕BPF完成很多事情。最主要的功能是加载bpf程序(cmd=BPF_PROG_LOAD)，其次是围绕MAP的一系列操作，包括创建MAP(cmd=BPF_MAP_CREATE)、MAP元素查询(cmd=BPF_MAP_LOOKUP_ELEM)、MAP元素值更新(cmd=BPF_MAP_UPDATE_ELEM)等。

当cmd=BPF_MAP_CREATE时，即bpf执行创建MAP的操作后，bpf调用会返回一个文件描述符fd，通过该fd后续可以操作新创建的MAP。通过fd访问map，这个很unix！

当然这么底层的系统调用，一般BPF用户态开发人员无需接触到，像libbpf就包装了一系列的map操作函数，这些函数不会暴露map fd给用户，简化了使用方法，提升了使用体验。

下面我们先来看一下如何用C语言实现基于map的BPF用户态与内核态的数据交换。

三. 使用C基于libbpf使用map的示例

这个示例改造自helloworld示例。原helloworld示例在execve这个系统调用被调用时输出一条内核日志(在/sys/kernel/debug/tracing/trace_pipe中可以查看到)，用户态程序并没有与内核态程序做任何数据交换。

在这个新示例(execve_counter)中，我们依然跟踪系统调用execve，不同的是我们对execve进行调用计数，并将技术存储在BPF MAP中。而用户态部分程序则读取该MAP中的计数并定时输出计数值。

我们先来看看BPF内核态部分的源码：

// https://github.com/bigwhite/experiments/tree/master/ebpf-examples/execve-counter/execve_counter.bpf.c

#include <linux/bpf.h>
#include <bpf/bpf_helpers.h>

typedef __u64 u64;
typedef char stringkey[64];

struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 128);
    //__type(key, stringkey);
    stringkey* key;
    __type(value, u64);
} execve_counter SEC(".maps");

SEC("tracepoint/syscalls/sys_enter_execve")
int bpf_prog(void *ctx) {
  stringkey key = "execve_counter";
  u64 *v = NULL;
  v = bpf_map_lookup_elem(&execve_counter, &key);
  if (v != NULL) {
    *v += 1;
  }
  return 0;
}

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

和helloworld示例不同，我们在新示例中定义了一个map结构execve_counter，通过SEC宏将其标记为BPF MAP变量。

这个map结构有四个字段：

• type: 使用的BPF MAP类型(参见前面的bpf_map_type枚举类型)，这里我们使用BPF_MAP_TYPE_HASH，即一个hash散列表结构；
• max_entries：map内的key-value对的最大数量；
• key: 指向key内存空间的指针。这里我们自定义了一个类型stringkey(char[64])来表示每个key元素的类型；
• value: 指向value内存空间的指针，这里value元素的类型为u64，一个64位整型。

内核态函数bpf_prog的实现也比较简单：在上面的map中查询”execve_counter”这个key，如果查到了，则将得到的value指针指向的内存中的值加1。

我们再来看看execve_counter这个示例的用户态部分的程序源码：

// https://github.com/bigwhite/experiments/tree/master/ebpf-examples/execve_counter/execve_counter.c

#include <stdio.h>
#include <unistd.h>
#include <sys/resource.h>
#include <bpf/libbpf.h>
#include <linux/bpf.h>
#include "execve_counter.skel.h"

typedef __u64 u64;
typedef char stringkey[64];

static int libbpf_print_fn(enum libbpf_print_level level, const char *format, va_list args)
{
    return vfprintf(stderr, format, args);
}

int main(int argc, char **argv)
{
    struct execve_counter_bpf *skel;
    int err;

    libbpf_set_strict_mode(LIBBPF_STRICT_ALL);
    /* Set up libbpf errors and debug info callback */
    libbpf_set_print(libbpf_print_fn);

    /* Open BPF application */
    skel = execve_counter_bpf__open();
    if (!skel) {
        fprintf(stderr, "Failed to open BPF skeleton\n");
        return 1;
    }

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

    /* init the counter */
    stringkey key = "execve_counter";
    u64 v = 0;
    err = bpf_map__update_elem(skel->maps.execve_counter, &key, sizeof(key), &v, sizeof(v), BPF_ANY);
    if (err != 0) {
        fprintf(stderr, "Failed to init the counter, %d\n", err);
        goto cleanup;
    }

    /* Attach tracepoint handler */
    err = execve_counter_bpf__attach(skel);
    if (err) {
        fprintf(stderr, "Failed to attach BPF skeleton\n");
        goto cleanup;
    }

    for (;;) {
            // read counter value from map
            err = bpf_map__lookup_elem(skel->maps.execve_counter, &key, sizeof(key), &v, sizeof(v), BPF_ANY);
            if (err != 0) {
               fprintf(stderr, "Lookup key from map error: %d\n", err);
               goto cleanup;
            } else {
               printf("execve_counter is %llu\n", v);
            }

            sleep(5);
    }

cleanup:
    execve_counter_bpf__destroy(skel);
    return -err;
}

map是在execve_counter_bpf__load中完成的创建，跟踪代码你会发现(参考libbpf源码)，最终会调用bpf系统调用创建map。

和helloworld示例不同的是，我们在attach handler之前，先使用libbpf封装的bpf_map__update_elem初始化了bpf map中的key(初始化为0，如果没有这一步，第一次bpf程序执行时，会提示找不到key)。

然后attach handler后，我们在一个循环中每隔5s通过bpf_map__lookup_elem查询一下key=”execve_counter”的值并输出到控制台。

用户态程序之所以可以直接使用map，是因为bpftool基于execve_counter.bpf.c生成的execve_counter.skel.h中包含了map的各种信息。

接下来我们执行make编译一下这个ebpf程序，然后执行并观察输出：

$sudo ./execve_counter
libbpf: loading object 'execve_counter_bpf' from buffer
libbpf: elf: section(3) tracepoint/syscalls/sys_enter_execve, size 192, link 0, flags 6, type=1
libbpf: sec 'tracepoint/syscalls/sys_enter_execve': found program 'bpf_prog' at insn offset 0 (0 bytes), code size 24 insns (192 bytes)
libbpf: elf: section(4) .reltracepoint/syscalls/sys_enter_execve, size 16, link 22, flags 0, type=9
libbpf: elf: section(5) .rodata, size 64, link 0, flags 2, type=1
libbpf: elf: section(6) .maps, size 32, link 0, flags 3, type=1
libbpf: elf: section(7) license, size 13, link 0, flags 3, type=1
libbpf: license of execve_counter_bpf is Dual BSD/GPL
libbpf: elf: section(13) .BTF, size 898, link 0, flags 0, type=1
libbpf: elf: section(15) .BTF.ext, size 176, link 0, flags 0, type=1
libbpf: elf: section(22) .symtab, size 744, link 1, flags 0, type=2
libbpf: looking for externs among 31 symbols...
libbpf: collected 0 externs total
libbpf: map 'execve_counter': at sec_idx 6, offset 0.
libbpf: map 'execve_counter': found type = 1.
libbpf: map 'execve_counter': found key [9], sz = 64.
libbpf: map 'execve_counter': found value [13], sz = 8.
libbpf: map 'execve_counter': found max_entries = 128.
libbpf: map 'execve_c.rodata' (global data): at sec_idx 5, offset 0, flags 480.
libbpf: map 1 is "execve_c.rodata"
libbpf: sec '.reltracepoint/syscalls/sys_enter_execve': collecting relocation for section(3) 'tracepoint/syscalls/sys_enter_execve'
libbpf: sec '.reltracepoint/syscalls/sys_enter_execve': relo #0: insn #15 against 'execve_counter'
libbpf: prog 'bpf_prog': found map 0 (execve_counter, sec 6, off 0) for insn #15
libbpf: map 'execve_counter': created successfully, fd=4
libbpf: map 'execve_c.rodata': created successfully, fd=5
execve_counter is 0
execve_counter is 0
execve_counter is 9
execve_counter is 23
... ...

注：如果不知道如何编译execve_counter这个示例，请先移步《使用C语言从头开发一个Hello World级别的eBPF程序》了解其构建原理。

bpftool工具提供了查看map的特性，我们可以通过它查看示例创建的map：

$sudo bpftool map
114: hash  name execve_counter  flags 0x0
    key 64B  value 8B  max_entries 128  memlock 20480B
    btf_id 120
116: array  name execve_c.rodata  flags 0x80
    key 4B  value 64B  max_entries 1  memlock 4096B
    frozen

我们还可以dump一下整个map：

$sudo bpftool map dump id 114
[{
        "key": "execve_counter",
        "value": 23
    }
]

我们看到，整个map中就一个键值对(key=”execve_counter”)，其值与示例的用户态部分程序输出的一致。

好了，有了C示例作为基础，我们再来看看如何基于Go来实现这个示例。

四. 使用Go基于cilium/ebpf实现execve-counter示例

使用Go开发BPF用户态部分程序要容易的多，cilium/ebpf提供了的包用起来很简单。如果还不知道如何用Go开发ebpf用户态部分的套路，请先移步《使用Go语言开发eBPF程序》一文了解一下。

Go语言示例的必不可少的原料是execve_counter.bpf.c，这个C源码文件与上面的execve_counter示例中的execve_counter.bpf.c的唯一差别就是include的头文件改成了common.h：

$diff execve_counter.bpf.c ../execve-counter/execve_counter.bpf.c
1,2c1,2
<
< #include "common.h"
---
> #include <linux/bpf.h>
> #include <bpf/bpf_helpers.h>

基于原料execve_counter.bpf.c，bpf2go工具会生成用户态部分所需的Go源码，比如：bpfObject中包含的bpf map实例：

// bpfMaps contains all maps after they have been loaded into the kernel.
//
// It can be passed to loadBpfObjects or ebpf.CollectionSpec.LoadAndAssign.
type bpfMaps struct {
    ExecveCounter *ebpf.Map `ebpf:"execve_counter"`
}

最后，我们在main包main函数中直接使用这些生成的与bpf objects相关的Go函数即可，下面是main.go部分源码：

// https://github.com/bigwhite/experiments/tree/master/ebpf-examples/execve-counter-go/main.go

// $BPF_CLANG, $BPF_CFLAGS and $BPF_HEADERS are set by the Makefile.
//go:generate bpf2go -cc $BPF_CLANG -cflags $BPF_CFLAGS -target bpfel,bpfeb bpf execve_counter.bpf.c -- -I $BPF_HEADERS
func main() {
    stopper := make(chan os.Signal, 1)
    signal.Notify(stopper, os.Interrupt, syscall.SIGTERM)

    // Allow the current process to lock memory for eBPF resources.
    if err := rlimit.RemoveMemlock(); err != nil {
        log.Fatal(err)
    }

    // Load pre-compiled programs and maps into the kernel.
    objs := bpfObjects{}
    if err := loadBpfObjects(&objs, nil); err != nil {
        log.Fatalf("loading objects: %s", err)
    }
    defer objs.Close()

    // init the map element
    var key [64]byte
    copy(key[:], []byte("execve_counter"))
    var val int64 = 0
    if err := objs.bpfMaps.ExecveCounter.Put(key, val); err != nil {
        log.Fatalf("init map key error: %s", err)
    }

    // attach to xxx
    kp, err := link.Tracepoint("syscalls", "sys_enter_execve", objs.BpfProg, nil)
    if err != nil {
        log.Fatalf("opening tracepoint: %s", err)
    }
    defer kp.Close()

    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            if err := objs.bpfMaps.ExecveCounter.Lookup(key, &val); err != nil {
                log.Fatalf("reading map error: %s", err)
            }
            log.Printf("execve_counter: %d\n", val)

        case <-stopper:
            // Wait for a signal and close the perf reader,
            // which will interrupt rd.Read() and make the program exit.
            log.Println("Received signal, exiting program..")
            return
        }
    }
}

在main函数，我们通过objs.bpfMaps.ExecveCounter直接访问map实例，并通过其Put和Lookup方法可以直接操作map。这里要注意的是key的类型必须与execve_counter.bpf.c中的key类型(char[64])保持内存布局一致，不能直接用string类型，否则会在执行时报下面错误：

init map key error: can't marshal key: string doesn't marshal to 64 bytes

编译和执行execve-counter-go和helloworld-go别无二致：

$make
$go run -exec sudo main.go bpf_bpfel.go

2022/07/17 16:59:52 execve_counter: 0
2022/07/17 16:59:57 execve_counter: 14
^C2022/07/17 16:59:59 Received signal, exiting program..

五. 小结

本文介绍了eBPF内核态部分与用户态部分进行数据交换的主要方法：BPF MAP机制。这里的MAP不是狭义的一种hash散列表，而是一个抽象数据结构容器，目前支持二十几种数据结构，大家可以根据自己的需求挑选适当的结构（可查询手册了解各种数据结构的特点)。

MAP本质上也是由bpf系统调用创建的，bpf程序只需要声明map的key、value、type等组成信息即可。用户态可以通过bpf系统调用返回的fd操作map，libbpf和cilium/ebpf等封装了对fd的操作，这样简化了API的使用。

内核中map的update操作不是原子的，因此当有多个bpf程序并发访问一个map时，需要同步操作。bpf提供了bpf_spin_lock来实现对map操作的同步。我们可以在value类型中加入bpf_spin_lock来同步对value的修改，就像下面的例子(例子来自《Linux Observability with BPF》一书)：

struct concurrent_element {
    struct bpf_spin_lock semaphore;
    int count;
}

struct bpf_map_def SEC("maps") concurrent_map = {
    .type = BPF_MAP_TYPE_HASH,
    .key_size = sizeof(int),
    .value_size = sizeof(struct concurrent_element),
    .max_entries = 100,
};

int bpf_program(struct pt_regs *ctx) {
      intkey=0;
      struct concurrent_element init_value = {};
      struct concurrent_element *read_value;
      bpf_map_create_elem(&concurrent_map, &key, &init_value, BPF_NOEXIST);
      read_value = bpf_map_lookup_elem(&concurrent_map, &key);
      bpf_spin_lock(&read_value->semaphore);
      read_value->count += 100;
      bpf_spin_unlock(&read_value->semaphore);
}

本文涉及代码可以在这里下载。

六. 参考资料

• 《揭秘BPF map前生今世》 – https://www.ebpf.top/post/map_internal/
• 《边缘网络eBPF超能力：eBPF map原理与性能解析》 – https://mp.weixin.qq.com/s/Is84xGHFExE1BPkbPpKjwg
• bpf系统调用说明 – https://man7.org/linux/man-pages/man2/bpf.2.html
• 官方bpf map参考手册 – https://www.kernel.org/doc/html/latest/bpf/maps.html
• bpftool参考手册 – https://www.mankier.com/8/bpftool
• 《Building BPF applications with libbpf-bootstrap》 – https://nakryiko.com/posts/libbpf-bootstrap/#bpf-maps

“Gopher部落”知识星球旨在打造一个精品Go学习和进阶社群！高品质首发Go技术文章，“三天”首发阅读权，每年两期Go语言发展现状分析，每天提前1小时阅读到新鲜的Gopher日报，网课、技术专栏、图书内容前瞻，六小时内必答保证等满足你关于Go语言生态的所有需求！2022年，Gopher部落全面改版，将持续分享Go语言与Go应用领域的知识、技巧与实践，并增加诸多互动形式。欢迎大家加入！

我爱发短信：企业级短信平台定制开发专家 https://tonybai.com/。smspush : 可部署在企业内部的定制化短信平台，三网覆盖，不惧大并发接入，可定制扩展； 短信内容你来定，不再受约束, 接口丰富，支持长短信，签名可选。2020年4月8日，中国三大电信运营商联合发布《5G消息白皮书》，51短信平台也会全新升级到“51商用消息平台”，全面支持5G RCS消息。

著名云主机服务厂商DigitalOcean发布最新的主机计划，入门级Droplet配置升级为：1 core CPU、1G内存、25G高速SSD，价格5$/月。有使用DigitalOcean需求的朋友，可以打开这个链接地址：https://m.do.co/c/bff6eed92687 开启你的DO主机之路。

Gopher Daily(Gopher每日新闻)归档仓库 – https://github.com/bigwhite/gopherdaily

我的联系方式：

• 微博：https://weibo.com/bigwhite20xx
• 博客：tonybai.com
• github: https://github.com/bigwhite

商务合作方式：撰稿、出书、培训、在线课程、合伙创业、咨询、广告合作。