eBPFEndpoint
南风未起 2026-08-23 19:39:22 cni
# 1.apply
# 1.1.时机
dataplane将event推送各模块，会执行d.apply()驱动各模块下发变更，覆盖iptables/route/ipset/eBPF，感兴趣可以看dataplane。func (d *InternalDataplane) loopUpdatingDataplane() { ... for { select { case msg := <-d.toDataplane: d.onDatastoreMessage(msg) case ifaceUpdate := <-d.ifaceUpdates: d.onIfaceMonitorMessage(ifaceUpdate) ... } if d.datastoreInSync && d.ifaceMonitorInSync && d.dataplaneNeedsSync { // Dataplane is out-of-sync, check if we're throttled. if d.applyThrottle.Admit() { ... // Actually apply the changes to the dataplane. d.apply() ... } else { if !beingThrottled { log.Info("Dataplane updates throttled") beingThrottled = true } } } } } func (d *InternalDataplane) apply() { ... // 这里会触发manager的apply for _, mgr := range d.allManagers { mgr.CompleteDeferredWork() ... } ... }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39这里主要看
manager触发时机，ipset/route/xdp/iptables比较特殊，走的是单独分支
# 1.2.下发
m.CompleteDeferredWork()会执行首次初始化对齐内存和内核状态，初始化完成后基于内存状态进行增量同步和BPF挂载，维护网卡相关Map。func (m *bpfEndpointManager) CompleteDeferredWork() error { // Do one-off initialisation. m.startupOnce.Do(func() { // 首次初始化,启动BPF Map清理协程+扫描已挂载程序+加载ifState缓存 m.dp.ensureStarted() // 非数据网卡/Pod网卡/L3网卡,清理未知网卡的BPF挂载,其实同步阶段已经清理过一次 m.initUnknownIfaces.Iter(func(iface string) error { // 挂载过BPF if ai, ok := m.initAttaches[iface]; ok { // 执行cgo交互内核,执行TC/XDP程序卸载+元数据清理 m.cleanupOldAttach(iface, ai) ... delete(m.initAttaches, iface) return set.RemoveItem } return nil }) // 内核加载的ifstate条目同步,已不存在的网卡标记清理 m.syncIfStateMap() m.initUnknownIfaces = nil // iface counter同步 m.syncIfaceCounters() ... }) // dirty iface挂载BPF程序(主机网卡,eth0,bofout,lo...) m.applyProgramsToDirtyDataInterfaces() // Pod网卡挂载BPF m.updateWEPsInDataplane() ... // CTLB兼容模式 if m.ctlbWorkaroundMode != ctlbWorkaroundDisabled { // dirty service路由设置 m.dirtyServices.Iter(func(svc serviceKey) error { for _, ip := range m.services[svc] { // service cidr-->bpfin m.dp.setRoute(ip) } return set.RemoveItem }) } // iface state map更新提交到内核 m.ifStateMap.ApplyAllChanges() ... // BPF及路由设置完成的wep,渲染特殊iptables chain,表示Pod流量Accept if m.happyWEPsDirty { // 渲染iptables规则(cali-wl-iface-allow chain) chains := m.ruleRenderer.WorkloadInterfaceAllowChains(m.happyWEPs) // 更新chain(内存侧,延迟到iptables manager批量apply时写内核) m.iptablesFilterTable.UpdateChains(chains) m.happyWEPsDirty = false } ... // conntrack map迁移(升级场景) m.copyDeltaOnce.Do(func() { // 旧版本BPF CTMap条目拷贝到新版BPF Map,迁移完成关闭旧ctMapFD及pin文件 m.bpfmaps.CtMap.CopyDeltaFromOldMap() ... }) return nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
注意
eBPF模式下依然需要iptables chain，用于放行就绪的Pod接口流量，确保Forward链中来自Pod网卡的流量不会被iptables丢弃
# 2.状态对齐
# 2.1.ensure
m.ensureStarted()会启动后台协程清理过期的jumpMap，扫描同步已有网卡挂载的BPF程序，加载内核存在的ifStateMap内容重建内存状态。func (m *bpfEndpointManager) ensureStarted() { // 间隔10s清理一次/sys/fs/bpf/tc未使用数据 m.mapCleanupRunner.Start(context.Background()) ... // calico管理网卡挂载的tc/xdp m.initAttaches, err = bpf.ListCalicoAttached() ... // 基于iterator+KV序列化封装读取内核数据 m.ifStateMap.LoadCacheFromDataplane() ... } func (r *Runner) Start(ctx context.Context) { go r.loop(ctx) } func (r *Runner) loop(ctx context.Context) { ... for { select { case <-ctx.Done(): return // m.updateWEPsInDataplane()通知触发 case <-r.triggerC: triggered = true // 间隔10s触发一次 case <-timerC: timerC = nil } ... // 自上一次触发超过10s if triggered && delayToNextClean <= 0 { // 触发一次 r.callback(ctx) triggered = false lastTriggerTime = r.time.Now() // 否则,重置一下定时器 } else if timerC == nil { if timer == nil { timer = r.time.NewTimer(delayToNextClean) } else { timer.Reset(delayToNextClean) } timerC = timer.Chan() } } } // callback回调的是这里 func(ctx context.Context) { m.cleanupLock.Lock() defer m.cleanupLock.Unlock() // 这个函数之前出现过 // 1.扫描/sys/fs/bpf/tc的jumpMap文件 // 2.扫描网卡挂载的BPF程序及内核加载的BPF程序 // 3.清理网卡挂载未使用的驱动jumpMap文件或空目录 bpf.CleanUpMaps() }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
补充
mapCleanupRunner是限流清理器，响应updateWEPsInDataplane的主动触发，也有10秒保底周期清理，防止Pin文件泄漏
# 2.2.listAttached
ListCalicoAttached()通过bpftool工具链扫描当前系统中所有calico管理的TC/XDP挂载，用于Felix重启或升级时识别遗留状态。// list all programs that are attached to TC or XDP and are related to Calico. That is, they have jumpmap pinned // in our dir hierarchy. func ListCalicoAttached() (map[string]EPAttachInfo, error) { // 执行bpftool net -j列出网卡tc/xdp挂载 aTC, aXDP, err := ListTcXDPAttachedProgs() ... // 整理挂载点 for _, p := range aTC { attachedProgIDs.Add(p.ID) } for _, p := range aXDP { attachedProgIDs.Add(p.ID) } // 1.扫描/sys/fs/bpf/tc目录,匹配jumpMap内核文件句柄 // 2.执行bpftool map show pinned命令获取jumpMap元数据。 maps, err := ListPerEPMaps() ... // 执行bpftool --json --pretty prog show获取内核eBPF程序详细信息 // [ // { // "id": 123, // "type": "xdp", // "name": "calico_xdp_prog", // "tag": "abcdef1234567890", // "gpl_compatible": true, // "loaded_at": "2026-08-11T10:30:00+0800", // "uid": 0, // "bytes_xlated": 1024, // "jited": true, // "bytes_jited": 512, // "bytes_memlock": 4096, // "map_ids": [456, 789], // "btf_id": 42 // } // ] allProgs, err := GetAllProgs() ... for _, p := range allProgs { // 仅保留网卡挂载的 if !attachedProgIDs.Contains(p.Id) { continue } // calico管理的网卡挂载 for _, m := range p.MapIds { if _, ok := maps[m]; ok { caliProgs.Add(p.Id) break } } } // 记录dev-->tc挂载 for _, p := range aTC { if caliProgs.Contains(p.ID) { ai[p.DevName] = EPAttachInfo{TCId: p.ID} } } // 记录dev-->xdp挂载 for _, p := range aXDP { if caliProgs.Contains(p.ID) { info := ai[p.DevName] info.XDPId = p.ID info.XDPMode = p.Mode ai[p.DevName] = info } } return ai, nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
注意
这里加载的
calico管理网卡挂载的BPF程序卸载阶段会用到，挂载过的才会执行卸载
# 2.3.loadCache
CachingMap是BPF Map的抽象，采用三缓冲实现差分提交，LoadCacheFromDataplane在启动时从内核全量加载当前Map内容作为对比基线。// loads the contents of the DP map into the dataplane cache, allowing it to be queried with // GetDataplaneCache and IterDataplaneCache. func (c *CachingMap[K, V]) LoadCacheFromDataplane() error { // 重置一下pendingUpdate/pendingDel c.initCache() // 加载内核ifStateMap数据 dp, err := c.dpMap.Load() ... c.cacheOfDataplane = dp // 计算差异(desiredState对比cache,生成pendingUpdates/pendingDeletions) c.recalculatePendingOperations() return nil } func (m *TypedMap[K, V]) Load() (map[K]V, error) { memMap := make(map[K]V) // 执行C.bpf_maps_map_load_multi获取内核ifstateMap的KV数据放到内存 // 这里的封装可以避免直接操作内核缓冲区/KV字节序列化/迭代批次 err := m.MapWithExistsCheck.Iter(func(kb, vb []byte) IteratorAction { // 读出的KV利用构造器解析放入内存 memMap[m.kConstructor(kb)] = m.vConstructor(vb) return IterNone }) return memMap, err } // compares the dataplane cache against he desired state and adds entries to pendingUpdates/pendingDeletions. func (c *CachingMap[K, V]) recalculatePendingOperations() { // Look for any discrepancies and queue up updates. for k, desiredVal := range c.desiredStateOfDataplane { actualVal := c.cacheOfDataplane[k] // 同步的网卡期望和内核状态对不上 if actualVal != desiredVal { c.pendingUpdates[k] = desiredVal } } // Scan for any dataplane keys that are not in the desired map at all and queue up deletions. for k, actualVal := range c.cacheOfDataplane { desiredVal, ok := c.desiredStateOfDataplane[k] // 内核多出的ifaceState if !ok { c.pendingDeletions[k] = actualVal } } }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
# 3.Map同步
# 3.1.ifstate
m.syncIfStateMap()会扫描内核ifaceStateMap条目，将仍然存在的网卡条目注册到desiredState(标记为需要保留)，不存在的标记为删除。func (m *bpfEndpointManager) syncIfStateMap() { // 扫描内核加载的ifstate状态 m.ifStateMap.IterDataplaneCache(func(k ifstate.Key, v ifstate.Value) { ifindex := int(k.IfIndex()) _, err := net.InterfaceByIndex(ifindex) // 网卡已经不存在 if err != nil { // "net" does not export the strings or err types :( if strings.Contains(err.Error(), "no such network interface") { // 清理ifstate期望状态 m.ifStateMap.DeleteDesired(k) } // 网卡存在 } else { // 注册到期望态 m.ifStateMap.SetDesired(k, v) } }) } // sets the desired state of the given key to the given value. This is an in-memory operation, func (c *CachingMap[K, V]) SetDesired(k K, v V) { c.desiredStateOfDataplane[k] = v if c.cacheOfDataplane == nil { return // Initial sync is pending, we're not tracking deltas yet. } // 由pendingDel清理 delete(c.pendingDeletions, k) // Check if we think we need to update the dataplane as a result. currentVal, ok := c.cacheOfDataplane[k] if ok && currentVal == v { // Dataplane already agrees with the new value so clear any pending update. delete(c.pendingUpdates, k) return } c.pendingUpdates[k] = v } // DeleteDesired deletes the given key from the desired state of the dataplane. This is an in-memory operation, // it doesn't actually touch the dataplane. func (c *CachingMap[K, V]) DeleteDesired(k K) { delete(c.desiredStateOfDataplane, k) if c.cacheOfDataplane == nil { return // Initial sync is pending, we're not tracking deltas yet. } // 由pendingUpdate清理 delete(c.pendingUpdates, k) // Check if we need to update the dataplane. currentVal, ok := c.cacheOfDataplane[k] if !ok { // We don't think this value is in the dataplane so clear any pending delete. delete(c.pendingDeletions, k) return } c.pendingDeletions[k] = currentVal }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60注意
SetDesired/DeleteDesired是内存操作，仅更新Pending队列，实际BPF系统调用在ApplyAllChanges批量执行
# 3.2.ifcounter
m.syncIfaceCounters会扫描内核cali_v4_counters文件数据，清理已不存在网卡的计数条目，防止BPF Map出现空间泄漏及指标虚统计问题。func (m *bpfEndpointManager) syncIfaceCounters() error { // 获取主机所有网卡 ifaces, err := net.Interfaces() ... for i := range ifaces { exists.Add(ifaces[i].Index) } // 执行C.bpf_maps_map_load_multi获取内核ifcounterMap内容 err = m.bpfmaps.CountersMap.Iter(func(k, v []byte) maps.IteratorAction { ... copy(key[:], k) // counter对应网卡没了,执行C.bpf_maps_map_call交互del接口清理网卡计数数据 if !exists.Contains(key.IfIndex()) { return maps.IterDelete } return maps.IterNone }) ... return nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24注意
不同于
ifState使用三缓冲模型，ifCounters Map会在迭代读取内核数据阶段就执行清理
# 4.hep网卡
# 4.1.attach
m.applyProgramsToDirtyDataInterfaces()会处理标记为dirty的主机网卡，为每个接口挂载TC Ingress/Egress和XDP程序。func (m *bpfEndpointManager) applyProgramsToDirtyDataInterfaces() { ... m.dirtyIfaceNames.Iter(func(iface string) error { // 仅处理主机数据网卡(eth0/bpfout/lo...)和L3网卡 if !m.isDataIface(iface) && !m.isL3Iface(iface) { return nil } // 网卡不是UP,从dirty集合移除 if !m.ifaceIsUp(iface) { return set.RemoveItem } ... go func() { ... // tc qdisc show检查,没有执行C.bpf_tc_hook_create创建 m.dp.ensureQdisc(iface) ... // 获取hep策略 if hep, hepExists := m.hostIfaceToEpMap[iface]; hepExists { hepPtr = &hep } ... go func() { ... // hep网卡挂载tc ingress(外部→主机) // CT查找→NAT查找(DNAT)→策略匹配→FIB转发/重定向 m.attachDataIfaceProgram(iface, hepPtr, PolDirnIngress) }() ... go func() { ... // hep网卡挂载xdp(仅Untracked策略,防IP欺骗) m.attachXDPProgram(iface, hepPtr) }() // hep网卡挂载tc egress(主机→外部) // CT查找→SNAT→策略匹配→FIB/encap m.attachDataIfaceProgram(iface, hepPtr, PolDirnEgress) ... if err == nil { // 设置/proc/sys/net/ipv4/conf/%s/accept_local=1接收本地流量 _ = m.dp.setAcceptLocal(iface, true) } ... }() return nil }) ... // 更新ifstate isReady状态,清理dirty for iface, err := range errs { isReady := true if err == nil { m.dirtyIfaceNames.Discard(iface) } else { isReady = false if isLinkNotFoundError(err) { m.dirtyIfaceNames.Discard(iface) } } m.withIface(iface, func(i *bpfInterface) bool { i.dpState.isReady = isReady m.updateIfaceStateMap(iface, i) return false // no need to enforce dirty }) } }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
注意
挂载成功设置
accept_local=1是为了让流量经bpfout处理后源IP变为tunnel IP(本地地址)时能被正常接收
# 4.2.xdpAttach
m.attachXDPProgram()仅处理HostEndpoint的Untracked Tiers(TC前执行,用于防IP欺骗和早期丢弃)，挂载逻辑比TC简单。func (m *bpfEndpointManager) attachXDPProgram(ifaceName string, ep *proto.HostEndpoint) error { ap := &xdp.AttachPoint{ Iface: ifaceName, LogLevel: m.bpfLogLevel, Modes: m.xdpModes, } if ep != nil && len(ep.UntrackedTiers) == 1 { // 有Untracked策略,确保XDP程序挂载并更新策略 jumpMapFD, err := m.dp.ensureProgramAttached(ap) ... // 这里是动态编译Prog策略程序,放在后面说,TC也会用到 rules := polprog.Rules{ ForHostInterface: true, HostNormalTiers: m.extractTiers(ep.UntrackedTiers[0], PolDirnIngress, false), ForXDP: true, } // 编译Prog策略程序加载到内核,关联到jumpMapFD,这样Pin路径的jumpMap文件就能看到 return m.dp.updatePolicyProgram(jumpMapFD, rules, "xdp", ap) } else { // 无Untracked策略,确保XDP程序已卸载 return m.dp.ensureNoProgram(ap) } }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26xdp AttachPoint挂载点生成要简单一点，会基于xdp模式优先级逐个尝试挂载xdp程序，策略程序关联和TC程序一样基于jumpMap记录。// Ensure TC/XDP program is attached to the specified interface and return its jump map FD. func (m *bpfEndpointManager) ensureProgramAttached(ap attachPoint) (maps.FD, error) { // iface关联jumpMapFD jumpMapFD := m.getJumpMapFD(ap) if jumpMapFD != 0 { // 执行C.bpf_xdp_program_id检查xdp挂载 attached, err := ap.IsAttached() ... // 未挂载过 if !attached { // 关闭关联jumpMapFD jumpMapFD.Close() ... // 清理iface.dpState.jumpMapFDs缓存 m.setJumpMapFD(ap, 0) jumpMapFD = 0 // Trigger program to be re-added below. } } // 重新挂载 if jumpMapFD == 0 { ... // 执行网卡xdp挂载 progID, err := ap.AttachProgram() ... // 获取jumpMapFD(bpftool prog show id <progID> --json获取jumpMapID,系统调用获取FD) jumpMapFD, err = FindJumpMap(progID, ap.IfaceName()) ... // 设置到iface.dpState.jumpMapFDs m.setJumpMapFD(ap, jumpMapFD) } return jumpMapFD, nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39ap.AttachProgram()会基于挂载点配置编译xdp程序，设置BPF map全局变量，将加载到内核的xdp程序关联到主机网卡及记录挂载元数据。func (ap *AttachPoint) AttachProgram() (int, error) { ... // 预编译的xdp程序路径/usr/lib/calico/bpf/xdp_xxx.o filename := ap.FileName() preCompiledBinary := path.Join(bpf.ObjectDir, filename) // 打开xdp程序,提取程序Map obj, err := libbpf.OpenObject(preCompiledBinary) ... defer obj.Close() for m, err := obj.FirstMap(); m != nil && err == nil; m, err = m.NextMap() { // 内部映射(libbpf .rodata常量段) if m.IsMapInternal() { // 设置.rodata全局配置(仅IfaceName, XDP全局数据比TC简单) ConfigureProgram(m, ap.Iface) ... continue } // 设置pin路径/sys/fs/bpf/tc/iface_xdp,关联jumpMap会Pin到这里 pinDir := bpf.MapPinDir(m.Type(), m.Name(), ap.Iface, bpf.HookXDP) m.SetPinPath(path.Join(pinDir, m.Name())) ... } // XDP挂载检查: 读取/var/run/calico/bpf/prog/{iface}_xdp.json元数据, // 对比progID/object路径/.o文件SHA256 hash,全部一致则跳过重挂载 progID, isAttached := ap.AlreadyAttached(preCompiledBinary) if isAttached { return progID, nil } // 加载BPF程序至内核 obj.Load() ... // 执行C.bpf_update_jump_map更新jumpMap // calico_xdp_norm_pol_tail(正常策略)/calico_xdp_accepted_entrypoint(允许后处理)/calico_xdp_drop(丢弃) updateJumpMap(obj) ... // 获取网卡挂载的旧 xdp progID oldID, err := ap.ProgramID() ... for _, mode := range ap.Modes { // 执行C.bpf_program_attach_xdp挂载xdp BPF程序,这里相当于替换 // 按优先级尝试: XDP_FLAGS_HW_MODE(Offload) → XDP_FLAGS_DRV_MODE(Driver) → XDP_FLAGS_SKB_MODE(Generic) progID, err = obj.AttachXDP(ap.Iface, ap.ProgramName(), oldID, unix.XDP_FLAGS_REPLACE|uint(mode)) ... break } ... // xdp挂载元数据写入/var/run/calico/bpf/prog/iface_xdp.json bpf.RememberAttachedProg(ap, preCompiledBinary, progID) ... return progID, nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62m.ensureNoProgram()在没有Untracked会执行xdp程序卸载，会解绑网卡关联xdp程序，清理用户侧的json挂载元数据。// Ensure that the specified attach point does not have our program. func (m *bpfEndpointManager) ensureNoProgram(ap attachPoint) error { // Clean up jump map FD if there is one. jumpMapFD := m.getJumpMapFD(ap) if jumpMapFD != 0 { // Close the jump map FD. jumpMapFD.Close() ... m.setJumpMapFD(ap, 0) ... } // Ensure interface does not have our program attached. ap.DetachProgram() ... return err } func (ap *AttachPoint) DetachProgram() error { // Get the current XDP program ID, if any. progID, err := ap.ProgramID() ... // 已卸载 if progID == DetachedID { return nil } // 确认是Calico的XDP程序(对比/var/run/calico/bpf/prog/xx.json元数据) ourProg, err := bpf.AlreadyAttachedProg(ap, path.Join(bpf.ObjectDir, ap.FileName()), progID) ... // Try to remove our XDP program in all modes for _, mode := range ap.Modes { // 执行C.bpf_xdp_detach卸载网卡xdp挂载 libbpf.DetachXDP(ap.Iface, uint(mode)) ... // 执行C.bpf_xdp_program_id获取网卡挂载xdpID curProgId, err := ap.ProgramID() ... // 卸载完成 if curProgId == DetachedID { removalSucceeded = true break } } ... // 清理/var/run/calico/bpf/prog/iface_xdp.json元数据 bpf.ForgetAttachedProg(ap.IfaceName(), bpf.HookXDP) ... return nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
问题
TC BPF挂载会主要卸载旧程序，这里好像更新旧程序，未主动卸载
# 4.3.tcAttach
m.attachDataIfaceProgram是主机网卡TC挂载的核心函数，会构建AttachPoint配置、执行挂载和确认及生成Prog程序关联到jumpMap。func (m *bpfEndpointManager) attachDataIfaceProgram(ifaceName string, ep *HostEndpoint, p PolDirection) error { // 主机IP是BPF运行的必要参数(SNAT/DNAT) if m.hostIP == nil { return fmt.Errorf("unknown host IP") } // 生成AttachPoint配置,BPF会基于这份配置编译 ap := m.calculateTCAttachPoint(p, ifaceName) ap.HostIP = m.hostIP ap.TunnelMTU = uint16(m.vxlanMTU) ap.ExtToServiceConnmark = uint32(m.bpfExtToServiceConnmark) ip, err := m.getInterfaceIP(ifaceName) if err != nil { ap.IntfIP = m.hostIP } else { ap.IntfIP = *ip } ap.NATin = uint32(m.natInIdx) ap.NATout = uint32(m.natOutIdx) // 挂载tc程序(返回Jump Map FD) jumpMapFD, err := m.dp.ensureProgramAttached(ap) ... // ep不为空,更新关联策略 if ep != nil { rules := polprog.Rules{ ForHostInterface: true, } // 将关联的Policy/Profile策略整理成rules m.addHostPolicy(&rules, ep, p) // 编译生成Prog程序加载到内核,关联到jumpMapFD,这样Pin路径的jumpMap文件就能看到 return m.dp.updatePolicyProgram(jumpMapFD, rules, p.RuleDir(), ap) } // 无ep,清理jumpMap关联策略 m.dp.removePolicyProgram(jumpMapFD, ap) ... return nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43m.ensureProgramAttached()检查TC程序挂载，没有挂载执行C.bpf_tc_program_attach关联到网卡上，同时也会记录挂载元数据到json。// Ensure TC/XDP program is attached to the specified interface and return its jump map FD. func (m *bpfEndpointManager) ensureProgramAttached(ap attachPoint) (maps.FD, error) { // iface关联jumpMapID jumpMapFD := m.getJumpMapFD(ap) if jumpMapFD != 0 { // 执行tc qdisc show/tc filter show检查iface是否仍挂载tc ingress/egress attached, err := ap.IsAttached() ... // 未挂载过(缓存FD过期,如程序被外部清理 if !attached { // 关闭jumpMapFD jumpMapFD.Close() ... // 清理iface.dpState.jumpMapFDs缓存 m.setJumpMapFD(ap, 0) jumpMapFD = 0 // Trigger program to be re-added below. } } // 重新挂载 if jumpMapFD == 0 { ... // 加载、编译及关联网卡 progID, err := ap.AttachProgram() ... // 执行bpftool prog show -id progID --json获取关联jumpMapFD jumpMapFD, err = FindJumpMap(progID, ap.IfaceName()) ... // 更新ifstate记录的jumpMapFD m.setJumpMapFD(ap, jumpMapFD) } return jumpMapFD, nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39m.AttachProgram负责从预编译.o文件加载TC挂载到网卡，编译的程序会加载到内核，挂载元数据会更新到用户侧的json文件用于卸载对比。func (ap *AttachPoint) AttachProgram() (int, error) { filename := ap.FileName(4) // 生成BPF目标文件路径/usr/lib/calico/bpf/xxx.o binaryToLoad := path.Join(bpf.ObjectDir, filename) // 执行tc filter show dev eth0 ingress获取旧挂载 progsToClean, err := ap.listAttachedPrograms() ... // TC挂载检查: 读取/var/run/calico/bpf/prog/{iface}_{hook}.json元数据, // 对比progID/object路径/.o文件SHA256 hash/config字符串,全部一致则跳过重挂载 progID, isAttached := ap.AlreadyAttached(binaryToLoad) if isAttached { return progID, nil } // BPF程序激活及加载到内核 obj, err := ap.loadObject(4, binaryToLoad) ... if ap.IPv6Enabled { filename := ap.FileName(6) obj, err := ap.loadObject(6, path.Join(bpf.ObjectDir, filename)) ... } // 执行C.bpf_tc_program_attach(libbpf)将BPF程序挂载到网卡clsact的ingress/egress hook progId, err := obj.AttachClassifier(SectionName(ap.Type, ap.ToOrFrom), ap.Iface, ap.Hook == bpf.HookIngress) ... // 执行tc filter del卸载旧挂载(避免重复挂载) ap.detachPrograms(progsToClean) ... // BPF挂载元数据写入用户侧/var/run/calico/bpf/prog/iface_hook.json bpf.RememberAttachedProg(ap, binaryToLoad, progId) ... return progId, nil } func (ap *AttachPoint) loadObject(ipVer int, file string) (*libbpf.Obj, error) { // 执行C.bpf_obj_open打开BPF程序,获取bpf_object obj, err := libbpf.OpenObject(file) ... // 遍历map for m, err := obj.FirstMap(); m != nil && err == nil; m, err = m.NextMap() { // 内部映射(libbpf .rodata常量段) if m.IsMapInternal() { // 设置.rodata全局配置(HostIP/IntfIP/TunnelMTU/NATin/NATout/VXLANPort/PSNAT...) ap.ConfigureProgram(m) ... continue } // 执行C.bpf_map_set_max_entries设置map映射大小 ap.setMapSize(m) ... // 计算BPF文件系统的Pin路径(/sys/fs/bpf/tc/[iface]_[igr|egr]/[map_name]) pinDir := bpf.MapPinDir(m.Type(), m.Name(), ap.Iface, ap.Hook) // 执行bpf_map_set_pin_path设置BPF文件系统映射 m.SetPinPath(path.Join(pinDir, m.Name())) ... } // 执行C.bpf_obj_load将目标BPF加载到内核(触发map创建+程序验证) obj.Load() ... // 将iface关联的BPF Prog索引写入jumpMap(通过bpf_map_update_elem设置tail call目标) // 主入口程序直接挂载在clsact上,Jump Map仅存放tail call子目标程序 ap.updateJumpMap(ipVer, obj) ... return obj, nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
注意
BPF TC Hook这里挂在有点复杂，目前仅看了一小部分BPF C侧程序，仅了解关联链路，后续可能还是得基于排查问题回顾
# 5.wep网卡
# 5.1.update
m.updateWEPsInDataplane处理所有dirty Pod网卡，挂载TC程序及应用网络策略，注意不会挂载XDP程序，XDP只作用于主机网卡防护。func (m *bpfEndpointManager) updateWEPsInDataplane() { ... m.dirtyIfaceNames.Iter(func(ifaceName string) error { // 仅处理wep(Pod veth) if !m.isWorkloadIface(ifaceName) { return nil } ... go func(ifaceName string) { ... // 应用网络策略(挂载TC+加载策略到Jump Map) m.applyPolicy(ifaceName) ... // wep网卡设置/proc/sys/net/ipv4/conf/%s/accept_local=1,接收本地流量 m.dp.setAcceptLocal(ifaceName, true) ... }(ifaceName) return nil }) ... if m.dirtyIfaceNames.Len() > 0 { // 触发一下mapCleanupRunner,清理jump maps m.mapCleanupRunner.Trigger() } for ifaceName, err := range errs { iface := m.nameToIface[ifaceName] wlID := iface.info.endpointID // 更新一下期望ifstate m.updateIfaceStateMap(ifaceName, &iface) // 挂载正常 if err == nil { // 更新happyWEPs,表示网卡就绪(后续渲染iptables ACCEPT链) if wlID != nil && m.allWEPs[*wlID] != nil { ... m.happyWEPs[*wlID] = m.allWEPs[*wlID] m.happyWEPsDirty = true } // 清理dirty m.dirtyIfaceNames.Discard(ifaceName) // 挂载异常 } else { // 清理happyWEPs if wlID != nil && m.happyWEPs[*wlID] != nil { ... delete(m.happyWEPs, *wlID) m.happyWEPsDirty = true } // wep网卡不存在,清理dirty if isLinkNotFoundError(err) { m.dirtyIfaceNames.Discard(ifaceName) } } } }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
注意
挂载成功的
Pod网卡会加入happyWEPs集合，表示可以接收流量，此时会渲染iptables ACCEPT chain
# 5.2.doApply
m.applyPolicy()会执行wildcard HEP策略合并和host endpoint策略抑制，挂载TC程序时方向反转(host ingress对应Pod egress)。func (m *bpfEndpointManager) applyPolicy(ifaceName string) error { ... // 执行挂载 m.doApplyPolicy(ifaceName, &isReady) ... // 更新iface m.withIface(ifaceName, func(iface *bpfInterface) (forceDirty bool) { iface.dpState.isReady = isReady return false // already dirty }) ... return err } func (m *bpfEndpointManager) doApplyPolicy(ifaceName string, isReady *bool) error { ... // 状态检查及dirty标记 m.withIface(ifaceName, func(iface *bpfInterface) (forceDirty bool) { ifaceUp = iface.info.ifaceIsUp() endpointID = iface.info.endpointID // wep网卡不是UP状态 if !ifaceUp { // jumpMapFD关掉(清理旧FD) for _, fd := range iface.dpState.jumpMapFDs { fd.Close() ... } iface.dpState.jumpMapFDs = nil } return false }) ... if !ifaceUp { // Interface is gone, nothing to do. return nil } // 确保clsact qdisc存在(先检查后创建) m.dp.ensureQdisc(ifaceName) ... // 获取wep端点信息,涉及网卡数据/关联策略 if endpointID != nil { wep = m.allWEPs[*endpointID] } ... go func() { ... // TC egress挂载(PolDirnIngress→HookEgress) m.attachWorkloadProgram(ifaceName, wep, PolDirnIngress) }() go func() { ... // TC ingress挂载(PolDirnEgress→HookIngress) m.attachWorkloadProgram(ifaceName, wep, PolDirnEgress) }() ... *isReady = true return nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
注意
wep网卡的策略方向和hep相反，这里按照host视角来看流量方向，Host Egress对应Pod Ingress
# 5.3.attach
m.attachWorkloadProgram()会反向挂载TC Ingress/Egress程序，Pod网卡设置了策略会合并主机通配策略生成Prog程序关联jumpMap。func (m *bpfEndpointManager) attachWorkloadProgram(ifaceName string, endpoint *proto.WorkloadEndpoint, polDirection PolDirection) error { if m.hostIP == nil { return fmt.Errorf("unknown host IP") } // 计算BPF编译挂载点 ap := m.calculateTCAttachPoint(polDirection, ifaceName) ap.HostIP = m.hostIP ap.TunnelMTU = uint16(m.vxlanMTU) ap.IntfIP = calicoRouterIP // WEP使用固定的169.254.1.1作为接口网关 ap.ExtToServiceConnmark = uint32(m.bpfExtToServiceConnmark) // 挂载tc程序 jumpMapFD, err := m.dp.ensureProgramAttached(ap) ... if endpoint != nil { profileIDs = endpoint.ProfileIds if len(endpoint.Tiers) != 0 { tier = endpoint.Tiers[0] } } // 渲染rules(WEP策略+合并wildcard HEP策略) rules := m.extractRules(tier, profileIDs, polDirection) // If host-* endpoint is configured, add in its policy. if m.wildcardExists { // wildcard HEP(如host-*)的策略反向合并 m.addHostPolicy(&rules, &m.wildcardHostEndpoint, polDirection.Inverse()) } // If workload egress and DefaultEndpointToHostAction is ACCEPT or DROP, suppress the normal // host-* endpoint policy. if polDirection == PolDirnEgress && m.epToHostAction != "RETURN" { rules.SuppressNormalHostPolicy = true } // If host -> workload, always suppress the normal host-* endpoint policy. if polDirection == PolDirnIngress { rules.SuppressNormalHostPolicy = true } // 基于rules生成policy prog指令及加载到内核,更新jumpMap return m.dp.updatePolicyProgram(jumpMapFD, rules, polDirection.RuleDir(), ap) }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
注意
这里的抑制主机通配策略指的是，
wep关联的策略优先，主机通配策略仅wep策略允许才生效
# 6.prog程序
# 6.1.extractTiers
xdp挂载生成Prog策略会用到m.extractTiers()，会根据wep定义的策略及关联的Policy/Profile渲染rules供Prog程序编译。//[ // { // "Name": "security", // "EndAction": "TierEndDeny", // "Policies": [ // { // "Name": "allow-frontend", // "Rules": [ // { // "Rule": { // "Action": "Allow", // "RuleId": "sec-fe-in-0", // "IpVersion": 4, // "Protocol": { "Number": 6 }, // "SrcNet": ["10.0.0.0/8"], // "DstPorts": [{ "First": 443, "Last": 443 }] // }, // "MatchID": "0x9c95410e2c265a5f" // }, // { // "Rule": { // "Action": "Deny", // "RuleId": "sec-fe-in-1", // "SrcNet": ["203.0.113.0/24"] // }, // "MatchID": "0x715af38b986e5c6a" // } // ] // } // ] // } //] func (m *bpfEndpointManager) extractTiers(tier *TierInfo, direction PolDirection, endTierDrop bool) []Tier { // 方向字符串Ingress/Egress,用于给MatchID加盐 dir := direction.RuleDir() if tier == nil { return } // 根据direction选出这个tier中需要展开的Policy名列表 directionalPols := tier.IngressPolicies if direction == PolDirnEgress { directionalPols = tier.EgressPolicies } if len(directionalPols) > 0 { // polprog.Tier,同一个Tier所有Policy顺序归在这一组里 // Tier末尾补EndAction(Deny/Pass) polTier := polprog.Tier{ Name: tier.Name, Policies: make([]polprog.Policy, len(directionalPols)), } for i, polName := range directionalPols { // 根据(tier, polName)取出Policy本体 pol := m.policies[proto.PolicyID{Tier: tier.Name, Name: polName}] ... // 根据direction选Policy内部真正需要的规则列表: // - 入向: InboundRules // - 出向: OutboundRules if direction == PolDirnIngress { prules = pol.InboundRules } else { prules = pol.OutboundRules } // polprog.Policy,对应N条Rule policy := polprog.Policy{ Name: polName, Rules: make([]polprog.Rule, len(prules)), } for ri, r := range prules { // Rule包一层: 保存匹配条件本身 // 生成一个稳定的 64-bit MatchID,用于写BPF RuleCountersMap的key policy.Rules[ri] = polprog.Rule{ Rule: r, MatchID: m.dp.ruleMatchID(dir, r.Action, "Policy", polName, ri), } } polTier.Policies[i] = policy } // 设置Tier末尾的EndAction,表明Policy都不匹配时怎么处理 // - TierEndDeny: 直接丢弃(Normal/FWD Tier) // - TierEndPass: 继续进入下一个Tier(Untracked/PreDNAT) if endTierDrop { polTier.EndAction = polprog.TierEndDeny } else { polTier.EndAction = polprog.TierEndPass } rTiers = append(rTiers, polTier) } return }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98注意
这里是基于
Policy生成的出入站策略，HEP/WEP网卡都可能用到，Profile有单独的渲染规则
# 6.2.extractRules
wep网卡挂载TC程序也涉及Prog策略生成，会多出Profile Rule渲染，类似于Policy渲染，基于方向选出Profile本体对象的出入站规则。func (m *bpfEndpointManager) extractRules(tier *TierInfo, profileNames []string, direction PolDirection) Rules { ... // When there is applicable normal policy that does not explicitly Allow or Deny traffic, // traffic is dropped. r.Tiers = m.extractTiers(tier, direction, EndTierDrop) r.Profiles = m.extractProfiles(profileNames, direction) return r } //[ // { // "Name": "ns.default", // "Rules": [ // { // "Rule": { // "Action": "Allow", // "RuleId": "ns-default-in-0", // "SrcNet": ["10.244.0.0/16"] // }, // "MatchID": "0xBA2497E387567C31" // }, // { // "Rule": { // "Action": "Allow", // "RuleId": "ns-default-in-1", // "Protocol": { "Number": 1 } // }, // "MatchID": "0x2FF071E88C9340B1" // }, // { // "Rule": { // "Action": "Allow", // "RuleId": "ns-default-in-2", // "SrcNet": ["10.0.0.0/24"], // "Protocol": { "Number": 6 }, // "DstPorts": [{ "First": 9090, "Last": 9090 }] // }, // "MatchID": "0xD904A09924253971" // } // ] // } //] func (m *bpfEndpointManager) extractProfiles(profileNames []string, direction PolDirection) []Profile { // 方向字符串Ingress/Egress dir := direction.RuleDir() if count := len(profileNames); count > 0 { ... for i, profName := range profileNames { // 取Profile对象本体 prof := m.profiles[proto.ProfileID{Name: profName}] ... // 方向选择: Ingress→InboundRules/Egress→OutboundRules if direction == PolDirnIngress { prules = prof.InboundRules } else { prules = prof.OutboundRules } profile := polprog.Profile{ Name: profName, Rules: make([]polprog.Rule, len(prules)), } for ri, r := range prules { profile.Rules[ri] = polprog.Rule{ Rule: r, MatchID: m.dp.ruleMatchID(dir, r.Action, "Profile", profName, ri), } } rProfiles[i] = profile } } return }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78policy/profile本质都是取inBound/outBound Rules渲染为Prog程序可识别格式
# 6.3.addHostPolicy
m.addHostPolicy()会将HostEndpoint关联的Policy/Profiles渲染为polprog.Rules，Prog程序基于这份配置编译限制出入站流量。func (m *bpfEndpointManager) addHostPolicy(rules *polprog.Rules, hostEndpoint *HostEndpoint, p PolDirection) { // 加载hep preNat策略(DNAT前执行,用于NodePort等特殊流量放行) if len(hostEndpoint.PreDnatTiers) == 1 { rules.HostPreDnatTiers = m.extractTiers(hostEndpoint.PreDnatTiers[0], p, NoEndTierDrop) } // 加载hep forward策略(转发阶段执行) if len(hostEndpoint.ForwardTiers) == 1 { rules.HostForwardTiers = m.extractTiers(hostEndpoint.ForwardTiers[0], p, EndTierDrop) } // 加载hep normal策略(常规策略) if len(hostEndpoint.Tiers) == 1 { rules.HostNormalTiers = m.extractTiers(hostEndpoint.Tiers[0], p, EndTierDrop) } // 加载hep profile策略 rules.HostProfiles = m.extractProfiles(hostEndpoint.ProfileIds, p) }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20注意
这个一般是
HEP网卡再用，渲染的比较全，涉及policy/profile两种格式
# 6.4.updateProg
m.updatePolicyProgram()基于渲染的rules编译Prog BPF字节码，将字节码加载到内核使策略程序生效，更新jumpMapFD引用实现关联。func (m *bpfEndpointManager) updatePolicyProgram(fd maps.FD, rules Rules, polDir string, ap attachPoint) error { ... // IPV4/IPV6处理 for _, ipFamily := range ipVersions { // p4i_iface/p4e_iface/p6i_iface/p6e_iface progName := policyProgramName(ap.IfaceName(), polDir, ipFamily) // 基于rules编译Prog BPF字节码程序 insns, err := m.doUpdatePolicyProgram(progName, fd, rules, ipFamily) ... } return nil } func (m *bpfEndpointManager) doUpdatePolicyProgram(pn string, fd FD, Rules, family IPVersion) (Insns, error){ ... // prog构建器 pg := polprog.NewBuilder(m.ipSetIDAlloc, m.bpfmaps.IpsetsMap.MapFD(),m.bpfmaps.StateMap.MapFD(),fd, opts...) if family == proto.IPVersion_IPV6 { pg.EnableIPv6Mode() } // 基于rules生成BPF指令(直接生成eBPF字节码,不需要clang编译) insns, err := pg.Instructions(rules) ... // 编译的BPF类型: SCHED_CLS(TC)或XDP progType := unix.BPF_PROG_TYPE_SCHED_CLS if rules.ForXDP { progType = unix.BPF_PROG_TYPE_XDP } // 系统调用,加载BPF程序至内核,返回progID progFD, err := bpf.LoadBPFProgramFromInsns(insns, pn, "Apache-2.0", uint32(progType)) ... defer func() { // 写入jumpMap引用可以确保prog程序存活,本地FD可关闭 progFD.Close() ... }() ... // 序列化progID binary.LittleEndian.PutUint32(v, uint32(progFD)) // 系统调用,更新jumpMap Policy slot指向新的策略程序,TC/XDP基于tail call尾调用跳转 maps.UpdateMapEntry(fd, jumpPolicyKey(family), v) ... return insns, nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
注意
这里编译
BPF指令和加载到内核基本都是系统调用C侧，就不深入看了
# 6.5.removeProg
m.removePolicyProgram()就简单多了，仅需要更新jumpMapFD实现引用关联解除，内核自己就会回收未使用的Prog BPF程序。func (m *bpfEndpointManager) removePolicyProgram(jumpMapFD maps.FD, ap attachPoint) error { ... // IPV4/IPV6 for _, ipFamily := range ipVersions { m.doRemovePolicyProgram(jumpMapFD, ipFamily) ... } return nil } func (m *bpfEndpointManager) doRemovePolicyProgram(jumpMapFD maps.FD, ipFamily proto.IPVersion) error { maps.DeleteMapEntryIfExists(jumpMapFD, jumpPolicyKey(ipFamily), 4) ... return nil } func DeleteMapEntryIfExists(mapFD FD, k []byte, valueSize int) error { DeleteMapEntry(mapFD, k, valueSize) ... return err } func DeleteMapEntry(mapFD FD, k []byte, valueSize int) error { ... C.bpf_maps_map_call(unix.BPF_MAP_DELETE_ELEM, C.uint(mapFD), unsafe.Pointer(&k[0]), unsafe.Pointer(nil), 0) ... return nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33注意
这里的
jumpMap更新的不是Pin目录的文件，还是基于系统调用通知内核解除关联，Pin文件自己会更新