eBPFEndpoint

1.入口

1.1.manager

• eBPFEndPointManager对象会作为cailico基于BPF接管内核栈流量的代理，后续网卡对BPF程序的挂载及相关数据的存取会基于代理对象下发。

• func newBPFEndpointManager(dp bpfDataplane, config *Config,...) (*bpfEndpointManager, error) {
      // liveness回调(这里是HealthAggregator)
  	if livenessCallback == nil {
  		livenessCallback = func() {}
  	}
      
  	m := &bpfEndpointManager{
  		...
  		ipSetIDAlloc:            ipSetIDAlloc,
  		...
          // ipsets.Map-->cali_v4_ip_sets    		state.Map-->cali_state
          // arp.Map-->cali_v4_arp				counters.Map-->cali_counters
          // failsafes.Map-->cali_v4_fsafes   	nat.FrontendMap-->cali_v4_nat_fe
          // nat.BackendMap-->cali_v4_nat_be		nat.AffinityMap-->cali_v4_nat_aff
          // routes.Map-->cali_v4_routes			conntrack.Map-->cali_v4_ct
          // nat.SendRecvMsgMap-->cali_v4_srmsg	nat.AllNATsMsgMap-->cali_v4_ct_nats
          // ifstate.Map-->cali_iface				counters.PolicyMap-->cali_rule_ctrs
  		bpfmaps:                 bpfmaps,
          // 这里会对ifStateMap存取进行封装,避免直接操作byte
  		ifStateMap: cachingmap.New[ifstate.Key, ifstate.Value](ifstate.MapParams.Name,
  			maps.NewTypedMap[ifstate.Key, ifstate.Value](
  				bpfmaps.IfStateMap.(maps.MapWithExistsCheck), ifstate.KeyFromBytes, ifstate.ValueFromBytes,
  			)),
  		ruleRenderer:        iptablesRuleRenderer, 					// ruleRenderer
  		iptablesFilterTable: iptablesFilterTable,					// iptables filter table
  		onStillAlive:        livenessCallback,						// HealthAggregator
  		...
  		dirtyRules:            set.New[polprog.RuleMatchID](),
  		arpMap:                bpfmaps.ArpMap,						// arp bpf map相关参数
  	}
  
  	m.xdpModes = []bpf.XDPMode{ bpf.XDPOffload, bpf.XDPDriver, bpf.XDPGeneric }
  
  	// Clean all the files under /var/run/calico/bpf/prog to remove any information from the
  	// previous execution of the bpf dataplane, and make sure the directory exists.
  	bpf.CleanAttachedProgDir()
  
  	// Normally this endpoint manager uses its own dataplane implementation
  	if m.dp == nil {
  		m.dp = m
  	}
  
      // BPF Map GC runner,间隔10s清理孤儿Jump Map和空目录
      // 防止接口异常删除后pin文件残留占用BPF资源
  	m.mapCleanupRunner = ratelimited.NewRunner(10s, func(ctx context.Context) {
  		// So that we serialise with AttachProgram()
  		...
  		bpf.CleanUpMaps()
  	})
  
      // CTLB兼容处理,部分内核对CTLB Hook存在兼容问题,此时需回退到Workaround模式,基于一对特殊veth pair处理NAT
  	if config.FeatureGates != nil {
  		switch config.FeatureGates["BPFConnectTimeLoadBalancingWorkaround"] {
  		case "enabled":
  			m.ctlbWorkaroundMode = ctlbWorkaroundEnabled	// tcp+udp流量均启用
  		case "udp":
  			m.ctlbWorkaroundMode = ctlbWorkaroundUDPOnly	// udp流量启用
  		}
  	}
  
      // CTLB Workaround
  	if m.ctlbWorkaroundMode != ctlbWorkaroundDisabled {
          // 初始化routeTable,绑定bpfInDev设备,监听路由变化
          // removeExternalRoutes=true: 同步时自动清理非Calico管理的路由,见routetable分析
          
  		m.routeTable = routetable.New([]string{bpfInDev}, 4, false, // vxlan
  			config.NetlinkTimeout, nil, // deviceRouteSourceAddress
  			config.DeviceRouteProtocol, true, // removeExternalRoutes
  			unix.RT_TABLE_MAIN, opReporter, featureDetector)
          // service cidr缓存
  		m.services = make(map[serviceKey][]ip.V4CIDR)
          // 标记待同步service
  		m.dirtyServices = set.New[serviceKey]()
  
          // 关闭rp_fiter(calico eBPF自己检查rp_filter)
          // rp_filter作为反向路径过滤,基于路由限制报文的出入口是否符合预期
          // CTLB模式流量的入网卡和出网卡不一致,rp_filter会丢弃流量
          // 0:不开启	1:报文出入口网卡必须一致	2:基于路由决定出口网卡
          // /proc/sys/net/ipv4/conf/%s/rp_filter设为0
  		m.dp.setRPFilter("all", 0)
          ...
  
          // 创建bpfin.cali/bpfout.cali veth对、ARP邻居和qdisc...
  		m.dp.ensureBPFDevices()
          ...
  	}
  	...
  
  	return m, nil
  }
  
  // 清理/var/run/calico/bpf/prog/下不存在网卡的json元数据
  func CleanAttachedProgDir() {
      // 生成/var/run/calico/bpf/prog目录
  	os.MkdirAll(RuntimeProgDir, 0600)
      ...
  
      // 获取所有iface
  	interfaces, err := net.Interfaces()
  	...
  
  	expectedJSONFiles := set.New[string]()
  	for _, iface := range interfaces {
          // iface网卡BPF挂载点
  		for _, hook := range Hooks {
              // 为每个网卡的ingress/egress/xdp三个hook点生成期望的json文件名
  			expectedJSONFiles.Add(RuntimeJSONFilename(iface.Name, hook))
  		}
  	}
  
      // 清理不匹配的json文件(已删除网卡的残留元数据)
  	filepath.Walk(RuntimeProgDir, func(p string, info os.FileInfo, err error) error {
  		...
  		if !expectedJSONFiles.Contains(p) {
  			os.Remove(p)
  			...
  		}
  
  		return nil
  	})
  	...
  }
  

• 

• 注意

  bpfEndpointMgr初始化会清理一次/var/run/calico/bpf/prog/的过期元数据，后续基于10s间隔的mapCleanupRunner清理

1.2.cleanmap

• CleanUpMaps()会扫描/sys/fs/bpf/tc的jumpMap文件，网卡挂载的BPF程序及内核加载的BPF程序，清理网卡挂载未使用的文件或空目录。

• // scans for cali_jump maps that are still pinned to filesystem but no longer referenced by our BPF programs.
  func CleanUpMaps() {
  	// 扫描/sys/fs/bpf/tc,获取BPF jumpMap的mapID-->jumpMap路径
  	mapIDToPath, err := ListPerEPMaps()
  	...
  
      // 获取网卡attach的BPF程序(TC/XDP)
  	aTc, aXdp, err := ListTcXDPAttachedProgs()
  	...
  	attachedProgs := set.New[int]()
  	for _, prog := range aTc {
  		attachedProgs.Add(prog.ID)
  	}
  	for _, prog := range aXdp {
  		attachedProgs.Add(prog.ID)
  	}
  
      // bpftool prog list --json会列出内核已加载的BPF程序
      // {
      //   "id": 1024,
      //   "name": "calico_tc_ingress",
      //   "map_ids": [50,51,68]
      // }
  	progsJSON, err := exec.Command("bpftool", "prog", "list", "--json").Output()
  	...
  	var progs []struct {
  		ID   int    `json:"id"`
  		Name string `json:"name"`
  		Maps []int  `json:"map_ids"`
  	}
  	json.Unmarshal(progsJSON, &progs)
  	...
      
      // 遍历内核加载BPF程序
  	for _, p := range progs {
          // 未挂载的程序可能detach但map仍被引用的中间态
  		if !attachedProgs.Contains(p.ID) {
  			continue
  		}
          
          // 走到这里,p.ID是正在网卡运行的BPF程序,正在使用的map不能删,剔除掉
  		for _, id := range p.Maps {
  			delete(mapIDToPath, id)
  		}
  	}
  
  	// 走到这里只剩下孤儿jumpMap,清理对应pin文件
  	for id, p := range mapIDToPath {
  		os.Remove(p)
  		...
  	}
  	...
  
      // 再次扫描/sys/fs/bpf/tc
      // 先假设所有calico管理的目录都是空的,遇到文件时将其父目录标记为"非空"
  	err = filepath.Walk("/sys/fs/bpf/tc", func(p string, info os.FileInfo, err error) error {
  		...
          
          // calico管理的目录
  		if info.IsDir() && pinDirRegex.MatchString(info.Name()) {
              // 暂认为是空目录
  			p := path.Clean(p)
  			emptyAutoDirs.Add(p)
          // 非calico管理目录或文件类型
  		} else {
              // 利用文件修正空目录
  			dirPath := path.Clean(path.Dir(p))
  			if emptyAutoDirs.Contains(dirPath) {
  				emptyAutoDirs.Discard(dirPath)
  			}
  		}
  		return nil
  	})
  	...
  
      // /sys/fs/bpf/tc将空目录清理掉
  	emptyAutoDirs.Iter(func(p string) error {
  		os.Remove(p)
  		...
  		return nil
  	})
  }
  

• ListPerEPMaps()扫描/sys/fs/bpf/tc目录，匹配jumpMap内核文件句柄，执行bpftool map show pinned命令获取jumpMap表元数据。

• // returns all pod jumpMap,with mapID to jumpMap filepath.
  func ListPerEPMaps() (map[int]string, error) {
  	mapIDToPath := make(map[int]string)
      // 扫描/sys/fs/bpf/tc已pinned的BPF map,梳理endpoint独立的jumpMap跳转表,构建mapID-->bpffs路径映射
  	err := filepath.Walk("/sys/fs/bpf/tc", func(p string, info os.FileInfo, err error) error {
  		...
          // 全局共享map跳过
  		if strings.Contains(p, "globals") {
  			return nil
  		}
          // 前缀为cali_jump3的文件,这是jumpMap的内核句柄
  		if strings.HasPrefix(info.Name(), maps.JumpMapName()) {
              // # bpftool map show pinned xxx获取map结构
              // 32: array  name cali_jump3_veth7ae2f3
              //     key 4B  value 8B  max_entries 64  flags 0x0
              
              // # bpftool map dump pinned xxx获取map内容
              // key: 00 00 00 00  value: 30 81 01 00 00 00 00 00  // idx0 → Policy校验函数地址
              // key: 01 00 00 00  value: 78 82 01 00 00 00 00 00  // idx1 → DNAT转发函数地址
              // key: 02 00 00 00  value: b0 82 01 00 00 00 00 00  // idx2 → SNAT转换函数地址
              // key: 03 00 00 00  value: 00 00 00 00 00 00 00 00  // idx3 → 地址0，模块未启用，不跳转
              // key: 04 00 00 00  value: 28 83 01 00 00 00 00 00  // idx4 → Metrics指标采集函数地址
  			out, err := exec.Command("bpftool", "map", "show", "pinned", p).Output()
  			...
              
              // 取BPF mapID
  			idStr := string(bytes.Split(out, []byte(":"))[0])
  			id, err := strconv.Atoi(idStr)
  			...
              // mapID-->junpMap表路径
  			mapIDToPath[id] = p
  		}
  		return nil
  	})
  
  	return mapIDToPath, err
  }
  

• ListTcXDPAttachedProgs()会执行bpftool net -j列出网卡挂载的BPF，内容包括tc/xdp挂载到的网卡、Hook及BPF程序内核ID。

• // returns all programs attached to TC or XDP hooks.
  func ListTcXDPAttachedProgs() (TcList, XDPList, error) {
  	// Find all the programs that are attached to interfaces.
      // bpftool net -j会输出所有网卡attach的BPF程序(TC/XDP...)
      // [
      //   {
      //     "tc": [
      //       {"ifname":"vethxxx","hook":"ingress","prog_id":123,...},
      //       {"ifname":"vethxxx","hook":"egress","prog_id":456,...}
      //     ],
      //     "xdp": [
      //       {"ifname":"eth0","mode":"native","prog_id":789,...}
      //     ]
      //   }
      // ]
  	out, err := exec.Command("bpftool", "net", "-j").Output()
  	...
  
  	var attached []struct {
  		TC  TcList  `json:"tc"`
  		XDP XDPList `json:"xdp"`
  	}
  
  	json.Unmarshal(out, &attached)
  	...
  
  	return attached[0].TC, attached[0].XDP, nil
  }
  

• 

• 清理

  这套逻辑基于间隔10s的timer触发，内核加载和网卡运行的BPF程序取交集，将正在使用的jumpMap文件去除，清理剩余的jumpMap文件

1.3.device

• ensureBPFDevices()会创建CTLB降级的veth pair设备及相关的ARP邻居和路由，设备上会检查及创建tc clsact供后续TC BPF挂载，

• func (m *bpfEndpointManager) ensureBPFDevices() error {
      // CTLB禁用时不创建
  	if m.ctlbWorkaroundMode == ctlbWorkaroundDisabled {
  		return nil
  	}
      ...
  
      // 获取bpfInDev设备
  	bpfin, err := netlink.(bpfInDev)
  	if err != nil {
          // 没有的话创建一对veth pair: bpfin.cali <-> bpfout.cali
  		la := netlink.NewLinkAttrs()
  		la.Name = bpfInDev
  		nat := &netlink.Veth{
  			LinkAttrs: la,
  			PeerName:  bpfOutDev,
  		}
  		netlink.LinkAdd(nat)
          ...
          
  		bpfin, err = netlink.LinkByName(bpfInDev)
  		...
  	}
      
      // 启用bpfInDev
  	if state := bpfin.Attrs().OperState; state != netlink.OperUp {
  		netlink.LinkSetUp(bpfin)
          ...
  	}
      
      // 启用bpfOutDev
  	bpfout, err = netlink.LinkByName(bpfOutDev)
  	...
  	if state := bpfout.Attrs().OperState; state != netlink.OperUp {
  		netlink.LinkSetUp(bpfout)
          ...
  	}
  
  	m.natInIdx = bpfin.Attrs().Index
  	m.natOutIdx = bpfout.Attrs().Index
  
      // 更新BPF ArpMap,bpfInDev--->bpfOutDev,这是给BPF用的
      // 用于BPF程序在bpf_redirect时重写L2头
      // key:
      //   IPv4 = 0.0.0.0
      //   ifindex = bpfin index
      // value:
      //   src MAC = bpfin MAC
      //   dst MAC = bpfout MAC
  	anyV4, _ := ip.CIDRFromString("0.0.0.0/0")
  	_ = m.arpMap.Update(
  		bpfarp.NewKey(anyV4.Addr().AsNetIP(), uint32(m.natInIdx)).AsBytes(),
  		bpfarp.NewValue(bpfin.Attrs().HardwareAddr, bpfout.Attrs().HardwareAddr).AsBytes(),
  	)
  
      // ip neigh add 169.254.1.1 lladdr <bpfout-mac> dev bpfin permanent  永久arp邻居用于将流量转到bpfou 
  	// bpfInDev和bpfOutDev是内部虚拟设备,这里就是告知内核bpfin发包给169.254.1.1的流量直接使用bpfOutDev的mac
  	arp := &netlink.Neigh{
  		State:        netlink.NUD_PERMANENT,
  		IP:           net.IPv4(169, 254, 1, 1),
  		HardwareAddr: bpfout.Attrs().HardwareAddr,
  		LinkIndex:    bpfin.Attrs().Index,
  	}
  	netlink.NeighAdd(arp)
      ...
  
      // bpfInDev和bpfOutDev设备相关参数设备
      // set /proc/sys/net/ipv4/conf/%s/route_localnet=1  支持处理localhost流量
      // set /proc/sys/net/ipv4/neigh/%s/proxy_delay=0	关闭ARP延迟
      // set /proc/sys/net/ipv4/conf/%s/proxy_arp=1 		ARP代理,host回答ARP请求
      // set /proc/sys/net/ipv4/conf/%s/forwarding=1		允许内核转发经过接口的IPV4流量
      // set /proc/sys/net/ipv4/conf/%s/rp_filter=0		关闭rp_filter
  	configureInterface(bpfInDev, 4, "0", writeProcSys)
      ...
  	configureInterface(bpfOutDev, 4, "0", writeProcSys)
      ...
  
      // bpfin网卡创建tc clsact,这个qdisc后续不会发挥作用,bpfin也不会挂TC
  	m.ensureQdisc(bpfInDev)
  	...
      // lo网卡创建tc clsact(TC依赖)
  	m.ensureQdisc("lo")
  	...
  
  	// ip route add 169.254.1.1/32 dev bpfin
      // 目标为169.254.1.1的报文直接从bpfin发出,配合routeTable将Service CIDR路由到169.254.1.1,统一走bpfin管道
  	cidr, _ := ip.CIDRFromString("169.254.1.1/32")
  	m.routeTable.RouteUpdate(bpfInDev, routetable.Target{
  		Type: routetable.TargetTypeLinkLocalUnicast,
  		CIDR: cidr,
  	})
  
  	return nil
  }
  
  func (m *bpfEndpointManager) ensureQdisc(iface string) error {
  	return tc.EnsureQdisc(iface)
  }
  
  // makes sure that qdisc is attached to the given interface
  func EnsureQdisc(ifaceName string) error {
      // 检查网卡有没有clsact
      // qdisc show dev bpfin clsact
  	hasQdisc, err := HasQdisc(ifaceName)
  	...
  	if hasQdisc {
  		return nil
  	}
      
      // qdisc add dev bpfin clsact
  	return libbpf.CreateQDisc(ifaceName)
  }
  

• 

• 补充

  1.CTLB本质上用于处理主机访问的service流量，connect阶段实现提前NAT，绕过TC BPF处理

  2.eBPF模式下，service流量解析依赖网卡挂载的TC程序，CTLB未生效就必须将service流量经过网卡TC处理

  3.bpfin/bpfout就是兼容处理的特殊网卡对，路由将主机service流量引入bpfin-bpfout端，利用bpfout网卡上的TC Hook实现NAT

2.事件驱动

2.1.推送

• dataplane章节提到过，calGraph和ifaceMonitor计算的事件会触发推送，执行mgr.OnUpdate()将路由或网络接口变化传递给下游。

• func (d *InternalDataplane) loopUpdatingDataplane() {
  	...
  
  	for {
  		select {
          // calGraph计算的事件
  		case msg := <-d.toDataplane:
              // 更新缓存,标记同步
  			d.onDatastoreMessage(msg)
          // ifaceMonitor监听的网卡事件
  		case ifaceUpdate := <-d.ifaceUpdates:
              // 更新缓存,标记同步
  			d.onIfaceMonitorMessage(ifaceUpdate)
  		...
  		}
  		...
  	}
  }
  
  // called when get message from calGraph it opportunistically processes a match of msg from its channel.
  func (d *InternalDataplane) onDatastoreMessage(msg interface{}) {
  	...
  
      // 执行注册的所有manager的OnUpdate回调
  	d.processMsgFromCalcGraph(msg)
      
      // 批量drain,最多一次处理100条消息
  	drainChan(d.toDataplane, d.processMsgFromCalcGraph)
  	...
  }
  
  // called when we get message from interface monitor it opportunistically processes match msg from its channel.
  func (d *InternalDataplane) onIfaceMonitorMessage(ifaceUpdate any) {
  	...
  
  	// 执行注册的所有manager的OnUpdate回调
  	d.processIfaceUpdate(ifaceUpdate)
      
      // 批量drain,最多一次处理100条消息
  	drainChan(d.ifaceUpdates, d.processIfaceUpdate)
  
      // 网卡事件额外标记
  	d.dataplaneNeedsSync = true
  	...
  }
  

• 

• 补充

  calGraph基于APIServer对象变化计算事件，ifaceMonitor监听主机网卡生成事件，两者驱动manager更新内存dirty区

2.2.分发

• m.OnUpdate()根据消息类型分发到具体处理函数，更新内存索引及标记接口dirty，这里更新的都是内存的dirty区，后续会基于Apply执行。

• // 来自calGraph监听对象变化计算的事件和ifaceMonitor监听主机网卡的事件
  func (m *bpfEndpointManager) OnUpdate(msg interface{}) {
  	switch msg := msg.(type) {
  	// 主机网卡变化
  	case *ifaceStateUpdate:
  		m.onInterfaceUpdate(msg)
  	case *ifaceAddrsUpdate:
  		m.onInterfaceAddrsUpdate(msg)
  
  	// pod网卡变化
  	case *proto.WorkloadEndpointUpdate:
  		m.onWorkloadEndpointUpdate(msg)
  	case *proto.WorkloadEndpointRemove:
  		m.onWorkloadEnpdointRemove(msg)
  
  	// Policies策略对象变化
  	case *proto.ActivePolicyUpdate:
  		m.onPolicyUpdate(msg)
  	case *proto.ActivePolicyRemove:
  		m.onPolicyRemove(msg)
  
      // Profiles策略对象变化(SA/NS级别)
  	case *proto.ActiveProfileUpdate:
  		m.onProfileUpdate(msg)
  	case *proto.ActiveProfileRemove:
  		m.onProfileRemove(msg)
  
      // 主机IP变化
  	case *proto.HostMetadataUpdate:
  		if msg.Hostname == m.hostname {
  			ip := net.ParseIP(msg.Ipv4Addr)
  			if ip != nil {
  				m.hostIP = ip
  				...
                  // HostIP变化会影响Global Map和TC NAT
                  // 所有接口的BPF程序都需要重新加载
  				for ifaceName := range m.nameToIface {
  					m.dirtyIfaceNames.Add(ifaceName)
  				}
  				...
  			}
  		}
  
      // service变化(NAT路由)
  	case *proto.ServiceUpdate:
  		m.onServiceUpdate(msg)
  	case *proto.ServiceRemove:
  		m.onServiceRemove(msg)
  	case *proto.RouteUpdate:
  		m.onRouteUpdate(msg)
  	}
  }
  

• 

• 补充

  1.Policy对象：对应NetworkPolicy，用于设置网络策略及隔离规则，类似设置白黑名单

  2.Profile对象：Calico的策略对象，基于Namespace+SA设置隐式的访问规则

3.网卡处理

3.1.onIfaceAddr

• m.onInterfaceAddrsUpdate()将主机网卡IPv4地址变化更新到ifaceToIpMap缓存及标记网卡dirty，网卡地址变化会影响TC程序全局数据。

• func (m *bpfEndpointManager) onInterfaceAddrsUpdate(update *ifaceAddrsUpdate) {
  	...
  
      // addr变化
  	if update.Addrs != nil && update.Addrs.Len() > 0 {
          // 匹配一下IPV4地址
  		update.Addrs.Iter(func(item string) error {
  			ip := net.ParseIP(item)
  			if ip.To4() != nil {
  				ipAddrs = append(ipAddrs, ip)
  			}
  			return nil
  		})
  		sort.Slice(ipAddrs, func(i, j int) bool { return bytes.Compare(ipAddrs[i], ipAddrs[j]) < 0 })
  		if len(ipAddrs) > 0 {
              // 更新iface--->IP映射,排序后的第一个作为网卡主IP
  			ip, ok := m.ifaceToIpMap[update.Name]
  			if !ok || !ip.Equal(ipAddrs[0]) {
  				m.ifaceToIpMap[update.Name] = ipAddrs[0]
                  // 标记dirty,Apply阶段处理
  				m.dirtyIfaceNames.Add(update.Name)
  			}
  
  		}
      // addr删除
  	} else {
  		_, ok := m.ifaceToIpMap[update.Name]
          // 清理映射
  		if ok {
  			delete(m.ifaceToIpMap, update.Name)
              // 加入暂存区
  			m.dirtyIfaceNames.Add(update.Name)
  		}
  	}
  }
  

3.2.onIfaceUpdate

• m.onInterfaceUpdate()根据主机网卡的UP/DOWN状态清理的旧的BPF挂载及更新接口的BPF Map，内存侧的iface缓存也会做相应对齐。

• func (m *bpfEndpointManager) onInterfaceUpdate(update *ifaceStateUpdate) {
  	...
  
      // 网卡被删除了
  	if update.State == ifacemonitor.StateNotPresent {
          // 清理/var/run/calico/bpf/prog/iface_{xdp,ingress,engress}.json文件,这是用户态维护的挂载元数据
  		bpf.ForgetIfaceAttachedProg(update.Name)
          ...
  	}
  
      // 仅处理数据接口(eth0/bpfout/lo...)、Pod网卡或L3隧道网卡(tun0/wireguard.cali...)
  	if !m.isDataIface(update.Name) && !m.isWorkloadIface(update.Name) && !m.isL3Iface(update.Name) {
          // UP状态未直接口
  		if update.State == ifacemonitor.StateUp {
              // 挂载过BPF,清理一下挂载及用户侧元数据
  			if ai, ok := m.initAttaches[update.Name]; ok {
  				m.cleanupOldAttach(update.Name, ai)
                  ...
                  delete(m.initAttaches, update.Name)
  			}
  		}
          
          // 记录到未知网卡集合
  		if m.initUnknownIfaces != nil {
  			m.initUnknownIfaces.Add(update.Name)
  		}
  		return
  	}
  
      // withIface封装了iface状态更新+dirty标记逻辑
  	m.withIface(update.Name, func(iface *bpfInterface) (forceDirty bool) {
  		ifaceIsUp := update.State == ifacemonitor.StateUp
  		// UP状态
  		if ifaceIsUp {
              // 清理下挂载状态,后续要重新挂
  			delete(m.initAttaches, update.Name)
  			switch update.Name {
              // CTLB专用网卡不修改rp_filter,已经在创建阶段配置过
  			case bpfInDev, bpfOutDev:
  				// do nothing
              // 主机网卡
  			default:
                  // set /proc/sys/net/ipv4/conf/<iface>/rp_filter=2
                  // 允许Service流量经BPF DNAT后从不同网卡返回(DSR流量)
  				m.dp.setRPFilter(update.Name, 2)
                  ...
  			}
              
              // set /proc/sys/net/ipv4/conf/<iface>/accept_local=1,允许接收本机地址流量(Pod访问Svc后DNAT回本节点)
  			_ = m.dp.setAcceptLocal(update.Name, true)
              // 支持HEP统配,将未绑定HEP的主网卡关联到到通配hostEndpoint
  			if _, hostEpConfigured := m.hostIfaceToEpMap[update.Name]; m.wildcardExists && !hostEpConfigured {
  				m.addHEPToIndexes(update.Name, &m.wildcardHostEndpoint)
  				m.hostIfaceToEpMap[update.Name] = m.wildcardHostEndpoint
  			}
              
              // 更新一下iface的index和UP状态
  			iface.info.ifIndex = update.Index
  			iface.info.isUP = true
              // 更新一下ifaceState BPFMap
  			m.updateIfaceStateMap(update.Name, iface)
          // DOWN状态
  		} else {
              // 支持HEP统配,,清理绑定通配HEP的主网卡映射
  			if m.wildcardExists && reflect.DeepEqual(m.hostIfaceToEpMap[update.Name], m.wildcardHostEndpoint) {
  				m.removeHEPFromIndexes(update.Name, &m.wildcardHostEndpoint)
  				delete(m.hostIfaceToEpMap, update.Name)
  			}
              
              // 清理网卡xdp/ingress/egress的规则命中计数器(counters.PolicyMap),这里直接交互内核执行删除
  			m.deleteIfaceCounters(update.Name, iface.info.ifIndex)
              // 更新一下网卡状态
  			iface.dpState.isReady = false
  			iface.info.isUP = false
              // 更新一下ifaceState BPFMap
  			m.updateIfaceStateMap(update.Name, iface)
  			iface.info.ifIndex = 0
  		}
          
  		return true // Force interface to be marked dirty in case we missed a transition during a resync.
  	})
  }
  
  func (m *bpfEndpointManager) withIface(ifaceName string, fn func(iface *bpfInterface) (forceDirty bool)) {
  	iface := m.nameToIface[ifaceName]
  	ifaceCopy := iface
  	dirty := fn(&iface)
  	...
      
      // 网卡下线
  	if reflect.DeepEqual(iface, zeroIface) {
          // 清理name--->iface映射
  		delete(m.nameToIface, ifaceName)
  	} else {
  		// 更新name-->iface映射
  		m.nameToIface[ifaceName] = iface
  	}
  
      // 状态变化或forceDirty,标记dirty
  	dirty = dirty || iface.info != ifaceCopy.info
  	if !dirty {
  		return
  	}
  	m.dirtyIfaceNames.Add(ifaceName)
  }
  

• 

• BPF挂载

  1.内核侧：挂载的BPF程序可以基于bpftool net show查询，用到的BPF Map规则文件会Pin到/sys/fs/bpf持久化

  2.用户侧：元数据会存放在/var/run/calico/bpf/prog/iface_{xdp,ingress,engress}.json，卸载BPF程序会用到

3.3.cleanupAttach

• m.cleanupOldAttach()检测到三无网卡执行BPF程序卸载，xdp/tc实现独立的卸载逻辑，同步阶段的三无产品及Apply阶段场景也会调用。

• func (m *bpfEndpointManager) cleanupOldAttach(iface string, ai bpf.EPAttachInfo) error {
  	if ai.XDPId != 0 {
  		ap := xdp.AttachPoint{
  			Iface: iface,
  			// Try all modes in this order
  			Modes: []bpf.XDPMode{bpf.XDPGeneric, bpf.XDPDriver, bpf.XDPOffload},
  		}
  
          // 卸载XDP程序
  		m.dp.ensureNoProgram(&ap)
          ...
  	}
  	if ai.TCId != 0 {
  		ap := tc.AttachPoint{
  			Iface: iface,
  			Hook:  bpf.HookEgress,
  		}
  
          // 卸载TCEngress程序
  		m.dp.ensureNoProgram(&ap)
          ...
  
  		ap.Hook = bpf.HookIngress
  
          // 卸载TCIngress程序
  		m.dp.ensureNoProgram(&ap)
          ...
  	}
  
  	return nil
  }
  
  // Ensure that the specified attach point does not have our program.
  func (m *bpfEndpointManager) ensureNoProgram(ap attachPoint) error {
  
  	// 获取对应的jumpMap文件句柄
  	jumpMapFD := m.getJumpMapFD(ap)
  	if jumpMapFD != 0 {
  		// 尝试关闭文件句柄
  		if err := jumpMapFD.Close(); err == nil {
              // 清理缓存网卡关联的jumpMap文件句柄
  			m.setJumpMapFD(ap, 0)
  		}
          ...
  	}
  
  	// 卸载网卡挂载的BPF程序
  	err := ap.DetachProgram()
  	...
  
  	return err
  }
  

• 

• 注意

  这里的jumpMapFD需要关注，TC程序执行策略检查会调用tail call jump跳转到动态编译的Prog Hook，这个Prog Hook后面会介绍

3.4.detachProgram

• ap.DetachProgram()由xdp/tc独立实现，本质是执行cgo程序交互内核接口，循环执行清理命令，另外还会清理用户侧维护的挂载元数据。

• // xdp卸载实现
  func (ap *AttachPoint) DetachProgram() error {
  	// 执行C.bpf_xdp_program_id交互内核接口获取xdp程序ID
  	progID, err := ap.ProgramID()
  	...
      
      // 网卡未挂载xdp
  	if progID == DetachedID {
  		return nil
  	}
  
      // 检查网卡挂载XDP程序条件:
      // 1.读取/var/run/calico/bpf/prog/iface_hook.json文件内容及解析到data
      // 2.计算/usr/lib/calico/bpf/xdp_level.o文件路径对应hash值
      // 3.data.hash=hash&程序路径一致(/usr/lib/calico/bpf/xdp_level.o)&data.ID=progID&data.config=ap内容
  	ourProg, err := bpf.AlreadyAttachedProg(ap, path.Join(bpf.ObjectDir, ap.FileName()), progID)
  	if !ourProg {
  		return fmt.Errorf("XDP expected program ID does match with current one: %w", err)
  	}
      ...
  
  	// 根据模式优先级尝试detach
  	for _, mode := range ap.Modes {
          // 执行C.bpf_xdp_detach交互内核卸载网卡xdp程序
  		libbpf.DetachXDP(ap.Iface, uint(mode))
  		...
          
          // 执行C.bpf_xdp_program_id再次获取网卡挂载xdp程序ID
  		curProgId, err := ap.ProgramID()
  		...
  
          // ID=0代表卸载完成
  		if curProgId == DetachedID {
  			removalSucceeded = true
  			break
  		}
  	}
  	...
  
  	// 清理/var/run/calico/bpf/prog/iface_xdp.json文件,这是用户态维护的挂载元数据
  	bpf.ForgetAttachedProg(ap.IfaceName(), bpf.HookXDP)
      ...
      
  	return nil
  }
  
  // tc卸载实现
  func (ap *AttachPoint) DetachProgram() error {
      // tc filter show dev <iface> <ingress|egress> 列出所有BPF filter
      // 正则取出优先级和句柄handle
  	progsToClean, err := ap.listAttachedPrograms()
  	...
  
  	return ap.detachPrograms(progsToClean)
  }
  
  func (ap *AttachPoint) detachPrograms(progsToClean []attachedProg) error {
  	...
      
  	for _, p := range progsToClean {
  		attemptCleanup := func() error {
              // tc filter del dev <iface> <hook> pref <pref> handle <handle> bpf
  			err := ExecTC("filter", "del", "dev", ap.Iface, ap.Hook.String(), "pref", p.pref, "handle", p.handle, "bpf")
  			return err
  		}
          
  		attemptCleanup()
  		...
  	}
  	...
  
  	return nil
  }
  

• 

• xdp/tc卸载程序都是执行cgo函数，基于iface元数据通知内核卸载，挂载其实也是一样的

4.hostendpoint

4.1.endpointMgr

• m.OnHEPUpdate()由endpointManager计算hostEndpoint变化触发，BPF模式开启endPointManager仅计算，不生成iptables策略。

• func (d *InternalDataplane) apply() {
  	...
      
  	for _, mgr := range d.allManagers {
  		if handler, ok := mgr.(UpdateBatchResolver); ok {
              // endpointManager此阶段计算host→HEP映射
  			handler.ResolveUpdateBatch()
  			...
  			d.reportHealth()
  		}
  	}
  	...
  }
  
  // endpointManager负责解析主机网卡与HostEndpoint的对应关系
  func (m *endpointManager) ResolveUpdateBatch() error {
  	...
  
      // ifaceMonitor注意到主机网卡变化就会触发这里
  	if m.hostEndpointsDirty {
  		m.newIfaceNameToHostEpID = m.resolveHostEndpoints()
  	}
  
  	return nil
  }
  
  func (m *endpointManager) resolveHostEndpoints() map[string]proto.HostEndpointID {
  	...
  
      // BPF模式下endpointMgr不生成iptables,仅计算映射通知bpfEndpointManager
  	if m.bpfEndpointManager != nil {
  		hostIfaceToEpMap := map[string]proto.HostEndpoint{}
  		for ifaceName, id := range newIfaceNameToHostEpID {
  			hostIfaceToEpMap[ifaceName] = *m.rawHostEndpoints[id]
  		}
          // 通知
  		m.bpfEndpointManager.OnHEPUpdate(hostIfaceToEpMap)
  	}
  
  	return newIfaceNameToHostEpID
  }
  

• 不太清楚HEP为什么独立及endpointManager/eBPFEndpointManager共存，可能有历史原因，其实OnUpdate是可以拿到HEP事件自己计算

4.2.onHEPUpdate

• m.OnHEPUpdate()根据上游推送的事件更新hostIfaceToEpMap和policiesToWorkloads/profilesToWorkloads索引，标记网卡dirty。

• func (m *bpfEndpointManager) OnHEPUpdate(hostIfaceToEpMap map[string]proto.HostEndpoint) {
  	if m == nil {
  		return
  	}
  
  	// wildcard HEP(host-*): 匹配所有未显式绑定HEP的数据/L3接口
  	wildcardHostEndpoint, wildcardExists := hostIfaceToEpMap[allInterfaces]
  	if wildcardExists {
  		for ifaceName := range m.nameToIface {
              // 主机/三层网卡未关联hostEndpoint
  			if exist := hostIfaceToEpMap[ifaceName]; (m.isDataIface(ifaceName)||m.isL3Iface(ifaceName))&&!exist{
                  // 关联到通配hostEndpoint
  				hostIfaceToEpMap[ifaceName] = wildcardHostEndpoint
  			}
  		}
          // 清理通配HEP,后续只处理具体网卡
  		delete(hostIfaceToEpMap, allInterfaces)
  	}
  
  	// wildcard HEP变更,所有workload接口受影响(workload的host侧策略)
  	if (wildcardExists != m.wildcardExists) || !reflect.DeepEqual(wildcardHostEndpoint, m.wildcardHostEndpoint){
  		// policiesToWorkloads/profilesToWorkloads映射清理
          m.removeHEPFromIndexes(allInterfaces, &m.wildcardHostEndpoint)
          // 更新通配HEP
  		m.wildcardHostEndpoint = wildcardHostEndpoint
  		m.wildcardExists = wildcardExists
          // 重新注册policiesToWorkloads/profilesToWorkloads映射
  		m.addHEPToIndexes(allInterfaces, &wildcardHostEndpoint)
  		for ifaceName := range m.nameToIface {
              // Pod网卡标记dirty
  			if m.isWorkloadIface(ifaceName) {
  				m.dirtyIfaceNames.Add(ifaceName)
  			}
  		}
  	}
  
  	// 处理已有HEP的变更/删除
  	for ifaceName, existingEp := range m.hostIfaceToEpMap {
  		newEp, stillExists := hostIfaceToEpMap[ifaceName]
          // HEP无变化
  		if stillExists && reflect.DeepEqual(newEp, existingEp) {
  			log.Debugf("No change to host endpoint for ifaceName=%v", ifaceName)
  		} else {
              // policiesToWorkloads/profilesToWorkloads映射清理
  			m.removeHEPFromIndexes(ifaceName, &existingEp)
              // 主机网卡还存在
  			if stillExists {
                  // 重新注册policiesToWorkloads/profilesToWorkloads映射
  				m.addHEPToIndexes(ifaceName, &newEp)
                  // 更新关联HEP
  				m.hostIfaceToEpMap[ifaceName] = newEp
  			} else {
  				delete(m.hostIfaceToEpMap, ifaceName)
  			}
              // 主机网卡标记待更新
  			m.dirtyIfaceNames.Add(ifaceName)
  		}
  		delete(hostIfaceToEpMap, ifaceName)
  	}
  
  	// 处理新增HEP
  	for ifaceName, newEp := range hostIfaceToEpMap {
          // 只处理数据接口(eth0/bpfout/lo...)/L3接口(tun0/wireguard.cali...)
  		if !m.isDataIface(ifaceName) && !m.isL3Iface(ifaceName) {
  			continue
  		}
  
          // 重新注册policiesToWorkloads/profilesToWorkloads映射
  		m.addHEPToIndexes(ifaceName, &newEp)
  		m.hostIfaceToEpMap[ifaceName] = newEp
          // 主机网卡标记待更新
  		m.dirtyIfaceNames.Add(ifaceName)
  	}
  }
  

• 

• wildcard HEP(host-*)是特殊HostEndpoint，匹配所有主机网卡，未显式创建HEP的网卡自动应用wildcard HEP的策略

5.workendpoint

5.1.onUpdate

• workloadEndpoint其实对应Pod网卡，calico cni plugin完成veth和IP分配会主动创建wep对象存储Pod网卡的完整网络信息。

• // adds/updates the workload in the cache along with index from active policy to workloads using that policy.
  func (m *bpfEndpointManager) onWorkloadEndpointUpdate(msg *proto.WorkloadEndpointUpdate) {
  	wlID := *msg.Id
  	oldWEP := m.allWEPs[wlID]
      // 清理旧的wep关联的policy/profile
  	m.removeWEPFromIndexes(wlID, oldWEP)
  
  	wl := msg.Endpoint
  	m.allWEPs[wlID] = wl
      // 注册新的wep关联的policy/profile
  	m.addWEPToIndexes(wlID, wl)
      // 更新iface上的endpointID并标记dirty
  	m.withIface(wl.Name, func(iface *bpfInterface) bool {
  		iface.info.endpointID = &wlID
  		return true // Force interface to be marked dirty in case policies changed.
  	})
  }
  
  func (m *bpfEndpointManager) removeWEPFromIndexes(wlID proto.WorkloadEndpointID, wep *proto.WorkloadEndpoint) {
  	if wep == nil {
  		return
  	}
  
      // // 清理tier中的ingress/egress policy索引
  	for _, t := range wep.Tiers {
          // 解绑wep的ingressPolicy---policiesToWorkloads
  		m.removePolicyToEPMappings(t.IngressPolicies, wlID)
          // 解绑wep的egressPolicy---policiesToWorkloads
  		m.removePolicyToEPMappings(t.EgressPolicies, wlID)
  	}
  
      // 清理profile索引
  	m.removeProfileToEPMappings(wep.ProfileIds, wlID)
  
      // 重置iface上的endpointID及标记dirty
  	m.withIface(wep.Name, func(iface *bpfInterface) bool {
  		iface.info.endpointID = nil
  		return false
  	})
  }
  
  func (m *bpfEndpointManager) addWEPToIndexes(wlID proto.WorkloadEndpointID, wl *proto.WorkloadEndpoint) {
  	for _, t := range wl.Tiers {
          // 注册wep的ingressPolicy---policiesToWorkloads
  		m.addPolicyToEPMappings(t.IngressPolicies, wlID)
          // 注册wep的egressPolicy---policiesToWorkloads
  		m.addPolicyToEPMappings(t.EgressPolicies, wlID)
  	}
      
      // 注册wep的profile---profilesToWorkloads
  	m.addProfileToEPMappings(wl.ProfileIds, wlID)
  }
  

• 

• 网络策略变化不会直接下发到内核，先存到暂存区，后面Apply一起下发

5.2.onRemove

• m.onWorkloadEnpdointRemove()相反，会清理wep关联的polocy/profile映射及健康工作负载缓存，内存iface对象关联的wep会重置。

• // removes the workload from the cache and the index, which maps from policy to workload.
  func (m *bpfEndpointManager) onWorkloadEnpdointRemove(msg *proto.WorkloadEndpointRemove) {
  	wlID := *msg.Id
  	oldWEP := m.allWEPs[wlID]
      // 1.扫描wep关联的polocy/profile
      // 2.清理policiesToWorkloads缓存
  	m.removeWEPFromIndexes(wlID, oldWEP)
  	delete(m.allWEPs, wlID)
  
     	// 清理happyWEPs(成功编程BPF的WEP集合)
  	if m.happyWEPs[wlID] != nil {
  		delete(m.happyWEPs, wlID)
  		m.happyWEPsDirty = true
  	}
  
      // 重置iface关联endpointID及标记网卡dirty
  	m.withIface(oldWEP.Name, func(iface *bpfInterface) bool {
  		iface.info.endpointID = nil
  		return false
  	})
  	...
  }
  

• 目前来看，happyWEPs用来加速的，用于快速扫描网络正常的Pod网卡配置iptables accept chain

6.policy

6.1.update

• policy/profile更新就相对简单，缓存policy/profile网络规则，受影响的Pod网卡标记dirty，后面Apply阶段会基于dirty网卡工作。

• // onPolicyUpdate stores the policy in the cache and marks any endpoints using it dirty.
  func (m *bpfEndpointManager) onPolicyUpdate(msg *proto.ActivePolicyUpdate) {
  	polID := *msg.Id
  	m.policies[polID] = msg.Policy
      // 标记iface待更新
  	m.markEndpointsDirty(m.policiesToWorkloads[polID], "policy")
  	...
  }
  
  // onProfileUpdate stores the profile in the cache and marks any endpoints that use it as dirty.
  func (m *bpfEndpointManager) onProfileUpdate(msg *proto.ActiveProfileUpdate) {
  	profID := *msg.Id
  	m.profiles[profID] = msg.Profile
      // 标记iface待更新
  	m.markEndpointsDirty(m.profilesToWorkloads[profID], "profile")
  	...
  }
  

• 网卡的网络策略会基于缓存的iface-->wep-->policy/profile串起来生成规则

6.2.remove

• policy/profile清理正好相反，暂存区的规则缓存及wep-->policy/profile映射会删除，受影响的网卡会标记位dirty，后面会更新网络规则。

• // removes the policy from the cache and marks any endpoints using it dirty.
  func (m *bpfEndpointManager) onPolicyRemove(msg *proto.ActivePolicyRemove) {
  	polID := *msg.Id
      // 标记iface待更新
  	m.markEndpointsDirty(m.policiesToWorkloads[polID], "policy")
      // 清理缓存
  	delete(m.policies, polID)
  	delete(m.policiesToWorkloads, polID)
  	...
  }
  
  // removes the profile from the cache and marks any endpoints that were using it as dirty.
  func (m *bpfEndpointManager) onProfileRemove(msg *proto.ActiveProfileRemove) {
  	profID := *msg.Id
      // 标记iface待更新
  	m.markEndpointsDirty(m.profilesToWorkloads[profID], "profile")
      // 清理缓存
  	delete(m.profiles, profID)
  	delete(m.profilesToWorkloads, profID)
  	...
  }
  

• policy/profile主要用来限制Pod访问策略，比如限制来源、限制出入站规则...

7.service

7.1.update

• m.onServiceUpdate()用于CTLB兼容场景，收集clusterIP/LBIP调整service路由，利用路由将主机的service流量委托给bpfxx设备。

• func (m *bpfEndpointManager) onServiceUpdate(update *proto.ServiceUpdate) {
      // CTLB禁用不处理,说明此时不是eBPF模式
  	if m.ctlbWorkaroundMode == ctlbWorkaroundDisabled {
  		return
  	}
  
      // UDPOnly模式跳过无UDP端口的Service
  	if m.ctlbWorkaroundMode == ctlbWorkaroundUDPOnly {
  		hasUDP := false
  		for _, port := range update.Ports {
  			if port.Protocol == "UDP" {
  				hasUDP = true
  				break
  			}
  		}
  
  		if !hasUDP {
  			return // skip services that do not have UDP ports
  		}
  	}
  	...
  
      // 收集ClusterIP和LoadBalancerIP
  	if update.ClusterIp != "" {
  		ips = append(ips, update.ClusterIp)
  	}
  	if update.LoadbalancerIp != "" {
  		ips = append(ips, update.LoadbalancerIp)
  	}
  
  	key := serviceKey{name: update.Name, namespace: update.Namespace}
  	...
  
      // 将合法IPv4转换为V4CIDR
  	for _, i := range ips {
  		cidr, err := ip.ParseCIDROrIP(i)
          ...
  		cidrv4, ok := cidr.(ip.V4CIDR)
          ...
          ips4 = append(ips4, cidrv4)
  	}
  
  	// 对比新旧IP,删除已移除IP的路由
  	for _, old := range m.services[key] {
  		...
          // svc的新旧IP对比
  		for _, svcIP := range ips4 {
  			if old == svcIP {
  				exists = true
  				break
  			}
  		}
          
          // 清理旧地址路由
  		if !exists {
  			m.dp.delRoute(old)
  		}
  	}
  
      // 注册新地址
  	m.services[key] = ips4
      // 加入待更新暂存区
  	m.dirtyServices.Add(key)
  }
  
  // 路由委托给routeTable配置
  func (m *bpfEndpointManager) delRoute(cidr ip.V4CIDR) {
  	m.routeTable.RouteRemove(bpfInDev, cidr)
  }
  

• 

• 注意

  可以明确，CTLB未生效情况下，clusterIP/LBIP由于没有关联的主机网卡，无法回退至TC NAT，所以会创建bpfin/bpfout基于路由兜底

7.2.remote

• m.onServiceRemove()相对简单，service对应clusterIP/loadBalanceIP旧地址广播给routeMgr清理路由，删除service地址缓存。

• func (m *bpfEndpointManager) onServiceRemove(update *proto.ServiceRemove) {
      // 这里还是会检查service流量是否接管
  	if m.ctlbWorkaroundMode == ctlbWorkaroundDisabled {
  		return
  	}
  
  	key := serviceKey{name: update.Name, namespace: update.Namespace}
  
      // 清理service地址路由
  	for _, svcIP := range m.services[key] {
  		m.dp.delRoute(svcIP)
  	}
  
      // 解注册地址
  	delete(m.services, key)
  }
  
  func (m *bpfEndpointManager) delRoute(cidr ip.V4CIDR) {
      // 会将svcIP放到暂存区,由routeTable清理路由
  	m.routeTable.RouteRemove(bpfInDev, cidr)
  }
  

• 这里的地址路由用于将service流量转到bpfin—>bpfout，走TC NAT解析为PodIP

7.3.route

• m.onRouteUpdate()处理隧道设备地址变化，设备用于跨节点流量转发，更新后需要重新加载bpfout的TC程序，确保NAT后的地址走到封包设备。

• func (m *bpfEndpointManager) onRouteUpdate(update *proto.RouteUpdate) {
      // IPIP/VXLAN设备路由,目标地址就是隧道设备地址
  	if update.Type == proto.RouteType_LOCAL_TUNNEL {
          // 解析目标地址
  		ip, _, err := net.ParseCIDR(update.Dst)
  		...
          
          // 记录一下隧道设备地址
  		m.tunnelIP = ip
          // bpfout的TC程序依赖tunnelIP封装跨节点报文
  		m.dirtyIfaceNames.Add(bpfOutDev)
  	}
  }
  

• 注意

  这里可以理解为，bpfout TC NAT后的跨节点PodIP需要封包，要打上隧道设备标志，确保路由过去隧道设备接受封装报文