14.深入k8s：kube-proxy ipvs及其源码分析

Posted on by luozhiyun

82062770_p0

转载请声明出处哦~，本篇文章发布于luozhiyun的博客：https://www.luozhiyun.com

源码版本是1.19

这一篇是讲service，但是基础使用以及基本概念由于官方实在是写的比较完整了，我没有必要复述一遍，所以还不太清楚的小伙伴们可以去看官方的文档：https://kubernetes.io/docs/concepts/services-networking/service/。

IPVS 概述

在 Kubernetes 集群中，每个 Node 运行一个 kube-proxy 进程。kube-proxy 负责为 Service 实现了一种 VIP（虚拟 IP）的形式。

从官方文档介绍来看：

从 Kubernetes v1.0 开始，您已经可以使用 userspace 代理模式。 Kubernetes v1.1 添加了 iptables 模式代理，在 Kubernetes v1.2 中，kube-proxy 的 iptables 模式成为默认设置。 Kubernetes v1.8 添加了 ipvs 代理模式。

现在我们看的源码是基于1.19，所以现在默认是ipvs代理模式。

LVS是国内章文嵩博士开发并贡献给社区的，主要由ipvs和ipvsadm组成。

ipvs是工作在内核态的4层负载均衡，基于内核底层netfilter实现，netfilter主要通过各个链的钩子实现包处理和转发。ipvs由ipvsadm提供简单的CLI接口进行ipvs配置。由于ipvs工作在内核态，只处理四层协议，因此只能基于路由或者NAT进行数据转发，可以把ipvs当作一个特殊的路由器网关，这个网关可以根据一定的算法自动选择下一跳。

IPVS vs IPTABLES

  • iptables 使用链表，ipvs 使用哈希表；
  • iptables 只支持随机、轮询两种负载均衡算法而 ipvs 支持的多达 8 种；
  • ipvs 还支持 realserver 运行状况检查、连接重试、端口映射、会话保持等功能。

IPVS用法

IPVS可以通过ipvsadm 命令进行配置，如-L列举，-A添加，-D删除。

如下命令创建一个service实例172.17.0.1:32016，-t指定监听的为TCP端口，-s指定算法为轮询算法rr(Round Robin)，ipvs支持简单轮询(rr)、加权轮询(wrr)、最少连接(lc)、源地址或者目标地址散列(sh、dh)等10种调度算法。

ipvsadm -A -t 172.17.0.1:32016 -s rr

在添加调度算法的时候还需要用-r指定server地址，-w指定权值，-m指定转发模式，-m设置masquerading表示NAT模式（-g为gatewaying，即直连路由模式），如下所示：

ipvsadm -a -t 172.17.0.1:32016 -r 10.244.1.2:8080 -m -w 1
ipvsadm -a -t 172.17.0.1:32016 -r 10.244.1.3:8080 -m -w 1
ipvsadm -a -t 172.17.0.1:32016 -r 10.244.3.2:8080 -m -w 1

Service ClusterIP原理

不清楚iptables调用链的同学可以先看看：https://www.zsythink.net/archives/1199了解一下。

我们这里使用上一篇HPA的一个例子：

apiVersion: apps/v1 
kind: Deployment
metadata:
  name: hpatest
spec:
  replicas: 1
  selector:
    matchLabels:
      app: hpatest     
  template: 
    metadata:
      labels:
        app: hpatest
    spec:
      containers:
      - name: hpatest
        image: nginx
        imagePullPolicy: IfNotPresent
        command: ["/bin/sh"]
        args: ["-c","/usr/sbin/nginx; while true;do echo `hostname -I` > /usr/share/nginx/html/index.html; sleep 120;done"]
        ports: 
        - containerPort: 80
        resources:
          requests:
            cpu: 1m
            memory: 100Mi
          limits:
            cpu: 3m
            memory: 400Mi  
---
apiVersion: v1
kind: Service
metadata:
  name: hpatest-svc
spec:
  selector:
    app: hpatest
  ports:
  - port: 80
    targetPort: 80
    protocol: TCP

创建好之后，看看svc：

# kubectl get svc
NAME          TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)    AGE
hpatest-svc   ClusterIP   10.68.196.212   <none>        80/TCP     2m44s

然后我们查看ipvs配置情况：

# ipvsadm -S -n | grep 10.68.196.212

-A -t 10.68.196.212:80 -s rr
-a -t 10.68.196.212:80 -r 172.20.0.251:80 -m -w 1

-S表示输出所保存的规则，-n表示以数字的形式输出ip和端口。可以看到ipvs的LB IP为ClusterIP，算法为rr，RS为Pod的IP。使用的模式为NAT模式。

当我们创建Service之后，kube-proxy 首先会在宿主机上创建一个虚拟网卡（叫作：kube-ipvs0），并为它分配 Service VIP 作为 IP 地址，如下所示：

# ip addr show kube-ipvs0
7: kube-ipvs0: <BROADCAST,NOARP> mtu 1500 qdisc noop state DOWN group default
    link/ether 12:bb:85:91:96:4d brd ff:ff:ff:ff:ff:ff
    ...
    inet 10.68.196.212/32 brd 10.68.196.212 scope global kube-ipvs0
       valid_lft forever preferred_lft forever

下面看看ClusterIP如何传递的。使用命令：iptables -t nat -nvL可以看到由很多Chain，ClusterIP访问方式为：

PREROUTING --> KUBE-SERVICES --> KUBE-CLUSTER-IP --> INPUT --> KUBE-FIREWALL --> POSTROUTING

当使用命令链接服务时：

 curl  10.68.196.212:80

由于10.96.54.11就在本地，所以会以这个IP作为出口地址，即源IP和目标IP都是10.96.54.11，此时相当于：

 10.68.196.212:xxxx ->  10.68.196.212:80

然后经过ipvs，ipvs会从RS ip列中选择其中一个Pod ip作为目标IP:

10.68.196.212:xxxx ->  10.68.196.212:80
                   |
                   | IPVS
                   v
172.20.0.251:xxxx   ->  172.20.0.251:80

查看OUTPUT规则：

# iptables-save 
-A OUTPUT -m comment --comment "kubernetes service portals" -j KUBE-SERVICES
-A KUBE-SERVICES ! -s 172.20.0.0/16 -m comment --comment "Kubernetes service cluster ip + port for masquerade purpose" -m set --match-set KUBE-CLUSTER-IP dst,dst -j KUBE-MARK-MASQ
-A KUBE-MARK-MASQ -j MARK --set-xmark 0x4000/0x4000

如上规则的意思就是除了Pod以外访问ClusterIP的包都打上0x4000/0x4000。

到了POSTROUTING链:

-A POSTROUTING -m comment --comment "kubernetes postrouting rules" -j KUBE-POSTROUTING
-A KUBE-POSTROUTING -m comment --comment "kubernetes service traffic requiring SNAT" -m mark --mark 0x4000/0x4000 -j MASQUERADE

如上规则的意思就是只要匹配mark0x4000/0x4000的包都做SNAT，由于172.20.0.251是从flannel.1出去的，因此源ip会改成flannel.1的ip 172.20.0.0：

10.68.196.212:xxxx -> 10.68.196.212:80
                    |
                    | IPVS
                    v
10.68.196.212:xxxx  -> 172.20.0.251:80
                     |
                     | MASQUERADE
                     v
172.20.0.0:xxxx     -> 172.20.0.251:80

最后通过VXLAN隧道发到Pod的Node上，转发给Pod的veth，回包通过路由到达源Node节点，源Node节点通过之前的MASQUERADE再把目标IP还原为172.20.0.251。

kube-proxy ipvs 源码分析

初始化ipvs

文件位置：cmd/kube-proxy/app/server_others.go

kube-proxy启动的时候会调用NewProxyServer初始化ipvs 代理：

func newProxyServer(
    config *proxyconfigapi.KubeProxyConfiguration,
    cleanupAndExit bool,
    master string) (*ProxyServer, error) {

    ...
    //获取代理模式userspace iptables ipvs
    proxyMode := getProxyMode(string(config.Mode), canUseIPVS, iptables.LinuxKernelCompatTester{})
    ...
    //代理模式是iptables
    if proxyMode == proxyModeIPTables {
        ...
    //  代理模式是ipvs
    } else if proxyMode == proxyModeIPVS {
        klog.V(0).Info("Using ipvs Proxier.")
        //判断是够启用了 ipv6 双栈
        if utilfeature.DefaultFeatureGate.Enabled(features.IPv6DualStack) {
            ...
        } else {
            ...
            //初始化 ipvs 模式的 proxier
            proxier, err = ipvs.NewProxier(
                ...
            )
        }
        ...
    } else {
        ...
    }

    return &ProxyServer{
        ...
    }, nil
}

NewProxyServer方法会根据proxyMode来选择是IPVS还是IPTables，ipvs会调用ipvs.NewProxier方法来初始化一个proxier。

NewProxier

func NewProxier(... ) (*Proxier, error) {
    ...  
    //对于 SNAT iptables 规则生成 masquerade 标记
    masqueradeValue := 1 << uint(masqueradeBit)
    ...
    //设置默认调度算法 rr
    if len(scheduler) == 0 {
        klog.Warningf("IPVS scheduler not specified, use %s by default", DefaultScheduler)
        scheduler = DefaultScheduler
    }
    // healthcheck服务器对象创建
    serviceHealthServer := healthcheck.NewServiceHealthServer(hostname, recorder)

    ...
    //初始化 proxier
    proxier := &Proxier{
        ...
    } 
    //初始化 ipset 规则
    proxier.ipsetList = make(map[string]*IPSet)
    for _, is := range ipsetInfo {
        proxier.ipsetList[is.name] = NewIPSet(ipset, is.name, is.setType, isIPv6, is.comment)
    }
    burstSyncs := 2
    klog.V(2).Infof("ipvs(%s) sync params: minSyncPeriod=%v, syncPeriod=%v, burstSyncs=%d",
        ipt.Protocol(), minSyncPeriod, syncPeriod, burstSyncs)
    //初始化 syncRunner
    proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs)
    //启动 gracefuldeleteManager
    proxier.gracefuldeleteManager.Run()
    return proxier, nil
}

这个方法主要做了如下几件事：

  1. 对于 SNAT iptables 规则生成 masquerade 标记；
  2. 设置默认调度算法 rr；
  3. healthcheck服务器对象创建；
  4. 初始化 proxier；
  5. 初始化 ipset 规则；
  6. 初始化 syncRunner；
  7. 启动 gracefuldeleteManager；

这个方法在初始化syncRunner的时候设置了proxier.syncProxyRules方法作为一个参数构建了同步运行器syncRunner。

调用同步运行器

文件位置：cmd/kube-proxy/app/server.go

func (s *ProxyServer) Run() error {
    ...
    //调用ipvs的SyncLoop方法
    go s.Proxier.SyncLoop()

    return <-errCh
}

kube-proxy在启动的时候会初始化完ProxyServer 对象后，会调用runLoop方法，然后调用到ProxyServer的Run方法中，最后调用ipvs的SyncLoop方法。

func (proxier *Proxier) SyncLoop() { 
    ...
    proxier.syncRunner.Loop(wait.NeverStop)     //执行NewBoundedFrequencyRunner对象Loop
}

func (bfr *BoundedFrequencyRunner) Loop(stop <-chan struct{}) {
    bfr.timer.Reset(bfr.maxInterval)
    for {
        select {
        case <-stop:
            bfr.stop()
            return
        case <-bfr.timer.C():           //定时器方式执行
            bfr.tryRun()
        case <-bfr.run:                 //按需方式执行（发送运行指令信号）
            bfr.tryRun()
        }
    }
}

func (bfr *BoundedFrequencyRunner) tryRun() {
    bfr.mu.Lock()
    defer bfr.mu.Unlock()

  //限制条件允许运行func
    if bfr.limiter.TryAccept() {
         bfr.fn()                                 // 重点执行部分，调用func，上下文来看此处就是
                                                  // 对syncProxyRules()的调用
        bfr.lastRun = bfr.timer.Now()             // 记录运行时间
        bfr.timer.Stop()                          
        bfr.timer.Reset(bfr.maxInterval)          // 重设下次运行时间
        klog.V(3).Infof("%s: ran, next possible in %v, periodic in %v", bfr.name, bfr.minInterval, bfr.maxInterval)
        return
    }

  //限制条件不允许运行，计算下次运行时间
  elapsed := bfr.timer.Since(bfr.lastRun)    // elapsed:上次运行时间到现在已过多久
  nextPossible := bfr.minInterval - elapsed  // nextPossible:下次运行至少差多久（最小周期）
  nextScheduled := bfr.maxInterval - elapsed // nextScheduled:下次运行最迟差多久(最大周期)
    klog.V(4).Infof("%s: %v since last run, possible in %v, scheduled in %v", bfr.name, elapsed, nextPossible, nextScheduled)

    if nextPossible < nextScheduled {
        bfr.timer.Stop()
        bfr.timer.Reset(nextPossible)
        klog.V(3).Infof("%s: throttled, scheduling run in %v", bfr.name, nextPossible)
    }
}

SyncLoop方法会调用到proxier的syncRunner实例设置的syncProxyRules方法。

同步运行器

syncProxyRules方法比较长，所以这里就分开来一步步的讲，跟好代码节奏来就好了。

代码位置：pkg/proxy/ipvs/proxier.go

    //更新 service 与 endpoint变化信息
    serviceUpdateResult := proxy.UpdateServiceMap(proxier.serviceMap, proxier.serviceChanges)
    endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges)

    staleServices := serviceUpdateResult.UDPStaleClusterIP 
    // 合并 service 列表
    for _, svcPortName := range endpointUpdateResult.StaleServiceNames {
        if svcInfo, ok := proxier.serviceMap[svcPortName]; ok && svcInfo != nil && conntrack.IsClearConntrackNeeded(svcInfo.Protocol()) {
            klog.V(2).Infof("Stale %s service %v -> %s", strings.ToLower(string(svcInfo.Protocol())), svcPortName, svcInfo.ClusterIP().String())
            staleServices.Insert(svcInfo.ClusterIP().String())
            for _, extIP := range svcInfo.ExternalIPStrings() {
                staleServices.Insert(extIP)
            }
        }
    }

这里是同步与新更service和endpoints，然后 合并 service 列表。

    //nat链
    proxier.natChains.Reset()
    //nat规则
    proxier.natRules.Reset()
    //filter链
    proxier.filterChains.Reset()
    //filter规则
    proxier.filterRules.Reset()

    // Write table headers.
    writeLine(proxier.filterChains, "*filter")
    writeLine(proxier.natChains, "*nat")
    //创建kubernetes的表连接链数据
    proxier.createAndLinkeKubeChain()

这里会重置链表规则，然后调用createAndLinkeKubeChain方法创建kubernetes的表连接链数据，下面我们看看createAndLinkeKubeChain方法：

func (proxier *Proxier) createAndLinkeKubeChain() {
    //通过iptables-save获取现有的filter和NAT表存在的链数据
    existingFilterChains := proxier.getExistingChains(proxier.filterChainsData, utiliptables.TableFilter)
    existingNATChains := proxier.getExistingChains(proxier.iptablesData, utiliptables.TableNAT)

    // Make sure we keep stats for the top-level chains
    //里面保存了NAT表链和Filter表链
    // NAT表链： KUBE-SERVICES / KUBE-POSTROUTING / KUBE-FIREWALL
    //          KUBE-NODE-PORT / KUBE-LOAD-BALANCER / KUBE-MARK-MASQ
    // Filter表链： KUBE-FORWARD
    for _, ch := range iptablesChains {
        //不存在则创建链，创建顶层链
        if _, err := proxier.iptables.EnsureChain(ch.table, ch.chain); err != nil {
            klog.Errorf("Failed to ensure that %s chain %s exists: %v", ch.table, ch.chain, err)
            return

        }
        //nat表写链
        if ch.table == utiliptables.TableNAT {
            if chain, ok := existingNATChains[ch.chain]; ok {
                writeBytesLine(proxier.natChains, chain)
            } else {
                writeLine(proxier.natChains, utiliptables.MakeChainLine(kubePostroutingChain))
            }
        // filter表写链
        } else {
            if chain, ok := existingFilterChains[KubeForwardChain]; ok {
                writeBytesLine(proxier.filterChains, chain)
            } else {
                writeLine(proxier.filterChains, utiliptables.MakeChainLine(KubeForwardChain))
            }
        }
    }
    // 默认链下创建kubernete服务专用跳转规则
    // iptables -I OUTPUT -t nat --comment "kubernetes service portals" -j KUBE-SERVICES
    // iptables -I PREROUTING -t nat --comment "kubernetes service portals" -j KUBE-SERVICES
    // iptables -I POSTROUTING -t nat --comment "kubernetes postrouting rules" -j KUBE-POSTROUTING
    // iptables -I FORWARD -t filter --comment "kubernetes forwarding rules" -j KUBE-FORWARD
    for _, jc := range iptablesJumpChain {
        args := []string{"-m", "comment", "--comment", jc.comment, "-j", string(jc.to)}
        if _, err := proxier.iptables.EnsureRule(utiliptables.Prepend, jc.table, jc.from, args...); err != nil {
            klog.Errorf("Failed to ensure that %s chain %s jumps to %s: %v", jc.table, jc.from, jc.to, err)
        }
    }

}

createAndLinkeKubeChain方法首先会获取现存的filter和NAT表，然后再遍历iptablesChains。

iptablesChains里面保存了NAT表链和Filter表链：NAT表链 KUBE-SERVICES / KUBE-POSTROUTING / KUBE-FIREWALL KUBE-NODE-PORT / KUBE-LOAD-BALANCER / KUBE-MARK-MASQ；Filter表链 KUBE-FORWARD；

然后再根据iptablesJumpChain创建跳转规则。

下面回到syncProxyRules往下走。

    // 创建 dummy interface kube-ipvs0
    _, err = proxier.netlinkHandle.EnsureDummyDevice(DefaultDummyDevice)
    if err != nil {
        klog.Errorf("Failed to create dummy interface: %s, error: %v", DefaultDummyDevice, err)
        return
    }

    // 创建默认的 ipset 规则，http://ipset.netfilter.org/
    for _, set := range proxier.ipsetList {
        if err := ensureIPSet(set); err != nil {
            return
        }
        set.resetEntries()
    }

设置默认Dummy接口，并确定ipsets规则已存在的集合，ipset相关可以看：http://ipset.netfilter.org/。

下面会遍历proxier.serviceMap，对每一个服务创建 ipvs 规则，比较长，也分开说。

for svcName, svc := range proxier.serviceMap {
    ...
    //基于此服务的有效endpoint列表，更新KUBE-LOOP-BACK的ipset集，以备后面生成相应iptables规则(SNAT伪装地址)
    for _, e := range proxier.endpointsMap[svcName] {
        ep, ok := e.(*proxy.BaseEndpointInfo)
        if !ok {
            klog.Errorf("Failed to cast BaseEndpointInfo %q", e.String())
            continue
        }
        if !ep.IsLocal {
            continue
        }
        epIP := ep.IP()
        epPort, err := ep.Port()
        // Error parsing this endpoint has been logged. Skip to next endpoint.
        if epIP == "" || err != nil {
            continue
        }
        entry := &utilipset.Entry{
            IP:       epIP,
            Port:     epPort,
            Protocol: protocol,
            IP2:      epIP,
            SetType:  utilipset.HashIPPortIP,
        }
        // 校验KUBE-LOOP-BACK集合entry记录项
        if valid := proxier.ipsetList[kubeLoopBackIPSet].validateEntry(entry); !valid {
            klog.Errorf("%s", fmt.Sprintf(EntryInvalidErr, entry, proxier.ipsetList[kubeLoopBackIPSet].Name))
            continue
        }
        // 插入此entry记录至active记录队列
        proxier.ipsetList[kubeLoopBackIPSet].activeEntries.Insert(entry.String())
    }
    ...
}

这一段是根据有效endpoint列表，更新KUBE-LOOP-BACK的ipset集，以备后面生成相应iptables规则(SNAT伪装地址)；

for svcName, svc := range proxier.serviceMap {
    ...
    //构建ipset entry
    entry := &utilipset.Entry{
        IP:       svcInfo.ClusterIP().String(),
        Port:     svcInfo.Port(),
        Protocol: protocol,
        SetType:  utilipset.HashIPPort,
    }
    // add service Cluster IP:Port to kubeServiceAccess ip set for the purpose of solving hairpin.
    // proxier.kubeServiceAccessSet.activeEntries.Insert(entry.String())
    // 类型校验ipset entry
    if valid := proxier.ipsetList[kubeClusterIPSet].validateEntry(entry); !valid {
        klog.Errorf("%s", fmt.Sprintf(EntryInvalidErr, entry, proxier.ipsetList[kubeClusterIPSet].Name))
        continue
    }
    // 名为KUBE-CLUSTER-IP的ipset集插入entry,以备后面统一生成IPtables规则
    proxier.ipsetList[kubeClusterIPSet].activeEntries.Insert(entry.String())
    // ipvs call
    // 构建ipvs虚拟服务器VS服务对象
    serv := &utilipvs.VirtualServer{
        Address:   svcInfo.ClusterIP(),
        Port:      uint16(svcInfo.Port()),
        Protocol:  string(svcInfo.Protocol()),
        Scheduler: proxier.ipvsScheduler,
    }
    // Set session affinity flag and timeout for IPVS service
    // 设置IPVS服务的会话保持标志和超时时间
    if svcInfo.SessionAffinityType() == v1.ServiceAffinityClientIP {
        serv.Flags |= utilipvs.FlagPersistent
        serv.Timeout = uint32(svcInfo.StickyMaxAgeSeconds())
    }
    // We need to bind ClusterIP to dummy interface, so set `bindAddr` parameter to `true` in syncService()
    // 将clusterIP绑定至dummy虚拟接口上，syncService()处理中需置bindAddr地址为True
    if err := proxier.syncService(svcNameString, serv, true, bindedAddresses); err == nil {
        activeIPVSServices[serv.String()] = true
        activeBindAddrs[serv.Address.String()] = true
        // ExternalTrafficPolicy only works for NodePort and external LB traffic, does not affect ClusterIP
        // So we still need clusterIP rules in onlyNodeLocalEndpoints mode.
        //同步endpoints信息，IPVS为VS更新realServer
        if err := proxier.syncEndpoint(svcName, false, serv); err != nil {
            klog.Errorf("Failed to sync endpoint for service: %v, err: %v", serv, err)
        }
    } else {
        klog.Errorf("Failed to sync service: %v, err: %v", serv, err)
    }
    ...
}

ipset集KUBE-CLUSTER-IP更新,以备后面生成相应iptables规则。

for svcName, svc := range proxier.serviceMap {
    ...
    //为 load-balancer类型创建 ipvs 规则
    for _, ingress := range svcInfo.LoadBalancerIPStrings() {
        if ingress != "" { 
            // 构建ipset entry
            entry = &utilipset.Entry{
                IP:       ingress,
                Port:     svcInfo.Port(),
                Protocol: protocol,
                SetType:  utilipset.HashIPPort,
            } 
            if valid := proxier.ipsetList[kubeLoadBalancerSet].validateEntry(entry); !valid {
                klog.Errorf("%s", fmt.Sprintf(EntryInvalidErr, entry, proxier.ipsetList[kubeLoadBalancerSet].Name))
                continue
            }
            //KUBE-LOAD-BALANCER ipset集更新
            proxier.ipsetList[kubeLoadBalancerSet].activeEntries.Insert(entry.String()) 
            //服务指定externalTrafficPolicy=local时,KUBE-LOAD-BALANCER-LOCAL ipset集更新
            if svcInfo.OnlyNodeLocalEndpoints() {
                if valid := proxier.ipsetList[kubeLoadBalancerLocalSet].validateEntry(entry); !valid {
                    klog.Errorf("%s", fmt.Sprintf(EntryInvalidErr, entry, proxier.ipsetList[kubeLoadBalancerLocalSet].Name))
                    continue
                }
                proxier.ipsetList[kubeLoadBalancerLocalSet].activeEntries.Insert(entry.String())
            }
            // 服务的LoadBalancerSourceRanges被指定时，基于源IP保护的防火墙策略开启，KUBE-LOAD-BALANCER-FW ipset集更新
            if len(svcInfo.LoadBalancerSourceRanges()) != 0 { 
                if valid := proxier.ipsetList[kubeLoadbalancerFWSet].validateEntry(entry); !valid {
                    klog.Errorf("%s", fmt.Sprintf(EntryInvalidErr, entry, proxier.ipsetList[kubeLoadbalancerFWSet].Name))
                    continue
                }
                proxier.ipsetList[kubeLoadbalancerFWSet].activeEntries.Insert(entry.String())
                allowFromNode := false
                for _, src := range svcInfo.LoadBalancerSourceRanges() {
                    // ipset call
                    entry = &utilipset.Entry{
                        IP:       ingress,
                        Port:     svcInfo.Port(),
                        Protocol: protocol,
                        Net:      src,
                        SetType:  utilipset.HashIPPortNet,
                    } 
                    // 枚举所有源CIDR白名单列表，KUBE-LOAD-BALANCER-SOURCE-CIDR ipset集更新
                    //cidr：https://cloud.google.com/kubernetes-engine/docs/how-to/flexible-pod-cidr
                    if valid := proxier.ipsetList[kubeLoadBalancerSourceCIDRSet].validateEntry(entry); !valid {
                        klog.Errorf("%s", fmt.Sprintf(EntryInvalidErr, entry, proxier.ipsetList[kubeLoadBalancerSourceCIDRSet].Name))
                        continue
                    }
                    proxier.ipsetList[kubeLoadBalancerSourceCIDRSet].activeEntries.Insert(entry.String())

                    // ignore error because it has been validated
                    _, cidr, _ := net.ParseCIDR(src)
                    if cidr.Contains(proxier.nodeIP) {
                        allowFromNode = true
                    }
                }
                ...
            } 
            // ipvs call
            // 构建ipvs 虚拟主机对象
            serv := &utilipvs.VirtualServer{
                Address:   net.ParseIP(ingress),
                Port:      uint16(svcInfo.Port()),
                Protocol:  string(svcInfo.Protocol()),
                Scheduler: proxier.ipvsScheduler,
            }
            ...
        }
    }
    ...
}

这里为 load-balancer类型创建 ipvs 规则，LoadBalancerSourceRanges和externalTrafficPolicy=local被指定时将对KUBE-LOAD-BALANCER-LOCAL、KUBE-LOAD-BALANCER-FW、KUBE-LOAD-BALANCER-SOURCE-CIDR、KUBE-LOAD-BALANCER-SOURCE-IP ipset集更新,以备后面生成相应iptables规则。

    ...
    //同步 ipset 记录，清理 conntrack
    for _, set := range proxier.ipsetList {
        set.syncIPSetEntries()
    }

    //创建 iptables 规则数据
    proxier.writeIptablesRules()
    // 合并iptables规则
    proxier.iptablesData.Reset()
    proxier.iptablesData.Write(proxier.natChains.Bytes())
    proxier.iptablesData.Write(proxier.natRules.Bytes())
    proxier.iptablesData.Write(proxier.filterChains.Bytes())
    proxier.iptablesData.Write(proxier.filterRules.Bytes())

    klog.V(5).Infof("Restoring iptables rules: %s", proxier.iptablesData.Bytes())
    //基于iptables格式化规则数据，使用iptables-restore刷新iptables规则
    err = proxier.iptables.RestoreAll(proxier.iptablesData.Bytes(), utiliptables.NoFlushTables, utiliptables.RestoreCounters)
    ...

最后这里会刷新iptables 规则，然后创建 iptables 规则数据，将合并iptables规则，iptables-restore刷新iptables规则。

总结

这一篇没有怎么讲service是怎么运行的，怎么使用的，而是选择讲了kube-proxy的ipvs代理是怎么做的，以及在开头讲了ipvs与iptables区别与关系，看不懂的同学需要自己去补充一下iptables相关的知识，文中的 ipvs 的知识我也是现学的，如果有讲解不好的地方欢迎指出。

Reference

https://kubernetes.io/docs/tasks/administer-cluster/dns-debugging-resolution/

https://kubernetes.io/docs/tasks/debug-application-cluster/debug-service/

https://kubernetes.io/docs/concepts/services-networking/service/

https://cloud.google.com/kubernetes-engine/docs/how-to/flexible-pod-cidr

https://github.com/kubernetes/kubernetes/tree/master/pkg/proxy/ipvs

https://zh.wikipedia.org/zh-cn/%E7%BD%91%E7%BB%9C%E5%9C%B0%E5%9D%80%E8%BD%AC%E6%8D%A2

https://segmentfault.com/a/1190000009043962

http://ipset.netfilter.org/