endpointMgr
南风未起 2026-03-23 19:39:22 cni
# 1.简介
# 1.1.定义
endpointManager会基于wep/hep生成iptables chain规则，利用callback向xdpState/sockmapState/eBPFEndpointMgr上报数据。// endpointManager manages the dataplane resources that belong to each endpoint as well as // the "dispatch chains" that fan out packets to the right per-endpoint chain. type endpointManager struct { // Config. ... wlIfacesRegexp *regexp.Regexp // calico podIface正则 kubeIPVSSupportEnabled bool // ipvs支持标记 floatingIPsEnabled bool // floatingIP支持标记 // Our dependencies. rawTable iptablesTable // raw table mangleTable iptablesTable // mangle table filterTable iptablesTable // filter table ruleRenderer rules.RuleRenderer // 生成iptables rule routeTable routetable.RouteTableInterface // L2/L3 route writeProcSys procSysWriter // proc/sys修改 osStat func(path string) (os.FileInfo, error) // os资源检测 epMarkMapper rules.EndpointMarkMapper // endpoint流量标记 // Pending updates, cleared in CompleteDeferredWork as the data is copied to the activeXYZ fields. pendingWlEpUpdates map[proto.WorkloadEndpointID]*proto.WorkloadEndpoint // workload endpoint变化 pendingIfaceUpdates map[string]ifacemonitor.State // iface状态变化 // Active state, updated in CompleteDeferredWork. activeWlEndpoints map[proto.WorkloadEndpointID]*proto.WorkloadEndpoint // 活跃的workload endoint activeWlIfaceNameToID map[string]proto.WorkloadEndpointID // iface-->workload activeUpIfaces set.Set[string] // up状态iface activeWlIDToChains map[proto.WorkloadEndpointID][]*iptables.Chain // workload-->iptables chain activeWlDispatchChains map[string]*iptables.Chain // workload-->dispatch chain activeEPMarkDispatchChains map[string]*iptables.Chain // mark-->dispatch chain // Workload endpoint that would be locally active but are 'shadowed' by other endpoints with same ifaceName. shadowedWlEndpoints map[proto.WorkloadEndpointID]*proto.WorkloadEndpoint // iface冲突的workload // contains names of workload interfaces that need to have their configuration (sysctls etc.) refreshed. wlIfaceNamesToReconfigure set.Set[string] // 需配置sysctl的iface ... // maps interface names to lists of source IPs that we accept from these interfaces sourceSpoofingConfig map[string][]string // iface允许伪装的IP // rpfSkipChainDirty is set to true when the rpf status of some endpoints is updated rpfSkipChainDirty bool // RPF检查白名单 // default configuration for new interfaces,used to reset kernel settings when source spoofing is disabled defaultRPFilter string // 激活RPFilter标记 // maps host interface name to the set of IPs on that interface (reported from the dataplane). hostIfaceToAddrs map[string]set.Set[string] // hostIface IP // rawHostEndpoints contains the raw (i.e. not resolved to interface) host endpoints. rawHostEndpoints map[proto.HostEndpointID]*proto.HostEndpoint // 未解析的hostendpoint // hostEndpointsDirty is set to true when host endpoints are updated. hostEndpointsDirty bool // 标记hostendpoint变化 // activeHostIfaceToChains maps host interface name to the chains that we've programmed. activeHostIfaceToRawChains map[string][]*iptables.Chain activeHostIfaceToFiltChains map[string][]*iptables.Chain activeHostIfaceToMangleIngressChains map[string][]*iptables.Chain activeHostIfaceToMangleEgressChains map[string][]*iptables.Chain // Dispatch chains that we've programmed for host endpoints. activeHostRawDispatchChains map[string]*iptables.Chain activeHostFilterDispatchChains map[string]*iptables.Chain activeHostMangleDispatchChains map[string]*iptables.Chain // activeHostEpIDToIfaceNames records which interfaces we resolved each host endpoint to. activeHostEpIDToIfaceNames map[proto.HostEndpointID][]string // activeIfaceNameToHostEpID records which endpoint we resolved each host interface to. activeIfaceNameToHostEpID map[string]proto.HostEndpointID // hostIface-->hostendpoint newIfaceNameToHostEpID map[string]proto.HostEndpointID // 最新的hostIface-->hostendpoint needToCheckDispatchChains bool // dispatch chain过期检查 needToCheckEndpointMarkChains bool // epMark chain过期检查 ... // Callbacks callbacks endpointManagerCallbacks // xdpState/sockmapState回调 bpfEnabled bool // eBPF dataplane标记 bpfEndpointManager hepListener // eBPF endpointMgr }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注意
callbacks主要由xdpState/sockmapState注册回调，bpfEndpointManager是eBPF Dataplane注册的回调
# 1.2.初始化
newEndpointManager会实例化epMgr及内存状态缓存，触发调用会调整ep chain及交互routeTable调整L2/L3路由，实现ep流量管理。func newEndpointManager(rawTable, mangleTable, filterTable iptablesTable, ruleRenderer rules.RuleRenderer, routeTable routetable.RouteTableInterface, ipVersion uint8, epMarkMapper rules.EndpointMarkMapper, kubeIPVSSupportEnabled bool, wlInterfacePrefixes []string, ..., defaultRPFilter string, bpfEnabled bool, bpfEndpointManager hepListener, callbacks *common.Callbacks, floatingIPsEnabled bool, ) *endpointManager { return newEndpointManagerWithShims( rawTable, mangleTable, filterTable, ruleRenderer, routeTable, ipVersion, epMarkMapper, kubeIPVSSupportEnabled, wlInterfacePrefixes, onWorkloadEndpointStatusUpdate, writeProcSys, os.Stat, defaultRPFilter, bpfEnabled, bpfEndpointManager, callbacks, floatingIPsEnabled, ) } func newEndpointManagerWithShims(rawTable, mangleTable, filterTable, ruleRenderer rules.RuleRenderer, routeTable routetable.RouteTableInterface, ipVersion uint8, epMarkMapper rules.EndpointMarkMapper, kubeIPVSSupportEnabled bool, wlInterfacePrefixes []string, ..., procSysWriter procSysWriter, osStat func(name string) (os.FileInfo, error), defaultRPFilter string, bpfEnabled bool, bpfEndpointManager hepListener, callbacks *common.Callbacks, floatingIPsEnabled bool) *endpointManager { // workload iface正则匹配 wlIfacesPattern := "^(" + strings.Join(wlInterfacePrefixes, "|") + ").*" wlIfacesRegexp := regexp.MustCompile(wlIfacesPattern) return &endpointManager{ ... wlIfacesRegexp: wlIfacesRegexp, kubeIPVSSupportEnabled: kubeIPVSSupportEnabled, bpfEnabled: bpfEnabled, bpfEndpointManager: bpfEndpointManager, floatingIPsEnabled: floatingIPsEnabled, rawTable: rawTable, mangleTable: mangleTable, filterTable: filterTable, ruleRenderer: ruleRenderer, routeTable: routeTable, writeProcSys: procSysWriter, osStat: osStat, epMarkMapper: epMarkMapper, // Pending updates, we store these up as OnUpdate is called, then process them // in CompleteDeferredWork and transfer the important data to the activeXYX fields. pendingWlEpUpdates: map[proto.WorkloadEndpointID]*proto.WorkloadEndpoint{}, pendingIfaceUpdates: map[string]ifacemonitor.State{}, activeUpIfaces: set.New[string](), activeWlEndpoints: map[proto.WorkloadEndpointID]*proto.WorkloadEndpoint{}, activeWlIfaceNameToID: map[string]proto.WorkloadEndpointID{}, activeWlIDToChains: map[proto.WorkloadEndpointID][]*iptables.Chain{}, shadowedWlEndpoints: map[proto.WorkloadEndpointID]*proto.WorkloadEndpoint{}, wlIfaceNamesToReconfigure: set.New[string](), epIDsToUpdateStatus: set.NewBoxed[any](), sourceSpoofingConfig: map[string][]string{}, rpfSkipChainDirty: true, defaultRPFilter: defaultRPFilter, hostIfaceToAddrs: map[string]set.Set[string]{}, rawHostEndpoints: map[proto.HostEndpointID]*proto.HostEndpoint{}, hostEndpointsDirty: true, activeHostIfaceToRawChains: map[string][]*iptables.Chain{}, activeHostIfaceToFiltChains: map[string][]*iptables.Chain{}, activeHostIfaceToMangleIngressChains: map[string][]*iptables.Chain{}, activeHostIfaceToMangleEgressChains: map[string][]*iptables.Chain{}, // Caches of the current dispatch chains indexed by chain name. We use these to // calculate deltas when we need to update the chains. activeWlDispatchChains: map[string]*iptables.Chain{}, activeHostFilterDispatchChains: map[string]*iptables.Chain{}, activeHostMangleDispatchChains: map[string]*iptables.Chain{}, activeHostRawDispatchChains: map[string]*iptables.Chain{}, activeEPMarkDispatchChains: map[string]*iptables.Chain{}, needToCheckDispatchChains: true, // Need to do start-of-day update. needToCheckEndpointMarkChains: true, // Need to do start-of-day update. OnEndpointStatusUpdate: onWorkloadEndpointStatusUpdate, callbacks: newEndpointManagerCallbacks(callbacks, ipVersion), } }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注意
loopUpdatingDataplane-->processMsgFromCalcGraph-->mgr.OnUpdate<--processIfaceUpdate<--ifaceMonitor
# 1.3.onUpdate
m.OnUpdate()会将endpoint/iface事件统一接入，基于类型更新本地缓存，利用pending queue+dirty机制延迟触发同步和计算。func (m *endpointManager) OnUpdate(protoBufMsg interface{}) { ... switch msg := protoBufMsg.(type) { // workload endpoint update/remove,更新待处理缓存 case *proto.WorkloadEndpointUpdate: m.pendingWlEpUpdates[*msg.Id] = msg.Endpoint case *proto.WorkloadEndpointRemove: m.pendingWlEpUpdates[*msg.Id] = nil // host endpoint update case *proto.HostEndpointUpdate: // 上报给xdpState/sockmapState m.callbacks.InvokeUpdateHostEndpoint(*msg.Id) // 更新raw cache m.rawHostEndpoints[*msg.Id] = msg.Endpoint // 标记污点 m.hostEndpointsDirty = true ... // host endpoint remove case *proto.HostEndpointRemove: // 上报给xdpState/sockmapState m.callbacks.InvokeRemoveHostEndpoint(*msg.Id) // 清理raw cache delete(m.rawHostEndpoints, *msg.Id) // 标记host污点 m.hostEndpointsDirty = true ... // iface state uodate case *ifaceStateUpdate: // 更新iface cache m.pendingIfaceUpdates[msg.Name] = msg.State // iface addr update case *ifaceAddrsUpdate: // 仅关心host iface if m.wlIfacesRegexp.MatchString(msg.Name) { return } // 更新或清理host addr缓存 if msg.Addrs != nil { m.hostIfaceToAddrs[msg.Name] = msg.Addrs } else { delete(m.hostIfaceToAddrs, msg.Name) } // 标记host污点 m.hostEndpointsDirty = true } }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注意
m.callbacks.Invoke会触发xdpState/sockmapState增删处理
# 2.应用
# 2.1.updateBatch
m.ResolveUpdateBatch()会匹配iface-->hostEndpoint，更新内存映射关系及将iface-->hostEndpoint对同步给eBPFEndpointMgr。func (m *endpointManager) ResolveUpdateBatch() error { // 扫描pendingIfaceUpdates for ifaceName, state := range m.pendingIfaceUpdates { // up iface if state == ifacemonitor.StateUp { // 注册到activeUpIfaces m.activeUpIfaces.Add(ifaceName) // workload iface if m.wlIfacesRegexp.MatchString(ifaceName) { // 注册到待配置sysctl列表 m.wlIfaceNamesToReconfigure.Add(ifaceName) } // down iface } else { // 由activeUpIfaces解注册 m.activeUpIfaces.Discard(ifaceName) } // iface标记需要report status m.markEndpointStatusDirtyByIface(ifaceName) // clean pendingIfaceUpdates delete(m.pendingIfaceUpdates, ifaceName) } // 首次/OnUpdate.HostEndpoint/OnUpdate.ifaceAddrsUpdate if m.hostEndpointsDirty { // 重新计算host endpoint m.newIfaceNameToHostEpID = m.resolveHostEndpoints() } return nil } func (m *endpointManager) resolveHostEndpoints() map[string]proto.HostEndpointID { ... // 扫描hostIfaceToAddrs for ifaceName, ifaceAddrs := range m.hostIfaceToAddrs { bestHostEpId := proto.HostEndpointID{} HostEpLoop: // 扫描rawHostEndpoints for id, hostEp := range m.rawHostEndpoints { // hep.Name == "*" if forAllInterfaces(hostEp) { continue } // iface匹配多个host endpoint,仅取hep.endpointID最小的 if (bestHostEpId.EndpointId != "") && (bestHostEpId.EndpointId < id.EndpointId) { // We already have a HostEndpointId that is better than // this one, so no point looking any further. continue } // iface匹配到hostep if hostEp.Name == ifaceName { // The HostEndpoint has an explicit name that matches the // interface. bestHostEpId = id continue // hostep不匹配 } else if hostEp.Name != "" { // The HostEndpoint has an explicit name that isn't this interface. continue } // hostep匹配ifaceAddr for _, wantedList := range [][]string{hostEp.ExpectedIpv4Addrs, hostEp.ExpectedIpv6Addrs} { for _, wanted := range wantedList { if ifaceAddrs.Contains(wanted) { // The HostEndpoint expects an IP address that is on this interface. bestHostEpId = id continue HostEpLoop } } } } // 记录iface对应hostep if bestHostEpId.EndpointId != "" { newIfaceNameToHostEpID[ifaceName] = bestHostEpId } } ... // Similar loop to find the best all-interfaces host endpoint. for id, hostEp := range m.rawHostEndpoints { // hep.Name != "*" if !forAllInterfaces(hostEp) { continue } // 多个全局hostep,选hostepID最小的 if (bestHostEpId.EndpointId != "") && (bestHostEpId.EndpointId < id.EndpointId) { // We already have a HostEndpointId that is better than this one, so no point looking any further. continue } bestHostEpId = id } // 记录全局iface对应hostep if bestHostEpId.EndpointId != "" { newIfaceNameToHostEpID[allInterfaces] = bestHostEpId } // eBPF dataplane if m.bpfEndpointManager != nil { // Construct map of interface names to host endpoints, and pass to the BPF endpoint manager. hostIfaceToEpMap := map[string]proto.HostEndpoint{} for ifaceName, id := range newIfaceNameToHostEpID { // Note, dereference the proto.HostEndpoint here so that the data lifetime // is decoupled from the validity of the pointer here. hostIfaceToEpMap[ifaceName] = *m.rawHostEndpoints[id] } // iface-->hostEP回调通知到eBPFEndpointMgr m.bpfEndpointManager.OnHEPUpdate(hostIfaceToEpMap) } return newIfaceNameToHostEpID }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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
注意
resolveHostEndpoints基于ifaceState/ifaceAddr匹配最佳的hostEndpoint，eBPF Dataplane模式会将结果推给eBPFEPMgr
# 2.2.deferredWork
m.CompleteDeferredWork()是endpoint变更的提交入口，会将OnUpdate更新的dirty数据统一处理，同步到dataplane/kernel。func (m *endpointManager) CompleteDeferredWork() error { // 合并/解析待处理的workload endpoint m.resolveWorkloadEndpoints() // 首次/OnUpdate.HostEndpoint/OnUpdate.ifaceAddrsUpdate if m.hostEndpointsDirty { // 同步host endpoint m.updateHostEndpoints() // 重置dirty标记 m.hostEndpointsDirty = false } // 首次启动/resolveWorkloadEndpoints解析 if m.rpfSkipChainDirty { // 更新RPFSkip相关chain m.updateRPFSkipChain() // 重置dirty标记 m.rpfSkipChainDirty = false } // IPVS & 首次启动/resolveWorkloadEndpoints解析/updateHostEndpoints解析 if m.kubeIPVSSupportEnabled && m.needToCheckEndpointMarkChains { // endpoint mark规则调整 m.resolveEndpointMarks() // 重置dirty标记 m.needToCheckEndpointMarkChains = false } // send any endpoint status updates to report status combined m.updateEndpointStatuses() return nil } func (m *endpointManager) updateRPFSkipChain() { // cali-rpf-skip chain及rule chain := &iptables.Chain{ Name: rules.ChainRpfSkip, Rules: make([]iptables.Rule, 0), } // 扫描source spoofing for interfaceName, addresses := range m.sourceSpoofingConfig { for _, addr := range addresses { // 注册rule(-A cali-rpf-skip -i eth0 -s 10.0.0.0/24 -j ACCEPT) chain.Rules = append(chain.Rules, iptables.Rule{ Match: iptables.Match().InInterface(interfaceName).SourceNet(addr), Action: iptables.AcceptAction{}, }) } } // 更新raw table m.rawTable.UpdateChain(chain) } func (t *Table) UpdateChain(chain *Chain) { ... // chain引用更新及注册到dirty t.increfReferredChains(chain.Rules) // old chain引用清理 if oldChain := t.chainNameToChain[chain.Name]; oldChain != nil { ... t.decrefReferredChains(oldChain.Rules) } // 更新chainName-->new chain映射 t.chainNameToChain[chain.Name] = chain .... // chain被引用,注册到dirty if t.chainRefCounts[chain.Name] > 0 { t.dirtyChains.Add(chain.Name) } ... }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
注意
dataplane.loop收到dpconnector消息会利用回调mgr.OnUpdate分发给不同manager，dataplane.apply阶段触发变更提交
# 2.3.resolvewep
m.resolveWorkloadEndpoints()会将workload endpoint同步到linux内核，设置iptables chain及回调注入route。func (m *endpointManager) resolveWorkloadEndpoints() { // workload endpoint变化 if len(m.pendingWlEpUpdates) > 0 { // 标记dispatch chain检查 m.needToCheckDispatchChains = true } // endpoint清理回调 removeActiveWorkload := func(..., oldWorkload *proto.WorkloadEndpoint, id proto.WorkloadEndpointID) { // 回调通知xdpState/sockmapState endpoint删除 m.callbacks.InvokeRemoveWorkload(oldWorkload) // workload endpoint关联的iptables filter chain引用及dirtyChains更新 m.filterTable.RemoveChains(m.activeWlIDToChains[id]) delete(m.activeWlIDToChains, id) // workload endpoint不为空 if oldWorkload != nil { // endpoint关联mark释放 m.epMarkMapper.ReleaseEndpointMark(oldWorkload.Name) // 触发endpoint iface route清理 m.routeTable.SetRoutes(oldWorkload.Name, nil) // endpoint iface标记sysctl清理 m.wlIfaceNamesToReconfigure.Discard(oldWorkload.Name) // endpoint iface-->endpoint映射清理 delete(m.activeWlIfaceNameToID, oldWorkload.Name) // endpoint标记rpf skip清理 if m.hasSourceSpoofingConfiguration(oldWorkload.Name) { delete(m.sourceSpoofingConfig, oldWorkload.Name) m.rpfSkipChainDirty = true } } // endpoint active列表清理 delete(m.activeWlEndpoints, id) } // 扫描待更新的workload endpoint for len(m.pendingWlEpUpdates) > 0 { // Handle pending workload endpoint updates. for id, workload := range m.pendingWlEpUpdates { oldWorkload := m.activeWlEndpoints[id] if workload != nil { // iface endpoint冲突 if existingId, ok := m.activeWlIfaceNameToID[workload.Name]; ok && existingId != id { // 最新的endpointID大 if wlIdsAscending(&existingId, &id) { // 暂存endpoint m.shadowedWlEndpoints[id] = workload // 由pending提前消费掉 delete(m.pendingWlEpUpdates, id) continue } // 暂存old endpoint m.shadowedWlEndpoints[existingId] = m.activeWlEndpoints[existingId] // old endpoint内存状态及route清理 removeActiveWorkload(logCxt, m.activeWlEndpoints[existingId], existingId) } // endpoint Name变化 if oldWorkload != nil && oldWorkload.Name != workload.Name { // 释放old endpoint mark m.epMarkMapper.ReleaseEndpointMark(oldWorkload.Name) // iptable dataplane模式 if !m.bpfEnabled { // old endpoint关联的iptables filter chain引用及dirtyChains更新 m.filterTable.RemoveChains(m.activeWlIDToChains[id]) // endpoint标记rpf skip清理 if m.hasSourceSpoofingConfiguration(oldWorkload.Name) { delete(m.sourceSpoofingConfig, workload.Name) m.rpfSkipChainDirty = true } } // old endpoint route清理 m.routeTable.SetRoutes(oldWorkload.Name, nil) // old endpoint标记的sysctl清理 m.wlIfaceNamesToReconfigure.Discard(oldWorkload.Name) // old endpoint active列表清理 delete(m.activeWlIfaceNameToID, oldWorkload.Name) } ... // 提取endpoint关联的策略 if len(workload.Tiers) > 0 { ingressPolicyNames = workload.Tiers[0].IngressPolicies egressPolicyNames = workload.Tiers[0].EgressPolicies } ... // iptables dataplane if !m.bpfEnabled { // 生成endpoint关联的iptables chain chains := m.ruleRenderer.WorkloadEndpointToIptablesChains( workload.Name, m.epMarkMapper, adminUp, ingressPolicyNames, egressPolicyNames, workload.ProfileIds) // chains注册到filter table m.filterTable.UpdateChains(chains) m.activeWlIDToChains[id] = chains // endpoint rpf skip配置更新 if len(workload.AllowSpoofedSourcePrefixes) > 0 && !m.hasSourceSpoofingConfiguration(workload.Name) { m.sourceSpoofingConfig[workload.Name] = workload.AllowSpoofedSourcePrefixes m.rpfSkipChainDirty = true } else if m.hasSourceSpoofingConfiguration(workload.Name) && len(workload.AllowSpoofedSourcePrefixes) == 0 { delete(m.sourceSpoofingConfig, workload.Name) m.rpfSkipChainDirty = true } } ... // floatingIP注册 for _, natInfo := range natInfos { // 支持floatingIP或openstack集群 if m.floatingIPsEnabled || id.OrchestratorId == apiv3.OrchestratorOpenStack { // 合并floatingIP和endpointIP if !alreadyCopied { ipStrings = append([]string(nil), ipStrings...) alreadyCopied = true } ipStrings = append(ipStrings, natInfo.ExtIp+addrSuffix) } } ... // endpoint mac解析 if workload.Mac != "" { var err error mac, err = net.ParseMAC(workload.Mac) ... } ... // active endpoint生成路由条目 if adminUp { // 10.244.1.5/32 → ee:ee:ee:ee:ee:01 for _, s := range ipStrings { routeTargets = append(routeTargets, routetable.Target{ CIDR: ip.MustParseCIDROrIP(s), DestMAC: mac, }) } } // endpoint route推到routeTable处理 m.routeTable.SetRoutes(workload.Name, routeTargets) // 标记endpoint iface需要重新配置 m.wlIfaceNamesToReconfigure.Add(workload.Name) // 标记endpoint/iface active m.activeWlEndpoints[id] = workload m.activeWlIfaceNameToID[workload.Name] = id delete(m.pendingWlEpUpdates, id) // 通知xdpState/sockmapState切换endpoint状态 m.callbacks.InvokeUpdateWorkload(oldWorkload, workload) } else { // endpoint状态清理 removeActiveWorkload(logCxt, oldWorkload, id) delete(m.pendingWlEpUpdates, id) delete(m.shadowedWlEndpoints, id) // old endpoint非空 if oldWorkload != nil { // 由shadowedWlEndpoints选出替代的endpoint bestShadowedId := proto.WorkloadEndpointID{} for sId, sWorkload := range m.shadowedWlEndpoints { if sWorkload.Name == oldWorkload.Name { if bestShadowedId.EndpointId == "" || wlIdsAscending(&sId, &bestShadowedId) { bestShadowedId = sId } } } // 替代的endpoint加入pendingWlEpUpdates if bestShadowedId.EndpointId != "" { m.pendingWlEpUpdates[bestShadowedId] = m.shadowedWlEndpoints[bestShadowedId] delete(m.shadowedWlEndpoints, bestShadowedId) } } } // Update or deletion, make sure we update the interface status. m.epIDsToUpdateStatus.Add(id) } } // iptables dataplane & endpoint changed if !m.bpfEnabled && m.needToCheckDispatchChains { // 生成endpoint dispatch chain转到fw/tw chain newDispatchChains := m.ruleRenderer.WorkloadDispatchChains(m.activeWlEndpoints) m.updateDispatchChains(m.activeWlDispatchChains, newDispatchChains, m.filterTable) m.needToCheckDispatchChains = false // Set flag to update endpoint mark chains. m.needToCheckEndpointMarkChains = true } // 需要重新配置的endpoint iface m.wlIfaceNamesToReconfigure.Iter(func(ifaceName string) error { err := m.configureInterface(ifaceName) if err != nil { m.interfaceExistsInProcSys(ifaceName) ... return nil } return set.RemoveItem }) }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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
注意
这里比较重要的是基于基于
workload endpoint状态生成iptables chain、设置route及回调通知xdpstate/sockmapstate状态
# 2.3.updatehep
m.updateHostEndpoints()负责将host endpoint配置(policy/tier/iface)转换为raw/mangle/filter chains及更新对应table。func (m *endpointManager) updateHostEndpoints() { ... // 扫描hostendpoint for ifaceName, id := range newIfaceNameToHostEpID { // endpoint metadata ep := m.rawHostEndpoints[id] // 关联raw/untracked策略 if len(ep.UntrackedTiers) > 0 { // iface != "*" if ifaceName != allInterfaces { newUntrackedIfaceNameToHostEpID[ifaceName] = id } } // 关联mangle策略 if len(ep.PreDnatTiers) > 0 { // Similar optimisation (or neatness) for pre-DNAT policy. newPreDNATIfaceNameToHostEpID[ifaceName] = id } // Record that this host endpoint is in use, for status reporting. newHostEpIDToIfaceNames[id] = append(newHostEpIDToIfaceNames[id], ifaceName) ... } // iptables dataplane if !m.bpfEnabled { ... // 扫描hostendpoint for ifaceName, id := range newIfaceNameToHostEpID { // host endpoint metadata hostEp := m.rawHostEndpoints[id] ... // endpoint关联filter policy if len(hostEp.Tiers) > 0 { ingressPolicyNames = hostEp.Tiers[0].IngressPolicies egressPolicyNames = hostEp.Tiers[0].EgressPolicies } // endpoint关联forward policy if len(hostEp.ForwardTiers) > 0 { ingressForwardPolicyNames = hostEp.ForwardTiers[0].IngressPolicies egressForwardPolicyNames = hostEp.ForwardTiers[0].EgressPolicies } // 生成iptables chain filtChains := m.ruleRenderer.HostEndpointToFilterChains( ifaceName, m.epMarkMapper, ingressPolicyNames, egressPolicyNames, ingressForwardPolicyNames, egressForwardPolicyNames, hostEp.ProfileIds, ) // 更新filter table if !reflect.DeepEqual(filtChains, m.activeHostIfaceToFiltChains[ifaceName]) { m.filterTable.UpdateChains(filtChains) } // 更新iface-->filter chain newHostIfaceFiltChains[ifaceName] = filtChains delete(m.activeHostIfaceToFiltChains, ifaceName) // 生成mangle chain mangleChains := m.ruleRenderer.HostEndpointToMangleEgressChains( ifaceName, egressPolicyNames, hostEp.ProfileIds, ) // 更新mangle table if !reflect.DeepEqual(mangleChains, m.activeHostIfaceToMangleEgressChains[ifaceName]) { m.mangleTable.UpdateChains(mangleChains) } // 更新iface-->mangle chain newHostIfaceMangleEgressChains[ifaceName] = mangleChains delete(m.activeHostIfaceToMangleEgressChains, ifaceName) } ... // mangle prenat规则 for ifaceName, id := range newPreDNATIfaceNameToHostEpID { hostEp := m.rawHostEndpoints[id] ... // Update the mangle table for preDNAT policy. if len(hostEp.PreDnatTiers) > 0 { ingressPolicyNames = hostEp.PreDnatTiers[0].IngressPolicies } // 渲染mangle chain mangleChains := m.ruleRenderer.HostEndpointToMangleIngressChains( ifaceName, ingressPolicyNames, ) // 更新mangle table if !reflect.DeepEqual(mangleChains, m.activeHostIfaceToMangleIngressChains[ifaceName]) { m.mangleTable.UpdateChains(mangleChains) } newHostIfaceMangleIngressChains[ifaceName] = mangleChains delete(m.activeHostIfaceToMangleIngressChains, ifaceName) } // filter/manage chain更新 for ifaceName, chains := range m.activeHostIfaceToFiltChains { m.filterTable.RemoveChains(chains) } for ifaceName, chains := range m.activeHostIfaceToMangleEgressChains { m.mangleTable.RemoveChains(chains) } for ifaceName, chains := range m.activeHostIfaceToMangleIngressChains { m.mangleTable.RemoveChains(chains) } m.callbacks.InvokeInterfaceCallbacks(m.activeIfaceNameToHostEpID, newIfaceNameToHostEpID) m.activeHostIfaceToFiltChains = newHostIfaceFiltChains m.activeHostIfaceToMangleEgressChains = newHostIfaceMangleEgressChains m.activeHostIfaceToMangleIngressChains = newHostIfaceMangleIngressChains } // ipv4或eBPF未启用 if m.ipVersion == 4 || !m.bpfEnabled /* BPF enforces RPF on its own */ { ... // Build iptables chains for untracked host endpoint policy. for ifaceName, id := range newUntrackedIfaceNameToHostEpID { hostEp := m.rawHostEndpoints[id] ... // 获取untracked policy if len(hostEp.UntrackedTiers) > 0 { ingressPolicyNames = hostEp.UntrackedTiers[0].IngressPolicies egressPolicyNames = hostEp.UntrackedTiers[0].EgressPolicies } ... // 渲染raw chain if m.bpfEnabled { rawChains = append(rawChains, m.ruleRenderer.HostEndpointToRawEgressChain( ifaceName, egressPolicyNames, )) } else { rawChains = m.ruleRenderer.HostEndpointToRawChains( ifaceName, ingressPolicyNames, egressPolicyNames, ) } // 更新raw table chain if !reflect.DeepEqual(rawChains, m.activeHostIfaceToRawChains[ifaceName]) { m.rawTable.UpdateChains(rawChains) } newHostIfaceRawChains[ifaceName] = rawChains delete(m.activeHostIfaceToRawChains, ifaceName) } // Remove untracked policy iptables chains that are no longer wanted. for ifaceName, chains := range m.activeHostIfaceToRawChains { m.rawTable.RemoveChains(chains) } m.activeHostIfaceToRawChains = newHostIfaceRawChains } // Remember the host endpoints that are now in use. m.activeIfaceNameToHostEpID = newIfaceNameToHostEpID m.activeHostEpIDToIfaceNames = newHostEpIDToIfaceNames // ipv4或eBPF未启用 if m.ipVersion == 4 || !m.bpfEnabled { ... // 渲染raw dispatch chain if m.bpfEnabled { newRawDispatchChains = m.ruleRenderer.ToHostDispatchChains(newUntrackedIfaceNameToHostEpID, "") } else { newRawDispatchChains = m.ruleRenderer.HostDispatchChains(newUntrackedIfaceNameToHostEpID, "", false) } // 更新raw table dispatch chain m.updateDispatchChains(m.activeHostRawDispatchChains, newRawDispatchChains, m.rawTable) } if m.bpfEnabled { // Code after this point is for other dispatch chains and IPVS endpoint marking, // which aren't needed in BPF mode. return } // Rewrite the filter dispatch chains if they've changed. defaultIfaceName := "" if _, ok := newIfaceNameToHostEpID[allInterfaces]; ok { // All-interfaces host endpoint is active. defaultIfaceName = allInterfaces delete(newIfaceNameToHostEpID, allInterfaces) } // all-interface filter dispatch chain newFilterDispatchChains := m.ruleRenderer.HostDispatchChains(newIfaceNameToHostEpID, defaultIfaceName, true) // all-interface mangle dispatch chain newMangleEgressDispatchChains := m.ruleRenderer.ToHostDispatchChains(newIfaceNameToHostEpID, defaultIfaceName) // 更新filter table m.updateDispatchChains(m.activeHostFilterDispatchChains, newFilterDispatchChains, m.filterTable) // Set flag to update endpoint mark chains. m.needToCheckEndpointMarkChains = true // Rewrite the mangle dispatch chains if they've changed. defaultIfaceName = "" if _, ok := newPreDNATIfaceNameToHostEpID[allInterfaces]; ok { // All-interfaces host endpoint is active. Arrange for it to be the // default. This is handled the same as the filter dispatch chains above. defaultIfaceName = allInterfaces delete(newPreDNATIfaceNameToHostEpID, allInterfaces) } // 渲染mangle chain newMangleIngressDispatchChains := m.ruleRenderer.FromHostDispatchChains(newPreDNATIfaceNameToHostEpID, defaultIfaceName) newMangleDispatchChains := append(newMangleIngressDispatchChains, newMangleEgressDispatchChains...) // 更新mangle table m.updateDispatchChains(m.activeHostMangleDispatchChains, newMangleDispatchChains, m.mangleTable) }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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
补充
这里生成大量的
iptables chain/rule拦截及校验流量，后续需部署calico结合pod网卡数据具体分析
# 2.4.endpointMark
m.resolveEndpointMarks()用于iptables dataplane模式维护endpoint mark标记相关的dispatch chain，以跳转对应处理链。func (m *endpointManager) resolveEndpointMarks() { if m.bpfEnabled { return } // Render endpoint mark chains for active workload and host endpoint. newEndpointMarkDispatchChains := m.ruleRenderer.EndpointMarkDispatchChains(m.epMarkMapper, m.activeWlEndpoints, m.activeIfaceNameToHostEpID) m.updateDispatchChains(m.activeEPMarkDispatchChains, newEndpointMarkDispatchChains, m.filterTable) } // In some scenario, e.g. packet goes to an kubernetes ipvs service ip. func (r *DefaultRuleRenderer) EndpointMarkDispatchChains(...) []*Chain { // Extract endpoint names. wlNames := make([]string, 0, len(wlEndpoints)) for _, endpoint := range wlEndpoints { wlNames = append(wlNames, endpoint.Name) } hepNames := make([]string, 0, len(hepEndpoints)) for ifaceName := range hepEndpoints { hepNames = append(hepNames, ifaceName) } // mark dispatch chain渲染 return r.endpointMarkDispatchChains( wlNames, hepNames, epMarkMapper, SetEndPointMarkPfx, WorkloadFromEndpointPfx, HostFromEndpointForwardPfx, ChainDispatchSetEndPointMark, ChainDispatchFromEndPointMark, ) }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
注意
workload/host endpoint均对应set mark chain和mark dispatch chain，基于mark将不同endpoint流量分发到处理链