【kubernetes/k8s源码分析】kubelet cri源码分析

深藏阁楼爱情的钟 2021-11-04 16:54 420阅读 0赞

### CRI基本原理 ###

>       早期的 kubernetes 使用 docker 作为默认的 runtime，后来又加入 rkt，每加入一种新运行时，k8s 都要修改接口。container runtime 加入，不同的容器的实现和功能差异很大。标准就是 [CRI][]（Container RunTime Interface）。k8s 不再关心底层的容器运行时，kubelet只感知到 CRI server，而 CRI server 只需遵循 CRI 标准实现

CRI 接口定义在 k8s.io/cri-api/pkg/apis/services.go，看看具体定义了哪些接口

### RuntimeService 接口 ###

**    RuntimeVersioner，负责返回 runtime 名字，版本以及 API 版本，一个接口函数 Version**

*  Version(apiVersion string) (\*runtimeapi.VersionResponse, error)

**    ContainerManager，对 container 的操作接口，简单明了**

*  CreateContainer(podSandboxID string, config \*runtimeapi.ContainerConfig, sandboxConfig \*runtimeapi.PodSandboxConfig) (string, error)
 *  StartContainer(containerID string) error
 *  StopContainer(containerID string, timeout int64) error
 *  RemoveContainer(containerID string) error
 *  ListContainers(filter \*runtimeapi.ContainerFilter) (\[\]\*runtimeapi.Container, error)
 *  ContainerStatus(containerID string) (\*runtimeapi.ContainerStatus, error)
 *  UpdateContainerResources(containerID string, resources \*runtimeapi.LinuxContainerResources) error
 *  ExecSync(containerID string, cmd \[\]string, timeout time.Duration) (stdout \[\]byte, stderr \[\]byte, err error)
 *  Exec(\*runtimeapi.ExecRequest) (\*runtimeapi.ExecResponse, error)
 *  Attach(req \*runtimeapi.AttachRequest) (\*runtimeapi.AttachResponse, error)
 *  ReopenContainerLog(ContainerID string) error

**    PodSandboxManager，负责操作 sandbox （Pause 容器或者虚拟机），是线程安全的，包括方法如下：**

*  RunPodSandbox(config \*runtimeapi.PodSandboxConfig, runtimeHandler string) (string, error)
 *  StopPodSandbox(podSandboxID string) error
 *  RemovePodSandbox(podSandboxID string) error
 *  PodSandboxStatus(podSandboxID string) (\*runtimeapi.PodSandboxStatus, error)
 *  ListPodSandbox(filter \*runtimeapi.PodSandboxFilter) (\[\]\*runtimeapi.PodSandbox, error)
 *  PortForward(\*runtimeapi.PortForwardRequest) (\*runtimeapi.PortForwardResponse, error)

状态查询接口等

>     // RuntimeService interface should be implemented by a container runtime.
>     // The methods should be thread-safe.
>     type RuntimeService interface {
>        RuntimeVersioner
>        ContainerManager
>        PodSandboxManager
>        ContainerStatsManager
>     
>        // UpdateRuntimeConfig updates runtime configuration if specified
>        UpdateRuntimeConfig(runtimeConfig *runtimeapi.RuntimeConfig) error
>        // Status returns the status of the runtime.
>        Status() (*runtimeapi.RuntimeStatus, error)
>     }

### ImageManagerService 接口 ###

定义了镜像的接口，包括查询，下载，删除等操作

// ImageManagerService interface should be implemented by a container image
    // manager.
    // The methods should be thread-safe.
    type ImageManagerService interface {
    	// ListImages lists the existing images.
    	ListImages(filter *runtimeapi.ImageFilter) ([]*runtimeapi.Image, error)
    	// ImageStatus returns the status of the image.
    	ImageStatus(image *runtimeapi.ImageSpec) (*runtimeapi.Image, error)
    	// PullImage pulls an image with the authentication config.
    	PullImage(image *runtimeapi.ImageSpec, auth *runtimeapi.AuthConfig, podSandboxConfig *runtimeapi.PodSandboxConfig) (string, error)
    	// RemoveImage removes the image.
    	RemoveImage(image *runtimeapi.ImageSpec) error
    	// ImageFsInfo returns information of the filesystem that is used to store images.
    	ImageFsInfo() ([]*runtimeapi.FilesystemUsage, error)
    }

> RunKubelet
> 
>         -->  createAndInitKubelet
> 
>                  -->  NewMainKubelet
> 
>                          -->  getRuntimeAndImageServices （第 1 章节讲解）
> 
>                          -->  NewKubeGenericRuntimeManager

# 1. getRuntimeAndImageServices 函数 #

NewRemoteRuntimeService 实例化 remote runtime 服务，实现了上述的 RuntimeService 接口

NewRemoteImageService 实例化 remote 镜像服务，实现了上述的 ImageManagerService 接口

func getRuntimeAndImageServices(remoteRuntimeEndpoint string, remoteImageEndpoint string, runtimeRequestTimeout metav1.Duration) (internalapi.RuntimeService, internalapi.ImageManagerService, error) {
    	rs, err := remote.NewRemoteRuntimeService(remoteRuntimeEndpoint, runtimeRequestTimeout.Duration)
    	if err != nil {
    		return nil, nil, err
    	}
    	is, err := remote.NewRemoteImageService(remoteImageEndpoint, runtimeRequestTimeout.Duration)
    	if err != nil {
    		return nil, nil, err
    	}
    	return rs, is, err
    }

###     1.1 NewRemoteRuntimeService 函数 ###

实现是与 remote endpoint 建立 GRPC 连接，实例化 RemoteRuntimeService，实现了 RuntimeService 接口，实现路径在

pkg/kubelet/remote/remote\_runtime.go 中，接口的实现是调用 GRPC 客户端发送 GRPC 请求

// NewRemoteRuntimeService creates a new internalapi.RuntimeService.
    func NewRemoteRuntimeService(endpoint string, connectionTimeout time.Duration) (internalapi.RuntimeService, error) {
    	klog.V(3).Infof("Connecting to runtime service %s", endpoint)
    	addr, dailer, err := util.GetAddressAndDialer(endpoint)
    	if err != nil {
    		return nil, err
    	}
    	ctx, cancel := context.WithTimeout(context.Background(), connectionTimeout)
    	defer cancel()
    
    	conn, err := grpc.DialContext(ctx, addr, grpc.WithInsecure(), grpc.WithDialer(dailer), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMsgSize)))
    	if err != nil {
    		klog.Errorf("Connect remote runtime %s failed: %v", addr, err)
    		return nil, err
    	}
    
    	return &RemoteRuntimeService{
    		timeout:       connectionTimeout,
    		runtimeClient: runtimeapi.NewRuntimeServiceClient(conn),
    		logReduction:  logreduction.NewLogReduction(identicalErrorDelay),
    	}, nil
    }

**     1.1.1 RunPodSandbox 创建 sandbox**

向 remote server 发送 GRPC RunPodSandboxRequest 请求

// RunPodSandbox creates and starts a pod-level sandbox. Runtimes should ensure
    // the sandbox is in ready state.
    func (r *RemoteRuntimeService) RunPodSandbox(config *runtimeapi.PodSandboxConfig, runtimeHandler string) (string, error) {
    	// Use 2 times longer timeout for sandbox operation (4 mins by default)
    	// TODO: Make the pod sandbox timeout configurable.
    	ctx, cancel := getContextWithTimeout(r.timeout * 2)
    	defer cancel()
    
    	resp, err := r.runtimeClient.RunPodSandbox(ctx, &runtimeapi.RunPodSandboxRequest{
    		Config:         config,
    		RuntimeHandler: runtimeHandler,
    	})
    	if err != nil {
    		klog.Errorf("RunPodSandbox from runtime service failed: %v", err)
    		return "", err
    	}
    
    	if resp.PodSandboxId == "" {
    		errorMessage := fmt.Sprintf("PodSandboxId is not set for sandbox %q", config.GetMetadata())
    		klog.Errorf("RunPodSandbox failed: %s", errorMessage)
    		return "", errors.New(errorMessage)
    	}
    
    	return resp.PodSandboxId, nil
    }

###      1.2 NewRemoteImageService 函数 ###

实现方式是与 remote 建立 GRPC 连接，实例化 RemoteImageService，实现了上述的 ImageManagerService 接口

，实现路径在 pkg/kubelet/remote/remote\_image.go 中，接口的实现是调用 GRPC 客户端发送 GRPC 请求

// NewRemoteImageService creates a new internalapi.ImageManagerService.
    func NewRemoteImageService(endpoint string, connectionTimeout time.Duration) (internalapi.ImageManagerService, error) {
    	klog.V(3).Infof("Connecting to image service %s", endpoint)
    	addr, dailer, err := util.GetAddressAndDialer(endpoint)
    	if err != nil {
    		return nil, err
    	}
    
    	ctx, cancel := context.WithTimeout(context.Background(), connectionTimeout)
    	defer cancel()
    
    	conn, err := grpc.DialContext(ctx, addr, grpc.WithInsecure(), grpc.WithDialer(dailer), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMsgSize)))
    	if err != nil {
    		klog.Errorf("Connect remote image service %s failed: %v", addr, err)
    		return nil, err
    	}
    
    	return &RemoteImageService{
    		timeout:     connectionTimeout,
    		imageClient: runtimeapi.NewImageServiceClient(conn),
    	}, nil
    }

# 2. NewKubeGenericRuntimeManager 函数 #

实例化 kubeGenericRuntimeManager，实现了 KubeGenericRuntime 接口，定义在 pkg/kubelet/kuberuntime/kuberuntime\_manager.go 中，接口包括，容器运行时，以及命令接口

// NewKubeGenericRuntimeManager creates a new kubeGenericRuntimeManager
    func NewKubeGenericRuntimeManager(
    	recorder record.EventRecorder,
    	livenessManager proberesults.Manager,
    	seccompProfileRoot string,
    	containerRefManager *kubecontainer.RefManager,
    	machineInfo *cadvisorapi.MachineInfo,
    	podStateProvider podStateProvider,
    	osInterface kubecontainer.OSInterface,
    	runtimeHelper kubecontainer.RuntimeHelper,
    	httpClient types.HttpGetter,
    	imageBackOff *flowcontrol.Backoff,
    	serializeImagePulls bool,
    	imagePullQPS float32,
    	imagePullBurst int,
    	cpuCFSQuota bool,
    	cpuCFSQuotaPeriod metav1.Duration,
    	runtimeService internalapi.RuntimeService,
    	imageService internalapi.ImageManagerService,
    	internalLifecycle cm.InternalContainerLifecycle,
    	legacyLogProvider LegacyLogProvider,
    	runtimeClassManager *runtimeclass.Manager,
    ) (KubeGenericRuntime, error)

赋值 runtime

>     klet.containerRuntime = runtime
>     klet.streamingRuntime = runtime
>     klet.runner = runtime

[CRI]: https://yq.aliyun.com/go/articleRenderRedirect?url=https%3A%2F%2Fgithub.com%2Fkubernetes%2Fcommunity%2Fblob%2Fmaster%2Fcontributors%2Fdevel%2Fsig-node%2Fcontainer-runtime-interface.md