iptables
南风未起 2024-10-30 20:36:22 kubeproxy
# 1.iptables
# 1.1.简介
iptables是linux内核的包过滤防火墙系统，基于iptables可以添加和删除具体的规则，iptables维护4表5链，防火墙策略规则分别写入这些表和链实现包过滤和地址转发。--- filter表 控制数据包准入,决定数据包是否可以到达目标进程端口，可以控制的链路是INPUT/FORWARD/OUTPUT --- nat表 控制数据包地址转换,可以修改源和目标的IP地址,实现包路由,可以控制的链路有PREROUTING/OUTPUT/POSTROUTING --- mangle表 修改数据包的原数据,比如TTL,可以控制的链路有PREROUTING/INPUT/OUTPUT/FORWARD/POSTROUTING --- raw表 控制nat表中连接追踪机制的启用状况,能基于数据包的状态进行规则设定,可以控制的链路有PREROUTING/OUTPUT --- input链 入栈数据过滤,对路由策略分派的包到达目标进程端口前进行匹配处理 --- output链 出栈数据过滤,对本机目标进程端口的包转发给请求方 --- forward链 转发数据过滤,对路由策略分配的包进行路由转发 --- prerouting链 路由前过滤,包到达网口进行规则匹配 --- postrouting链 路由后过滤,包离开网口进行匹配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
注意
1.外部数据包会经过
prerouting链和input链处理2.本机数据包到外部会经过
output链和postrouting链处理3.防火墙作为路由转发数据会经过
prerouting链、forward链和postrouting链
# 1.2.链扩展
iptables模式下，kubeproxy会根据service和endpoints实时刷新规则，利用filter表和nat表扩充kube-services相关链，扩充的链最终又插入原本的5链实现数据包劫持。
注意
iptables会基于cidr检查外部流量，注入标记以便postrouting返回包时进行源地址伪装
# 1.3.流量倾斜
iptables模式下，请求流量第一次到达会选择endpoints中的某个pod作为目的地，处理完成会进行连接释放。但tcp连接建立及重新用于后续请求的情况下，长连接模式会造成流量倾斜，已建立的连接不会再调用iptables进行负载，导致流量固定送到某个pod。
注意
udp连接是无状态的，service和endpoint删除时，相对于tcp连接需要多做一步，释放conntracker，避免连接残留
# 2.iptproxier
# 2.1.newProxier
proxier实现Provider定义的serviceHandler/endpointHandler/endpointSliceHandler/nodeHandler，用于监听到资源变化时增量触发规则更新，本质上执行的是sync()和syncLoop()，基于iptables工具进行规则管理。// Provider is the interface provided by proxier implementations. type Provider interface { config.EndpointSliceHandler config.ServiceHandler config.NodeHandler // Sync immediately synchronizes the Provider's current state to proxy rules. Sync() // SyncLoop runs periodic work for proxy rules. SyncLoop() } // NewProxier returns a new Proxier given an iptables Interface instance. func NewProxier(...) (*Proxier, error) { // 1.设置内核loopback回环参数 if utilproxy.ContainsIPv4Loopback(nodePortAddresses) { utilproxy.EnsureSysctl(sysctl, sysctlRouteLocalnet, 1) ... } // 2.开启br_netfilter,bridge流量经过iptables sysctl.GetSysctl(sysctlBridgeCallIPTables) ... // 3.外部流量标记(0x4000) masqueradeValue := 1 << uint(masqueradeBit) masqueradeMark := fmt.Sprintf("%#08x", masqueradeValue) serviceHealthServer := healthcheck.NewServiceHealthServer(hostname, recorder, nodePortAddresses) ... // 初始化proxier proxier := &Proxier{ // 缓存及增量事件 serviceMap: make(proxy.ServiceMap), serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, recorder, nil), endpointsMap: make(proxy.EndpointsMap), endpointsChanges: proxy.NewEndpointChangeTracker(hostname, newEndpointInfo, ipFamily, recorder, nil), // 执行周期(30s) syncPeriod: syncPeriod, // iptables相关 iptables: ipt, masqueradeAll: masqueradeAll, masqueradeMark: masqueradeMark, exec: exec, localDetector: localDetector, ... } ... // 初始化runner proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, time.Hour, burstSyncs) // 后台监控iptables规则链健康 go ipt.Monitor(kubeProxyCanaryChain, []utiliptables.Table{utiliptables.TableMangle, utiliptables.TableNAT, utiliptables.TableFilter}, proxier.syncProxyRules, syncPeriod, wait.NeverStop) ... return proxier, nil } // iptables自愈机制 func (runner *runner) Monitor(canary Chain, tables []Table, reloadFunc func(), interval time.Duration, stopCh <-chan struct{}) { for { // 间隔30s向mangle/nat/filter表创建KUBE-PROXY-CANARY链,直至成功 _ = utilwait.PollImmediateUntil(interval, func() (bool, error) { for _, table := range tables { runner.EnsureChain(table, canary) ... } return true, nil }, stopCh) // 间隔30s检测KUBE-PROXY-CANARY链存在 err := utilwait.PollUntil(interval, func() (bool, error) { // 检测mangle表 runner.ChainExists(tables[0], canary) ... // mangle表不存在,间隔100ms检查其它表 err := utilwait.PollImmediate(iptablesFlushPollTime, iptablesFlushTimeout, func() (bool, error) { for i := 1; i < len(tables); i++ { runner.ChainExists(tables[i], canary) ... } return true, nil }) ... // mangle/filter/nat表不存在KUBE-PROXY-CANARY链 return true, nil }, stopCh) // stopCh关闭 if err != nil { // 清理KUBE-PROXY-CANARY链 for _, table := range tables { _ = runner.DeleteChain(table, canary) } return } // 执行syncProxyRules重建 reloadFunc() } }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
注意
1.
proxier最重要的就是syncRunner，用于周期或事件触发syncProxyRules()执行2.
dualProxier是ipv4proxier和ipv6proxier的封装，内部会依次执行双栈的sync()和syncLoop()
# 2.2.eventWatch
proxier实现了serviceHandler/endpointSliceHandler/nodeHandler相关接口，service/endpointSlice/node变更时触发回调驱动syncLoop()进行规则更新。// service相关 func (proxier *Proxier) OnServiceAdd(service *v1.Service) { proxier.OnServiceUpdate(nil, service) } func (proxier *Proxier) OnServiceUpdate(oldService, service *v1.Service) { if proxier.serviceChanges.Update(oldService, service) && proxier.isInitialized() { proxier.Sync() } } func (proxier *Proxier) OnServiceDelete(service *v1.Service) { proxier.OnServiceUpdate(service, nil) } //endpointSlice相关 func (proxier *Proxier) OnEndpointSliceAdd(endpointSlice *discovery.EndpointSlice) { if proxier.endpointsChanges.EndpointSliceUpdate(endpointSlice, false) && proxier.isInitialized() { proxier.Sync() } } func (proxier *Proxier) OnEndpointSliceUpdate(_, endpointSlice *discovery.EndpointSlice) { if proxier.endpointsChanges.EndpointSliceUpdate(endpointSlice, false) && proxier.isInitialized() { proxier.Sync() } } func (proxier *Proxier) OnEndpointSliceDelete(endpointSlice *discovery.EndpointSlice) { if proxier.endpointsChanges.EndpointSliceUpdate(endpointSlice, true) && proxier.isInitialized() { proxier.Sync() } } // 信号通知 func (proxier *Proxier) Sync() { ... proxier.syncRunner.Run() } func (bfr *BoundedFrequencyRunner) Run() { // 发送处理信号 select { case bfr.run <- struct{}{}: default: } } // node相关 func (proxier *Proxier) OnNodeAdd(node *v1.Node) { ... proxier.OnNodeUpdate(nil,node) } func (proxier *Proxier) OnNodeUpdate(oldNode, node *v1.Node) { // 非本节点 if node.Name != proxier.hostname { return } // nodeLabel无变化 if reflect.DeepEqual(proxier.nodeLabels, node.Labels) { return } proxier.mu.Lock() // 更新nodeLabel proxier.nodeLabels = map[string]string{} for k, v := range node.Labels { proxier.nodeLabels[k] = v } proxier.mu.Unlock() // 规则刷新 proxier.syncProxyRules() } // OnNodeDelete is called whenever deletion of an existing node // object is observed. func (proxier *Proxier) OnNodeDelete(node *v1.Node) { // 非本节点 if node.Name != proxier.hostname { return } proxier.mu.Lock() // 重置nodeLabel proxier.nodeLabels = nil proxier.mu.Unlock() // 规则刷新 proxier.syncProxyRules() }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
注意
1.
service/endpointSlice的增量事件基于syncRunner.sync()通知信号触发更新2.
node的增量事件对比nodeLabel差异，调用proxier.syncProxyRules()更新规则
# 2.3.syncProxyRules
proxier.syncProxyRules()负责iptables规则更新，基于最新的service/endpoints生成及下放iptables策略，确保流量能正确地转发到目标应用。// SyncLoop runs periodic work.This is expected to run as a goroutine or as the main loop of the app. func (proxier *Proxier) SyncLoop() { ... proxier.syncRunner.Loop(wait.NeverStop) } // Loop handles the periodic timer and run requests.This is expected to be called as a goroutine. func (bfr *BoundedFrequencyRunner) Loop(stop <-chan struct{}) { // 重置timer定时为30s bfr.timer.Reset(bfr.maxInterval) for { select { // 同步终止 case <-stop: // 关闭timmer及rateLimiter bfr.stop() return // 30s周期 case <-bfr.timer.C(): bfr.tryRun() // 增量事件信号 case <-bfr.run: bfr.tryRun() // 重试信号 case <-bfr.retry: bfr.doRetry() } } } // 规则同步 func (bfr *BoundedFrequencyRunner) tryRun() { bfr.mu.Lock() defer bfr.mu.Unlock() // 限制访问速率(0.033s生成1枚令牌,最大2枚) if bfr.limiter.TryAccept() { // 执行syncProxyRules bfr.fn() // 重置定时器30s bfr.timer.Reset(bfr.maxInterval) return } ... // 重置定时器min定时剩余时间,30-since(lastRun) bfr.timer.Stop() bfr.timer.Reset(nextScheduled) } // This is where all of the iptables-save/restore calls happen. func (proxier *Proxier) syncProxyRules() { proxier.mu.Lock() defer proxier.mu.Unlock() // service/endpoints未完成同步 if !proxier.isInitialized() { return } ... // 1.最新状态更新至serviceMap // 2.清理change.previous中change.current存在的资源(避免update事件误判为delete) // 3.serviceMap清理change.previous中的资源(delete事件),收集state udp clusterIP // 4.重置change.items // 5.构建add/update结果 serviceUpdateResult := proxier.serviceMap.Update(proxier.serviceChanges) // 1.清理endpoint中change.previous包含的旧端点 // 2.change.current最新端点合并到endpointMap // 3.收集udp过期端点(change.previous存在的ready端点,change.current不存在) // 4.收集过期udp服务(change.previous未就绪,change.current就绪,旧的conntrack可能存在) // 5.收集本地就绪端点 endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges) // 需要清理的udp服务clusterIP列表 conntrackCleanupServiceIPs := serviceUpdateResult.UDPStaleClusterIP // 需要清理conntrack的udp nodeport conntrackCleanupServiceNodePorts := sets.NewInt() // 遍历过期的udp service for _, svcPortName := range endpointUpdateResult.StaleServiceNames { // 过期的udp服务存在(udp/sctp) if svcInfo, ok := proxier.serviceMap[svcPortName]; ok && svcInfo != nil && conntrack.IsClearConntrackNeeded(svcInfo.Protocol()) { // 收集需清理的udp service clusterIP conntrackCleanupServiceIPs.Insert(svcInfo.ClusterIP().String()) // 收集udp service externalIP for _, extIP := range svcInfo.ExternalIPStrings() { conntrackCleanupServiceIPs.Insert(extIP) } // 收集udp service LBIP for _, lbIP := range svcInfo.LoadBalancerIPStrings() { conntrackCleanupServiceIPs.Insert(lbIP) } // 收集需清理的udp nodeport nodePort := svcInfo.NodePort() if svcInfo.Protocol() == v1.ProtocolUDP && nodePort != 0 { conntrackCleanupServiceNodePorts.Insert(nodePort) } } } ... defer func() { if !success { // 重试信号 proxier.syncRunner.RetryAfter(proxier.syncPeriod) } }() // 检查filter表和nat表目标链 for _, jump := range iptablesJumpChains { // 检查及创建目标链 proxier.iptables.EnsureChain(jump.table, jump.dstChain) ... // 检查及创建filter表和nat表中5链的目标规则 // iptables -t filter -I INPUT -m conntrack --ctstate NEW -m comment --comment "kubernetes externally- // visible service portals" -j KUBE-EXTERNAL-SERVICES proxier.iptables.EnsureRule(utiliptables.Prepend, jump.table, jump.srcChain, args...) ... } // 检查及创建KUBE-MARK-DROP链存在 for _, ch := range iptablesEnsureChains { // KUBE-MARK-DROP链检查--iptables -t nat -N KUBE-MARK-DROP proxier.iptables.EnsureChain(ch.table, ch.chain) ... } // 下面将写入iptables规则,不会提前返回 ... // 重置缓冲区 proxier.existingFilterChainsData.Reset() // 执行iptables-save -t filter获取现有链 err := proxier.iptables.SaveInto(utiliptables.TableFilter, proxier.existingFilterChainsData) ... // 缓存filter已有的链 existingFilterChains = utiliptables.GetChainLines(utiliptables.TableFilter, proxier.existingFilterChainsData.Bytes()) ... // 重置iptablesData proxier.iptablesData.Reset() // 执行iptables-save -t nat获取现有链 err = proxier.iptables.SaveInto(utiliptables.TableNAT, proxier.iptablesData) ... // 缓存nat表已有的链 existingNATChains = utiliptables.GetChainLines(utiliptables.TableNAT, proxier.iptablesData.Bytes()) ... // 计算filter表kube-services/kube-nodeports/kube-postrouting/kube-mark-masq链定义 for _, chainName := range []utiliptables.Chain{kubeServicesChain, kubeExternalServicesChain, kubeForwardChain, kubeNodePortsChain} { // 保留链及计数器 if chain, ok := existingFilterChains[chainName]; ok { proxier.filterChains.WriteBytes(chain) // 写入空链计数器 } else { proxier.filterChains.Write(utiliptables.MakeChainLine(chainName)) } } // 计算nat表kube-services/kube-nodeports/kube-postrouting/kube-mark-masq链定义 for _, chainName := range []utiliptables.Chain{kubeServicesChain, kubeNodePortsChain, kubePostroutingChain, KubeMarkMasqChain} { // 保留链及计数器 if chain, ok := existingNATChains[chainName]; ok { proxier.natChains.WriteBytes(chain) // 写入空链计数器 } else { proxier.natChains.Write(utiliptables.MakeChainLine(chainName)) } } // kube-postrouting标记匹配链规则 proxier.natRules.Write( "-A", string(kubePostroutingChain), "-m", "mark", "!", "--mark", fmt.Sprintf("%s/%s", proxier.masqueradeMark, proxier.masqueradeMark), "-j", "RETURN", ) // kube-postrouting标记重置链规则 proxier.natRules.Write( "-A", string(kubePostroutingChain), "-j", "MARK", "--xor-mark", proxier.masqueradeMark, ) // kube-postrouting源地址伪装链规则 masqRule := []string{ "-A", string(kubePostroutingChain), "-m", "comment", "--comment", "kubernetes service traffic requiring SNAT", "-j", "MASQUERADE", } // 源地址转换使用全范围端口随机化 if proxier.iptables.HasRandomFully() { masqRule = append(masqRule, "--random-fully") } proxier.natRules.Write(masqRule) // kube-mark-masq标记注入链规则 proxier.natRules.Write( "-A", string(KubeMarkMasqChain), "-j", "MARK", "--or-mark", proxier.masqueradeMark, ) // 活跃的nat chain activeNATChains := map[utiliptables.Chain]bool{} // use a map as a set // 长度64的字符串切片,用于存放临时的iptables命令参数,避免频繁扩容 args := make([]string, 64) // 计算最新的endpoint chain长度 proxier.endpointChainsNumber = 0 for svcName := range proxier.serviceMap { proxier.endpointChainsNumber += len(proxier.endpointsMap[svcName]) } // 获取节点可用的nodeport地址(零地址/nodeportAddresses覆盖的网卡地址) nodeAddresses, err := utilproxy.GetNodeAddresses(proxier.nodePortAddresses, proxier.networkInterfacer) ... // 保留符合当前proxier协议的nodeport地址 isIPv6 := proxier.iptables.IsIPv6() for addr := range nodeAddresses { if utilproxy.IsZeroCIDR(addr) && isIPv6 == netutils.IsIPv6CIDRString(addr) { nodeAddresses = sets.NewString(addr) break } } // service链规则 for svcName, svc := range proxier.serviceMap { svcInfo, ok := svc.(*serviceInfo) ... protocol := strings.ToLower(string(svcInfo.Protocol())) svcNameString := svcInfo.nameString allEndpoints := proxier.endpointsMap[svcName] // 分类endpoints // 1.service流量策略为cluster(默认)的ready endpoint // 2.service流量策略为local的ready endpoint // 3.本节点可路由到的endpoint(cluster/local/cluster+local) // 4.集群范围存在ready endpoint clusterEndpoints, localEndpoints, allLocallyReachableEndpoints, hasEndpoints := proxy.CategorizeEndpoints(allEndpoints, svcInfo, proxier.nodeLabels) // 生成endpoint chain规则 for _, ep := range allLocallyReachableEndpoints { epInfo, ok := ep.(*endpointsInfo) ... // kube-sep-xxx endpointChain := epInfo.ChainName // 计算nat表链数据 if chain, ok := existingNATChains[endpointChain]; ok { proxier.natChains.WriteBytes(chain) } else { proxier.natChains.Write(utiliptables.MakeChainLine(endpointChain)) } activeNATChains[endpointChain] = true args = append(args[:0], "-A", string(endpointChain)) args = proxier.appendServiceCommentLocked(args, svcNameString) // 转到kube-mark-masq的链规则 // -A chainName -s 10.244.1.117/32 -m comment --comment svcName -j chainName proxier.natRules.Write( args, "-s", epInfo.IP(), "-j", string(KubeMarkMasqChain)) // 客户端亲和性 // -A chainName -m comment --comment svcName -m recent --name chainName --set if svcInfo.SessionAffinityType() == v1.ServiceAffinityClientIP { args = append(args, "-m", "recent", "--name", string(endpointChain), "--set") } // -A chainName -m comment --comment svcName -m recent --name chainName --set -m tcp -p tcp -j DNAT // --to-destination 10.244.1.5:80 // 地址转换链规则 args = append(args, "-m", protocol, "-p", protocol, "-j", "DNAT", "--to-destination", epInfo.Endpoint) // 更新nat rules proxier.natRules.Write(args) } ... // internalPolicyChain初始值为kube-svc-xxx,internalTraffic取local时调整为kube-svl-xxx internalTrafficChain := internalPolicyChain // externalPolicyChain初始值为kube-svc-xxx,externalTraffic取local时调整为kube-svl-xxx // externalTrafficChain强制取kube-ext-xxx,避免流量黑洞 externalTrafficChain := svcInfo.externalChainName // eventually jumps to externalPolicyChain // service的kube-svc-xxx链(nodeport/lb) if hasEndpoints && svcInfo.UsesClusterEndpoints() { // Create the Cluster traffic policy chain, retaining counters if possible. if chain, ok := existingNATChains[clusterPolicyChain]; ok { proxier.natChains.WriteBytes(chain) } else { proxier.natChains.Write(utiliptables.MakeChainLine(clusterPolicyChain)) } activeNATChains[clusterPolicyChain] = true } // service的kube-svl-xxx链(clusterIP) if hasEndpoints && svcInfo.UsesLocalEndpoints() { if chain, ok := existingNATChains[localPolicyChain]; ok { proxier.natChains.WriteBytes(chain) } else { proxier.natChains.Write(utiliptables.MakeChainLine(localPolicyChain)) } activeNATChains[localPolicyChain] = true } // kube-ext-xxx链处理(nodePort/lb/externalIP) if hasEndpoints && svcInfo.ExternallyAccessible() { // 计算nat kube-ext-xxx链 if chain, ok := existingNATChains[externalTrafficChain]; ok { proxier.natChains.WriteBytes(chain) } else { proxier.natChains.Write(utiliptables.MakeChainLine(externalTrafficChain)) } activeNATChains[externalTrafficChain] = true // externalPolicy=cluster,生成kube-ext-xxx链的mark规则进行源地址伪装 if !svcInfo.ExternalPolicyLocal() { proxier.natRules.Write( "-A", string(externalTrafficChain), "-m", "comment", "--comment", fmt.Sprintf("masquerade traffic for %s external destinations", svcNameString), "-j", string(KubeMarkMasqChain)) // externalPolicy=local,只选本机的endpoint } else { // 本地流量检测(恒为true) if proxier.localDetector.IsImplemented() { // pod访问service外部地址流量处理,短路到kube-svc-xxx规则,避免无法到达endpoint proxier.natRules.Write( "-A", string(externalTrafficChain), "-m", "comment", "--comment", fmt.Sprintf("pod traffic for %s external destinations", svcNameString), proxier.localDetector.IfLocal(), "-j", string(clusterPolicyChain)) } // 本机进程访问service外部地址流量处理,进行源地址伪装,确保流量正确到达 proxier.natRules.Write( "-A", string(externalTrafficChain), "-m", "comment", "--comment", fmt.Sprintf("masquerade LOCAL traffic for %s external destinations", svcNameString), "-m", "addrtype", "--src-type", "LOCAL", "-j", string(KubeMarkMasqChain)) // 本机进程访问外部地址流量处理,转到kube-svc-xxx,确保流量正确送达 proxier.natRules.Write( "-A", string(externalTrafficChain), "-m", "comment", "--comment", fmt.Sprintf("route LOCAL traffic for %s external destinations", svcNameString), "-m", "addrtype", "--src-type", "LOCAL", "-j", string(clusterPolicyChain)) } // 其它流量转到kube-svc-xxx或kube-svl-xxx proxier.natRules.Write( "-A", string(externalTrafficChain), "-j", string(externalPolicyChain)) } // clusterIP流量捕获 if hasEndpoints { args = append(args[:0], "-m", "comment", "--comment", fmt.Sprintf("%s cluster IP", svcNameString), "-m", protocol, "-p", protocol, "-d", svcInfo.ClusterIP().String(), "--dport", strconv.Itoa(svcInfo.Port()), ) // 所有流向clusterIP流程进行源地址伪装(kube-svc-xxx/kube-svl-xxx) if proxier.masqueradeAll { proxier.natRules.Write( "-A", string(internalTrafficChain), args, "-j", string(KubeMarkMasqChain)) // 非本节点流量进行源地址伪装 } else if proxier.localDetector.IsImplemented() { // This masquerades off-cluster traffic to a service VIP. The idea // is that you can establish a static route for your Service range, // routing to any node, and that node will bridge into the Service // for you. Since that might bounce off-node, we masquerade here. proxier.natRules.Write( "-A", string(internalTrafficChain), args, proxier.localDetector.IfNotLocal(), "-j", string(KubeMarkMasqChain)) } // 打完标记转到kube-svc-xxx/kube-svl-xxx proxier.natRules.Write( "-A", string(kubeServicesChain), args, "-j", string(internalTrafficChain)) // service未关联可用endpoint } else { // 转到clusterIP流量拒绝 proxier.filterRules.Write( "-A", string(kubeServicesChain), "-m", "comment", "--comment", fmt.Sprintf("%s has no endpoints", svcNameString), "-m", protocol, "-p", protocol, "-d", svcInfo.ClusterIP().String(), "--dport", strconv.Itoa(svcInfo.Port()), "-j", "REJECT", ) } // service-externalIP流量捕获 for _, externalIP := range svcInfo.ExternalIPStrings() { // service关联可用endpoint,externalIP流量转到kube-ext-xxx if hasEndpoints { // Send traffic bound for external IPs to the "external // destinations" chain. proxier.natRules.Write( "-A", string(kubeServicesChain), "-m", "comment", "--comment", fmt.Sprintf("%s external IP", svcNameString), "-m", protocol, "-p", protocol, "-d", externalIP, "--dport", strconv.Itoa(svcInfo.Port()), "-j", string(externalTrafficChain)) // service未关联可用endpoint，externalIP流量拒绝 } else { // No endpoints. proxier.filterRules.Write( "-A", string(kubeExternalServicesChain), "-m", "comment", "--comment", fmt.Sprintf("%s has no endpoints", svcNameString), "-m", protocol, "-p", protocol, "-d", externalIP, "--dport", strconv.Itoa(svcInfo.Port()), "-j", "REJECT", ) } } // lbIP流量捕获 if len(svcInfo.LoadBalancerIPStrings()) > 0 && hasEndpoints { // 流量默认转到kube-ext-xxx链 nextChain := externalTrafficChain // service指定LBSourceRanges限制 if len(svcInfo.LoadBalancerSourceRanges()) > 0 { // 写入防火墙链kube-fw-xxx fwChain := svcInfo.firewallChainName if chain, ok := existingNATChains[fwChain]; ok { proxier.natChains.WriteBytes(chain) } else { proxier.natChains.Write(utiliptables.MakeChainLine(fwChain)) } activeNATChains[fwChain] = true // nextChain指向防火墙链 nextChain = svcInfo.firewallChainName } // lbIP规则刷新 for _, lbip := range svcInfo.LoadBalancerIPStrings() { // kube-service-chains链转向目标链 proxier.natRules.Write( "-A", string(kubeServicesChain), "-m", "comment", "--comment", fmt.Sprintf("%s loadbalancer IP", svcNameString), "-m", protocol, "-p", protocol, "-d", lbip, "--dport", strconv.Itoa(svcInfo.Port()), "-j", string(nextChain)) // kube-fw-xxx防火墙链刷新 if len(svcInfo.LoadBalancerSourceRanges()) > 0 { args = append(args[:0], "-A", string(nextChain), "-m", "comment", "--comment", fmt.Sprintf("%s loadbalancer IP", svcNameString), ) // firewall filter based on each source range allowFromNode := false for _, src := range svcInfo.LoadBalancerSourceRanges() { // 写入限制src的防火墙规则 proxier.natRules.Write(args, "-s", src, "-j", string(externalTrafficChain)) _, cidr, err := netutils.ParseCIDRSloppy(src) // 检查是否允许来自本节点的流量 if cidr.Contains(proxier.nodeIP) { allowFromNode = true } } // 来自lbip的流量劫持到kube-ext-xxx链 if allowFromNode { proxier.natRules.Write( args, "-s", lbip, "-j", string(externalTrafficChain)) } // 其它流量转到kube-mark-drop标记链用于丢弃 proxier.natRules.Write(args, "-j", string(KubeMarkDropChain)) } } // service无可用endpoint,拒绝lbIP的流量 } else { // No endpoints. for _, lbip := range svcInfo.LoadBalancerIPStrings() { proxier.filterRules.Write( "-A", string(kubeExternalServicesChain), "-m", "comment", "--comment", fmt.Sprintf("%s has no endpoints", svcNameString), "-m", protocol, "-p", protocol, "-d", lbip, "--dport", strconv.Itoa(svcInfo.Port()), "-j", "REJECT", ) } } // nodeport流量劫持 if svcInfo.NodePort() != 0 && len(nodeAddresses) != 0 { // nodeport流量转到kube-ext-xxx链 if hasEndpoints { proxier.natRules.Write( "-A", string(kubeNodePortsChain), "-m", "comment", "--comment", svcNameString, "-m", protocol, "-p", protocol, "--dport", strconv.Itoa(svcInfo.NodePort()), "-j", string(externalTrafficChain)) // service无可用endpoint,拒绝nodeport流量 } else { // No endpoints. proxier.filterRules.Write( "-A", string(kubeExternalServicesChain), "-m", "comment", "--comment", fmt.Sprintf("%s has no endpoints", svcNameString), "-m", "addrtype", "--dst-type", "LOCAL", "-m", protocol, "-p", protocol, "--dport", strconv.Itoa(svcInfo.NodePort()), "-j", "REJECT", ) } } // filter表healthcheck端口规则(kube-nodeport-chains) if svcInfo.HealthCheckNodePort() != 0 { // no matter if node has local endpoints, healthCheckNodePorts // need to add a rule to accept the incoming connection proxier.filterRules.Write( "-A", string(kubeNodePortsChain), "-m", "comment", "--comment", fmt.Sprintf("%s health check node port", svcNameString), "-m", "tcp", "-p", "tcp", "--dport", strconv.Itoa(svcInfo.HealthCheckNodePort()), "-j", "ACCEPT", ) } // kube-svc-xxx链规则更新 if svcInfo.UsesClusterEndpoints() { // Write rules jumping from clusterPolicyChain to clusterEndpoints proxier.writeServiceToEndpointRules(svcNameString, svcInfo, clusterPolicyChain, clusterEndpoints, args) } // kube-svl-xxx规则更新 if svcInfo.UsesLocalEndpoints() { if len(localEndpoints) != 0 { // Write rules jumping from localPolicyChain to localEndpointChains proxier.writeServiceToEndpointRules(svcNameString, svcInfo, localPolicyChain, localEndpoints, args) } else { // serice关联endpoint但不再本节点,拒绝kube-svl流量 if hasEndpoints { // Blackhole all traffic since there are no local endpoints args = append(args[:0], "-A", string(localPolicyChain), "-m", "comment", "--comment", fmt.Sprintf("%s has no local endpoints", svcNameString), "-j", string(KubeMarkDropChain), ) proxier.natRules.Write(args) } } } } // kube-proxy不活跃的链删除 for chain := range existingNATChains { // 非活跃链 if !activeNATChains[chain] { chainString := string(chain) // 非kube-proxy链跳过 if !isServiceChainName(chainString) { continue } // 重新写入链(相当于清空规则) proxier.natChains.WriteBytes(existingNATChains[chain]) // 删除链 proxier.natRules.Write("-X", chainString) } } // kube-services转向kube-nodeports流量规则 for address := range nodeAddresses { // cidr为零地址,直接将转向本节点的流量由kube-service-chains转到kube-nodeport-chains if utilproxy.IsZeroCIDR(address) { proxier.natRules.Write( "-A", string(kubeServicesChain), "-m", "comment", "--comment", "kubernetes service nodeports; NOTE: this must be the last rule in this chain", "-m", "addrtype", "--dst-type", "LOCAL", "-j", string(kubeNodePortsChain)) // Nothing else matters after the zero CIDR. break } ... // 将转向本节点的流量,哪些dst可以由kube-services转到kube-nodeports proxier.natRules.Write( "-A", string(kubeServicesChain), "-m", "comment", "--comment", "kubernetes service nodeports; NOTE: this must be the last rule in this chain", "-d", address, "-j", string(kubeNodePortsChain)) } // filter表kube-forward-chain规则 // 状态无效的包丢弃 proxier.filterRules.Write( "-A", string(kubeForwardChain), "-m", "conntrack", "--ctstate", "INVALID", "-j", "DROP", ) // 标记0x4000的流量允许通过kube-forward链 proxier.filterRules.Write( "-A", string(kubeForwardChain), "-m", "comment", "--comment", "kubernetes forwarding rules", "-m", "mark", "--mark", fmt.Sprintf("%s/%s", proxier.masqueradeMark, proxier.masqueradeMark), "-j", "ACCEPT", ) // 允许已建立连接的数据包通过,不再从新匹配mark标记 proxier.filterRules.Write( "-A", string(kubeForwardChain), "-m", "comment", "--comment", "kubernetes forwarding conntrack rule", "-m", "conntrack", "--ctstate", "RELATED,ESTABLISHED", "-j", "ACCEPT", ) // 同步filter表和nat表规则 proxier.iptablesData.Reset() proxier.iptablesData.WriteString("*filter\n") proxier.iptablesData.Write(proxier.filterChains.Bytes()) proxier.iptablesData.Write(proxier.filterRules.Bytes()) proxier.iptablesData.WriteString("COMMIT\n") proxier.iptablesData.WriteString("*nat\n") proxier.iptablesData.Write(proxier.natChains.Bytes()) proxier.iptablesData.Write(proxier.natRules.Bytes()) proxier.iptablesData.WriteString("COMMIT\n") // 规则写入内核iptables // iptables-restore --noflush --counters < xxxx err = proxier.iptables.RestoreAll(proxier.iptablesData.Bytes(), utiliptables.NoFlushTables, utiliptables.RestoreCounters) if err != nil { ... return } success = true ... // 同步需要健康检查的services proxier.serviceHealthServer.SyncServices(serviceUpdateResult.HCServiceNodePorts) ... // 同步需要健康检查的endpoints proxier.serviceHealthServer.SyncEndpoints(endpointUpdateResult.HCEndpointsLocalIPSize) ... // udp serviceIP旧conntrack条目清理(service规则) // conntrack -D -p udp --orig-dst 10.96.0.1 (-f inet6) for _, svcIP := range conntrackCleanupServiceIPs.UnsortedList() { conntrack.ClearEntriesForIP(proxier.exec, svcIP, v1.ProtocolUDP) ... } // udp nodeport旧conntrack清理(service规则) // conntrack -D -p udp --dport 30000 (-f inet6) for _, nodePort := range conntrackCleanupServiceNodePorts.UnsortedList() { conntrack.ClearEntriesForPort(proxier.exec, nodePort, isIPv6, v1.ProtocolUDP) ... } // udp ready endpoint删除/udp endpoint由无ready转为存在ready,需清理旧的conntrack,避免网络连接状态缓存影响流量 // conntrack -D -p udp --dport 30001 --dst-nat 10.0.0.5 // 1.清理endpoint nodeport对应的nat条目 // 2.清理clusterIP的nat条目 // 3.清理externalIP的nat条目 // 4.清理lbIP的nat条目 proxier.deleteEndpointConnections(endpointUpdateResult.StaleEndpoints) }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
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
注意
1.
kubeproxy会初始化扩展链，针对扩展链添加规则进行网络地址转换及路由，扩展链顶链写入原有链规则进行数据包劫持2.
endpoint对应的链规则dst为pod，具体的跨节点转发依赖cni