内置kube-proxy

eBPF的学习涉及内置kube-proxy协作，尤其是NAT相关的BPF Map更新会由这里处理，建议先了解该章节NAT Map处理逻辑

1.简介

1.1.背景

• 这里的kubeproxy不是原生的，是calico BPF模式内置的一个轻量替代品，绕过iptables/ipvs将service--endpoint同步到BPF Map。

• 

• 其实就是将本应同步到iptables/ipvs的内容写到BPF Map

1.2.启动

• calico BPF模式下，dataplane初始化会激活内置kubeproxy和conntrackScanner，同步的service/endpoint写到对应BPF Map。

• func NewIntDataplaneDriver(config Config) *InternalDataplane {
  	...
  
  	if config.BPFEnabled {
  		...
  
  		// service map更新
  		if config.KubeClientSet != nil {
  			// We have a Kubernetes connection, start watching services and populating the NAT maps.
  			kp, err := bpfproxy.StartKubeProxy(
  				config.KubeClientSet,
  				config.Hostname,
  				bpfMaps,
  				bpfproxyOpts...,
  			)
  			...
              
              // 这里会注册一下回调,hostIP变化会影响出主机流量的SNAT,需更新BPF MAP
  			bpfRTMgr.setHostIPUpdatesCallBack(kp.OnHostIPsUpdate)
              // 这里会注册路由回调,路由变化影响Nodeport流量
  			bpfRTMgr.setRoutesCallBacks(kp.OnRouteUpdate, kp.OnRouteDelete)
  			// 陈旧conntrack条目清理,避免影响正常流量,iptables模式下交给内核管的,这里需要自己管
  			conntrackScanner.AddUnlocked(bpfconntrack.NewStaleNATScanner(kp))
  			conntrackScanner.Start()
  		}
          ...
  	}
  	...
  
  	return dp
  }
  
  // StartKubeProxy start a new kube-proxy if there was no error
  func StartKubeProxy(k8s kubernetes.Interface, hostname string,
  	bpfMaps *bpfmap.Maps, opts ...Option) (*KubeProxy, error) {
  
  	kp := &KubeProxy{
  		k8s:         k8s,
  		hostname:    hostname,
  		frontendMap: bpfMaps.FrontendMap.(maps.MapWithExistsCheck),
  		backendMap:  bpfMaps.BackendMap.(maps.MapWithExistsCheck),
  		affinityMap: bpfMaps.AffinityMap,
  		ctMap:       bpfMaps.CtMap,
  		opts:        opts,
          rt:          NewRTCache(), // 路由缓存(供远程NodePort展开用)
  
  		hostIPUpdates: make(chan []net.IP, 1),
  		exiting:       make(chan struct{}),
  	}
  	...
      
  	go func() {
          // 后台启动,等待首次hostIPs更新
  		kp.start()
  		...
  	}()
  
  	return kp, nil
  }
  

• 注意

  路由变化会影响kubeproxy的rtCache，这个路由缓存影响的是NodePort这种流量的下一跳，会生成一条特殊的fe-be

1.3.start

• kp.start()会延迟启动，hostIP列表变化一次才会执行kp.run()实现三层BPF Map装配，这里会基于双重消费确保主机地址列表是最新的。

• func (kp *KubeProxy) start() error {
  
  	// wait for the initial update
  	hostIPs := <-kp.hostIPUpdates
  
      // 运行
  	kp.run(hostIPs)
  	...
      
  	go func() {
  		defer kp.wg.Done()
  		for {
              // hostIP同步异常,终止proxy
  			hostIPs, ok := <-kp.hostIPUpdates
  			if !ok {
  				kp.proxy.Stop()
  				return
  			}
  			...
  
              // 通知重新基于hostIP同步
  			go func() {
  				defer close(stopped)
  				kp.proxy.Stop()
  			}()
  
          // 内层处理
  		waitforstop:
  			for {
  				select {
                  // 获取新的hostIP
  				case hostIPs, ok = <-kp.hostIPUpdates:
  					if !ok {
  						return
  					}
  				case <-kp.exiting:
  					return
                  // 基于新的hostIP同步一次,回到外层
  				case <-stopped:
  					kp.run(hostIPs)
  					...
  					break waitforstop
  				}
  			}
  		}
  	}()
  
  	return nil
  }
  

• HostIP列表变化需要重新生成NodePort FrontendKey，所以这里必须停掉Proxy重跑

1.4.run

• kp.run()会组装hostIPs，基于封装的fe/be map加载内核实际NAT Entry，对比descibe计算pending差异，激活proxy同步配置。

• func (kp *KubeProxy) run(hostIPs []net.IP) error {
  	...
  
      // 组装NodePort IP列表: 本机所有IP + 255.255.255.255 wildcard
  	withLocalNP := make([]net.IP, len(hostIPs), len(hostIPs)+1)
  	copy(withLocalNP, hostIPs)
      // podNPIP = 255.255.255.255
  	withLocalNP = append(withLocalNP, podNPIP)
  
      // BPF Map封装,内部基于next批量读BPF Map KV
  	feCache := cachingmap.New[nat.FrontendKey, nat.FrontendValue](nat.FrontendMapParameters.Name,
  		maps.NewTypedMap[nat.FrontendKey, nat.FrontendValue](
  			kp.frontendMap, nat.FrontendKeyFromBytes, nat.FrontendValueFromBytes,
  		))
  	beCache := cachingmap.New[nat.BackendKey, nat.BackendValue](nat.BackendMapParameters.Name,
  		maps.NewTypedMap[nat.BackendKey, nat.BackendValue](
  			kp.backendMap, nat.BackendKeyFromBytes, nat.BackendValueFromBytes,
  		))
  
      // syncer初始化
  	syncer, err := NewSyncer(withLocalNP, feCache, beCache, kp.affinityMap, kp.rt)
  	...
      
      // 激活proxy同步
  	proxy, err := New(kp.k8s, syncer, kp.hostname, kp.opts...)
  	...
  
      // 这里会记录下,hostIP变化会停掉syncer重新跑
  	kp.proxy = proxy
  	kp.syncer = syncer
  
  	return nil
  }
  
  // NewSyncer returns a new Syncer
  func NewSyncer(...) (*Syncer, error) {
  
  	s := &Syncer{
  		bpfSvcs:     svcsmap,
  		bpfEps:      epsmap,
  		bpfAff:      affmap,
  		rt:          rt,
  		nodePortIPs: uniqueIPs(nodePortIPs),
  		prevSvcMap:  make(map[svcKey]svcInfo),
  		prevEpsMap:  make(k8sp.EndpointsMap),
  		stop:        make(chan struct{}),
  	}
  
      // 这里会加载一下内核fe/be map内容
  	s.loadOrigs()
      ...
  
  	return s, nil
  }
  
  func (s *Syncer) loadOrigs() error {
      // fe map加载,基于describe和actual对比计算pendingUpdate/pendingDeletion
  	s.bpfEps.LoadCacheFromDataplane()
  	...
      
      // be map加载,基于describe和actual对比计算pendingUpdate/pendingDeletion
  	s.bpfSvcs.LoadCacheFromDataplane()
  	...
      
  	return nil
  }
  

• 

• Syncer作为中间层将NAT数据同步到BPF Map，Proxy将监听事件推给Syncer

2.proxy

2.1.initialize

• proxy复用原生kube-proxy的ServiceConfig/EndpointSliceConfig，通过informer监听service/endpoint资源变化推到Syncer。

• // New returns a new Proxy for the given k8s interface
  func New(k8s kubernetes.Interface, dp DPSyncer, hostname string, opts ...Option) (Proxy, error) {
  	...
  
  	p := &proxy{
  		k8s:      k8s,
  		dpSyncer: dp,
  		hostname: hostname,
  		svcMap:   make(k8sp.ServicePortMap),
  		epsMap:   make(k8sp.EndpointsMap),
  		recorder: new(loggerRecorder),
  		minDPSyncPeriod: 30 * time.Second, // XXX revisit the default
  		stopCh: make(chan struct{}),
  	}
  	...
  
  	// 这是一个限流runner,对syncer做了包装,间隔1h同步一次,间隔30s生成一个令牌,减少BPF Map写入压力
      // svc/eps变化立即通知触发一次,不过会受到令牌桶限流限制,实现最小30s合并执行一次,最长1h周期同步一次
  	p.runner = async.NewBoundedFrequencyRunner("dp-sync-runner",
  		p.invokeDPSyncer, p.minDPSyncPeriod, time.Hour /* XXX might be infinite? */, 1)
      // 提供给外部的runner触发方式
  	dp.SetTriggerFn(p.runner.Run)
  
      // svc health checker
  	p.svcHealthServer = healthcheck.NewServiceHealthServer(p.hostname, p.recorder, []string{"0.0.0.0/0"})
  
      // change跟踪,确认svc/endpoint变化及历史状态累积
  	p.epsChanges = k8sp.NewEndpointChangeTracker(p.hostname, nil, v1.IPv4Protocol, p.recorder, nil)
  	p.svcChanges = k8sp.NewServiceChangeTracker(nil, v1.IPv4Protocol, p.recorder, nil)
  
  	noProxyName, err := labels.NewRequirement(apis.LabelServiceProxyName, selection.DoesNotExist, nil)
  	if err != nil {
  		return nil, errors.Errorf("noProxyName selector: %s", err)
  	}
      
      // 仅匹配kube-proxy的service,部分service基于这个注解会委托给其它controller
  	noHeadlessEndpoints, err := labels.NewRequirement(v1.IsHeadlessService, selection.DoesNotExist, nil)
  	...
  	labelSelector := labels.NewSelector()
  	labelSelector = labelSelector.Add(*noProxyName, *noHeadlessEndpoints)
  
      // svc监听
  	informerFactory := informers.NewSharedInformerFactoryWithOptions(k8s, p.syncPeriod,
  		informers.WithTweakListOptions(func(options *metav1.ListOptions) {
  			options.LabelSelector = labelSelector.String()
  		}))
  	svcConfig := config.NewServiceConfig(
  		informerFactory.Core().V1().Services(),
  		p.syncPeriod,
  	)
      // 回调注册
  	svcConfig.RegisterEventHandler(p)
  	...
  
      // endpoint监听
  	epsConfig := config.NewEndpointSliceConfig(informerFactory.Discovery().V1().EndpointSlices(), p.syncPeriod)
  	epsConfig.RegisterEventHandler(p)
  	...
  
      // BPF Map同步
  	p.startRoutine(func() { p.runner.Loop(p.stopCh) })
      // svc/informer监听,基于eventHandler触发一下syncDP
  	p.startRoutine(func() { epsRunner.Run(p.stopCh) })
  	p.startRoutine(func() { informerFactory.Start(p.stopCh) })
  	p.startRoutine(func() { svcConfig.Run(p.stopCh) })
  
  	return p, nil
  }
  

• 

• 注意

  这里的runner.Loop()基于事件驱动，svc/eps tracker基于历史状态和当前状态计算出有事件变更，会更新tracker.items及发同步信号

2.2.syncer

• runner.Loop()基于周期信号、变更信号及重试信号触发invoke同步，执行时会基于令牌桶限流，未申请到令牌会重置Timer等下一轮触发。

• // Loop handles the periodic timer and run requests.  This is expected to be called as a goroutine.
  func (bfr *BoundedFrequencyRunner) Loop(stop <-chan struct{}) {
      // 重置定时器,间隔1h
  	bfr.timer.Reset(bfr.maxInterval)
  	for {
  		select {
  		case <-stop:
  			bfr.stop()
  			return
          // 1h周期触发
  		case <-bfr.timer.C():
  			bfr.tryRun()
          // svc/eps变化触发
  		case <-bfr.run:
  			bfr.tryRun()
          // 基于retryTime重置Timer
  		case <-bfr.retry:
  			bfr.doRetry()
  		}
  	}
  }
  
  // assumes the lock is not held
  func (bfr *BoundedFrequencyRunner) doRetry() {
  	...
  	if bfr.retryTime.IsZero() {
  		return
  	}
  
  	// 重置Timer用于重试
  	retryInterval := bfr.retryTime.Sub(bfr.timer.Now())
  	bfr.retryTime = time.Time{}
  	if retryInterval < bfr.timer.Remaining() {
  		bfr.timer.Stop()
  		bfr.timer.Reset(retryInterval)
  	}
  }
  
  // assumes the lock is not held
  func (bfr *BoundedFrequencyRunner) tryRun() {
  	...
  
      // 申请令牌(30s生成一个)
  	if bfr.limiter.TryAccept() {
  		// 执行invoke同步
  		bfr.fn()
          // 重置Timer,1h后再触发
  		bfr.lastRun = bfr.timer.Now()
  		bfr.timer.Stop()
  		bfr.timer.Reset(bfr.maxInterval)
  		return
  	}
  	...
  }
  

• 

• 这里的核心就是bfr.fn()，其实就是限流器初始化传入的invokeDPSyncer

2.3.invoke

• p.invokeDPSyncer()是Map更新入口，svc/eps首次同步、Timer到期、svc/eps增量变化及重试都会触发，将最新的NAT数据发到内核。

• func (p *proxy) invokeDPSyncer() {
      // 同步完成才触发,避免fe--be不匹配
  	if !p.isInitialized() {
  		return
  	}
  	...
  
      // 将svc/eps tracker跟踪的pending合并到结果集
  	svcUpdateResult := p.svcMap.Update(p.svcChanges)
  	epsUpdateResult := p.epsMap.Update(p.epsChanges)
  
      // 更新svc/eps health checker,这会作为BPF Map更新参考,访问通过的才会加入BPF Map
  	p.svcHealthServer.SyncServices(svcUpdateResult.HCServiceNodePorts)
      ...
  	p.svcHealthServer.SyncEndpoints(epsUpdateResult.HCEndpointsLocalIPSize)
      ...
  
      // 下发到内核BPF Map,这会作为CTLB/TC NAT的数据来源
  	p.dpSyncer.Apply(DPSyncerState{
  		SvcMap:   p.svcMap,
  		EpsMap:   p.epsMap,
  		NodeZone: p.nodeZone,
  	})
  	...
  
      // kubeproxy更新健康,避免被重启
  	if p.healthzServer != nil {
  		p.healthzServer.Updated()
  	}
  }
  

• 

• 注意

  tracker.items会记录变化的svc/eps前后状态，下发内核前会合并结果KV以明确删除的旧状态和增量的新状态，health checker作为参考

3.merge

3.1.svcMap

• sm.Update()会将svcChange tracker跟踪的变更合并到svcMap，收集过期的udp clusterIP和健康检查端口，作为后续计算的svc来源。

• // Update updates ServicePortMap base on the given changes.
  func (sm ServicePortMap) Update(changes *ServiceChangeTracker) (result UpdateServiceMapResult) {
  	...
      
      // tracker跟踪的svc变更合并到servicePortMap,更新UDPStaleClusterIP
      // UDPStaleClusterIP: 记录被删除的UDP ClusterIP,用于清理conntrack残留条目
  	sm.apply(changes, result.UDPStaleClusterIP)
  	...
      
  	// 收集启用health checker的service port
      // LoadBalancer和externalTrafficPolicy=Local场景下,LB Controller可以设置这个NodePort端口
      // 这个端口可以探测这个节点有没有Pod,确保LB VIP可以将流量直接打到有Pod的节点
  	for svcPortName, info := range sm {
          // service设置了HealthCheckNodePort
  		if info.HealthCheckNodePort() != 0 {
  			result.HCServiceNodePorts[svcPortName.NamespacedName] = uint16(info.HealthCheckNodePort())
  		}
  	}
  
  	return result
  }
  
  // apply the changes to ServicePortMap and update the stale udp cluster IP set.
  // change.previous = ServicePortMap{
  //     {"ns/svc:http",  {ClusterIP:10.96.0.10, Port:80,  Proto:TCP,  HCNodePort:30080}},
  //     {"ns/svc:dns",   {ClusterIP:10.96.0.10, Port:53,  Proto:UDP,  HCNodePort:30080}},
  // }
  // change.current = ServicePortMap{
  //     {"ns/svc:http",  {ClusterIP:10.96.0.10, Port:80,  Proto:TCP,  HCNodePort:30080}},
  //     {"ns/svc:https", {ClusterIP:10.96.0.10, Port:443, Proto:TCP,  HCNodePort:30080}},
  // }
  // 合并结果为,svcMap保留80+443配置,清理53配置
  func (sm *ServicePortMap) apply(changes *ServiceChangeTracker, UDPStaleClusterIP sets.String) {
  	...
      
      // tracker跟踪的svc变更
  	for _, change := range changes.items {
  		...
          
          // current service port合并到svcMap
  		sm.merge(change.current)
  		// previous service过滤掉和current service重叠的端口
  		change.previous.filter(change.current)
          // 清理previous service port,这里删除的UDP ClusterIP会收集
  		sm.unmerge(change.previous, UDPStaleClusterIP)
  	}
      
  	// 重置svc tracker.items
  	changes.items = make(map[types.NamespacedName]*serviceChange)
  	...
  }
  

• 

• current和previous基于端口对比，格式为{namespace,name,port}

3.2.epsMap

• epsMap是svc关联的就绪地址列表，eps.Update()会将eps tracker增量变更合并到epsMap，还会收集过期的ep用于conntrack清理。

• // Update updates endpointsMap base on the given changes.
  func (em EndpointsMap) Update(changes *EndpointChangeTracker) (result UpdateEndpointMapResult) {
  	...
  
      // tracker跟踪的eps变更合并到epsMap,同步记录StaleEndpoints和StaleServiceNames
      // StaleEndpoints: 被删除的UDP端点(ep:port -> svcPortName)
      // StaleServiceNames: 从0个ready endpoint恢复到有ready endpoint的UDP Service
  	em.apply(changes, &result.StaleEndpoints, &result.StaleServiceNames, &result.LastChangeTriggerTimes)
  	...
      
      // 筛选本节点的Pod
  	localIPs := em.getLocalReadyEndpointIPs()
  	for nsn, ips := range localIPs {
          // 记录svc-->本地Pod数量,配合HealthCheckNodePort探测用
  		result.HCEndpointsLocalIPSize[nsn] = len(ips)
  	}
  
  	return result
  }
  
  // apply the changes to EndpointsMap and updates stale endpoints and service-endpoints pair.
  func (em EndpointsMap) apply(ect *EndpointChangeTracker, staleEndpoints *[]ServiceEndpoint,
  	staleServiceNames *[]ServicePortName, lastChangeTriggerTimes *map[types.NamespacedName][]time.Time) {
  	...
      
      // eps tracker跟踪的变化(eps模式由cache取,ep模式由items取)
  	changes := ect.checkoutChanges()
      
  	for _, change := range changes {
  		...
          // epsMap清理旧的endpoint
  		em.unmerge(change.previous)
          // epsMap合并新的endpoint
  		em.merge(change.current)
          // 检测过期的udp endpoint/service
  		detectStaleConnections(change.previous, change.current, staleEndpoints, staleServiceNames)
  	}
  	...
  }
  
  // detectStaleConnections modifies <staleEndpoints> and <staleServices> with detected stale connections.
  func detectStaleConnections(oldEndpointsMap, newEndpointsMap EndpointsMap, staleEndpoints *[]ServiceEndpoint, staleServiceNames *[]ServicePortName) {
  	
      // UDP是无连接的,conntrack不会自己清理,这里会检测过期的udp endpoint
  	for svcPortName, epList := range oldEndpointsMap {
  		if svcPortName.Protocol != v1.ProtocolUDP {
  			continue
  		}
  
  		for _, ep := range epList {
  			// 旧的端点不是ready,说明没有发过流量,不会有stale conntrack
  			if !ep.IsReady() {
  				continue
  			}
  			...
              
            	// 检查ep还在不在
  			for i := range newEndpointsMap[svcPortName] {
  				if newEndpointsMap[svcPortName][i].Equal(ep) {
  					stale = false
  					break
  				}
  			}
              
              // ep不在了,说明过期
  			if stale {
  				*staleEndpoints = append(*staleEndpoints, ServiceEndpoint{ ep.String(), svcPortName })
  			}
  		}
  	}
  
  	// 检测过期的service,还是只关心udp的
  	for svcPortName, epList := range newEndpointsMap {
  		if svcPortName.Protocol != v1.ProtocolUDP {
  			continue
  		}
  		...
  
          // Service1: VIP:8080 → 10.1.1.1:8080 VIP:8080 → 10.1.1.2:8080  Pod都挂在了
          // Service2: VIP:8080 → 10.1.1.3:8080  Pod就绪
          // 此时10.1.1.1:8080 10.1.1.2:8080残留conntrack
  		if epReady > 0 && oldEpReady == 0 {
  			*staleServiceNames = append(*staleServiceNames, svcPortName)
  		}
  	}
  }
  

• 

• 本地Pod统计

  PLB VIP会访问HealthCheckNodePort，对应的Listener监听后基于svc关联本地Pod数量应答，PLB根据应答节点调整节点后端池

3.3.checker

• health checker是端口http服务，LoadBalancer+externalTrafficLocal的service需要访问Pod所处节点才能正常，由这个服务应答。

• // 这个listener用于应带本节点有没有目标Pod,配置LB VIP更新NodeIP后端池
  func (hcs *server) SyncServices(newServices map[types.NamespacedName]uint16) error {
  	...
  
  	// 缓存的svc
  	for nsn, svc := range hcs.services {
          // svc HealthCheckNodePort不存在或变化
  		if port, found := newServices[nsn]; !found || port != svc.port {
  			// 停掉旧的listerner
  			_ = svc.closeAll()
  			delete(hcs.services, nsn)
  
  		}
  	}
  
  	// svc--->HealthCheckNodePort
  	for nsn, port := range newServices {
          // 注册过
  		if hcs.services[nsn] != nil {
  			continue
  		}
  
          // 注册一个listener监听HealthCheckNodePort流量
  		svc := &hcInstance{nsn: nsn, port: port}
  		svc.listenAndServeAll(hcs)
  		...
          
          // 缓存一下svc HealthCheckNodePort
  		hcs.services[nsn] = svc
  	}
  	return nil
  }
  
  // 这个就比较简单了,仅更新svc-->localPod数量映射,LB VIP周期访问就知道这个节点要不要加到后端池
  func (hcs *server) SyncEndpoints(newEndpoints map[types.NamespacedName]int) error {
  	...
  
      // svc-->len(localPod)
  	for nsn, count := range newEndpoints {
  		if hcs.services[nsn] == nil {
  			continue
  		}
  		
          // 更新一下svc本地Pod数量缓存
  		hcs.services[nsn].endpoints = count
  	}
      
      // 清理一下svc本地Pod数量缓存
  	for nsn, hci := range hcs.services {
  		if _, found := newEndpoints[nsn]; !found {
  			hci.endpoints = 0
  		}
  	}
  	return nil
  }
  

• 

• 这个端口健康检查在原生kube-proxy好像没碰到过，目前也是第一次看到这个用法

4.syncer

4.1.apply

• s.Apply()是syncer入口，尝试基于内核BPF Map和svcMap/epsMap重建prev状态及对比更新内核数据，还会用后台协程清理过期亲和后端。

• // Apply applies the new state
  func (s *Syncer) Apply(state DPSyncerState) error {
      // 首次启动
  	if !s.synced {
          // 基于BPF Map/svcMap/epsMap还原preMap状态
          // 重启后增量更新,避免全量刷写导致已有连接中断
  		s.startupSync(state)
          ...
  	} else {
  		...
          
          // 本地diff用上一次的newSvcMap/newEpsMap
  		s.prevSvcMap = s.newSvcMap
  		s.prevEpsMap = s.newEpsMap
  	}
  	...
  
      // 更新内核BPF Map
  	s.apply(state)
      ...
  
  	// we are fully synced now
  	if !s.synced {
  		s.synced = true
  	}
  
      // 清理过期/无效的会话亲和(Affinity)条目
      // 必须在apply()之后执行,确保stickySvcs/stickyEps更新完毕
  	return s.cleanupSticky()
  }
  
  func (s *Syncer) cleanupSticky() error {
  	..
  
      // 加载内核bpfAffMap内容
  	s.bpfAff.Iter(func(k, v []byte) maps.IteratorAction {
  		...
  
  		fend, ok := s.stickySvcs[key.FrontendAffinityKey()]
          // svc fend不在了,清理bpfAffMap
  		if !ok {
  			return maps.IterDelete
  		}
  
          // svc fend对应ep backend不在了,清理bpfAffMap
  		if _, ok := s.stickyEps[fend.id][val.Backend()]; !ok {
  			return maps.IterDelete
  		}
  
          // 距上次访问超出StickyMaxAgeSeconds,清理bpfAffMap
  		if now-val.Timestamp() > fend.timeo {
  			return maps.IterDelete
  		}
          
  		return maps.IterNone
  	})
  	...
      
  	return nil
  }
  

• 

• 注意

  这里比较重要的是s.apply()，会转换数据格式将service/endpoint更新到内核

4.1.writeSvc

• s.applySvc()根据本地优先原则将svc/backend写入内存BPF Map侧的pendingUpdate，另外还会维护设置亲和的svc和关联的eps。

• func (s *Syncer) applySvc(skey svcKey, sinfo k8sp.ServicePort, eps []k8sp.Endpoint) error {
  	...
  
  	old, exists := s.prevSvcMap[skey]
      // svc无变化,复用svcID
  	if exists && ServicePortEqual(old.svc, sinfo) {
  		id = old.id
  	} else {
  		id = s.newSvcID()
  	}
      
      // 更新内存的bpfSvc和bpfEps pending区
  	count, local, err := s.updateService(skey, sinfo, id, eps)
  	...
  
      // 更新newSvcMap
  	s.newSvcMap[skey] = svcInfo{ id: id, count: count, localCount: local, svc: sinfo }
  	...
  
  	return nil
  }
  
  func (s *Syncer) updateService(skey svcKey, sinfo ServicePort, id uint32, eps []Endpoint) (int, int, error) {
  	...
  
      // service设置亲和,初始化stickyEps
  	if sinfo.SessionAffinityType() == v1.ServiceAffinityClientIP {
  		s.stickyEps[id] = make(map[nat.BackendValue]struct{})
  	}
  
      // 先处理本地Pod,这里其实是一个优化,本地Pod匹配直接基于ordinal = prandom % loca计算,可以更快
  	for _, ep := range eps {
  		if !ep.GetIsLocal() {
  			continue
  		}
  
  		// backend就绪
  		if ep.IsReady() {
              // 更新bpfEps pending区和stickyEps
  			s.writeSvcBackend(id, uint32(cnt), ep)
              ...
  		}
  		...
  	}
  
      // 处理远端Pod
  	for _, ep := range eps {
  		if ep.GetIsLocal() {
  			continue
  		}
  
  		// backend就绪
  		if ep.IsReady() {
              // 更新bpfEps pending区和stickyEps
  			s.writeSvcBackend(id, uint32(cnt), ep)
              ...
  		}
  		...
  	}
  	...
  
      // 更新bpfSvc pending区
  	s.writeSvc(sinfo, id, cnt, local, flags)
      ...
  
      // 记录clusterIP记录的svc-->backends
  	if !hasSvcKeyExtra(skey, svcTypeNodePortRemote) {
  		s.newEpsMap[skey.sname] = cpEps
  	}
  
  	return cnt, local, nil
  }
  

• 

• 注意

  这里更新的都是clusterIP类型的主条目，其它类型会生成派生条目更新

6.4.writeDrv

• s.applyDerived()会生成LBIP/ExternalIP/NodePort的派生条目，LBIP/ExternalIP涉及的CIDR Range情况，会生成cidr entry。

• func (s *Syncer) applyDerived(sname k8sp.ServicePortName, t svcType, sinfo k8sp.ServicePort) error {
  	
      // 获取clusterIP条目,复用{ svcID,count,local }三元组
  	svc, ok := s.newSvcMap[getSvcKey(sname, "")]
  	...
  	skey = getSvcKey(sname, getSvcKeyExtra(t, sinfo.ClusterIP().String()))
  	...
  
      // 设置派生类型的flag
  	switch t {
  	case svcTypeNodePort, svcTypeLoadBalancer, svcTypeNodePortRemote:
  		if sinfo.ExternalPolicyLocal() {
  			flags |= nat.NATFlgExternalLocal
  		}
  		if sinfo.InternalPolicyLocal() {
  			flags |= nat.NATFlgInternalLocal
  		}
  	}
  
  	newInfo := svcInfo{
  		id:         svc.id,
  		count:      count,
  		localCount: local,
  		svc:        sinfo,
  	}
  
      // 派生条目复用主条目的backends,所以这里只更新bpfSvc
  	s.writeSvc(sinfo, svc.id, count, local, flags)
      ...
      
      // LBIP/ExternalIP类型的Range情况
  	if svcTypeLoadBalancer == t || svcTypeExternalIP == t {
  		s.writeLBSrcRangeSvcNATKeys(sinfo, svc.id, count, local, flags)
  		...
  	}
  
  	s.newSvcMap[skey] = newInfo
  	...
  
  	return nil
  }
  
  func (s *Syncer) writeLBSrcRangeSvcNATKeys(svc ServicePort, svcID uint32, count, local int, flags uint32) error{
  	...
      
      // service亲和时间设置
  	if svc.SessionAffinityType() == v1.ServiceAffinityClientIP {
  		affinityTimeo = uint32(svc.StickyMaxAgeSeconds())
  	}
  
  	if len(svc.LoadBalancerSourceRanges()) == 0 {
  		return nil
  	}
      
      // 不同源CIDR创建Frontend key
      // key = {clusterIP, Port, Proto, SrcCIDR}
  	keys, err := getSvcNATKeyLBSrcRange(svc)
  	...
      
      // 复用主条目的{ svcID, count, local }
  	val := nat.NewNATValueWithFlags(svcID, uint32(count), uint32(local), affinityTimeo, flags)
  	for _, key := range keys {
  		...
          
          // 更新bpfSvcs.pendingUpdate
  		s.bpfSvcs.SetDesired(key, val)
  	}
      
      // 主条目重新设置为BlackHole,BPF会丢弃clusterIP的流量
      // 限制匹配src_cidr的流量才能通过,其他流量被黑洞
  	key, err = getSvcNATKey(svc)
  	...
  	val = nat.NewNATValue(svcID, nat.BlackHoleCount, uint32(0), uint32(0))
  	s.bpfSvcs.SetDesired(key, val)
  	return nil
  }
  

• 

• 注意

  LB/ExternalIP类型设置SrcRange时，clusterIP Fe会标记为BlackHole(黑洞流量)，流量会被BPF丢弃

6.2.publish

• s.apply()会生成service/endpint对应的BPF Map格式数据发到内核，service对应的fe/be entry会覆盖更新，nodePort特殊处理。

• func (s *Syncer) apply(state DPSyncerState) error {
  	...
  
      // 先视为svc/eps需要全部删除(pendingDeletions)
  	s.bpfSvcs.DeleteAllDesired()
  	s.bpfEps.DeleteAllDesired()
  
  	// insert or update existing services
  	for sname, sinfo := range state.SvcMap {
          // service的zone注解
  		hintsAnnotation := sinfo.HintsAnnotation()
  		skey := getSvcKey(sname, "")
  		...
          
          // svc命中ep收集
  		for _, ep := range state.EpsMap[sname] {
              // 获取endpoint zone hints
  			zoneHints := ep.GetZoneHints()
              // ep就绪或正在终止才处理
              // Terminating的Pod也加入endpoints列表,支持优雅终止期间的连接draining
              // 但只有IsReady()的endpoint才真正写入Backend Map
  			if ep.IsReady() || ep.IsTerminating() {
                  // svc-node-ep zone匹配(nodeZone=nil也视为视为匹配)
  				if ShouldAppendTopologyAwareEndpoint(nodeZone, hintsAnnotation, zoneHints) {
  					eps = append(eps, ep)
  				}
  			}
  		}
  
          // clusterIP条目更新
  		s.applySvc(skey, sinfo, eps)
  		...
  
          // ============ 派生条目,复用主条目的svcID,复用主条目的backend ============
          // 多个入口共享同一个后端池和svcID,避免重复分配backend slot
          
          // 派生service LBIP
  		for _, lbIP := range sinfo.LoadBalancerIPStrings() {
  			if lbIP != "" {
  				extInfo := serviceInfoFromK8sServicePort(sinfo)
  				extInfo.clusterIP = net.ParseIP(lbIP)
                  // 派生LBIP条目更新
  				s.applyDerived(sname, svcTypeLoadBalancer, extInfo)
  				...
  			}
  		}
          
  		// 派生externalIP
  		for _, extIP := range sinfo.ExternalIPStrings() {
  			extInfo := serviceInfoFromK8sServicePort(sinfo)
  			extInfo.clusterIP = net.ParseIP(extIP)
              // externalIP条目更新
  			s.applyDerived(sname, svcTypeExternalIP, extInfo)
  			...
  		}
  
          // 派生nodePort
  		if nport := sinfo.NodePort(); nport != 0 {
  			for _, npip := range s.nodePortIPs {
  				npInfo := serviceInfoFromK8sServicePort(sinfo)
  				npInfo.clusterIP = npip
  				npInfo.port = nport
                  // podNPIP(255.255.255.255)是NodePort的wildcard(通配符) FE条目
                  // Pod访问<任意节点IP>:<NodePort>时:
                  //   1. 第一次NAT lookup: 用原始目标IP(节点IP)查 Frontend Map → 不命中
                  //   2. 检测到流量来自Pod或Host/CTLB,查路由表确认目标IP是一个host节点
                  //   3. 将nat_key.addr替换为255.255.255.255再次查找 → 命中通配条目
                  //   4. 使用该条目的svcID/backend列表完成 DNAT
                  // - internalTrafficPolicy=Cluster(默认): 通配条目存在,其backends包含所有Pod(本地+远端)
                  //   Pod访问任意节点IP的NodePort都能通过wildcard fallback命中
                  // - internalTrafficPolicy=Local,通配条目不存在
                  //   每个远端节点IP单独创建NodePortRemote FE条目(每个条目只含该节点上的Pod)
                  //   这样Pod访问远端NodePort时可直接DNAT到该节点上的Pod,不经过额外转发
  				if npip.Equal(podNPIP) && sinfo.InternalPolicyLocal() {
  					// do not program the meta entry, program each node
  					// separately
  					continue
  				}
                  
                   // nodePort条目更新
  				s.applyDerived(sname, svcTypeNodePort, npInfo)
  				...
  			}
              
              // InternalPolicy=Local时,展开远程NodePort(每个远端节点IP一个FE条目)
  			if sinfo.InternalPolicyLocal() {
                  // 写入newSvcMap
  				if miss := s.expandAndApplyNodePorts(sname, sinfo, eps, nport, s.rt.Lookup); miss != nil {
  					expNPMisses = append(expNPMisses, miss)
  				}
  			}
  		}
  	}
  
  	// 清理内核feMap过期数据
  	s.bpfSvcs.ApplyDeletionsOnly()
  	...
  	// 更新内核beMap增量数据
  	s.bpfEps.ApplyUpdatesOnly()
  	...
  	// 更新内核feMap增量数据
  	s.bpfSvcs.ApplyUpdatesOnly()
  	...
  	// 清理内核beMap过期数据
  	s.bpfEps.ApplyDeletionsOnly()
  	...
  
      // 路由未就绪的nodePort加入fixup队列,路由就绪补充entry缺失数据
  	s.runExpandNPFixup(expNPMisses)
  
  	return nil
  }
  

• 

• 注意

  这里的BPF Map更新顺序要严格遵守，避免出现svc指向无效backend或引用未就绪backend

6.5.expand

• s.expandAndApplyNodePorts()会为远端节点生成独立的fe-bes条目，NodePort流量到达本节点直接DNAT到目标节点Pod，避免SNAT。

• // NodePort流量到达一个没有本地Pod的节点时,传统kube-proxy会做SNAT转发到有Pod的节点,回来的流量又需要SNAT回来,多了一跳且丢失源IP
  // Calico DSR/Direct模式解决方案:
  //   - 不为255.255.255.255创建条目
  //   - 为每个远端节点IP创建一个NodePortRemote Frontend条目
  //   - 该条目的backend只包含该远端节点上的Pod
  //   - 这样当流量到达节点A,而Pod在节点B,节点A直接DNAT到节点B上的Pod,不做SNAT(DSR模式)
  //   - 回包直接从Pod所在节点返回,不经过节点A
  func (s *Syncer) expandAndApplyNodePorts(...) *expandMiss {
  
      // 根据路由表将endpoint按节点分组
  	ipToEp, miss := s.expandNodePorts(sname, sinfo, eps, nport, rtLookup)
  
  	for node, neps := range ipToEp {
          // 生成远端节点的fe-bes条目
  		s.applyExpandedNP(sname, sinfo, neps, node, nport)
          ...
  	}
  
  	return miss
  }
  
  func (s *Syncer) expandNodePorts(...) (map[ip.V4Addr][]k8sp.Endpoint, *expandMiss) {
  	...
      
  	for _, ep := range eps {
  		ipv4 := ip.FromString(ep.IP()).(ip.V4Addr)
  
          // 基于rtCache路由找一下Pod所在节点
          // 路由归bpfRouteManager维护,通过OnRouteUpdate回调更新RTCache
  		rt, ok := rtLookup(ipv4)
          // 未命中,记录miss
  		if !ok {
  			if miss == nil {
  				miss = &expandMiss{
  					sname: sname,
  					sinfo: sinfo,
  					nport: nport,
  				}
  			}
  			miss.eps = append(miss.eps, ep)
  			continue
  		}
  
  		flags := rt.Flags()
  		// Pod路由&找到对应节点,分组记录
  		if flags&routes.FlagWorkload != 0 && flags&routes.FlagLocal == 0 {
  			nodeIP := rt.NextHop().(ip.V4Addr)
  			ipToEp[nodeIP] = append(ipToEp[nodeIP], ep)
  			...
  		}
  	}
  	return ipToEp, miss
  }
  
  func (s *Syncer) applyExpandedNP(...) error {
      // {svcName, "NodePortRemote:" + nodeIP}
  	skey := getSvcKey(sname, getSvcKeyExtra(svcTypeNodePortRemote, node.String()))
  	si := serviceInfoFromK8sServicePort(sinfo)
      // 注意这里,clusterIP替换为nodeIP
  	si.clusterIP = node.AsNetIP()
  	si.port = nport
  
      // remote nodePort条目更新
      // 分配新的svcID(skey不同且不是主条目),backend列表仅有节点上的Pod
  	s.applySvc(skey, si, eps)
      ...
  
  	return nil
  }
  

• 

• 注意

  这里未命中的miss需要特别关注下，会启动一个后台协程waitRTCache阻塞，路由推过来再触发一次nodePortRemote条目更新

6.6.runExpand

• s.runExpandNPFixup()会执行nodePortRemote entry修复，rtCache路由就绪再触发一次publish补充缺失的nodePort svc条目。

• // expandNodePorts依赖RTCache中的路由信息来确定Pod节点
  // svc/eps可能先于路由到达,此时路由查不到,产生miss
  // fixup goroutine会阻塞等待RTCache更新触发重试
  func (s *Syncer) runExpandNPFixup(misses []*expandMiss) {
  	if len(misses) == 0 {
  		return
  	}
      ...
  
  	// start the fixer routine and exit
  	go func() {
  		...
  
  		for {
  			// 基于condition阻塞,rtCache更新会唤醒
  			s.rt.WaitAfter(ctx, func(lookup func(addr ip.Addr) (routes.Value, bool)) bool {
  				...
  				for _, m := range misses {
                      // 再次尝试展开
  					if _, miss := s.expandNodePorts(m.sname, m.sinfo, m.eps, m.nport, lookup); miss != nil {
  						again = append(again, miss)
  						if !reflect.DeepEqual(m.eps, miss.eps) {
  							missesChanged = true
  						}
  					} else {
  						missesChanged = true
  					}
  				}
  
                  // rtCache补充了一些路由,触发publish
  				if missesChanged && s.triggerFn != nil {
  					s.triggerFn()
  				}
                  
  				misses = again
  				return len(misses) == 0 // block or not block
  			})
  
  			if len(misses) == 0 || ctx.Err() != nil {
  				return
  			}
  		}
  	}()
  }
  

• 

• 补充

  rtCache更新尝试修复一部分，直到全部修复，否则继续阻塞等待下一轮