Proxy 服务框架

程序入口与初始化

kube-proxy使用通用Cobra框架构建一个标准的CLI应用.同kubernets基础组件(如:scheduler)一样的方式来创建应用和运行应用，因此应用command构建层不将对其深入分析，我们将重心聚焦在后面的proxyserver创建与proxyserver run代码分析。

!FILENAME cmd/kube-proxy/proxy.go:35

  1. func main() {
  2.     rand.Seed(time.Now().UnixNano())
  4.     command := app.NewProxyCommand()  //Cobra命令风格应用对象创建
  5.   //......
  7.   //对前面实例化Command对象的执行
  8.     if err := command.Execute(); err != nil {
  9.         fmt.Fprintf(os.Stderr, "error: %v\n", err)
  10.         os.Exit(1)
  11.     }
  12. }

NewProxyCommand()返回命令command对象,command.Execute()则调用定义的Run: func{},执行内部的opts.Run().

!FILENAME cmd/kube-proxy/app/server.go:370

  1. func NewProxyCommand() *cobra.Command {
  2.   // 创建options对象(全局的应用配置对象)
  3.     opts := NewOptions()
  5.     cmd := &cobra.Command{
  6.         Use: "kube-proxy",
  7.         Long: `The Kubernetes network proxy runs on each node...略`,
  8.         Run: func(cmd *cobra.Command, args []string) {   //Command应用RUN命令func定义
  9.             verflag.PrintAndExitIfRequested()
  10.             utilflag.PrintFlags(cmd.Flags())
  12.             if err := initForOS(opts.WindowsService); err != nil {
  13.                 klog.Fatalf("failed OS init: %v", err)
  14.             }
  15.       // 完成所有需求的options配置(外置配置文件、自定义主机名处理、特征功能开关设置等)
  16.             if err := opts.Complete(); err != nil {
  17.                 klog.Fatalf("failed complete: %v", err)
  18.             }
  19.       // 校验kube-proxy配置的有效性
  20.             if err := opts.Validate(args); err != nil {
  21.                 klog.Fatalf("failed validate: %v", err)
  22.             }
  23.       klog.Fatal(opts.Run())  //应用真正执行Run(),后面进行分析
  24.         },
  25.     }
  27.     var err error
  28.     opts.config, err = opts.ApplyDefaults(opts.config)   //应用默认配置，完成配置的初始化工作
  29.     if err != nil {
  30.         klog.Fatalf("unable to create flag defaults: %v", err)
  31.     }
  33.     opts.AddFlags(cmd.Flags())
  35.     cmd.MarkFlagFilename("config", "yaml", "yml", "json")
  37.   return cmd  //返回命令command对象
  38. }

NewOptions() 全局配置对象创建

!FILENAME cmd/kube-proxy/app/server.go:184

  1. func NewOptions() *Options {
  2.     return &Options{
  3.         config:      new(kubeproxyconfig.KubeProxyConfiguration), //实例化配置config对象
  4.         healthzPort: ports.ProxyHealthzPort,                      //Healthz端口
  5.         metricsPort: ports.ProxyStatusPort,                       //metrics端口
  6.         scheme:      scheme.Scheme,
  7.         codecs:      scheme.Codecs,
  8.         CleanupIPVS: true,
  9.     }
  10. }
  12. //完整的配置结构体与命令执行时用户传递的命令选项相对应
  13. type KubeProxyConfiguration struct {
  14.     metav1.TypeMeta
  15.     FeatureGates map[string]bool              //功能特征模块开关
  16.     BindAddress string                        //默认所有接口0.0.0.0 
  17.     HealthzBindAddress string                 //默认0.0.0.0:10256
  18.     MetricsBindAddress string                 //默认127.0.0.1:10249
  19.     EnableProfiling bool                      //"/debug/pprof"
  20.     ClusterCIDR string                        //ClusterCIDR
  21.     HostnameOverride string                   //自定义Hostname
  22.     ClientConnection apimachineryconfig.ClientConnectionConfiguration  //apiserver client
  23.   IPTables KubeProxyIPTablesConfiguration   //IPTABLES配置项(地址伪装、同步周期等)
  24.   IPVS KubeProxyIPVSConfiguration           //IPVS配置项(同步周期、调度器等)
  25.     OOMScoreAdj *int32                        //修改OOMScoreAdj分值
  26.     Mode ProxyMode                            //proxy模式
  27.     PortRange string                          //端口range
  28.     ResourceContainer string                  //"Default: /kube-proxy"
  29.     UDPIdleTimeout metav1.Duration            //UDP空闲超时
  30.     Conntrack KubeProxyConntrackConfiguration //Conntrack对象
  31.     ConfigSyncPeriod metav1.Duration          //同步周期
  32.     NodePortAddresses []string                //Node地址
  33. }
  35. const (
  36.     ProxyHealthzPort = 10256
  37.   ProxyStatusPort = 10249
  38. )

opts.Run() 创建proxy服务并启动

!FILENAME cmd/kube-proxy/app/server.go:250

  1. func (o *Options) Run() error {
  2.     if len(o.WriteConfigTo) > 0 {
  3.         return o.writeConfigFile()           //写配置文件
  4.     }
  6.     proxyServer, err := NewProxyServer(o)  //基于配置创建proxy服务
  7.     if err != nil {
  8.         return err
  9.     }
  11.     return proxyServer.Run()              //运行proxy服务，后续详细分析proxyserver执行代码逻辑
  12. }

上面已完成对kube-proxy第一层(CLI创建、配置项初始化、启动)分析,下面进入proxy Server创建与启动运行代码分析( kube-proxy第二层 )。

Proxy Server创建

NewProxyServer() 非windows版本代码

!FILENAME cmd/kube-proxy/app/server_others.go:55

  1. func NewProxyServer(o *Options) (*ProxyServer, error) {
  2.     return newProxyServer(o.config, o.CleanupAndExit, o.CleanupIPVS, o.scheme, o.master)
  3. }
  5. func newProxyServer(
  6.     config *proxyconfigapi.KubeProxyConfiguration,
  7.     cleanupAndExit bool,
  8.     cleanupIPVS bool,
  9.     scheme *runtime.Scheme,
  10.     master string) (*ProxyServer, error) {
  12.     if config == nil {
  13.         return nil, errors.New("config is required")
  14.     }
  16.     if c, err := configz.New(proxyconfigapi.GroupName); err == nil {
  17.         c.Set(config)
  18.     } else {
  19.         return nil, fmt.Errorf("unable to register configz: %s", err)
  20.     }
  22.   //协议IPV4 or IPV6
  23.     protocol := utiliptables.ProtocolIpv4
  24.     if net.ParseIP(config.BindAddress).To4() == nil {
  25.         klog.V(0).Infof("IPv6 bind address (%s), assume IPv6 operation", config.BindAddress)
  26.         protocol = utiliptables.ProtocolIpv6
  27.     }
  29.   // 关键依赖工具实现的api接口iptables/ipvs/ipset/dbus
  30.   // 从此处则可以看出kube-proxy底层技术的依赖（当然不同的proxy-mode，实现技术也不一样 ）
  31.     var iptInterface utiliptables.Interface
  32.     var ipvsInterface utilipvs.Interface
  33.     var kernelHandler ipvs.KernelHandler
  34.     var ipsetInterface utilipset.Interface
  35.     var dbus utildbus.Interface
  38.   // exec命令执行器对象创建
  39.     execer := exec.New()
  40.   // dbus对象创建（linux实现进程间通信机制）
  41.     dbus = utildbus.New()
  42.   // iptables操作对象创建
  43.     iptInterface = utiliptables.New(execer, dbus, protocol)
  44.   // IPVS
  45.     kernelHandler = ipvs.NewLinuxKernelHandler()
  46.   // ipset
  47.     ipsetInterface = utilipset.New(execer)
  48.   // IPVS环境检测
  49.     canUseIPVS, _ := ipvs.CanUseIPVSProxier(kernelHandler, ipsetInterface)
  50.     if canUseIPVS {
  51.         ipvsInterface = utilipvs.New(execer)
  52.     }
  54.     // We omit creation of pretty much everything if we run in cleanup mode
  55.     if cleanupAndExit {
  56.         return &ProxyServer{
  57.             execer:         execer,
  58.             IptInterface:   iptInterface,
  59.             IpvsInterface:  ipvsInterface,
  60.             IpsetInterface: ipsetInterface,
  61.             CleanupAndExit: cleanupAndExit,
  62.         }, nil
  63.     }
  65.   //api client
  66.     client, eventClient, err := createClients(config.ClientConnection, master)
  67.     if err != nil {
  68.         return nil, err
  69.     }
  71.     //主机名
  72.     hostname, err := utilnode.GetHostname(config.HostnameOverride)
  73.     if err != nil {
  74.         return nil, err
  75.     }
  76.   // 事件广播器
  77.     eventBroadcaster := record.NewBroadcaster()
  78.   // Create event recorder
  79.     recorder := eventBroadcaster.NewRecorder(scheme, v1.EventSource{Component: "kube-proxy", Host: hostname})
  81.     nodeRef := &v1.ObjectReference{
  82.         Kind:      "Node",
  83.         Name:      hostname,
  84.         UID:       types.UID(hostname),
  85.         Namespace: "",
  86.     }
  88.     var healthzServer *healthcheck.HealthzServer
  89.     var healthzUpdater healthcheck.HealthzUpdater
  91.   //创建默认的healthzServer服务对象
  92.     if len(config.HealthzBindAddress) > 0 {
  93.         healthzServer = healthcheck.NewDefaultHealthzServer(config.HealthzBindAddress, 2*config.IPTables.SyncPeriod.Duration, recorder, nodeRef)
  94.         healthzUpdater = healthzServer
  95.     }
  97.     var proxier proxy.ProxyProvider
  98.     var serviceEventHandler proxyconfig.ServiceHandler
  99.     var endpointsEventHandler proxyconfig.EndpointsHandler
  101.   // proxyMode模式配置获取
  102.     proxyMode := getProxyMode(string(config.Mode), iptInterface, kernelHandler, ipsetInterface, iptables.LinuxKernelCompatTester{})
  103.   // 节点绑定IP
  104.     nodeIP := net.ParseIP(config.BindAddress)
  105.     if nodeIP.IsUnspecified() {
  106.         nodeIP = utilnode.GetNodeIP(client, hostname)
  107.     }
  109.     if proxyMode == proxyModeIPTables {                   // proxyMode为"IPTables"
  110.         klog.V(0).Info("Using iptables Proxier.")
  111.         if config.IPTables.MasqueradeBit == nil {
  112.             // MasqueradeBit must be specified or defaulted.
  113.             return nil, fmt.Errorf("unable to read IPTables MasqueradeBit from config")
  114.         }
  116.         //创建iptables proxier对象
  117.         proxierIPTables, err := iptables.NewProxier(
  118.             iptInterface,
  119.             utilsysctl.New(),
  120.             execer,
  121.             config.IPTables.SyncPeriod.Duration,
  122.             config.IPTables.MinSyncPeriod.Duration,
  123.             config.IPTables.MasqueradeAll,
  124.             int(*config.IPTables.MasqueradeBit),
  125.             config.ClusterCIDR,
  126.             hostname,
  127.             nodeIP,
  128.             recorder,
  129.             healthzUpdater,
  130.             config.NodePortAddresses,
  131.         )
  132.         if err != nil {
  133.             return nil, fmt.Errorf("unable to create proxier: %v", err)
  134.         }
  135.         metrics.RegisterMetrics()
  136.     //iptables proxier对象和事件处理
  137.         proxier = proxierIPTables               
  138.         serviceEventHandler = proxierIPTables
  139.         endpointsEventHandler = proxierIPTables
  140.         // No turning back. Remove artifacts that might still exist from the userspace Proxier.
  141.         klog.V(0).Info("Tearing down inactive rules.")
  142.         //模式转换清理userspace/ipvs模式所创建iptables规则
  143.         userspace.CleanupLeftovers(iptInterface)
  144.         if canUseIPVS {
  145.             ipvs.CleanupLeftovers(ipvsInterface, iptInterface, ipsetInterface, cleanupIPVS)
  146.         }
  147.     } else if proxyMode == proxyModeIPVS {                  // proxyMode为"IPVS"
  148.         klog.V(0).Info("Using ipvs Proxier.")
  149.     //ipvs proxier对象创建
  150.         proxierIPVS, err := ipvs.NewProxier(
  151.             iptInterface,
  152.             ipvsInterface,
  153.             ipsetInterface,
  154.             utilsysctl.New(),
  155.             execer,
  156.             config.IPVS.SyncPeriod.Duration,
  157.             config.IPVS.MinSyncPeriod.Duration,
  158.             config.IPVS.ExcludeCIDRs,
  159.             config.IPTables.MasqueradeAll,
  160.             int(*config.IPTables.MasqueradeBit),
  161.             config.ClusterCIDR,
  162.             hostname,
  163.             nodeIP,
  164.             recorder,
  165.             healthzServer,
  166.             config.IPVS.Scheduler,
  167.             config.NodePortAddresses,
  168.         )
  169.         if err != nil {
  170.             return nil, fmt.Errorf("unable to create proxier: %v", err)
  171.         }
  172.         metrics.RegisterMetrics()
  173.     //ipvs proxier对象和事件处理
  174.         proxier = proxierIPVS
  175.         serviceEventHandler = proxierIPVS
  176.         endpointsEventHandler = proxierIPVS
  177.         klog.V(0).Info("Tearing down inactive rules.")
  178.     //模式转换清理userspace/iptables模式规则
  179.         userspace.CleanupLeftovers(iptInterface)
  180.         iptables.CleanupLeftovers(iptInterface)
  181.     } else {                                              // proxyMode为"userspace"
  182.         klog.V(0).Info("Using userspace Proxier.")
  183.     //创建RR模式负载均衡
  184.         loadBalancer := userspace.NewLoadBalancerRR()
  185.     //设置EndpointsConfigHandler(endpoints事件处理)
  186.         endpointsEventHandler = loadBalancer
  187.     //创建userspace proxier对象
  188.         proxierUserspace, err := userspace.NewProxier(
  189.             loadBalancer,
  190.             net.ParseIP(config.BindAddress),
  191.             iptInterface,
  192.             execer,
  193.             *utilnet.ParsePortRangeOrDie(config.PortRange),
  194.             config.IPTables.SyncPeriod.Duration,
  195.             config.IPTables.MinSyncPeriod.Duration,
  196.             config.UDPIdleTimeout.Duration,
  197.             config.NodePortAddresses,
  198.         )
  199.         if err != nil {
  200.             return nil, fmt.Errorf("unable to create proxier: %v", err)
  201.         }
  202.     //userspace proxier对象和service事件处理
  203.         serviceEventHandler = proxierUserspace
  204.         proxier = proxierUserspace
  206.         klog.V(0).Info("Tearing down inactive rules.")
  207.         //模式转换清理iptables/ipvs模式所创建iptables规则
  208.         iptables.CleanupLeftovers(iptInterface)
  209.         if canUseIPVS {
  210.             ipvs.CleanupLeftovers(ipvsInterface, iptInterface, ipsetInterface, cleanupIPVS)
  211.         }
  212.     }
  214.   //注册reloadfunc为proxier的sync()同步方法
  215.     iptInterface.AddReloadFunc(proxier.Sync)
  217.   // 构建ProxyServer对象
  218.     return &ProxyServer{
  219.         Client:                 client,                  //apiServer client
  220.         EventClient:            eventClient,             //事件client
  221.         IptInterface:           iptInterface,            //iptables接口
  222.         IpvsInterface:          ipvsInterface,           //ipvs接口
  223.         IpsetInterface:         ipsetInterface,          //ipset接口
  224.         execer:                 execer,                  //exec命令执行器   
  225.         Proxier:                proxier,                 //proxier创建对象
  226.         Broadcaster:            eventBroadcaster,        //事件广播器
  227.         Recorder:               recorder,                //事件记录器
  228.         ConntrackConfiguration: config.Conntrack,        //Conntrack配置
  229.         Conntracker:            &realConntracker{},      //Conntrack对象
  230.         ProxyMode:              proxyMode,               //proxy模式
  231.         NodeRef:                nodeRef,                 //node节点reference信息
  232.         MetricsBindAddress:     config.MetricsBindAddress,        //metric服务地址配置
  233.         EnableProfiling:        config.EnableProfiling,           //debug/pprof配置
  234.         OOMScoreAdj:            config.OOMScoreAdj,               //OOMScoreAdj值配置
  235.         ResourceContainer:      config.ResourceContainer,         //容器资源配置
  236.         ConfigSyncPeriod:       config.ConfigSyncPeriod.Duration, //同步周期配置
  237.         ServiceEventHandler:    serviceEventHandler,              //处理service事件proxier对象
  238.         EndpointsEventHandler:  endpointsEventHandler,            //处理endpoints事件proxier对象
  239.         HealthzServer:          healthzServer,                    //健康检测服务
  240.     }, nil
  241. }

Proxy Server运行

!FILENAME cmd/kube-proxy/app/server.go:481

  1. func (s *ProxyServer) Run() error {
  2.     klog.Infof("Version: %+v", version.Get())
  4.   // 如果CleanupAndExit设置为true,则删除存在的所有iptables规则项，然后应用退出。
  5.     if s.CleanupAndExit {
  6.         encounteredError := userspace.CleanupLeftovers(s.IptInterface)
  7.         encounteredError = iptables.CleanupLeftovers(s.IptInterface) || encounteredError
  8.         encounteredError = ipvs.CleanupLeftovers(s.IpvsInterface, s.IptInterface, s.IpsetInterface, s.CleanupIPVS) || encounteredError
  9.         if encounteredError {
  10.             return errors.New("encountered an error while tearing down rules.")
  11.         }
  12.         return nil
  13.     }
  15.   // 根据启动参数配置"oom-score-adj"分值调整，取值区间[-1000, 1000]
  16.     var oomAdjuster *oom.OOMAdjuster
  17.     if s.OOMScoreAdj != nil {
  18.         oomAdjuster = oom.NewOOMAdjuster()
  19.         if err := oomAdjuster.ApplyOOMScoreAdj(0, int(*s.OOMScoreAdj)); err != nil {
  20.             klog.V(2).Info(err)
  21.         }
  22.     }
  24.   //"resource-container"设置是否运行在容器里
  25.     if len(s.ResourceContainer) != 0 {
  26.         if err := resourcecontainer.RunInResourceContainer(s.ResourceContainer); err != nil {
  27.             klog.Warningf("Failed to start in resource-only container %q: %v", s.ResourceContainer, err)
  28.         } else {
  29.             klog.V(2).Infof("Running in resource-only container %q", s.ResourceContainer)
  30.         }
  31.     }
  33.   //事件广播器
  34.     if s.Broadcaster != nil && s.EventClient != nil {
  35.         s.Broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: s.EventClient.Events("")})
  36.     }
  38.     // 根据配置启动healthz健康检测服务
  39.     if s.HealthzServer != nil {
  40.         s.HealthzServer.Run()
  41.     }
  43.   // 根据配置启动metrics服务。URI: "/proxyMode" 与 "/metrics"
  44.     if len(s.MetricsBindAddress) > 0 {
  45.         mux := mux.NewPathRecorderMux("kube-proxy")
  46.         healthz.InstallHandler(mux)
  47.         mux.HandleFunc("/proxyMode", func(w http.ResponseWriter, r *http.Request) {
  48.             fmt.Fprintf(w, "%s", s.ProxyMode)
  49.         })
  50.         mux.Handle("/metrics", prometheus.Handler())
  51.         if s.EnableProfiling {
  52.             routes.Profiling{}.Install(mux)
  53.         }
  54.         configz.InstallHandler(mux)
  55.         go wait.Until(func() {
  56.             err := http.ListenAndServe(s.MetricsBindAddress, mux)
  57.             if err != nil {
  58.                 utilruntime.HandleError(fmt.Errorf("starting metrics server failed: %v", err))
  59.             }
  60.         }, 5*time.Second, wait.NeverStop)
  61.     }
  63.   // 如果需要(命令选项或配置项)调节conntrack配置值
  64.     if s.Conntracker != nil {
  65.         max, err := getConntrackMax(s.ConntrackConfiguration)
  66.         if err != nil {
  67.             return err
  68.         }
  69.         if max > 0 {
  70.             err := s.Conntracker.SetMax(max)
  71.             if err != nil {
  72.                 if err != readOnlySysFSError {
  73.                     return err
  74.                 }
  76.                 const message = "DOCKER RESTART NEEDED (docker issue #24000): /sys is read-only: " +
  77.                     "cannot modify conntrack limits, problems may arise later."
  78.                 s.Recorder.Eventf(s.NodeRef, api.EventTypeWarning, err.Error(), message)
  79.             }
  80.         }
  81.      //设置conntracker的TCPEstablishedTimeout
  82.         if s.ConntrackConfiguration.TCPEstablishedTimeout != nil && s.ConntrackConfiguration.TCPEstablishedTimeout.Duration > 0 {
  83.             timeout := int(s.ConntrackConfiguration.TCPEstablishedTimeout.Duration / time.Second)
  84.             if err := s.Conntracker.SetTCPEstablishedTimeout(timeout); err != nil {
  85.                 return err
  86.             }
  87.         }
  88.     //设置conntracker的TCPCloseWaitTimeout
  89.         if s.ConntrackConfiguration.TCPCloseWaitTimeout != nil && s.ConntrackConfiguration.TCPCloseWaitTimeout.Duration > 0 {
  90.             timeout := int(s.ConntrackConfiguration.TCPCloseWaitTimeout.Duration / time.Second)
  91.             if err := s.Conntracker.SetTCPCloseWaitTimeout(timeout); err != nil {
  92.                 return err
  93.             }
  94.         }
  95.     }
  97.    // informer机制获取与监听Services和Endpoints的配置与事件信息
  98.    // 注册ServiceEventHandler服务事件的处理
  99.    // 注册EndpointsEventHandler端点事件的处理
  100.     informerFactory := informers.NewSharedInformerFactory(s.Client, s.ConfigSyncPeriod)
  101.     serviceConfig := config.NewServiceConfig(informerFactory.Core().V1().Services(), s.ConfigSyncPeriod)
  102.     serviceConfig.RegisterEventHandler(s.ServiceEventHandler)
  103.     go serviceConfig.Run(wait.NeverStop)
  105.     endpointsConfig := config.NewEndpointsConfig(informerFactory.Core().V1().Endpoints(), s.ConfigSyncPeriod)
  106.     endpointsConfig.RegisterEventHandler(s.EndpointsEventHandler)
  107.     go endpointsConfig.Run(wait.NeverStop)
  109.     go informerFactory.Start(wait.NeverStop)
  111.   // "新生儿降生的哭声"，作者命名比较有生活情趣^_^
  112.   //  服务成功启动，将启动事件广播。
  113.   //   s.Recorder.Eventf(s.NodeRef, api.EventTypeNormal, "Starting", "Starting kube-proxy.")
  114.   //      nodeRef := &v1.ObjectReference{
  115.     //        Kind:      "Node",
  116.     //        Name:      hostname,
  117.     //        UID:       types.UID(hostname),
  118.     //       Namespace: "",
  119.     //   }
  120.     s.birthCry()
  122.   // Proxier(代理服务提供者)进行循环配置同步与处理proxy逻辑（本文聚焦应用主框架，后有专篇分析Proxier）
  123.     // 此时将进入了proxier运行，默认使用的iptables模式的proxier对象
  124.     s.Proxier.SyncLoop()
  125.     return nil
  126. }

s.Proxier.SyncLoop()的运行将进入kube-proxy第三层(service实现机制层),Proxier实例化对象是在proxy server对象创建时通过config配置文件或”-proxy-mode”指定（userspace / iptables / ipvs模式），而默认使用的iptables模式proxier对象。第三层的代码分析将针对三种模式设立专篇分析，此处为关键部分请记住此处，在后面proxier的分析文章重点关注。

在第二层的框架层我们还须关注kube-proxy与kubernetes集群同步信息的机制informer。kube-proxy组件在proxy server的run()运行创建service、endpoints的informer对其list同步数据和watch监听事件(add、delete、update)，调用注册的proxier handler进行处理(第三层proxier的同步处理机制调用与触发)。

同步规则机制(Informer)

kube-proxy同样使用client-go标准的ApiServer同步方式，创建informer,注册事件处理器handler，持续监控watch事件并调用handler处理事件add/update/delete，后端处理则由proxier(userspace/iptables/ipvs)实现。因为endpoints的同步与service同步方式一致，则下面仅说明service代码实现。

!FILENAME pkg/proxy/config/config.go:174

  1. func NewServiceConfig(serviceInformer coreinformers.ServiceInformer, resyncPeriod time.Duration) *ServiceConfig {
  2.     result := &ServiceConfig{
  3.         lister:       serviceInformer.Lister(),                //监听器
  4.         listerSynced: serviceInformer.Informer().HasSynced,    //监听器同步状态值
  5.     }
  7.   //在服务informer上添加了资源事件的处理器handleFunc。（Add/update/delete）
  8.     serviceInformer.Informer().AddEventHandlerWithResyncPeriod(
  9.         cache.ResourceEventHandlerFuncs{
  10.             AddFunc:    result.handleAddService,
  11.             UpdateFunc: result.handleUpdateService,
  12.             DeleteFunc: result.handleDeleteService,
  13.         },
  14.         resyncPeriod,     
  15.     )
  17.     return result
  18. }

!FILENAME pkg/proxy/config/config.go:119

  1. func (c *ServiceConfig) Run(stopCh <-chan struct{}) {
  2.     defer utilruntime.HandleCrash()
  4.     klog.Info("Starting service config controller")
  5.     defer klog.Info("Shutting down service config controller")
  7.     if !controller.WaitForCacheSync("service config", stopCh, c.listerSynced) {
  8.         return
  9.     }
  11.     for i := range c.eventHandlers {
  12.         klog.V(3).Info("Calling handler.OnServiceSynced()")
  13.         c.eventHandlers[i].OnServiceSynced()               //服务与proxier处理同步
  14.     }
  16.     <-stopCh
  17. }

HandleAddService()新增事件处理

!FILENAME pkg/proxy/config/config.go:217

  1. func (c *ServiceConfig) handleAddService(obj interface{}) {
  2.     service, ok := obj.(*v1.Service)
  3.     if !ok {
  4.         utilruntime.HandleError(fmt.Errorf("unexpected object type: %v", obj))
  5.         return
  6.     }
  7.     for i := range c.eventHandlers {
  8.         klog.V(4).Info("Calling handler.OnServiceAdd")
  9.         c.eventHandlers[i].OnServiceAdd(service)            //服务新增事件与proxier处理同步
  10.     }
  11. }

HandleUpdateService()更新事件处理

!FILENAME pkg/proxy/config/config.go:229

  1. func (c *ServiceConfig) handleUpdateService(oldObj, newObj interface{}) {
  2.     oldService, ok := oldObj.(*v1.Service)
  3.     if !ok {
  4.         utilruntime.HandleError(fmt.Errorf("unexpected object type: %v", oldObj))
  5.         return
  6.     }
  7.     service, ok := newObj.(*v1.Service)
  8.     if !ok {
  9.         utilruntime.HandleError(fmt.Errorf("unexpected object type: %v", newObj))
  10.         return
  11.     }
  12.     for i := range c.eventHandlers {
  13.         klog.V(4).Info("Calling handler.OnServiceUpdate")
  14.         c.eventHandlers[i].OnServiceUpdate(oldService, service) //服务更新事件与proxier处理同步 
  15.     }
  16. }

HandleDeleteService()删除事件处理

!FILENAME pkg/proxy/config/config.go:246

  1. func (c *ServiceConfig) handleDeleteService(obj interface{}) {
  2.     service, ok := obj.(*v1.Service)
  3.     if !ok {
  4.         tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
  5.         if !ok {
  6.             utilruntime.HandleError(fmt.Errorf("unexpected object type: %v", obj))
  7.             return
  8.         }
  9.         if service, ok = tombstone.Obj.(*v1.Service); !ok {
  10.             utilruntime.HandleError(fmt.Errorf("unexpected object type: %v", obj))
  11.             return
  12.         }
  13.     }
  14.     for i := range c.eventHandlers {
  15.         klog.V(4).Info("Calling handler.OnServiceDelete")
  16.         c.eventHandlers[i].OnServiceDelete(service)         //服务删除事件与proxier处理同步
  17.     }
  18. }

第三层proxier分析，请参看iptables、ipvs、userspace-mode proxier分析文档。

〜本文END〜