xdpstate
南风未起 2026-08-26 19:39:22 cni
# 1.简介
# 1.1.xdp
xdp(express data path)会挂到网卡驱动前，不分配sk_buff、不进入协议栈及不查路由前提下丢弃垃圾包，避免无意义的解析浪费CPU。
注意
相比于传统的流量解析，
xdp很早就可以做出丢包决策，绕过sk_buff分配及L2/L3/L4头部解析
# 1.2.模式
xdp根据位置支持Native/Generic/Offload三种模式，由于挂载位置差异性能提升不同，Offload提升最大(CPU不参与)，瓶颈位于网卡硬件。--- Native XDP XDP加载到NIC的接收路径最前端,位于sk_buff分配前,直接操作原始帧,相比iptables快10～100倍,需网卡驱动支持 --- Generic XDP XDP作为内核协议栈的Hook挂载,位于sk_buff分配后,相比iptables快2~5倍,不支持XDP_TX/XDP_REDIRECT动作,对网卡无要求 --- Offload XDP XDP编译为固件指令加载到网卡硬件执行,不占用CPU资源,性能提升最高,需硬件支持Smart NIC1
2
3
4
5
6
7
8
注意
执行位置越早，内核程序跳过越多，CPU开销越少，吞吐越高
# 1.3.行为
xdp程序的返回值决定了包的命运，主要涉及丢弃、转发及放行策略，支持drop/pass/redirect/tx/aborted五种动作实现流量校验及处理。动作码 值 含义 典型场景 XDP_ABORTED0 出错，丢弃并记录 trace 程序异常、调试 XDP_DROP1 静默丢弃 DDoS 防护、黑名单 XDP_PASS2 交给内核协议栈 正常流量放行 XDP_TX3 从同一网卡发回去 SYN Cookie、反射 XDP_REDIRECT4 转发到另一个网卡/CPU 负载均衡、AF_XDP 限制
1.
xdp仅看得到单个包的内容，基于IP分片的TCP段无法重组及处理，会直接放行2.
xdp位于协议栈前，无法理解socket概念，无法利用已建立连接这种有状态防火墙规则及过滤3.
xdp_tx不会自动分片，利用bpf_xdp_adjust_head补充封装头导致包超出MTU，网卡会默认丢弃，必须检查包长度4.
eBPF验证器允许最大100万条指令，栈空间仅512字节，网卡仅支持挂一个xdp程序，功能组合必须利用tail call链式调用
# 2.实现
# 2.1.模式
calico有两套XDP实现，xdpstate是早期的xdp优化，作为iptables模式的预过滤器用于网卡驱动尽早丢弃恶意流量减轻压力。--- 非BPF模式(xdpstate,基于bpftool加载,用于deny黑名单) /sys/fs/bpf/calico/ ├── calico_failsafe_ports_v1 // 全局HASH └── xdp/ ├── prefilter_v1_<iface> // XDP程序prog pinned └── <iface>_IPv4_v1_blacklist // 接口LPM_TRIE黑名单 --- BPF模式(基于libbpf+tail call加载调用,Allow+Deny完整untracked policy) /sys/fs/bpf/tc/ ├── globals/ │ ├── cali_state (PERCPU_ARRAY) // per-CPU共享状态 │ ├── cali_counters (PERCPU_HASH) // per-iface计数器 │ ├── cali_v4_fsafes (LPM_TRIE) // failsafe CIDR │ └── cali_v4_ipsets (HASH) // IPSet成员 └── <iface>_xdp/ └── cali_jump3 (PROG_ARRAY) // Tail Call跳转表1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16注意
xdpstate作为早期实现并不完善，未提供策略程序tail call能力，Pin Map是单独管理的，未和TC共用
# 2.2.非BPF
xdp.c有单独的实现，基于bpftool+ip link set xdp pinned加载，单程序线性执行，没有tail call/状态/计数/metadata。// 提取TCP/UDP目的端口到dport{proto, port} static CALI_BPF_INLINE int extract_ports(__u32 len, struct iphdr *h, struct protoport *dport) { ... switch (h->protocol) { // TCP协议 case IPPROTO_TCP: // TCP包太短,绕过取端口 if (len < sizeof(struct ethhdr) + sizeof(*h) + sizeof(struct tcphdr)) return 1; // 取端口 dport->port = port_to_host(((struct tcphdr*)((void*)h + sizeof(*h)))->dest); break; // UDP协议 case IPPROTO_UDP: // 取端口,UDP包走到这里可以安全访问,之前已检查过长度 dport->port = port_to_host(((struct udphdr*)((void*)h + sizeof(*h)))->dest); break; default: return 0; } return 1; } // XDP入口,收到一个包执行一次,返回XDP_PASS或XDP_DROP __attribute__((section("prefilter_func"))) enum xdp_action prefilter(struct xdp_md* xdp) { ... // 长度不足以太头+IP头+UDP头(42B),格式异常DROP if (data + sizeof(*ehdr) + sizeof(*ihdr) + sizeof(struct udphdr) > data_end) return XDP_DROP; // 非IPv4(ARP/IPv6/VLAN等)直接PASS,不做拦截 if (be16_to_host(ETH_P_IP) != ehdr->h_proto) return XDP_PASS; // IP头起始位置 ihdr = data + sizeof(*ehdr); // 取TCP/UDP目的端口 if (extract_ports(xdp->data_end - xdp->data, ihdr, &dport)) // 查calico_failsafe_ports_v1白名单,命中PASS if (bpf_map_lookup_elem(&calico_failsafe_ports, &dport)) return XDP_PASS; // 查<iface>_IPv4_v1_blacklist黑名单,命中DROP ip4val_to_lpm(&sip, 32, ihdr->saddr); if (bpf_map_lookup_elem(&calico_prefilter_v4, &sip)) return XDP_DROP; // PASS,交给TC/网络栈处理 return XDP_PASS; }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
注意
这里可以看到，
xdpstate相关的xdp.c实现不会用到tail call跳转策略程序，仅支持黑白名单检查
# 2.3.BPF
xdp.c实现更规范，基于libbpf加载， 利用tail call拆分多个程序，和tc复用global map记录状态，支持策略程序跳转。// bpf_tail_call 动态JIT策略程序(Policy/Profile) // ├─ 匹配到DENY规则,bpf_tail_call DROP入口程序,执行cali_counters更新及DROP // ├─ 匹配到ALLOW规则,bpf_tail_call ACCEPT入口程序,执行metadata标记及PASS,TC层看到标记放行 // └─ 无匹配,return XDP_PASS,交给TC层继续处理 #define CALI_JUMP_TO(xdp, idx) ({ // 查cali_jump3 struct cali_jump *j = cali_jump_get(); \ if (j) // tail call跳转policy策略程序 bpf_tail_call(xdp, &j->cali_jump3, idx); /* 成功不返回，失败fallthrough */ }) // XDP主入口,收到一个入站包执行一次,返回XDP_PASS/XDP_DROP static CALI_BPF_INLINE int calico_xdp(struct xdp_md *xdp) { // ctx存在栈上 struct cali_tc_ctx ctx = { .state = state_get(), // 读取cali_state Map .counters = counters_get(xdp->ingress_ifindex), // 读取cali_counters Map .fwd = {.res = XDP_PASS}, // 默认放行 .ipheader_len = IP_SIZE, // IP头长度默认值(20字节) .xdp = xdp, // 保存XDP上下文指针 }; // state map异常,保守放行 if (!ctx.state) return XDP_PASS; // counter map异常,暂时DROP if (!ctx.counters) return XDP_DROP; // state map上一个包的解析结果会残留,清理一下 __builtin_memset(ctx.state, 0, sizeof(*ctx.state)); // counter map计数更新 counter_inc(&ctx, COUNTER_TOTAL_PACKETS); // 数据包检查,错包/畸形包丢弃 if (parse_packet_ip(&ctx) == PARSING_ERROR) goto deny; // IP五元组到cali_state tc_state_fill_from_iphdr(&ctx); // 1.TCP/UDP/ICMP解析,端口和类型写入cali_state // 2.VXLAN/IPIP查cali_v4_routes检查对端是不是calico节点,执行allow/deny if (tc_state_fill_from_nexthdr(&ctx) == PARSING_ERROR) goto deny; // 入站查cali_v4_fsafes白名单 if (is_failsafe_in(ctx.state->ip_proto, ctx.state->dport, ctx.state->ip_src)) goto allow; // 出站查cali_v4_fsafes白名单 if (is_failsafe_out(ctx.state->ip_proto, ctx.state->sport, ctx.state->ip_src)) goto allow; // 策略程序跳转 CALI_JUMP_TO(xdp, PROG_INDEX_POLICY); allow: return XDP_PASS; deny: return XDP_DROP; }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
61CALI_JUMP_TO会基于cali_jump关联FD跳转到JIT编译策略程序，策略程序匹配Allow/Deny会继续跳转xdp.c预编译程序。// 策略程序JIT编译策略后原子替换这个slot SEC("xdp/policy") int calico_xdp_norm_pol_tail(struct xdp_md *xdp) { return XDP_PASS; } // JIT策略判定Allow SEC("xdp/accept") int calico_xdp_accepted_entrypoint(struct xdp_md *xdp) { // 初始化栈上下文 struct cali_tc_ctx ctx = { .counters = counters_get(xdp->ingress_ifindex), // 读cali_counters .fwd = {.res = XDP_PASS}, // 默认结果放行 .ipheader_len = IP_SIZE }; // cali_counters map异常DROP if (!ctx.counters) return XDP_DROP; // 数据包的headroom元数据区域写CALI_META_ACCEPTED_BY_XDP标记 // TC层的BPF程序收到包后会检查这个标记,匹配会直接跳过Untracked策略+conntrack创建 xdp2tc_set_metadata(xdp, CALI_META_ACCEPTED_BY_XDP); // 更新cali_counters counter_inc(&ctx, CALI_REASON_ACCEPTED_BY_POLICY); // 放行交给内核协议栈→TC层 return XDP_PASS; } // JIT策略判定Deny SEC("xdp/drop") int calico_xdp_drop(struct xdp_md *xdp) { // 初始化栈上下文 struct cali_tc_ctx ctx = { .state = state_get(), // 读cali_state .counters = counters_get(xdp->ingress_ifindex), // 读cali_counters .ipheader_len = IP_SIZE }; // state/counter map异常DROP if (!ctx.state || !ctx.counters) return XDP_DROP; // 更新cali_counters counter_inc(&ctx, CALI_REASON_DROPPED_BY_POLICY); // 返回DROP return XDP_DROP; }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
注意
1.
policy/profile由eBPFEndpointMgr生成动态JIT策略程序，基于allow/deny cidr编译2.
policy/profile引用ipset还会查询cali_v4_ip_sets获取cidr，基于cidr执行编译3.编译的策略程序
FD基于bpf_map_update_elem原子替换cali_jump引用，CALI_JUMP_TO读取执行跳转4.策略条件命中
allow/deny cidr会执行tail call跳转到xdp.o预编译的ACCEPT/DROP程序处理
# 3.初始化
# 3.1.入口
dataplane初始化会根据配置启用或清理xdpstate，执行环境检查及回调注册，xdpstate有单独的xdp挂载及map存取实现。func NewIntDataplaneDriver(config Config) *InternalDataplane { ... // xdpstate启用 if config.XDPEnabled { // kernel ≧ 4.16,小端序CPU bpf.SupportsXDP() ... // BPF模式有单独的XDP实现 if !config.BPFEnabled { // 初始化xdpstate st, err := NewXDPState(config.XDPAllowGeneric) ... dp.xdpState = st // 注册回调,感知iface/hep变化 dp.xdpState.PopulateCallbacks(callbacks) // 注册到manager,上游触发onUpdate dp.RegisterManager(st) } } ... // 未启用xdpstate if !config.BPFEnabled && dp.xdpState == nil { // 初始化临时xdpstate xdpState, err := NewXDPState(config.XDPAllowGeneric) ... // 清理挂载及残留map,避免影响正常流量 xdpState.WipeXDP(); err != nil { } } // 注册资源回调 func (x *xdpState) PopulateCallbacks(cbs *common.Callbacks) { if x.ipV4State != nil { // 注册iface/hep回调 cbIDs := []*common.CbID{ cbs.AddInterfaceV4.Append(x.ipV4State.addInterface), cbs.RemoveInterfaceV4.Append(x.ipV4State.removeInterface), cbs.UpdateInterfaceV4.Append(x.ipV4State.updateInterface), cbs.UpdateHostEndpointV4.Append(x.ipV4State.updateHostEndpoint), cbs.RemoveHostEndpointV4.Append(x.ipV4State.removeHostEndpoint), } // 这里会记录,用于卸载阶段使用 x.ipV4State.cbIDs = append(x.ipV4State.cbIDs, cbIDs...) } }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
注意
BPF模式不会用xdpstate，这是比较早期的实现，BPF会用完整版的xdp执行流量检查
# 3.2.initial
NewXDPState()会实例化xdp state对象，维护IPV4状态缓存、内核状态、待同步差异及BPF操作队列，实现后续xdp状态同步。func NewXDPState(allowGenericXDP bool) (*xdpState, error) { // bpf加载,sockmap那里介绍过 lib, err := bpf.NewBPFLib("/usr/lib/calico/bpf/") ... // 初始化xdpstate return NewXDPStateWithBPFLibrary(lib, allowGenericXDP), nil } func NewXDPStateWithBPFLibrary(library bpf.BPFDataplane, allowGenericXDP bool) *xdpState { return &xdpState{ ipV4State: newXDPIPState(4), // 状态缓存队列 common: xdpStateCommon{ programTag: "", needResync: true, bpfLib: library, xdpModes: getXDPModes(allowGenericXDP), // Offload/Driver/Generic }, } } func newXDPIPState(ipFamily int) *xdpIPState { return &xdpIPState{ ipFamily: ipFamily, ipsetIDsToMembers: newIPSetIDsToMembers(), // ipsetID-->member currentState: newXDPSystemState(), // 已同步的内核状态 pendingDiffState: newXDPPendingDiffState(), // 回调收集的差异 bpfActions: newXDPBPFActions(), // 待执行的BPF动作 cbIDs: 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注意
xdpIPState是核心状态对象，记录内核当前状态、当前差异及待执行动作，后续的apply就是基于这里的状态执行更新
# 3.3.safeport
dataplane初始化iptables通用环境时，会幂等创建全局failsafe hash map，这个map设置白名单允许绕过xdp流量检测。func (d *InternalDataplane) setUpIptablesNormal() { ... if d.xdpState != nil { // calico_failsafe_ports_v1创建 if !d.setXDPFailsafePorts() { // 失败清理xdp数据 d.shutdownXDPCompletely() ... } } } // 初始化非BPF模式XDP的全局failsafe端口白名单 func (d *InternalDataplane) setXDPFailsafePorts() error { // bpftool map create... /sys/fs/bpf/calico/calico_failsafe_ports_v1 // 全局复用,key = { proto, dport },命中即放行 d.xdpState.common.bpfLib.NewFailsafeMap() // 入站failsafe端口(22/179/4789...） for _, p := range d.config.RulesConfig.FailsafeInboundHostPorts { // 转IP协议号(TCP=6, UDP=17) proto, _ := stringToProtocol(p.Protocol) // bpftool map update...写入{proto, port}到failsafe map d.xdpState.common.bpfLib.UpdateFailsafeMap(uint8(proto), p.Port) } 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注意
calico_failsafe_ports_v1记录的是{proto,port}而非addr，命中会直接放行流量
# 3.4.update
xdp.OnUpdate()会接收dataplane转发的msg，将ipset/policy变化记录到pendingDiffState，提取deny规则更新黑名单。func (d *InternalDataplane) loopUpdatingDataplane() { ... for { select { // calGraph计算的事件 case msg := <-d.toDataplane: d.onDatastoreMessage(msg)--->x.OnUpdate // 监听到的iface事件 case ifaceUpdate := <-d.ifaceUpdates: d.onIfaceMonitorMessage(ifaceUpdate)--->x.OnUpdate ... } ... } } func (x *xdpState) OnUpdate(protoBufMsg interface{}) { switch msg := protoBufMsg.(type) { // ipset member更新 case *proto.IPSetDeltaUpdate: // pendingReplaces及pendingAdds注册 x.ipV4State.addMembersIPSet(msg.Id, membersToSet(msg.AddedMembers)) // pendingReplaces清理,pendingDeletions注册 x.ipV4State.removeMembersIPSet(msg.Id, membersToSet(msg.RemovedMembers)) // ipset更新 case *proto.IPSetUpdate: // pendingReplaces更新,pendingAdds及pendingDeletions清理 x.ipV4State.replaceIPSet(msg.Id, membersToSet(msg.Members)) // ipset移除 case *proto.IPSetRemove: // pendingReplaces/pendingAdds/pendingDeletions清理 x.ipV4State.removeIPSet(msg.Id) // policy更新 case *proto.ActivePolicyUpdate: // 更新PoliciesToUpdate,清理PoliciesToRemove x.ipV4State.updatePolicy(*msg.Id, msg.Policy) // policy移除 case *proto.ActivePolicyRemove: // 清理PoliciesToUpdate,注册PoliciesToRemove x.ipV4State.removePolicy(*msg.Id) } } func (s *xdpIPState) updatePolicy(policyID proto.PolicyID, policy *proto.Policy) { // 由移除列表清理 s.pendingDiffState.PoliciesToRemove.Discard(policyID) // xdp支持接管的policy rule注册到PoliciesToUpdate // 仅处理第一个Untracked Tier的第一条Inbound Policy的第一条Inbound Rule // deny+未指定协议/端口/目的端+仅引用一个源ipset,这个ipset的成员IP会写到<iface>_IPv4_v1_blacklist if xdpRules, ok := xdpRulesFromProtoRules(policy.InboundRules, policy.OutboundRules); ok { s.pendingDiffState.PoliciesToUpdate[policyID] = &xdpRules } else { s.pendingDiffState.PoliciesToUpdate[policyID] = nil } } func (s *xdpIPState) removePolicy(policyID proto.PolicyID) { delete(s.pendingDiffState.PoliciesToUpdate, policyID) s.pendingDiffState.PoliciesToRemove.Add(policyID) }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
注意
xdp会接管可识别的Policy(Deny、IPV4...)，这是<iface>_IPv4_v1_blacklist黑名单IP数据来源
# 4.process
# 4.1.apply
dataplane.apply()会触发xdpState将状态缓存转换为BPF Action，以执行xdp程序加载、附着及相关的BPF Map更新。func (d *InternalDataplane) apply() { ... // xdpstate启用 if d.xdpState != nil { // 标记一下强制刷新 if d.forceXDPRefresh { // 下一轮全量对比,而非增量 d.xdpState.QueueResync() d.forceXDPRefresh = false } ... // pendingDiffState--->BPF Action d.xdpState.ProcessPendingDiffState(d.endpointsSourceV4) // xdp attach/detach和blacklist IP增删 d.applyXDPActions() ... // ipset member更新到<iface>_IPv4_v1_blacklist // 基于iface--policy--ipset匹配引用IP的网卡,更新计数 d.xdpState.ProcessMemberUpdates() // 重置pendingDiffState d.xdpState.DropPendingDiffState() // ipset member更新失败 if err != nil { // 尝试apply剩余动作 d.applyXDPActions() ... } // currentState=newState,用于下一次对比 d.xdpState.UpdateState() ... // apply失败 if applyXDPError != nil { // xdp卸载及map清理 d.shutdownXDPCompletely() ... } } ... } // attempts to disable XDP state. This could fail in cases where XDP isn't working properly. func (d *InternalDataplane) shutdownXDPCompletely() error { if d.xdpState == nil { return nil } // xdp callback清理 if d.callbacks != nil { d.xdpState.DepopulateCallbacks(d.callbacks) } ... for i := 0; i < 10; i++ { // xdp卸载及map清理 err = d.xdpState.WipeXDP() if err == nil { // 清理成功xdpState重置为nil d.xdpState = nil return nil } // 间隔100ms重试 time.Sleep(waitInterval) } return fmt.Errorf("Failed to wipe the XDP state", maxTries, waitInterval, err) }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
注意
xdp action失败会执行shutdownXDPCompletely清理，重试清理xdp数据，失败不会阻塞，而是留到下次loop触发
# 4.2.diffstate
x.ProcessPendingDiffState()会基于pendingDiffState生成new desired state及BPF Action，调整内存侧状态缓存。func (x *xdpState) ProcessPendingDiffState(epSourceV4 endpointsSource) { if x.ipV4State != nil { // 基于状态缓存计算差异 x.ipV4State.processPendingDiffState(epSourceV4) } } func (s *xdpIPState) processPendingDiffState(epSource endpointsSource) { ... // 同步的hep对象 rawHep := epSource.GetRawHostEndpoints() ... // 新接口,processHostEndpointChange for ifaceName, hepID := range pds.NewIfaceNameToHostEpID { s.processHostEndpointChange(ifaceName, &xdpIfaceData{}, hepID, rawHep[hepID], changeInMaps) processedIfaces.Add(ifaceName) } // 删除接口,uninstall+removeMap pds.IfaceNamesToDrop.Iter(func(ifName string) error { if data, ok := cs.IfaceNameToData[ifName]; ok && data.NeedsXDP() { ba.UninstallXDP.Add(ifName); ba.RemoveMap.Add(ifName) } delete(newCs.IfaceNameToData, ifName) processedIfaces.Add(ifName) return nil }) // iface HEP变化,processHostEndpointChange for ifaceName, newEpID := range pds.IfaceEpIDChange { data := cs.IfaceNameToData[ifaceName] s.processHostEndpointChange(ifaceName, &data, newEpID, rawHep[newEpID], changeInMaps) processedIfaces.Add(ifaceName) } // HEP更新,processHostEndpointChange pds.UpdatedHostEndpoints.Iter(func(hepID proto.HostEndpointID) error { for ifaceName, data := range cs.IfaceNameToData { if processedIfaces.Contains(ifaceName) || data.EpID != hepID { continue } s.processHostEndpointChange(ifaceName, &data, hepID, rawHep[hepID], changeInMaps) processedIfaces.Add(ifaceName) } return nil }) // Policy删除 pds.PoliciesToRemove.Iter(func(policyID proto.PolicyID) error { delete(newCs.XDPEligiblePolicies, policyID); return nil }) // Policy更新(逐接口计算refDelta，跳过已处理iface) for policyID, rules := range pds.PoliciesToUpdate { for ifaceName, data := range cs.IfaceNameToData { if processedIfaces.Contains(ifaceName) { continue } // 验证policyID属于该HEP及更新计数 oldSetIDs := data.PoliciesToSetIDs[policyID] if oldSetIDs != nil { oldSetIDs.Iter(func(setID string) error { m[setID] -= 1; return nil }) } if rules != nil { // 基于支持的rules获取关联的ipsetID newSetIDs := getSetIDs(rules) newSetIDs.Iter(func(setID string) error { m[setID] += 1; return nil }) newCs.IfaceNameToData[ifaceName].PoliciesToSetIDs[policyID] = newSetIDs } else { delete(newCs.IfaceNameToData[ifaceName].PoliciesToSetIDs, policyID) } } // 更新支持的xdp policy rules if rules != nil { newCs.XDPEligiblePolicies[policyID] = *rules } else { delete(newCs.XDPEligiblePolicies, policyID) } } // Install/Uninstall切换 ifacesWithUpdatedPolicies.Iter(func(ifaceName string) error { // xdp policy deny rules不为空才挂载xdp,否则没有意义 oldNeedsXDP := cs.IfaceNameToData[ifaceName].NeedsXDP() newNeedsXDP := newCs.IfaceNameToData[ifaceName].NeedsXDP() if oldNeedsXDP && !newNeedsXDP { ba.UninstallXDP.Add(ifaceName); ba.RemoveMap.Add(ifaceName) } if !oldNeedsXDP && newNeedsXDP { ba.InstallXDP.Add(ifaceName); ba.CreateMap.Add(ifaceName) } return nil }) // changeInMaps,也就是xdp deny rules发生变化,需要更新IP for ifaceName, ips := range changeInMaps { // iface未被删 if !ba.RemoveMap.Contains(ifaceName) { // 更新iface-ipsetID引用计数,后续执行action会展开ipset关IP,写入IP:count for setID, refCount := range ips { switch { case refCount > 0: ba.AddToMap[ifaceName][setID] = uint32(refCount) case refCount < 0: ba.RemoveFromMap[ifaceName][setID] = uint32(-refCount) } } } } }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
99
100
101
102
注意
processPendingDiffState会将pendingDiffState渲染为BPF Action及xdpState内存预期状态
# 4.3.change
x.processHostEndpointChange()根据Policy计算iface引用的ipset，更新xdp状态、BPF Map及相关的BPF Action。func (s *xdpIPState) processHostEndpointChange(...) { ... // 统计old ipsetID for _, setIDs := range oldData.PoliciesToSetIDs { setIDs.Iter(func(setID string) error { oldSetIDs[setID] += 1 return nil }) } newPolicyIDs := getPolicyIDs(newEP) ... // 扫描newPolicyID for _, policyID := range newPolicyIDs { // 获取xdp policy rules rules := s.getLatestRulesForPolicyID(policyID) if rules == nil { continue } // rule关联的ipsetIDs rulesSetIDs := getSetIDs(rules) // 记录policyID-->ruleSetIDs policiesToSetIDs[policyID] = rulesSetIDs // 更新new ipsetID引用 rulesSetIDs.Iter(func(setID string) error { newSetIDs[setID] += 1 return nil }) } // 初始化iface state newData := xdpIfaceData{ EpID: newHepID, PoliciesToSetIDs: policiesToSetIDs, } // 更新IfaceNameToData s.newCurrentState.IfaceNameToData[ifaceName] = newData ... // xdp policy rules存在才需要挂载 oldNeedsXDP := oldData.NeedsXDP() newNeedsXDP := newData.NeedsXDP() // iface没有xdp rules,生成uninstall action if oldNeedsXDP && !newNeedsXDP { // iface uninstallXDP s.bpfActions.UninstallXDP.Add(ifaceName) // iface remove BPFMap s.bpfActions.RemoveMap.Add(ifaceName) // 否则生成install action } else if !oldNeedsXDP && newNeedsXDP { // iface installXDP s.bpfActions.InstallXDP.Add(ifaceName) // iface create BPFMap s.bpfActions.CreateMap.Add(ifaceName) } // 计算refDelta = new - old m, ok := changeInMaps[ifaceName] if !ok { m = make(map[string]int) changeInMaps[ifaceName] = m } for setID, refCount := range newSetIDs { m[setID] += refCount } for setID, refCount := range oldSetIDs { m[setID] -= refCount } } func (s *xdpIPState) getLatestRulesForPolicyID(policyID proto.PolicyID) *xdpRules { // policyID-->policy rules // 优先由PoliciesToUpdate匹配,其次由XDPEligiblePolicies计算结果匹配 rules, ok := s.pendingDiffState.PoliciesToUpdate[policyID] if ok { return rules } else { xdpRules, ok := s.newCurrentState.XDPEligiblePolicies[policyID] if ok { return &xdpRules } else { 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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
注意
pendingDiffState暂存的policy rules会解析为setIDs更新changeInMaps，xdp状态切换会更新BPF Action
# 4.4.member
x.ProcessMemberUpdates()基于last cache/pendingState计算memberAdd/memberDrop，更新<iface>_IPv4_v1_blacklist数据。func (x *xdpState) ProcessMemberUpdates() error { if x.ipV4State != nil { memberCacheV4 := newXDPMemberCache(x.ipV4State.getBpfIPFamily(), x.common.bpfLib) // member更新 err := x.ipV4State.processMemberUpdates(memberCacheV4) ... } return nil } func (s *xdpIPState) processMemberUpdates(memberCache *xdpMemberCache) error { // merge member变化{ setID → (toAdd, toDrop) } // pendingReplace vs cache --> { setID → (toAdd, toDrop) } // pendingAdds --> { setID → toAdd } // pendingDrops --> { setID → toDrop } changes := s.getMemberChanges() for setID, change := range changes { // ipset影响的iface ifacesToRefCounts := s.getAffectedIfaces(setID) for iface, refCount := range ifacesToRefCounts { // iface drop ipset member miDelete := &memberIterSet{ members: change.toDrop, refCount: refCount } // 执行bpftool map delete...清理<iface>_IPv4_v1_blacklist member processMemberDeletions(memberCache, iface, miDelete) ... // iface add ipset member miAdd := &memberIterSet{ members: change.toAdd, refCount: refCount } // 执行bpftool map update...更新<iface>_IPv4_v1_blacklist member processMemberAdds(memberCache, iface, miAdd) ... } } // 基于pending更新cache,清空pending s.ipsetIDsToMembers.UpdateCache() 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
注意
cache记录lastState，member.pending会对比cache找到ipset add/drp member，refCount是受影响的iface数量
# 5.handle
# 5.1.apply
d.applyXDPActions()会重新同步ipsets，修复及对齐BPF Map和XDP状态的一致性，基于内存xdp action状态执行相应内核策略。func (d *InternalDataplane) applyXDPActions() error { ... for i := 0; i < 10; i++ { // 基于标记重新同步xdp状态 d.xdpState.ResyncIfNeeded(d.ipsetsSourceV4) ... // 执行BPF同步 d.xdpState.ApplyBPFActions(d.ipsetsSourceV4) ... } return err } func (x *xdpState) ResyncIfNeeded(ipsSourceV4 ipsetsSource) error { ... // 不需要重新同步 if !x.common.needResync { return nil } ... for i := 0; i < 10; i++ { // 重新同步 x.tryResync(newConvertingIPSetsSource(ipsSourceV4)) ... } ... x.common.needResync = false 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
注意
xdp tag获取会attach xdp到临时网卡，读取完成会清理temp iface及detach xdp
# 5.2.resync
s.tryResync()会扫描内核存在的xdp prog及BPF Map，检查Tag/Mode及Pin Map正确性，更新内存ipset member用于后续下发。func (x *xdpState) tryResync(ipsSourceV4 ipsetsSource) error { if x.common.programTag == "" { // 临时建一对veth pair,加载及附着xdp程序,基于bpftool prog show...获取Tag,用完会清理veth pair tag := x.common.bpfLib.GetXDPObjTagAuto() ... x.common.programTag = tag } // 执行xdp状态同步 if x.ipV4State != nil { x.ipV4State.tryResync(&x.common, ipsSourceV4) ... } return nil } func (s *xdpIPState) tryResync(common *xdpStateCommon, ipsSource ipsetsSource) error { ... // ipsetID-->member的pending/cache重制 s.ipsetIDsToMembers.Clear() // 内核扫描 resyncState, err := s.newXDPResyncState(common.bpfLib, ipsSource, common.programTag, common.xdpModes) ... // 修复prog/map挂载一致性 s.fixupXDPProgramAndMapConsistency(resyncState) // 修复blacklist member s.fixupBlocklistContents(resyncState) return nil } func (s *xdpIPState) newXDPResyncState(bpfLib bpf.BPFDataplane, ...) (*xdpResyncState, error) { // 枚举挂XDP程序的接口 xdpIfaces, _ := bpfLib.GetXDPIfaces() for _, iface := range xdpIfaces { // bpftool prog show...获取xdp tag tag, tagErr := bpfLib.GetXDPTag(iface) // ip link show...获取xdp mode mode, modeErr := bpfLib.GetXDPMode(iface) // tag/mode检查 ifacesWithProgs[iface] = progInfo{bogus: len(bogosityReasons) > 0} } // 列出/sys/fs/bpf/calico/xdp/的*_IPv4_v1_blacklist ifacesWithPinnedMaps, _ := bpfLib.ListCIDRMaps(s.getBpfIPFamily()) for _, iface := range ifacesWithPinnedMaps { // <iface>_IPv4_v1_blacklist格式检查 mapOk, _ := bpfLib.IsValidMap(iface, s.getBpfIPFamily()) ... // bpftool map show...获取<iface>_IPv4_v1_blacklist FD mapID, _ := bpfLib.GetCIDRMapID(iface, s.getBpfIPFamily()) // bpftool prog show...获取Pin mapFD mapIDs, _ := bpfLib.GetMapsFromXDP(iface) // 检查一下prog Pin mapID和实际的mapID是否匹配 mapMismatch := !contains(mapIDs, mapID) // 读出<iface>_IPv4_v1_blacklist记录的{key→refcount} var mapContents map[bpf.CIDRMapKey]uint32 ... // 记录一下 ifacesWithMaps[iface] = mapInfo{bogus: mapBogus, mismatched: mapMismatch, contents: mapContents} } ... // ipset关联member for _, data := range s.newCurrentState.IfaceNameToData { for _, setIDs := range data.PoliciesToSetIDs { setIDs.Iter(func(setID string) error { if visited.Contains(setID) { return nil } members, _ := s.getIPSetMembers(setID, ipsSource) ipsetMembers[setID] = members visited.Add(setID); return nil }) } } return &xdpResyncState{ifacesWithProgs, ifacesWithMaps, ipsetMembers}, 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
78
79
注意
这里主要检查
iface已绑定xdp的tag/mode和关联Pin Map的正确性，加载blacklist member和期望member内容到内存
# 5.3.fixup
fixup基于内核和期望网卡状态决定install/uninstall xdp及create/remove map，若blacklist member更新还会生成增量action。// xdp+blacklist修复 func (s *xdpIPState) fixupXDPProgramAndMapConsistency(resyncState *xdpResyncState) { // 收集内存期望和内核附着xdp的所有网卡 ifaces := s.getIfaces(resyncState, giNS|giWX|giIX|giUX|giWM|giCM|giRM) ifaces.Iter(func(iface string) error { ... // iface实际未挂载xdp,也没有期望挂载 if !hasXDP && !shouldHaveXDP { // remove blacklist action if mapExists { s.bpfActions.RemoveMap.Add(iface) } return } // iface实际未挂载xdp,期望挂载 if !hasXDP && shouldHaveXDP { // install xdp action s.bpfActions.InstallXDP.Add(iface) // bogus blacklist remove action/new map create action if !mapExists { s.bpfActions.CreateMap.Add(iface) } else if mapBogus { s.bpfActions.RemoveMap.Add(iface); s.bpfActions.CreateMap.Add(iface) } return } // iface挂载xdp,期望不挂载 if hasXDP && !shouldHaveXDP { // uninstall xdp action s.bpfActions.UninstallXDP.Add(iface) // old blacklist remove action if mapExists { s.bpfActions.RemoveMap.Add(iface) } return } // blacklist正常,始终期望挂xdp if hasXDP && !hasBogusXDP && shouldHaveXDP { // blacklist不存在 if !mapExists { // uninstall--->install action s.bpfActions.UninstallXDP.Add(iface) s.bpfActions.InstallXDP.Add(iface) // blacklist create action s.bpfActions.CreateMap.Add(iface) // blacklist异常 } else if mapBogus { // uninstall--->install action s.bpfActions.UninstallXDP.Add(iface) s.bpfActions.InstallXDP.Add(iface) // blacklist remove--->create action s.bpfActions.RemoveMap.Add(iface) s.bpfActions.CreateMap.Add(iface) // blacklist FD和xdp关联不匹配 } else if mapMismatch { // uninstall--->install action // 复用blacklist,基于blacklist重装 s.bpfActions.UninstallXDP.Add(iface) s.bpfActions.InstallXDP.Add(iface); } // else: Good program + correct map → 无动作 return } // blacklist异常,始终期望挂xdp if hasXDP && hasBogusXDP && shouldHaveXDP { // uninstall--->install action s.bpfActions.UninstallXDP.Add(iface) s.bpfActions.InstallXDP.Add(iface); // blacklist不存在,create action if !mapExists { s.bpfActions.CreateMap.Add(iface) } // blacklist异常,remove--->create action else if mapBogus { s.bpfActions.RemoveMap.Add(iface); s.bpfActions.CreateMap.Add(iface) } return } return nil }) } // blacklist member修复 func (s *xdpIPState) fixupBlocklistContents(resyncState *xdpResyncState) { // 收集内存期望和内核附着xdp的所有网卡 ifaces := s.getIfaces(resyncState, giNS) ifaces.Iter(func(iface string) error { // blacklist期望新建 if s.bpfActions.CreateMap.Contains(iface) { // 设置iface→setID→refCount s.fixupBlocklistContentsFreshMap(iface) // blacklist删除或更新 } else { // 内核 vs 期望,blacklist正常的话,更新一下iface→setID→refCount // 正常的blacklist member基于MemberToAdd/MemberToDrop更新,不需要AddToMap/RemoveFromMap全量重建 s.fixupBlocklistContentsExistingMap(resyncState, iface) } return nil }) // AddToMap/RemoveFromMap条目修正 for _, m := range []map[string]map[string]uint32{s.bpfActions.AddToMap, s.bpfActions.RemoveFromMap} { for iface := range m { // iface不需要挂载xdp if !ifaces.Contains(iface) { // blacklist member增删不用执行 delete(m, iface) } } } }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
99
100
101
102
103
104
105
106
注意
xdp要不要安装，基于iface policy/profile有没有满足条件的deny规则，xdp是用来流量检查的，无规则网卡附着没有意义
# 5.4.action
x.ApplyBPFActions()会将计算的xdp action应用到Linux Kernel，维护xdp program/xdp map的增删及map member更新。func (x *xdpState) ApplyBPFActions(ipsSource ipsetsSource) error { memberCacheV4 := newXDPMemberCache(x.ipV4State.getBpfIPFamily(), x.common.bpfLib) // 执行xdp action err := x.ipV4State.bpfActions.apply(memberCacheV4, x.ipV4State.ipsetIDsToMembers, newConvertingIPSetsSource(ipsSource), x.common.xdpModes) x.ipV4State.bpfActions = newXDPBPFActions() // 执行后清空 ... return nil } func (a *xdpBPFActions) apply(memberCache *xdpMemberCache, ...) error { // 卸载时遍历所有模式 allXDPModes := getXDPModes(true) // 卸载XDP a.UninstallXDP.Iter(func(iface string) error { // ip link show dev <iface> <mode> off // prog文件删除 for _, mode := range allXDPModes { memberCache.bpfLib.RemoveXDP(iface, mode) } return nil }) // blacklist Pin文件清理 a.RemoveMap.Iter(func(iface string) error { memberCache.bpfLib.RemoveCIDRMap(iface, memberCache.GetFamily()); return nil }) // blacklist Pin Map创建 a.CreateMap.Iter(func(iface string) error { // bpftool map create memberCache.bpfLib.NewCIDRMap(iface, memberCache.GetFamily()); return nil }) // blacklist member补充 for iface, memberMap := range a.MembersToAdd { // bpftool map update processMemberAdds(...) } // blacklist member批量补充 for iface, setIDMap := range a.AddToMap { // bpftool map update for setID, refCount := range setIDMap { members, _ := getIPSetMembers(ipsetIDsToMembers, setID, ipsSource) processMemberAdds(memberCache, iface, &memberIterSet{members, refCount}) } } // blacklist member清理 for iface, memberMap := range a.MembersToDrop { // bpftool map delete processMemberDeletions(...) } // blacklist member批量清理 for iface, setIDMap := range a.RemoveFromMap { // bpftool map delete for setID, refCount := range setIDMap { members, _ := ipsetIDsToMembers.GetCached(setID) processMemberDeletions(memberCache, iface, &memberIterSet{members, refCount}) } } // xdp挂载 a.InstallXDP.Iter(func(iface string) error { // bpftool prog load <objPath> <progPath> type xdp pin <blacklist> // ip link set dev <ifName> <mode> pinned <progPath> for _, mode := range xdpModes { if err := memberCache.bpfLib.LoadXDPAuto(iface, mode); err == nil { break } } return nil }) 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
73
74
75
注意
1.
memberToAdd/memberToDrop基于bpfMap/lastState对比计算，记录的是member refCount变化2.
addToMap/removeFromMap基于ipset refCount对比计算，记录的是ipset allMember refCount变化
# 5.5.wide
NewIntDataplaneDriver()检查XDP状态，不支持会执行xdpState.WipeXDP()同步及清理xdp残留的状态，避免旧的XDP程序影响流量。func (x *xdpState) WipeXDP() error { savedIPV4State := x.ipV4State // 期望状态先重置 x.ipV4State = newXDPIPState(4) x.ipV4State.newCurrentState = newXDPSystemState() defer func() { x.ipV4State = savedIPV4State }() ipsSource := &nilIPSetsSource{} // 再扫描一次内核,由于期望重置,会生成uninstall+remove action x.tryResync(ipsSource) // 执行清理 x.ApplyBPFActions(ipsSource) x.QueueResync() return nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17注意
widexdp会临时重置dsw，基于dsw状态生成uninstall/removeMap action清理内核xdp数据