上篇文章 kubernetes service 原理解析 已经分析了 service 原理以 kube-proxy 中三种模式的原理，本篇文章会从源码角度分析 kube-proxy 的设计与实现。

kubernetes 版本: v1.16

kube-proxy 启动流程

前面的文章已经说过 kubernetes 中所有组件都是通过其 run() 方法启动主逻辑的，run() 方法调用之前会进行解析命令行参数、添加默认值等。下面就直接看 kube-proxy 的 run() 方法：

  • 若启动时指定了 --write-config-to 参数，kube-proxy 只将启动的默认参数写到指定的配置文件中，然后退出
  • 初始化 ProxyServer 对象
  • 如果启动参数 --cleanup 设置为 true，则清理 iptables 和 ipvs 规则并退出

k8s.io/kubernetes/cmd/kube-proxy/app/server.go:290

  1. func (o *Options) Run() error {
  2.     defer close(o.errCh)
  3.     // 1.如果指定了 --write-config-to 参数，则将默认的配置文件写到指定文件并退出
  4.     if len(o.WriteConfigTo) > 0 {
  5.         return o.writeConfigFile()
  6.     }
  8.     // 2.初始化 ProxyServer 对象
  9.     proxyServer, err := NewProxyServer(o)
  10.     if err != nil {
  11.         return err
  12.     }
  14.     // 3.如果启动参数 --cleanup 设置为 true，则清理 iptables 和 ipvs 规则并退出
  15.     if o.CleanupAndExit {
  16.         return proxyServer.CleanupAndExit()
  17.     }
  19.     o.proxyServer = proxyServer
  20.     return o.runLoop()
  21. }

Run() 方法中主要调用了 NewProxyServer() 方法来初始化 ProxyServer，然后会调用 runLoop() 启动主循环，继续看初始化 ProxyServer 的具体实现：

  • 初始化 iptables、ipvs 相关的 interface
  • 若启用了 ipvs 则检查内核版本、ipvs 依赖的内核模块、ipset 版本，内核模块主要包括：ip_vs，ip_vs_rr,ip_vs_wrr,ip_vs_sh,nf_conntrack_ipv4,nf_conntrack，若没有相关模块，kube-proxy 会尝试使用 modprobe 自动加载
  • 根据 proxyMode 初始化 proxier，kube-proxy 启动后只运行一种 proxier

k8s.io/kubernetes/cmd/kube-proxy/app/server_others.go:57

  1. func NewProxyServer(o *Options) (*ProxyServer, error) {
  2.     return newProxyServer(o.config, o.CleanupAndExit, o.master)
  3. }
  5. func newProxyServer(
  6.     config *proxyconfigapi.KubeProxyConfiguration,
  7.     cleanupAndExit bool,
  8.     master string) (*ProxyServer, error) {
  9.     ......
  11.     if c, err := configz.New(proxyconfigapi.GroupName); err == nil {
  12.         c.Set(config)
  13.     } else {
  14.         return nil, fmt.Errorf("unable to register configz: %s", err)
  15.     }
  17.     ......
  19.     // 1.关键依赖工具 iptables/ipvs/ipset/dbus
  20.     var iptInterface utiliptables.Interface
  21.     var ipvsInterface utilipvs.Interface
  22.     var kernelHandler ipvs.KernelHandler
  23.     var ipsetInterface utilipset.Interface
  24.     var dbus utildbus.Interface
  26.     // 2.执行 linux 命令行的工具
  27.     execer := exec.New()
  29.     // 3.初始化 iptables/ipvs/ipset/dbus 对象
  30.     dbus = utildbus.New()
  31.     iptInterface = utiliptables.New(execer, dbus, protocol)
  32.     kernelHandler = ipvs.NewLinuxKernelHandler()
  33.     ipsetInterface = utilipset.New(execer)
  35.     // 4.检查该机器是否支持使用 ipvs 模式
  36.     canUseIPVS, _ := ipvs.CanUseIPVSProxier(kernelHandler, ipsetInterface)
  37.     if canUseIPVS {
  38.         ipvsInterface = utilipvs.New(execer)
  39.     }
  41.     if cleanupAndExit {
  42.         return &ProxyServer{
  43.             ......
  44.         }, nil
  45.     }
  47.     // 5.初始化 kube client 和 event client
  48.     client, eventClient, err := createClients(config.ClientConnection, master)
  49.     if err != nil {
  50.         return nil, err
  51.     }
  52.     ......
  54.     // 6.初始化 healthzServer
  55.     var healthzServer *healthcheck.HealthzServer
  56.     var healthzUpdater healthcheck.HealthzUpdater
  57.     if len(config.HealthzBindAddress) > 0 {
  58.         healthzServer = healthcheck.NewDefaultHealthzServer(config.HealthzBindAddress, 2*config.IPTables.SyncPeriod.Duration, recorder, nodeRef)
  59.         healthzUpdater = healthzServer
  60.     }
  62.     // 7.proxier 是一个 interface，每种模式都是一个 proxier
  63.     var proxier proxy.Provider
  65.     // 8.根据 proxyMode 初始化 proxier
  66.     proxyMode := getProxyMode(string(config.Mode), kernelHandler, ipsetInterface, iptables.LinuxKernelCompatTester{})
  67.     ......
  69.     if proxyMode == proxyModeIPTables {
  70.         klog.V(0).Info("Using iptables Proxier.")
  71.         if config.IPTables.MasqueradeBit == nil {
  72.             return nil, fmt.Errorf("unable to read IPTables MasqueradeBit from config")
  73.         }
  75.         // 9.初始化 iptables 模式的 proxier
  76.         proxier, err = iptables.NewProxier(
  77.             .......
  78.         )
  79.         if err != nil {
  80.             return nil, fmt.Errorf("unable to create proxier: %v", err)
  81.         }
  82.         metrics.RegisterMetrics()
  83.     } else if proxyMode == proxyModeIPVS {
  84.         // 10.判断是够启用了 ipv6 双栈
  85.         if utilfeature.DefaultFeatureGate.Enabled(features.IPv6DualStack) {
  86.             ......
  87.             // 11.初始化 ipvs 模式的 proxier
  88.             proxier, err = ipvs.NewDualStackProxier(
  89.                 ......
  90.             )
  91.         } else {
  92.             proxier, err = ipvs.NewProxier(
  93.                 ......
  94.             )
  95.         }
  96.         if err != nil {
  97.             return nil, fmt.Errorf("unable to create proxier: %v", err)
  98.         }
  99.         metrics.RegisterMetrics()
  100.     } else {
  101.         // 12.初始化 userspace 模式的 proxier
  102.         proxier, err = userspace.NewProxier(
  103.             ......
  104.         )
  105.         if err != nil {
  106.             return nil, fmt.Errorf("unable to create proxier: %v", err)
  107.         }
  108.     }
  110.     iptInterface.AddReloadFunc(proxier.Sync)
  111.     return &ProxyServer{
  112.         ......
  113.     }, nil
  114. }

runLoop() 方法主要是启动 proxyServer。

k8s.io/kubernetes/cmd/kube-proxy/app/server.go:311

  1. func (o *Options) runLoop() error {
  2.     // 1.watch 配置文件变化
  3.     if o.watcher != nil {
  4.         o.watcher.Run()
  5.     }
  7.     // 2.以 goroutine 方式启动 proxyServer
  8.     go func() {
  9.         err := o.proxyServer.Run()
  10.         o.errCh <- err
  11.     }()
  13.     for {
  14.         err := <-o.errCh
  15.         if err != nil {
  16.             return err
  17.         }
  18.     }
  19. }

o.proxyServer.Run() 中会启动已经初始化好的所有服务：

  • 设定进程 OOMScore，可通过命令行配置，默认值为 --oom-score-adj="-999"
  • 启动 metric server 和 healthz server，两者分别监听 10256 和 10249 端口
  • 设置内核参数 nf_conntrack_tcp_timeout_established 和 nf_conntrack_tcp_timeout_close_wait
  • 将 proxier 注册到 serviceEventHandler、endpointsEventHandler 中
  • 启动 informer 监听 service 和 endpoints 变化
  • 执行 s.Proxier.SyncLoop()，启动 proxier 主循环

k8s.io/kubernetes/cmd/kube-proxy/app/server.go:527

  1. func (s *ProxyServer) Run() error {
  2.     ......
  4.     // 1.进程 OOMScore，避免进程因 oom 被杀掉，此处默认值为 -999
  5.     var oomAdjuster *oom.OOMAdjuster
  6.     if s.OOMScoreAdj != nil {
  7.         oomAdjuster = oom.NewOOMAdjuster()
  8.         if err := oomAdjuster.ApplyOOMScoreAdj(0, int(*s.OOMScoreAdj)); err != nil {
  9.             klog.V(2).Info(err)
  10.         }
  11.     }
  12.     ......
  14.     // 2.启动 healthz server
  15.     if s.HealthzServer != nil {
  16.         s.HealthzServer.Run()
  17.     }
  19.     // 3.启动 metrics server
  20.     if len(s.MetricsBindAddress) > 0 {
  21.         ......
  22.         go wait.Until(func() {
  23.             err := http.ListenAndServe(s.MetricsBindAddress, proxyMux)
  24.             if err != nil {
  25.                 utilruntime.HandleError(fmt.Errorf("starting metrics server failed: %v", err))
  26.             }
  27.         }, 5*time.Second, wait.NeverStop)
  28.     }
  30.     // 4.配置 conntrack，设置内核参数 nf_conntrack_tcp_timeout_established 和 nf_conntrack_tcp_timeout_close_wait
  31.     if s.Conntracker != nil {
  32.         max, err := getConntrackMax(s.ConntrackConfiguration)
  33.         if err != nil {
  34.             return err
  35.         }
  36.         if max > 0 {
  37.             err := s.Conntracker.SetMax(max)
  38.             ......
  39.         }
  41.         if s.ConntrackConfiguration.TCPEstablishedTimeout != nil && s.ConntrackConfiguration.TCPEstablishedTimeout.Duration > 0 {
  42.             timeout := int(s.ConntrackConfiguration.TCPEstablishedTimeout.Duration / time.Second)
  43.             if err := s.Conntracker.SetTCPEstablishedTimeout(timeout); err != nil {
  44.                 return err
  45.             }
  46.         }
  47.         if s.ConntrackConfiguration.TCPCloseWaitTimeout != nil && s.ConntrackConfiguration.TCPCloseWaitTimeout.Duration > 0 {
  48.             timeout := int(s.ConntrackConfiguration.TCPCloseWaitTimeout.Duration / time.Second)
  49.             if err := s.Conntracker.SetTCPCloseWaitTimeout(timeout); err != nil {
  50.                 return err
  51.             }
  52.         }
  53.     }
  55.     ......
  57.     // 5.启动 informer 监听 Services 和 Endpoints 或者 EndpointSlices 信息
  58.     informerFactory := informers.NewSharedInformerFactoryWithOptions(s.Client, s.ConfigSyncPeriod,
  59.         informers.WithTweakListOptions(func(options *metav1.ListOptions) {
  60.             options.LabelSelector = labelSelector.String()
  61.         }))
  64.     // 6.将 proxier 注册到 serviceConfig、endpointsConfig 中
  65.     serviceConfig := config.NewServiceConfig(informerFactory.Core().V1().Services(), s.ConfigSyncPeriod)
  66.     serviceConfig.RegisterEventHandler(s.Proxier)
  67.     go serviceConfig.Run(wait.NeverStop)
  69.     if utilfeature.DefaultFeatureGate.Enabled(features.EndpointSlice) {
  70.         endpointSliceConfig := config.NewEndpointSliceConfig(informerFactory.Discovery().V1alpha1().EndpointSlices(), s.ConfigSyncPeriod)
  71.         endpointSliceConfig.RegisterEventHandler(s.Proxier)
  72.         go endpointSliceConfig.Run(wait.NeverStop)
  73.     } else {
  74.         endpointsConfig := config.NewEndpointsConfig(informerFactory.Core().V1().Endpoints(), s.ConfigSyncPeriod)
  75.         endpointsConfig.RegisterEventHandler(s.Proxier)
  76.         go endpointsConfig.Run(wait.NeverStop)
  77.     }
  79.     // 7.启动 informer
  80.     informerFactory.Start(wait.NeverStop)
  82.     s.birthCry()
  84.     // 8.启动 proxier 主循环
  85.     s.Proxier.SyncLoop()
  86.     return nil
  87. }

回顾一下整个启动逻辑：

  1. o.Run() --> o.runLoop() --> o.proxyServer.Run() --> s.Proxier.SyncLoop()

o.Run() 中调用了 NewProxyServer() 来初始化 proxyServer 对象，其中包括初始化每种模式对应的 proxier，该方法最终会调用 s.Proxier.SyncLoop() 执行 proxier 的主循环。

proxier 的初始化

看完了启动流程的逻辑代码，接着再看一下各代理模式的初始化，上文已经提到每种模式都是一个 proxier，即要实现 proxy.Provider 对应的 interface，如下所示：

  1. type Provider interface {
  2.     config.EndpointsHandler
  3.     config.EndpointSliceHandler
  4.     config.ServiceHandler
  6.     Sync()
  7.     SyncLoop()
  8. }

首先要实现 service、endpoints 和 endpointSlice 对应的 handler，也就是对 OnAdd、OnUpdate、OnDelete 、OnSynced 四种方法的处理，详细的代码在下文进行讲解。EndpointSlice 是在 v1.16 中新加入的一个 API。Sync() 和 SyncLoop() 是主要用来处理iptables 规则的方法。

iptables proxier 初始化

首先看 iptables 模式的 NewProxier()方法，其函数的具体执行逻辑为：

  • 设置相关的内核参数route_localnet、bridge-nf-call-iptables
  • 生成 masquerade 标记
  • 设置默认调度算法 rr
  • 初始化 proxier 对象
  • 使用 BoundedFrequencyRunner 初始化 proxier.syncRunner，将 proxier.syncProxyRules 方法注入，BoundedFrequencyRunner 是一个管理器用于执行用户注入的函数，可以指定运行的时间策略。

k8s.io/kubernetes/pkg/proxy/iptables/proxier.go:249

  1. func NewProxier(ipt utiliptables.Interface,
  2.     ......
  3. ) (*Proxier, error) {
  4.     // 1.设置相关的内核参数
  5.     if val, _ := sysctl.GetSysctl(sysctlRouteLocalnet); val != 1 {
  6.         ......
  7.     }
  9.     if val, err := sysctl.GetSysctl(sysctlBridgeCallIPTables); err == nil && val != 1 {
  10.         ......
  11.     }
  13.     // 2.设置 masqueradeMark，默认为 0x00004000/0x00004000
  14.     // 用来标记 k8s 管理的报文，masqueradeBit 默认为 14
  15.     // 标记 0x4000 的报文（即 POD 发出的报文)，在离开 Node 的时候需要进行 SNAT 转换
  16.     masqueradeValue := 1 << uint(masqueradeBit)
  17.     masqueradeMark := fmt.Sprintf("%#08x/%#08x", masqueradeValue, masqueradeValue)
  19.     ......
  21.     endpointSlicesEnabled := utilfeature.DefaultFeatureGate.Enabled(features.EndpointSlice)
  23.     healthChecker := healthcheck.NewServer(hostname, recorder, nil, nil)
  25.     // 3.初始化 proxier
  26.     isIPv6 := ipt.IsIpv6()
  27.     proxier := &Proxier{
  28.         ......
  29.     }
  30.     burstSyncs := 2
  32.     // 4.初始化 syncRunner，BoundedFrequencyRunner 是一个定时执行器，会定时执行
  33.     // proxier.syncProxyRules 方法,syncProxyRules 是每个 proxier 实际刷新iptables 规则的方法
  34.     proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs)
  35.     return proxier, nil
  36. }

ipvs proxier 初始化

ipvs NewProxier() 方法主要逻辑为：

  • 设定内核参数，route_localnet、br_netfilter、bridge-nf-call-iptables、conntrack、conn_reuse_mode、ip_forward、arp_ignore、arp_announce 等
  • 和 iptables 一样，对于 SNAT iptables 规则生成 masquerade 标记
  • 设置默认调度算法 rr
  • 初始化 proxier 对象
  • 初始化 ipset 规则
  • 初始化 syncRunner 将 proxier.syncProxyRules 方法注入
  • 启动 gracefuldeleteManager 定时清理 RS (realServer) 记录

k8s.io/kubernetes/pkg/proxy/ipvs/proxier.go:316

  1. func NewProxier(ipt utiliptables.Interface,
  2.     ......
  3. ) (*Proxier, error) {
  5.     // 1.设定内核参数
  6.     if val, _ := sysctl.GetSysctl(sysctlRouteLocalnet); val != 1 {
  7.         ......
  8.     }
  9.     ......
  11.     // 2.生成 masquerade 标记
  12.     masqueradeValue := 1 << uint(masqueradeBit)
  13.     masqueradeMark := fmt.Sprintf("%#08x/%#08x", masqueradeValue, masqueradeValue)
  15.     // 3.设置默认调度算法 rr
  16.     if len(scheduler) == 0 {
  17.         scheduler = DefaultScheduler
  18.     }
  20.     healthChecker := healthcheck.NewServer(hostname, recorder, nil, nil) // use default implementations of deps
  22.     endpointSlicesEnabled := utilfeature.DefaultFeatureGate.Enabled(features.EndpointSlice)
  24.     // 4.初始化 proxier
  25.     proxier := &Proxier{
  26.         ......
  27.     }
  28.     // 5.初始化 ipset 规则
  29.     proxier.ipsetList = make(map[string]*IPSet)
  30.     for _, is := range ipsetInfo {
  31.         proxier.ipsetList[is.name] = NewIPSet(ipset, is.name, is.setType, isIPv6, is.comment)
  32.     }
  33.     burstSyncs := 2
  35.     // 6.初始化 syncRunner
  36.     proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs)
  38.     // 7.启动 gracefuldeleteManager
  39.     proxier.gracefuldeleteManager.Run()
  40.     return proxier, nil
  41. }

userspace proxier 初始化

userspace NewProxier() 方法主要逻辑为：

  • 初始化 iptables 规则
  • 初始化 proxier
  • 初始化 syncRunner 将 proxier.syncProxyRules 方法注入

k8s.io/kubernetes/pkg/proxy/userspace/proxier.go:187

  1. func NewProxier(......) (*Proxier, error) {
  2.     return NewCustomProxier(loadBalancer, listenIP, iptables, exec, pr, syncPeriod, minSyncPeriod, udpIdleTimeout, nodePortAddresses, newProxySocket)
  3. }
  5. func NewCustomProxier(......) (*Proxier, error) {
  6.     ......
  8.     // 1.设置打开文件数
  9.     err = setRLimit(64 * 1000)
  10.     if err != nil {
  11.         return nil, fmt.Errorf("failed to set open file handler limit: %v", err)
  12.     }
  14.     proxyPorts := newPortAllocator(pr)
  16.     return createProxier(loadBalancer, listenIP, iptables, exec, hostIP, proxyPorts, syncPeriod, minSyncPeriod, udpIdleTimeout, makeProxySocket)
  17. }
  19. func createProxier(loadBalancer LoadBalancer, listenIP net.IP, iptables iptables.Interface, exec utilexec.Interface, hostIP net.IP, proxyPorts PortAllocator, syncPeriod, minSyncPeriod, udpIdleTimeout time.Duration, makeProxySocket ProxySocketFunc) (*Proxier, error) {
  20.     if proxyPorts == nil {
  21.         proxyPorts = newPortAllocator(utilnet.PortRange{})
  22.     }
  24.     // 2.初始化 iptables 规则
  25.     if err := iptablesInit(iptables); err != nil {
  26.         return nil, fmt.Errorf("failed to initialize iptables: %v", err)
  27.     }
  29.     if err := iptablesFlush(iptables); err != nil {
  30.         return nil, fmt.Errorf("failed to flush iptables: %v", err)
  31.     }
  33.     // 3.初始化 proxier
  34.     proxier := &Proxier{
  35.         ......
  36.     }
  38.     // 4.初始化 syncRunner
  39.     proxier.syncRunner = async.NewBoundedFrequencyRunner("userspace-proxy-sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, numBurstSyncs)
  40.     return proxier, nil
  41. }

proxier 接口实现

handler 的实现

上文已经提到过每种 proxier 都需要实现 interface 中的几个方法，首先看一下 ServiceHandler、EndpointsHandler 和 EndpointSliceHandler 相关的，对于 service、endpoints 和 endpointSlices 三种对象都实现了 OnAdd、OnUpdate、OnDelete 和 OnSynced 方法。

  1. // 1.service 相关的方法
  2. func (proxier *Proxier) OnServiceAdd(service *v1.Service) {
  3.     proxier.OnServiceUpdate(nil, service)
  4. }
  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)
  14. }
  16. func (proxier *Proxier) OnServiceSynced(){
  17.     ......
  18.     proxier.syncProxyRules()
  19. }
  21. // 2.endpoints 相关的方法
  22. func (proxier *Proxier) OnEndpointsAdd(endpoints *v1.Endpoints) {
  23.     proxier.OnEndpointsUpdate(nil, endpoints)
  24. }
  26. func (proxier *Proxier) OnEndpointsUpdate(oldEndpoints, endpoints *v1.Endpoints) {
  27.     if proxier.endpointsChanges.Update(oldEndpoints, endpoints) && proxier.isInitialized() {
  28.         proxier.Sync()
  29.     }
  30. }
  32. func (proxier *Proxier) OnEndpointsDelete(endpoints *v1.Endpoints) {
  33.     proxier.OnEndpointsUpdate(endpoints, nil)
  34. }
  36. func (proxier *Proxier) OnEndpointsSynced() {
  37.     ......
  38.     proxier.syncProxyRules()
  39. }
  41. // 3.endpointSlice 相关的方法
  42. func (proxier *Proxier) OnEndpointSliceAdd(endpointSlice *discovery.EndpointSlice) {
  43.     if proxier.endpointsChanges.EndpointSliceUpdate(endpointSlice, false) && proxier.isInitialized() {
  44.         proxier.Sync()
  45.     }
  46. }
  48. func (proxier *Proxier) OnEndpointSliceUpdate(_, endpointSlice *discovery.EndpointSlice) {
  49.     if proxier.endpointsChanges.EndpointSliceUpdate(endpointSlice, false) && proxier.isInitialized() {
  50.         proxier.Sync()
  51.     }
  52. }
  54. func (proxier *Proxier) OnEndpointSliceDelete(endpointSlice *discovery.EndpointSlice) {
  55.     if proxier.endpointsChanges.EndpointSliceUpdate(endpointSlice, true) && proxier.isInitialized() {
  56.         proxier.Sync()
  57.     }
  58. }
  60. func (proxier *Proxier) OnEndpointSlicesSynced() {
  61.     ......
  62.     proxier.syncProxyRules()
  63. }

在启动逻辑的 Run() 方法中 proxier 已经被注册到了 serviceConfig、endpointsConfig、endpointSliceConfig 中，当启动 informer，cache 同步完成后会调用 OnSynced() 方法，之后当 watch 到变化后会调用 proxier 中对应的 OnUpdate() 方法进行处理，OnSynced() 会直接调用 proxier.syncProxyRules() 来刷新iptables 规则，而 OnUpdate() 会调用 proxier.syncRunner.Run() 方法，其最终也是调用 proxier.syncProxyRules() 方法刷新规则的，这种转换是在 BoundedFrequencyRunner 中体现出来的，下面看一下具体实现。

Sync() 以及 SyncLoop() 的实现

每种 proxier 的 Sync() 以及 SyncLoop() 方法如下所示，都是调用 syncRunner 中的相关方法，而 syncRunner 在前面的 NewProxier() 中已经说过了，syncRunner 是调用 async.NewBoundedFrequencyRunner() 方法初始化，至此，基本上可以确定了所有的核心都是在 BoundedFrequencyRunner 中实现的。

  1. func NewProxier() (*Proxier, error) {
  2.     ......
  3.     proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs)
  4.     ......
  5. }
  7. // Sync()
  8. func (proxier *Proxier) Sync() {
  9.     proxier.syncRunner.Run()
  10. }
  12. // SyncLoop()
  13. func (proxier *Proxier) SyncLoop() {
  14.     if proxier.healthzServer != nil {
  15.         proxier.healthzServer.UpdateTimestamp()
  16.     }
  17.     proxier.syncRunner.Loop(wait.NeverStop)
  18. }

NewBoundedFrequencyRunner()是其初始化的函数，其中的参数 minInterval和 maxInterval 分别对应 proxier 中的 minSyncPeriod 和 syncPeriod，两者的默认值分别为 0s 和 30s，其值可以使用 --iptables-min-sync-period 和 --iptables-sync-period 启动参数来指定。

k8s.io/kubernetes/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)}
  3.     // 执行定时器
  4.     <-timer.C()
  5.     // 调用 construct() 函数
  6.     return construct(name, fn, minInterval, maxInterval, burstRuns, timer)
  7. }
  9. func construct(name string, fn func(), minInterval, maxInterval time.Duration, burstRuns int, timer timer) *BoundedFrequencyRunner {
  10.     if maxInterval < minInterval {
  11.         panic(fmt.Sprintf("%s: maxInterval (%v) must be >= minInterval (%v)", name, maxInterval, minInterval))
  12.     }
  13.     if timer == nil {
  14.         panic(fmt.Sprintf("%s: timer must be non-nil", name))
  15.     }
  17.     bfr := &BoundedFrequencyRunner{
  18.         name:        name,
  19.         fn:          fn,               // 被调用的函数，proxier.syncProxyRules
  20.         minInterval: minInterval,
  21.         maxInterval: maxInterval,
  22.         run:         make(chan struct{}, 1),
  23.         timer:       timer,
  24.     }
  25.     // 由于默认的 minInterval = 0，此处使用 nullLimiter
  26.     if minInterval == 0 {
  27.         bfr.limiter = nullLimiter{}
  28.     } else {
  29.         // 采用“令牌桶”算法实现流控机制
  30.         qps := float32(time.Second) / float32(minInterval)
  31.         bfr.limiter = flowcontrol.NewTokenBucketRateLimiterWithClock(qps, burstRuns, timer)
  32.     }
  33.     return bfr
  34. }

在启动流程 Run() 方法最后调用的 s.Proxier.SyncLoop() 最终调用的是 BoundedFrequencyRunner 的 Loop()方法，如下所示：

k8s.io/kubernetes/pkg/util/async/bounded_frequency_runner.go:169

  1. func (bfr *BoundedFrequencyRunner) Loop(stop <-chan struct{}) {
  2.     bfr.timer.Reset(bfr.maxInterval)
  3.     for {
  4.         select {
  5.         case <-stop:
  6.             bfr.stop()
  7.             return
  8.         case <-bfr.timer.C():   // 定时器
  9.             bfr.tryRun()
  10.         case <-bfr.run:       // 接收 channel
  11.             bfr.tryRun()
  12.         }
  13.     }
  14. }

proxier 的 OnUpdate() 中调用的 syncRunner.Run() 其实只是在 bfr.run 这个带 buffer 的 channel 中发送了一条数据，在 BoundedFrequencyRunner 的 Loop()方法中接收到该数据后会调用 bfr.tryRun() 进行处理：

k8s.io/kubernetes/pkg/util/async/bounded_frequency_runner.go:191

  1. func (bfr *BoundedFrequencyRunner) Run() {
  2.     select {
  3.     case bfr.run <- struct{}{}:   // 向 channel 发送信号
  4.     default:
  5.     }
  6. }

而 tryRun() 方法才是最终调用 syncProxyRules() 刷新iptables 规则的。

k8s.io/kubernetes/pkg/util/async/bounded_frequency_runner.go:211

  1. func (bfr *BoundedFrequencyRunner) tryRun() {
  2.     bfr.mu.Lock()
  3.     defer bfr.mu.Unlock()
  5.     if bfr.limiter.TryAccept() {
  6.         // 执行 fn() 即 syncProxyRules() 刷新iptables 规则
  7.         bfr.fn()
  8.         bfr.lastRun = bfr.timer.Now()
  9.         bfr.timer.Stop()
  10.         bfr.timer.Reset(bfr.maxInterval)
  11.         return
  12.     }
  14.     elapsed := bfr.timer.Since(bfr.lastRun)    // how long since last run
  15.     nextPossible := bfr.minInterval - elapsed  // time to next possible run
  16.     nextScheduled := bfr.maxInterval - elapsed // time to next periodic run
  18.     if nextPossible < nextScheduled {
  19.         bfr.timer.Stop()
  20.         bfr.timer.Reset(nextPossible)
  21.     }
  22. }

通过以上分析可知，syncProxyRules() 是每个 proxier 的核心方法，启动 informer cache 同步完成后会直接调用 proxier.syncProxyRules() 刷新iptables 规则，之后如果 informer watch 到相关对象的变化后会调用 BoundedFrequencyRunner 的 tryRun()来刷新iptables 规则，定时器每 30s 会执行一次iptables 规则的刷新。

总结

本文主要介绍了 kube-proxy 的启动逻辑以及三种模式 proxier 的初始化，还有最终调用刷新iptables 规则的 BoundedFrequencyRunner，可以看到其中的代码写的很巧妙。而每种模式下的iptables 规则是如何创建、刷新以及转发的是如何实现的会在后面的文章中进行分析。