ipsetMgr

1.简介

1.1.原理

• ipset是Linux提供的高效数据存储和查询工具，配合iptables可以将大量IP/Addr组织为集合及快速匹配，避免逐条匹配的性能开销。

• --- 优势
      1.ipset基于哈希表或位图实现,查询和操作的时间复杂度接近O(1)
      2.无需加载iptables规则就可以动态修改集合内容
      3.支持单个IP地址、子网和IP:Port组合的多种数据格式
      4.iptables规则可以引用整个集合,简化配置
  

• 注意

  ipset可以高效管理规则，不过规则隐藏，增大了网络流量异常排查的复杂度

1.2.类型

• ipset支持多种组合，涉及IP、CIDR、MAC、IP:Port及Set类型，Set类型可以存储其它的集合数据，作为嵌套集合管理ipset规则。

• 集合类型	描述	示例
  Hash:IP	IP地址存储	192.168.1.1
  Hash:Net	子网存储	192.168.1.0/24
  Hash:IP,Port	IP:Port组合存储	192.168.1.1,tcp:80
  Hash:Mac	Mac地址存储	00:1A:2B:3C:4D:5E
  List:Set	存储其它集合，作为嵌套集合	子集合名称

• 注意

  ipset集合的规则会基于iptables规则引用，实现网络流量拦截及管理

1.3.定义

• ipsets维护的是待处理规则的内存状态，支持后续批量更新、删除及同步ipset rule至dataplane，管理或替换cni相关的ipset。

• // IPSets manages a whole "plane" of IP sets, i.e. all the IPv4 sets, or all the IPv6 IP sets.
  type IPSets struct {
  	...
  
      // ID-->ipset映射
  	ipSetIDToIPSet       map[string]*ipSet
      // Name-->ipset映射
  	mainIPSetNameToIPSet map[string]*ipSet
  
      // ipset name集合
  	existingIPSetNames set.Set[string]
      ...
  
  	// 修改过的ipset id
  	dirtyIPSetIDs  set.Set[string]
      // 全量重建标识
  	resyncRequired bool
  
  	// 待删除的临时ipset
  	pendingTempIPSetDeletions set.Set[string]
  	// 待删除的ipset
  	pendingIPSetDeletions set.Set[string]
  
  	// ipset cli执行器
  	newCmd cmdFactory
  	...
  
  	// ipset restore结果捕获
  	restoreInCopy bytes.Buffer
  	stdoutCopy bytes.Buffer
  	stderrCopy bytes.Buffer
  	...
  
      // 白名单(点名处理的ipset name)
  	neededIPSetNames set.Set[string]
  }
  
  func NewIPSets(ipVersionConfig *IPVersionConfig, recorder logutils.OpRecorder) *IPSets {
  	return NewIPSetsWithShims(
  		ipVersionConfig,
  		recorder,
  		newRealCmd,
  		time.Sleep,
  	)
  }
  
  // NewIPSetsWithShims is an internal test constructor.
  func NewIPSetsWithShims(*IPVersionConfig, logutils.OpRecorder, cmdFactory, func(time.Duration)) *IPSets {
  	familyStr := string(ipVersionConfig.Family)
  	return &IPSets{
  		IPVersionConfig: ipVersionConfig,
  		ipSetIDToIPSet:       map[string]*ipSet{},
  		mainIPSetNameToIPSet: map[string]*ipSet{},
  		dirtyIPSetIDs:             set.New[string](),
  		pendingTempIPSetDeletions: set.New[string](),
  		pendingIPSetDeletions:     set.New[string](),
  		newCmd:                    cmdFactory,
  		sleep:                     sleep,
  		existingIPSetNames:        set.New[string](),
  		resyncRequired:            true,
  		...
  	}
  }
  

• 注意

  上游推送的ipset会和内存状态对比，暂存批量下发到Linux内核

2.应用

1.4.apply

• s.ApplyUpdates()基于退避重试执行ipset状态同步及修复，利用内存期望状态不断尝试将ipset变更应用到Linux内核，必要时触发全量同步。

• func (s *IPSets) ApplyUpdates() {
  	success := false
      // 退避时间(1ms,2^10ms)
  	retryDelay := 1 * time.Millisecond
  	backOff := func() {
  		s.sleep(retryDelay)
  		retryDelay *= 2
  	}
      
      // 重试10次
  	for attempt := 0; attempt < 10; attempt++ {
  		...
          
          // 强制同步(queueSync/tryUpdate失败)
  		if s.resyncRequired {
              ...
              
  			// 对比期望及更新差异
  			numProblems, err := s.tryResync()
              // 失败延迟重试
  			if err != nil {
  				backOff()
  				continue
  			}
  			...
              
              // 重置强制同步标记
  			s.resyncRequired = false
  		}
  
          // 临时ipset清理
  		numTempSets := s.pendingTempIPSetDeletions.Len()
  		if numTempSets > 0 {
  			s.tryTempIPSetDeletions()
  		}
  
          // dirty ipset更新到底层
  		if err := s.tryUpdates(); err != nil {
  			// 失败延迟重试
  			s.resyncRequired = true
  			...
  			backOff()
  			continue
  		}
  
  		success = true
  		break
  	}
  	...
  }
  

• 

• 注意

  s.tryResync()会执行tryUpdates()失败或queueSync()标记s.resyncRequired参数后执行

1.5.resync

• s.tryResync()会执行ipset list获取内核ipset规则，基于内存期望状态进行对齐，相关的差异会修复更新到内存，确保内存数据一致性。

• // tryResync attempts to bring our state into sync with the dataplane.  It scans the contents of the
  // IP sets in the dataplane and queues up updates to any IP sets that are out-of-sync.
  func (s *IPSets) tryResync() (numProblems int, err error) {
  	...
      
      // 流式执行ipset list
  	cmd := s.newCmd("ipset", "list")
  	// Grab stdout as a pipe so we can stream through the (potentially very large) output.
  	out, err := cmd.StdoutPipe()
  	...
  	// Capture error output into a buffer.
  	cmd.SetStderr(&stderr)
      ...
  	cmd.Start()
  	...
      
  	// 清空过期ipsetName
  	s.existingIPSetNames.Clear()
      
  	// Use a scanner to chunk the input into lines.
  	scanner := bufio.NewScanner(out)
  	ipSetName := ""
  
  	// 逐行解析ipset list结果
      // 	Name: test-100
      //	Type: hash:ip
      //	Revision: 4
      //	Header: family inet hashsize 1024 maxelem 65536
      //	Size in memory: 224
      //	References: 0
      //	Members:
      //	10.0.0.2
      //	10.0.0.1
  	for scanner.Scan() {
  		line := scanner.Text()
          
          // ipsetName登记
  		if strings.HasPrefix(line, "Name:") {
  			ipSetName = strings.Split(line, " ")[1]
  			s.existingIPSetNames.Add(ipSetName)
  		}
          
  		if strings.HasPrefix(line, "Members:") {
  			// 本节点接管的ipset
  			ipSet := s.mainIPSetNameToIPSet[ipSetName]
              
              // ipset无法识别
  			if ipSet == nil || ipSet.members == nil {
  				
                  // skip members内容
  				for scanner.Scan() {
  					line := scanner.Bytes()
  					if len(line) == 0 {
  						// End of members
  						break
  					}
  				}
  				ipSetName = ""
  				continue
  			}
              
              // 初始化ipset member集合
  			dataplaneMembers := set.NewBoxed[IPSetMember]()
  			for scanner.Scan() {
  				line := scanner.Text()
  				if line == "" {
  					// End of members
  					break
  				}
                  // member提取及存入集合
  				canonMember := ipSet.Type.CanonicaliseMember(line)
  				dataplaneMembers.Add(canonMember)
  			}
              ...
  
  			// 基于内存状态对齐期望
  			ipSet.members.Iter(func(m IPSetMember) error {
                  // ipset位于内存和内核
  				if dataplaneMembers.Contains(m) {
  					// 剔除差异
  					dataplaneMembers.Discard(m)
  					return nil
  				}
  
                  // ipset待删除,内核未找到
  				if ipSet.pendingDeletions.Contains(m) {
  					// 剔除删除差异,说明删除完成
  					ipSet.pendingDeletions.Discard(m)
  					return set.RemoveItem
  				}
  				...
                  
                  // 追加内核待补充ipset
  				s.dirtyIPSetIDs.Add(ipSet.SetID)
  				ipSet.pendingAdds.Add(m)
  				return set.RemoveItem
  			})
  
  			// 基于内核状态对齐期望
  			dataplaneMembers.Iter(func(m IPSetMember) error {
  				// 内核多出ipset同步到内存
  				ipSet.members.Add(m)
  
                  // ipset已标记追加内核
  				if ipSet.pendingAdds.Contains(m) {
  					// 剔除差异,内核已存在ipset
  					ipSet.pendingAdds.Discard(m)
  					return nil
  				}
  				...
  
                  // 内核多出ipset注册到待删除队列
  				s.dirtyIPSetIDs.Add(ipSet.SetID)
  				ipSet.pendingDeletions.Add(m)
  				return nil
  			})
  		}
  	}
      ...
      
  	cmd.Wait()
  	...
  
      // orphan ipset清理
      // 遍历所有已知ipset(dataplane-->onUpdate-->addOrReplace-->s.ipSetIDToIPSet)
  	expectedIPSets := set.NewBoxed[string]()
  	for _, ipSet := range s.ipSetIDToIPSet {
          // 过滤未允许的ipset(配置指定使用的保留)
  		if !s.ipSetNeeded(ipSet.SetID) {
  			continue
  		}
          // 注册到期望集合
  		expectedIPSets.Add(ipSet.MainIPSetName)
  	}
  
  	// 待清理ipset加入期望集合
  	s.pendingIPSetDeletions.Iter(func(item string) error {
  		expectedIPSets.Add(item)
  		return nil
  	})
  
  	// 扫描内核ipset
  	s.existingIPSetNames.Iter(func(setName string) error {
          // ipsetName前缀未匹配,无法接管
  		if !s.IPVersionConfig.OwnsIPSet(setName) {
  			return nil
  		}
          
          // 期望集合注册过
  		if expectedIPSets.Contains(setName) {
  			return nil
  		}
          
          // 期望集合未注册的ipset都是待删除的
          // 临时ipset删除登记
  		if s.IPVersionConfig.IsTempIPSetName(setName) {
  			s.pendingTempIPSetDeletions.Add(setName)
  		}
          
          // 其余ipset删除登记
  		s.pendingIPSetDeletions.Add(setName)
  		return nil
  	})
  
  	return
  }
  

• 

• 注意

  s.tryResync()会获取内核ipset，基于内存已知状态进行对比，将内核待追加或待删除的数据进行对齐

1.6.update

• s.tryUpdates()基于ipset restore进行状态同步，用于将内存记录的ipset变更批量同步到Linux内核，减少进程fork，确保规则一致性。

• // attempts to do the updates as a single 'ipset restore' session in order to minimise process forking overhead.
  func (s *IPSets) tryUpdates() error {
  	needUpdates := false
      // 未限制ipsetName
  	if s.neededIPSetNames == nil {
          // 有变更就标记更新
  		needUpdates = s.dirtyIPSetIDs.Len() > 0
      // 限制ipsetName
  	} else {
          // 变更必须发生在限制集合
  		s.dirtyIPSetIDs.Iter(func(setID string) error {
  			if s.ipSetNeeded(setID) {
  				needUpdates = true
  				return set.StopIteration
  			}
  			return nil
  		})
  	}
      
  	if !needUpdates {
  		return nil
  	}
  	...
  
  	// Set up an ipset restore session.
  	cmd := s.newCmd("ipset", "restore")
  	// Get the pipe for stdin.
  	rawStdin, err := cmd.StdinPipe()
  	...
      // "Tee" the data that we write to stdin to a buffer so we can dump it to the log on failure.
  	stdin := io.MultiWriter(&s.restoreInCopy, rawStdin)
      ...
      
  	// 执行ipset restore命令
  	cmd.Start()
  	...
  
  	// 扫描dirty ipset
  	s.dirtyIPSetIDs.Iter(func(setID string) error {
          // 检查更新条件
  		if !s.ipSetNeeded(setID) {
  			return nil
  		}
          
          // ipset转为ipset restore line写入stdin
  		ipSet := s.ipSetIDToIPSet[setID]
  		s.writeUpdates(ipSet, stdin)
  		...
          
  		return nil
  	})
      
  	// 提交restore内容
  	stdin.Write([]byte("COMMIT\n"))
  	...
  
  	// 扫描dirty ipset同步到内存状态
  	s.dirtyIPSetIDs.Iter(func(setID string) error {
          // 更新条件检查
  		if !s.ipSetNeeded(setID) {
  			return nil
  		}
  		ipSet := s.ipSetIDToIPSet[setID]
          
          // replace模式
  		if ipSet.pendingReplace != nil {
              // member更新
  			ipSet.members = ipSet.pendingReplace
  			ipSet.pendingReplace = nil
  			// 标记存在
  			s.existingIPSetNames.Add(ipSet.MainIPSetName)
          // patch模式
  		} else {
              // 待追加ipset同步到member
  			ipSet.pendingAdds.Iter(func(m IPSetMember) error {
  				ipSet.members.Add(m)
  				return set.RemoveItem
  			})
              // 待清理ipset由member剔除
  			ipSet.pendingDeletions.Iter(func(m IPSetMember) error {
  				ipSet.members.Discard(m)
  				return set.RemoveItem
  			})
  		}
  		return set.RemoveItem
  	})
  
  	return nil
  }
  
  func (s *IPSets) writeUpdates(ipSet *ipSet, w io.Writer) error {
  	...
  
      // ipset member更新
  	if ipSet.pendingReplace == nil {
  		// 增量更新member
  		if ipSet.pendingAdds.Len() == 0 && ipSet.pendingDeletions.Len() == 0 {
  			// We hit this case if an IP is added, then removed before we actually
  			// write it, nothing to do.
  			return nil
  		}
          
          // del blacklist 1.1.1.1 --exist
  		// add blacklist 3.3.3.3
  		return s.writeDeltas(ipSet, w, logCxt)
  	}
      
      // ipset创建
      // create blacklist hash:ip family inet maxelem 65536
      // create tmp_set_1234 hash:ip family inet maxelem 65536
      // add tmp_set_1234 10.0.0.1
      // add tmp_set_1234 10.0.0.2
      // add tmp_set_1234 10.0.0.3
      // swap blacklist tmp_set_1234
      // destroy tmp_set_1234
  	return s.writeFullRewrite(ipSet, w, logCxt)
  }
  

• 

• 注意

  ipset变更会基于restore重建ipset，member变更基于增量数据restore调整ipset规则

1.5.deletions

• s.tryTempIPSetDeletions()和s.ApplyDeletions()主要负责清理过期及临时ipset，确保内核和内存ipset规则的状态一致性。

• // tryTempIPSetDeletions tries to delete any temporary IP sets found by the last resync.
  func (s *IPSets) tryTempIPSetDeletions() {
      // 临时ipset删除
  	s.pendingTempIPSetDeletions.Iter(func(setName string) error {
  		// ipset还存在
  		if s.existingIPSetNames.Contains(setName) {
  			// ipset destroy清理及更新内存状态
  			s.deleteIPSet(setName)
              ...
              
              // 剔除差异
              s.pendingIPSetDeletions.Discard(setName)
  		}
  		// Always remove the item so we don't retry until the next timed resync.
  		return set.RemoveItem
  	})
  }
  
  // ApplyDeletions tries to delete any IP sets that are no longer needed.
  func (s *IPSets) ApplyDeletions() {
  	s.pendingIPSetDeletions.Iter(func(setName string) error {
          // ipset还存在
  		if s.existingIPSetNames.Contains(setName) {
              // ipset destroy清理及更新内存状态
  			s.deleteIPSet(setName)
              ...
  		}
  		// Always remove the item so we don't retry until the next timed resync.
  		return set.RemoveItem
  	})
  	...
  }
  
  func (s *IPSets) deleteIPSet(setName string) error {
  	cmd := s.newCmd("ipset", "destroy", string(setName))
  	cmd.CombinedOutput()
      ...
      
  	// update the cache.
  	s.existingIPSetNames.Discard(setName)
      
      // ipset处于内存期望状态
  	if ipSet := s.mainIPSetNameToIPSet[setName]; ipSet != nil {
  		// 重建pendingReplace
  		if ipSet.pendingReplace == nil {
  			// member收集
  			ipSet.pendingReplace = ipSet.members
  			ipSet.members = nil
  			ipSet.pendingAdds.Iter(func(m IPSetMember) error {
  				ipSet.pendingReplace.Add(m)
  				return set.RemoveItem
  			})
  			ipSet.pendingDeletions.Iter(func(m IPSetMember) error {
  				ipSet.pendingReplace.Discard(m)
  				return set.RemoveItem
  			})
  		}
  	}
  	return nil
  }
  

• 

• 注意

  pendingReplace重建是为了避免内存期望ipset误删，重建后tryUpdate会生成ipset对应规则进行修复