【containerd 源码分析】containerd cri 启动注册流程源码分析

谁践踏了优雅 2023-08-17 15:30 166阅读 0赞

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70][]

# CRI #

CRI（Container Runtime Interface）是 Kubernetes 定义的与 contianer runtime 进行交互的接口，将 Kubernetes 与特定的容器解耦。Kubernetes早期的版本，对于容器环境的支持是通过 hard code 方式直接调用 Docker API，支持更多的容器运行时和更精简的容器运行时，Kubernetes 提出了CRI。  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70 1][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70 2][]

Containerd 在 1.0 及以前版本将 dockershim 和 docker daemon 替换为 cri-containerd + containerd，在 1.1 版本直接将 cri-containerd 内置在 Containerd 中，作为一个 CRI 插件

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70 3][]

Containerd 内置的 CRI 插件实现了 Kubelet CRI 接口中的 Image Service 和 Runtime Service，通过内部接口管理容器和镜像，调用 CNI 插件给 Pod 配置网络

>     // grpcServices are all the grpc services provided by cri containerd.
>     type grpcServices interface {
>        runtime.RuntimeServiceServer
>        runtime.ImageServiceServer
>     }

# 1. init 函数注册 CRI 服务插件 #

路径 github.com/containerd/cri/cri.go，注册 CRI 服务插件，类型为 io.containerd.grpc.v1，ID 为 cri

> \# ctr plugins ls  
> TYPE                            ID                    PLATFORMS      STATUS      
> io.containerd.content.v1        content               -              ok          
> io.containerd.grpc.v1           version               -              ok          
> **io.containerd.grpc.v1           cri                   linux/amd64    ok**

// Register CRI service plugin
    func init() {
    	config := criconfig.DefaultConfig()
    	plugin.Register(&plugin.Registration{
    		Type:   plugin.GRPCPlugin,
    		ID:     "cri",
    		Config: &config,
    		Requires: []plugin.Type{
    			plugin.ServicePlugin,
    		},
    		InitFn: initCRIService,
    	})
    }

###      1.1 initCRIService 函数 ###

配置信息，配置文件 /etc/containerd 目录下 config.toml

*  Snapshotter:overlayfs
 *  DefaultRuntimeName:runc
 *  ContainerdRootDir：/var/lib/containerd
 *  ContainerdEndpoint：/run/containerd/containerd.sock
 *  RootDir：/var/lib/containerd/io.containerd.grpc.v1.cri
 *  StateDir：/run/containerd/io.containerd.grpc.v1.cri

func initCRIService(ic *plugin.InitContext) (interface{}, error) {
    	ic.Meta.Platforms = []imagespec.Platform{platforms.DefaultSpec()}
    	ic.Meta.Exports = map[string]string{"CRIVersion": constants.CRIVersion}
    	ctx := ic.Context
    	pluginConfig := ic.Config.(*criconfig.PluginConfig)
    	c := criconfig.Config{
    		PluginConfig:       *pluginConfig,
    		ContainerdRootDir:  filepath.Dir(ic.Root),
    		ContainerdEndpoint: ic.Address,
    		RootDir:            ic.Root,
    		StateDir:           ic.State,
    	}

**     1.1.1 getServicesOpts**

getServicesOpts 函数从 plugin 上下文获取服务选项，类型为** **io.containerd.service.v1，这个时实现了内部的服务

// getServicesOpts get service options from plugin context.
    func getServicesOpts(ic *plugin.InitContext) ([]containerd.ServicesOpt, error) {
    	plugins, err := ic.GetByType(plugin.ServicePlugin)
    	if err != nil {
    		return nil, errors.Wrap(err, "failed to get service plugin")
    	}

**     1.1.1.1 定义 map 一系列服务**

*  content-service
 *  images-service
 *  snapshots-service
 *  containers-service
 *  tasks-service
 *  diff-service
 *  namespaces-service
 *  leases-service

for s, fn := range map[string]func(interface{}) containerd.ServicesOpt{
    	services.ContentService: func(s interface{}) containerd.ServicesOpt {
    		return containerd.WithContentStore(s.(content.Store))
    	},
    	services.ImagesService: func(s interface{}) containerd.ServicesOpt {
    		return containerd.WithImageService(s.(images.ImagesClient))
    	},
    	services.SnapshotsService: func(s interface{}) containerd.ServicesOpt {
    		return containerd.WithSnapshotters(s.(map[string]snapshots.Snapshotter))
    	},
    	services.ContainersService: func(s interface{}) containerd.ServicesOpt {
    		return containerd.WithContainerService(s.(containers.ContainersClient))
    	},
    	services.TasksService: func(s interface{}) containerd.ServicesOpt {
    		return containerd.WithTaskService(s.(tasks.TasksClient))
    	},
    	services.DiffService: func(s interface{}) containerd.ServicesOpt {
    		return containerd.WithDiffService(s.(diff.DiffClient))
    	},
    	services.NamespacesService: func(s interface{}) containerd.ServicesOpt {
    		return containerd.WithNamespaceService(s.(namespaces.NamespacesClient))
    	},
    	services.LeasesService: func(s interface{}) containerd.ServicesOpt {
    		return containerd.WithLeasesService(s.(leases.Manager))
    	},
    }

**      1.1.2 containerd.New 函数**

路径 github.com/containerd/containerd/client.go，创建一个 client 连接到 containerd 实例

client, err := containerd.New(
    	"",
    	containerd.WithDefaultNamespace(constants.K8sContainerdNamespace),
    	containerd.WithDefaultPlatform(criplatforms.Default()),
    	containerd.WithServices(servicesOpts...),
    )

**      1.1.3 NewCRIService 函数实例化 criService **

// NewCRIService returns a new instance of CRIService
    func NewCRIService(config criconfig.Config, client *containerd.Client) (CRIService, error) {
    	var err error
    	c := &criService{
    		config:             config,
    		client:             client,
    		os:                 osinterface.RealOS{},
    		sandboxStore:       sandboxstore.NewStore(),
    		containerStore:     containerstore.NewStore(),
    		imageStore:         imagestore.NewStore(client),
    		snapshotStore:      snapshotstore.NewStore(),
    		sandboxNameIndex:   registrar.NewRegistrar(),
    		containerNameIndex: registrar.NewRegistrar(),
    		initialized:        atomic.NewBool(false),
    	}

**     实现了接口 CRIService，包括如下：定义 pkg/server/services.go**

>     // CRIService is the interface implement CRI remote service server.
>     type CRIService interface {
>        Run() error
>        // io.Closer is used by containerd to gracefully stop cri service.
>        io.Closer
>        plugin.Service
>        grpcServices
>     }

**     grpcServices 实现了接口 Runtime 和 image 服务**

>     // grpcServices are all the grpc services provided by cri containerd.
>     type grpcServices interface {
>        runtime.RuntimeServiceServer
>        runtime.ImageServiceServer
>     }

# 2. Run 函数启动 CRI 服务 #

// Run starts the CRI service.
    func (c *criService) Run() error {
    	logrus.Info("Start subscribing containerd event")
    	c.eventMonitor.subscribe(c.client)
    
    	logrus.Infof("Start recovering state")
    	if err := c.recover(ctrdutil.NamespacedContext()); err != nil {
    		return errors.Wrap(err, "failed to recover state")
    	}
    
    	// Start event handler.
    	logrus.Info("Start event monitor")
    	eventMonitorErrCh := c.eventMonitor.start()

###     2.1 recover 函数 ###

当 CRI down 掉重启进行恢复工作

// NOTE: The recovery logic has following assumption: when the cri plugin is down:
    // 1) Files (e.g. root directory, netns) and checkpoint maintained by the plugin MUST NOT be
    // touched. Or else, recovery logic for those containers/sandboxes may return error.
    // 2) Containerd containers may be deleted, but SHOULD NOT be added. Or else, recovery logic
    // for the newly added container/sandbox will return error, because there is no corresponding root
    // directory created.
    // 3) Containerd container tasks may exit or be stoppped, deleted. Even though current logic could
    // tolerant tasks being created or started, we prefer that not to happen.
    
    // recover recovers system state from containerd and status checkpoint.
    func (c *criService) recover(ctx context.Context) error {

**     2.1.1 对所有 sandbox 进行恢复**

**    client.Containers 列出所有 sandbox 的容器，根据 lable **io.cri-containerd.kind = sandbox

loadSandbox 加载 sandbox 的 metadata，NewSandbox 实例化 Sandbox 包括 metadata，status， 网络 namespace

sandboxStore.Add 把 sandbox 加入到缓存中

// Recover all sandboxes.
    sandboxes, err := c.client.Containers(ctx, filterLabel(containerKindLabel, containerKindSandbox))
    	if err != nil {
    		return errors.Wrap(err, "failed to list sandbox containers")
    	}
    
    for _, sandbox := range sandboxes {
    	sb, err := c.loadSandbox(ctx, sandbox)
    	if err != nil {
    		log.G(ctx).WithError(err).Errorf("Failed to load sandbox %q", sandbox.ID())
    		continue
    	}
    	log.G(ctx).Debugf("Loaded sandbox %+v", sb)
    	if err := c.sandboxStore.Add(sb); err != nil {
    		return errors.Wrapf(err, "failed to add sandbox %q to store", sandbox.ID())
    	}
    	if err := c.sandboxNameIndex.Reserve(sb.Name, sb.ID); err != nil {
    		return errors.Wrapf(err, "failed to reserve sandbox name %q", sb.Name)
    	}
    }

**     2.1.2 恢复所有 container**

列出所有 container，根据 label io.cri-containerd.kind = container 过滤

// Recover all containers.
    containers, err := c.client.Containers(ctx, filterLabel(containerKindLabel, containerKindContainer))
    if err != nil {
    	return errors.Wrap(err, "failed to list containers")
    }
    for _, container := range containers {
    	cntr, err := c.loadContainer(ctx, container)
    	if err != nil {
    		log.G(ctx).WithError(err).Errorf("Failed to load container %q", container.ID())
    		continue
    	}
    	log.G(ctx).Debugf("Loaded container %+v", cntr)
    	if err := c.containerStore.Add(cntr); err != nil {
    		return errors.Wrapf(err, "failed to add container %q to store", container.ID())
    	}
    	if err := c.containerNameIndex.Reserve(cntr.Name, cntr.ID); err != nil {
    		return errors.Wrapf(err, "failed to reserve container name %q", cntr.Name)
    	}
    }

**     2.1.3 恢复所有 image，加载到 cache**

// Recover all images.
    cImages, err := c.client.ListImages(ctx)
    if err != nil {
    	return errors.Wrap(err, "failed to list images")
    }
    c.loadImages(ctx, cImages)

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70]: /images/20230810/f78790bf8c79446e8e25eb5d34afcbe6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70 1]: /images/20230810/81fde7a18fb443399149b833dcd81d86.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70 2]: /images/20230810/a87a6432407844ef8ceafd8e7aa650ab.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70 3]: /images/20230810/7d389cee074e4c47bb457de16c1489e6.png