dataplane

1.入口

1.1.start

• felix会订阅datastore相关事件。基于变更事件计算增量diff，经过多次数据转发写入iptables/ebpf规则和路由条目，确保应用流量正常。

• // Run is the entry point to run a Felix instance.
  func Run(configFile string, gitVersion string, buildDate string, gitRevision string) {
  	...
  
  configRetry:
  	for {
  		...
  
          // 1.加载ENV配置
  		envConfig := config.LoadConfigFromEnvironment(os.Environ())
  		// 2.读取配置文件
  		fileConfig, err := config.LoadConfigFile(configFile)
  		...
          
  		// 3.解析及合并配置
  		_, err = configParams.UpdateFrom(envConfig, config.EnvironmentVariable)
  		...
  		_, err = configParams.UpdateFrom(fileConfig, config.ConfigFile)
  		...
  
          // 4.连接datastore(etcd/kube)
  		v3Client, err = client.New(datastoreConfig)
  		...
  		backendClient = v3Client.(interface{ Backend() bapi.Client }).Backend()
          
          // 5.加载及合并配置(clusterInformation/felixConfiguration/calico-node)
  		for {
  			globalConfig, hostConfig := loadConfigFromDatastore(
  				ctx, backendClient, datastoreConfig, configParams.FelixHostname)
  			...
              
  			configParams.UpdateFrom(globalConfig, config.DatastoreGlobal)
  			...
  			configParams.UpdateFrom(hostConfig, config.DatastorePerHost)
  			...
              
  			break
  		}
          
          // 6.参数校验
  		configParams.Validate()
  		...
  
  		// 7.获取所有IPPool
  		ippoolKVPList, err := backendClient.List(ctx, model.ResourceListOptions{Kind: apiv3.KindIPPool}, "")
  		...
          
          // 8.计算封装模式
  		encapCalculator := calc.NewEncapsulationCalculator(configParams, ippoolKVPList)
  		configParams.Encapsulation.IPIPEnabled = encapCalculator.IPIPEnabled()
  		configParams.Encapsulation.VXLANEnabled = encapCalculator.VXLANEnabled()
  		configParams.Encapsulation.VXLANEnabledV6 = encapCalculator.VXLANEnabledV6()
  		...
          
          // 9.配置被更新,重连datastore
  		backendClient, err = backend.NewClient(datastoreConfig)
  		...
  
  		// 10.初始化typha扫描器
  		typhaDiscoverer = createTyphaDiscoverer(configParams, k8sClientSet)
  		...
  
  		break configRetry
  	}
  	...
  
      // 11.BPF支持检查
  	if configParams.BPFEnabled {
  		if err := dp.SupportsBPF(); err != nil {
  			configParams.OverrideParam("BPFEnabled", "false")
  			...
  		}
  	}
  	...
  
      // 12.激活dataplane driver(iptables/eBPF)
  	dpDriver, dpDriverCmd = dp.StartDataplaneDriver(
  		configParams.Copy(), // Copy to avoid concurrent access.
  		healthAggregator, configChangedRestartCallback, fatalErrorCallback, k8sClientSet)
  	...
      
      // 13.初始化dataplane连接器(calGraph-->connector-->dataplane driver)
  	dpConnector := newConnector(
  		configParams.Copy(), // Copy to avoid concurrent access.
  		connToUsageRepUpdChan, backendClient, v3Client, dpDriver, failureReportChan)
  	...
      
      // 14.初始化policy变化订阅(calGraph-->toPolicySync-->policySyncProcessor-->client) 
  	calcGraphClientChannels := []chan<- interface{}{dpConnector.ToDataplane}
  	if configParams.IsLeader() && configParams.PolicySyncPathPrefix != "" {
  		toPolicySync := make(chan interface{})
  		policySyncUIDAllocator := policysync.NewUIDAllocator()
  		policySyncProcessor = policysync.NewProcessor(toPolicySync)
  		policySyncServer = policysync.NewServer(policySyncProcessor.JoinUpdates, policySyncUIDAllocator.NextUID)
  		policySyncAPIBinder = binder.NewBinder(configParams.PolicySyncPathPrefix)
  		policySyncServer.RegisterGrpc(policySyncAPIBinder.Server())
  		calcGraphClientChannels = append(calcGraphClientChannels, toPolicySync)
  	}
      ...
      
  	//
  	// Syncer -chan-> Validator -chan-> Calc graph -chan->   dataplane
  	//        KVPair            KVPair             protobufs
  
  	// Get a Syncer from the datastore, or a connection to our remote sync daemon, Typha,
  	// which will feed the calculation graph with updates, bringing Felix into sync.
  
      syncerToValidator := calc.NewSyncerCallbacksDecoupler()
      // 15.typha模式
  	if typhaDiscoverer.TyphaEnabled() {
  		// Use a remote Syncer, via the Typha server.
  		typhaConnection = syncclient.New(
  			typhaDiscoverer,
  			buildinfo.GitVersion,
  			configParams.FelixHostname,
  			fmt.Sprintf("Revision: %s; Build date: %s", buildinfo.GitRevision, buildinfo.BuildDate),
  			syncerToValidator,
  			&syncclient.Options{
  				ReadTimeout:  configParams.TyphaReadTimeout,
  				WriteTimeout: configParams.TyphaWriteTimeout,
  				KeyFile:      configParams.TyphaKeyFile,
  				CertFile:     configParams.TyphaCertFile,
  				CAFile:       configParams.TyphaCAFile,
  				ServerCN:     configParams.TyphaCN,
  				ServerURISAN: configParams.TyphaURISAN,
  			},
  		)
      // 16.syncer模式
  	} else {
  		// Use the syncer locally.
  		syncer = felixsyncer.New(backendClient, datastoreConfig.Spec, syncerToValidator,configParams.IsLeader())
  		configParams.SetUseNodeResourceUpdates(true)
  	}...
  
  	// 16.backend启动
  	if syncer != nil {
  		syncer.Start()
  	} else {
  		startTime := time.Now()
  		for attempt := 1; ; attempt++ {
              ...
  			// Try to connect to Typha, this actually tries all available Typha instances before it returns.
  			typhaConnection.Start(context.Background())
  			...
              
  			break
  		}
  		...
  
          // 协商增量资源更新
  		supportsNodeResourceUpdates, err := typhaConnection.SupportsNodeResourceUpdates(10 * time.Second)
  		...
  		configParams.SetUseNodeResourceUpdates(supportsNodeResourceUpdates)
  		...
  	}
  
  	// 17.初始化Policy规则图计算模块
  	asyncCalcGraph := calc.NewAsyncCalcGraph(configParams.Copy(), calcGraphClientChannels, healthAggregator)
      ...
  
  	// 18.初始化校验器,位于syncer-->calGraph之间
  	validator := calc.NewValidationFilter(asyncCalcGraph, configParams)
  
      // 19.建立syncer-->validator链接
  	go syncerToValidator.SendTo(validator)
      
      // 20.启动calGraph
  	asyncCalcGraph.Start()
  	...
  
  	// 21.启动dataplane connector作为calGraph-->dataplane桥梁
  	dpConnector.Start()
  
      // 22.启动policy syncer供外部client查询策略
  	if policySyncProcessor != nil {
  		policySyncProcessor.Start()
  		sc := make(chan *sync.WaitGroup)
  		stopSignalChans = append(stopSignalChans, sc)
  		go policySyncAPIBinder.SearchAndBind(sc)
  	}
  
      // 23.向dataplane发送第一条消息(配置)
  	dpConnector.ToDataplane <- configParams.ToConfigUpdate()
  	...
  
  	// 24.注册优雅退出
  	monitorAndManageShutdown(failureReportChan, dpDriverCmd, stopSignalChans)
  }
  

• 

• 注意

  syncer/typha同步的事件会根据上图流向转到dataplane和policySyncerProcessor

1.2.dataplane

• dp.StartDataplaneDriver()负责分配标记为、路由表等资源，初始化内置或外部dataplane及启动，基于不同模式管理容器网络流量。

• func StartDataplaneDriver(...) (DataplaneDriver, *exec.Cmd) {
  
      // 1.仅leader执行dataplane,非leader同步dataplane计算的规则
  	if !configParams.IsLeader() {
  		// Return an inactive dataplane, since we're not the leader.
  		return &inactive.InactiveDataplane{}, nil
  	}
  
      // 2.iptables/ebpf dataplane
  	if configParams.UseInternalDataplaneDriver {
  		// 2.1.ipvs模式检查
  		kubeIPVSSupportEnabled := false
  		if ifacemonitor.IsInterfacePresent(intdataplane.KubeIPVSInterface) {
  			if !configParams.BPFEnabled {
  				kubeIPVSSupportEnabled = true
  			}
  		}
  		...
  
          // 2.2.iptable初始标志位
  		allowedMarkBits := configParams.IptablesMarkMask
          
          // 2.3.剔除eBPF固定占用位(0x1ff00000)
  		if configParams.BPFEnabled {
              
  			...
  			allowedMarkBits ^= allowedMarkBits & tcdefs.MarksMask
  		}
  
          // 2.3.初始化markMgr
  		markBitsManager := markbits.NewMarkBitsManager(allowedMarkBits, "felix-iptables")
  		...
  
          // 2.4.跨chain放行标志(0x00100000)
  		markAccept, _ = markBitsManager.NextSingleBitMark()
  
          // 2.5.端点链放行标志(0x00200000)
  		markPass, _ = markBitsManager.NextSingleBitMark()
  
          // 2.6.多条rule计算的临时标记(0x00400000/0x00800000)
  		markScratch0, _ = markBitsManager.NextSingleBitMark()
  		markScratch1, _ = markBitsManager.NextSingleBitMark()
  
          // 2.7.wireGuard流量标记(0x01000000)
  		if configParams.WireguardEnabled || configParams.WireguardEnabledV6 {
  			markWireguard, _ = markBitsManager.NextSingleBitMark()
  			...
  		}
  		...
  
          // 2.8.非calico-ep标记(0x02000000)
  		markEndpointMark, allocated :=markBitsManager.NextBlockBitsMark(markBitsManager.AvailableMarkBitCount())
  		if kubeIPVSSupportEnabled {
  			...
  			// Take lowest bit position (position 1) from endpoint mark mask reserved for non-calico endpoint.
  			markEndpointNonCaliEndpoint = uint32(1) << uint(bits.TrailingZeros32(markEndpointMark))
  		}
  
  		// 2.9.初始化路由索引分配器
  		reservedTables := []idalloc.IndexRange{{Min: 253, Max: 255}}
  		routeTableIndexAllocator := idalloc.NewIndexAllocator(configParams.RouteTableIndices(), reservedTables)
  		...
          
  		// 2.10.WireGuard分配路由表ID 
  		if idx, err := routeTableIndexAllocator.GrabIndex(); err == nil {
  			wireguardEnabled = configParams.WireguardEnabled
  			wireguardTableIndex = idx
  		}
  		...
          
  		if idx, err := routeTableIndexAllocator.GrabIndex(); err == nil {
  			wireguardEnabledV6 = configParams.WireguardEnabledV6
  			wireguardTableIndexV6 = idx
  		}
  		...
  
  		if k8sClientSet != nil {
  
  			// 2.11.获取kubernetes node
  			felixNode, err := k8sClientSet.CoreV1().Nodes().Get(context.Background(), felixHostname, ...)
  			...
  
              // 2.12.提取zoneLabel
  			felixNodeZone = felixNode.Labels[coreV1.LabelTopologyZone]
  		}
  
  		dpConfig := intdataplane.Config{ ... }
  
          // 2.13.DSR模式(Client → Node → Pod → Node → Client转为Client → Node → Pod → Client)
  		if configParams.BPFExternalServiceMode == "dsr" {
  			dpConfig.BPFNodePortDSREnabled = true
              // 不走DSR的网段
  			dpConfig.BPFDSROptoutCIDRs = configParams.BPFDSROptoutCIDRs
  		}
  
          // 2.14.启动dataplane
  		intDP := intdataplane.NewIntDataplaneDriver(dpConfig)
  		intDP.Start()
  		...
  
  		return intDP, nil
      // 3.启动外部dataplane
  	} else {
  		return extdataplane.StartExtDataplaneDriver(configParams.DataplaneDriver)
  	}
  }
  

• 

• 注意

  intdataplane.NewIntDataplaneDriver()初始化及启动的是外部dataplane，这里只分析internalDataplane

2.启动

2.1.initialize

• intdataplane.NewIntDataplaneDriver()根据配置初始化数据平面，涉及iptables/route/eBPF/service接管...，跟踪所有网络功能。

• func NewIntDataplaneDriver(config Config) *InternalDataplane {
      // 1.初始化ruleRenader(policy-->iptables)
      ruleRenderer := config.RuleRendererOverride
  	if ruleRenderer == nil {
  		ruleRenderer = rules.NewRenderer(config.RulesConfig)
  	}
      
      // 2.ep标记,区分calico endpoint
  	epMarkMapper := rules.NewEndpointMarkMapper(IptablesMarkEndpoint,IptablesMarkNonCaliEndpoint)
  
      // 3.探测宿主机MTU(1460)
  	hostMTU, err := findHostMTU(config.MTUIfacePattern)
  	...
      
      // 4.设置不同模式的MTU
  	ConfigureDefaultMTUs(hostMTU, &config)
      
      // 5.取最小作为Pod MTU写入/var/lib/calico/mtu
  	podMTU := determinePodMTU(config)
  	writeMTUFile(podMTU)
      ...
  
      // 6.初始化dataplane
  	dp := &InternalDataplane{ ... }
  	dp.applyThrottle.Refill()
  	dp.ifaceMonitor.StateCallback = dp.onIfaceStateChange
  	dp.ifaceMonitor.AddrCallback = dp.onIfaceAddrsChange
  	dp.ifaceMonitor.InSyncCallback = dp.onIfaceInSync
  
      // 7.iptables后端检测
  	backendMode := environment.DetectBackend(config.LookPathOverride, cmdshim.NewRealCmd, IptablesBackend)
  
  	// Most iptables tables need the same options.
  	iptablesOptions := iptables.TableOptions{... BackendMode: backendMode ... }
  
      // 8.BPF模式匹配待清理的kubeproxy规则
  	if config.BPFEnabled && config.BPFKubeProxyIptablesCleanupEnabled {
  		// If BPF-mode is enabled, clean up kube-proxy's rules too.
  		iptablesOptions.ExtraCleanupRegexPattern = rules.KubeProxyInsertRuleRegex
  		iptablesOptions.HistoricChainPrefixes = append(iptablesOptions.HistoricChainPrefixes, rules.KubeProxyChainPrefixes...)
  	}
  	...
  
      // 9.NAT特殊处理(匹配待清理的SNAT/DNAT/MASQ规则)
  	iptablesNATOptions := iptablesOptions
  	if iptablesNATOptions.ExtraCleanupRegexPattern == "" {
  		iptablesNATOptions.ExtraCleanupRegexPattern = rules.HistoricInsertedNATRuleRegex
  	} else {
  		iptablesNATOptions.ExtraCleanupRegexPattern += "|" + rules.HistoricInsertedNATRuleRegex
  	}
  	...
      
      // 10.关联四表对象
  	mangleTableV4 := iptables.NewTable("mangle", 4, rules.RuleHashPrefix, iptablesLock, featureDetector,
  		iptablesOptions)
      
  	natTableV4 := iptables.NewTable("nat", 4, rules.RuleHashPrefix, iptablesLock, featureDetector,
  		iptablesNATOptions)
      
  	rawTableV4 := iptables.NewTable("raw", 4, rules.RuleHashPrefix, iptablesLock, featureDetector,
  		iptablesOptions)
      
  	filterTableV4 := iptables.NewTable("filter", 4, rules.RuleHashPrefix, iptablesLock, featureDetector,
  		iptablesOptions)
    	...
  	
      // 11.初始化ipset对象
  	ipSetsV4 := ipsets.NewIPSets(ipSetsConfigV4, dp.loopSummarizer)
      
      // 12.注册四表及IPSET
  	dp.iptablesNATTables = append(dp.iptablesNATTables, natTableV4)
  	dp.iptablesRawTables = append(dp.iptablesRawTables, rawTableV4)
  	dp.iptablesMangleTables = append(dp.iptablesMangleTables, mangleTableV4)
  	dp.iptablesFilterTables = append(dp.iptablesFilterTables, filterTableV4)
  	dp.ipSets = append(dp.ipSets, ipSetsV4)
  
      // 13.valan模式
  	if config.RulesConfig.VXLANEnabled {
  		...
          // 13.1.valan路由表初始化
  		if !config.RouteSyncDisabled {
  			routeTableVXLAN = routetable.New([]string{"^vxlan.calico$"}, 4, true, config.NetlinkTimeout,
  				config.DeviceRouteSourceAddress, config.DeviceRouteProtocol, true, unix.RT_TABLE_MAIN,
  				dp.loopSummarizer, featureDetector, routetable.WithLivenessCB(dp.reportHealth))
  		} else {
  			routeTableVXLAN = &routetable.DummyTable{}
  		}
  
          // 13.2.初始化vxlanMgr(device/fdb/arp/route/vtep维护)
  		vxlanManager := newVXLANManager(
  			ipSetsV4, routeTableVXLAN, "vxlan.calico", config, dp.loopSummarizer, 4, featureDetector)
  		
          // 13.3.启动同步协程(检测/修复device)
          go vxlanManager.KeepVXLANDeviceInSync(config.VXLANMTU, dataplaneFeatures.ChecksumOffloadBroken, 10*time.Second)
          
          // 13.4.注册到dataplane
  		dp.RegisterManager(vxlanManager)
  	} else {
  		// 13.5.清理旧设备
  		go cleanUpVXLANDevice("vxlan.calico")
  	}
  	...
  
      // 14.设置回调
  	callbacks := common.NewCallbacks()
  	dp.callbacks = callbacks
      
      // 15.初始化XDP加速
  	if config.XDPEnabled {
          // 15.1.XDP支持检查
  		if err := bpf.SupportsXDP(); err != nil {
  			config.XDPEnabled = false
          // 15.2.iptables模式
  		} else if !config.BPFEnabled {
              // 初始化xdpStateMgr
  			st, err := NewXDPState(config.XDPAllowGeneric)
  			...
              dp.xdpState = st
              
              // 补充callback及注册
              dp.xdpState.PopulateCallbacks(callbacks)
              dp.RegisterManager(st)
  		}
  	}
  
  	// 16.iptables模式+XDP残留清理
  	if !config.BPFEnabled && dp.xdpState == nil {
          // 清理残留的XDP程序
  		xdpState, err := NewXDPState(config.XDPAllowGeneric)
  		...
          xdpState.WipeXDP()
          ...
  	}
  
      // 17.尝试开启sockMap加速
  	if config.SidecarAccelerationEnabled {
          // 17.1.检测sockMap支持
  		bpf.SupportsSockmap()
          ...
          
          // 17.2.初始化socktMapStateMgr
          st, err := NewSockmapState()
          ...
          dp.sockmapState = st
          
          // 17.3.注册回调
          dp.sockmapState.PopulateCallbacks(callbacks)
  
          // 17.4.加载eBPF程序实现加速
          dp.sockmapState.SetupSockmapAcceleration()
          ...
  	}
  
      // 18.清理过期的sockMap
  	if dp.sockmapState == nil {
  		st, err := NewSockmapState()
  		...
          st.WipeSockmap(bpf.FindInBPFFSOnly)
  	}
  
      // 19.初始化ipsetMgr及注册
  	ipsetsManager := common.NewIPSetsManager(ipSetsV4, config.MaxIPSetSize)
  	dp.RegisterManager(ipsetsManager)
  
      // 20.iptables模式
  	if !config.BPFEnabled {
  		// 20.1.设置ipsetMgr
  		dp.ipsetsSourceV4 = ipsetsManager
  		// 20.2.注册hostIPMgr及policyMgr
  		dp.RegisterManager(newHostIPManager(config.RulesConfig.WorkloadIfacePrefixes, rules.IPSetIDThisHostIPs,
  			ipSetsV4, config.MaxIPSetSize))
  		dp.RegisterManager(newPolicyManager(rawTableV4, mangleTableV4, filterTableV4, ruleRenderer, 4))
  
          // 20.3.清理上一次残留的BPF资源
  		// Clean up any leftover BPF state.
  		err := bpfnat.RemoveConnectTimeLoadBalancer("")
  		...
  		tc.CleanUpProgramsAndPins()
     	// 21.BPF模式
  	} else {
  		// 21.1.注册iptables policyMgr作为出口策略
  		dp.RegisterManager(newRawEgressPolicyManager(rawTableV4, ruleRenderer, 4,
  			func(neededIPSets set.Set[string]) {
  				ipSetsV4.SetFilter(neededIPSets)
  			}))
  	}
  	...
  
      // 22.Pod网卡匹配表达式
  	for i, r := range config.RulesConfig.WorkloadIfacePrefixes {
  		interfaceRegexes[i] = "^" + r + ".*"
  	}
  
      // 23.读取反向包过滤配置
  	defaultRPFilter, err := os.ReadFile("/proc/sys/net/ipv4/conf/default/rp_filter")
  	if err != nil {
  		defaultRPFilter = []byte{'1'}
  	}
  	...
  
      // 24.设置BPF表大小
  	bpfipsets.SetMapSize(config.BPFMapSizeIPSets)
  	bpfnat.SetMapSizes(config.BPFMapSizeNATFrontend, config.BPFMapSizeNATBackend, config.BPFMapSizeNATAffinity)
  	bpfroutes.SetMapSize(config.BPFMapSizeRoute)
  	bpfconntrack.SetMapSize(config.BPFMapSizeConntrack)
  	bpfifstate.SetMapSize(config.BPFMapSizeIfState)
  	...
  
      // 25.BPF模式
  	if config.BPFEnabled {
          // 25.1.初始化BPF MAP
  		bpfMaps = bpfmap.CreateBPFMaps()
  		...
          
  		// 25.2.初始化BPF IPSET及注册
  		ipSetIDAllocator := idalloc.New()
  		ipSetsV4 := bpfipsets.NewBPFIPSets(ipSetsConfigV4,ipSetIDAllocator,bpfMaps.IpsetsMap, dp.loopSummarizer)
  		dp.ipSets = append(dp.ipSets, ipSetsV4)
          ipsetsManager.AddDataplane(ipSetsV4)
          
          // 25.3.初始化BPF RouteMgr及注册
  		bpfRTMgr := newBPFRouteManager(&config, bpfMaps, dp.loopSummarizer)
  		dp.RegisterManager(bpfRTMgr)
  
          // 25.4.IPIP模式禁用FIB Lookup(BPF对L3 Device支持不完整)
  		fibLookupEnabled := !config.RulesConfig.IPIPEnabled
          
          // 25.5.初始化安全规则管理及注册,允许内部流量走白名单
  		failsafeMgr := failsafes.NewManager(bpfMaps.FailsafesMap,
  			config.RulesConfig.FailsafeInboundHostPorts,
  			config.RulesConfig.FailsafeOutboundHostPorts,
  			dp.loopSummarizer,
  		)
  		dp.RegisterManager(failsafeMgr)
  
          // 25.6.Pod网卡匹配正则
  		workloadIfaceRegex := regexp.MustCompile(strings.Join(interfaceRegexes, "|"))
          // 初始化BPF EPMgr及注册
  		bpfEndpointManager, err = newBPFEndpointManager(
  			nil, &config, bpfMaps, fibLookupEnabled, workloadIfaceRegex, ipSetIDAllocator,
  			ruleRenderer, filterTableV4, dp.reportHealth, dp.loopSummarizer, featureDetector)
  		...
  		dp.RegisterManager(bpfEndpointManager)
  		...
  
          // 25.7.初始化conntrack scanner,扫描conntrack map及清理
  		conntrackScanner := bpfconntrack.NewScanner(bpfMaps.CtMap,
  			bpfconntrack.NewLivenessScanner(config.BPFConntrackTimeouts, config.BPFNodePortDSREnabled))
  		conntrackScanner.Scan()
  
  		// 25.8.kube-proxy承接        
  		bpfproxyOpts := []bpfproxy.Option{
  			bpfproxy.WithMinSyncPeriod(config.KubeProxyMinSyncPeriod),
  		}
  
          // DSR功能
  		if config.BPFNodePortDSREnabled {
  			bpfproxyOpts = append(bpfproxyOpts, bpfproxy.WithDSREnabled())
  		}
  
          // 流量拓扑转发
  		if len(config.NodeZone) != 0 {
  			bpfproxyOpts = append(bpfproxyOpts, bpfproxy.WithTopologyNodeZone(config.NodeZone))
  		}
  
          // 启动BPF kube-proxy
  		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/Route变化,更新NAT
  			bpfRTMgr.setHostIPUpdatesCallBack(kp.OnHostIPsUpdate)
  			bpfRTMgr.setRoutesCallBacks(kp.OnRouteUpdate, kp.OnRouteDelete)
              
              // 清理Svc删除的NAT State
  			conntrackScanner.AddUnlocked(bpfconntrack.NewStaleNATScanner(kp))
              // 启动contrack scanner loop
  			conntrackScanner.Start()
  		}
          ...
  
          // 启用BPF CTLB
  		if config.BPFConnTimeLBEnabled {
  			...
              
  			// 加载eBPF对象Attach到CGroup
  			bpfnat.InstallConnectTimeLoadBalancer(
  				config.BPFCgroupV2, config.BPFLogLevel, config.BPFConntrackTimeouts.UDPLastSeen, excludeUDP)
  			...
          // 由CGroupV2卸载CTLB
  		} else {
  			// Deactivate the connect-time load balancer.
  			nat.RemoveConnectTimeLoadBalancer(config.BPFCgroupV2)
  			...
  		}
  	}
  	...
  
      // 26.routeMgr初始化
  	if !config.RouteSyncDisabled {
  		routeTableV4 = routetable.New(interfaceRegexes, 4, false, config.NetlinkTimeout,
  			config.DeviceRouteSourceAddress, config.DeviceRouteProtocol, config.RemoveExternalRoutes, unix.RT_TABLE_MAIN,
  			dp.loopSummarizer, featureDetector, routetable.WithLivenessCB(dp.reportHealth),
  			routetable.WithRouteCleanupGracePeriod(routeCleanupGracePeriod))
  	} else {
  		routeTableV4 = &routetable.DummyTable{}
  	}
  
      // 27.EPMgr初始化及注册
  	epManager := newEndpointManager(...)
  	dp.RegisterManager(epManager)
  	dp.endpointsSourceV4 = epManager
      
      // 28.初始化FloatingIPMgr(SNAT/DNAT)及注册
  	dp.RegisterManager(newFloatingIPManager(natTableV4, ruleRenderer, 4, config.FloatingIPsEnabled))
      // 29.初始化MASQMgr
  	dp.RegisterManager(newMasqManager(ipSetsV4, natTableV4, ruleRenderer, config.MaxIPSetSize, 4))
  	
      // 29.IPIP模式
      if config.RulesConfig.IPIPEnabled {
  		// 初始化IPIPMgr及注册
  		dp.ipipManager = newIPIPManager(ipSetsV4, config.MaxIPSetSize, config.ExternalNodesCidrs)
  		dp.RegisterManager(dp.ipipManager) // IPv4-only
  	} else {
  		// 清理IPIP遗留配置
  		if config.RulesConfig.FelixConfigIPIPEnabled == nil {
  			// Start a cleanup goroutine not to block felix if it needs to retry
  			go cleanUpIPIPAddrs()
  		}
  	}
  
  	// 30.wireGuardMgr初始化及注册
  	cryptoRouteTableWireguard := wireguard.New(config.Hostname, &config.Wireguard, 4, config.NetlinkTimeout,
  		config.DeviceRouteProtocol, func(publicKey wgtypes.Key) error {
  			if publicKey == zeroKey {
  				dp.fromDataplane <- &proto.WireguardStatusUpdate{PublicKey: "", IpVersion: 4}
  			} else {
  				dp.fromDataplane <- &proto.WireguardStatusUpdate{PublicKey: publicKey.String(), IpVersion: 4}
  			}
  			return nil
  		},
  		dp.loopSummarizer,
  		featureDetector,
  	)
  	dp.wireguardManager = newWireguardManager(cryptoRouteTableWireguard, config, 4)
  	dp.RegisterManager(dp.wireguardManager) // IPv4
  
      // 31.初始化svcLoopMgr及注册,防止无限转发
  	dp.RegisterManager(newServiceLoopManager(filterTableV4, ruleRenderer, 4))
  	...
      // IPV6相关的类同IPV4,这里不展示了
      ...
  	
      // 32.汇总iptables table
  	dp.allIptablesTables = append(dp.allIptablesTables, dp.iptablesMangleTables...)
  	dp.allIptablesTables = append(dp.allIptablesTables, dp.iptablesNATTables...)
  	dp.allIptablesTables = append(dp.allIptablesTables, dp.iptablesFilterTables...)
  	dp.allIptablesTables = append(dp.allIptablesTables, dp.iptablesRawTables...)
  	...
  
  	return dp
  }
  

• 

• 注意

  NewIntDataplaneDriver()本质上是根据不同工作模式初始化manager注册到dataplane，用于后续规则生成及同步

2.2.start

• d.Start()是felix dataplane组件启动入口，用于初始化配置，启动多个后台任务监听内核网络状态变化及维护数据面。

• func (d *InternalDataplane) Start() {
  	// 1.静态配置加载
  	d.doStaticDataplaneConfig()
  
  	// 2.dataplane主循环
  	go d.loopUpdatingDataplane()
  	...
      
      // 3.网卡监控
  	go d.ifaceMonitor.MonitorInterfaces()
      
      // 4.MTU监控
  	go d.monitorHostMTU()
  }
  

• 注意

  d.loopUpdatingDataplane()是felix核心模块，负责真正的策略生成及同步

2.3.staticcfg

• d.doStaticDataplaneConfig()会修改内核配置，修改iptables规则及BPF流量管理，维护IPIP隧道设备及周期检查。

• // sets up the kernel and our static iptables chains. Should be called
  // once at start of day before starting the main loop.
  func (d *InternalDataplane) doStaticDataplaneConfig() {
  	// 1.配置内核参数
  	d.configureKernel()
  
      // 2.BPF模式
  	if d.config.BPFEnabled {
          // 2.1.写入iptables filter规则(标记已连接流量)
          // -A FORWARD 1 -m conntrack --ctstate ESTABLISHED,RELATED -m comment -j MARK --set-xmark <Mark>/<Mask>
          // -A OUTPUT 1 -m conntrack --ctstate ESTABLISHED,RELATED -m comment -j MARK --set-xmark <Mark>/<Mask>
  		d.setUpIptablesBPFEarly()
          
          // 2.2.BPF iptable规则更新(^cali,Filter/NAT/RAW/Manage)
  		d.setUpIptablesBPF()
  	} else {
          // 2.3.普通iptables规则更新(^cali,Filter/NAT/RAW/Manage)
  		d.setUpIptablesNormal()
  	}
  
      // 3.IPIP设备维护
  	if d.config.RulesConfig.IPIPEnabled {
  		go d.ipipManager.KeepIPIPDeviceInSync(d.config.IPIPMTU, d.config.RulesConfig.IPIPTunnelAddress)
  	}
  }
  
  func (d *InternalDataplane) configureKernel() {
      // 1.加载nf_conntrack_proto_sctp模块(避免SCTP流量被标记无效)
  	mp := newModProbe(moduleConntrackSCTP, newRealCmd)
  	out, err := mp.Exec()
  
      // 2.开启IPV4/IPV6转发
  	writeProcSys("/proc/sys/net/ipv4/ip_forward", "1")
  	...
  	if d.config.IPv6Enabled {
  		writeProcSys("/proc/sys/net/ipv6/conf/all/forwarding", "1")
  		...
  	}
  
      // 3.禁止非特权用户加载BPF
  	if d.config.BPFEnabled && d.config.BPFDisableUnprivileged {
  		writeProcSys("/proc/sys/kernel/unprivileged_bpf_disabled", "1")
  		...
  	}
      
      // 4.加载WireGuard模块
  	if d.config.Wireguard.Enabled || d.config.Wireguard.EnabledV6 {
  		// wireguard module is available in linux kernel >= 5.6
  		mpwg := newModProbe(moduleWireguard, newRealCmd)
  		out, err = mpwg.Exec()
  	}
  }
  
  // KeepIPIPDeviceInSync is a goroutine that configures the IPIP tunnel device, then periodically
  // checks that it is still correctly configured.
  func (d *ipipManager) KeepIPIPDeviceInSync(mtu int, address net.IP) {
  	for {
  		err := d.configureIPIPDevice(mtu, address)
  		if err != nil {
  			time.Sleep(1 * time.Second)
  			continue
  		}
  		time.Sleep(10 * time.Second)
  	}
  }
  
  // configureIPIPDevice ensures the IPIP tunnel device is up and configures correctly.
  func (d *ipipManager) configureIPIPDevice(mtu int, address net.IP) error {
  	// 1.获取tun0设备
  	link, err := d.dataplane.LinkByName("tunl0")
  	if err != nil {
  		
  		d.dataplane.RunCmd("ip", "tunnel", "add", "tunl0", "mode", "ipip")
  		...
  		link, err = d.dataplane.LinkByName("tunl0")
  		...
  	}
  	...
  	
      // 2.MTU差异
  	if oldMTU != mtu {
  		// 设置MTU
  		d.dataplane.LinkSetMTU(link, mtu)
          ...
  	}
   
      // 3.tun0设备激活
  	if attrs.Flags&net.FlagUp == 0 {
  		d.dataplane.LinkSetUp(link)
          ...
  	}
  
      // 4.地址设置
  	d.setLinkAddressV4("tunl0", address)
      ...
      
  	return nil
  }
  

• 

• 注意

  这里集中负责的还是内核模块加载、iptables chain/rule设置及ipip tun0设备维护

2.4.monitorIface

• d.MonitorInterfaces()会监听link/route变化及周期全量同步，根据iface状态及IP变化推送事件至loopUpdatingDataplane处理。

• func (m *InterfaceMonitor) MonitorInterfaces() {
  	// Reconnection loop.
  	for {
  		...
  		{
  			...
              
              // 1.订阅iface/route变化
  			m.netlinkStub.Subscribe(updates, routeUpdates)
              ...
              
              // 2.iface/route事件拆分及有效性检查
  			go FilterUpdates(filterUpdatesCtx, filteredRouteUpdates, routeUpdates, filteredUpdates, updates)
  		}
  
  		// 3.同步已有iface/route
  		m.resync()
  		...
  
  		m.InSyncCallback()
      // 4.推送事件至ifaceUpdates
  	readLoop:
  		for {
  			select {
              // 5.iface事件处理
  			case update, ok := <-filteredUpdates:
  				if !ok {
  					break readLoop
  				}
  				m.handleNetlinkUpdate(update)
              // 6.route事件处理
  			case routeUpdate, ok := <-filteredRouteUpdates:
  				if !ok {
  					break readLoop
  				}
  				m.handleNetlinkRouteUpdate(routeUpdate)
              // 7.周期同步iface/route
  			case <-m.resyncC:
  				m.resync()
  				...
  			}
  		}
  		...
  	}
  }
  
  func (m *InterfaceMonitor) resync() error {
  	// 1.获取节点iface
  	links, err := m.netlinkStub.LinkList()
  	...
  	currentIfaces := set.New[string]()
      
      // 2.更新iface缓存(基于route获取网卡地址),推送事件至ifaceUpdates
  	for _, link := range links {
  		attrs := link.Attrs()
  		if attrs == nil {
  			continue
  		}
  		currentIfaces.Add(attrs.Name)
  		m.storeAndNotifyLink(true, link)
  	}
      
      // 3.iface过期处理
  	for ifIndex, info := range m.ifaceIdxToInfo {
  		name := info.Name
  		if currentIfaces.Contains(name) {
  			continue
  		}
  
          // 推送事件至ifaceUpdates
  		m.StateCallback(name, StateNotPresent, ifIndex)
  		if info.TrackAddrs {
  			// We were tracking addresses for this interface before but now it's gone.  Signal that.
  			m.AddrCallback(name, nil)
  		}
          
          // 清理缓存
  		delete(m.ifaceNameToIdx, name)
  		delete(m.ifaceIdxToInfo, ifIndex)
  	}
  
  	return nil
  }
  

• 

• 注意

  resync()周期同步的iface规则及nelink.subscribe()监听的节点网卡规则会推到channel供主循环消费以更新流量管理策略

2.3.monitormtu

• d.monitorHostMTU()用于定期(30s)检测和监控Iface MTU，MTU发生变化会触发配置变更回调，通知系统优雅重启以适应新的网络环境。

• func (d *InternalDataplane) monitorHostMTU() {
  	for {
          // 1.获取Iface MTU
  		mtu, err := findHostMTU(d.config.MTUIfacePattern)
  		...
          
          // 2.MTU变化
          if d.config.hostMTU != mtu {
              // 发送终止信号优雅退出(30s)
  			d.config.ConfigChangedRestartCallback()
  		}
  		time.Sleep(30 * time.Second)
  	}
  }
  
  // findHostMTU auto-detects the smallest host interface MTU.
  func findHostMTU(matchRegex *regexp.Regexp) (int, error) {
  	// 1.获取iface
  	links, err := netlink.LinkList()
  	...
  
  	// 2.匹配的iface对应的最小MTU
  	for _, l := range links {
  		// Skip links that we know are not external interfaces.
  		fields := log.Fields{"mtu": l.Attrs().MTU, "name": l.Attrs().Name}
  		if matchRegex == nil || !matchRegex.MatchString(l.Attrs().Name) {
  			continue
  		}
  		if l.Attrs().MTU < smallest || smallest == 0 {
  			smallest = l.Attrs().MTU
  		}
  	}
  
      // 3.1460作为MTU默认值
  	if smallest == 0 {
  		// We failed to find a usable interface. Default the MTU of the host
  		// to 1460 - the smallest among common cloud providers.
  		return 1460, nil
  	}
  	return smallest, nil
  }
  

• 

• 补充

  这里逻辑相对清晰，不再赘述，下一章主要分析felix主循环d.loopUpdatingDataplane()