kubelet与containerd交互概览

下图展示了kubelet如何与containerd进行交互，containred通过cri插件实现了CRI的接口。

1. kubelet通过grpc调用containerd的RunSanbox方法，创建了Pod运行的基本环境，包括pause容器和containerd-shim。
2. kubelet通过grpc调用containerd的CreateContainer和StartContainer创建Pod中的容器，包括initContainer和normalContainer。

创建pod

创建一个使用busybox镜像且cmd设置sleep 3000000的pod。下图是使用bcc工具exec-snoop抓取了containerd创建Pod的具体流程，其中省略了部分runc调用的细节。
1、 调用container-shim start 启动用于创建runc的containerd-shim，第一个containerd-shim（pid 20792）退出后，第二个containerd-shim（pid 20799）的父进程就变成了1，这样containerd-shim就与containerd脱离了关系，重启containerd也不会影响containerd-shim进程。
2、通过ttrpc调用第二个containerd-shim（pid 20799）的Newtask方法，会调用runc create。
3、再通过ttrpc调用第二个containerd-shim（pid 20799）的Start方法，会调用runc start启动pause容器。
4、以同样的方式启动Pod中定义的container，注意这次的containerd-shim只有start调用，原因是1中创建的第二个containerd-shim（pid 20799）是可以与runc交互的进程，这里不用重复创建

注意：containerd-shim 20802 和 20803 是线程

systemd(1)─┬
           ├─containerd(1694)─┬─{containerd}(1696)
           ├─containerd-shim(20799)─┬─pause(20821)
           │                        ├─sleep(20852)
           │                        ├─{containerd-shim}(20802)
           │                        ├─{containerd-shim}(20803)
containerd-shim  20792  1694     0 /usr/bin/containerd-shim-runc-v2 -namespace k8s.io -address /run/containerd/containerd.sock -publish-binary /usr/bin/containerd -id 193727d35be73014e4adfe4ba93aa20b3834654a65990b9de9fba123592044aa -debug start
containerd-shim  20799  20792    0 /usr/bin/containerd-shim-runc-v2 -namespace k8s.io -id 193727d35be73014e4adfe4ba93aa20b3834654a65990b9de9fba123592044aa -address /run/containerd/containerd.sock
runc             20811  20799    0 /usr/bin/runc --root /run/containerd/runc/k8s.io --log /run/containerd/io.containerd.runtime.v2.task/k8s.io/193727d35be73014e4adfe4ba93aa20b3834654a65990b9de9fba123592044aa/log.json --log-format json --systemd-cgroup create --bundle /run/containerd/io.containerd.runtime.v2.task/k8s.io/193727d35be73014e4adfe4ba93aa20b3834654a65990b9de9fba123592044aa --pid-file /run/containerd/io.containerd.runtime.v2.task/k8s.io/193727d35be73014e4adfe4ba93aa20b3834654a65990b9de9fba123592044aa/init.pid 193727d35be73014e4adfe4ba93aa20b3834654a65990b9de9fba123592044aa/run/containerd/io.containerd.runtime.v2.task/k8s.io/193727d35be73014e4adfe4ba93aa20b3834654a65990b9de9fba123592044aa --pid-file /run/containerd/io.containerd.runtime.v2.task/k8s.io/193727d35be73014e4adfe4ba93aa20b3834654a65990b9de9fba123592044aa/init.pid 193727d35be73014e4adfe4ba93aa20b3834654a65990b9de9fba123592044aa
runc             20827  20799    0 /usr/bin/runc --root /run/containerd/runc/k8s.io --log /run/containerd/io.containerd.runtime.v2.task/k8s.io/193727d35be73014e4adfe4ba93aa20b3834654a65990b9de9fba123592044aa/log.json --log-format json --systemd-cgroup start 193727d35be73014e4adfe4ba93aa20b3834654a65990b9de9fba123592044aa
pause            20821  20799    0 /pause
containerd-shim  20834  1694     0 /usr/bin/containerd-shim-runc-v2 -namespace k8s.io -address /run/containerd/containerd.sock -publish-binary /usr/bin/containerd -id f2ba685582ea81fb6568da6cec0ee3086dd8e15db11214ddfed19f4ded41725b -debug start
runc             20841  20799    0 /usr/bin/runc --root /run/containerd/runc/k8s.io --log /run/containerd/io.containerd.runtime.v2.task/k8s.io/f2ba685582ea81fb6568da6cec0ee3086dd8e15db11214ddfed19f4ded41725b/log.json --log-format json --systemd-cgroup create --bundle /run/containerd/io.containerd.runtime.v2.task/k8s.io/f2ba685582ea81fb6568da6cec0ee3086dd8e15db11214ddfed19f4ded41725b --pid-file /run/containerd/io.containerd.runtime.v2.task/k8s.io/f2ba685582ea81fb6568da6cec0ee3086dd8e15db11214ddfed19f4ded41725b/init.pid f2ba685582ea81fb6568da6cec0ee3086dd8e15db11214ddfed19f4ded41725b
runc             20859  20799    0 /usr/bin/runc --root /run/containerd/runc/k8s.io --log /run/containerd/io.containerd.runtime.v2.task/k8s.io/f2ba685582ea81fb6568da6cec0ee3086dd8e15db11214ddfed19f4ded41725b/log.json --log-format json --systemd-cgroup start f2ba685582ea81fb6568da6cec0ee3086dd8e15db11214ddfed19f4ded41725b
sleep            20852  20799    0 /bin/sleep 30000000

containerd的CRI插件

containerd的CRI的实现部分通过插件的形式集成到了containerd中，这种插件化的方式使得扩展containerd变得相对容易，并且核心代码几乎不需要变动。CRI注册了自己特有的grpc方法到containerd中，这样外部就可以通过调用统一的containerd的grpc方法进而调用到CRI的grpc方法。

containerd-shim

containerd启动containerd-shim，containerd-shim再启动一个containerd-shim，此时先前containerd-shim退出。新的container-shim进程的父进程就变成了1，此containerd-shim就是用户init程序的父进程。

这里讨论的其实都是 containerd-shim-runc-v2。这是containerd现在用的shim。containerd代码库中还有一个containerd-shim，这个是dockerd在用的 https://stackoverflow.com/questions/67585798/what-process-is-the-parent-process-of-containerd-shim

func (m *TaskManager) Create(ctx context.Context, id string, opts runtime.CreateOpts) (_ runtime.Task, retErr error) {
    bundle, err := NewBundle(ctx, m.root, m.state, id, opts.Spec.Value)
     //  启动containerd-shim进程
    shim, err := m.startShim(ctx, bundle, id, opts)
    //  向第二个containerd-shim发送create指令
    t, err := shim.Create(ctx, opts)
    if err := m.tasks.Add(ctx, t); err != nil 
    return t, nil
}
func (b *binary) Start(ctx context.Context, opts *types.Any, onClose func()) (_ *shim, err error) {
    args := []string{"-id", b.bundle.ID}
    args = append(args, "start")
    cmd, err := client.Command(
        ctx,
        b.runtime,
        b.containerdAddress,
        b.containerdTTRPCAddress,
        b.bundle.Path,
        opts,
        args...,
    )
    out, err := cmd.CombinedOutput()
    // 返回一个unix socket
    address := strings.TrimSpace(string(out))
    conn, err := client.Connect(address, client.AnonDialer)
   // 新建与unix socket的连接
    client := ttrpc.NewClient(conn, ttrpc.WithOnClose(onCloseWithShimLog))
    return &shim{
        bundle:  b.bundle,
        client:  client,
        task:    task.NewTaskClient(client),
        events:  b.events,
        rtTasks: b.rtTasks,
    }
}
// 第二个containerd-shim启动grpc服务
func (s *Client) Serve() {
    shimapi.RegisterTaskService(server, s.service)
}

containerd-shim 向 containerd转发事件

cri服务处理事件，进程退出，oom等，并更新container或者sandbox的状态。之后kubelet就可以调用CRI接口获取container的状态信息，包括oom状态码等。

// containerd-shim
func run(id string, initFunc Init, config Config) error {
    ttrpcAddress := os.Getenv(ttrpcAddressEnv)
    // 建立到containerd的连接 
    publisher, err := NewPublisher(ttrpcAddress)
}
// 开始处理shim的事件并转发到containerd
func (l *RemoteEventsPublisher) processQueue() {
  forwardRequest（）
}
func (l *RemoteEventsPublisher) forwardRequest(ctx context.Context, req *v1.ForwardRequest) error {
    // 到containerd的events的grpc接口
    service, err := l.client.EventsService()
    service.Forward(fCtx, req)
}
// containerd ：
// 发布消息
func (e *Exchange) Publish(ctx context.Context, topic string, event events.Event) (err error) {
    validateTopic(topic)
    return e.broadcaster.Write(&envelope)
}
// 向订阅者发布消息
func (b *Broadcaster) run() {
    for {
        select {
        case event := <-b.events:
            for _, sink := range b.sinks {
                sink.Write(event)
                }
            }
        }
}
// cri订阅事件
func (em *eventMonitor) subscribe(subscriber events.Subscriber) {
    // note: filters are any match, if you want any match but not in namespace foo
    // then you have to manually filter namespace foo
    filters := []string{
        `topic=="/tasks/oom"`,
        `topic~="/images/"`,
    }
    em.ch, em.errCh = subscriber.Subscribe(em.ctx, filters...)
}
// cri处理容器事件，并更新容器状态，
func (em *eventMonitor) start() <-chan error {
    go func() {
        for {
            select {
            case e := <-em.ch:
                id, evt, err := convertEvent(e.Event)
                em.handleEvent(evt)
                       }
        }
    }()
}