Cilium CNI: tc ReloadDatapath 工作原理解析

// pkg/datapath/loader/loader.go

func (l *Loader) CompileAndLoad(ctx context.Context, ep datapath.Endpoint, stats *metrics.SpanStat) error {
	if ep == nil {
		log.Fatalf("LoadBPF() doesn't support non-endpoint load")
	}

	dirs := directoryInfo{
		Library: option.Config.BpfDir,     // /var/lib/cilium/bpf，存放 BPF 模版文件
		Runtime: option.Config.StateDir,   // /var/run/cilium，存放 endpoint 运行状态
		State:   ep.StateDir(),            // /var/run/cilium/state/{endpoint-id}
		Output:  ep.StateDir(),
	}
	return l.compileAndLoad(ctx, ep, &dirs, stats)
}

func (l *Loader) compileAndLoad(ctx context.Context, ep datapath.Endpoint, dirs *directoryInfo) error {
	err := compileDatapath(ctx, dirs, ep.IsHost(), ep.Logger(Subsystem))  // 编译 BPF 程序
	err = l.reloadDatapath(ctx, ep, dirs)  // 加载 BPF 程序
	return err
}

// pkg/datapath/loader/loader.go

func (l *Loader) reloadDatapath(ctx context.Context, ep datapath.Endpoint, dirs *directoryInfo) error {
	// 替换当前 BPF 程序
	objPath := path.Join(dirs.Output, "bpf_lxc.o")

	// endpoint 是否为 host endpoint
	if ep.IsHost() {
		objPath = path.Join(dirs.Output, "bpf_host.o")
		l.reloadHostDatapath(ctx, ep, objPath)  // 重载 cilium_host 上的 BPF 程序
	} else {
		progs := []progDefinition

		if ep.RequireEgressProg() {
			progs = append(progs, progDefinition{progName: "cil_to_container", direction: "egress"})
		} else {
			err := RemoveTCFilters(ep.InterfaceName(), netlink.HANDLE_MIN_EGRESS)  // 移除接口 egress 方向上所有的 filters
		}

		finalize, err := replaceDatapath(ctx, ep.InterfaceName(), objPath, progs, "")  // 重载 endpoint 接口上的 BPF 程序
		defer finalize()
	}

	if ep.RequireEndpointRoute() {
		if ip := ep.IPv4Address(); ip.IsValid() {  // 获取 endpoint 的 ipv4 地址
			upsertEndpointRoute(ep, *iputil.AddrToIPNet(ip))
		}
		if ip := ep.IPv6Address(); ip.IsValid() {
			upsertEndpointRoute(ep, *iputil.AddrToIPNet(ip))
		}
	}

	return nil
}

// pkg/endpoint/endpoint.go

func parseEndpoint(ctx context.Context, owner regeneration.Owner, ...) (*Endpoint, error) {
	// ...

	// 若有 key 为 "reserved:host" label 的 endpoint 即为 host endpoint
	ep.isHost = ep.HasLabels(labels.LabelHost)

	// ...
}

// pkg/datapath/loader/loader.go

func (l *Loader) reloadHostDatapath(ctx context.Context, ep datapath.Endpoint, objPath string) error {
	nbInterfaces := len(option.Config.GetDevices()) + 2  // default: 2
	symbols := make([]string, 2, nbInterfaces)
	directions := make([]string, 2, nbInterfaces)
	objPaths := make([]string, 2, nbInterfaces)
	interfaceNames := make([]string, 2, nbInterfaces)
	symbols[0], symbols[1] = "cil_to_host", "cil_from_host"
	directions[0], directions[1] = "ingress", "egress"
	objPaths[0], objPaths[1] = objPath, objPath
	interfaceNames[0], interfaceNames[1] = ep.InterfaceName(), ep.InterfaceName()

	if _, err := netlink.LinkByName("cilium_net"); err != nil {
		return err  // cilium_net 和 cilium_host 成对出现，若对端接口不存在，则直接返回错误
	} else {
		// 对于 cilium_net 接口来说，其只需要在 ingress 方向上加载 BPF 程序即可
		interfaceNames = append(interfaceNames, "cilium_net")
		symbols = append(symbols, "cil_to_host")
		directions = append(directions, "ingress")
		secondDevObjPath := path.Join(ep.StateDir(), "bpf_host_cilium_net.o")
		err := patchHostNetdevDatapath(ep, objPath, secondDevObjPath, "cilium_net", nil)  // 填充一些接口信息
		objPaths = append(objPaths, secondDevObjPath)
	}

	bpfMasqIPv4Addrs := node.GetMasqIPv4AddrsWithDevices()

	// 默认情况下该配置项为空，故一般不执行此循环
	for _, device := range option.Config.GetDevices() {
		if _, err := netlink.LinkByName(device); err != nil {
			continue
		}

		netdevObjPath := path.Join(ep.StateDir(), "bpf_netdev_"+device+".o")
		err := patchHostNetdevDatapath(ep, objPath, netdevObjPath, device, bpfMasqIPv4Addrs)
		objPaths = append(objPaths, netdevObjPath)
		interfaceNames = append(interfaceNames, device)
		symbols = append(symbols, "cil_from_netdev")
		directions = append(directions, "ingress")

		// ... 判断是否需要加载 cil_to_netdev 到接口 egress 方向
	}

	// 针对每个接口，分别重载属于该接口、接口方向的 BPF 程序
	for i, interfaceName := range interfaceNames {
		symbol := symbols[i]
		progs := []progDefinition
		finalize, err := replaceDatapath(ctx, interfaceName, objPaths[i], progs, "")  // ***
		defer finalize()
	}

	return nil
}

// pkg/datapath/loader/netlink.go

func replaceDatapath(ctx context.Context, ifName, objPath string, progs []progDefinition, xdpMode string) (func(), error) {

	link, err := netlink.LinkByName(ifName)

	// 从磁盘加载 eBPF ELF 文件，并解析为 CollectionSpec
	spec, err := bpf.LoadCollectionSpec(objPath)

	for _, prog := range progs {
		if spec.Programs[prog.progName] == nil {  // 查询重载的程序是否包含 BPF 程序中
			return nil, // not-found
		}
	}

	// 加载 CollectionSpec 至内核，并 pin 在 bpffs 的 TCGlobalsPath 路径上
	finalize := func() {}
	opts := ebpf.CollectionOptions{
		Maps: ebpf.MapOptions{PinPath: bpf.TCGlobalsPath()},
	}
	coll, err := bpf.LoadCollection(spec, opts)
	if errors.Is(err, ebpf.ErrMapIncompatible) {
		// 若路径上原有的 spec 与现加载的 spec 不同，就尝试重新加载新的 spec
		err := bpf.StartBPFFSMigration(bpf.TCGlobalsPath(), spec)

		finalize = func() {
			bpf.FinalizeBPFFSMigration(bpf.TCGlobalsPath(), spec, false)  // 删除现有加载 maps
		}

		// 上述重新加载完毕后，再次重试加载 CollectionSpec
		coll, err = bpf.LoadCollection(spec, opts)
	}
	var ve *ebpf.VerifierError
	if errors.As(err, &ve) {
		// Verifier error
	}
	defer coll.Close()

	for _, prog := range progs {
		// 将程序挂载到接口上
		if err := attachProgram(link, coll.Programs[prog.progName], prog.progName, directionToParent(prog.direction), xdpModeToFlag(xdpMode)); err != nil {
			bpf.FinalizeBPFFSMigration(bpf.TCGlobalsPath(), spec, true)  // 回滚到原有 maps
			return nil, err
		}
	}

	return finalize, nil
}

func attachProgram(link netlink.Link, prog *ebpf.Program, progName string, qdiscParent uint32, xdpFlags uint32) error {
	if prog == nil {
		return errors.New("cannot attach a nil program")
	}

	if xdpFlags != 0 {
		// 挂载程序到 XDP
		netlink.LinkSetXdpFdWithFlags(link, prog.FD(), int(xdpFlags))
		return nil
	}

	err := replaceQdisc(link)  // 替换接口现有的 clsact qdisc

	filter := &netlink.BpfFilter{
		FilterAttrs: netlink.FilterAttrs{
			LinkIndex: link.Attrs().Index,
			Parent:    qdiscParent,
			Handle:    1,
			Protocol:  unix.ETH_P_ALL,
			Priority:  option.Config.TCFilterPriority,
		},
		Fd:           prog.FD(),
		Name:         fmt.Sprintf("%s-%s", progName, link.Attrs().Name),
		DirectAction: true,  // 启用 direct-action 模式
	}

	err := netlink.FilterReplace(filter)  // 替换现有的 tc filter

	return nil
}

// daemon/cmd/endpoint.go

func (d *Daemon) createEndpoint(ctx context.Context, owner regeneration.Owner, epTemplate *models.EndpointChangeRequest) (*endpoint.Endpoint, int, error) {
	if option.Config.EnableEndpointRoutes {  // default: "false"

		// 是否对每个 endpoint 都插入一条路由，而非使用经过 cilium_host 的路由
		epTemplate.DatapathConfiguration.InstallEndpointRoute = true  // 对应 RequireEndpointRoute()

		// 由于直接通过 endpoint 的接口路由，绕过了 cilium_host 接口，所以 BPF 程序需要挂载在 endpoint 接口的 egress 方向
		epTemplate.DatapathConfiguration.RequireEgressProg = true  // 对应 RequireEgressProg()

		// ...
	}
	// ...
}