浅谈 K8s 网络模型CNI协议

$ mkdir -p /etc/cni/net.d
$ cat >/etc/cni/net.d/10-mynet.conf <<EOF
{
 "cniVersion": "0.2.0", // CNI Spec 版本
 "name": "mynet", // 自定义名称
 "type": "bridge", // 插件类型 bridge
 "bridge": "cni0", // 网桥名称
 "isGateway": true, // 是否作为网关
 "ipMasq": true, // 是否设置 IP 伪装
 "ipam": {
  "type": "host-local", // IPAM 类型 host-local
  "subnet": "10.22.0.0/16", // 子网段
  "routes": [
   { "dst": "0.0.0.0/0" } // 目标路由段
  ]
 }
}
EOF
$ cat >/etc/cni/net.d/99-loopback.conf <<EOF
{
 "cniVersion": "0.2.0", // CNI Spec 版本
 "name": "lo", // 自定义名称
 "type": "loopback" // 插件类型 loopback
}
EOF

// kubernetes/cmd/kubelet/kubelet.go
func main() {
 command := app.NewKubeletCommand()

 // kubelet uses a config file and does its own special
 // parsing of flags and that config file. It initializes
 // logging after it is done with that. Therefore it does
 // not use cli.Run like other, simpler commands.
 code := run(command)
 os.Exit(code)
}

// kubernetes/pkg/kubelet/pleg/pleg.go
// 通过 PLEG 进行 Pod 生命周期事件管理
type PodLifecycleEventGenerator interface {
 Start() // 通过 relist 获取所有 Pods 并计算事件类型
 Watch() chan *PodLifecycleEvent // 监听 eventChannel，传递给下游消费者
 Healthy() (bool, error)
}

// kubernetes/pkg/kubelet/pleg/generic.go
// 生产者：获取所有 Pods 列表，计算出对应的事件类型，进行 Sync
func (g *GenericPLEG) relist() {
 klog.V(5).InfoS("GenericPLEG: Relisting")
 ...
 // 获取当前所有 Pods 列表
 podList, err := g.runtime.GetPods(true)
 if err != nil {
  klog.ErrorS(err, "GenericPLEG: Unable to retrieve pods")
  return
 }

 for pid := range g.podRecords {
  allContainers := getContainersFromPods(oldPod, pod)
  for _, container := range allContainers {

   // 计算事件类型：running/exited/unknown/non-existent
   events := computeEvents(oldPod, pod, &container.ID)
   for _, e := range events {
    updateEvents(eventsByPodID, e)
   }
  }
 }

 // 遍历所有事件
 for pid, events := range eventsByPodID {
  for i := range events {
   // Filter out events that are not reliable and no other components use yet.
   if events[i].Type == ContainerChanged {
    continue
   }
   select {
   case g.eventChannel <- events[i]: // 生产者：发送到事件 channel，对应监听的 goroutine 会消费
   default:
    metrics.PLEGDiscardEvents.Inc()
    klog.ErrorS(nil, "Event channel is full, discard this relist() cycle event")
   }
  }
 }
 ...
}

// kubernetes/pkg/kubelet/kubelet.go
// 消费者：根据 channel 获取的各类事件，进行 Pod Sync
func (kl *Kubelet) syncLoopIteration(configCh <-chan kubetypes.PodUpdate, handler SyncHandler,
 syncCh <-chan time.Time, housekeepingCh <-chan time.Time, plegCh <-chan *pleg.PodLifecycleEvent) bool {
 select {
 ...
 // 消费者：监听 plegCh 的事件
 case e := <-plegCh:
  if e.Type == pleg.ContainerStarted {
   // 更新容器的最后启动时间
   kl.lastContainerStartedTime.Add(e.ID, time.Now())
  }
  if isSyncPodWorthy(e) {
   if pod, ok := kl.podManager.GetPodByUID(e.ID); ok {
    klog.V(2).InfoS("SyncLoop (PLEG): event for pod", "pod", klog.KObj(pod), "event", e)

    // 进行相关 Pod 事件的 Sync
    handler.HandlePodSyncs([]*v1.Pod{pod})
   } else {
    // If the pod no longer exists, ignore the event.
    klog.V(4).InfoS("SyncLoop (PLEG): pod does not exist, ignore irrelevant event", "event", e)
   }
  }

  // 容器销毁事件处理：清除 Pod 内相关 Container
  if e.Type == pleg.ContainerDied {
   if containerID, ok := e.Data.(string); ok {
    kl.cleanUpContainersInPod(e.ID, containerID)
   }
  }
  ...
 }
 return true
}

// kubernetes/pkg/kubelet/dockershim/docker_sandbox.go
// 启动运行 Pod Sandbox
func (ds *dockerService) RunPodSandbox(ctx context.Context, r *runtimeapi.RunPodSandboxRequest) (*runtimeapi.RunPodSandboxResponse, error) {
 config := r.GetConfig()

 // Step 1: 拉取基础镜像(infra-container: k8s.gcr.io/pause:3.6)
 image := defaultSandboxImage
 if err := ensureSandboxImageExists(ds.client, image); err != nil {
  return nil, err
 }

 // Step 2: 创建 Sandbox 容器
 createConfig, err := ds.makeSandboxDockerConfig(config, image)
 if err != nil {
  return nil, fmt.Errorf("failed to make sandbox docker config for pod %q: %v", config.Metadata.Name, err)
 }
 createResp, err := ds.client.CreateContainer(*createConfig)
 if err != nil {
  createResp, err = recoverFromCreationConflictIfNeeded(ds.client, *createConfig, err)
 }

 // Step 3: 创建 Sandbox 检查点(用于记录当前执行到哪一步了)
 if err = ds.checkpointManager.CreateCheckpoint(createResp.ID, constructPodSandboxCheckpoint(config)); err != nil {
  return nil, err
 }

 // Step 4: 启动 Sandbox 容器
 err = ds.client.StartContainer(createResp.ID)
 if err != nil {
  return nil, fmt.Errorf("failed to start sandbox container for pod %q: %v", config.Metadata.Name, err)
 }

 // Step 5: 对 Sandbox 容器进行网络配置
 err = ds.network.SetUpPod(config.GetMetadata().Namespace, config.GetMetadata().Name, cID, config.Annotations, networkOptions)
 if err != nil {
  // 如果网络配置失败，则回滚：删除建立起来的 Pod 网络
  err = ds.network.TearDownPod(config.GetMetadata().Namespace, config.GetMetadata().Name, cID)
  if err != nil {
   errList = append(errList, fmt.Errorf("failed to clean up sandbox container %q network for pod %q: %v", createResp.ID, config.Metadata.Name, err))
  }

  // 停止容器运行
  err = ds.client.StopContainer(createResp.ID, defaultSandboxGracePeriod)
  ...
 }

 return resp, nil
}