MetalLB 工作原理解析

// internal/k8s/controllers/service_controller.go

func (r *ServiceReconciler) reconcileService(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	// ...
	var service *v1.Service

	// 根据 Endpoint 提供的 NamespacedName 对象寻找对应的 Service 对象
	service, err := r.serviceFor(ctx, req.NamespacedName)
	if err != nil {					\
		return ctrl.Result{}, err		 \
	}						  -->-- r.Get(ctx, name, &res)

        // 若 MetalLB 的配置文件中指定了 LoadBalancerClass，则比对它和 Service 的是否一致
        // 只有一致或无指定配置时才可通过，默认情况下，配置文件不指定该字段
	if filterByLoadBalancerClass(service, r.LoadBalancerClass) {
		return ctrl.Result{}, nil
	}

	// 根据 Service 获取其所代理的 Endpoints 或 EndpointSlice
	epSlices, err := epsOrSlicesForServices(ctx, r, req.NamespacedName, r.Endpoints)
	if err != nil {
		return ctrl.Result{}, err
	}
	// 此时根据 Service 是否为空，可以判断出此次调谐是对 Service 的删除还是更新

	// 对 Service 进行处理，包括 IP 地址的分配和回收
	res := r.Handler(r.Logger, req.NamespacedName.String(), service, epSlices)
	switch res {
	case SyncStateError:
		return ctrl.Result{}, retryError
	case SyncStateReprocessAll:
		// 重新进行全量的调谐
		r.forceReload()
		return ctrl.Result{}, nil
	case SyncStateErrorNoRetry:
		return ctrl.Result{}, nil
	}
	return ctrl.Result{}, nil
}

ctrl.NewControllerManagedBy(mgr).
    For(&v1.Service{}).
    Watches(&source.Kind{Type: &v1.Endpoints{}},
        handler.EnqueueRequestsFromMapFunc(func(obj client.Object) []reconcile.Request {
            endpoints, ok := obj.(*v1.Endpoints)
            if !ok {
                return []reconcile.Request{}
            }
            name := types.NamespacedName{Name: endpoints.Name, Namespace: endpoints.Namespace}
            return []reconcile.Request
        })).
    Watches(&source.Channel{Source: r.Reload}, &handler.EnqueueRequestForObject{}).
    Complete(r)

func (r *ServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	if !isReloadReq(req) {
		return r.reconcileService(ctx, req)
	}
	return r.reprocessAll(ctx, req)
}

// internal/k8s/controllers/service_controller_reload.go
func isReloadReq(req ctrl.Request) bool {
	if req.Name == "reload" && req.Namespace == "metallbreload" {
		return true
	}
	return false
}

// controller/main.go

func (c *controller) SetBalancer(l log.Logger, name string, svcRo *v1.Service, _ epslices.EpsOrSlices) controllers.SyncState {
	// 对于空的 Service 即触发回收 IP 操作
	if svcRo == nil {  // Read only
		c.deleteBalancer(l, name) --->---
                                                  \
                                                   \
                                                    c.ips.Unassign(name)

                // 触发后进行全量调谐，因为可能存在其他 LB 类型的 Service 在等待 IP 地址的分配
		return controllers.SyncStateReprocessAll
	}

	// 在分配 IP 地址之前，先确保地址池是配置过的
	if c.pools == nil || c.pools.ByName == nil {
		return controllers.SyncStateSuccess
	}

	svc := svcRo.DeepCopy()
	successRes := controllers.SyncStateSuccess
	// 检查该服务是否被分配过 IP 地址
	wasAllocated := c.isServiceAllocated(name) --->---
        // 获取与分配 IP                                    \
	c.convergeBalancer(l, name, svc)                    \
                                                    	 c.ips.Pool(key) != ""

        // convergeBalancer 可能会取消对 Service 的 IP 分配，若此种情况发生
	if wasAllocated && !c.isServiceAllocated(name) {
		// 被回收的 IP 地址可能还会被其他 LB 类型的 Service 使用，所以再进行全量调谐
		successRes = controllers.SyncStateReprocessAll
	}

	// 对于没有发生任何变化的 Service，则直接返回
	if reflect.DeepEqual(svcRo, svc) {
		return successRes
	}

	toWrite := svcRo.DeepCopy()
	// 最后再次与 svcRo 的 Status 字段进行比对，发生变化了则直接进行更新；因为 svc 在 convergeBalancer 中可能会发生变化
	if !reflect.DeepEqual(svcRo.Status, svc.Status) {
		toWrite.Status = svc.Status
	}
	// Annotations 字段也是，发生变化了则直接进行更新
	if !reflect.DeepEqual(svcRo.Annotations, svc.Annotations) {
		toWrite.Annotations = svc.Annotations
	}
	// 只有上述两个字段发生了更新时，才会引发这两者的不同，进而才会进行更新
	if !reflect.DeepEqual(toWrite, svcRo) {
		if err := c.client.UpdateStatus(svc); err != nil {
			return controllers.SyncStateError
		}
		return successRes
	}

	return successRes
}

// controller/service.go

// #1
func (c *controller) convergeBalancer(l log.Logger, key string, svc *v1.Service) {
	lbIPs := []net.IP{}
	var err error

	// 对于非 LoadBalancer 类型的 Service，可提前返回；同时还清除了 Service 的状态信息
	if svc.Spec.Type != v1.ServiceTypeLoadBalancer {
		c.clearServiceState(key, svc) --->---
		return				      \
	}					       \
                                                        \
                                                        func (c *controller) clearServiceState(key string, svc *v1.Service) {
                                                            c.ips.Unassign(key)
                                                            delete(svc.Annotations, annotationIPAllocateFromPool)  // => "metallb.universe.tf/ip-allocated-from-pool"
                                                            svc.Status.LoadBalancer = v1.LoadBalancerStatus{}
                                                          }

        // MetalLB 会根据 ClusterIP 的类型来决定使用的地址族，故对于没有 ClusterIP 的 Service 则直接返回
	if len(svc.Spec.ClusterIPs) == 0 && svc.Spec.ClusterIP == "" {
		c.clearServiceState(key, svc)
		return
	}

	// ...

// #2
	// 获取所有在 Status 中 Ingress 字段出现的 IP 地址
	for i := range svc.Status.LoadBalancer.Ingress {
		ip := svc.Status.LoadBalancer.Ingress[i].IP
		if len(ip) != 0 {
			lbIPs = append(lbIPs, net.ParseIP(ip))
		}
	}
	// 若 IP 地址为空，或是所有 IP 地址的解析都不正确，则会清除当前 Service 的状态
	if len(lbIPs) == 0 {
		c.clearServiceState(key, svc)
	} else {
		// 确定当前 LB IP 的 IP 地址家族
		lbIPsIPFamily, err := ipfamily.ForAddressesIPs(lbIPs)
		// 确定 ClusterIP 的 IP 地址家族
		clusterIPsIPFamily, err := ipfamily.ForService(svc)
		if err != nil {
			return
		}
		// 若 LB IP 和 ClsuterIP 的 IP 地址家族不一致，则非有效的 IP 地址
		if lbIPsIPFamily != clusterIPsIPFamily || lbIPsIPFamily == ipfamily.Unknown {
			c.clearServiceState(key, svc)
			lbIPs = []net.IP{}
		}
	}

	// ...

// #3
	// 对于现有的 LB IP，它们可能随着配置的更该而不再适用，所以需要再次进行检查并提供再次分配 LB IP 的机会
	if len(lbIPs) != 0 {
		// 地址分配的操作是幂等的，详细说明见下节内容
		if err = c.ips.Assign(key, svc, lbIPs, k8salloc.Ports(svc), k8salloc.SharingKey(svc), k8salloc.BackendKey(svc)); err != nil {
			c.clearServiceState(key, svc)
			lbIPs = []net.IP{}
		}
		// 对于地址池 annotation 被修改的情况，意味着需要使用一个新的地址池进行地址分配
		desiredPool := svc.Annotations[annotationAddressPool]  // => "metallb.universe.tf/address-pool"
		if len(lbIPs) != 0 && desiredPool != "" && c.ips.Pool(key) != desiredPool {
			c.clearServiceState(key, svc)
			lbIPs = []net.IP{}
		}
		// 获取期望的 LB IP
		desiredLbIPs, _, err := getDesiredLbIPs(svc)
		if err != nil {
			return
		}
		// 若存在期望的 LB IP，且当前 LB IP 与期望的 LB IP 不同，则清空现有状态
		if len(desiredLbIPs) > 0 && !isEqualIPs(lbIPs, desiredLbIPs) {
			c.clearServiceState(key, svc)
			lbIPs = []net.IP{}
		}
	}

    // ...

// #4
	// 到此为止，对于没有 LB IP 的 Service 才进行地址分配，详细说明见下节内容
	if len(lbIPs) == 0 {
		lbIPs, err = c.allocateIPs(key, svc)
		if err != nil {
			return
		}
	}

	// IP 分配失败
	if len(lbIPs) == 0 {
		c.clearServiceState(key, svc)
		return
	}

	// 检查该分配 IP 对应的地址池是否存在
	pool := c.ips.Pool(key)
	if pool == "" || c.pools == nil || c.pools.IsEmpty(pool) {
		c.clearServiceState(key, svc)
		return
	}

	// 最后，记录分配的 IP 到 Service 的 Status 和 Annotations 字段
	lbIngressIPs := []v1.LoadBalancerIngress{}
	for _, lbIP := range lbIPs {
		lbIngressIPs = append(lbIngressIPs, v1.LoadBalancerIngress{IP: lbIP.String()})
	}
	svc.Status.LoadBalancer.Ingress = lbIngressIPs
	if svc.Annotations == nil {
		svc.Annotations = make(map[string]string)
	}
	svc.Annotations[annotationIPAllocateFromPool] = pool  // => "metallb.universe.tf/ip-allocated-from-pool"
}

// controller/service.go

func (c *controller) allocateIPs(key string, svc *v1.Service) ([]net.IP, error) {
	// 确定 Service 所使用的 IP 地址类型，确定方式见上文
	serviceIPFamily, err := ipfamily.ForService(svc)
	if err != nil {
		return nil, err
	}
	desiredLbIPs, desiredLbIPFamily, err := getDesiredLbIPs(svc)
	if err != nil {
		return nil, err
	}

	// 若用户指定了期望 LB IP，则先尝试分配这个 IP
	if len(desiredLbIPs) > 0 {
		if serviceIPFamily != desiredLbIPFamily {
			return nil, // err
		}
		if err := c.ips.Assign(key, svc, desiredLbIPs, k8salloc.Ports(svc), k8salloc.SharingKey(svc), k8salloc.BackendKey(svc)); err != nil {
			return nil, err
		}
		return desiredLbIPs, nil
	}
	// 否则，从地址池中分配一个 IP 地址
	desiredPool := svc.Annotations[annotationAddressPool]
	if desiredPool != "" {
		ips, err := c.ips.AllocateFromPool(key, svc, serviceIPFamily, desiredPool, k8salloc.Ports(svc), k8salloc.SharingKey(svc), k8salloc.BackendKey(svc))
		if err != nil {
			return nil, err
		}
		return ips, nil
	}

	// 若地址池没有被指定，则从所有跟该 Service 相关的地址池中分配
	return c.ips.Allocate(key, svc, serviceIPFamily, k8salloc.Ports(svc), k8salloc.SharingKey(svc), k8salloc.BackendKey(svc))
}

type controller struct {
	client service
	pools  *config.Pools
	ips    *allocator.Allocator
}

// internal/allocator/allocator.go
type Allocator struct {
	pools *config.Pools

	allocated       map[string]*alloc          // svc -> alloc，记录已分配的 IP 信息
	sharingKeyForIP map[string]*key            // ip.String() -> assigned sharing key
	portsInUse      map[string]map[Port]string // ip.String() -> Port -> svc
	servicesOnIP    map[string]map[string]bool // ip.String() -> svc -> allocated?
	poolIPsInUse    map[string]map[string]int  // poolName -> ip.String() -> number of users
}

type alloc struct {
	pool  string
	ips   []net.IP
	ports []Port
	key   --->---
}                     \
                       \
                      type key struct {
                          sharing string
                          backend string
                        }

// internal/allocator/allocator.go

func (a *Allocator) Allocate(svcKey string, svc *v1.Service, serviceIPFamily ipfamily.Family, ports []Port, sharingKey, backendKey string) ([]net.IP, error) {
	// 对于已经被分配地址的 Service，这里再次尝试指定地址
	if alloc := a.allocated[svcKey]; alloc != nil {
		// 指定的还是原来已经分配的地址，这里的主要目的是对原地址的合法性再次进行校验；若校验通过，Allocator.allocated 字段虽然会更新，但是内容不变
		if err := a.Assign(svcKey, svc, alloc.ips, ports, sharingKey, backendKey); err != nil {
			return nil, err
		}
		return alloc.ips, nil
	}
	// 获取 serviceAllocation 中规定的，与当前 Service 各种原数据或命名空间相匹配的地址池，并按照地址池的优先级降序排序
	pinnedPools := a.pinnedPoolsForService(svc)
	for _, pool := range pinnedPools {
                // 只要从一个地址池中分配 IP 成功，则直接返回该分配的 IP
		if ips, err := a.AllocateFromPool(svcKey, svc, serviceIPFamily, pool.Name, ports, sharingKey, backendKey); err == nil {
			return ips, nil
		}
	}
	// 遍历所有地址池，过滤掉所有非租户的地址池或不会自动分配 IP 的地址池
	for _, pool := range a.pools.ByName {
		if !pool.AutoAssign || pool.ServiceAllocations != nil {
			continue
		}
		if ips, err := a.AllocateFromPool(svcKey, svc, serviceIPFamily, pool.Name, ports, sharingKey, backendKey); err == nil {
			return ips, nil
		}
	}

	return nil, errors.New("no available IPs")
}

func (a *Allocator) AllocateFromPool(svcKey string, svc *v1.Service, serviceIPFamily ipfamily.Family, poolName string, ports []Port, sharingKey, backendKey string) ([]net.IP, error) {
	if alloc := a.allocated[svcKey]; alloc != nil {
		// ...
		if err := a.Assign(svcKey, svc, alloc.ips, ports, sharingKey, backendKey); err != nil {
			return nil, err
		}
		return alloc.ips, nil
	}

	// 获取该指定的地址池对象
	pool := a.pools.ByName[poolName]
	ips := []net.IP{}
	// 根据 IP 地址家族决定分配的地址类型
	ipfamilySel := make(map[ipfamily.Family]bool)
	switch serviceIPFamily {
	case ipfamily.DualStack:
		ipfamilySel[ipfamily.IPv4], ipfamilySel[ipfamily.IPv6] = true, true
	default:
		ipfamilySel[serviceIPFamily] = true
	}

	for _, cidr := range pool.CIDR {
		// 地址池的 CIDR 要在和目的 IP 地址类型相同时，才能被分配
		cidrIPFamily := ipfamily.ForCIDR(cidr)
		if _, ok := ipfamilySel[cidrIPFamily]; !ok {
			continue
		}
		ip := a.getIPFromCIDR(cidr, pool.AvoidBuggyIPs, svcKey, ports, sharingKey, backendKey)  // 获取 IP
		if ip != nil {
			ips = append(ips, ip)
			delete(ipfamilySel, cidrIPFamily)
		}
	}

	// 存在没有被分配的 IP 地址类型，说明地址池已耗尽
	if len(ipfamilySel) > 0 {
		return nil, // err
	}
	err := a.Assign(svcKey, svc, ips, ports, sharingKey, backendKey)  // 将分配后的 IP 指定给 Service
	if err != nil {
		return nil, err
	}
	return ips, nil
}

func (a *Allocator) Assign(svcKey string, svc *v1.Service, ips []net.IP, ports []Port, sharingKey, backendKey string) error {

	// check ...

	alloc := &alloc{
		pool:  pool.Name,
		ips:   ips,
		ports: make([]Port, len(ports)),
		key:   *sk,
	}
	copy(alloc.ports, ports)
	a.assign(svcKey, alloc)
	return nil
}

// internal/allocator/allocator.go

func (a *Allocator) checkSharing(svc string, ip string, ports []Port, sk *key) error {
	if existingSK := a.sharingKeyForIP[ip]; existingSK != nil {
		// 检查 sharing-key 是否相同
		if err := sharingOK(existingSK, sk); err != nil {
			// ...
		}

		// 检查端口是否被占用，端口由协议和端口号两部分组成
		for _, port := range ports {
			if curSvc, ok := a.portsInUse[ip][port]; ok && curSvc != svc {
				return // err
			}
		}
	}
	return nil
}

// speaker/main.go

func newController(cfg controllerConfig) (*controller, error) {
	handlers := map[config.Proto]Protocol{
		config.BGP: &bgpController{/*...*/},
	}
	protocols := []config.Proto{config.BGP}

	if !cfg.DisableLayer2 {  // 虽然有 Layer2 模式的开关，但在实现中并没有发现该配置的可设置项
		a, err := layer2.New(cfg.Logger)  // 初始化 Layer2 Announcer
		handlers[config.Layer2] = &layer2Controller{/*...*/}
		protocols = append(protocols, config.Layer2)
	}

	ret := &controller{  // 初始化 speaker 的 controller
		// ...
                protocolHandlers: handlers,
		announced:        map[config.Proto]map[string]bool{},
		svcIPs:           map[string][]net.IP{},
		protocols:        protocols,
	}
	ret.announced[config.BGP] = map[string]bool{}
	ret.announced[config.Layer2] = map[string]bool{}

	return ret, nil
}

type Protocol interface {
	SetConfig(log.Logger, *config.Config) error
	ShouldAnnounce(log.Logger, string, []net.IP, *config.Pool, *v1.Service, epslices.EpsOrSlices) string
	SetBalancer(log.Logger, string, []net.IP, *config.Pool, service, *v1.Service) error
	DeleteBalancer(log.Logger, string, string) error
	SetNode(log.Logger, *v1.Node) error
}

for _, protocol := range c.protocols {
    if st := c.handleService(l, name, lbIPs, svc, pool, eps, protocol); st == controllers.SyncStateError {
        return st
    }
}

// speaker/layer2_controller.go

func (c *layer2Controller) ShouldAnnounce(l log.Logger, name string, toAnnounce []net.IP, pool *config.Pool, svc *v1.Service, eps epslices.EpsOrSlices) string {
	// 检查 Endpoint 或 EndpointSlice 是否处于 Ready 状态
	if !activeEndpointExists(eps) {
		return "notOwner"
	}

	// 检查 speaker 所在 Node 是否匹配地址池中 L2Advertisements 的 NodeSelector
	if !poolMatchesNodeL2(pool, c.myNode) {
		return "notOwner"
	}

	// 选出所有匹配地址池中 L2Advertisements NodeSelector 的 speaker Node
	forPool := speakersForPool(c.sList.UsableSpeakers(), pool)  // 当然是从所有有效的 speaker 中选
	var nodes []string
	// 根据不同的外部流量策略，选出候选 Node
	if svc.Spec.ExternalTrafficPolicy == v1.ServiceExternalTrafficPolicyTypeLocal {
		// 对于 local 类型，只有 Endpoints 出现在的 Node 才可作为候选
		nodes = usableNodes(eps, forPool)
	} else {
		// 对于 cluster 类型，上述所有 Node 都可作为候选
		nodes = nodesWithActiveSpeakers(forPool)
	}
	ipString := toAnnounce[0].String()
	// 根据 node 名 + LB IP 的哈希值对 nodes 进行排序
	sort.Slice(nodes, func(i, j int) bool {
		hi := sha256.Sum256([]byte(nodes[i] + "#" + ipString))
		hj := sha256.Sum256([]byte(nodes[j] + "#" + ipString))
		return bytes.Compare(hi[:], hj[:]) < 0
	})

	// 若当前 speaker Node 是排序后 Node 列表中的第一个，则就该由本 speaker 来承担宣告工作
	if len(nodes) > 0 && nodes[0] == c.myNode {
		return ""
	}

	return "notOwner"
}

// internal/layer2/announcer.go

func (a *Announce) updateInterfaces() {
	ifs, err := net.Interfaces()
	// ...

	for _, intf := range ifs {
		ifi := intf

		if ifi.Flags&net.FlagUp == 0 {  // 是否启动
			continue
		}
		if _, err = os.Stat("/sys/class/net/" + ifi.Name + "/master"); !os.IsNotExist(err) {  // 是否存在
			continue
		}
		f, err := os.ReadFile("/sys/class/net/" + ifi.Name + "/flags")  // 是否支持 ARP
		if err == nil {
			flags, _ := strconv.ParseUint(string(f)[:len(string(f))-1], 0, 32)
			// NOARP flag
			if flags&0x80 != 0 {
				continue
			}
		}
		if ifi.Flags&net.FlagBroadcast != 0 {  // 是否支持广播
			keepARP[ifi.Index] = true
		}

		// ...

		//初始化并保存所有接口对应的 Responder
		if keepARP[ifi.Index] && a.arps[ifi.Index] == nil {
			resp, err := newARPResponder(a.logger, &ifi, a.shouldAnnounce)
			a.arps[ifi.Index] = resp
		}
		if keepNDP[ifi.Index] && a.ndps[ifi.Index] == nil {
			resp, err := newNDPResponder(a.logger, &ifi, a.shouldAnnounce)
			a.ndps[ifi.Index] = resp
		}
	}

    // ...
}

// speaker/layer2_controller.go

func (c *layer2Controller) SetBalancer(l log.Logger, name string, lbIPs []net.IP, pool *config.Pool, client service, svc *v1.Service) error {
	// 获取 Announcer 统计的接口
	ifs := c.announcer.GetInterfaces()
	for _, lbIP := range lbIPs {
		// 获取该 LB IP 对应的 IPAdvertisement，里面记录了规定使用的接口
		ipAdv := ipAdvertisementFor(lbIP, c.myNode, pool.L2Advertisements)
		// 对比看两者接口是否匹配
		if !ipAdv.MatchInterfaces(ifs...) {
			continue
		}
		c.announcer.SetBalancer(name, ipAdv)  // 对外进行广播
	}
	return nil
}

func ipAdvertisementFor(ip net.IP, localNode string, l2Advertisements []*config.L2Advertisement) layer2.IPAdvertisement {
	ifs := sets.Set[string]{}  // 记录规定使用的接口
	for _, l2 := range l2Advertisements {
		// 跳过不属于该 Node 的地址池
		if matchNode := l2.Nodes[localNode]; !matchNode {
			continue
		}
		// 若要使用所有接口，不设置任何配置即可
		if l2.AllInterfaces {
			return layer2.NewIPAdvertisement(ip, true, sets.Set[string]{})
		}
		ifs = ifs.Insert(l2.Interfaces...)
	}
	return layer2.NewIPAdvertisement(ip, false, ifs)
}

// internal/layer2/announcer.go

func (a *Announce) SetBalancer(name string, adv IPAdvertisement) {  // name 为 Service name
	// 向 spamCh 写入数据，触发 spamLoop 发送 ARP 响应
	defer a.doSpam(adv)  --->--- a.spamCh <- adv

	a.Lock()
	defer a.Unlock()

	// 一个 Service 的 ipAdvertisement 可能会更新很多次，但只处理第一次
	if ipAdvertisements, ok := a.ips[name]; ok {
		for i := range ipAdvertisements {
			if adv.ip.Equal(a.ips[name][i].ip) {
				a.ips[name][i] = adv // 对于已有的，覆盖原来的值，以防接口变化了
				return
			}
		}
	}
	a.ips[name] = append(a.ips[name], adv)

	// 记录该 IP 的引用次数
	a.ipRefcnt[adv.ip.String()]++

	// ... 执行 defer
}

func (a *Announce) gratuitous(adv IPAdvertisement) {
	a.RLock()
	defer a.RUnlock()

	ip := adv.ip
	// 若当前 Node 对于 ip 的引用计数为 0，说明该 Node 不是进行广播的
	if a.ipRefcnt[ip.String()] <= 0 {
		return
	}

	if ip.To4() != nil {
		for _, client := range a.arps {
			// 只使用与规定接口匹配的 responder 接口
			if !adv.matchInterface(client.intf) {
				continue
			}
			client.Gratuitous(ip)
		}
	} else {
		// 至于 ipv6 类型，处理方式也同上
		for _, client := range a.ndps {
			if !adv.matchInterface(client.intf) {
				continue
			}
			client.Gratuitous(ip)
		}
	}
}

// internal/layer2/arp.go

func (a *arpResponder) processRequest() dropReason {
	pkt, eth, err := a.conn.Read()
	if err != nil {
		return dropReasonError
	}

	// 忽略 ARP 响应
	if pkt.Operation != arp.OperationRequest {
		return dropReasonARPReply
	}

	// 忽略非广播型并且目的 MAC 地址为当前节点的 ARP 请求
	if !bytes.Equal(eth.Destination, ethernet.Broadcast) && !bytes.Equal(eth.Destination, a.hardwareAddr) {
		return dropReasonEthernetDestination
	}

	// 忽略 Announcer 规定忽略的 ARP 请求
	reason := a.announce(pkt.TargetIP, a.intf)
	if reason != dropReasonNone {
		return reason
	}

	a.conn.Reply(pkt, a.hardwareAddr, pkt.TargetIP)  // 对 ARP 请求进行响应
	return dropReasonNone
}

// internal/layer2/announcer.go

func (a *Announce) shouldAnnounce(ip net.IP, intf string) dropReason {
	a.RLock()
	defer a.RUnlock()
	ipFound := false
	for _, ipAdvertisements := range a.ips {
		for _, i := range ipAdvertisements {
			if i.ip.Equal(ip) {
				ipFound = true
				if i.matchInterface(intf) {  // 是合法的 IP 但非规定的接口
					return dropReasonNone
				}
			}
		}
	}
	if ipFound {
		return dropReasonNotMatchInterface
	}
	return dropReasonAnnounceIP
}

// internal/layer2/arp.go

func (a *arpResponder) Gratuitous(ip net.IP) error {
	for _, op := range []arp.Operation{arp.OperationRequest, arp.OperationReply} {
		pkt, err := arp.NewPacket(op, a.hardwareAddr, ip, ethernet.Broadcast, ip)
		a.conn.WriteTo(pkt, ethernet.Broadcast)
	}
	return nil
}

// internal/layer2/ndp.go

func (n *ndpResponder) Gratuitous(ip net.IP) error {
	err := n.advertise(net.IPv6linklocalallnodes, ip, true)  // 特殊的 ipv6 多播地址
	return err
}

func (n *ndpResponder) advertise(dst, target net.IP, gratuitous bool) error {
	m := &ndp.NeighborAdvertisement{
		Solicited:     !gratuitous,
		Override:      gratuitous,  // Should clients replace existing cache entries
		TargetAddress: target,
		Options: []ndp.Option{
			&ndp.LinkLayerAddress{
				Direction: ndp.Target,
				Addr:      n.hardwareAddr,
			},
		},
	}
	return n.conn.WriteTo(m, nil, dst)
}

func (n *ndpResponder) Watch(ip net.IP) error {
	// ...
	group, err := ndp.SolicitedNodeMulticast(ip)

	if n.solicitedNodeGroups[group.String()] == 0 {
		n.conn.JoinGroup(group)
	}
	n.solicitedNodeGroups[group.String()]++
	return nil
}

func (n *ndpResponder) Unwatch(ip net.IP) error {
	// ...
	group, err := ndp.SolicitedNodeMulticast(ip)

	n.solicitedNodeGroups[group.String()]--
	if n.solicitedNodeGroups[group.String()] == 0 {
		n.conn.LeaveGroup(group)
	}
	return nil
}

func (n *ndpResponder) processRequest() dropReason {
	msg, _, src, err := n.conn.ReadFrom()
	if err != nil {
		return dropReasonError
	}

	// 只处理 NS 类型的消息
	ns, ok := msg.(*ndp.NeighborSolicitation)
	if !ok {
		return dropReasonMessageType
	}

	// 提取发送者的源 MAC 地址
	var nsLLAddr net.HardwareAddr
        for _, o := range ns.Options {
		lla, ok := o.(*ndp.LinkLayerAddress)
		if !ok {
			continue
		}
		if lla.Direction != ndp.Source {
			continue
		}
		nsLLAddr = lla.Addr
		break
	}
	if nsLLAddr == nil {
		return dropReasonNoSourceLL
	}

	// announce 方法与上文 ARP Responder 中的一样
	reason := n.announce(ns.TargetAddress, n.intf)
	if reason != dropReasonNone {
		return reason
	}

	n.advertise(src, ns.TargetAddress, false)  // 回复 NA 类型的消息，单播地址
	// ...
	return dropReasonNone
}

// internal/speakerlist/speakerlist.go

func (sl *SpeakerList) Start(client *k8s.Client) {
	sl.client = client

	// 初始化 pod IP 列表，即在 metallb-system 命名空间下的 speaker pod 的 IP
	iplist, err := sl.mlSpeakers()
	sl.mlMux.Lock()
	sl.mlSpeakerIPs = iplist
	sl.mlMux.Unlock()

	go sl.updateSpeakerIPs()
	go sl.memberlistWatchEvents()
	go sl.joinMembers()
}

func (sl *SpeakerList) UsableSpeakers() map[string]bool {
	if sl.ml == nil {
		return nil
	}
	activeNodes := map[string]bool{}
	for _, n := range sl.ml.Members() {  // memberlist method
		activeNodes[n.Name] = true
	}
	return activeNodes
}

// speaker/bgp_controller.go

func (c *bgpController) syncPeers(l log.Logger) error {
	var (
		errs          int
		needUpdateAds bool
	)

	// 遍历所有 peers，这些 peers 是当前最新的
	for _, p := range c.peers {
		// 匹配每个 peer 上的 NodeSeletor，决定该 Node 是否对当前 peer 生效
		shouldRun := false
		if len(p.cfg.NodeSelectors) == 0 {
			shouldRun = true
		}
		for _, ns := range p.cfg.NodeSelectors {
			if ns.Matches(c.nodeLabels) {
				shouldRun = true
				break
			}
		}

		// 若 session 非空但是 Node 已经不生效了，则关闭当前 session
		if p.session != nil && !shouldRun {
			p.session.Close()  // --->--- conn.Close()
			p.session = nil
		} else if p.session == nil && shouldRun {
			// 若 session 不存在但是 Node 在生效中，则创建新的 session
			var routerID net.IP
			if p.cfg.RouterID != nil {
				routerID = p.cfg.RouterID
			}
			s, err := c.sessionManager.NewSession(c.logger,  // 创建 session 并尝试进行连接
				bgp.SessionParameters{
					// ...
				},
			)
                        p.session = s
                        needUpdateAds = true
		}
	}

	// 对于有新创建 session 的情况，需要重新发送一次广播
	if needUpdateAds {
		err := c.updateAds()
	}
	return nil
}

// internal/bgp/bgp.go

type SessionManager interface {
	NewSession(logger log.Logger, args SessionParameters) (Session, error)
	SyncBFDProfiles(profiles map[string]*config.BFDProfile) error
}

// speaker/bgp_controller.go

func (c *bgpController) updateAds() error {
	var allAds []*bgp.Advertisement
	for _, ads := range c.svcAds {
		allAds = append(allAds, ads...)
	}
	for _, peer := range c.peers {
		if peer.session == nil {
			continue
		}
		// 针对已建立 session 的 peer 进行 IP 广播
		if err := peer.session.Set(allAds...); err != nil {
			return err
		}
	}
	return nil
}

// internal/bgp/native/native.go

func (s *session) Set(advs ...*bgp.Advertisement) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	newAdvs := map[string]*bgp.Advertisement{}
	for _, adv := range advs {
		// 遍历该 IP 对应的所有 peers，看当前 peer 是否在其中，若在则匹配
		if !adv.MatchesPeer(s.SessionName) {
			continue
		}
		// 目前只能广播 ipv4 类型的 IP 地址
		err := validate(adv)
		if err != nil {
			return err
		}
		newAdvs[adv.Prefix.String()] = adv
	}

	s.new = newAdvs
	s.cond.Broadcast()

	return nil
}

// internal/bgp/frr/config.go

func debouncer(body func(config *frrConfig) error, reload <-chan reloadEvent, reloadInterval time.Duration, failureRetryInterval time.Duration, l log.Logger) {
	go func() {
		var config *frrConfig
		var timeOut <-chan time.Time
		timerSet := false
		for {
			select {
			case newCfg, ok := <-reload:
				if !ok { // the channel was closed
					return
				}
				if newCfg.useOld && config == nil {  // useOld 字段由配置的定时验证方法进行设置，若配置出现任何问题，则该字段为 true
					continue
				}
				if !newCfg.useOld && reflect.DeepEqual(newCfg.config, config) {  // 忽略配置不变的请求
					continue
				}
				if !newCfg.useOld {
					config = newCfg.config  // 压缩配置的方法很粗暴，就是直接使用该时间段内最新的配置
				}
				if !timerSet {  // 设置等待时间
					timeOut = time.After(reloadInterval)
					timerSet = true
				}
			case <-timeOut:
				err := body(config)  // 写入 FRR 配置文件
				if err != nil {  // 若出现错误则进行重试
					timeOut = time.After(failureRetryInterval)  // 重试间隔 5s，不可配置
					timerSet = true
					continue
				}
				timerSet = false
			}
		}
	}()
}

// internal/bgp/frr/frr.go

func (sm *sessionManager) SyncBFDProfiles(profiles map[string]*metallbconfig.BFDProfile) error {
	sm.Lock()
	defer sm.Unlock()
	sm.bfdProfiles = make([]BFDProfile, 0)
	for _, p := range profiles {
		frrProfile := configBFDProfileToFRR(p)  // CR 翻译为 FRR 配置
		sm.bfdProfiles = append(sm.bfdProfiles, *frrProfile)
	}
	sort.Slice(sm.bfdProfiles, func(i, j int) bool {
		return sm.bfdProfiles[i].Name < sm.bfdProfiles[j].Name
	})

	frrConfig, err := sm.createConfig()  // 根据当前 manager 的状态生成一份最新的配置文件
	sm.reloadConfig <- reloadEvent{config: frrConfig}  // 写入配置通道，之后完成写入配置
	return nil
}