自己动手写Kubernetes的CNI

2024-12-15 约 2069 字预计阅读 5 分钟

概述

近期公司在容器网络上有些新的场景，例如需要在容器中插入一个 ipip tunnel，并且配置好一个公司内部的 vip。这一步，之前内部的实现是通过在需要这种网络模式的 Pod 里，根据一定的规则通过 webhook patch 进去一个 init-container，但近期的实践中发现，某些组件的 Operator 在实现原地重启的逻辑的时候，会限制用户 patch 一些元素到 Pod 里，比如说会限制 patch init-container 这个元素，因为在做原地更新的时候，这些组件可能会对比原地更新前后的 Pod 的 Spec，有可能是让你不能 patch 任何东西进去的。

那么为了解决上面提到的这个问题，我们考虑到自行实现一个 CNI，通过 CNI 把容器内部的 ipip tunnel 设备配置好，这也符合 CNI 的规范和语义，因此最后我们的方案确定为自定义 CNI 来配置 ipip tunnel。

CNI流程

因为我司的容器需要多个 CNI，所以是用 chain 的方式来链式调用，相比于修改现有的 CNI，我们相对比较熟悉这种链式调用的方式，这也是我们考虑自定义 CNI 的原因之一。

/%E8%87%AA%E5%B7%B1%E5%8A%A8%E6%89%8B%E5%86%99kubernetes%E7%9A%84cni/img.png

假设我们定义的 CNI 名为 tun，下面是我们的 CNI 配置文件。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18


{
  "name": "generic-veth",
  "cniVersion": "0.1.0",
  "plugins": [
    {
      "type": "contivk8s.bin"
    },
    {
      "type": "cilium-cni"
    },
    {
      "type": "tun",
      "capabilities": {
        "io.kubernetes.cri.pod-annotations": true
      }
    }
  ]
}

另外从上面的流程图看到，tun 需要获取 Pod 的 Annotations，也就是了解 Pod 对应的业务和部署池等信息，通过这些信息，需要到 VGW 获取 VIP，这个假设的前提也是业务部署池等信息相对不会变化，而 VIP 是有可能变化的，如果 VIP 可以保证不变，那么 tun 可以不通过 HTTP 的方式去获取 VIP，毕竟这一步是有消耗的，如果 VIP 确保不变，那么也可以通过 Pod 的 Annotations 写死往后传。需要注意的是，capabilities 是必须要配置的，否则 CNI 无法获取 Pod 上的 Annotations。

代码实现

下面是 tun 代码实现的伪代码，可以根据代码的逻辑，看到 ipip tunnel 配置的过程。

  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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260


package main

import (
    ...
)

const logFilePath = "/var/log/tun.log"

func init() {
	// 创建日志文件
	file, err := os.OpenFile(logFilePath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
	if err != nil {
		fmt.Printf("Could not open log file: %v\n", err)
		os.Exit(1)
	}

	// 设置日志输出到文件
	log.SetOutput(file)
	log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile)
}

// GetHtVip 获取 VIP 列表
func GetHtVip(domainName, vipPoolId string) []string {
    ...
	return vips
}

type NetConf struct {
	types.NetConf
	RuntimeConfig json.RawMessage `json:"runtimeConfig,omitempty"`
}

func parseNetConf(bytes []byte) (*NetConf, error) {
	conf := &NetConf{}
	if err := json.Unmarshal(bytes, conf); err != nil {
		log.Printf("Failed to parse network config: %v", err)
		return nil, fmt.Errorf("failed to parse network config: %v", err)
	}
	log.Println("Parsed network config successfully")

	if conf.RawPrevResult != nil {
		log.Println("Parsing previous result")
		if err := version.ParsePrevResult(&conf.NetConf); err != nil {
			log.Printf("Failed to parse prevResult: %v", err)
			return nil, fmt.Errorf("failed to parse prevResult: %v", err)
		}

		if _, err := current.NewResultFromResult(conf.PrevResult); err != nil {
			log.Printf("Failed to convert result to current version: %v", err)
			return nil, fmt.Errorf("failed to convert result to current version: %v", err)
		}
		log.Println("Previous result parsed successfully")
	} else {
		log.Println("No previous result to parse")
	}
	return conf, nil
}

func cmdAdd(args *skel.CmdArgs) error {
	// 解析网络配置
	conf, err := parseNetConf(args.StdinData)
	if err != nil {
		log.Printf("Error parsing network config: %v", err)
		return err
	}
	log.Println("Parsed network config successfully")

	// 设置 VIP
	var annotations map[string]map[string]string
	if err := json.Unmarshal(conf.RuntimeConfig, &annotations); err != nil {
		log.Printf("Failed to decode runtimeConfig: %v", err)
		return fmt.Errorf("failed to decode runtimeConfig: %w", err)
	}
	log.Println("Decoded runtimeConfig successfully")

	var vips []string
	if podAnnotations, ok := annotations["io.kubernetes.cri.pod-annotations"]; ok {
		businessDomain := podAnnotations["business_domain"]
		poolID := podAnnotations["pool_id"]
		serviceType := podAnnotations["service_type"]
		log.Printf("Retrieving VIPs for business domain: %s, pool ID: %s", businessDomain, poolID)

		// 仅当 service_type 为 vgw 时才获取 VIP
		if serviceType == "vgw" {
			vips = GetHtVip(businessDomain, poolID)
		} else {
			log.Println("Service type is not 'vgw', returning previous result")
			return types.PrintResult(conf.PrevResult, conf.CNIVersion)
		}
	}
	log.Printf("Retrieved VIPs: %v", vips)

	if len(vips) == 0 {
		log.Println("No VIPs found, exiting cmdAdd")
		return nil
	}

	tunnelName := "tunl0"
	log.Printf("Configuring tunnel: %s", tunnelName)

	// 进入指定的网络命名空间
	err = ns.WithNetNSPath(args.Netns, func(_ ns.NetNS) error {
		// 查找隧道接口
		link, err := netlink.LinkByName(tunnelName)
		if err != nil {
			log.Printf("Failed to find tunnel: %v", err)
			return fmt.Errorf("failed to find tunnel: %v", err)
		}
		log.Printf("Found tunnel: %s", tunnelName)

		// 启用隧道接口
		if err := netlink.LinkSetUp(link); err != nil {
			log.Printf("Failed to set tunnel up: %v", err)
			return fmt.Errorf("failed to set tunnel up: %v", err)
		}
		log.Printf("Tunnel %s is now up", tunnelName)

		// 添加 VIP 地址
		for _, vip := range vips {
			addr, err := netlink.ParseAddr(fmt.Sprintf("%s/32", vip))
			if err != nil {
				log.Printf("Failed to parse address %s: %v", vip, err)
				return fmt.Errorf("failed to parse address: %v", err)
			}

			// 获取当前接口上的地址列表
			existingAddrs, err := netlink.AddrList(link, netlink.FAMILY_V4)
			if err != nil {
				log.Printf("Failed to list addresses on link %s: %v", link.Attrs().Name, err)
				return fmt.Errorf("failed to list addresses: %v", err)
			}

			// 检查地址是否已经存在
			exists := false
			for _, existingAddr := range existingAddrs {
				if existingAddr.IP.Equal(addr.IP) {
					exists = true
					break
				}
			}

			// 如果地址不存在，则添加
			if !exists {
				if err := netlink.AddrAdd(link, addr); err != nil {
					log.Printf("Failed to add address %s to tunnel: %v", vip, err)
					return fmt.Errorf("failed to add address: %v", err)
				}
				log.Printf("Added address %s to tunnel", vip)
			} else {
				log.Printf("Address %s already exists on tunnel, skipping addition", vip)
			}
		}

		// 修改默认路由的 advmss 值
		if err := modifyDefaultRouteAdvmss(); err != nil {
			return err
		}
		// 配置 ARP 参数
		if err := configureTunParameters(tunnelName); err != nil {
			return err
		}

		return nil
	})

	if err != nil {
		log.Printf("Error configuring network namespace: %v", err)
		return err
	}
	log.Println("Configured network namespace successfully")

	// 创建结果对象
	var result types.Result

	// 如果有前一个结果，则合并
	if conf.PrevResult != nil {
		log.Println("Found previous result, merging with current configuration")
		// 先将 PrevResult 转换为 current.Result
		prevResult, err := current.NewResultFromResult(conf.PrevResult)
		if err != nil {
			log.Printf("Failed to convert previous result: %v", err)
			return err
		}

		// 将当前接口添加到前一个结果
		prevResult.Interfaces = append(prevResult.Interfaces, &current.Interface{
			Name:    tunnelName,
			Mac:     "00:00:00:00:00:00",
			Sandbox: args.Netns,
		})

		// 返回合并后的结果
		result = prevResult
	} else {
		log.Println("No previous result found, creating new result")
		result = &current.Result{
			CNIVersion: conf.CNIVersion,
			Interfaces: []*current.Interface{
				{
					Name:    tunnelName,
					Mac:     "00:00:00:00:00:00",
					Sandbox: args.Netns,
				},
			},
		}
	}

	log.Println("Returning result from cmdAdd")
	return types.PrintResult(result, conf.CNIVersion)
}

func modifyDefaultRouteAdvmss() error {
    ...
	// 修改默认路由的 advmss
	for _, route := range routes {
		// 检查是否为默认路由
		if route.Dst.String() == "0.0.0.0/0" {
			log.Printf("Found default route: %+v", route)
			route.AdvMSS = 1420 // 设置 advmss 值
			if err := netlink.RouteReplace(&route); err != nil {
				log.Printf("Failed to modify default route: %v", err)
				return fmt.Errorf("failed to modify default route: %v", err)
			}
			log.Printf("Modified default route advmss to %d", route.AdvMSS)
			break // 修改完默认路由后退出
		}
	}
	return nil
}

func configureTunParameters(tunnelName string) error {
	log.Println("Configuring ARP parameters")
    ...
}

func cmdDel(args *skel.CmdArgs) error {
	log.Println("Starting cmdDel")

    ...

	log.Println("cmdDel completed successfully")
	return nil
}

func cmdCheck(args *skel.CmdArgs) error {
	log.Println("Starting cmdCheck")

	...

	log.Println("cmdCheck completed successfully")
	return nil
}

func main() {
	skel.PluginMainFuncs(skel.CNIFuncs{
		Add:   cmdAdd,
		Check: cmdCheck,
		Del:   cmdDel,
	}, version.All, bv.BuildString("tunnel"))
}

其他tips

需要注意如果还在使用 Docker 来作为 Kubernetes 的容器运行时，要注意 setupPodNetwork，具体是由 Kubelet 还是 Containerd 配置的，在新版本的 CRI 标准下，Kubelet 已经不负责配置 sandbox 的网络配置，而是交给下游的 Containerd 来实现。可以参考 sandbox_run.go#L531。

参考资料

CNI官网

💡赞赏支持

微信打赏

支付宝打赏

目录

自己动手写Kubernetes的CNI

概述

CNI流程

代码实现

其他tips

参考资料