flannel
南风未起 2025-07-22 19:39:22 cni
# 1.简介
# 1.1.flannel
flannel是coreos团队设计的一个网络规划服务，负责划分节点网段及存储网络配置，确保集群中的不同节点运行的pod具有唯一的IP地址。简单来说，flannel会创建一个大型内部网络，各节点基于独立网段分配内部IP，不同节点基于路由封包通信。
注意
flannel本质上是一种L3 overlay网络，基于原始数据包封包至其它网络包进行路由转发和通信
# 1.2.架构
flannel设计分为两部分，运行于节点的flanneld(pod)负责子网分配、路由配置及网络封装，可执行程序flannel plugin负责交互运行时提供容器网络管理能力。
设计
1.
flanneld以pod运行于各节点，以分配节点子网及同步路由，部分工作模式下参与网络封装2.
flannel plugin作为可执行文件位于/opt/cni/bin，供运行时创建pause容器分配网络调用3.
flannel plugin基于flanneld渲染网络配置渲染委托插件配置，基于委托插件完成容器网络管理核心功能
# 2.工作模式
# 2.1.udp模式
flanneld会激活udp服务(8285)监听来自其它节点的数据包，打开/dev/net/tun设备(flannel0)作为用户空间和内核空间的数据包交互通道。pod跨节点通信时，cni0的流量根据route table路由到flannel0，flannel0基于套接字将数据包交给flanneld封包(IP->udp)，经物理网卡发送到其他节点。
注意
1.发送阶段经历
cni0-->(flannel0-->flanneld)-->eth02.接收阶段经历
eth0-->(flanneld-->flannel0)-->cni03.由于传输过程数据包经历
2次用户态和内核态切换，且flanneld接入封包拆包，网络效率非常低，已被官方弃用
# 2.2.vxlan模式
flanneld会激活vxlan隧道(flannel.1)，用于将跨节点的pod数据包封装在udp数据包进行传输。相比于udp模式，vxlan协议的封包和拆包基于内核态完成，减少了用户态和内核态切换，效率相对来说有所提升。
注意
1.发送阶段经历
cni0-->flannel.1-->eth02.接收阶段经历
eth0-->flannel.1-->cni03.由于
vxlan协议的封包和拆包基于内核态完成，减少用户态和内核态切换，效率相对提高
# 2.3.host-gw模式
flanneld会将主机作为网关实现跨节点网络通信，集群节点的网络必须位于L2直接可达，性能极高。相对的，节点二层网络互通限制了host-gw模式应用，无法适用于节点网段划分场景，集群规模膨胀的路由表动态更新也是不小的压力。
注意
1.发送阶段经历
cni0-->eth02.接收阶段经历
eth0-->cni03.由于将主机作为网关，
host-gw模式无需进行封包和拆包，性能相对较高
# 2.4.ip-ip模式
flanneld会激活IPIP隧道封装和传输容器流量，基于简单的IP封装协议将原始数据包封装在另一个IP数据包，实现容器跨节点通信。与vxlan相比，vxlan会把其它节点的flannel.1设备作为网关，ipip直接将节点作为网关。
注意
1.发送阶段经历
cni0-->flannel.ipip-->eth02.接收阶段经历
eth0-->flannel.ipip-->cni03.数据包的封包和拆包基于内核态完成，性能相比
udp模式好4.
ip-ip属于实验性模式，不推荐使用
# 2.5.ipsec模式
flanneld会激活strongswan的charon进程进行密钥交换，更新节点的安全关联(SA)及动态策略(Policy)。容器流量转发时，基于XFRM策略匹配src和dst，以将数据包送到ESP加密模块处理，经主机网卡发出。
注意
1.发送阶段经历
cni0-->ipsec加密-->eth02.接收阶段经历
eth0-->ipsec解密-->cni03.数据包的加密和解密基于内核完成，策略维护存在一定开销
4.
ipsec属于实验性模式，不推荐使用
# 2.6.wireguard模式
flanneld激活wireGuard网络接口(flannel-wg)以匹配cni0的流量，数据包由wireguard加密经物理网卡发送。wireGuard会点对点连接集群节点，保证流量机密性和完整性。
注意
1.发送阶段经历
cni0-->flannel-wg-->eth02.接收阶段经历
eth0-->flannel-wg-->cni03.数据包的加密和解密基于内核完成
# 3.主流程
# 3.1.start
flannel会初始化subnetManager和backendManager，前者用于交互etcd/API以管理子网，后者用于根据网络配置初始化对应backend以配置网络。func main() { ... // 初始化subnet管理器 sm, err := newSubnetManager(ctx) ... // 获取网络配置(net-conf.json) config, err := getConfig(ctx, sm) ... // 获取默认网卡或指定网卡(节点流量出入) extIface, err = ipmatch.LookupExtIface(opts.publicIP, "", "", ipStack, optsPublicIP) // 初始化backend管理器 bm := backend.NewManager(ctx, sm, extIface) // 加载及缓存backend对象 be, err := bm.GetBackend(config.BackendType) ... // 加载及初始化network对象 bn, err := be.RegisterNetwork(ctx, &wg, config) ... // 获取已分配的IPV4子网信息 if config.EnableIPv4 { flannelIPv4Net, err = config.GetFlannelNetwork(&bn.Lease().Subnet) ... } // 获取已分配的IPV6子网信息 if config.EnableIPv6 { flannelIpv6Net, err = config.GetFlannelIPv6Network(&bn.Lease().IPv6Subnet) ... } // iptables snat规则 if opts.ipMasq { // 读取上一次已分配IPV4子网(subnet.env) prevNetworks := ReadCIDRsFromSubnetFile(opts.subnetFile, "FLANNEL_NETWORK") prevSubnet := ReadCIDRFromSubnetFile(opts.subnetFile, "FLANNEL_SUBNET") // 读取上一次已分配IPV6子网(subnet.env) prevIPv6Networks := ReadIP6CIDRsFromSubnetFile(opts.subnetFile, "FLANNEL_IPV6_NETWORK") prevIPv6Subnet := ReadIP6CIDRFromSubnetFile(opts.subnetFile, "FLANNEL_IPV6_SUBNET") // snat/masq规则写入 err = trafficMngr.SetupAndEnsureMasqRules(flannelIPv4Net, prevSubnet, prevNetworks, flannelIpv6Net, prevIPv6Subnet, prevIPv6Networks, bn.Lease(), opts.iptablesResyncSeconds) ... } // iptables转发规则写入 if opts.iptablesForwardRules { trafficMngr.SetupAndEnsureForwardRules( flannelIPv4Net, flannelIpv6Net, opts.iptablesResyncSeconds) } // subnet信息持久化(/run/flannel/subnet.env) sm.HandleSubnetFile(opts.subnetFile, config, opts.ipMasq, bn.Lease().Subnet, bn.Lease().IPv6Subnet,bn.MTU()) ... // 启动backend对象 go func() { bn.Run(ctx) wg.Done() }() ... // lease续约 sm.CompleteLease(ctx, bn.Lease(), &wg) wg.Wait() os.Exit(0) }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
注意
1.
flannel支持kube subnetManager和etcd subnetManager两种形式，基于启动参数初始化，用于subnet lease管理2.
backendManager基于net-conf.json声明的backendType初始化backend，进一步注册backendNetwork及启动3.
flannel会对比subnet.env和lease子网及主网配置，根据差异更新iptables的masq和forward规则
# 3.2.subnetManager
subnetManager用于监听集群变化以维护网络信息，支持kubeAPI和etcd两种形式。实现上，kubeAPI基于informer监听node和cidr变化，etcd基于原生的watch机制感知node和cidr变化，由backend消费事件以更新程序内部的subnet信息。// initial subnetManager to update. func newSubnetManager(ctx context.Context) (subnet.Manager, error) { // kubeAPI模式(默认) if opts.kubeSubnetMgr { return kube.NewSubnetManager(ctx, opts.kubeApiUrl, opts.kubeConfigFile, opts.kubeAnnotationPrefix, opts.netConfPath, opts.setNodeNetworkUnavailable) } ... // etcd模式 return etcd.NewLocalManager(ctx, cfg, prevSubnet, prevIPv6Subnet, opts.subnetLeaseRenewMargin) }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16etcd subnetManager会注册watch监听subnet变化，根据subnet配置生成lease推送到backend消费处理。// create etcd subnetManager func NewLocalManager(...) (subnet.Manager, error) { // 注册client r, err := newEtcdSubnetRegistry(ctx, config, nil) ... // 构造etcd subnetManager结构体 return newLocalManager(r, prevSubnet, prevIPv6Subnet, subnetLeaseRenewMargin), nil }1
2
3
4
5
6
7
8kube subnetManager会开启nodeInformer及注册eventHandler，根据net-conf.json与node保存的网络配置对比，生成lease事件推送给backend。// subnetManager based on kube mode. func NewSubnetManager(...) (subnet.Manager, error) { ... // 初始化client c, err := clientset.NewForConfig(cfg) ... // 获取nodeName nodeName := os.Getenv("NODE_NAME") if nodeName == "" { // 获取env podName podName := os.Getenv("POD_NAME") // 获取env podNamespace podNamespace := os.Getenv("POD_NAMESPACE") ... // 获取pod pod, err := c.CoreV1().Pods(podNamespace).Get(ctx, podName, metav1.GetOptions{}) ... // 获取nodeName以annotate nodeName = pod.Spec.NodeName ... } // 读取net-conf.json配置 netConf, err := os.ReadFile(netConfPath) ... // 解析subnet配置 sc, err := subnet.ParseConfig(string(netConf)) ... // 初始化基于kube的subnetManager sm, err := newKubeSubnetManager(ctx, c, sc, nodeName, prefix) ... // 标识节点网络不可用 sm.setNodeNetworkUnavailable = setNodeNetworkUnavailable // alloc模式,仅分配IP if sm.disableNodeInformer { log.Infof("Node controller skips sync") } else { ... // 启动subnetManager监听 go sm.Run(ctx) // node资源同步完成 // 1s检查一次直到10min超时 err = wait.Poll(time.Second, nodeControllerSyncTimeout, func() (bool, error) { return sm.nodeController.HasSynced(), nil }) ... } return sm, 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
补充
1.
kubeAPI模式，flannel的数据以annotation形式存在node，执行变更操作以事件形式监听2.
etcd模式，flannel的数据存在KV，网络变更基于watch同步
# 3.3.backend
flannel会将backend构造函数注册到constructorMap，基于backendManager抽象backend初始化过程，加载backend对象及注册。func main() { ... // 加载及缓存backend对象 be, err := bm.GetBackend(config.BackendType) ... // 加载及初始化network对象 bn, err := be.RegisterNetwork(ctx, &wg, config) ... } // 基于backendManager加载及缓存backend对象 func (bm *manager) GetBackend(backendType string) (Backend, error) { ... // net-conf.json定义的类型 betype := strings.ToLower(backendType) // backend已初始化 if be, ok := bm.active[betype]; ok { return be, nil } // 加载backendFunc befunc, ok := constructors[betype] ... // 初始化backend be, err := befunc(bm.sm, bm.extIface) ... // 缓存 bm.active[betype] = be bm.wg.Add(1) go func() { <-bm.ctx.Done() ... // 终止解注册 delete(bm.active, betype) ... bm.wg.Done() }() return be, 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
补充
backendManager维护backend单例对象，backend.RegistryNetwork()构造backendNetwork对象实现网络代理
# 3.4.setupMasq
flannel会维护SNAT规则，以基于iptables管理网络转发的出口伪装规则，保证pod出去访问外部网络时能够正确返回。// msq规则维护 func (iptm IPTablesManager) SetupAndEnsureMasqRules(...) error { // node IPV4网络配置不为空 if flannelIPv4Net.String() != "" { ... // 网络配置发生变化 if prevNetwork != flannelIPv4Net && prevSubnet != currentlease.Subnet { ... // 旧规则回收 iptm.deleteIP4Tables(iptm.masqRules(prevNetworks, newLease)) ... } // 创建新的IPV4 chain iptm.CreateIP4Chain("nat", "FLANNEL-POSTRTG") // 最新规则渲染回调 getRules := func() []trafficmngr.IPTablesRule { return iptm.masqRules([]ip.IP4Net{flannelIPv4Net}, currentlease) } // 5s间隔同步及下发最新masq规则 go iptm.setupAndEnsureIP4Tables(getRules, resyncPeriod) } // node IPV6网络配置不为空 if flannelIPv6Net.String() != "" { ... // 网络配置发生变化,回收旧的iptables masq规则 if prevIPv6Network != flannelIPv6Net && prevIPv6Subnet != currentlease.IPv6Subnet { ... // 旧规则回收 iptm.deleteIP6Tables(iptm.masqIP6Rules(prevIPv6Networks, newLease)) ... } // 创建新的IPV6 chain iptm.CreateIP6Chain("nat", "FLANNEL-POSTRTG") // 最新规则渲染回调 getRules := func() []trafficmngr.IPTablesRule { return iptm.masqIP6Rules([]ip.IP6Net{flannelIPv6Net}, currentlease) } // 5s间隔同步及下发最新masq规则 go iptm.setupAndEnsureIP6Tables(getRules, resyncPeriod) } return nil } // 清理给定旧规则 func (iptm IPTablesManager) deleteIP4Tables(rules []trafficmngr.IPTablesRule) error { // iptables对象 ipt, err := iptables.New() ... // iptables-restore对象 iptRestore, err := NewIPTablesRestoreWithProtocol(iptables.ProtocolIPv4) ... // 删除rules teardownIPTables(ipt, iptRestore, rules) ... return nil } // 刷入最新规则 func (iptm IPTablesManager) setupAndEnsureIP4Tables(...) { // 执行规则渲染 rules := getRules() ... // iptables对象 ipt, err := iptables.New() ... // iptables-restore对象 iptRestore, err := NewIPTablesRestoreWithProtocol(iptables.ProtocolIPv4) ... // 规则构建+iptables-restore --noflush批量写入 err = ipTablesBootstrap(ipt, iptRestore, rules) ... defer func() { // 退出前规则清理 teardownIPTables(ipt, iptRestore, rules) ... }() for { // masq规则检查及同步 ensureIPTables(ipt, iptRestore, getRules()) ... // 间隔5s time.Sleep(time.Duration(resyncPeriod) * time.Second) } }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
补充
flannel维护的masq进行snat，以确保回包SYN/ACL匹配，避免丢弃
# 3.5.setupForward
flannel维护filter forward规则，以限制转发进出flannel网络网段的流量，确保pod<-->外部/节点的数据包能被内核正确转发。func (iptm IPTablesManager) SetupAndEnsureForwardRules(...) { if flannelIPv4Network.String() != "" { // 创建chain iptm.CreateIP4Chain("filter", "FLANNEL-FWD") // 规则渲染 getRules := func() []trafficmngr.IPTablesRule { return iptm.forwardRules(flannelIPv4Network.String()) } // 异步下发 go iptm.setupAndEnsureIP4Tables(getRules, resyncPeriod) } if flannelIPv6Network.String() != "" { // 创建chain iptm.CreateIP6Chain("filter", "FLANNEL-FWD") // 渲染规则 getRules := func() []trafficmngr.IPTablesRule { return iptm.forwardRules(flannelIPv6Network.String()) } // 异步下发 go iptm.setupAndEnsureIP6Tables(getRules, resyncPeriod) } } // forward规则更新 func (iptm IPTablesManager) setupAndEnsureIP4Tables(...) { // 执行规则渲染 rules := getRules() ... // iptables对象 ipt, err := iptables.New() ... // iptables-restore对象 iptRestore, err := NewIPTablesRestoreWithProtocol(iptables.ProtocolIPv4) ... // 规则构建+iptables-restore --noflush批量写入 err = ipTablesBootstrap(ipt, iptRestore, rules) ... defer func() { // 退出前规则清理 teardownIPTables(ipt, iptRestore, rules) ... }() for { // masq规则检查及同步 ensureIPTables(ipt, iptRestore, getRules()) ... // 间隔5s time.Sleep(time.Duration(resyncPeriod) * time.Second) } }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
补充
flannel拦截forward流量，允许pod<-->任意通过
# 3.6.CompleteLease
sm.CompleteLease()会持续监听租约变化，以维护subnet有效状态，ksr.CompleteLease()基于nodeInformer监听，无需续约lease声明子网占用；esr.CompleteLease()维护subnet分配，基于lease有效期标记占用，是续约的主要实现者。// kubeAPI监听 func (ksm *kubeSubnetManager) CompleteLease(ctx context.Context, lease *lease.Lease, wg *sync.WaitGroup) error { ... // subnetManager初始化setNodeNetworkUnavailable=true if !ksm.setNodeNetworkUnavailable { // not set NodeNetworkUnavailable NodeCondition return nil } // 设置node NetworkUnavailableCondition=false condition := v1.NodeCondition{ Type: v1.NodeNetworkUnavailable, Status: v1.ConditionFalse, Reason: "FlannelIsUp", Message: "Flannel is running on this node", LastTransitionTime: metav1.Now(), LastHeartbeatTime: metav1.Now(), } // patch到节点 raw, err := json.Marshal(&[]v1.NodeCondition{condition}) ... patch := []byte(fmt.Sprintf(`{"status":{"conditions":%s}}`, raw)) _, err = ksm.client.CoreV1().Nodes().PatchStatus(ctx, ksm.nodeName, patch) return err }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24esr.CompleteLease()利用watch监听子网变化，驱动esr.reNewLease()基于事件或周期维护subnet状态，避免subnet再分配或篡改。// etcd监听 func (m *LocalManager) CompleteLease(ctx context.Context, myLease *lease.Lease, wg *sync.WaitGroup) error { // Use the subnet manager to start watching leases. evts := make(chan lease.Event) wg.Add(1) // watch子网租约变化 go func() { l := myLease subnet.WatchLease(ctx, m, l.Subnet, l.IPv6Subnet, evts) wg.Done() }() // 续期边界60min renewMargin := time.Duration(m.subnetLeaseRenewMargin) * time.Minute // 执行续约的时间 dur := time.Until(myLease.Expiration) - renewMargin for { select { // lease续约 case <-time.After(dur): // 刷新subnet及关联lease m.RenewLease(ctx, myLease) ... // 计算下一次lease刷新时间 dur = time.Until(myLease.Expiration) - renewMargin // watch Lease case e, ok := <-evts: ... switch e.Type { // ADD case lease.EventAdded: // 更新lease到期时间 myLease.Expiration = e.Lease.Expiration // 计算下一次lease刷新时间 dur = time.Until(myLease.Expiration) - renewMargin ... // DEL事件 case lease.EventRemoved: // 退出监听 return errInterrupted } } } } // etcd lease监听 func WatchLease(ctx context.Context, sm Manager, sn ip.IP4Net, sn6 ip.IP6Net, receiver chan lease.Event) { ... // watch subnet变化构造lease(当前节点) go func() { sm.WatchLease(ctx, sn, sn6, leaseWatchChan) ... }() // lease推送到channel供外部消费 for watchResults := range leaseWatchChan { for _, wr := range watchResults { // 原始租约 if len(wr.Snapshot) > 0 { receiver <- lease.Event{ Type: lease.EventAdded, Lease: wr.Snapshot[0], } // ADD/DEL事件租约 } else if len(wr.Events) > 0 { receiver <- wr.Events[0] } else { log.V(2).Info("WatchLease: empty event received") } } } close(receiver) } // subnet lease刷新 func (m *LocalManager) RenewLease(ctx context.Context, lease *lease.Lease) error { // 更新lease子网配置 exp, err := m.registry.updateSubnet(ctx, lease.Subnet, lease.IPv6Subnet, &lease.Attrs, subnetTTL, 0) ... // 设置过期时间(now+24h) lease.Expiration = exp return nil } // subnet KV刷新 func (esr *etcdSubnetRegistry) updateSubnet(ctx context.Context, sn ip.IP4Net, sn6 ip.IP6Net, attrs *lease.LeaseAttrs, ttl time.Duration, asof int64) (time.Time, error) { ... // 申请24h过期lease lresp, lerr := esr.cli.Grant(ctx, int64(ttl.Seconds())) ... // 更新KV及关联lease _, perr := esr.kv().Put(ctx, key, string(value), etcd.WithLease(lresp.ID)) // 失败 if perr != nil { // 撤销lease申请 _, rerr := esr.cli.Revoke(ctx, lresp.ID) ... return time.Time{}, perr } // 计算lease过期时间 exp := time.Now().Add(time.Duration(lresp.TTL) * time.Second) return exp, 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
补充
1.
kube subnetManager无需维护lease，基于nodeInformer感知subnet状态，仅更新node networkCond可用2.
etcd subnetManager由于基于watch感知subnet变化及有效性，所以会异步watch subnet，关联lease及动态更新
# 4.udp实现
udp模式基于flannel udpServer监听数据包，将数据包封装在udp包内发到目标节点指定端口，目标结果的flannel udpServer会解析数据包传递给容器。
注意
udp模式由于2次切换用户态和内核态效率很低，目前已不推荐使用，这里不再进一步分析
# 5.vxlan实现
# 5.1.原理
vxlan(virtual extensible lan)是一种基于隧道的网络模式，数据包会被封装在udp数据包，以实现跨节点传输，是官方flannel的默认模式。
注意
vxlan封装udp数据包时，原始数据包的MAC地址会替换为源端和对端flannel.1网卡地址，外部包会加入源端和对端节点IP，以传输和解析
# 5.2.registerNetwork
vxlan.RegisterNetwork()会解析vxlan配置，创建flannel.1虚拟网卡及绑定物理网卡，设置虚拟网卡接口地址作为隧道端点，申请subnet租约作为podcidr，最终返回network对象用于后续路由下发。// network注册 func (be *VXLANBackend) RegisterNetwork(...) (backend.Network, error) { ... // 解析vxlan配置 json.Unmarshal(config.Backend, &cfg) ... // 获取flannel.1虚拟网卡MAC地址 macStr := be.subnetMgr.GetStoredMacAddress(ctx) if macStr != "" { hwAddr, err = net.ParseMAC(macStr) ... } // IPV4 vxlan设备创建 if config.EnableIPv4 { ... // 创建vxlan设备 dev, err = newVXLANDevice(&devAttrs) ... // 标记vxlan dr模式 dev.directRouting = cfg.DirectRouting } // IPV6 vxlan设备创建 if config.EnableIPv6 { ... // 创建vxlan设备 v6Dev, err = newVXLANDevice(&v6DevAttrs) ... // 标记vxlan dr模式 v6Dev.directRouting = cfg.DirectRouting } ... // 生成lease对象 lease, err := be.subnetMgr.AcquireLease(ctx, subnetAttrs) ... // vxlan设备接口地址配置 if config.EnableIPv4 { // 获取subnet所属network net, err := config.GetFlannelNetwork(&lease.Subnet) ... // ipv4 vxlan设备接口地址(设置/32类型host-only地址,不会产生广播路由) dev.Configure(ip.IP4Net{IP: lease.Subnet.IP, PrefixLen: 32}, net) ... } if config.EnableIPv6 { // subnet所属network net, err := config.GetFlannelIPv6Network(&lease.IPv6Subnet) ... // ipv6 vxlan设备接口地址(设置/128类型host-only地址,不会产生广播路由) v6Dev.ConfigureIPv6(ip.IP6Net{IP: lease.IPv6Subnet.IP, PrefixLen: 128}, net) ... } return newNetwork(be.subnetMgr, be.extIface, dev, v6Dev, ip.IP4Net{}, lease, cfg.MTU) }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
53etcd subnetManager尝试申请租约以声明subnet占用，优先复用已有的空闲subnet，已有的subnet未被占用就不再划分子网，避免网段浪费。// etcd租约申请 func (m *LocalManager) AcquireLease(ctx context.Context, attrs *lease.LeaseAttrs) (*lease.Lease, error) { // 获取网络配置 config, err := m.GetNetworkConfig(ctx) ... // 重试10次获取 for i := 0; i < raceRetries; i++ { // 尝试申请lease l, err := m.tryAcquireLease(ctx, config, attrs.PublicIP, attrs) switch err { case nil: return l, nil ... } } return nil, errors.New("Max retries reached trying to acquire a subnet") } // lease申请 func (m *LocalManager) tryAcquireLease(...) (*lease.Lease, error) { // 获取已分配的子网租约 leases, _, err := m.registry.getSubnets(ctx) ... // 节点IP已绑定租约(节点已分配子网) if l := findLeaseByIP(leases, extIaddr); l != nil { // lease配置兼容,调整TTL复用 if isSubnetConfigCompat(config, l.Subnet) && isIPv6SubnetConfigCompat(config, l.IPv6Subnet) { ... // 非永久lease设置TTL if !l.Expiration.IsZero() { // 24h过期 ttl = subnetTTL } // 更新subnet及续期关联lease exp, err := m.registry.updateSubnet(ctx, l.Subnet, l.IPv6Subnet, attrs, ttl, 0) ... l.Attrs = *attrs l.Expiration = exp return l, nil // 不兼容 } else { // 删除旧的subnet KV m.registry.deleteSubnet(ctx, l.Subnet, l.IPv6Subnet) ... } } ... // 尝试复用已分配过的subnet lease if !m.previousSubnet.Empty() { // 查找已分配的subnet关联lease,未被节点占用 if l := findLeaseBySubnet(leases, m.previousSubnet); l == nil { // 网络配置兼容,复用subnet,不再重新分配 if isSubnetConfigCompat(config, m.previousSubnet) && isIPv6SubnetConfigCompat(config, m.previousIPv6Subnet) { sn = m.previousSubnet sn6 = m.previousIPv6Subnet } } } // 无法复用已有的subnet if sn.Empty() { // 重新分配一个subnet sn, sn6, err = m.allocateSubnet(config, leases) ... } // 创建subnet KV及关联lease exp, err := m.registry.createSubnet(ctx, sn, sn6, attrs, subnetTTL) switch { case err == nil: return &lease.Lease{...}, 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
74kube subnetManager维护node注解的网络配置，与etcd不同，kubeAPI模式仅读取node.spec分配的podcidr，不再负责划分节点网络。// AcquireLease adds the flannel specific node annotationsand returns a lease. func (ksm *kubeSubnetManager) AcquireLease(ctx context.Context, attrs *lease.LeaseAttrs) (*lease.Lease, error) { ... // alloc模式,根据节点名获取node if ksm.disableNodeInformer { cachedNode, err = ksm.client.CoreV1().Nodes().Get(ctx, ksm.nodeName,reversion{value:0}) // 其它模式,开启nodeInformer监听,获取node缓存 } else { cachedNode, err = ksm.nodeStore.Get(ksm.nodeName) } ... // backendData转为json bd, err = attrs.BackendData.MarshalJSON() ... v6Bd, err = attrs.BackendV6Data.MarshalJSON() ... // 解析节点podcidr cidr,cidrv6 := parseNodeSpecCIDR() // 节点annotation值与期望配置不一致 if n.annotationChanged() { // 更新node注解数据 updateNodeAnnotation(ctx,n) // 旧node序列化数据 oldData, err := json.Marshal(cachedNode) ... // 新node序列化数据 newData, err := json.Marshal(n) ... // patch到节点 patchBytes, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, v1.Node{}) .. ksm.client.CoreV1().Nodes().Patch(ctx, ksm.nodeName, types.StrategicMergePatchType, patchBytes, metav1.PatchOptions{}, "status") ... } // 初始化租约(24h) lease := &lease.Lease{ Attrs: *attrs, Expiration: time.Now().Add(24 * time.Hour), } // 设置lease subnet if cidr != nil && ksm.enableIPv4 { ipnet := ip.FromIPNet(cidr) net, err := ksm.subnetConf.GetFlannelNetwork(&ipnet) ... lease.Subnet = ip.FromIPNet(cidr) } if ipv6Cidr != nil { ip6net := ip.FromIP6Net(ipv6Cidr) net, err := ksm.subnetConf.GetFlannelIPv6Network(&ip6net) ... lease.IPv6Subnet = ip.FromIP6Net(ipv6Cidr) } // 仅vxlan/host-gw/wireguard后端支持双栈 if attrs.BackendType != "vxlan" && attrs.BackendType != "host-gw" && attrs.BackendType != "wireguard" { lease.EnableIPv4 = true lease.EnableIPv6 = false } return lease, 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
补充
1.
vxlan.RegisterNetwork()负责创建vxlan设备，执行smr.AcquireLease()申请子网及租约，更新vxlan接口地址2.
etcd subnetManager会维护集群节点子网分配，基于lease对象标识占用，分配会优先复用已存储未占用的subnet，以避免虚悬子网3.
kube subnetManager仅维护node annotation值，节点子网分配由controller-manager负责，ksr仅读取使用
# 5.3.run
vxlan.RegisterNetwork()会完成前期准备工作，创建vxlan设备、设置接口地址及维护子网，封装的network会执行Run()维护网络配置。// 网络配置维护 func (nw *network) Run(ctx context.Context) { ... wg.Add(1) go func() { // lease监听 subnet.WatchLeases(ctx, nw.subnetMgr, nw.SubnetLease, events) wg.Done() }() defer wg.Wait() for { // event evtBatch, ok := <-events ... // event处理 nw.handleSubnetEvents(evtBatch) } } // watchLeases based on etcd or kube. func WatchLeases(ctx context.Context, sm Manager, ownLease *lease.Lease, receiver chan []lease.Event) { ... go func() { // lease监听 err := sm.WatchLeases(ctx, leaseWatchChan) ... }() // event同步 for watchResults := range leaseWatchChan { for _, wr := range watchResults { ... // ADD/DEL事件,同步到lw缓存,追加到batch if len(wr.Events) > 0 { batch = lw.Update(wr.Events) // 同步快照状态(ADD/DEL结果追加到batch) } else { batch = lw.Reset(wr.Snapshot) } ... // ADD/DEL事件推送 if len(batch) > 0 { receiver <- batch } } } close(receiver) }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
50sm.WatchLeases()会监听子网变化，esr.WatchLeases()基于watch监听，ksr.WatchLeases()仅转发nodeInformer监听到的事件。// etcd watch(与CompleteLease续期流程一致) func (m *LocalManager) WatchLeases(ctx context.Context, receiver chan []lease.LeaseWatchResult) error { // 获取原始leases(快照) wr, err := m.registry.leasesWatchReset(ctx) ... // 推送快照 receiver <- []lease.LeaseWatchResult{wr} // 计算待watch的next版本 nextIndex, err := getNextIndex(wr.Cursor) ... // watch所有subnet变化 err = m.registry.watchSubnets(ctx, receiver, nextIndex) ... return nil } // kube watch waits for the kubeSubnetManager to provide event in case something relevant changed in node data. func (ksm *kubeSubnetManager) WatchLeases(ctx context.Context, receiver chan []lease.LeaseWatchResult) error { for { select { // nodeInformer监听事件 case event := <-ksm.events: // 推送 receiver <- []lease.LeaseWatchResult{ { Events: []lease.Event{event}, }} case <-ctx.Done(): close(receiver) return ctx.Err() } } }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
33nw.handleSubnetEvents()基于监听到的lease event动态维护节点网络配置，调整routing/arp/fdb表，确保pod流量路由到正确网卡。// subnet event事件处理 func (nw *network) handleSubnetEvents(batch []lease.Event) { // 遍历 for _, event := range batch { ... // IPV4路由与邻居项准备 if event.Lease.EnableIPv4 && nw.dev != nil { // 解析vxlan网卡配置 json.Unmarshal(attrs.BackendData, &vxlanAttrs) ... // 构造vxlan路由 // 其它节点podcidr --> vxlan vxlanRoute = netlink.Route{ LinkIndex: nw.dev.link.Attrs().Index, Scope: netlink.SCOPE_UNIVERSE, Dst: sn.ToIPNet(), Gw: sn.IP.ToIP(), } vxlanRoute.SetFlag(syscall.RTNH_F_ONLINK) // 构造直接路由 // 主机位于二层网络,基于直接路由转发,不经过vxlan网卡封包 directRoute = netlink.Route{ Dst: sn.ToIPNet(), Gw: attrs.PublicIP.ToIP(), } // 检测直接路由支持 if nw.dev.directRouting { dr, err := ip.DirectRouting(attrs.PublicIP.ToIP()) ... directRoutingOK = dr } } // IPV6路由与邻居项准备 if event.Lease.EnableIPv6 && nw.v6Dev != nil { ... } switch event.Type { // ADD(新加入节点) case lease.EventAdded: // IPV4网络配置 if event.Lease.EnableIPv4 { // 直接路由(host-gw) if directRoutingOK { ... // 配置路由表 retry.Do(func() error { return netlink.RouteReplace(&directRoute) }) ... // vxlan网络配置 } else { ... // arp表配置 retry.Do(func() error { return nw.dev.AddARP(neighbor{IP: sn.IP, MAC: net.HardwareAddr(vxlanAttrs.VtepMAC)}) }); err != nil { continue } // fdb表配置 if err := retry.Do(func() error { return nw.dev.AddFDB(neighbor{IP: attrs.PublicIP, MAC: net.HardwareAddr(vxlanAttrs.VtepMAC)}) }); err != nil { ... // 失败,销毁arp记录 retry.Do(func() error { return nw.dev.DelARP(neighbor{IP: event.Lease.Subnet.IP, MAC: net.HardwareAddr(vxlanAttrs.VtepMAC)}) }) ... continue } // vxlan路由配置 if err := retry.Do(func() error { return netlink.RouteReplace(&vxlanRoute) }); err != nil { ... // 销毁arp nw.dev.DelARP(neighbor{IP: event.Lease.Subnet.IP, MAC: net.HardwareAddr(vxlanAttrs.VtepMAC)}) ... // 销毁fdb nw.dev.DelFDB(neighbor{IP: event.Lease.Attrs.PublicIP, MAC: net.HardwareAddr(vxlanAttrs.VtepMAC)}) ... continue } } } // IPV6网络配置 if event.Lease.EnableIPv6 { ... } // 节点销毁/重建 case lease.EventRemoved: // IPV4网络配置调整 if event.Lease.EnableIPv4 { // 直接路由(host-gw) if directRoutingOK { ... // DEL路由记录 retry.Do(func() error { return netlink.RouteDel(&directRoute) }) // vxlan网络 } else { ... // DEL ARP记录 retry.Do(func() error { return nw.dev.DelARP(neighbor{IP: sn.IP, MAC: net.HardwareAddr(vxlanAttrs.VtepMAC)}) }) ... // DEL FDB记录 retry.Do(func() error { return nw.dev.DelFDB(neighbor{IP: attrs.PublicIP, MAC: net.HardwareAddr(vxlanAttrs.VtepMAC)}) }) ... // DEL ROUTE记录 retry.Do(func() error { return netlink.RouteDel(&vxlanRoute) }) ... } } if event.Lease.EnableIPv6 { ... } default: log.Error("internal error: unknown event type: ", int(event.Type)) } } }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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
补充
1.
esr.WatchLeases()基于etcd订阅subnets空间所有子网变化，以感知节点上下线，动态更新网络配置2.
ksr.WatchLeases()基于nodeInformer订阅node变化，以感知节点上下线，动态更新网络配置3.
registryNetwork()准备vxlan设备，run()更像是维护经过vxlan设备的流量
# 6.host-gw实现
# 6.1.原理
host-gw(host gateway)模式基于主机网关实现路由，依赖底层物理网络的路由实现节点间的直接通信，不进行隧道封装，性能相对较高。
注意
host-gw模式将节点物理网卡作为网关，依赖底层物理网络直接路由
# 6.2.registryNetwork
hw.RegisterNetwork()负责网络对象的准备工作，真正的网络配置放在Run()，由于借助二层可达网络转发，相对vxlan模式简单一些。func (be *HostgwBackend) RegisterNetwork(...) (backend.Network, error) { ... // IPV4配置 if config.EnableIPv4 { // 节点网卡IP attrs.PublicIP = ip.FromIP(be.extIface.ExtAddr) // 路由回调 n.GetRoute = func(lease *lease.Lease) *netlink.Route { return &netlink.Route{ Dst: lease.Subnet.ToIPNet(), Gw: lease.Attrs.PublicIP.ToIP(), LinkIndex: n.LinkIndex, } } } // IPV6配置 if config.EnableIPv6 { ... } // 租约申请(etcd/kubeAPI) l, err := be.sm.AcquireLease(ctx, &attrs) switch err { case nil: n.SubnetLease = l ... } return n, 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
补充
host-gw不会创建虚拟设备，lease申请仅对应subnet分配，供后续路由配置
# 6.3.run
host-gw会初始化routeNetwork对象，用于watchLeases事件进一步配置主机路由，以基于底层物理网络转发数据包，实现跨节点通信。func (n *RouteNetwork) Run(ctx context.Context) { ... wg.Add(1) go func() { // lease监听(etcd/kube) subnet.WatchLeases(ctx, n.SM, n.SubnetLease, evts) wg.Done() }() ... // 路由检查 wg.Add(1) go func() { n.routeCheck(ctx) wg.Done() }() defer wg.Wait() for { // 事件处理 evtBatch, ok := <-evts ... n.handleSubnetEvents(evtBatch) } } // lease监听 func WatchLeases(ctx context.Context, sm Manager, ownLease *lease.Lease, receiver chan []lease.Event) { // 初始化owner lease(网络配置屏蔽自身网段) lw := &lease.LeaseWatcher{ OwnLease: ownLease, } ... // lease监听 go func() { sm.WatchLeases(ctx, leaseWatchChan) ... }() // 事件推送 for watchResults := range leaseWatchChan { for _, wr := range watchResults { ... if len(wr.Events) > 0 { batch = lw.Update(wr.Events) } else { batch = lw.Reset(wr.Snapshot) } ... if len(batch) > 0 { receiver <- batch } } } close(receiver) } // 路由检查 func (n *RouteNetwork) routeCheck(ctx context.Context) { for { select { case <-ctx.Done(): return // 10s触发一次,误删路由重新添加 case <-time.After(routeCheckRetries * time.Second): n.checkSubnetExistInV4Routes() n.checkSubnetExistInV6Routes() } } } // IPV4路由检查 func (n *RouteNetwork) checkSubnetExistInV4Routes() { n.checkSubnetExistInRoutes(n.routes, netlink.FAMILY_V4) } func (n *RouteNetwork) checkSubnetExistInRoutes(routes []netlink.Route, ipFamily int) { // 获取当前已配置路由 routeList, err := netlink.RouteList(nil, ipFamily) ... // 遍历缓存路由列表 for _, route := range routes { exist := false // 遍历系统路由 for _, r := range routeList { ... // 标记已添加缓存路由 if routeEqual(r, route) { exist = true break } } // 缓存路由未添加 if !exist { // 重新注册路由 netlink.RouteAdd(&route) ... } } ... } // 事件处理 func (n *RouteNetwork) handleSubnetEvents(batch []lease.Event) { for _, evt := range batch { switch evt.Type { // ADD子网事件 case lease.EventAdded: ... // IPV4路由注册 if evt.Lease.EnableIPv4 { // 基于lease构造route route := n.GetRoute(&evt.Lease) // 注册及添加到缓存 routeAdd(route, netlink.FAMILY_V4, n.addToRouteList, n.removeFromV4RouteList) } // IPV6路由注册 if evt.Lease.EnableIPv6 { // 基于lease构造route route := n.GetV6Route(&evt.Lease) // 注册及添加到缓存 routeAdd(route, netlink.FAMILY_V6, n.addToV6RouteList, n.removeFromV6RouteList) } // DEL子网事件 case lease.EventRemoved: ... // IPV4路由销毁 if evt.Lease.EnableIPv4 { // 构造route route := n.GetRoute(&evt.Lease) // 移除缓存route n.removeFromV4RouteList(*route) // 销毁路由 netlink.RouteDel(route) ... } // IPV6路由销毁 if evt.Lease.EnableIPv6 { // 构造route route := n.GetV6Route(&evt.Lease) // 移除缓存route n.removeFromV6RouteList(*route) // 路由销毁 netlink.RouteDel(route) ... } default: log.Error("Internal error: unknown event type: ", int(evt.Type)) } } }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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
补充
host-gw模式会根据node/subnet变化事件动态注册或销毁路由，已注册路由缓存到列表间隔10s周期确认及补充
# 7.ip-ip实现
# 7.1.原理
IPIP是一种简单的协议，基于IPIP隧道封装和传输容器流量，跨主机通信时，原始的数据包封装在另一个IP(主机)数据包，目标主机收到封装的数据包会解封装出原始的数据包，基于网桥转发给容器。
注意
ipip本身是一种三层隧道协议，用于跨广播域转发数据包，突破host-gw限制
# 7.2.registryNetwork
IPIP模式相对vxlan模式，将网关由flannel.x虚拟设备替换为节点，原始数据包经过flannel.ipip封包至新的IP数据包进行发送。func (be *IPIPBackend) RegisterNetwork(...) (backend.Network, error) { ... // network初始化 n := &backend.RouteNetwork{ SimpleNetwork: backend.SimpleNetwork{ ExtIface: be.extIface, }, SM: be.sm, BackendType: backendType, } ... // lease申请 l, err := be.sm.AcquireLease(ctx, attrs) switch err { case nil: n.SubnetLease = l ... } // 获取集群网络 net, err := config.GetFlannelNetwork(&n.SubnetLease.Subnet) ... // 配置ipip设备 link, err := be.configureIPIPDevice(n.SubnetLease, net) ... n.Mtu = link.MTU n.LinkIndex = link.Index // 构造路由回调 n.GetRoute = func(lease *lease.Lease) *netlink.Route { // ipip设备路由 route := netlink.Route{ Dst: lease.Subnet.ToIPNet(), Gw: lease.Attrs.PublicIP.ToIP(), LinkIndex: n.LinkIndex, Flags: int(netlink.FLAG_ONLINK), } // 直接路由模式 if cfg.DirectRouting { // 检查支持直接路由 dr, err := ip.DirectRouting(lease.Attrs.PublicIP.ToIP()) ... // 经物理网卡直接发送,不再经过ipip设备 if dr { route.LinkIndex = n.ExtIface.Iface.Index } } return &route } return n, nil } // ipip设备配置 func (be *IPIPBackend) configureIPIPDevice(lease *lease.Lease, flannelnet ip.IP4Net) (*netlink.Iptun, error) { // ipip设备定义 link := &netlink.Iptun{LinkAttrs: netlink.LinkAttrs{Name: tunnelName}, Local: be.extIface.IfaceAddr} // 创建IPIP设备 if err := netlink.LinkAdd(link); err != nil { ... // ipip设备存在 existing, err := netlink.LinkByName(tunnelName) ... // 设备配置不一致 ipip, ok := existing.(*netlink.Iptun) ... if ipip.Local == nil || !ipip.Local.Equal(be.extIface.IfaceAddr) || (ipip.Remote != nil && ipip.Remote.String() != "0.0.0.0") { ... // 先删除设备 netlink.LinkDel(existing) ... // 重新添加设备 netlink.LinkAdd(link) ... } } // 计算MTU(上层网卡-20) expectMTU := be.extIface.Iface.MTU - 20 ... oldMTU := link.Attrs().MTU if oldMTU > expectMTU || oldMTU == 0 { // 调整设备MTU netlink.LinkSetMTU(link, expectMTU) ... link.Attrs().MTU = expectMTU } // 隧道设备分配/32地址 ip.EnsureV4AddressOnLink(ip.IP4Net{IP: lease.Subnet.IP, PrefixLen: 32}, flannelnet, link) ... // 激活隧道设备(UP) netlink.LinkSetUp(link) ... return link, 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
注意
1.
ipip模块加载时，内核会自动在每个网络命名空间创建默认ipip设备tun0，接收的ipip协议包没有精确匹配的设备会交给tun02.
ipip设备可重命名tun0实现，考虑到tun0可能占用，重新创建flannel.ipip设备，设置local属性
# 7.3.run
be.RegistryNetwork会构建ipip隧道及设备地址，其余的工作就围绕subnet变化展开，根据网络动态变化更新路由，这里复用host-gw实现。// route动态维护 func (n *RouteNetwork) Run(ctx context.Context) { ... wg.Add(1) go func() { // subnet变化监听 subnet.WatchLeases(ctx, n.SM, n.SubnetLease, evts) wg.Done() }() ... wg.Add(1) go func() { // 缓存route周期检查 n.routeCheck(ctx) wg.Done() }() defer wg.Wait() for { evtBatch, ok := <-evts ... // subnet变化同步 n.handleSubnetEvents(evtBatch) } }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注意
ipip模式与host-gw模式共用routeNetwork(pod-->node)动态维护，区别在于host-gw流量无需流经隧道设备封包
# 8.plugin
# 8.1.简介
cni核心实现包含两部分：cni pod和cni plugin，前者运行于节点维护主机网络配置及隧道设备，后者作为可执行文件供cri调用，配置主机与容器间网络及分配容器IP。其中，cni plugin位于/opt/cni/bin目录，包含cni官方提供的插件及各cni实现的插件。--- ipam(地址分配) host-local，dhcp，static --- main(主机与容器网络配置) bridge、ptp、ipvlan、macvlan、host-device --- meta(三方插件,生成其它cni插件运行的配置文件) bandwidth、firewall、portmap、tuning、sbr、vrf1
2
3
4
5
6
7
8
注意
flannel是委托实现的，flannel-cni-plugin插件仅将配置进行读取和补充及渲染出真正需要执行的插件配置
# 8.2.原理
flannel plugin是委托插件，根据入参及flanneld渲染配置转化真正执行插件，组合一般为网桥插件(bridge)和IPAM插件(host-local)，以创建网桥及veth设备，执行IPAM插件分配地址，根据分配地址配置容器路由，同时将vethx隧道接入网桥，确保容器网络访问外部正常。
注意
1.
flanneld运行于节点动态感知节点网段，创建隧道设备及维护路由，确保cni0的流量正常转到节点外部2.
flannel plugin由cri调用，根据入参及flanneld维护配置渲染真正执行插件及调用3.
main/ipam插件真正执行，创建网桥及容器网络，根据容器分配IP配置网络命名空间内路由，确保网络命名空间内流量经eth0转到cni04.
ipam执行分配IP保存到/var/lib/cni/network，用于唯一分配检查及后续回收
# 8.3.网络创建
cmdAdd是cni实现容器网络添加的核心功能，cri调度pause容器会基于cni plugin配置容器网络环境，以确保容器接入指定网络与外部互通。func cmdAdd(args *skel.CmdArgs) error { // 解析输入(subnet.env和data持久化位置) n, err := loadFlannelNetConf(args.StdinData) ... // 加载subnet.env配置 fenv, err := loadFlannelSubnetEnv(n.SubnetFile) ... return doCmdAdd(args, n, fenv) } func doCmdAdd(args *skel.CmdArgs, n *NetConf, fenv *subnetEnv) error { // 设置默认配置 n.Delegate["name"] = n.Name if !hasKey(n.Delegate, "type") { n.Delegate["type"] = "bridge" } if !hasKey(n.Delegate, "ipMasq") { // if flannel is not doing ipmasq, we should ipmasq := !*fenv.ipmasq n.Delegate["ipMasq"] = ipmasq } if !hasKey(n.Delegate, "mtu") { mtu := fenv.mtu n.Delegate["mtu"] = mtu } if n.Delegate["type"].(string) == "bridge" { if !hasKey(n.Delegate, "isGateway") { n.Delegate["isGateway"] = true } } if n.CNIVersion != "" { n.Delegate["cniVersion"] = n.CNIVersion } // 获取ipam配置 ipam, err := getDelegateIPAM(n, fenv) ... n.Delegate["ipam"] = ipam // 持久化及执行网桥插件 return delegateAdd(args.ContainerID, n.DataDir, n.Delegate) }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
注意
flannel plugin仅读取入参及subnet.env文件，组装网桥插件及IPAM插件执行所需配置，剩余的网络配置和地址分配就委托给它们执行
# 8.4.网络删除
cmdDel是cni实现容器网络删除的核心功能，cri删除pause容器会基于cni plugin清理容器网络环境，回收容器网卡及分配IP。func cmdDel(args *skel.CmdArgs) error { // 加载入参配置 nc, err := loadFlannelNetConf(args.StdinData) ... if nc.RuntimeConfig != nil { if nc.Delegate == nil { nc.Delegate = make(map[string]interface{}) } nc.Delegate["runtimeConfig"] = nc.RuntimeConfig } // 执行删除 return doCmdDel(args, nc) } func doCmdDel(args *skel.CmdArgs, n *NetConf) error { // 加载容器插件配置 cleanup, netConfBytes, err := consumeScratchNetConf(args.ContainerID, n.DataDir) ... // 清理持久化的插件配置 defer func() { cleanup(err) }() // 解析插件配置 nc := &types.NetConf{} json.Unmarshal(netConfBytes, nc) ... // 委托网桥插件执行 return invoke.DelegateDel(context.TODO(), nc.Type, netConfBytes, 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
注意
flannel plugin执行回收会基于渲染网桥及IPAM插件配置，先基于网桥插件执行设备回收，再执行IPAM地址回收，避免直接回收地址冲突