server

1.service

1.1.简介

• endpoints controller负责生成和维护endpoints对象，监听service和对应pod变化，动态调整endpoints的后端IP池，service的容器发现基于endpoints实现。kubeproxy会监听service和endpoints的更新，调用proxyer刷新主机规则，实现流量转发。
• 

• 注意

  1.kubernetes使用endpointSlice，kubeproxy会监听endpointSlice，否则会监听endpoint

  2.kubernetes启动service topology，kubeproxy会监听node，实现服务基于集群的node topology进行流量路由

1.2.工作模式

• --- clusterIP模式
  	向服务分配集群内部的虚拟IP地址供集群内其它服务访问,clusterIP仅对集群内部有效,无法直接从集群外部访问
  
  --- headless-service模式
  	clusterIP设置为None，对应的service不会分配集群内的虚拟IP，此时service域名解析直接转到后端Pod
  
  --- nodeport模式
  	集群的每个节点开一个静态端口，端口流量转发到service，提供外部访问能力
  
  --- loadbalancer模式
  	LB模式依赖外部负载组件，nodeport和clusterIP是其附属资源，本质是将service和LB的IP绑定，LB将流量转发到nodeport
  
  --- externalName
  	集群内部访问外部服务,无selector和endpoints，svc域名和外部域名解析为CNAME记录
  
  --- ingress模式
  	ingress会作为service的service，根据不同的url将请求转发到不同service，提供外部访问能力
  

• 

• 注意

  1.service.spec.publishNotReadyAddresses支持关联未ready pod，用于向statefulset应用设置srv dns记录实现对等发现

  2.externalName模式支持将svc域名和外部域名关联，集群内访问svc基于coredns转发到集群外

  3.service的实际路由转发由kube-proxy组件实现，service仅以一种VIP的形式存在

1.3.负载模式

• service仅以VIP的形式存在，路由转发基于kubeproxy后端的代理模块实现，支持userspace/iptables/ipvs/kernelspace四种方案。

• --- userspace模式
  	1.访问服务的请求到达节点进入内核iptables，回到用户空间由proxy转发到目标应用
  	2.kubeproxy监听的转发端口在用户空间，依赖iptables将访问服务的连接重定向给proxy
  	3.流量由用户空间进出内核造成性能损耗
  
  --- iptables模式
  	1.基于netfilter实现，流量基于iptables规则路由到目标应用
  	2.iptables模块基于DNAT模块实现service-->pod路由，免去内核到用户态切换，不支持无响应重试
  	3.大规模service会产生太多的iptables规则，规则链表遍历更新会引起时延，造成明显的性能问题
  
  ---ipvs模式
  	1.内核支持的一种服务虚拟化技术构建于netfilter基础上的内核传输层
  	2.基于高效的哈希结构存储网络路由规则，相比iptables显著减少规则的同步开销，提高集群的路由扩展规模
  	3.负载均衡算法支持轮询/最少连接/目标哈希/源哈希/预计延迟时间最短/从不排队
  	4.依赖iptables进行包过滤、地址伪装、SNAT功能，基于ipset存储路由或伪装的源或目标地址，确保iptables的规则数量恒定
  

• 

• 注意

  1.ipvs不会直接调用iptables生成规则链，基于其扩展ipset存储路由或伪装的流量源或目标地址

  2.ipvs模式规则链数量恒定，基于哈希结构高效查找和匹配规则

  3.ipvs支持服务的健康检查和连接重试

1.4.流量路由

• kubeproxy仅负责watch变更及更新规则链，流量转发基于filter和nat规则，iptables和ipvs均涉及调整规则。其中，filter表用于包过滤，控制某个链路是否准入。

• 

• nat表用于网络地址转换，修改包的源地址或目标地址，确保流量路由到正确位置，仅针对连接的第一个包生效，后续包直接复用跟踪结果，性能开销小。

• 

• 数据包经过iptables会逐次经过raw/mangle/nat/filter的chain进行决策及地址转换，kubeproxy主要修改filter/nat表的链规则。

• 

• 注意

  1.service仅提供同节点路由能力，跨节点依赖CNI

  2.service跨节点路由只是将目标地址伪装，基于CNI将流量转向其它节点应用，回包进行源地址伪装，恢复为clusterIP-->Pod

2.kubeproxy

2.1.简介

• kubeproxy主要监听service/endpoint/endpointSlice/node的事件，根据同步内容下放策略到机器，底层调用iptables规则修改4表5链，利用nat实现地址转换及路由。

• 

• 注意

  1.boundedFrequeneRunner是执行定时处理的调度器

  2.proxier提供syncProxyRules()执行规则刷新

2.2.proxyServer

• proxyServer是核心组件，负责管理和维护实际的转发规则及同步资源事件，内部关联userspace/iptables/ipvs三种工作模式，体现在不同的proxier实例，proxier会真正管理底层规则。

• // ProxyServer represents all the parameters required to start the Kubernetes proxy server.
  type ProxyServer struct {
      // 资源同步相关
  	Client                 clientset.Interface
  	EventClient            v1core.EventsGetter
      Broadcaster            events.EventBroadcaster
  	Recorder               events.EventRecorder
      
      // 网络相关接口
  	IptInterface           utiliptables.Interface
  	IpvsInterface          utilipvs.Interface
  	IpsetInterface         utilipset.Interface
  	execer                 exec.Interface
      
      // 流量代理相关
  	Proxier                proxy.Provider
      ProxyMode              string
      
      // node/pod网络相关
      NodeRef                *v1.ObjectReference
      localDetectorMode      kubeproxyconfig.LocalMode
      podCIDRs               []string // only used for LocalModeNodeCIDR
  
      // 连接跟踪相关
  	ConntrackConfiguration kubeproxyconfig.KubeProxyConntrackConfiguration
  	Conntracker            Conntracker // if nil, ignored
  	
  	...
      // OOM分数
  	OOMScoreAdj            *int32
      // 同步周期
  	ConfigSyncPeriod       time.Duration
      // 健康检查接口
  	HealthzServer          healthcheck.ProxierHealthUpdater
  }
  
  // NewProxyServer returns a new ProxyServer.
  func NewProxyServer(o *Options) (*ProxyServer, error) {
  	return newProxyServer(o.config, o.CleanupAndExit, o.master)
  }
  
  // create proxyServer for iptables/ipvs
  func newProxyServer(config *proxyconfigapi.KubeProxyConfiguration,cleanupAndExit bool,master string) (*ProxyServer, error) {
  
  	...
  
  	// Create a iptables utils.
  	execer := exec.New()
  
      // 1.初始化内核检查工具
  	kernelHandler = ipvs.NewLinuxKernelHandler()
      // 2.初始化ipset工具
  	ipsetInterface = utilipset.New(execer)
      // 3.ipvs检查及初始化
  	canUseIPVS, err := ipvs.CanUseIPVSProxier(kernelHandler, ipsetInterface, config.IPVS.Scheduler)
  	if canUseIPVS {
  		ipvsInterface = utilipvs.New()
  	}
  
  	// 4.清理模式
  	if cleanupAndExit {
  		return &ProxyServer{
  			execer:         execer,
  			IpvsInterface:  ipvsInterface,
  			IpsetInterface: ipsetInterface,
  		}, nil
  	}
  
  	...
      // 5.初始化kubeclient
  	client, eventClient, err := createClients(config.ClientConnection, master)
  	...
  
      // 6.获取nodeIP
  	nodeIP := detectNodeIP(client, hostname, config.BindAddress)
  
  	// 7.初始化事件发布
  	eventBroadcaster := events.NewBroadcaster(&events.EventSinkImpl{Interface: client.EventsV1()})
  	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, "kube-proxy")
  
      ...
      // 8.初始化healthzServer
  	if len(config.HealthzBindAddress) > 0 {
  		healthzServer = healthcheck.NewProxierHealthServer(config.HealthzBindAddress, 2*config.IPTables.SyncPeriod.Duration, recorder, nodeRef)
  	}
  
  	...
      // 9.解析支持模式(userspace-->iptables-->ipvs)
  	proxyMode := getProxyMode(string(config.Mode), canUseIPVS, iptables.LinuxKernelCompatTester{})
      // 10.外部流量检查基准(ClusterCIDR-->NodeCIDR-->BridgeInterface-->InterfaceNamePrefix)
  	detectLocalMode, err = getDetectLocalMode(config)
  	...
  
  	// 11.根据nodeIP的地址类型,初始化iptables工具或ip6tables工具
  	iptInterface = utiliptables.New(execer, primaryProtocol)
  	
      // 12.非userspace模式,初始化iptables和ip6tables以应对双栈
  	if proxyMode != proxyModeUserspace {
  		// Create iptables handlers for both families, one is already created
  		// Always ordered as IPv4, IPv6
  		if primaryProtocol == utiliptables.ProtocolIPv4 {
  			ipt[0] = iptInterface
  			ipt[1] = utiliptables.New(execer, utiliptables.ProtocolIPv6)
  		} else {
  			ipt[0] = utiliptables.New(execer, utiliptables.ProtocolIPv4)
  			ipt[1] = iptInterface
  		}
  
  		for _, perFamilyIpt := range ipt {
              // 命名执行失败,说明不支持双栈
  			if !perFamilyIpt.Present() {
  				dualStack = false
  			}
  		}
  	}
  
      // 13.iptables模式
  	if proxyMode == proxyModeIPTables {
  		// 检查地址伪装标志位
  		if config.IPTables.MasqueradeBit == nil {
  			// MasqueradeBit must be specified or defaulted.
  			return nil, fmt.Errorf("unable to read IPTables MasqueradeBit from config")
  		}
  
  		if dualStack {
  			...
              // 生成本地流量检测器
  			localDetectors, err = getDualStackLocalDetectorTuple(detectLocalMode, config, ipt, nodeInfo)
  			...
  
  			// 初始化双栈iptables proxier
  			proxier, err = iptables.NewDualStackProxier(
  				ipt,
  				utilsysctl.New(),
  				execer,
  				config.IPTables.SyncPeriod.Duration,
  				config.IPTables.MinSyncPeriod.Duration,
  				config.IPTables.MasqueradeAll,
  				int(*config.IPTables.MasqueradeBit),
  				localDetectors,
  				hostname,
  				nodeIPTuple(config.BindAddress),
  				recorder,
  				healthzServer,
  				config.NodePortAddresses,
  			)
  		} else {
  			...
              // 初始化单栈本地流量检测器
  			localDetector, err = getLocalDetector(detectLocalMode, config, iptInterface, nodeInfo)
  			...
  
  			// 初始化单栈iptables proxier
  			proxier, err = iptables.NewProxier(
  				iptInterface,
  				utilsysctl.New(),
  				execer,
  				config.IPTables.SyncPeriod.Duration,
  				config.IPTables.MinSyncPeriod.Duration,
  				config.IPTables.MasqueradeAll,
  				int(*config.IPTables.MasqueradeBit),
  				localDetector,
  				hostname,
  				nodeIP,
  				recorder,
  				healthzServer,
  				config.NodePortAddresses,
  			)
  		}
      // 14.ipvs模式
  	} else if proxyMode == proxyModeIPVS {
  		
  		if dualStack {
  			...
              // 初始化本地流量检测器
  			localDetectors, err = getDualStackLocalDetectorTuple(detectLocalMode, config, ipt, nodeInfo)
  			...
  
              // 初始化双栈ipvs proxier
  			proxier, err = ipvs.NewDualStackProxier(
  				ipt,
  				ipvsInterface,
  				ipsetInterface,
  				utilsysctl.New(),
  				execer,
  				config.IPVS.SyncPeriod.Duration,
  				config.IPVS.MinSyncPeriod.Duration,
  				config.IPVS.ExcludeCIDRs,
  				config.IPVS.StrictARP,
  				config.IPVS.TCPTimeout.Duration,
  				config.IPVS.TCPFinTimeout.Duration,
  				config.IPVS.UDPTimeout.Duration,
  				config.IPTables.MasqueradeAll,
  				int(*config.IPTables.MasqueradeBit),
  				localDetectors,
  				hostname,
  				nodeIPs,
  				recorder,
  				healthzServer,
  				config.IPVS.Scheduler,
  				config.NodePortAddresses,
  				kernelHandler,
  			)
  		} else {
  			...
              // 初始化本地流量检测器
  			localDetector, err = getLocalDetector(detectLocalMode, config, iptInterface, nodeInfo)
  			...
  
              // 单栈ipvs proxier
  			proxier, err = ipvs.NewProxier(
  				iptInterface,
  				ipvsInterface,
  				ipsetInterface,
  				utilsysctl.New(),
  				execer,
  				config.IPVS.SyncPeriod.Duration,
  				config.IPVS.MinSyncPeriod.Duration,
  				config.IPVS.ExcludeCIDRs,
  				config.IPVS.StrictARP,
  				config.IPVS.TCPTimeout.Duration,
  				config.IPVS.TCPFinTimeout.Duration,
  				config.IPVS.UDPTimeout.Duration,
  				config.IPTables.MasqueradeAll,
  				int(*config.IPTables.MasqueradeBit),
  				localDetector,
  				hostname,
  				nodeIP,
  				recorder,
  				healthzServer,
  				config.IPVS.Scheduler,
  				config.NodePortAddresses,
  				kernelHandler,
  			)
  		}
      // 15.userspace模式
  	} else {
  
  		// TODO this has side effects that should only happen when Run() is invoked.
  		proxier, err = userspace.NewProxier(
  			userspace.NewLoadBalancerRR(),
  			netutils.ParseIPSloppy(config.BindAddress),
  			iptInterface,
  			execer,
  			*utilnet.ParsePortRangeOrDie(config.PortRange),
  			config.IPTables.SyncPeriod.Duration,
  			config.IPTables.MinSyncPeriod.Duration,
  			config.UDPIdleTimeout.Duration,
  			config.NodePortAddresses,
  		)
  	}
  
      // userspace mode doesn't support endpointslice.
  	if proxyMode == proxyModeUserspace {
  		useEndpointSlices = false
  	}
  
  	return &ProxyServer{
  		Client:                 client,
  		EventClient:            eventClient,
  		IptInterface:           iptInterface,
  		IpvsInterface:          ipvsInterface,
  		IpsetInterface:         ipsetInterface,
  		execer:                 execer,
  		Proxier:                proxier,
  		Broadcaster:            eventBroadcaster,
  		Recorder:               recorder,
  		ConntrackConfiguration: config.Conntrack,
  		Conntracker:            &realConntracker{},
  		ProxyMode:              proxyMode,
  		NodeRef:                nodeRef,
  		MetricsBindAddress:     config.MetricsBindAddress,
  		BindAddressHardFail:    config.BindAddressHardFail,
  		EnableProfiling:        config.EnableProfiling,
          // OOM分数为-999
  		OOMScoreAdj:            config.OOMScoreAdj,
  		ConfigSyncPeriod:       config.ConfigSyncPeriod.Duration,
  		HealthzServer:          healthzServer,
  		UseEndpointSlices:      useEndpointSlices,
  		localDetectorMode:      detectLocalMode,
  		podCIDRs:               podCIDRs,
  	}, nil
  }
  

• 

• 注意

  1.NewProxyServer()主要根据工作模式初始化本地流量检测器和proxier

  2.proxier是真正管理底层规则的实例，推荐的是iptables/ipvs,userspace模式已逐渐废弃

2.3.runLoop

• kube-proxy启动会初始化proxyServer实例，调用runLoop()后台启动proxyServer监听service及同步规则。

• // Run runs the specified ProxyServer.
  func (o *Options) Run() error {
  	...
      // 初始化proxyServer
  	proxyServer, err := NewProxyServer(o)
  	...
  
  	o.proxyServer = proxyServer
      
      // 后台启动主循环
  	return o.runLoop()
  }
  
  // runLoop will watch on the update change of the proxy server's configuration file.Return an error when updated
  func (o *Options) runLoop() error {
      // proxy配置监听(内容修改/重命名)
  	if o.watcher != nil {
  		o.watcher.Run()
  	}
  
  	// 后台启动规则管理
  	go func() {
  		err := o.proxyServer.Run()
  		o.errCh <- err
  	}()
  
      // 监听规则管理运行状态
  	for {
  		err := <-o.errCh
  		if err != nil {
  			return err
  		}
  	}
  }
  
  // Run runs the specified ProxyServer.  This should never exit (unless CleanupAndExit is set).
  func (s *ProxyServer) Run() error {
  
  	// 1.设置OOM分数
  	var oomAdjuster *oom.OOMAdjuster
  	if s.OOMScoreAdj != nil {
          // 初始化oomAdjuster
  		oomAdjuster = oom.NewOOMAdjuster()
          // 修改kubeproxy对应OOM分数
  		oomAdjuster.ApplyOOMScoreAdj(0, int(*s.OOMScoreAdj))
          ...
  	}
  
      // 2.事件发布
  	if s.Broadcaster != nil && s.EventClient != nil {
  		s.Broadcaster.StartRecordingToSink(stopCh)
  	}
  	...
  
  	// 3.启动healthz server
  	serveHealthz(s.HealthzServer, errCh)
  
  	// 4.设置conntrack
  	if s.Conntracker != nil {
  		max, err := getConntrackMax(s.ConntrackConfiguration)
  		...
          // 设置内核参数nf_conntrack_max
  		if max > 0 {
  			s.Conntracker.SetMax(max)
  			...
  		}
  
          // 设置内核参数nf_conntrack_tcp_timeout_established
  		if s.ConntrackConfiguration.TCPEstablishedTimeout.Duration > 0 {
  			timeout := int(s.ConntrackConfiguration.TCPEstablishedTimeout.Duration / time.Second)
  			s.Conntracker.SetTCPEstablishedTimeout(timeout)
              ...
  		}
  
          // 设置内核参数nf_conntrack_tcp_timeout_close_wait
  		if s.ConntrackConfiguration.TCPCloseWaitTimeout.Duration > 0 {
  			timeout := int(s.ConntrackConfiguration.TCPCloseWaitTimeout.Duration / time.Second)
  			s.Conntracker.SetTCPCloseWaitTimeout(timeout)
              ...
  		}
  	}
  
      // 跳过不由kubeproxy管理的service
  	noProxyName, err := labels.NewRequirement(apis.LabelServiceProxyName, selection.DoesNotExist, nil)
  	...
      // 跳过headless service(直接基于后端池访问)
  	noHeadlessEndpoints, err := labels.NewRequirement(v1.IsHeadlessService, selection.DoesNotExist, nil)
  	...
  	labelSelector = labelSelector.Add(*noProxyName, *noHeadlessEndpoints)
  
      // 初始化informerFactory(过滤非kubeproxy管理资源)
  	informerFactory := informers.NewSharedInformerFactoryWithOptions(s.Client, s.ConfigSyncPeriod,
  		informers.WithTweakListOptions(func(options *metav1.ListOptions) {
  			options.LabelSelector = labelSelector.String()
  		}))
  
  	// service监听
  	serviceConfig := config.NewServiceConfig(informerFactory.Core().V1().Services(), s.ConfigSyncPeriod)
  	serviceConfig.RegisterEventHandler(s.Proxier)
  	go serviceConfig.Run(wait.NeverStop)
  
      // endpoints监听
  	if endpointsHandler, ok := s.Proxier.(config.EndpointsHandler); ok && !s.UseEndpointSlices {
  		endpointsConfig := config.NewEndpointsConfig(informerFactory.Core().V1().Endpoints(), s.ConfigSyncPeriod)
  		endpointsConfig.RegisterEventHandler(endpointsHandler)
  		go endpointsConfig.Run(wait.NeverStop)
      // endpointSlices监听
  	} else {
  		endpointSliceConfig := config.NewEndpointSliceConfig(informerFactory.Discovery().V1().EndpointSlices(), s.ConfigSyncPeriod)
  		endpointSliceConfig.RegisterEventHandler(s.Proxier)
  		go endpointSliceConfig.Run(wait.NeverStop)
  	}
  
  	// 启动informer
  	informerFactory.Start(wait.NeverStop)
  
      // 初始化nodeInformerFactory(本节点)
  	currentNodeInformerFactory := informers.NewSharedInformerFactoryWithOptions(s.Client, s.ConfigSyncPeriod,
  		informers.WithTweakListOptions(func(options *metav1.ListOptions) {
  			options.FieldSelector = fields.OneTermEqualSelector("metadata.name", s.NodeRef.Name).String()
  		}))
      // 监听node
  	nodeConfig := config.NewNodeConfig(currentNodeInformerFactory.Core().V1().Nodes(), s.ConfigSyncPeriod)
      // 注册更新proxier.podcidr回调
  	if s.localDetectorMode == kubeproxyconfig.LocalModeNodeCIDR {
  		nodeConfig.RegisterEventHandler(proxy.NewNodePodCIDRHandler(s.podCIDRs))
  	}
  	nodeConfig.RegisterEventHandler(s.Proxier)
  
  	go nodeConfig.Run(wait.NeverStop)
  
  	// 启动nodeInformerFactory
  	currentNodeInformerFactory.Start(wait.NeverStop)
  
  	// Birth Cry after the birth is successful
  	s.birthCry()
  
      // 启动proxier主循环(事件驱动刷新/定时刷新)
  	go s.Proxier.SyncLoop()
  
  	return <-errCh
  }
  

• 

• 注意

  1.kubeproxy会监听service/endpoints/node变化，触发proxyRules调整

  2.iptables/ipvs会定期(30s)同步规则，也会增量监听同步规则

  3.iptables/ipvs进行规则管理最终调用各自实现的syncProxyRules()

2.4.工作流程

• 

• 注意

  1.proxyServer负责总体调度，分发service/endpoint/node变更事件

  2.proxier负责同步变更事件，维护缓存中资源最新状态

  3.syncRunner负责定时或事件驱动更新规则