Iptables-mode Proxier

概述

kube-Proxy提供三种模式(userspace/iptables/ipvs)的proxier实现,userspace是早期的proxy模式，ipvs模式处于实验性阶段proxy模式，本文先从默认的内核级iptables proxier代码实现与逻辑分析开始，其它模式将用专文解析源码。

Iptables-mode Proxier的service配置和代码内都包含一些基础概念如clusterIP、nodeport、loadbalancer、Ingress、ClusterCIDR、onlyLocal、ExternalIP等，请在了解源码之前先熟悉其概念用途场景与类型区别，再看源码将对你理解proxy事半功倍。当然也需要对netfilter、iptables、connTrack等proxy基础依赖的工具熟悉。基础概念部分在本文将不深入介绍，有需求可自行查阅相关资料。

从kube-proxy组件整体框架层的代码来看，在ProxyServer.Run()最后走到了s.Proxier.SyncLoop()执行空间一直无限loop下去。而默认的ProxyServer配置的Proxier对象就是Iptables(if proxyMode == proxyModeIPTables)，将调用iptabls-mode的Proxier.SyncLoop()，SyncLoop()时间定时循环执行syncProxyRules()完成services、endpoints与iptables规则的同步操作。

Iptables-mode proxier的负载均衡机制是通过底层netfliter/iptables规则来实现的，通过Informer机制watch服务与端点信息的变更事件触发对iptables的规则的同步更新，如下代码逻辑示意图：

iptables-proxier

下面proxier源码分析,我们先从proxier的接口、实现类、实现类方法列表一窥究竟，从结构上看整体Proxier的框架。然后我们再详细分析proxier对象的产生时所定义的属性值、值类型和用途。有了前面的两项的了解后我们再来分析proxier类方法的实现，也就是proxier代理逻辑部分(关键逻辑部分在syncProxyRules()方法分析部分)。最后我们分析proxier底层内核iptables的runner实现，也就是proxy上层逻辑层最终会调用iptables命令去执行规则的操作部分。

Proxier 数据结构与类定义

ProxyProvider 代理提供者接口定义，需要实现两个proxy的关键方法Sync()和SyncLoop()

!FILENAME pkg/proxy/types.go:27

  1. type ProxyProvider interface {
  2.   // Sync 即时同步Proxy提供者的当前状态至proxy规则
  3.     Sync()
  4.     // SyncLoop 周期性运行
  5.   // 作为一个线程或应用主loop运行，无返回.
  6.     SyncLoop()
  7. }

Iptables-mode Proxier 为 ProxyProvider 接口实现类，proxier 类属性项比较多，我们先看一下注释用途与结构定义，在实例化proxier对象时我们再详看。

!FILENAME pkg/proxy/iptables/proxier.go:205

  1. type Proxier struct {
  2.     endpointsChanges *proxy.EndpointChangeTracker // 端点更新信息跟踪器
  3.     serviceChanges   *proxy.ServiceChangeTracker  // 服务更新信息跟踪器
  5.     mu           sync.Mutex                       // 保护同步锁
  6.     serviceMap   proxy.ServiceMap                 // 存放服务列表信息                ①
  7.     endpointsMap proxy.EndpointsMap               // 存放端点列表信息                ②
  8.     portsMap     map[utilproxy.LocalPort]utilproxy.Closeable //端口关闭接口map
  10.     endpointsSynced bool                          // ep同步状态
  11.     servicesSynced  bool                          // svc同步状态
  12.     initialized     int32                         // 初始化状态
  13.     syncRunner      *async.BoundedFrequencyRunner // 指定频率运行器，此处用于管理对 
  14.                                                 // syncProxyRules的调用
  16.     iptables       utiliptables.Interface         // iptables命令执行接口
  17.     masqueradeAll  bool                          
  18.     masqueradeMark string                         // SNAT地址伪装Mark
  19.     exec           utilexec.Interface             // exec命令执行工具接口
  20.     clusterCIDR    string                         // 集群CIDR
  21.     hostname       string                         // 主机名
  22.     nodeIP         net.IP                         // 节点IP地址
  23.     portMapper     utilproxy.PortOpener           // TCP/UTP端口打开与监听
  24.     recorder       record.EventRecorder           // 事件记录器
  25.     healthChecker  healthcheck.Server             // healthcheck服务器对象
  26.     healthzServer  healthcheck.HealthzUpdater     // Healthz更新器
  28.     precomputedProbabilities []string             //预计算可能性
  30.   //iptables规则与链数据(Filter/NAT)
  31.     iptablesData             *bytes.Buffer        
  32.     existingFilterChainsData *bytes.Buffer        
  33.     filterChains             *bytes.Buffer         
  34.     filterRules              *bytes.Buffer           
  35.     natChains                *bytes.Buffer        
  36.     natRules                 *bytes.Buffer      
  38.     endpointChainsNumber int                       
  40.     // Node节点IP与端口信息
  41.     nodePortAddresses []string                    
  43.     networkInterfacer utilproxy.NetworkInterfacer  //网络接口
  44. }

① ServiceMap和ServicePort定义

!FILENAME pkg/proxy/service.go:229

  1. type ServiceMap map[ServicePortName]ServicePort
  3. //String() =>  "NS/SvcName:PortName"
  4. ServicePortName{NamespacedName: svcName, Port: servicePort.Name}

ServiceSpec service.spec定义，在用户前端可定义service的spec配置项。

!FILENAME vendor/k8s.io/api/core/v1/types.go:3606

  1. type ServiceSpec struct {
  2.     Ports []ServicePort                   //服务端口列表
  3.     Selector map[string]string            //选择器
  4.     ClusterIP string                      //VIP 、 portal
  5.     Type ServiceType                      //服务类型   
  6.     ExternalIPs []string                  //外部IP列表，如外部负载均衡
  7.     SessionAffinity ServiceAffinity       //会话保持
  8.     LoadBalancerIP string                 //service类型为"LoadBalancer"时配置LB ip
  9.     LoadBalancerSourceRanges []string     //cloud-provider的限制client ip区间
  10.     ExternalName string 
  11.     ExternalTrafficPolicy ServiceExternalTrafficPolicyType 
  12.     HealthCheckNodePort int32              
  13.     PublishNotReadyAddresses bool 
  14.     SessionAffinityConfig *SessionAffinityConfig  //会话保持配置信息
  15. }

ServicePort类定义和ServicePort接口

!FILENAME vendor/k8s.io/api/core/v1/types.go:3563

  1. type ServicePort struct {
  2.     Name string 
  3.     Protocol Protocol 
  4.     Port int32 
  5.     TargetPort intstr.IntOrString 
  6.     NodePort int32 
  7. }
  9. //ServicePort接口
  10. type ServicePort interface {
  11.     // 返回服务字串，格式如: `IP:Port/Protocol`.
  12.     String() string
  13.     // 返回集群IP字串
  14.     ClusterIPString() string
  15.     // 返回协议
  16.     GetProtocol() v1.Protocol
  17.     // 返回健康检测端口
  18.     GetHealthCheckNodePort() int
  19. }

② EndpointsMap定义与Endpoint接口

!FILENAME pkg/proxy/endpoints.go:181

  1. type EndpointsMap map[ServicePortName][]Endpoint
  3. type Endpoint interface {
  4.     // 返回endpoint字串，格式 `IP:Port`.
  5.     String() string
  6.   // 是否本地
  7.     GetIsLocal() bool
  8.     // 返回IP
  9.     IP() string
  10.     // 返回端口
  11.     Port() (int, error)
  12.     // 检测两上endpoint是否相等
  13.     Equal(Endpoint) bool
  14. }

Endpoints结构与相关定义

!FILENAME vendor/k8s.io/api/core/v1/types.go:3710

  1. type Endpoints struct {
  2.     metav1.TypeMeta 
  3.     metav1.ObjectMeta 
  4.     Subsets []EndpointSubset  
  5. }
  7. type EndpointSubset struct {
  8.     Addresses []EndpointAddress            // EndpointAddress地址列表
  9.     NotReadyAddresses []EndpointAddress    
  10.     Ports []EndpointPort                   // EndpointPort端口列表
  11. }
  13. type EndpointAddress struct {           
  14.     IP string 
  15.     Hostname string 
  16.     NodeName *string
  17.     TargetRef *ObjectReference 
  18. }
  20. type EndpointPort struct {
  21.     Name string 
  22.     Port int32 
  23.     Protocol Protocol 
  24. }

Iptables-mode Proxier提供的方法列表，先大概从名称上来了解一下方法用途，后面我在逻辑部分对主要使用的方法再深入分析。

  1. func (proxier *Proxier) precomputeProbabilities(numberOfPrecomputed int) {/*...*/}
  2. func (proxier *Proxier) probability(n int) string{/*...*/}
  3. func (proxier *Proxier) Sync(){/*...*/}
  4. func (proxier *Proxier) SyncLoop(){/*...*/}
  5. func (proxier *Proxier) setInitialized(value bool){/*...*/}
  6. func (proxier *Proxier) isInitialized() bool{/*...*/}
  7. func (proxier *Proxier) OnServiceAdd(service *v1.Service){/*...*/}
  8. func (proxier *Proxier) OnServiceUpdate(oldService, service *v1.Service){/*...*/}
  9. func (proxier *Proxier) OnServiceDelete(service *v1.Service){/*...*/}
  10. func (proxier *Proxier) OnServiceSynced(){/*...*/}
  11. func (proxier *Proxier) OnEndpointsAdd(endpoints *v1.Endpoints){/*...*/}
  12. func (proxier *Proxier) OnEndpointsUpdate(oldEndpoints, endpoints *v1.Endpoints){/*...*/}
  13. func (proxier *Proxier) OnEndpointsDelete(endpoints *v1.Endpoints) {/*...*/}
  14. func (proxier *Proxier) OnEndpointsSynced() {/*...*/}
  15. func (proxier *Proxier) deleteEndpointConnections(connectionMap []proxy.ServiceEndpoint){/*...*/}
  16. func (proxier *Proxier) appendServiceCommentLocked(args []string, svcName string){/*...*/}
  17. func (proxier *Proxier) syncProxyRules(){/*...*/}

Proxier对象生成与运行

iptables Proxier 构建New方法，下面省略部分校验的代码，关注关键构造部分。

!FILENAME pkg/proxy/iptables/proxier.go:281

  1. func NewProxier(ipt utiliptables.Interface,
  2.     sysctl utilsysctl.Interface,
  3.     exec utilexec.Interface,
  4.     syncPeriod time.Duration,
  5.     minSyncPeriod time.Duration,
  6.     masqueradeAll bool,
  7.     masqueradeBit int,
  8.     clusterCIDR string,
  9.     hostname string,
  10.     nodeIP net.IP,
  11.     recorder record.EventRecorder,
  12.     healthzServer healthcheck.HealthzUpdater,
  13.     nodePortAddresses []string,
  14. ) (*Proxier, error) {
  16.     // ...以下为省略部分解析...
  17.   // sysctl对"net/ipv4/conf/all/route_localnet"设置 ，内核iptables支持   
  18.     // sysctl对"net/bridge/bridge-nf-call-iptables"设置，内核iptables支持
  20.   // 生产SNAT的IP伪装mark
  21.   // 如果节点IP为空，则kube-proxy的nodeIP的初始IP为127.0.0.1
  22.   // 集群CIDR检验是否为空，IPv6验证
  23.   // ...
  25.   //healthcheck服务器对象
  26.     healthChecker := healthcheck.NewServer(hostname, recorder, nil, nil) 
  28.   //proxier对象
  29.     proxier := &Proxier{
  30.         portsMap:                 make(map[utilproxy.LocalPort]utilproxy.Closeable),
  31.         serviceMap:               make(proxy.ServiceMap),   //svc存放map
  32.         serviceChanges:           proxy.NewServiceChangeTracker(newServiceInfo, &isIPv6, recorder),                                              //svc变化跟踪器
  33.         endpointsMap:             make(proxy.EndpointsMap), //ep存放map
  34.         endpointsChanges:         proxy.NewEndpointChangeTracker(hostname, newEndpointInfo, &isIPv6, recorder),                                     //ep变化跟踪器
  35.         iptables:                 ipt,
  36.         masqueradeAll:            masqueradeAll,
  37.         masqueradeMark:           masqueradeMark,
  38.         exec:                     exec,
  39.         clusterCIDR:              clusterCIDR,
  40.         hostname:                 hostname,
  41.         nodeIP:                   nodeIP,
  42.         portMapper:               &listenPortOpener{},       //服务端口监听器
  43.         recorder:                 recorder,
  44.         healthChecker:            healthChecker,
  45.         healthzServer:            healthzServer,
  46.         precomputedProbabilities: make([]string, 0, 1001),
  47.         iptablesData:             bytes.NewBuffer(nil),      //iptables配置规则数据
  48.         existingFilterChainsData: bytes.NewBuffer(nil),
  49.         filterChains:             bytes.NewBuffer(nil),
  50.         filterRules:              bytes.NewBuffer(nil),
  51.         natChains:                bytes.NewBuffer(nil),
  52.         natRules:                 bytes.NewBuffer(nil),
  53.         nodePortAddresses:        nodePortAddresses,
  54.         networkInterfacer:        utilproxy.RealNetwork{},   //网络接口
  55.     }
  56.     burstSyncs := 2
  57.     klog.V(3).Infof("minSyncPeriod: %v, syncPeriod: %v, burstSyncs: %d", minSyncPeriod, syncPeriod, burstSyncs)
  58.   //运行器执行指定频率对syncProxyRules调用来同步规则,关键的逻辑则在syncProxyRules()方法内，后面有详述方法                               
  59.     proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs)  
  60.     return proxier, nil
  61. }

proxier.SyncLoop() 我们知道proxy server的运行时最后的调用就是”s.Proxier.SyncLoop()”,此处我们来详细了解一下SyncLoop的proxier运行实现。

!FILENAME pkg/proxy/iptables/proxier.go:487

  1. func (proxier *Proxier) SyncLoop() {
  2.     if proxier.healthzServer != nil {
  3.         proxier.healthzServer.UpdateTimestamp()
  4.     }
  5.   // proxier.syncRunner在proxier对象创建时指定为async.NewBoundedFrequencyRunner(...)     ① 
  6.     proxier.syncRunner.Loop(wait.NeverStop) 
  7. }

① async.BoundedFrequencyRunner时间器循环func执行器。

proxier类结构内定义的proxier.syncRunner 类型为async.BoundedFrequencyRunner

!FILENAME pkg/util/async/bounded_frequency_runner.go:31

  1. type BoundedFrequencyRunner struct {
  2.     name        string        
  3.     minInterval time.Duration // 两次运行的最小间隔时间
  4.     maxInterval time.Duration // 两次运行的最大间隔时间
  6.     run chan struct{}         // 执行一次run
  8.     mu      sync.Mutex  
  9.     fn      func()            // 需要运行的func
  10.     lastRun time.Time         // 最近一次运行时间
  11.     timer   timer             // 定时器
  12.     limiter rateLimiter       // 按需限制运行的QPS
  13. }

BoundedFrequencyRunner实例化构建，通过传参按需来控制对func的调用。

!FILENAME pkg/util/async/bounded_frequency_runner.go:134

  1. func NewBoundedFrequencyRunner(name string, fn func(), minInterval, maxInterval time.Duration, burstRuns int) *BoundedFrequencyRunner {
  2.     timer := realTimer{Timer: time.NewTimer(0)}       // 立即tick
  3.     <-timer.C()                                       // 消费第一次tick,完成一执行run
  4.     return construct(name, fn, minInterval, maxInterval, burstRuns, timer)  // ->
  5. }
  7. //实例构建
  8. func construct(name string, fn func(), minInterval, maxInterval time.Duration, burstRuns int, timer timer) *BoundedFrequencyRunner {
  9.     if maxInterval < minInterval {
  10.         panic(fmt.Sprintf("%s: maxInterval (%v) must be >= minInterval (%v)", name, minInterval, maxInterval))
  11.     }
  12.     if timer == nil {
  13.         panic(fmt.Sprintf("%s: timer must be non-nil", name))
  14.     }
  16.     bfr := &BoundedFrequencyRunner{
  17.         name:        name,
  18.         fn:          fn,                               //被调用处理的func
  19.         minInterval: minInterval,
  20.         maxInterval: maxInterval,
  21.         run:         make(chan struct{}, 1),
  22.         timer:       timer,
  23.     }
  24.     if minInterval == 0 {                            //最小间隔时间如果不指定，将不受限制
  25.         bfr.limiter = nullLimiter{}
  26.     } else {
  27.         // TokenBucketRateLimiter的实现流控机制，有兴趣可以再深入了解机制，此处不展开
  28.         qps := float32(time.Second) / float32(minInterval)
  29.         bfr.limiter = flowcontrol.NewTokenBucketRateLimiterWithClock(qps, burstRuns, timer)
  30.     }
  31.     return bfr
  32. }

proxier.syncRunner.Loop() 时间器循环运行的实现

  1. func (bfr *BoundedFrequencyRunner) Loop(stop <-chan struct{}) {
  2.     klog.V(3).Infof("%s Loop running", bfr.name)
  3.     bfr.timer.Reset(bfr.maxInterval)
  4.     for {
  5.         select {
  6.         case <-stop:             // 停止channel实现关闭loop
  7.             bfr.stop()
  8.             klog.V(3).Infof("%s Loop stopping", bfr.name)
  9.             return
  10.         case <-bfr.timer.C():    //定时器tick运行
  11.             bfr.tryRun()           
  12.         case <-bfr.run:          // 执行一次运行
  13.             bfr.tryRun()
  14.         }
  15.     }
  16. }

BoundedFrequencyRunner.tryRun() 按指定频率对func的运行

!FILENAME:pkg/util/async/bounded_frequency_runner.go:211

  1. func (bfr *BoundedFrequencyRunner) tryRun() {
  2.     bfr.mu.Lock()
  3.     defer bfr.mu.Unlock()
  5.   //限制条件允许运行func
  6.     if bfr.limiter.TryAccept() {
  7.         bfr.fn()                                  // 调用func
  8.         bfr.lastRun = bfr.timer.Now()             // 记录运行时间
  9.         bfr.timer.Stop()                          
  10.         bfr.timer.Reset(bfr.maxInterval)          // 重设下次运行时间
  11.         klog.V(3).Infof("%s: ran, next possible in %v, periodic in %v", bfr.name, bfr.minInterval, bfr.maxInterval)
  12.         return
  13.     }
  15.   //限制条件不允许运行，计算下次运行时间
  16.   elapsed := bfr.timer.Since(bfr.lastRun)    // elapsed:上次运行时间到现在已过多久
  17.   nextPossible := bfr.minInterval - elapsed  // nextPossible:下次运行至少差多久（最小周期）
  18.   nextScheduled := bfr.maxInterval - elapsed // nextScheduled:下次运行最迟差多久(最大周期)
  19.     klog.V(4).Infof("%s: %v since last run, possible in %v, scheduled in %v", bfr.name, elapsed, nextPossible, nextScheduled)
  21.     if nextPossible < nextScheduled {
  22.         // Set the timer for ASAP, but don't drain here.  Assuming Loop is running,
  23.         // it might get a delivery in the mean time, but that is OK.
  24.         bfr.timer.Stop()
  25.         bfr.timer.Reset(nextPossible)
  26.         klog.V(3).Infof("%s: throttled, scheduling run in %v", bfr.name, nextPossible)
  27.     }
  28. }

Proxier 服务与端点更新 Tracker

kube-proxy需要及时同步services和endpoints的变化信息,前面我们看到proxier类对象有两个属性：serviceChanges和endpointsChanges是就是用来跟踪Service和Endpoint的更新信息，我们先来分析与之相关这两个类ServiceChangeTracker和EndpointChangeTracker。

ServiceChangeTracker 服务信息变更Tracker

!FILENAME pkg/proxy/service.go:143

  1. type ServiceChangeTracker struct {
  2.     // 同步锁保护items
  3.     lock sync.Mutex
  4.     // items为service变化记录map
  5.   items map[types.NamespacedName]*serviceChange
  6.     // makeServiceInfo允许proxier在处理服务时定制信息
  7.     makeServiceInfo makeServicePortFunc    
  8.     isIPv6Mode *bool                           //IPv6
  9.     recorder   record.EventRecorder            //事件记录器 
  10. }
  12. //serviceChange类型定义， <previous, current> 新旧服务对.
  13. type serviceChange struct {
  14.     previous ServiceMap
  15.     current  ServiceMap
  16. }
  17. //ServiceMap类型定义
  18. type ServiceMap map[ServicePortName]ServicePort
  20. //ServicePortName类型定义
  21. type ServicePortName struct {
  22.     types.NamespacedName
  23.     Port string
  24. }
  25. //NamespacedName类型定义
  26. type NamespacedName struct {
  27.     Namespace string
  28.     Name      string
  29. }
  31. //实例化ServiceChangeTracker对象
  32. func NewServiceChangeTracker(makeServiceInfo makeServicePortFunc, isIPv6Mode *bool, recorder record.EventRecorder) *ServiceChangeTracker {
  33.     return &ServiceChangeTracker{
  34.         items:           make(map[types.NamespacedName]*serviceChange),
  35.         makeServiceInfo: makeServiceInfo,
  36.         isIPv6Mode:      isIPv6Mode,
  37.         recorder:        recorder,
  38.     }
  39. }

EndpointChangeTracker.Update()

!FILENAME pkg/proxy/service.go:173

  1. func (ect *EndpointChangeTracker) Update(previous, current *v1.Endpoints) bool {
  2.     endpoints := current
  3.     if endpoints == nil {
  4.         endpoints = previous
  5.     }
  6.     // previous == nil && current == nil is unexpected, we should return false directly.
  7.     if endpoints == nil {
  8.         return false
  9.     }
  10.     namespacedName := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}
  12.     ect.lock.Lock()
  13.     defer ect.lock.Unlock()
  15.     change, exists := ect.items[namespacedName]
  16.     if !exists {
  17.         change = &endpointsChange{}
  18.         change.previous = ect.endpointsToEndpointsMap(previous)
  19.         ect.items[namespacedName] = change
  20.     }
  21.     change.current = ect.endpointsToEndpointsMap(current)
  22.     // if change.previous equal to change.current, it means no change
  23.     if reflect.DeepEqual(change.previous, change.current) {
  24.         delete(ect.items, namespacedName)
  25.     }
  26.     return len(ect.items) > 0
  27. }

EndpointChangeTracker 端点信息变更Tracker

!FILENAME pkg/proxy/endpoints.go:83

  1. type EndpointChangeTracker struct {
  2.     lock sync.Mutex
  3.     // kube-proxy运行的主机名.
  4.     hostname string
  5.     // items为"endpoints"变化记录map
  6.     items map[types.NamespacedName]*endpointsChange
  7.     makeEndpointInfo makeEndpointFunc
  8.     isIPv6Mode *bool
  9.     recorder   record.EventRecorder
  10. }
  11. //endpointsChange类型定义，<previous, current> 新旧端点对
  12. type endpointsChange struct {
  13.     previous EndpointsMap
  14.     current  EndpointsMap
  15. }
  16. //EndpointsMap类型定义
  17. type EndpointsMap map[ServicePortName][]Endpoint
  19. //Endpoint接口
  20. type Endpoint interface {
  21.     // 返回endpoint字串 如格式: `IP:Port`.
  22.     String() string
  23.     // 是否本地主机
  24.     GetIsLocal() bool
  25.     // 返回endpoint的IP部分
  26.     IP() string
  27.     // 返回endpoint的Port部分
  28.     Port() (int, error)
  29.     // 计算两个endpoint是否相等
  30.     Equal(Endpoint) bool
  31. }
  33. //实例化NewEndpointChangeTracker对象
  34. func NewEndpointChangeTracker(hostname string, makeEndpointInfo makeEndpointFunc, isIPv6Mode *bool, recorder record.EventRecorder) *EndpointChangeTracker {
  35.     return &EndpointChangeTracker{
  36.         hostname:         hostname,
  37.         items:            make(map[types.NamespacedName]*endpointsChange),
  38.         makeEndpointInfo: makeEndpointInfo,
  39.         isIPv6Mode:       isIPv6Mode,
  40.         recorder:         recorder,
  41.     }
  42. }

EndpointChangeTracker.Update()

!FILENAME pkg/proxy/endpoints.go:116

  1. func (ect *EndpointChangeTracker) Update(previous, current *v1.Endpoints) bool {
  2.     endpoints := current
  3.     if endpoints == nil {
  4.         endpoints = previous
  5.     }
  6.     // previous == nil && current == nil is unexpected, we should return false directly.
  7.     if endpoints == nil {
  8.         return false
  9.     }
  10.     namespacedName := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}
  12.     ect.lock.Lock()
  13.     defer ect.lock.Unlock()
  15.     change, exists := ect.items[namespacedName]
  16.     if !exists {
  17.         change = &endpointsChange{}
  18.         change.previous = ect.endpointsToEndpointsMap(previous)
  19.         ect.items[namespacedName] = change
  20.     }
  21.     change.current = ect.endpointsToEndpointsMap(current)
  22.     // if change.previous equal to change.current, it means no change
  23.     if reflect.DeepEqual(change.previous, change.current) {
  24.         delete(ect.items, namespacedName)
  25.     }
  26.     return len(ect.items) > 0
  27. }

Proxier服务与端点同步（Add / delete /update）都将调用ChangeTracker的update()来执行syncProxyRules

Proxier.OnServiceSynced()服务信息同步

!FILENAME pkg/proxy/iptables/proxier.go:528

  1. func (proxier *Proxier) OnServiceSynced() {
  2.     proxier.mu.Lock()
  3.     proxier.servicesSynced = true
  4.     proxier.setInitialized(proxier.servicesSynced && proxier.endpointsSynced)
  5.     proxier.mu.Unlock()
  7.     // 非周期同步，单次执行同步
  8.     proxier.syncProxyRules()
  9. }

service服务更新(add/delete/update),都调用Proxier.OnServiceUpdate()方法

!FILENAME pkg/proxy/iptables/proxier.go:513

  1. func (proxier *Proxier) OnServiceAdd(service *v1.Service) {
  2.   proxier.OnServiceUpdate(nil, service)    //传参一:oldService为nil
  3. }
  5. // update Service 包含Add / Delete
  6. func (proxier *Proxier) OnServiceUpdate(oldService, service *v1.Service) {
  7.     if proxier.serviceChanges.Update(oldService, service) && proxier.isInitialized() {
  8.         proxier.syncRunner.Run()     //单次执行
  9.     }
  10. }
  12. func (proxier *Proxier) OnServiceDelete(service *v1.Service) {
  13.   proxier.OnServiceUpdate(service, nil)   //传参二:newService为nil
  14. }

Proxier.OnEndpointsSynced()端点信息同步

!FILENAME pkg/proxy/iptables/proxier.go:552

  1. func (proxier *Proxier) OnEndpointsSynced() {
  2.     proxier.mu.Lock()
  3.     proxier.endpointsSynced = true
  4.     proxier.setInitialized(proxier.servicesSynced && proxier.endpointsSynced)
  5.     proxier.mu.Unlock()
  7.     // Sync unconditionally - this is called once per lifetime.
  8.     proxier.syncProxyRules()
  9. }

同service服务一样endpoint更新(add/delete/update),都调用Proxier.OnEndpointsUpdate()方法

!FILENAME pkg/proxy/iptables/proxier.go:538

  1. func (proxier *Proxier) OnEndpointsAdd(endpoints *v1.Endpoints) {
  2.   proxier.OnEndpointsUpdate(nil, endpoints)       //传参一:oldEndpoints为nil
  3. }
  5. // update endpoints，包含 Add / Delete
  6. func (proxier *Proxier) OnEndpointsUpdate(oldEndpoints, endpoints *v1.Endpoints) {
  7.     if proxier.endpointsChanges.Update(oldEndpoints, endpoints) && proxier.isInitialized() {
  8.         proxier.syncRunner.Run()
  9.     }
  10. }
  12. func (proxier *Proxier) OnEndpointsDelete(endpoints *v1.Endpoints) {
  13.   proxier.OnEndpointsUpdate(endpoints, nil)    //传参二:endpoints为nil
  14. }

syncProxyRule 同步配置与规则

proxier.syncProxyRules() 实现监听svc或ep更新配置到iptables规则的一致性同步机制功能，这也是iptables proxer最核心的逻辑代码。作者实现是利用了iptables-save/iptables-restore机制将现存的iptables配置和服务与端点同步的信息来生成相对应的iptables链与规则数据，每次同步执行写入可restore标准格式的规则数据后通过iptables-restore命令进行重设iptables规则。

这个同步规则处理代码比较长，我们后面将分解成小块来讲解。下面为syncProxyRules(部分已解析注释替代)代码框架说明，内注释的每块内容在后面都将有单独代码分析说明。

!FILENAME pkg/proxy/iptables/proxier.go:634

  1. func (proxier *Proxier) syncProxyRules() {
  2.   proxier.mu.Lock()
  3.     defer proxier.mu.Unlock()
  5.     start := time.Now()
  6.     defer func() {
  7.         metrics.SyncProxyRulesLatency.Observe(metrics.SinceInMicroseconds(start))
  8.         klog.V(4).Infof("syncProxyRules took %v", time.Since(start))
  9.     }()
  10.     // don't sync rules till we've received services and endpoints
  11.     if !proxier.endpointsSynced || !proxier.servicesSynced {
  12.         klog.V(2).Info("Not syncing iptables until Services and Endpoints have been received from master")
  13.         return
  14.     }
  16.   // 检测与更新变化(service/endpoints)
  17.   //...
  19.   // 创建与联接kube链
  20.   //...
  22.     // 获取现存在的Filter/Nat表链数据
  23.   //...
  26.   // 创建iptables-save/restore格式数据（表头、链）
  27.   //...
  29.   // 写kubernets特有的SNAT地址伪装规则
  30.   //...
  32.     // Accumulate NAT chains to keep.
  33.     activeNATChains := map[utiliptables.Chain]bool{} 
  34.     // Accumulate the set of local ports that we will be holding open once this update is complete
  35.     replacementPortsMap := map[utilproxy.LocalPort]utilproxy.Closeable{}
  37.     endpoints := make([]*endpointsInfo, 0)
  38.     endpointChains := make([]utiliptables.Chain, 0)
  40.     args := make([]string, 64)
  42.     // Compute total number of endpoint chains across all services.
  43.     proxier.endpointChainsNumber = 0
  44.     for svcName := range proxier.serviceMap {
  45.         proxier.endpointChainsNumber += len(proxier.endpointsMap[svcName])
  46.     }
  48.   // 为个"服务"创建rules（service portal规则的创建）
  49.     for svcName, svc := range proxier.serviceMap {
  50.      //...
  51.     }
  53.     // 删除不再使用的链
  54.   //...
  56.     // nodeports链
  57.   //...
  59.   // FORWARD策略
  60.   //...
  62.     // 配置clusterCIDR规则
  63.   //...
  65.     // 写结整标签
  66.   //...
  68.     //汇集与iptables-restore加载数据
  69.   //...
  71.   // 关闭过旧的本地端口，更新portmap数据
  72.     for k, v := range proxier.portsMap {
  73.         if replacementPortsMap[k] == nil {
  74.             v.Close()
  75.         }
  76.     }
  77.     proxier.portsMap = replacementPortsMap
  79.     // 更新healthz timestamp.
  80.     if proxier.healthzServer != nil {
  81.         proxier.healthzServer.UpdateTimestamp()
  82.     }
  84.     // 更新healthchecks.
  85.     if err := proxier.healthChecker.SyncServices(serviceUpdateResult.HCServiceNodePorts); err != nil {
  86.         klog.Errorf("Error syncing healthcheck services: %v", err)
  87.     }
  88.     if err := proxier.healthChecker.SyncEndpoints(endpointUpdateResult.HCEndpointsLocalIPSize); err != nil {
  89.         klog.Errorf("Error syncing healthcheck endpoints: %v", err)
  90.     }
  92.     // 完成清理工作
  93.     for _, svcIP := range staleServices.UnsortedList() {
  94.         if err := conntrack.ClearEntriesForIP(proxier.exec, svcIP, v1.ProtocolUDP); err != nil {
  95.             klog.Errorf("Failed to delete stale service IP %s connections, error: %v", svcIP, err)
  96.         }
  97.     }
  98.     proxier.deleteEndpointConnections(endpointUpdateResult.StaleEndpoints)
  99. }

下面将分解详述每块代码逻辑:

更新 service 和 endpoints ;返回更新结果

!FILENAME pkg/proxy/iptables/proxier.go:652

  1.     //更新SVC/EP
  2.   serviceUpdateResult := proxy.UpdateServiceMap(proxier.serviceMap, proxier.serviceChanges)
  3.     endpointUpdateResult := proxy.UpdateEndpointsMap(proxier.endpointsMap, proxier.endpointsChanges)
  5.     staleServices := serviceUpdateResult.UDPStaleClusterIP
  6.   // 从EndpointsMap更新结果返回中合并UDP协议废弃服务信息
  7.     for _, svcPortName := range endpointUpdateResult.StaleServiceNames {
  8.         if svcInfo, ok := proxier.serviceMap[svcPortName]; ok && svcInfo != nil && svcInfo.GetProtocol() == v1.ProtocolUDP {
  9.             klog.V(2).Infof("Stale udp service %v -> %s", svcPortName, svcInfo.ClusterIPString())
  10.             staleServices.Insert(svcInfo.ClusterIPString())
  11.         }
  12.     }

UpdateServiceMap() SVC 服务的更新实现

!FILENAME pkg/proxy/service.go:212

  1. func UpdateServiceMap(serviceMap ServiceMap, changes *ServiceChangeTracker) (result UpdateServiceMapResult) {
  2.     result.UDPStaleClusterIP = sets.NewString()   // 已废弃的UDP端口
  3.     serviceMap.apply(changes, result.UDPStaleClusterIP)  // 应用更新map->
  5.     result.HCServiceNodePorts = make(map[types.NamespacedName]uint16)
  6.     for svcPortName, info := range serviceMap {
  7.         if info.GetHealthCheckNodePort() != 0 {
  8.             result.HCServiceNodePorts[svcPortName.NamespacedName] = uint16(info.GetHealthCheckNodePort())    //健康检测的node Port
  9.         }
  10.     }
  12.     return result
  13. }

serviceMap.apply() 应用更新变化事件的服务项(merge->filter->unmerge)

!FILENAME pkg/proxy/service.go:268

  1. func (serviceMap *ServiceMap) apply(changes *ServiceChangeTracker, UDPStaleClusterIP sets.String) {
  2.     changes.lock.Lock()
  3.     defer changes.lock.Unlock()
  4.     for _, change := range changes.items {
  5.         serviceMap.merge(change.current)         //合并变更增加项到serviceMap              ①
  6.         change.previous.filter(change.current)   //过滤掉已处理更新变化的项，跳过unmerge处理  ②
  7.         serviceMap.unmerge(change.previous, UDPStaleClusterIP)  //删除废弃的项            ③
  8.     }
  9.     // clear changes after applying them to ServiceMap.
  10.     changes.items = make(map[types.NamespacedName]*serviceChange)
  11.     return
  12. }

① serviceMap.merge() 合并增加”other“内容到当前的serviceMap内。”即将变化的服务列表进行合并”。

!FILENAME pkg/proxy/service.go:301

  1. func (sm *ServiceMap) merge(other ServiceMap) sets.String {
  2.     existingPorts := sets.NewString()
  3.     for svcPortName, info := range other {
  4.         existingPorts.Insert(svcPortName.String())
  5.         _, exists := (*sm)[svcPortName]
  6.      //...
  7.         (*sm)[svcPortName] = info
  8.     }
  9.     return existingPorts
  10. }

② serviceMap.unmerge() 从当前map移除”other”存在的内容项。”即删除废弃的项“

!FILENAME pkg/proxy/service.go:330

  1. func (sm *ServiceMap) unmerge(other ServiceMap, UDPStaleClusterIP sets.String) {
  2.     for svcPortName := range other {
  3.         info, exists := (*sm)[svcPortName]
  4.         if exists {
  5.             if info.GetProtocol() == v1.ProtocolUDP {
  6.                 UDPStaleClusterIP.Insert(info.ClusterIPString())  //存储已废丢UDP服务的集群IP列表 
  7.             }
  8.             delete(*sm, svcPortName)
  9.         } //...
  10.     }
  11. }

③ serviceMap.filter() 基于”other”给定的服务端口名(key值)，过滤掉存在于serviceMap的项

!FILENAME pkg/proxy/service.go:319

  1. func (sm *ServiceMap) filter(other ServiceMap) {
  2.     for svcPortName := range *sm {
  3.         if _, ok := other[svcPortName]; ok {
  4.             delete(*sm, svcPortName)
  5.         }
  6.     }
  7. }

UpdateEndpointsMap() 端点更新的实现

!FILENAME pkg/proxy/endpoints.go:163

  1. func UpdateEndpointsMap(endpointsMap EndpointsMap, changes *EndpointChangeTracker) (result UpdateEndpointMapResult) {
  2.     result.StaleEndpoints = make([]ServiceEndpoint, 0)
  3.     result.StaleServiceNames = make([]ServicePortName, 0)
  5.     endpointsMap.apply(changes, &result.StaleEndpoints, &result.StaleServiceNames)
  7.     // TODO: If this will appear to be computationally expensive, consider
  8.     // computing this incrementally similarly to endpointsMap.
  9.     result.HCEndpointsLocalIPSize = make(map[types.NamespacedName]int)
  10.     localIPs := GetLocalEndpointIPs(endpointsMap)
  11.     for nsn, ips := range localIPs {
  12.         result.HCEndpointsLocalIPSize[nsn] = len(ips)
  13.     }
  15.     return result
  16. }

EndpointsMap.apply() 应用更新变化事件的端点项(merge->unmerge)

!FILENAME pkg/proxy/endpoints.go:242

  1. func (endpointsMap EndpointsMap) apply(changes *EndpointChangeTracker, staleEndpoints *[]ServiceEndpoint, staleServiceNames *[]ServicePortName) {
  2.     if changes == nil {
  3.         return
  4.     }
  5.     changes.lock.Lock()
  6.     defer changes.lock.Unlock()
  7.     for _, change := range changes.items {
  8.         endpointsMap.Unmerge(change.previous)                  // 删除                ①
  9.         endpointsMap.Merge(change.current)                     // 更新                ②
  10.         detectStaleConnections(change.previous, change.current, staleEndpoints, staleServiceNames)                                           //废弃查找              ③      
  11.     }
  12.     changes.items = make(map[types.NamespacedName]*endpointsChange)
  13. }

① EndpointsMap.Merge() 将”other”内指定项的值(端点列表)更新至EndpointMap

!FILENAME pkg/proxy/endpoints.go:259

  1. func (em EndpointsMap) Merge(other EndpointsMap) {
  2.     for svcPortName := range other {
  3.         em[svcPortName] = other[svcPortName]
  4.     }
  5. }

② EndpointsMap.Unmerge() 删除”other”内指定项

!FILENAME pkg/proxy/endpoints.go:266

  1. func (em EndpointsMap) Unmerge(other EndpointsMap) {
  2.     for svcPortName := range other {
  3.         delete(em, svcPortName)
  4.     }
  5. }

③ EndpointsMap.detectStaleConnections() 查找废弃后端连接信息项

!FILENAME pkg/proxy/endpoints.go:291

  1. func detectStaleConnections(oldEndpointsMap, newEndpointsMap EndpointsMap, staleEndpoints *[]ServiceEndpoint, staleServiceNames *[]ServicePortName) {
  2.     for svcPortName, epList := range oldEndpointsMap {
  3.         for _, ep := range epList {
  4.             stale := true
  5.             for i := range newEndpointsMap[svcPortName] {
  6.                 if newEndpointsMap[svcPortName][i].Equal(ep) {     //存在则stale为否
  7.                     stale = false
  8.                     break
  9.                 }
  10.             }
  11.             if stale {  
  12.                 klog.V(4).Infof("Stale endpoint %v -> %v", svcPortName, ep.String())
  13.                 *staleEndpoints = append(*staleEndpoints, ServiceEndpoint{Endpoint: ep.String(), ServicePortName: svcPortName})   //存储废弃的endpoint列表
  14.             }
  15.         }
  16.     }
  18.     for svcPortName, epList := range newEndpointsMap {
  19.     //对于UDP服务，如果后端变化从0至非0，可能存在conntrack项将服务的流量黑洞
  20.         if len(epList) > 0 && len(oldEndpointsMap[svcPortName]) == 0 {
  21.             *staleServiceNames = append(*staleServiceNames, svcPortName)
  22.       //存储废弃的服务名列表
  23.         }
  24.     }
  25. }

创建与联接 kube 链

  • filter表中INPUT链头部插入自定义链调转到KUBE-EXTERNAL-SERVICES链iptables -I “INPUT” -t “filter” -m “conntrack” —ctstate “NEW” -m comment —comment “kubernetes externally-visible service portals” -j “KUBE-EXTERNAL-SERVICES”

  • filter表中OUTPUT链头部插入自定义链调转到KUBE-SERVICE链iptables -I “OUTPUT” -t “filter” -m “conntrack” —ctstate “NEW” -m comment —comment “kubernetes service portals” -j “KUBE-SERVICES”

  • nat表中OUTPUT链头部插入自定义链调转到KUBE-SERVICES链iptables -I “OUTPUT” -t “nat” -m comment —comment “kubernetes service portals” -j “KUBE-SERVICES”

  • nat表中PREROUTING链头部插入自定义链调转到KUBE-SERVICES链iptables -I “PREROUTING” -t “nat” -m comment —comment “kubernetes service portals” -j “KUBE-SERVICES”

  • nat表中POSTROUTING链头部插入自定义链调转到KUBE-POSTROUTING链iptables -I “POSTROUTING” -t “nat” -m comment —comment “kubernetes postrouting rules” -j “KUBE-POSTROUTING”

  • filter表中FORWARD链头部插入自定义链调转到KUBE-FORWARD链iptables -I “FORWARD” -t “filter” -m comment —comment “kubernetes forwarding rules” -j “KUBE-FORWARD”

!FILENAME pkg/proxy/iptables/proxier.go:667

  1. //循环iptablesJumpChains定义
  2. for _, chain := range iptablesJumpChains {
  3.     //底层命令iptables -t $tableName -N $chainName
  4.         if _, err := proxier.iptables.EnsureChain(chain.table, chain.chain); err != nil {
  5.             klog.Errorf("Failed to ensure that %s chain %s exists: %v", chain.table, kubeServicesChain, err)
  6.             return
  7.         }
  8.         args := append(chain.extraArgs,
  9.             "-m", "comment", "--comment", chain.comment,
  10.             "-j", string(chain.chain),
  11.         )
  12.     //底层命令iptables  -I $chainName -t $tableName -m comment --comment $comment -j $chain
  13.         if _, err := proxier.iptables.EnsureRule(utiliptables.Prepend, chain.table, chain.sourceChain, args...); err != nil {
  14.             klog.Errorf("Failed to ensure that %s chain %s jumps to %s: %v", chain.table, chain.sourceChain, chain.chain, err)
  15.             return
  16.         }
  17.     }

创建 Iptables 基础数据

  • 获取现存在的Filter/Nat表链数据
  • 创建iptables-save/restore格式数据（表头、链）
  • 创建SNAT地址伪装规则

!FILENAME pkg/proxy/iptables/proxier.go:688

  1.     //现存在的filter表链获取
  2.     existingFilterChains := make(map[utiliptables.Chain][]byte)
  3.     proxier.existingFilterChainsData.Reset()
  4.     err := proxier.iptables.SaveInto(utiliptables.TableFilter, proxier.existingFilterChainsData)  //通过iptables-save方式来获取
  5.     if err != nil { 
  6.         klog.Errorf("Failed to execute iptables-save, syncing all rules: %v", err)
  7.     } else { 
  8.         existingFilterChains = utiliptables.GetChainLines(utiliptables.TableFilter, proxier.existingFilterChainsData.Bytes()) //输出结果
  9.     }
  11.     //同上，现存在的nat表链获取
  12.     existingNATChains := make(map[utiliptables.Chain][]byte)
  13.     proxier.iptablesData.Reset()
  14.     err = proxier.iptables.SaveInto(utiliptables.TableNAT, proxier.iptablesData)
  15.     if err != nil { 
  16.         klog.Errorf("Failed to execute iptables-save, syncing all rules: %v", err)
  17.     } else { 
  18.         existingNATChains = utiliptables.GetChainLines(utiliptables.TableNAT, proxier.iptablesData.Bytes())
  19.     }
  21.     // Reset all buffers used later.
  22.     // This is to avoid memory reallocations and thus improve performance.
  23.     proxier.filterChains.Reset()
  24.     proxier.filterRules.Reset()
  25.     proxier.natChains.Reset()
  26.     proxier.natRules.Reset()
  28.     // 写表头
  29.     writeLine(proxier.filterChains, "*filter")
  30.     writeLine(proxier.natChains, "*nat")

写链数据

fileter: “KUBE-SERVICES” / “KUBE-EXTERNAL-SERVICES”/ “KUBE-FORWARD” nat: “KUBE-SERVICES” / “KUBE-NODEPORTS” / “KUBE-POSTROUTING” / “KUBE-MARK-MASQ”

!FILENAME pkg/proxy/iptables/proxier.go:720

  1.     // 写chain链数据,将filter和Nat相关链格式化存放buffer
  2.     for _, chainName := range []utiliptables.Chain{kubeServicesChain, kubeExternalServicesChain, kubeForwardChain} {
  3.         if chain, ok := existingFilterChains[chainName]; ok {
  4.             writeBytesLine(proxier.filterChains, chain)
  5.         } else {
  6.       // iptables-save/restore格式的链行":$chainName - [0:0]"
  7.             writeLine(proxier.filterChains, utiliptables.MakeChainLine(chainName))
  8.         }
  9.     }
  10.     for _, chainName := range []utiliptables.Chain{kubeServicesChain, kubeNodePortsChain, kubePostroutingChain, KubeMarkMasqChain} {
  11.         if chain, ok := existingNATChains[chainName]; ok {
  12.             writeBytesLine(proxier.natChains, chain)
  13.         } else {
  14.             writeLine(proxier.natChains, utiliptables.MakeChainLine(chainName))
  15.         }
  16.     }

写地址伪装规则，在POSTROUTING阶段对地址进行MASQUERADE（基于接口动态IP的SNAT）处理，原始请求源IP将被丢失，被请求POD的应用看到为NodeIP或CNI设备IP(bridge/vxlan设备)

!FILENAME pkg/proxy/iptables/proxier.go:738

  1.   // 写kubernets特有的SNAT地址伪装规则
  2.   // -A KUBE-POSTROUTING -m comment --comment "kubernetes service traffic requiring SNAT" -m mark --mark 0x4000/0x4000 -j MASQUERADE
  3.     writeLine(proxier.natRules, []string{
  4.         "-A", string(kubePostroutingChain),
  5.         "-m", "comment", "--comment", `"kubernetes service traffic requiring SNAT"`,
  6.         "-m", "mark", "--mark", proxier.masqueradeMark,
  7.         "-j", "MASQUERADE",
  8.     }...)
  10.   //-A KUBE-MARK-MASQ -j MARK --set-xmark 0x4000/0x4000
  11.     writeLine(proxier.natRules, []string{
  12.         "-A", string(KubeMarkMasqChain),
  13.         "-j", "MARK", "--set-xmark", proxier.masqueradeMark,
  14.     }...)

为每个 service 创建 rules

先了解serviceInfo的完整定义说明

!FILENAME pkg/proxy/iptables/proxier.go:141

  1. type serviceInfo struct {
  2.     *proxy.BaseServiceInfo
  3.     // The following fields are computed and stored for performance reasons.
  4.     serviceNameString        string
  5.   servicePortChainName     utiliptables.Chain  // KUBE-SVC-XXXX16BitXXXX  服务链
  6.     serviceFirewallChainName utiliptables.Chain  // KUBE-FW-XXXX16BitXXXX   Firewall链
  7.     serviceLBChainName       utiliptables.Chain  // KUBE-XLB-XXXX16BitXXXX  SLB链
  8. }
  10. type BaseServiceInfo struct {
  11.   ClusterIP                net.IP                   //PortalIP(VIP)
  12.     Port                     int                      //portal端口
  13.     Protocol                 v1.Protocol              //协议
  14.     NodePort                 int                      //node节点端口     
  15.     LoadBalancerStatus       v1.LoadBalancerStatus    //LB Ingress
  16.     SessionAffinityType      v1.ServiceAffinity       //会话保持
  17.     StickyMaxAgeSeconds      int                      //保持最大时长
  18.     ExternalIPs              []string                 //ExternalIPs（指定的node上监听端口）
  19.     LoadBalancerSourceRanges []string                 //过滤源地址流量
  20.     HealthCheckNodePort      int                      //HealthCheck检测端口
  21.     OnlyNodeLocalEndpoints   bool
  22. }

为每个服务创建服务”KUBE-SVC-XXX…”和外部负载均衡”KUBE-XLB-XXX…”链

!FILENAME pkg/proxy/iptables/proxier.go:791

  1.         svcChain := svcInfo.servicePortChainName  //"KUBE-SVC-XXX..."
  2.         if hasEndpoints {
  3.             // Create the per-service chain, retaining counters if possible.
  4.             if chain, ok := existingNATChains[svcChain]; ok {
  5.                 writeBytesLine(proxier.natChains, chain)
  6.             } else {
  7.                 writeLine(proxier.natChains, utiliptables.MakeChainLine(svcChain))
  8.             }
  9.             activeNATChains[svcChain] = true
  10.         }
  12.         svcXlbChain := svcInfo.serviceLBChainName  // "KUBE-XLB-XXX…" 
  13.         if svcInfo.OnlyNodeLocalEndpoints {
  14.             // Only for services request OnlyLocal traffic
  15.             // create the per-service LB chain, retaining counters if possible.
  16.             if lbChain, ok := existingNATChains[svcXlbChain]; ok {
  17.                 writeBytesLine(proxier.natChains, lbChain)
  18.             } else {
  19.                 writeLine(proxier.natChains, utiliptables.MakeChainLine(svcXlbChain))
  20.             }
  21.             activeNATChains[svcXlbChain] = true
  22.         }

clusterIP流量的匹配，clusterIP为默认方式，仅资源集群内可访问。

!FILENAME pkg/proxy/iptables/proxier.go:815

  1.     //存在端点，写规则
  2.         if hasEndpoints {
  3.             args = append(args[:0],
  4.                 "-A", string(kubeServicesChain),
  5.                 "-m", "comment", "--comment", fmt.Sprintf(`"%s cluster IP"`, svcNameString),
  6.                 "-m", protocol, "-p", protocol,
  7.                 "-d", utilproxy.ToCIDR(svcInfo.ClusterIP),
  8.                 "--dport", strconv.Itoa(svcInfo.Port),
  9.             )
  10.      // proxier配置masqueradeAll
  11.      //  -A KUBE-SERVICES -m comment --comment "..." -m $prot -p $prot -d $clusterIP \
  12.      //     --dport $port -j KUBE-MARK-MASQ
  13.             if proxier.masqueradeAll {
  14.                 writeLine(proxier.natRules, append(args, "-j", string(KubeMarkMasqChain))...)
  15.             } else if len(proxier.clusterCIDR) > 0 {
  16.        // proxier配置clusterCIDR情况：
  17.        // -A KUBE-SERVICES ! -s $clusterCIDR -m comment --comment "..." -m $prot \
  18.        //    -p $prot -d $clusterIP --dport $port -j  KUBE-MARK-MASQ
  19.                 writeLine(proxier.natRules, append(args, "! -s", proxier.clusterCIDR, "-j", string(KubeMarkMasqChain))...)
  20.             }
  21.       //  -A KUBE-SERVICES -m comment --comment "..." -m $prot -p $prot -d $clusterIP \
  22.       //     --dport $port -j KUBE-SVC-XXXX16bitXXXX
  23.             writeLine(proxier.natRules, append(args, "-j", string(svcChain))...)
  24.         } else {
  25.       // 无Endpoints的情况，则创建REJECT规则
  26.       // -A KUBE-SERVICES -m comment --comment $svcName -m $prot -p $prot -d $clusterIP \
  27.       // --dport $port -j REJECT
  28.             writeLine(proxier.filterRules,
  29.                 "-A", string(kubeServicesChain),
  30.                 "-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString),
  31.                 "-m", protocol, "-p", protocol,
  32.                 "-d", utilproxy.ToCIDR(svcInfo.ClusterIP),
  33.                 "--dport", strconv.Itoa(svcInfo.Port),
  34.                 "-j", "REJECT",
  35.             )
  36.         }

服务是否启用ExternalIPs(指定的node上开启监听端口)

!FILENAME pkg/proxy/iptables/proxier.go:846

  1.     for _, externalIP := range svcInfo.ExternalIPs {
  2.       // 判断externalIP是否为本node的IP以及协议为SCTP，且端口是否已开启
  3.       // 如果未开启则在本地打开监听端口
  4.             if local, err := utilproxy.IsLocalIP(externalIP); err != nil {
  5.                 klog.Errorf("can't determine if IP is local, assuming not: %v", err)
  6.             } else if local && (svcInfo.GetProtocol() != v1.ProtocolSCTP) {
  7.                 lp := utilproxy.LocalPort{
  8.                     Description: "externalIP for " + svcNameString,
  9.                     IP:          externalIP,
  10.                     Port:        svcInfo.Port,
  11.                     Protocol:    protocol,
  12.                 }
  13.                 if proxier.portsMap[lp] != nil {
  14.                     klog.V(4).Infof("Port %s was open before and is still needed", lp.String())
  15.                     replacementPortsMap[lp] = proxier.portsMap[lp]
  16.                 } else {
  17.           //打开与监听本地端口
  18.                     socket, err := proxier.portMapper.OpenLocalPort(&lp)
  19.                     if err != nil {
  20.                         msg := fmt.Sprintf("can't open %s, skipping this externalIP: %v", lp.String(), err)
  22.                         proxier.recorder.Eventf(
  23.                             &v1.ObjectReference{
  24.                                 Kind:      "Node",
  25.                                 Name:      proxier.hostname,
  26.                                 UID:       types.UID(proxier.hostname),
  27.                                 Namespace: "",
  28.                             }, v1.EventTypeWarning, err.Error(), msg)
  29.                         klog.Error(msg)
  30.                         continue
  31.                     }
  32.                     replacementPortsMap[lp] = socket
  33.                 }
  34.             }
  36.       //存在端点，写规则
  37.             if hasEndpoints {
  38.                 args = append(args[:0],
  39.                     "-A", string(kubeServicesChain),        
  40.                     "-m", "comment", "--comment", fmt.Sprintf(`"%s external IP"`, svcNameString),
  41.                     "-m", protocol, "-p", protocol,
  42.                     "-d", utilproxy.ToCIDR(net.ParseIP(externalIP)),
  43.                     "--dport", strconv.Itoa(svcInfo.Port),
  44.                 )
  46.          // -A KUBE-EXTERNAL-SERVICES -m comment --comment "..." -m $prot -p $prot -d \
  47.         //    $externalIP --dport $port -j KUBE-MARK-MASQ 
  48.                 writeLine(proxier.natRules, append(args, "-j", string(KubeMarkMasqChain))...)
  50.         //  -A KUBE-EXTERNAL-SERVICES -m comment --comment "..." -m $prot -p $prot -d \
  51.         //     $externalIP --dport $port -m physdev ! --physdev-is-in \ 
  52.         //     -m addrtype ! --src-type Local -j KUBE-SVC-XXXX16bitXXXXX
  53.                 externalTrafficOnlyArgs := append(args,
  54.                     "-m", "physdev", "!", "--physdev-is-in",
  55.                     "-m", "addrtype", "!", "--src-type", "LOCAL")
  56.                 writeLine(proxier.natRules, append(externalTrafficOnlyArgs, "-j", string(svcChain))...)
  57.                 dstLocalOnlyArgs := append(args, "-m", "addrtype", "--dst-type", "LOCAL")
  59.         //  -A KUBE-EXTERNAL-SERVICES -m comment --comment "..." -m $prot -p $prot -d \
  60.         //     $externalIP --dport $port -m addrtype --dst-type Local 
  61.         //     -j KUBE-SVC-XXXX16bitXXXXX
  62.                 writeLine(proxier.natRules, append(dstLocalOnlyArgs, "-j", string(svcChain))...)
  63.             } else {
  64.         // 不存在端点信息则reject
  65.         // -A KUBE-EXTERNAL-SERVICES -m comment --comment "..." -m $prot -p $prot -d \
  66.         //    $externalIP --dport $port -j REJECT 
  67.                 writeLine(proxier.filterRules,
  68.                     "-A", string(kubeExternalServicesChain),
  69.                     "-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString),
  70.                     "-m", protocol, "-p", protocol,
  71.                     "-d", utilproxy.ToCIDR(net.ParseIP(externalIP)),
  72.                     "--dport", strconv.Itoa(svcInfo.Port),
  73.                     "-j", "REJECT",
  74.                 )
  75.             }
  76.         }

服务是否启用了外部负载均衡服务load-balancer ingress

!FILENAME pkg/proxy/iptables/proxier.go:917

  1.     //存在端点，写规则
  2.         if hasEndpoints {
  3.             fwChain := svcInfo.serviceFirewallChainName     //"KUBE-FW-XXXX16bitXXXXX"
  4.             for _, ingress := range svcInfo.LoadBalancerStatus.Ingress {
  5.                 if ingress.IP != "" {
  6.                     // 创建服务KUBE-FW-X链
  7.                     if chain, ok := existingNATChains[fwChain]; ok {
  8.                         writeBytesLine(proxier.natChains, chain)
  9.                     } else {  //原来不存在则新建
  10.                         writeLine(proxier.natChains, utiliptables.MakeChainLine(fwChain))
  11.                     }
  12.                     activeNATChains[fwChain] = true
  13.                     // The service firewall rules are created based on ServiceSpec.loadBalancerSourceRanges field.
  14.                     // This currently works for loadbalancers that preserves source ips.
  15.                     // For loadbalancers which direct traffic to service NodePort, the firewall rules will not apply.
  17.                     args = append(args[:0],
  18.                         "-A", string(kubeServicesChain),
  19.                         "-m", "comment", "--comment", fmt.Sprintf(`"%s loadbalancer IP"`, svcNameString),
  20.                         "-m", protocol, "-p", protocol,
  21.                         "-d", utilproxy.ToCIDR(net.ParseIP(ingress.IP)),
  22.                         "--dport", strconv.Itoa(svcInfo.Port),
  23.                     )          
  24.           // -A KUBE-SERVICES -m comment --comment "..." -m $prot -p $prot -d \
  25.           //    $ingresIP --dport $port -j KUBE-FW-XXXX16bitXXXXX 
  26.                     writeLine(proxier.natRules, append(args, "-j", string(fwChain))...)
  28.                     args = append(args[:0],
  29.                         "-A", string(fwChain),
  30.                         "-m", "comment", "--comment", fmt.Sprintf(`"%s loadbalancer IP"`, svcNameString),
  31.                     )
  33.            // 在KUBE-FW链，每个源匹配规则可能跳转至一个SVC或XLB链
  34.                     chosenChain := svcXlbChain
  35.                     // If we are proxying globally, we need to masquerade in case we cross nodes.
  36.                     // If we are proxying only locally, we can retain the source IP.
  37.                     if !svcInfo.OnlyNodeLocalEndpoints {
  38.             // -j "KUBE-MARK-MASQ" 地址伪装实现跨主机访问 
  39.                         writeLine(proxier.natRules, append(args, "-j", string(KubeMarkMasqChain))...)
  40.                         chosenChain = svcChain   // 选择为SVC链
  41.                     }
  43.                     if len(svcInfo.LoadBalancerSourceRanges) == 0 {
  44.             // 允许所有源，直接跳转 
  45.                         writeLine(proxier.natRules, append(args, "-j", string(chosenChain))...)
  46.                     } else {
  47.                         // 基于source range配置过滤 "-s $srcRanges"
  48.             allowFromNode := false
  49.                         for _, src := range svcInfo.LoadBalancerSourceRanges {
  50.                             writeLine(proxier.natRules, append(args, "-s", src, "-j", string(chosenChain))...)
  51.                             _, cidr, _ := net.ParseCIDR(src)
  52.                             if cidr.Contains(proxier.nodeIP) {
  53.                                 allowFromNode = true    //配置CIDR包含节点IP，则允许来自节点请求
  54.                             }
  55.                         }
  57.             // 添加 "-s $ingresIP" 来允许LB后端主机请求
  58.                         if allowFromNode {
  59.                             writeLine(proxier.natRules, append(args, "-s", utilproxy.ToCIDR(net.ParseIP(ingress.IP)), "-j", string(chosenChain))...)
  60.                         }
  61.                     }
  63.           // 条件ingress.IP为空"-j KUBE-MARK-DROP"
  64.                     writeLine(proxier.natRules, append(args, "-j", string(KubeMarkDropChain))...)
  65.                 }
  66.             }
  67.         }

服务是否启用了nodeport(在每个节点上都将开启一个nodeport端口)

!FILENAME pkg/proxy/iptables/proxier.go:989

  1.         if svcInfo.NodePort != 0 {
  2.       // 获取node addresses
  3.             addresses, err := utilproxy.GetNodeAddresses(proxier.nodePortAddresses, proxier.networkInterfacer)
  4.             if err != nil {
  5.                 klog.Errorf("Failed to get node ip address matching nodeport cidr: %v", err)
  6.                 continue
  7.             }
  9.             lps := make([]utilproxy.LocalPort, 0)
  10.             for address := range addresses {
  11.                 lp := utilproxy.LocalPort{
  12.                     Description: "nodePort for " + svcNameString,
  13.                     IP:          address,
  14.                     Port:        svcInfo.NodePort,
  15.                     Protocol:    protocol,
  16.                 }
  17.                 if utilproxy.IsZeroCIDR(address) {
  18.                     // Empty IP address means all
  19.                     lp.IP = ""
  20.                     lps = append(lps, lp)
  21.                     // If we encounter a zero CIDR, then there is no point in processing the rest of the addresses.
  22.                     break
  23.                 }
  24.                 lps = append(lps, lp)     //IP列表
  25.             }
  27.       // 为node节点的ips打开端口并保存持有socket句柄
  28.             for _, lp := range lps {
  29.                 if proxier.portsMap[lp] != nil {
  30.                     klog.V(4).Infof("Port %s was open before and is still needed", lp.String())
  31.                     replacementPortsMap[lp] = proxier.portsMap[lp]
  32.                 } else if svcInfo.GetProtocol() != v1.ProtocolSCTP {
  33.           // 打开和监听端口
  34.                     socket, err := proxier.portMapper.OpenLocalPort(&lp)
  35.                     if err != nil {
  36.                         klog.Errorf("can't open %s, skipping this nodePort: %v", lp.String(), err)
  37.                         continue
  38.                     }
  39.                     if lp.Protocol == "udp" {
  40.                 //清理udp conntrack记录
  41.                         err := conntrack.ClearEntriesForPort(proxier.exec, lp.Port, isIPv6, v1.ProtocolUDP)
  42.                         if err != nil {
  43.                             klog.Errorf("Failed to clear udp conntrack for port %d, error: %v", lp.Port, err)
  44.                         }
  45.                     }
  46.                     replacementPortsMap[lp] = socket //socket保存
  47.                 }
  48.             }
  50.       //存在端点，写规则
  51.             if hasEndpoints {
  52.         // -A KUBE-NODEPORTS -m comment --comment "..." -m $prot -p $prot --dport $nodePort
  53.                 args = append(args[:0],
  54.                     "-A", string(kubeNodePortsChain),
  55.                     "-m", "comment", "--comment", svcNameString,
  56.                     "-m", protocol, "-p", protocol,
  57.                     "--dport", strconv.Itoa(svcInfo.NodePort),
  58.                 )
  59.                 if !svcInfo.OnlyNodeLocalEndpoints {
  60.           //非本地nodeports则需SNAT规则添加,
  61.           // -j KUBE-MARK-MASQ -j KUBE-XLB-XXXX16bitXXXX
  62.                     writeLine(proxier.natRules, append(args, "-j", string(KubeMarkMasqChain))...)
  63.                     // Jump to the service chain.
  64.                     writeLine(proxier.natRules, append(args, "-j", string(svcChain))...)
  65.                 } else {
  66.                     loopback := "127.0.0.0/8"
  67.                     if isIPv6 {
  68.                         loopback = "::1/128"
  69.                     }
  70.            // 本地nodeports则规则添加,
  71.           // -s $loopback -j KUBE-MARK-MASQ -j KUBE-XLB-XXXX16bitXXXX
  72.                     writeLine(proxier.natRules, append(args, "-s", loopback, "-j", string(KubeMarkMasqChain))...)
  73.                     writeLine(proxier.natRules, append(args, "-j", string(svcXlbChain))...)
  74.                 }
  75.             } else {
  76.         // 无hasEndpoints，添加-j reject规则 
  77.         // -A KUBE-EXTERNAL-SERVICES -m comment --comment "..." -m addrtype \
  78.         //     --dst-type LOCAL -m $prot -p $prot --dport $nodePort -j REJECT
  79.                 writeLine(proxier.filterRules,
  80.                     "-A", string(kubeExternalServicesChain),
  81.                     "-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString),
  82.                     "-m", "addrtype", "--dst-type", "LOCAL",
  83.                     "-m", protocol, "-p", protocol,
  84.                     "--dport", strconv.Itoa(svcInfo.NodePort),
  85.                     "-j", "REJECT",
  86.                 )
  87.             }
  88.         }

基于服务名和协议，生成每个端点链

!FILENAME pkg/proxy/iptables/proxier.go:1087

  1.   for _, ep := range proxier.endpointsMap[svcName] {
  2.             epInfo, ok := ep.(*endpointsInfo)
  3.             if !ok {
  4.                 klog.Errorf("Failed to cast endpointsInfo %q", ep.String())
  5.                 continue
  6.             }
  7.             endpoints = append(endpoints, epInfo)
  8.       //基于服务名和协议生成端点链名称 "KUBE-SEP-XXXX16bitXXXX"
  9.             endpointChain = epInfo.endpointChain(svcNameString, protocol)
  10.             endpointChains = append(endpointChains, endpointChain)
  12.       // 创建端点链
  13.             if chain, ok := existingNATChains[utiliptables.Chain(endpointChain)]; ok {
  14.                 writeBytesLine(proxier.natChains, chain) 
  15.             } else {
  16.                 writeLine(proxier.natChains, utiliptables.MakeChainLine(endpointChain))
  17.             }
  18.             activeNATChains[endpointChain] = true
  19.         }

写SessionAffinity会话保持规则，实现在一段时间内保持session affinity，保持时间为180秒，通过添加“-m recent –rcheck –seconds 180 –reap”的iptables规则实现了会话保持。

!FILENAME pkg/proxy/iptables/proxier.go:1107

  1.     //SessionAffinityType设置为"ClientIP"，则写session保持规则
  2.     // -A KUBE-SVC-XXXX16bitXXXX -m recent -m comment –comment "..." \
  3.     //    --name KUBE-SEP-XXXX16bitXXXX --rcheck --seconds 180 --reap \
  4.     //    -j KUBE-SEP-XXXX16bitXXXX
  5.         if svcInfo.SessionAffinityType == v1.ServiceAffinityClientIP {
  6.             for _, endpointChain := range endpointChains {
  7.                 args = append(args[:0],
  8.                     "-A", string(svcChain),
  9.                 )
  10.                 proxier.appendServiceCommentLocked(args, svcNameString)
  11.                 args = append(args,
  12.                     "-m", "recent", "--name", string(endpointChain),
  13.                     "--rcheck", "--seconds", strconv.Itoa(svcInfo.StickyMaxAgeSeconds), "--reap",
  14.                     "-j", string(endpointChain),
  15.                 )
  16.                 writeLine(proxier.natRules, args...)
  17.             }
  18.         }

写负载均衡和DNAT规则，使用“-m statistic –-mode random -–probability ” iptables规则将后端POD组成一个基于概率访问的组合,实现服务访问的负载均衡功能效果。

  • 针对服务的每个端点在nat表内该service对应的自定义链“KUBE-SVC-XXXX16bitXXXX”中加入iptables规则。如果该服务对应的endpoints大于等于2，则添加负载均衡规则。

  • 针对选择非本地Node上的POD，需进行DNAT，将请求的目标地址设置成后选的POD的IP后进行路由。KUBE-MARK-MASQ将重设(伪装)源地址

    -A KUBE-SVC-XXXX16bitXXXX -m comment –comment “…” -m statistic —mode random —probability $prob -j KUBE-SEP-XXXX16bitXXXX

-A KUBE-SEP-XXXX16bitXXXX -m comment –comment “…” -s $epIp -j “KUBE-MARK-MASQ”

-A KUBE-SVC-XXXX16bitXXXX -m comment –comment “…” -m prot -p $prot -j DNAT —to-destination X.X.X.X:xxx

!FILENAME pkg/proxy/iptables/proxier.go:1123

  1.     // 写负载均衡和DNAT规则
  2.         n := len(endpointChains)
  3.         for i, endpointChain := range endpointChains {
  4.             epIP := endpoints[i].IP()
  5.             if epIP == "" {
  6.                 // Error parsing this endpoint has been logged. Skip to next endpoint.
  7.                 continue
  8.             }
  9.             // 每个服务生成的负载均衡规则，后端POD组成一个基于概率访问的组合
  10.       //  // -A KUBE-SVC-XXXX16bitXXXX -m comment –comment "..." 
  11.       //    -m statistic --mode random --probability $prob 
  12.       //    -j KUBE-SEP-XXXX16bitXXXX
  13.             args = append(args[:0], "-A", string(svcChain))
  14.             proxier.appendServiceCommentLocked(args, svcNameString)
  15.             if i < (n - 1) {   // 当端点大于或等于2 
  16.                 args = append(args,
  17.                     "-m", "statistic",
  18.                     "--mode", "random",
  19.                     "--probability", proxier.probability(n-i))
  20.             }
  21.             // The final (or only if n == 1) rule is a guaranteed match.
  22.             args = append(args, "-j", string(endpointChain))
  23.             writeLine(proxier.natRules, args...)
  25.       // 每个端点链规则
  26.       // -A KUBE-SEP-XXXX16bitXXXX -m comment –comment "..."  -s $epIp -j "KUBE-MARK-MASQ"
  27.             args = append(args[:0], "-A", string(endpointChain))
  28.             proxier.appendServiceCommentLocked(args, svcNameString)
  29.             // Handle traffic that loops back to the originator with SNAT.
  30.             writeLine(proxier.natRules, append(args,
  31.                 "-s", utilproxy.ToCIDR(net.ParseIP(epIP)),
  32.                 "-j", string(KubeMarkMasqChain))...)
  34.       // 如配置session保持"ClientIP"
  35.       // -m recent --name KUBE-SEP-XXXX16bitXXXX --set 
  36.             if svcInfo.SessionAffinityType == v1.ServiceAffinityClientIP {
  37.                 args = append(args, "-m", "recent", "--name", string(endpointChain), "--set")
  38.             }
  39.             // DNAT至最终的端点服务上
  40.       // -A KUBE-SVC-XXXX16bitXXXX -m comment –comment "..." 
  41.       //    -m $prot -p $prot -j DNAT --to-destination X.X.X.X:xxx
  42.             args = append(args, "-m", protocol, "-p", protocol, "-j", "DNAT", "--to-destination", endpoints[i].Endpoint)
  43.             writeLine(proxier.natRules, args...)
  44.         }
  46.     // 服务请求仅本地流量
  47.         localEndpoints := make([]*endpointsInfo, 0)
  48.         localEndpointChains := make([]utiliptables.Chain, 0)
  49.         for i := range endpointChains {
  50.             if endpoints[i].IsLocal {
  51.                 // These slices parallel each other; must be kept in sync
  52.                 localEndpoints = append(localEndpoints, endpoints[i])
  53.                 localEndpointChains = append(localEndpointChains, endpointChains[i])
  54.             }
  55.         }

启用clusterCIDR (Kube-proxy中的--cluster-dir指定的是集群中pod使用的网段，而pod使用的网段和apiserver中指定的service的cluster ip或vip网段不是同一个网段)

!FILENAME pkg/proxy/iptables/proxier.go:1179

  1.      // pod -> external VIP流量导向服务VIP(服务链)
  2.     // -A KUBE-XLB-XXXX16bitXXXX -m comment --comment "..." -s $clusterCIDR 
  3.     //    -j  KUBE-SVC-XXXX16bitXXXXX
  4.         if len(proxier.clusterCIDR) > 0 {
  5.             args = append(args[:0],
  6.                 "-A", string(svcXlbChain),
  7.                 "-m", "comment", "--comment",
  8.                 `"Redirect pods trying to reach external loadbalancer VIP to clusterIP"`,
  9.                 "-s", proxier.clusterCIDR,
  10.                 "-j", string(svcChain),
  11.             )
  12.             writeLine(proxier.natRules, args...)
  13.         }

生成本地端点链规则，本地源IP保持(当只在本地选择POD服务请求时，则不存在SNAT规则，可保持源地址IP信息。在nodePort或XLB时，可定义"externalTrafficPolicy": "Local"控制向属于这个service的本地的POD转发请求，如果本地没有POD能服务这个请求，请求将被DROP掉，客户端会发现请求超时没有响应。

!FILENAME pkg/proxy/iptables/proxier.go:1190

  1.     numLocalEndpoints := len(localEndpointChains)
  2.         if numLocalEndpoints == 0 {
  3.       // 无本地端点，将流量Drop(流量黑洞处理)
  4.       // -A KUBE-XLB-XXXX16bitXXXX -m comment --comment "..." -j KUBE-MARK-DROP
  5.             args = append(args[:0],
  6.                 "-A", string(svcXlbChain),
  7.                 "-m", "comment", "--comment",
  8.                 fmt.Sprintf(`"%s has no local endpoints"`, svcNameString),
  9.                 "-j",
  10.                 string(KubeMarkDropChain),
  11.             )
  12.             writeLine(proxier.natRules, args...)
  13.         } else {
  14.             // 本地端点会话保持开启
  15.       // -A KUBE-XLB-XXXX16bitXXXX -m comment --comment "..." -m recent \ 
  16.       //    --name KUBE-SEP-XXXX16bitXXXX \
  17.       //    --rcheck --seconds $StickyMaxAge --reap -j KUBE-SEP-XXXX16bitXXXX
  18.             if svcInfo.SessionAffinityType == v1.ServiceAffinityClientIP {
  19.                 for _, endpointChain := range localEndpointChains {
  20.                     writeLine(proxier.natRules,
  21.                         "-A", string(svcXlbChain),
  22.                         "-m", "comment", "--comment", svcNameString,
  23.                         "-m", "recent", "--name", string(endpointChain),
  24.                         "--rcheck", "--seconds", strconv.Itoa(svcInfo.StickyMaxAgeSeconds), "--reap",
  25.                         "-j", string(endpointChain))
  26.                 }
  27.             }
  29.       // 本地端点负载均衡处理"-m statistic --mode random --probability"
  30.       // 后端POD组成一个基于概率访问的组合
  31.             for i, endpointChain := range localEndpointChains {
  32.                 // Balancing rules in the per-service chain.
  33.                 args = append(args[:0],
  34.                     "-A", string(svcXlbChain),
  35.                     "-m", "comment", "--comment",
  36.                     fmt.Sprintf(`"Balancing rule %d for %s"`, i, svcNameString),
  37.                 )
  38.                 if i < (numLocalEndpoints - 1) {
  39.                     // Each rule is a probabilistic match.
  40.                     args = append(args,
  41.                         "-m", "statistic",
  42.                         "--mode", "random",
  43.                         "--probability", proxier.probability(numLocalEndpoints-i))
  44.                 }
  46.                 args = append(args, "-j", string(endpointChain))
  47.       // -A KUBE-XLB-XXXX16bitXXXX -m comment --comment "..." -m recent \ 
  48.       //    --name KUBE-SEP-XXXX16bitXXXX -m statistic --mode random \
  49.       //    --probability 0.50000000000  -j KUBE-SEP-XXXX16bitXXXX
  50.                 writeLine(proxier.natRules, args...)
  51.             }
  52.         }

配置收尾规则数据

删除不再使用的服务自定义kube链”KUBE-SVC-*“/“KUBE-SEP-*“/“KUBE-FW-*“/“KUBE-XLB-*“。

!FILENAME pkg/proxy/iptables/proxier.go:1237

  1.     for chain := range existingNATChains {
  2.         if !activeNATChains[chain] {
  3.             chainString := string(chain)
  4.             if !strings.HasPrefix(chainString, "KUBE-SVC-") && !strings.HasPrefix(chainString, "KUBE-SEP-") && !strings.HasPrefix(chainString, "KUBE-FW-") && !strings.HasPrefix(chainString, "KUBE-XLB-") {
  5.                 // Ignore chains that aren't ours.
  6.                 continue
  7.             }
  8.             // We must (as per iptables) write a chain-line for it, which has
  9.             // the nice effect of flushing the chain.  Then we can remove the
  10.             // chain.
  11.             writeBytesLine(proxier.natChains, existingNATChains[chain])
  12.             writeLine(proxier.natRules, "-X", chainString)
  13.         }
  14.     }

添加服务的nodeports规则(nat表-“KUBE-SERVICES”链)

!FILENAME pkg/proxy/iptables/proxier.go:1254

  1. // -A KUBE-SERVICES -m comment --comment "..." -m addrtype --dst-type LOCAL \
  2. //    -j KUBE-NODEPORTS
  3. //
  4. // -A KUBE-SERVICES -m comment --comment "..." -d $NODEIP -j KUBE-NODEPORTS
  5. //
  6. addresses, err := utilproxy.GetNodeAddresses(proxier.nodePortAddresses, proxier.networkInterfacer)
  7.     if err != nil {
  8.         klog.Errorf("Failed to get node ip address matching nodeport cidr")
  9.     } else {
  10.         isIPv6 := proxier.iptables.IsIpv6()
  11.         for address := range addresses {
  12.             // TODO(thockin, m1093782566): If/when we have dual-stack support we will want to distinguish v4 from v6 zero-CIDRs.
  13.             if utilproxy.IsZeroCIDR(address) {
  14.                 args = append(args[:0],
  15.                     "-A", string(kubeServicesChain),
  16.                     "-m", "comment", "--comment", `"kubernetes service nodeports; NOTE: this must be the last rule in this chain"`,
  17.                     "-m", "addrtype", "--dst-type", "LOCAL",
  18.                     "-j", string(kubeNodePortsChain))
  19.                 writeLine(proxier.natRules, args...)
  20.                 // Nothing else matters after the zero CIDR.
  21.                 break
  22.             }
  23.             // Ignore IP addresses with incorrect version
  24.             if isIPv6 && !utilnet.IsIPv6String(address) || !isIPv6 && utilnet.IsIPv6String(address) {
  25.                 klog.Errorf("IP address %s has incorrect IP version", address)
  26.                 continue
  27.             }
  28.             // create nodeport rules for each IP one by one
  29.             args = append(args[:0],
  30.                 "-A", string(kubeServicesChain),
  31.                 "-m", "comment", "--comment", `"kubernetes service nodeports; NOTE: this must be the last rule in this chain"`,
  32.                 "-d", address,
  33.                 "-j", string(kubeNodePortsChain))
  34.             writeLine(proxier.natRules, args...)
  35.         }
  36.     }

添加forward规则(filter表-“KUBE-FORWARD”链)

!FILENAME pkg/proxy/iptables/proxier.go:1289

  1. //-A KUBE-FORWARD -m comment --comment "..." -m mark --mark 0xFFFF/0xFFFF -j ACCEPT
  2. writeLine(proxier.filterRules,
  3.         "-A", string(kubeForwardChain),
  4.         "-m", "comment", "--comment", `"kubernetes forwarding rules"`,
  5.         "-m", "mark", "--mark", proxier.masqueradeMark,
  6.         "-j", "ACCEPT",
  7.     )

添加带clusterCIDR配置(源/目标)规则(filter表-“KUBE-FORWARD”链)

!FILENAME pkg/proxy/iptables/proxier.go:1297

  1. //Kube-proxy中的cluster-dir指定的是集群中pod使用的网段
  2. //pod使用的网段和service的cluster ip网段不是同一个网段
  3. //
  4. // -A KUBE-FORWARD -s $clusterCIDR -m comment --comment "..." -m conntrack --ctstate \ 
  5. //    RELATED,ESTABLISHED -j ACCEPT
  6. // -A KUBE-FORWARD -m comment --comment "..." -d $clusterCIDR -m conntrack --ctstate \ 
  7. //    RELATED,ESTABLISHED -j ACCEPT
  8. //
  9. if len(proxier.clusterCIDR) != 0 {
  10.         writeLine(proxier.filterRules,
  11.             "-A", string(kubeForwardChain),
  12.             "-s", proxier.clusterCIDR,           //指定源
  13.             "-m", "comment", "--comment", `"kubernetes forwarding conntrack pod source rule"`,
  14.             "-m", "conntrack",
  15.             "--ctstate", "RELATED,ESTABLISHED",
  16.             "-j", "ACCEPT",
  17.         )
  18.         writeLine(proxier.filterRules,
  19.             "-A", string(kubeForwardChain),
  20.             "-m", "comment", "--comment", `"kubernetes forwarding conntrack pod destination rule"`,
  21.             "-d", proxier.clusterCIDR,            //指定目标
  22.             "-m", "conntrack",
  23.             "--ctstate", "RELATED,ESTABLISHED",
  24.             "-j", "ACCEPT",
  25.         )
  26.     }

结尾标志写入

  1.     writeLine(proxier.filterRules, "COMMIT")
  2.     writeLine(proxier.natRules, "COMMIT")

汇集与加载 iptables 配置规则数据

!FILENAME pkg/proxy/iptables/proxier.go:1326

  1.   //汇集前面所处理的filter和nat表数据至iptablesData
  2.   proxier.iptablesData.Reset()
  3.     proxier.iptablesData.Write(proxier.filterChains.Bytes())
  4.     proxier.iptablesData.Write(proxier.filterRules.Bytes())
  5.     proxier.iptablesData.Write(proxier.natChains.Bytes())
  6.     proxier.iptablesData.Write(proxier.natRules.Bytes())
  8.     klog.V(5).Infof("Restoring iptables rules: %s", proxier.iptablesData.Bytes())
  9. // iptables-restore加载新配置(iptablesData)
  10.     err = proxier.iptables.RestoreAll(proxier.iptablesData.Bytes(), utiliptables.NoFlushTables, utiliptables.RestoreCounters)
  11.     if err != nil {
  12.         klog.Errorf("Failed to execute iptables-restore: %v", err)
  13.         // Revert new local ports.
  14.         klog.V(2).Infof("Closing local ports after iptables-restore failure")
  15.         utilproxy.RevertPorts(replacementPortsMap, proxier.portsMap)
  16.         return
  17.     }

IPtables 底层的 runner 实现

前面基本已看完整个proxy的执行流程，最后iptables proxier是如何使用系统层iptables命令进行底层的iptables规则CRUD操作（通俗的理解：iptables proxier实现都是在操作iptables命令生成相应的规则），下面我来看一下kuber-proxy组件底层iptables操作器的封装。

iptables 执行器

Interface**接口为运行iptables命令定义

!FILENAME pkg/util/iptables/iptables.go:45

  1. //接口与接口方法定义
  2. type Interface interface {
  3.     GetVersion() (string, error)
  4.     EnsureChain(table Table, chain Chain) (bool, error)
  5.     FlushChain(table Table, chain Chain) error
  6.     DeleteChain(table Table, chain Chain) error
  7.     EnsureRule(position RulePosition, table Table, chain Chain, args ...string) (bool, error)
  8.     DeleteRule(table Table, chain Chain, args ...string) error
  9.     IsIpv6() bool
  10.     SaveInto(table Table, buffer *bytes.Buffer) error
  11.     Restore(table Table, data []byte, flush FlushFlag, counters RestoreCountersFlag) error
  12.     RestoreAll(data []byte, flush FlushFlag, counters RestoreCountersFlag) error
  13.     AddReloadFunc(reloadFunc func())
  14.     Destroy()
  15. }

iptables Interface接实现类runner,完成对iptables命令执行器的定义

!FILENAME pkg/util/iptables/iptables.go:135

  1. // 类结构定义
  2. type runner struct {
  3.     mu              sync.Mutex              //同步锁 
  4.     exec            utilexec.Interface      //osExec命令执行
  5.     dbus            utildbus.Interface      //D-Bus操作API接口 
  6.     protocol        Protocol                //协议IPv4/IPv6
  7.     hasCheck        bool                    //"-C"检测命令flag
  8.   hasListener     bool                    //D-Bus信号监听是否开启(FirewallD start/restart)
  9.     waitFlag        []string                //iptables命令"wait"flag,等待xtables锁
  10.     restoreWaitFlag []string                //iptables-restore命令"wait"flag
  11.     lockfilePath    string                  //xtables锁文件位置
  13.     reloadFuncs []func()                    //定义reload处理func
  14.     signal      chan *godbus.Signal         //dbus信号
  15. }
  17. // Runner实现Iterface方法列表，后面将详细分析关键的方法实现代码逻辑
  18. func (runner *runner) GetVersion() (string, error)
  19. func (runner *runner) EnsureChain(table Table, chain Chain) (bool, error) 
  20. func (runner *runner) FlushChain(table Table, chain Chain) error 
  21. func (runner *runner) DeleteChain(table Table, chain Chain) error
  22. func (runner *runner) EnsureRule(position RulePosition, table Table, chain Chain, args ...string) (bool, error) 
  23. func (runner *runner) DeleteRule(table Table, chain Chain, args ...string) error
  24. func (runner *runner) IsIpv6() bool 
  25. func (runner *runner) SaveInto(table Table, buffer *bytes.Buffer) error
  26. func (runner *runner) Restore(table Table, data []byte, flush FlushFlag, counters RestoreCountersFlag) error 
  27. func (runner *runner) RestoreAll(data []byte, flush FlushFlag, counters RestoreCountersFlag) error
  28. func (runner *runner) AddReloadFunc(reloadFunc func()) 
  29. func (runner *runner) Destroy() 
  31. // Runner内部方法列表
  32. func (runner *runner) connectToFirewallD()
  33. func (runner *runner) restoreInternal(args []string, data []byte, flush FlushFlag, counters RestoreCountersFlag) error
  34. func (runner *runner) run(op operation, args []string) ([]byte, error)
  35. func (runner *runner) runContext(ctx context.Context, op operation, args []string) ([]byte, error)
  36. func (runner *runner) checkRule(table Table, chain Chain, args ...string) (bool, error)
  37. func (runner *runner) checkRuleWithoutCheck(table Table, chain Chain, args ...string) (bool, error) 
  38. func (runner *runner) checkRuleUsingCheck(args []string) (bool, error)
  39. func (runner *runner) dbusSignalHandler(bus utildbus.Connection)
  40. func (runner *runner) reload()

iptables runner对象的构造New() -> newInternal(),返回runner{…}实例化对象(Interface接口类型) ，完成了创建一个iptables的命令执行器生成工作。

!FILENAME pkg/util/iptables/iptables.go:152

  1. func newInternal(exec utilexec.Interface, dbus utildbus.Interface, protocol Protocol, lockfilePath string) Interface {
  2.     vstring, err := getIPTablesVersionString(exec, protocol)   //iptables版本获取
  3.     if err != nil {
  4.         klog.Warningf("Error checking iptables version, assuming version at least %s: %v", MinCheckVersion, err)
  5.         vstring = MinCheckVersion
  6.     }
  8.     if lockfilePath == "" {
  9.         lockfilePath = LockfilePath16x                      //默认锁文件位置"/run/xtables.lock"
  10.     }
  12.     runner := &runner{
  13.         exec:            exec,                                         //utilexec = osExec封装
  14.         dbus:            dbus,                                         //utildbus
  15.         protocol:        protocol,                                     //IPv4 or IPv6
  16.         hasCheck:        getIPTablesHasCheckCommand(vstring),          //"-C" flag是否指定
  17.         hasListener:     false,
  18.         waitFlag:        getIPTablesWaitFlag(vstring),                 //iptables -wait
  19.         restoreWaitFlag: getIPTablesRestoreWaitFlag(exec, protocol),   //iptables-restore -wait
  20.         lockfilePath:    lockfilePath,                                 //xtables锁文件位置
  21.     }
  22.     return runner
  23. }
  25. // 返回iptables exec命令执行器对象runner
  26. func New(exec utilexec.Interface, dbus utildbus.Interface, protocol Protocol) Interface {
  27.     return newInternal(exec, dbus, protocol, "")     
  28. }

iptables 执行器方法

runner.run() 这个是方法是runner最基础和公共调用的内部方法，也就是iptables命令执行os exec调用代码。run()有两个传参：1. 指定iptables操作command，2.参数列表。通过传参将组成一个完整的iptables命令进行exec调用执行。runContext()此方法内含有带context上下文和不带context两种执行方式。

!FILENAME pkg/util/iptables/iptables.go:218

  1. func (runner *runner) run(op operation, args []string) ([]byte, error) {
  2.     return runner.runContext(nil, op, args)
  3. }
  5. func (runner *runner) runContext(ctx context.Context, op operation, args []string) ([]byte, error) {
  6.     iptablesCmd := iptablesCommand(runner.protocol)  // "iptabels or ip6tables"
  7.     fullArgs := append(runner.waitFlag, string(op))
  8.     fullArgs = append(fullArgs, args...)
  9.     klog.V(5).Infof("running iptables %s %v", string(op), args)
  10.   // 根据是否传有Context上下文，调用不同的执行command/commandContext
  11.     if ctx == nil {
  12.         return runner.exec.Command(iptablesCmd, fullArgs...).CombinedOutput()
  13.     }
  14.     return runner.exec.CommandContext(ctx, iptablesCmd, fullArgs...).CombinedOutput()
  15. }
  17. //支持的iptables操作commands
  18. type operation string
  20. //runner.exec实现是osexec命令的执行
  21. func New() Interface {
  22.     return &executor{}
  23. }
  25. // Command is part of the Interface interface.
  26. func (executor *executor) Command(cmd string, args ...string) Cmd {
  27.     return (*cmdWrapper)(osexec.Command(cmd, args...))
  28. }
  30. // CommandContext is part of the Interface interface.
  31. func (executor *executor) CommandContext(ctx context.Context, cmd string, args ...string) Cmd {
  32.     return (*cmdWrapper)(osexec.CommandContext(ctx, cmd, args...))
  33. }

runner.GetVersion() 获取系统安装的iptables版本信息，格式为 “X.Y.Z”

!FILENAME pkg/util/iptables/iptables.go:218

  1. func (runner *runner) GetVersion() (string, error) {
  2.     return getIPTablesVersionString(runner.exec, runner.protocol)
  3. }
  5. func getIPTablesVersionString(exec utilexec.Interface, protocol Protocol) (string, error) {
  6.     // 执行命令"iptables or ip6tables --version"
  7.     iptablesCmd := iptablesCommand(protocol)   
  8.     bytes, err := exec.Command(iptablesCmd, "--version").CombinedOutput()
  9.     if err != nil {
  10.         return "", err
  11.     }
  12.   // 正则匹配，查找版本字符串，格式为 "X.Y.Z"
  13.     versionMatcher := regexp.MustCompile("v([0-9]+(\\.[0-9]+)+)")
  14.     match := versionMatcher.FindStringSubmatch(string(bytes))
  15.     if match == nil {
  16.         return "", fmt.Errorf("no iptables version found in string: %s", bytes)
  17.     }
  18.     return match[1], nil
  19. }

runner.EnsureChain() “-N” 检测指定的规则链是否存在，如果不存则创建此链，存在则返回true

!FILENAME pkg/util/iptables/iptables.go:223

  1. func (runner *runner) EnsureChain(table Table, chain Chain) (bool, error) {
  2.     fullArgs := makeFullArgs(table, chain)
  4.     runner.mu.Lock()
  5.     defer runner.mu.Unlock()
  7.   //执行"iptables -t $tableName -N $chainName"
  8.     out, err := runner.run(opCreateChain, fullArgs)
  9.     if err != nil {
  10.         if ee, ok := err.(utilexec.ExitError); ok {
  11.             if ee.Exited() && ee.ExitStatus() == 1 {
  12.                 return true, nil
  13.             }
  14.         }
  15.         return false, fmt.Errorf("error creating chain %q: %v: %s", chain, err, out)
  16.     }
  17.     return false, nil
  18. }

runner.FlushChain() “-F” 清空指定链

!FILENAME pkg/util/iptables/iptables.go:242

  1. func (runner *runner) FlushChain(table Table, chain Chain) error {
  2.     fullArgs := makeFullArgs(table, chain)
  3.   //...
  4.   //执行"iptables -t $tableName -F $chainName"
  5.     out, err := runner.run(opFlushChain, fullArgs)
  6.   //...
  7. }

runner.DeleteChain() “-X” 删除指定的链

!FILENAME pkg/util/iptables/iptables.go:256

  1. func (runner *runner) DeleteChain(table Table, chain Chain) error {
  2.     fullArgs := makeFullArgs(table, chain)
  3.   //...
  4.   //执行"iptables -t $tableName -X $chainName"
  5.     out, err := runner.run(opDeleteChain, fullArgs)
  6.   //...
  7. }

runner.EnsureRule() 检测规则是否存在，不存在则指定的”表内链上”, 指定position添加规则

!FILENAME pkg/util/iptables/iptables.go:271

  1. func (runner *runner) EnsureRule(position RulePosition, table Table, chain Chain, args ...string) (bool, error) {
  2.     fullArgs := makeFullArgs(table, chain, args...)
  4.     runner.mu.Lock()
  5.     defer runner.mu.Unlock()
  6.   // 检测规则是否存在
  7.     exists, err := runner.checkRule(table, chain, args...)
  8.     if err != nil {
  9.         return false, err
  10.     }
  11.     if exists {
  12.         return true, nil
  13.     }
  14.   // RulePosition "-I" "-A"
  15.   // 指定链序插入规则,执行"iptables  -I $chainName -t $tableName ... "
  16.   // 链末添加规则,执行"iptables  -A $chainName -t $tableName ... "    
  17.     out, err := runner.run(operation(position), fullArgs)  
  18.     if err != nil {
  19.         return false, fmt.Errorf("error appending rule: %v: %s", err, out)
  20.     }
  21.     return false, nil
  22. }
  24. //checkRule()先判断iptables是否支持"-C"flag,调用不同版本的检测rule的方法
  25. func (runner *runner) checkRule(table Table, chain Chain, args ...string) (bool, error) {
  26.     if runner.hasCheck {
  27.         return runner.checkRuleUsingCheck(makeFullArgs(table, chain, args...))
  28.     }
  29.     return runner.checkRuleWithoutCheck(table, chain, args...)
  30. }
  32. //支持"-C"flag
  33. func (runner *runner) checkRuleUsingCheck(args []string) (bool, error) {
  34.     ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
  35.     defer cancel()
  36.   //...
  37.   //执行"iptables -wait -C $chainName -t $tableName ... "
  38.     out, err := runner.runContext(ctx, opCheckRule, args)
  39.   //...
  40. }
  42.  //不支持"-C"flag，为了兼容iptables版本<1.4.11 
  43. func (runner *runner) checkRuleWithoutCheck(table Table, chain Chain, args ...string) (bool, error) {
  44.   // 'iptables-save -t $tableName'
  45.     iptablesSaveCmd := iptablesSaveCommand(runner.protocol)
  46.     klog.V(1).Infof("running %s -t %s", iptablesSaveCmd, string(table))
  47.     out, err := runner.exec.Command(iptablesSaveCmd, "-t", string(table)).CombinedOutput()
  48.     if err != nil {
  49.         return false, fmt.Errorf("error checking rule: %v", err)
  50.     }
  52.     //移除引号
  53.     var argsCopy []string
  54.     for i := range args {
  55.         tmpField := strings.Trim(args[i], "\"")
  56.         tmpField = trimhex(tmpField)
  57.         argsCopy = append(argsCopy, strings.Fields(tmpField)...)
  58.     }
  59.     argset := sets.NewString(argsCopy...)
  61.     for _, line := range strings.Split(string(out), "\n") {
  62.         var fields = strings.Fields(line)
  64.         //检测rule的链是否一致
  65.         if !strings.HasPrefix(line, fmt.Sprintf("-A %s", string(chain))) || len(fields) != len(argsCopy)+2 {
  66.             continue
  67.         }
  69.     // 移除所有引号
  70.         for i := range fields {
  71.             fields[i] = strings.Trim(fields[i], "\"")
  72.             fields[i] = trimhex(fields[i])
  73.         }
  75.         //字符集匹配查找是否存在
  76.         if sets.NewString(fields...).IsSuperset(argset) {
  77.             return true, nil
  78.         }
  79.         klog.V(5).Infof("DBG: fields is not a superset of args: fields=%v  args=%v", fields, args)
  80.     }
  82.     return false, nil
  83. }

runner.DeleteRule() “-D” 指定的”表中链上”删除规则

!FILENAME pkg/util/iptables/iptables.go:292

  1. func (runner *runner) DeleteRule(table Table, chain Chain, args ...string) error {
  2.     fullArgs := makeFullArgs(table, chain, args...)
  3.   //...
  4.   //检测规则是否存在
  5.     exists, err := runner.checkRule(table, chain, args...)
  6.   //...
  7.   //执行"iptables -D $chainName -t $tableName ..."
  8.     out, err := runner.run(opDeleteRule, fullArgs)
  9.   //...
  10. }

runner.SaveInto() 保存指定表的iptables规则集（buffer内）

!FILENAME pkg/util/iptables/iptables.go:317

  1. func (runner *runner) SaveInto(table Table, buffer *bytes.Buffer) error {
  2.   //...
  3.   // 执行 "iptables-save -t $tableName"
  4.     iptablesSaveCmd := iptablesSaveCommand(runner.protocol)
  5.     args := []string{"-t", string(table)}
  7.     cmd := runner.exec.Command(iptablesSaveCmd, args...)
  9.     cmd.SetStdout(buffer)
  10.     cmd.SetStderr(buffer)
  11.     return cmd.Run()
  12. }

runner.Restore() 装载指定表由iptables-save保存的规则集(从标准输入接收输入)

!FILENAME pkg/util/iptables/iptables.go:340

  1. func (runner *runner) Restore(table Table, data []byte, flush FlushFlag, counters RestoreCountersFlag) error {
  2.     // "iptables-restore -T $tableName"
  3.     args := []string{"-T", string(table)}
  4.     return runner.restoreInternal(args, data, flush, counters) //call and return
  5. }
  7. // restoreInternal()参数组装和iptables-restore命令恢复规则集data
  8. func (runner *runner) restoreInternal(args []string, data []byte, flush FlushFlag, counters RestoreCountersFlag) error {
  9.     runner.mu.Lock()
  10.     defer runner.mu.Unlock()
  12.     trace := utiltrace.New("iptables restore")
  13.     defer trace.LogIfLong(2 * time.Second)
  15.   //参数的组装 "--noflush" "--counters" "--wait"
  16.     if !flush {
  17.         args = append(args, "--noflush")
  18.     }
  19.     if counters {
  20.         args = append(args, "--counters")
  21.     }
  22.     if len(runner.restoreWaitFlag) == 0 {
  23.         locker, err := grabIptablesLocks(runner.lockfilePath)
  24.         if err != nil {
  25.             return err
  26.         }
  27.         trace.Step("Locks grabbed")
  28.         defer func(locker iptablesLocker) {
  29.             if err := locker.Close(); err != nil {
  30.                 klog.Errorf("Failed to close iptables locks: %v", err)
  31.             }
  32.         }(locker)
  33.     }
  36.     fullArgs := append(runner.restoreWaitFlag, args...)
  37.     iptablesRestoreCmd := iptablesRestoreCommand(runner.protocol)
  38.     klog.V(4).Infof("running %s %v", iptablesRestoreCmd, fullArgs)
  40.   // "iptables-restore -T $tableName --wait --noflush --counters < data"
  41.     cmd := runner.exec.Command(iptablesRestoreCmd, fullArgs...)
  42.   //从标准输入接受输入规则集data
  43.     cmd.SetStdin(bytes.NewBuffer(data))
  44.   //command对象执行与输出反馈
  45.     b, err := cmd.CombinedOutput()
  46.     if err != nil {
  47.         return fmt.Errorf("%v (%s)", err, b)
  48.     }
  49.     return nil
  50. }

runner.RestoreAll() 如同上Restore()，调用命令iptables-restore装载所有备份规则集

!FILENAME pkg/util/iptables/iptables.go:347

  1. func (runner *runner) RestoreAll(data []byte, flush FlushFlag, counters RestoreCountersFlag) error {
  2.     args := make([]string, 0)
  3.   //同上，无参数限制
  4.     return runner.restoreInternal(args, data, flush, counters) 
  5. }

runner.AddReloadFunc() 注册reload回调函数，实现iptables reload重新加载规则

!FILENAME pkg/util/iptables/iptables.go:679

  1. func (runner *runner) AddReloadFunc(reloadFunc func()) {
  2.     runner.mu.Lock()
  3.     defer runner.mu.Unlock()
  5.     //是否已启动监听
  6.     if !runner.hasListener {
  7.         runner.connectToFirewallD()  //启动D-bus监听
  8.     }
  9.     runner.reloadFuncs = append(runner.reloadFuncs, reloadFunc) //注册信号触发回调func
  10. }
  12. //通过Linux内核D-bus机制实现对FirewallD进程的信号监听与处理（实现reload iptables规则）
  13. func (runner *runner) connectToFirewallD() {
  14.     bus, err := runner.dbus.SystemBus()
  15.     if err != nil {
  16.         klog.V(1).Infof("Could not connect to D-Bus system bus: %s", err)
  17.         return
  18.     }
  19.     runner.hasListener = true
  21.   //SystemBus对象添加匹配规则定义（firewalld）
  22.     rule := fmt.Sprintf("type='signal',sender='%s',path='%s',interface='%s',member='Reloaded'", firewalldName, firewalldPath, firewalldInterface)
  23.     bus.BusObject().Call("org.freedesktop.DBus.AddMatch", 0, rule)
  25.     rule = fmt.Sprintf("type='signal',interface='org.freedesktop.DBus',member='NameOwnerChanged',path='/org/freedesktop/DBus',sender='org.freedesktop.DBus',arg0='%s'", firewalldName)
  26.     bus.BusObject().Call("org.freedesktop.DBus.AddMatch", 0, rule)
  28.     runner.signal = make(chan *godbus.Signal, 10)
  29.     bus.Signal(runner.signal)
  31.     go runner.dbusSignalHandler(bus)  //D-Bus信号监听处理Handler
  32. }
  34. //goroutine监听D-Bus信号，监听FirewallD发生变化和reload信号则reload规则集
  35. func (runner *runner) dbusSignalHandler(bus utildbus.Connection) {
  36.     firewalld := bus.Object(firewalldName, firewalldPath)
  38.     for s := range runner.signal {
  39.         if s == nil {
  40.             // 反注册dbus
  41.             bus.Signal(runner.signal)
  42.             return
  43.         }
  45.         switch s.Name {
  46.         case "org.freedesktop.DBus.NameOwnerChanged":  //信号：指定名称的拥有者发生了变化
  47.             name := s.Body[0].(string)
  48.             new_owner := s.Body[2].(string)
  50.       // 信号名称为"org.fedoraproject.FirewallD1"
  51.             if name != firewalldName || len(new_owner) == 0 {
  52.                 continue
  53.             }
  55.             firewalld.Call(firewalldInterface+".getDefaultZone", 0)
  57.       runner.reload()  //重新加载与同步规则(遍历调用runner.reloadFuncs())
  58.         case firewalldInterface + ".Reloaded":
  59.             runner.reload()
  60.         }
  61.     }
  62. }

runner.Destroy() D-bus监听注消

!FILENAME pkg/util/iptables/iptables.go:218

  1. func (runner *runner) Destroy() {
  2.     if runner.signal != nil {
  3.         runner.signal <- nil            //D-Bug信号channel置为空实现反注册
  4.     }
  5. }

上面为kube-proxy第三层的iptables Proxier代码分析所有内容，对于另外两种模式ipvs、userspace模式的proxier实现代码分析可查询userspace-mode proxier和ipvs-mode proxier文章内容。

~本文 END~