Source Code Analysis of Kubernetes Pod IP Management

1. Subnet Management by kube-controller-manager

kube-controller-manager has many controllers; the one related to Pod IP is NodeIpamController.

The NodeIpamController mainly manages the podcidr of nodes: when a new node joins the cluster, it allocates a subnet to that node; when a node is deleted, it reclaims the subnet.

The subnets of different nodes do not overlap, and each node can complete Pod IP allocation independently.

Below is an example of a running kube-controller-manager:

1
kubectl -n kube-system get pod kube-controller-manager -o yaml

The part about subnet configuration is:

1
2
3
4
5
6
7
8
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 specifies the range of Pod IPs with a mask length of 16. If reserved IPs are not considered, it means the cluster can hold at most 2^16 = 65536 pods.

These pods are distributed across several nodes. Next, node-cidr-mask-size is 24, so each node has only 32-24=8 bits left for pods, and each node can create at most 2^8=256 pods.

Correspondingly, the number of nodes this cluster can hold is 2^(32-16-8)=256 nodes.

When planning a cluster, you need to adjust these two parameters according to the scale of the cluster.

After enabling allocate-node-cidrs and setting cluster-cidr, kube-controller-manager allocates a subnet to each node and writes the result into the spec.podCIDR field.

1
2
3
4
spec:
  podCIDR: 10.234.58.0/24
  podCIDRs:
    - 10.234.58.0/24

Now let us analyze this process from the source code.

1. Starting NodeIpamController

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
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 just provides some monitoring metrics; the real startup logic is in the Run method.

1
2
3
4
5
func (nc *Controller) RunWithMetrics(ctx context.Context, controllerManagerMetrics *controllersmetrics.ControllerManagerMetrics) {
	controllerManagerMetrics.ControllerStarted("nodeipam")
	defer controllerManagerMetrics.ControllerStopped("nodeipam")
	nc.Run(ctx)
}

1
2
3
4
5
6
7
8
9
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()
}

1.2 Watching for Node Changes

While looking for the implementation of the cidrAllocator interface, I found three CIDR allocators: RangeAllocator for single-subnet allocation, MultiCIDRRangeAllocator for multiple CIDRs, and CloudCIDRAllocator for integrating with cloud providers.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
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)
	}
}

Let us look at the implementation of RangeAllocator here.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
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
}

In fact, the implementation of the RangeAllocator is similar to the controller you write when building an Operator: both watch resource changes through an informer and then call the corresponding methods.

1.3 Updating the Node’s podCIDR

What is special here is that the controller does not operate on the resource directly; instead it puts the change into a channel and then processes the status update through goroutines.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
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
}

The length of nodeCIDRUpdateChannel is 5000.

1
2
	cidrUpdateQueueSize = 5000
	nodeCIDRUpdateChannel: make(chan nodeReservedCIDRs, cidrUpdateQueueSize),

The logic that updates the Node Spec is handled by 30 goroutines.

1
2
3
4
	const cidrUpdateWorkers untyped int = 30
	for i := 0; i < cidrUpdateWorkers; i++ {
		go r.worker(ctx)
	}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
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
		}
	}
}

With cidrUpdateRetries = 3, it retries the update 3 times here; if the update keeps failing, the node is put back into the channel to wait for the next update.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
// 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")
}

The Patch method is used to update the Spec field of the node object.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
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
}

2. Network Configuration by kubelet

The figure above shows the process of Kubelet creating a Pod. Here we take the network configuration part of it for analysis:

The Pod is scheduled onto a node
kubelet calls the container runtime through cri to create the sandbox
The container runtime creates the sandbox
The container runtime calls cni to create the Pod network
IPAM manages the Pod IP

Now let us look at this process from the perspective of the source code implementation.

2.1 The Pod Is Scheduled onto a Node

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
apiVersion: v1
kind: Pod
metadata:
  labels:
    app: demo
    pod-template-hash: 7b9b5cf76b
  name: demo-7b9b5cf76b-5lpmj
  namespace: default
spec:
  containers:
    - image: shaowenchen/demo:ubuntu
  nodeName: node1

The scheduling process in Kubernetes is that kube-scheduler schedules the Pod onto a node according to the Pod’s resource requirements and the node’s resource situation, and writes the scheduling result into the pod.spec.nodeName field.

This part is not the focus of networking. I have also customized a scheduler in a production environment before; if you are interested, take a look at Tekton Optimization: Customizing the Cluster Scheduler.

2.2 kubelet Calls cri to Create the sandbox

SyncPod is the core method in kubelet; it calls cri to create or delete the pod according to the Pod’s status.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
// 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)
}

It calls the RunPodSandbox method of the RuntimeService interface to create the sandbox.

1
2
3
// 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)

After being wrapped by the runtimeService and instrumentedRuntimeService interfaces, it ultimately calls the RunPodSandbox method of remoteRuntimeService.

1
2
3
4
5
6
7
// 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,
	})

Here runtimeClient is an rpc client; it calls the container runtime through rpc to create the sandbox.

2.3 The container runtime Creates the sandbox

Taking containerd as an example, to create the sandbox:

1
2
3
4
5
6
7
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)
}

It calls CNI to create the network and create the sandbox.

 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
// 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)
}

2.4 The container runtime Calls cni to Create the Pod Network

In the previous step, before calling RunPodSandbox to create the sandbox, it first calls setupPodNetwork to configure the network. Let us expand on the implementation of setupPodNetwork here.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
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 implements the netPlugin interface

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
// containerd/go-cni/cni.go
func (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 spins up many goroutines; each goroutine calls the asynchAttach method, and the asynchAttach method calls the Attach method.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
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
}

It runs many goroutines calling CNI, but the rc channel has a length of 1, so the results are still processed one by one.

1
2
3
4
5
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}
}

It is only in the Attach method that the CNI plugin is really called.

1
2
3
4
5
6
7
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)
}

In https://github.com/containernetworking/cni/blob/main/libcni/api.go the CNI interface defines many methods, among which the most important are the AddNetwork and DelNetwork methods; the methods with List are batch operations.

1
2
3
4
5
6
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 is used to add a network interface to the container, creating a veth NIC on the host bound to the container’s ech0 NIC. DelNetwork is used to clean up the container-related network configuration when the container is deleted.

The core of CNI calling a plugin is the Exec interface, which directly calls the binary program.

1
2
3
4
5
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 passes the network configuration information to the CNI plugin in the form of standard input and environment variables. After the CNI plugin finishes processing, it writes the network configuration information to standard output; CRI parses the network configuration information from standard output and writes it into the container’s network configuration file.

Back to the implementation of the container runtime, containerd:

1
2
3
4
5
/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"

Here /etc/cni/net.d is the default location for CNI network configuration files, and /opt/cni/bin is the default search path for CNI network plugins.

1
2
3
4
ls /opt/cni/bin

bandwidth  calico       cilium-cni  firewall  host-device  install  loopback  portmap  sbr     tuning  vrf
bridge     calico-IPAM  dhcp        flannel   host-local   ipvlan   macvlan   ptp      static  vlan

1
2
3
4
5
6
7
cat /etc/cni/net.d/05-cilium.conf
{
  "cniVersion": "0.3.1",
  "name": "cilium",
  "type": "cilium-cni",
  "enable-debug": false
}

These configurations are used to initialize the netPlugin map[string]cni.CNI structure that CRI uses to obtain CNI plugins.

2.5 IPAM’s Management of Pod IPs

IPAM is short for IP Address Management; it is responsible for allocating ip addresses to containers. The IPAM component is usually a standalone binary, and it can also be implemented directly by a CNI plugin. In https://github.com/containernetworking/plugins/tree/main/plugins/ipam there are currently three implementations: host-local, dhcp, and static. Let us take host-local as an example:

Look at the CNI configuration file

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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" }]
  }
}

It specifies the CNI plugin type as host-local and specifies the Pod IP subnet as “10.234.58.0/24”.

Look at the CNI plugin’s storage directory

1
2
3
ls /var/lib/cni/networks

10.234.58.76  10.234.58.87  last_reserved_ip.0  lock

1
2
3
cat 10.234.58.76

b3b668af977bbeca6853122514044865793c056e81cccebf115dacffd25a8bcc

There is a group of files named after ip addresses, and the contents of the files are a string of characters. So what exactly are these?

How the files named after ip addresses are generated

When requesting a Pod IP, first obtain an available ip

1
2
3
4
5
func cmdAdd(args *skel.CmdArgs) error {
	for idx, rangeset := range ipamConf.Ranges {
		ipConf, err := allocator.Get(args.ContainerID, args.IfName, requestedIP)
	}
}

After obtaining an available ip, first try to store it in a local directory file

1
2
3
4
5
6
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)
	}
}

Write directly to the local file directory

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
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
	}
}

The content written is strings.TrimSpace(id) + LineBreak + ifname, where id is actually the container id, ifname is the NIC name, and LineBreak is the newline character.

Using the id, you can find the corresponding container on the host:

1
2
3
docker ps |grep b3b668

b3b668af977b   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

The purpose of the last_reserved_ip.0 file

1
2
3
cat last_reserved_ip.0

10.234.58.87

When obtaining an available IP, IPAM creates an iterator.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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
        }
    }

The iterator needs to rely on last_reserved_ip.0 to find the last allocated IP, and then start allocating after that IP.

1
2
3
4
5
6
7
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)
	}

Here lastIPFilePrefix = “last_reserved_ip.”

1
2
3
4
5
6
7
8
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
}

When host-local allocates ip addresses, it does so in a round-robin, incrementing fashion; if it allocates to the last IP, it starts allocating from the beginning again.

The lock file

1
2
3
4
type Store struct {
	*FileLock
	dataDir string
}

Every store operation acquires a lock, so IP allocation does not happen concurrently, ensuring uniqueness.

1
2
a.store.Lock()
defer a.store.Unlock()

3. Summary

This article mainly sorts out the Pod IP management process from kube-controller-manager to kubelet from the perspective of Pod IP management. The main contents are as follows:

kube-controller-manager allocates a Pod IP subnet to each node through the NodeIpamController; when planning a cluster, you need to adjust the cluster-cidr and node-cidr-mask-size parameters according to the cluster scale
kubelet calls the container runtime through cri to create the sandbox
The container runtime calls cni to create the Pod network
IPAM’s management of Pod IPs

At work, many paths you are familiar with may only be known at the level of the general flow, without knowing the specific implementation. Through source code analysis, you can understand the relevant details more deeply and also learn new knowledge.

For example, in the source code I saw the InPlacePodVerticalScaling parameter and found that it is an alpha feature of Kubernetes 1.27 that can adjust a Pod’s resource configuration without restarting the Pod; when writing an Operator to update the CR status, in suitable scenarios you can learn from the implementation of nodeCIDRUpdateChannel by putting the updated status into a channel and then handling the status update through a goroutine.