在 kube-controller-manager 有众多控制器,与 Pod IP 相关的源码是 NodeIpamController。
NodeIpamController 控制器主要是分析管理节点的 podcidr,当有新节点加入集群时,对P的管分配一个子网段给节点;当节点删除时,源码回收子网段。分析
每个节点的对P的管子网段不会重叠,每个节点都能够独立地完成 Pod IP 的分配。
下面看一个 kube-controller-manager 的运行示例:
kubectl -n kube-system get pod kube-controller-manager -o yaml
其中关于网段配置的部分为:
spec: containers: - command: - kube-controller-manager - --allocate-node-cidrs=true - --cluster-cidr=10.234.0.0/16 - --node-cidr-mask-size=24 - --service-cluster-ip-range=10.96.0.0/16
cluster-cidr 指定了 Pod IP 的范围,掩码位数 16,如果不考虑保留 IP,意味着集群最多可以容纳 2^16 = 65536 个 pod。
这些 Pod 分布在若干个节点上,接着看 node-cidr-mask-size 为 24,每个节点只剩下 32-24=8 位留给 pod,每个节点最多能创建 2^8=256 个 pod。
相应的,这个集群能够容纳的节点数量为 2^(32-16-8)=256 个节点。
在规划集群时,需要根据集群的规模来调整这两个参数。
开启 allocate-node-cidrs、设置 cluster-cidr 之后,kube-controller-manager 会给每个节点分配子网段,将结果写入 spec.podCIDR 字段。
spec: podCIDR: 10.234.58.0/24 podCIDRs: - 10.234.58.0/24
下面我们从源码分析一下这一过程。
func startNodeIpamController(ctx context.Context, controllerContext ControllerContext) (controller.Interface, bool, error) { // 如果 allocate-node-cidrs 没有开启会立即返回 if !controllerContext.ComponentConfig.KubeCloudShared.AllocateNodeCIDRs { return nil, false, nil } // 获取 clusterCIDR, serviceCIDR 启动 NodeIpamController nodeIpamController, err := nodeipamcontroller.NewNodeIpamController( ctx, controllerContext.InformerFactory.Core().V1().Nodes(), clusterCIDRInformer, controllerContext.Cloud, controllerContext.ClientBuilder.ClientOrDie("node-controller"), clusterCIDRs, serviceCIDR, secondaryServiceCIDR, nodeCIDRMaskSizes, ipam.CIDRAllocatorType(controllerContext.ComponentConfig.KubeCloudShared.CIDRAllocatorType), ) go nodeIpamController.RunWithMetrics(ctx, controllerContext.ControllerManagerMetrics) return nil, true, nil}
RunWithMetrics 只是提供了一些监控指标,真正的启动逻辑在 Run 方法中。
func (nc *Controller) RunWithMetrics(ctx context.Context, controllerManagerMetrics *controllersmetrics.ControllerManagerMetrics) { controllerManagerMetrics.ControllerStarted("nodeipam") defer controllerManagerMetrics.ControllerStopped("nodeipam") nc.Run(ctx)}
func (nc *Controller) Run(ctx context.Context) { if nc.allocatorType == ipam.IPAMFromClusterAllocatorType || nc.allocatorType == ipam.IPAMFromCloudAllocatorType { go nc.legacyIPAM.Run(ctx) } else { go nc.cidrAllocator.Run(ctx) } <-ctx.Done()}
在查找 cidrAllocator 接口实现的时候,我发现了三种 CIDR 分配器,分别是 RangeAllocator 适用单网段分配、MultiCIDRRangeAllocator 适用于多 CIDR、CloudCIDRAllocator 适用于对接云厂。
func New(ctx context.Context, kubeClient clientset.Interface, cloud cloudprovider.Interface, nodeInformer informers.NodeInformer, clusterCIDRInformer networkinginformers.ClusterCIDRInformer, allocatorType CIDRAllocatorType, allocatorParams CIDRAllocatorParams) (CIDRAllocator, error) { switch allocatorType { case RangeAllocatorType: return NewCIDRRangeAllocator(logger, kubeClient, nodeInformer, allocatorParams, nodeList) case MultiCIDRRangeAllocatorType: if !utilfeature.DefaultFeatureGate.Enabled(features.MultiCIDRRangeAllocator) { return nil, fmt.Errorf("invalid CIDR allocator type: %v, feature gate %v must be enabled", allocatorType, features.MultiCIDRRangeAllocator) } return NewMultiCIDRRangeAllocator(ctx, kubeClient, nodeInformer, clusterCIDRInformer, allocatorParams, nodeList, nil) case CloudAllocatorType: return NewCloudCIDRAllocator(logger, kubeClient, cloud, nodeInformer) default: return nil, fmt.Errorf("invalid CIDR allocator type: %v", allocatorType) }}
这里看看 RangeAllocator 的实现。
func NewCIDRRangeAllocator(logger klog.Logger, client clientset.Interface, nodeInformer informers.NodeInformer, allocatorParams CIDRAllocatorParams, nodeList *v1.NodeList) (CIDRAllocator, error) { nodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: controllerutil.CreateAddNodeHandler(func(node *v1.Node) error { return ra.AllocateOrOccupyCIDR(logger, node) }), UpdateFunc: controllerutil.CreateUpdateNodeHandler(func(_, newNode *v1.Node) error { if len(newNode.Spec.PodCIDRs) == 0 { return ra.AllocateOrOccupyCIDR(logger, newNode) } return nil }), DeleteFunc: controllerutil.CreateDeleteNodeHandler(logger, func(node *v1.Node) error { return ra.ReleaseCIDR(logger, node) }), }) return ra, nil}
其实 RangeAllocator 分配器的实现与写 Operator 时的控制器类似,都是通过 informer 来监听资源的变化,然后调用相应的方法。
这里比较特殊的是,控制器并不是直接操作资源,而是将变更放到了一个 channel 中,然后通过 goroutine 处理状态更新。
func (r *rangeAllocator) AllocateOrOccupyCIDR(logger klog.Logger, node *v1.Node) error { allocated := nodeReservedCIDRs{ nodeName: node.Name, allocatedCIDRs: make([]*net.IPNet, len(r.cidrSets)), } for idx := range r.cidrSets { podCIDR, err := r.cidrSets[idx].AllocateNext() allocated.allocatedCIDRs[idx] = podCIDR } // 将更新的内容放入 channel 中 r.nodeCIDRUpdateChannel <- allocated return nil}
nodeCIDRUpdateChannel 的长度是 5000。
cidrUpdateQueueSize = 5000 nodeCIDRUpdateChannel: make(chan nodeReservedCIDRs, cidrUpdateQueueSize),
而更新 Node Spec 的逻辑是通过 30 个 goroutine 来处理。
const cidrUpdateWorkers untyped int = 30 for i := 0; i < cidrUpdateWorkers; i++ { go r.worker(ctx) }
func (r *rangeAllocator) worker(ctx context.Context) { logger := klog.FromContext(ctx) for { select { case workItem, ok := <-r.nodeCIDRUpdateChannel: if !ok { logger.Info("Channel nodeCIDRUpdateChannel was unexpectedly closed") return } if err := r.updateCIDRsAllocation(logger, workItem); err != nil { // Requeue the failed node for update again. r.nodeCIDRUpdateChannel <- workItem } case <-ctx.Done(): return } }}
cidrUpdateRetries = 3 这里会重试 3 次更新,如果一直更新失败,会将节点重新放入 channel 中,等待下次更新。
// updateCIDRsAllocation assigns CIDR to Node and sends an update to the API server.func (r *rangeAllocator) updateCIDRsAllocation(logger klog.Logger, data nodeReservedCIDRs) error { // If we reached here, it means that the node has no CIDR currently assigned. So we set it. for i := 0; i < cidrUpdateRetries; i++ { if err = nodeutil.PatchNodeCIDRs(r.client, types.NodeName(node.Name), cidrsString); err == nil { logger.Info("Set node PodCIDR", "node", klog.KObj(node), "podCIDRs", cidrsString) return nil } } // 放回 pool 中 controllerutil.RecordNodeStatusChange(logger, r.recorder, node, "CIDRAssignmentFailed")}
使用 Patch 方法更新节点对象的 Spec 字段。
func PatchNodeCIDRs(c clientset.Interface, node types.NodeName, cidrs []string) error { // set the Pod cidrs list and set the old Pod cidr field patch := nodeForCIDRMergePatch{ Spec: nodeSpecForMergePatch{ PodCIDR: cidrs[0], PodCIDRs: cidrs, }, } patchBytes, err := json.Marshal(&patch) if err != nil { return fmt.Errorf("failed to json.Marshal CIDR: %v", err) } if _, err := c.CoreV1().Nodes().Patch(context.TODO(), string(node), types.StrategicMergePatchType, patchBytes, metav1.PatchOptions{ }); err != nil { return fmt.Errorf("failed to patch node CIDR: %v", err) } return nil}
图片
上图是 Kubelet 创建 Pod 的过程,这里截取其中对网络配置的部分进行分析:
下面从源码实现的角度来看看这个过程。
apiVersion: v1kind: Podmetadata: labels: app: demo pod-template-hash: 7b9b5cf76b name: demo-7b9b5cf76b-5lpmj namespace: defaultspec: containers: - image: hubimage/demo-ubuntu nodeName: node1
Kubernetes 中调度的过程是 kube-scheduler 根据 Pod 的资源需求和节点的资源情况,将 Pod 调度到某个节点上,并将调度结果写入 pod.spec.nodeName 字段。
这部分不是网络的重点,之前我也在生产环境下定制过调度器,感兴趣的话可以看看Tekton 优化之定制集群调度器 。
SyncPod 是 kubelet 中的核心方法,它会根据 Pod 的状态,调用 cri 创建或删除 pod。
// SyncPod syncs the running Pod into the desired Pod by executing following steps://// 1.计算沙箱和容器变化。// 2. 必要时关闭 Pod 沙箱。// 3. 关闭任何不应运行的容器。// 4.必要时创建沙箱。// 5.创建 ephemeral 容器。// 6. 创建 init 容器。// 7. 调整运行容器的大小(如果 InPlacePodVerticalScaling==true)// 8. 创建正常容器func (m *kubeGenericRuntimeManager) SyncPod(ctx context.Context, Pod *v1.Pod, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, backOff *flowcontrol.Backoff) (result kubecontainer.PodSyncResult) { // Step 4: Create a sandbox for the Pod if necessary. podSandboxID, msg, err = m.createPodSandbox(ctx, pod, podContainerChanges.Attempt)}
调用 RuntimeService 接口的 RunPodSandbox 方法创建 sandbox。
// createPodSandbox creates a Pod sandbox and returns (podSandBoxID, message, error).func (m *kubeGenericRuntimeManager) createPodSandbox(ctx context.Context, Pod *v1.Pod, attempt uint32) (string, string, error) { podSandBoxID, err := m.runtimeService.RunPodSandbox(ctx, podSandboxConfig, runtimeHandler)
经过 runtimeService、instrumentedRuntimeService 接口的封装,最终会调用 remoteRuntimeService 的 RunPodSandbox 方法。
// RunPodSandbox creates and starts a pod-level sandbox. Runtimes should ensure// the sandbox is in ready state.func (r *remoteRuntimeService) RunPodSandbox(ctx context.Context, config *runtimeapi.PodSandboxConfig, runtimeHandler string) (string, error) { resp, err := r.runtimeClient.RunPodSandbox(ctx, &runtimeapi.RunPodSandboxRequest{ Config: config, RuntimeHandler: runtimeHandler, })
这里的 runtimeClient 是一个 rpc client,通过 rpc 调用 container runtime 创建 sandbox。
以 containerd 为例,创建 sandbox:
func (in *instrumentedService) RunPodSandbox(ctx context.Context, r *runtime.RunPodSandboxRequest) (res *runtime.RunPodSandboxResponse, err error) { if err := in.checkInitialized(); err != nil { return nil, err } res, err = in.c.RunPodSandbox(ctrdutil.WithNamespace(ctx), r) return res, errdefs.ToGRPC(err)}
调用 CNI 创建网络,创建 sandbox。
// RunPodSandbox creates and starts a pod-level sandbox. Runtimes should ensure// the sandbox is in ready state.func (c *criService) RunPodSandbox(ctx context.Context, r *runtime.RunPodSandboxRequest) (_ *runtime.RunPodSandboxResponse, retErr error) { // 生成 sandbox id id := util.GenerateID() metadata := config.GetMetadata() name := makeSandboxName(metadata) // 获取 sandbox 的 oci 运行时 ociRuntime, err := c.getSandboxRuntime(config, r.GetRuntimeHandler()) sandboxInfo.Runtime.Name = ociRuntime.Type sandboxInfo.Sandboxer = ociRuntime.Sandboxer // 创建 sandbox 对象 sandbox := sandboxstore.NewSandbox( sandboxstore.Metadata{ ID: id, Name: name, Config: config, RuntimeHandler: r.GetRuntimeHandler(), }, sandboxstore.Status{ State: sandboxstore.StateUnknown, }, ) // 调用 CNI 插件,创建 sandbox 的网络 if !hostNetwork(config) && !userNsEnabled { var netnsMountDir = "/var/run/netns" sandbox.NetNS, err = netns.NewNetNS(netnsMountDir) // Save sandbox metadata to store if err := c.setupPodNetwork(ctx, &sandbox); err != nil { return nil, fmt.Errorf("failed to setup network for sandbox %q: %w", id, err) } } // 创建 sandbox err = c.nri.RunPodSandbox(ctx, &sandbox)}
在上一步骤中,调用 RunPodSandbox 创建 sandbox 之前,会先调用 setupPodNetwork 配置网络。这里展开看一下 setupPodNetwork 的实现。
func (c *criService) setupPodNetwork(ctx context.Context, sandbox *sandboxstore.Sandbox) error { var ( id = sandbox.ID config = sandbox.Config path = sandbox.NetNSPath netPlugin = c.getNetworkPlugin(sandbox.RuntimeHandler) err error result *cni.Result ) if c.config.CniConfig.NetworkPluginSetupSerially { result, err = netPlugin.SetupSerially(ctx, id, path, opts...) } else { result, err = netPlugin.Setup(ctx, id, path, opts...) }}
libcni 实现了 netPlugin 接口
// containerd/go-cni/cni.gofunc (c *libcni) Setup(ctx context.Context, id string, path string, opts ...NamespaceOpts) (*Result, error) { if err := c.Status(); err != nil { return nil, err } // 建一个新的网络命名空间 ns, err := newNamespace(id, path, opts...) if err != nil { return nil, err } // 调用 CNI 插件 result, err := c.attachNetworks(ctx, ns) if err != nil { return nil, err } return c.createResult(result)}
attachNetworks 起了很多协程,每个协程调用 asynchAttach 方法,asynchAttach 方法调用 Attach 方法。
func (c *libcni) attachNetworks(ctx context.Context, ns *Namespace) ([]*types100.Result, error) { var wg sync.WaitGroup var firstError error results := make([]*types100.Result, len(c.Networks())) rc := make(chan asynchAttachResult) for i, network := range c.Networks() { wg.Add(1) go asynchAttach(ctx, i, network, ns, &wg, rc) } for range c.Networks() { rs := <-rc if rs.err != nil && firstError == nil { firstError = rs.err } results[rs.index] = rs.res } wg.Wait() return results, firstError}
运行了很多协程调用 CNI,但 rc channel 的长度为 1,处理结果时却一个一个的。
func asynchAttach(ctx context.Context, index int, n *Network, ns *Namespace, wg *sync.WaitGroup, rc chan asynchAttachResult) { defer wg.Done() r, err := n.Attach(ctx, ns) rc <- asynchAttachResult{ index: index, res: r, err: err}}
Attach 方法中才真正开始调用 CNI 插件。
func (n *Network) Attach(ctx context.Context, ns *Namespace) (*types100.Result, error) { r, err := n.cni.AddNetworkList(ctx, n.config, ns.config(n.ifName)) if err != nil { return nil, err } return types100.NewResultFromResult(r)}
在 https://github.com/containernetworking/cni/blob/main/libcni/api.go 中 CNI 接口定义了很多方法,其中最重要的是 AddNetwork 和 DelNetwork 方法,带 List 的方法是批量操作。
type CNI interface { AddNetworkList(ctx context.Context, net *NetworkConfigList, rt *RuntimeConf) (types.Result, error) AddNetwork(ctx context.Context, net *NetworkConfig, rt *RuntimeConf) (types.Result, error) DelNetworkList(ctx context.Context, net *NetworkConfigList, rt *RuntimeConf) error DelNetwork(ctx context.Context, net *NetworkConfig, rt *RuntimeConf) error}
AddNetwork 用于为容器添加网络接口,在主机上创建 veth 网卡绑定到容器的 ech0 网卡上。DelNetwork 用于在容器删除时,清理容器相关的网络配置。
CNI 调用插件的核心是 Exec 接口,直接调用二进制程序。
type Exec interface { ExecPlugin(ctx context.Context, pluginPath string, stdinData []byte, environ []string) ([]byte, error) FindInPath(plugin string, paths []string) (string, error) Decode(jsonBytes []byte) (version.PluginInfo, error)}
CRI 以标准输入、环境变量的形式将网络配置信息传递给 CNI 插件。CNI 插件处理完成之后,将网络配置信息写入到标准输出中,CRI 将标准输出中的网络配置信息解析出来,写入到容器的网络配置文件中。
再回到 container runtime 的实现 containerd:
/usr/bin/containerd config dump |grep cni [plugins."io.containerd.grpc.v1.cri".cni] bin_dir = "/opt/cni/bin" conf_dir = "/etc/cni/net.d"
这里的 /etc/cni/net.d 是 CNI 网络配置文件的默认存放路径,/opt/cni/bin 是 CNI 网络插件的默认搜索路径。
ls /opt/cni/binbandwidth calico cilium-cni firewall host-device install loopback portmap sbr tuning vrfbridge calico-IPAM dhcp flannel host-local ipvlan macvlan ptp static vlan
cat /etc/cni/net.d/05-cilium.conf{ "cniVersion": "0.3.1", "name": "cilium", "type": "cilium-cni", "enable-debug": false}
这些配置用来初始化 CRI 获取 CNI 插件的 netPlugin map[string]cni.CNI 结构。
IPAM 是 IP Address Management 的缩写,负责为容器分配 ip 地址。IPAM 组件通常是一个独立的二进制文件,也可以直接由 CNI 插件实现。在 https://github.com/containernetworking/plugins/tree/main/plugins/ipam 中,目前有三种实现 host-local、dhcp、static。 这里以 host-local 为例:
cat /etc/cni/net.d/10-cni.conflist{ "name": "networks", "type": "cni", "ipam": { "type": "host-local", "subnet": "10.234.58.0/24", "routes": [{ "dst": "0.0.0.0/0" }] }}
指定了 CNI 插件的类型为 host-local,指定了 Pod IP 的网段为 "10.234.58.0/24" 。
ls /var/lib/cni/networks10.234.58.76 10.234.58.87 last_reserved_ip.0 lock
cat 10.234.58.76b3b668af977bbeca6853122514044865793c056e81cccebf115dacffd25a8bcc
这里有一组以 ip 命名的文件,而文件里面又是一串字符串。那么这些到底是什么呢?
申请一个 Pod IP 时,先获取一个可用 ip
func cmdAdd(args *skel.CmdArgs) error { for idx, rangeset := range ipamConf.Ranges { ipConf, err := allocator.Get(args.ContainerID, args.IfName, requestedIP) }}
获取到可用 ip 之后,先尝试着存储到本地目录文件中
func (a *IPAllocator) Get(id string, ifname string, requestedIP net.IP) (*current.IPConfig, error) { for { reservedIP, gw = iter.Next() reserved, err := a.store.Reserve(id, ifname, reservedIP.IP, a.rangeID) }}
直接写本地文件目录
func (s *Store) Reserve(id string, ifname string, ip net.IP, rangeID string) (bool, error) { fname := GetEscapedPath(s.dataDir, ip.String()) f, err := os.OpenFile(fname, os.O_RDWR|os.O_EXCL|os.O_CREATE, 0o600) if os.IsExist(err) { return false, nil } if _, err := f.WriteString(strings.TrimSpace(id) + LineBreak + ifname); err != nil { f.Close() os.Remove(f.Name()) return false, err }}
写入的内容为 strings.TrimSpace(id) + LineBreak + ifname,这里的 id 其实是容器的 id,ifname 是网卡名称,LineBreak 是换行符。
通过 id 在主机上可以找到对应的容器:
docker ps |grep b3b668b3b668af977b k8s.gcr.io/pause:3.5 "/pause" 6 weeks ago Up 6 weeks k8s_POD_xxx-5b795fd7dd-82hrh_kube-system_b127b65c-f0ca-48a7-9020-ada60dfa535a_0
cat last_reserved_ip.010.234.58.87
在获取可用 IP 时,IPAM 会创建一个迭代器。
func (a *IPAllocator) Get(id string, ifname string, requestedIP net.IP) (*current.IPConfig, error) { iter, err := a.GetIter() if err != nil { return nil, err } for { reservedIP, gw = iter.Next() if reservedIP == nil { break } }
而迭代器需要依靠 last_reserved_ip.0 找到上一次分配的 IP,然后从这个 IP 之后开始分配。
func (a *IPAllocator) GetIter() (*RangeIter, error) { lastReservedIP, err := a.store.LastReservedIP(a.rangeID) if err != nil && !os.IsNotExist(err) { log.Printf("Error retrieving last reserved ip: %v", err) } else if lastReservedIP != nil { startFromLastReservedIP = a.rangeset.Contains(lastReservedIP) }
这里的 lastIPFilePrefix = "last_reserved_ip."
func (s *Store) LastReservedIP(rangeID string) (net.IP, error) { ipfile := GetEscapedPath(s.dataDir, lastIPFilePrefix+rangeID) data, err := os.ReadFile(ipfile) if err != nil { return nil, err } return net.ParseIP(string(data)), nil}
host-local 分配 ip 时是按照轮询的方式,递增分配,如果分配到最后一个 IP,就又从头开始分配。
type Store struct { *FileLock dataDir string}
每次存储操作都会进行加锁,IP 分配不会并发进行,确保唯一性。
a.store.Lock()defer a.store.Unlock()
本篇主要是从 Pod IP 管理的角度,梳理了一下从 kube-controller-manager 到 kubelet 的 Pod IP 管理过程。主要内容如下:
在工作中很多熟悉的路径,可能仅仅只是知道大概的流程,不知道具体的实现。通过源码分析,可以更加深入地了解相关的细节,也能学习到新的知识。
比如,在源码中,我看到了 InPlacePodVerticalScaling 这个参数,发现是 Kubernetes 1.27 的一个 alpha 特性,可以在不重启 Pod 的情况下,调整 Pod 的资源配置;在写 Operator 更新 CR 状态时,在合适的场景下,可以学习 nodeCIDRUpdateChannel 的实现,将更新的状态放入 channel 中,然后通过 goroutine 处理状态更新。
责任编辑:武晓燕 来源: 陈少文 KubernetesPod管理(责任编辑:热点)
评价结果显示:零售业务对商业银行收益可持续能力的贡献不断增强
四川巴中恩阳机场新增航线直通18个城市 去年旅客吞吐量38.2万人次