ipam
南风未起 2025-12-11 19:39:22 cni
# 1.分配
# 1.1.cmdAdd
ipam.cmdAdd()用于容器网络地址分配，支持点名IP分配及未点名自动分配，双栈场景分配异常会触发回滚，避免部分成功造成的IP泄漏问题。func cmdAdd(args *skel.CmdArgs) error { ... // 1.加载配置 json.Unmarshal(args.StdinData, &conf) ... // 2.获取nodeName(配置/默认位置加载) nodename := utils.DetermineNodename(conf) ... // 3.生成wep身份对象 epIDs, err := utils.GetIdentifiers(args, nodename) ... // 4.计算wep对象名称——{node_name}-k8s-{strings.replace(pod_name, "-", "--")}-{wepIDs.Endpoint} epIDs.WEPName, err = epIDs.CalculateWorkloadEndpointName(false) ... // 5.唯一句柄(netName.containerID) handleID := utils.GetHandleID(conf.Name, args.ContainerID, epIDs.WEPName) ... // 6.加载IPAM配置 cnitypes.LoadArgs(args.Args, &ipamArgs) ... // 7.点名IP if ipamArgs.IP != nil { ... // 7.1.基于点名IP分配 assignIPWithLock := func() error { unlock := acquireIPAMLockBestEffort(conf.IPAMLockFile) defer unlock() return calicoClient.IPAM().AssignIP(ctx, assignArgs) } assignIPWithLock() ... // 7.2.回填结果 if ipamArgs.IP.To4() == nil { // It's an IPv6 address. ipNetwork = net.IPNet{IP: ipamArgs.IP, Mask: net.CIDRMask(128, 128)} r.IPs = append(r.IPs, &cniv1.IPConfig{ Address: ipNetwork }) } else { // It's an IPv4 address. ipNetwork = net.IPNet{IP: ipamArgs.IP, Mask: net.CIDRMask(32, 32)} r.IPs = append(r.IPs, &cniv1.IPConfig{ Address: ipNetwork }) } // 8.未点名IP } else { ... // 8.1.IP池加载(Pod注解定义-->ipam配置) v4pools, err := utils.ResolvePools(ctx, calicoClient, conf.IPAM.IPv4Pools, true) ... v6pools, err := utils.ResolvePools(ctx, calicoClient, conf.IPAM.IPv6Pools, false) ... // 8.2.分配参数初始化 assignArgs := ipam.AutoAssignArgs{ ... } ... // 8.3.开始自动分配 autoAssignWithLock := func(...) (*ipam.IPAMAssignments, *ipam.IPAMAssignments, error) { unlock := acquireIPAMLockBestEffort(conf.IPAMLockFile) defer unlock() return calicoClient.IPAM().AutoAssign(ctx, assignArgs) } v4Assignments, v6Assignments, err := autoAssignWithLock(calicoClient, ctx, assignArgs) ... // 8.4.双栈IPV4异常,回退IPV6地址 if num4 == 1 && v4Assignments != nil && len(v4Assignments.IPs) < num4 { if num6 == 1 && v6Assignments != nil && len(v6Assignments.IPs) > 0 { ... // Free the assigned IPv6 addresses when v4 address assignment fails. for _, v6 := range v6Assignments.IPs { v6IPs = append(v6IPs, ipam.ReleaseOptions{Address: v6.IP.String()}) } calicoClient.IPAM().ReleaseIPs(ctx, v6IPs...) ... } } // 8.5.双栈IPV6异常,回退IPV4地址 if num6 == 1 && v6Assignments != nil && len(v6Assignments.IPs) < num6 { if num4 == 1 && v4Assignments != nil && len(v4Assignments.IPs) > 0 { ... // Free the assigned IPv4 addresses when v4 address assignment fails. for _, v4 := range v4Assignments.IPs { v4IPs = append(v4IPs, ipam.ReleaseOptions{Address: v4.IP.String()}) } calicoClient.IPAM().ReleaseIPs(ctx, v4IPs...) ... } } // 8.6.回填IPV4结果 if num4 == 1 { ... ipV4Network := net.IPNet{IP: v4Assignments.IPs[0].IP, Mask: v4Assignments.IPs[0].Mask} r.IPs = append(r.IPs, &cniv1.IPConfig{ Address: ipV4Network }) } // 8.7.回填IPV6结果 if num6 == 1 { ... ipV6Network := net.IPNet{IP: v6Assignments.IPs[0].IP, Mask: v6Assignments.IPs[0].Mask} r.IPs = append(r.IPs, &cniv1.IPConfig{ Address: ipV6Network }) } } // Print result to stdout, in the format defined by the requested cniVersion. return cnitypes.PrintResult(r, conf.CNIVersion) }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
注意
calico-ipam分配核心逻辑分布在AssignIP()和AutoAssign()，分别负责点名IP分配和自动分配
# 1.2.assignIP
ipam.assignIP()负责将点名IP分配给容器，基于扩展CRD记录IP占用状态，确保分布式场景下的并发安全、一致性及可回收性。// AssignIP assigns the provided IP address to the provided host. func (c ipamClient) AssignIP(ctx context.Context, args AssignIPArgs) error { // 1.解析hostName hostname, err := decideHostname(args.Hostname) ... // 2.尝试获取IPPool(ETCD/CRD) pool, err := c.blockReaderWriter.getPoolForIP(args.IP, nil) ... // 3.获取或创建IPAMConfig对象(ETCD/CRD) cfg, err := c.GetIPAMConfig(ctx) ... // 4.划分IP所属的pool cidr blockCIDR := getBlockCIDRForAddress(args.IP, pool) for i := 0; i < datastoreRetries; i++ { // 4.1.获取IPAMBlock对象(ETCD/CRD) obj, err := c.blockReaderWriter.queryBlock(ctx, blockCIDR, "") if err != nil { // IPAMBlock对象未创建 if _, ok := err.(cerrors.ErrorResourceDoesNotExist); !ok { return err } // 获取或创建IPAMConfig对象(ETCD/CRD) cfg = c.GetIPAMConfig(ctx) ... // 获取或创建Pending BlockAffinity对象(ETCD/CRD),用于后续占用CIDR pa, err := c.blockReaderWriter.getPendingAffinity(ctx, hostname, blockCIDR) ... // 尝试创建IPAMBlock对象,成功确认BlockAffinity对象,否则回滚删除BlockAffinity对象 obj, err = c.blockReaderWriter.claimAffineBlock(ctx, pa, *cfg, args.HostReservedAttr) ... } // 4.2.解析IPAMBlock block := allocationBlock{obj.Value.(*model.AllocationBlock)} // 4.3.分配IP err = block.assign(cfg.StrictAffinity, args.IP, args.HandleID, args.Attrs, hostname) ... // 4.4.创建或更新IPAMHandle对象,设置占用block IP数量 if args.HandleID != nil { c.incrementHandle(ctx, *args.HandleID, blockCIDR, 1) } // 4.5.CAS更新IPAMBlock对象 _, err = c.blockReaderWriter.updateBlock(ctx, obj) if err != nil { // 别人先修改,重试整个流程 if _, ok := err.(cerrors.ErrorResourceUpdateConflict); ok { continue } // 更新失败,回退IPAMHandle对象的block IP计数 if args.HandleID != nil { c.decrementHandle(ctx, *args.HandleID, blockCIDR, 1, nil) ... } return err } return nil } return errors.New("Max retries hit - excessive concurrent IPAM requests") } func (b *allocationBlock) assign(affinityCheck bool, address cnet.IP, handleID *string, attrs map[string]string, host string) error { // 1.确认当前host允许由IPAMBlock分配IP if affinityCheck && b.Affinity != nil && !hostAffinityMatches(host, b.AllocationBlock) { return errors.New("Block host affinity does not match") } else if b.Affinity == nil { // IPAMBlock未绑定Host,限制亲和检查也不允许分配 if affinityCheck { return fmt.Errorf("Attempt to assign from block %v with no affinity", b.CIDR) } } // 2.IP转换为CIDR内索引 ordinal, err := b.IPToOrdinal(address) ... // 3.设置该位置版本号,避免并发锁定 b.SetSequenceNumberForOrdinal(ordinal) // 4.IP已经分配 if b.Allocations[ordinal] != nil { return cerrors.ErrorAlreadyExists{Err: "Address already assigned in block", Identifier:address.String()} } // 5.关联熟悉及标记占用 attrIndex := b.findOrAddAttribute(handleID, attrs) b.Allocations[ordinal] = &attrIndex // 6.由空闲列表释放IP for i, unallocated := range b.Unallocated { if unallocated == ordinal { b.Unallocated = append(b.Unallocated[:i], b.Unallocated[i+1:]...) break } } return nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
注意
相比
IP自动分配，点名IP分配简单一些，无需先查找可用IP
# 1.3.autoAssign
ipam.AutoAssign()基于节点可用的IPAMBlock自动分配IP及记录分配结果，分配的IP会更新到相关对象标记占用，避免重复分配。// AutoAssign automatically assigns one or more IP addresses as specified by the provided AutoAssignArgs. func (c ipamClient) AutoAssign(...) (*IPAMAssignments, *IPAMAssignments, error) { // 1.解析hostName hostname, err := decideHostname(args.Hostname) ... // 2.IPV4地址分配 if args.Num4 != 0 { // Assign IPv4 addresses. v4ia, err = c.autoAssign(ctx, args.Num4, args.HandleID, args.Attrs, args.IPv4Pools, 4, hostname, args.MaxBlocksPerHost, args.HostReservedAttrIPv4s, args.IntendedUse) ... } // 3.IPV6地址分配 if args.Num6 != 0 { // If no err assigning V4, try to assign any V6. v6ia, err = c.autoAssign(ctx, args.Num6, args.HandleID, args.Attrs, args.IPv6Pools, 6, hostname, args.MaxBlocksPerHost, args.HostReservedAttrIPv6s, args.IntendedUse) ... } return v4ia, v6ia, nil } func (c ipamClient) autoAssign(...) (*IPAMAssignments, error) { ... // 1.查询IPReservation设置的保留地址 reservations, err := c.getReservedIPs(ctx) ... // 2.node可用的BlockAffinity及用途匹配的所有IPPool pools, affBlocks, err := c.prepareAffinityBlocksForHost(ctx, requestedPools, version, host, rsvdAttr, use) ... // 3.获取IPAMConfig对象 config, err := c.GetIPAMConfig(ctx) ... // 4.合并节点Block申请限制 if config.MaxBlocksPerHost > 0 && maxNumBlocks > 0 && maxNumBlocks > config.MaxBlocksPerHost { // The global config is more restrictive, so use it instead. maxNumBlocks = config.MaxBlocksPerHost } else if maxNumBlocks == 0 { // No per-request value, so use the global one. maxNumBlocks = config.MaxBlocksPerHost } if maxNumBlocks == 0 { maxNumBlocks = 20 } ... // 5.IP分配 for len(ia.IPs) < num { ... // 5.1.明确剩余待分配IP数量 rem := num - len(ia.IPs) // 5.2.明确节点是否需申请新的IPAMBlock if maxNumBlocks > 0 && numBlocksOwned >= maxNumBlocks { s.allowNewClaim = false } // 5.3.匹配或申请IPAMBlock // a) nodeBlockAffinity-->IPAMBlock // b) poolUseable-->创建BlockAffinity-->创建IPAMBlock b, newlyClaimed, err := s.findOrClaimBlock(ctx, 1) ... // 5.4.基于Block分配IP,发生冲突CAS获取IPAMBlock最新状态 for i := 0; i < datastoreRetries; i++ { newIPs := c.assignFromExistingBlock(ctx, b, rem, handleID, attrs, host, config.StrictAffinity, reservations) ... ia.IPs = append(ia.IPs, newIPs...) rem = num - len(ia.IPs) break } } // 6.兜底分配(通常禁用IP租借行为) rem := num - len(ia.IPs) // 允许向其它节点IPAMBlock分配IP if config.StrictAffinity != true && rem != 0 { ... for _, p := range pools { // 6.1.生成随机CIDR生成器 newBlockCIDR := randomBlockGenerator(p, host) for rem > 0 { // 获取一个Block blockCIDR := newBlockCIDR() // 无可用Block if blockCIDR == nil { exhaustedPools = append(exhaustedPools, p.Spec.CIDR) break } // 跳过保留的Block if reservations.MatchesWholeCIDR(blockCIDR) { continue } // 查询对应IPAMBlock for i := 0; i < datastoreRetries; i++ { b, err := c.blockReaderWriter.queryBlock(ctx, *blockCIDR, "") ... // 基于IPAMBlock分配IP newIPs := c.assignFromExistingBlock(ctx, b, rem, handleID, attrs, host, false, reservations) ... ia.IPs = append(ia.IPs, newIPs...) rem = num - len(ia.IPs) break } } } ... } return ia, nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
注意
IPAM自动地址分配支持基于nodeBlock分配及otherNodeBlock借用两种模式，后者一般不会启用
# 1.4.randomBlock
randomBlockGenerator()会基于选中的IPPool利用BlockSize循环切割cidr，直至走完一轮分配，用于获取可用的cidr供IP分配。// Returns a generator that, when called, returns a random block from the given pool. func randomBlockGenerator(ipPool v3.IPPool, hostName string) func() *cnet.IPNet { // 1.解析poolCidr pool, err := cnet.ParseCIDR(ipPool.Spec.CIDR) ... // 2.pool覆盖IP数量 ones, size := pool.Mask.Size() numIP := new(big.Int).Exp(big.NewInt(2), big.NewInt(int64(size-ones)), nil) // 3.pool覆盖的CIDR数量 ones, size = blockMask.Size() blockSize := new(big.Int).Exp(big.NewInt(2), big.NewInt(int64(size-ones)), nil) // 4.不同CIDR覆盖IP数量 numBlocks := new(big.Int) numBlocks.Div(numIP, blockSize) ... // 5.生成随机起点 initialIndex.Rand(randm, numBlocks) i := initialIndex ... // 6.CIDR划分回调 return func() *cnet.IPNet { // 6.1.起始IP ip := cnet.IncrementIP(baseIP, big.NewInt(0).Mul(i, blockSize)) ... // 6.2.构造CIDR Block ipnet := net.IPNet{IP: ip.IP, Mask: blockMask} // 6.3.末尾检查 numDiff.Sub(numBlocks, i) // 6.4.走到末尾,循环到开头 if numDiff.Cmp(big.NewInt(1)) <= 0 { i = big.NewInt(0) } else { // Increment to the next block i.Add(i, big.NewInt(1)) } // 6.5.CIDR Block遍历完 if numReturned.Cmp(numBlocks) >= 0 { return nil } numReturned.Add(numReturned, big.NewInt(1)) // Return the block from this pool that corresponds with the index. return &cnet.IPNet{IPNet: ipnet} } }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
注意
cidr基于Pool和Blocksize切割，可拆分的网段是相对明确固定的
# 1.5.assignBlock
ipam.assignFromExistingBlock()负责亲和检查、IP分配及更新相关对象的状态，分配完成会CAS写回IPAMBlock对象数据。func (c ipamClient) assignFromExistingBlock(...) ([]net.IPNet, error) { // 1.IPAMBlock对应CIDR blockCIDR := block.Key.(model.BlockKey).CIDR // 2.类型断言 b := allocationBlock{block.Value.(*model.AllocationBlock)} // 3.IP分配 ips, err := b.autoAssign(num, handleID, host, attrs, affCheck, reservations) ... if len(ips) == 0 { return []net.IPNet{}, nil } // 4.分配完成,更新IPAMHandle计数 if handleID != nil { c.incrementHandle(ctx, *handleID, blockCIDR, num) } defer func() { // 5.出现异常回滚IPAMHandle计数 if err != nil && handleID != nil { c.decrementHandle(ctx, *handleID, blockCIDR, num, nil) } } // 更新IPAMBlock对象 block.Value = b.AllocationBlock _, err = c.blockReaderWriter.updateBlock(ctx, block) ... return ips, nil } func (b *allocationBlock) autoAssign(...) ([]cnet.IPNet, error) { // 1.亲和检查 if checkAffinityWithHost(affinityCheck,host,b.AllocationBlock) { // Affinity check is enabled but the host does not match - error. return nil, errors.New(s) } ... // 2.未分配IP的索引 for idx, ordinal := range b.Unallocated { // 2.1.分配完成 if len(ips) >= num { // Got enough IPs, finish copying the remaining ordinals. updatedUnallocated = append(updatedUnallocated, b.Unallocated[idx:]...) break } // 2.2.索引解析为IP addr := b.OrdinalToIP(ordinal) // 2.3.属于保留地址 if reservations.MatchesIP(addr) { updatedUnallocated = append(updatedUnallocated, ordinal) continue } // 2.4.关联属性标记占用 if attrIndexPtr == nil { attrIndex := b.findOrAddAttribute(handleID, attrs) attrIndexPtr = &attrIndex } b.Allocations[ordinal] = attrIndexPtr // 2.5.分配结果 ipNet := *mask ipNet.IP = addr.IP ips = append(ips, ipNet) // 2.6.设置对应版本号 b.SetSequenceNumberForOrdinal(ordinal) continue } b.Unallocated = updatedUnallocated return ips, nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
注意
IPAMBlock对象生成会初始化分配索引列表，IP分配就是占用或释放索引到对应列表进行记录
# 2.回收
# 2.1.cmdDel
ipam.cmdDel()用于释放分配的IP，回退IPAMHandle对象计数，更新IPAMBlock状态，必要时回收IPAMBlock及BlockAffinity对象。func cmdDel(args *skel.CmdArgs) error { // 1.加载配置 json.Unmarshal(args.StdinData, &conf) ... // 2.解析节点名 nodename := utils.DetermineNodename(conf) // 3.构造WEP身份对象 epIDs, err := utils.GetIdentifiers(args, nodename) ... // 4.计算WEP对象名称——{node_name}-k8s-{strings.replace(pod_name, "-", "--")}-{wepIDs.Endpoint} epIDs.WEPName, err = epIDs.CalculateWorkloadEndpointName(false) ... // 5.生成HandleID——netName.containerID handleID := utils.GetHandleID(conf.Name, args.ContainerID, epIDs.WEPName) ... // 6.释放IP calicoClient.IPAM().ReleaseByHandle(ctx, handleID) ... return nil } // ReleaseByHandle releases all IP addresses that have been assigned using the provided handle. func (c ipamClient) ReleaseByHandle(ctx context.Context, handleID string) error { // 1.获取IPAMHandle对象 obj, err := c.blockReaderWriter.queryHandle(ctx, handleID, "") ... // 2.基于handle.block关联IPAMBlock释放IP handle := allocationHandle{obj.Value.(*model.IPAMHandle)} for blockStr := range handle.Block { _, blockCIDR, _ := net.ParseCIDR(blockStr) c.releaseByHandle(ctx, *blockCIDR, ReleaseOptions{Handle: handleID}) ... } return nil } func (c ipamClient) releaseByHandle(ctx context.Context, blockCIDR net.IPNet, opts ReleaseOptions) error { for i := 0; i < datastoreRetries; i++ { // 1.获取IPAMBlock obj, err := c.blockReaderWriter.queryBlock(ctx, blockCIDR, "") ... // 2.释放IPAMBlock分配的IP block := allocationBlock{obj.Value.(*model.AllocationBlock)} num := block.releaseByHandle(opts) if num == 0 { // Block has no addresses with this handle, so // all addresses are already unallocated. return nil } // 3.IP均回收+没有关联的BlockAffinity if block.empty() && block.Affinity == nil { // 清理IPAMBlock对象 c.blockReaderWriter.deleteBlock(ctx, obj) ... // 4.更新IPAMBlock对象 } else { c.blockReaderWriter.updateBlock(ctx, obj) ... } // 5.更新IPAMHandle计数 c.decrementHandle(ctx, opts.Handle, blockCIDR, num, nil) ... // Determine whether or not the block's pool still matches the node. c.ensureConsistentAffinity(ctx, block.AllocationBlock) ... return nil } return errors.New("Hit max retries") }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
注意
IP释放会基于IPAMHandle、BlockAffinity、IPAMBlock及IPPool记录的状态进行网段索引释放，必要时回收过期或脏对象