【containerd 源码分析】containerd 启动流程分析

曾经终败给现在 2022-06-11 07:26 364阅读 0赞

github: [https://github.com/containerd/containerd][https_github.com_containerd_containerd] release-v1.3.0

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70][]

# **前言** #

*  dockerd是docker engine守护进程，dockerd启动时会启动containerd子进程，dockerd与containerd通过rpc进行通信
 *  ctr是containerd的cli
 *  containerd通过shim操作runc，runc真正控制容器生命周期，启动一个容器就会启动一个shim进程
 *  shim直接调用runc的包函数,shim与containerd之前通过rpc通信
 *  真正用户想启动的进程由runc的init进程启动，即`runc init [args ...]`

docker     ctr
      |         |
      V         V
    dockerd -> containerd ---> shim -> runc -> runc init -> process
                          |-- > shim -> runc -> runc init -> process
                          +-- > shim -> runc -> runc init -> process

containerd 只是一个守护进程，容器的实际运行时由 runC 控制。***containerd 主要职责是镜像管理（镜像、元信息等）、容器执行***（调用最终运行时组件执行）

![Center][]

# containerd 源码编译 #

需要安装依赖包：btrfs-tools，直接 make 即可生成 ctr containerd containerd-shim binaries 可执行文件

# 1. containerd main 函数 #

入口目录为 cmd/containerd/main.go 中 main 函数，默认配置文件 /etc/containerd/config.toml，包含三个子命令，configCommand，publishCommand，ociHook

func main() {
    	app := command.App()
    	if err := app.Run(os.Args); err != nil {
    		fmt.Fprintf(os.Stderr, "containerd: %s\n", err)
    		os.Exit(1)
    	}
    }

###     1.1 command App 函数 ###

App 函数返回 \*cli.App 实例

// App returns a *cli.App instance.
    func App() *cli.App {
    	app := cli.NewApp()
    	app.Name = "containerd"
    	app.Version = version.Version
    	app.Usage = usage
    	app.Description = `
    containerd is a high performance container runtime whose daemon can be started
    by using this command. If none of the *config*, *publish*, or *help* commands
    are specified, the default action of the **containerd** command is to start the
    containerd daemon in the foreground.
    
    
    A default configuration is used if no TOML configuration is specified or located
    at the default file location. The *containerd config* command can be used to
    generate the default configuration for containerd. The output of that command
    can be used and modified as necessary as a custom configuration.`

**     1.1.1 三个子命令**

configCommand 用于生成配置文件 **containerd config default > /etc/containerd/config.toml**，publishCommand，ociHook

app.Flags = append(app.Flags, serviceFlags()...)
    app.Commands = []cli.Command{
    	configCommand,
    	publishCommand,
    	ociHook,
    }

**     1.1.2 grpc address 不能为空**

> \[grpc\]  
>   address = "/run/containerd/containerd.sock"  
>   tcp\_address = ""  
>   tcp\_tls\_cert = ""  
>   tcp\_tls\_key = ""  
>   uid = 0  
>   gid = 0  
>   max\_recv\_message\_size = 16777216  
>   max\_send\_message\_size = 16777216
> 
> \[ttrpc\]  
>   address = ""  
>   uid = 0  
>   gid = 0

if config.GRPC.Address == "" {
    	return errors.Wrap(errdefs.ErrInvalidArgument, "grpc address cannot be empty")
    }
    if config.TTRPC.Address == "" {
    	// If TTRPC was not explicitly configured, use defaults based on GRPC.
    	config.TTRPC.Address = fmt.Sprintf("%s.ttrpc", config.GRPC.Address)
    	config.TTRPC.UID = config.GRPC.UID
    	config.TTRPC.GID = config.GRPC.GID
    }

# 2. server.New 函数创建以及初始化 containerd server #

初始化 contaienrd server，加载 timeout 配置

> \[timeouts\]  
>   "io.containerd.timeout.shim.cleanup" = "5s"  
>   "io.containerd.timeout.shim.load" = "5s"  
>   "io.containerd.timeout.shim.shutdown" = "3s"  
>   "io.containerd.timeout.task.state" = "2s"

// New creates and initializes a new containerd server
    func New(ctx context.Context, config *srvconfig.Config) (*Server, error) {
    	if err := apply(ctx, config); err != nil {
    		return nil, err
    	}
    	for key, sec := range config.Timeouts {
    		d, err := time.ParseDuration(sec)
    		if err != nil {
    			return nil, errors.Errorf("unable to parse %s into a time duration", sec)
    		}
    		timeout.Set(key, d)
    	}

###     2.1 LoadPlugins 加载插件 ###

// LoadPlugins loads all plugins into containerd and generates an ordered graph
    // of all plugins.
    func LoadPlugins(ctx context.Context, config *srvconfig.Config) ([]*plugin.Registration, error) {
    	// load all plugins into containerd
    	path := config.PluginDir
    	if path == "" {
    		path = filepath.Join(config.Root, "plugins")
    	}
    	if err := plugin.Load(path); err != nil {
    		return nil, err
    	}

**     2.1.1 注册插件 io.containerd.content.v1，**store 结构体 **实现了 Store 接口**

路径 containerd/content/local/store.go

> // Store combines the methods of content-oriented interfaces into a set that  
> // are commonly provided by complete implementations.  
> type Store interface \{  
>     Manager  
>     Provider  
>     IngestManager  
>     Ingester  
> \}

// load additional plugins that don't automatically register themselves
    plugin.Register(&plugin.Registration{
    	Type: plugin.ContentPlugin,
    	ID:   "content",
    	InitFn: func(ic *plugin.InitContext) (interface{}, error) {
    		ic.Meta.Exports["root"] = ic.Root
    		return local.NewStore(ic.Root)
    	},
    })

**     2.1.2 注册插件 io.containerd.metadata.v1，使用 bolt 数据库存储**

plugin.Register(&plugin.Registration{
    	Type: plugin.MetadataPlugin,
    	ID:   "bolt",
    	Requires: []plugin.Type{
    		plugin.ContentPlugin,
    		plugin.SnapshotPlugin,
    	},
    	Config: &srvconfig.BoltConfig{
    		ContentSharingPolicy: srvconfig.SharingPolicyShared,
    	},

**     2.1.3 遍历所有插件**

类型为 io.containerd.snapshotter.v1，或者 snapshot，实现在 containerd/snapshots/proxy/proxy.go，proxySnapshotter 结构体实现了 Snapshotter 接口

>     type Snapshotter interface {
>        Stat(ctx context.Context, key string) (Info, error)
>     
>        Update(ctx context.Context, info Info, fieldpaths ...string) (Info, error)
>     
>        Usage(ctx context.Context, key string) (Usage, error)
>     
>        Mounts(ctx context.Context, key string) ([]mount.Mount, error)
>     
>        Prepare(ctx context.Context, key, parent string, opts ...Opt) ([]mount.Mount, error)
>     
>        View(ctx context.Context, key, parent string, opts ...Opt) ([]mount.Mount, error)
>     
>        Commit(ctx context.Context, name, key string, opts ...Opt) error
>     
>        Remove(ctx context.Context, key string) error
>     
>        Walk(ctx context.Context, fn func(context.Context, Info) error) error
>     
>        Close() error
>     }

> \[plugins\]
> 
>   \[plugins."io.containerd.snapshotter.v1.devmapper"\]  
>     root\_path = ""  
>     pool\_name = ""  
>     base\_image\_size = ""

类型插件为 io.containerd.content.v1 或者 content，proxyContentStore 结构体实现了 Store 接口，实现在 containerd/content/local/store.go

>     // Store combines the methods of content-oriented interfaces into a set that
>     // are commonly provided by complete implementations.
>     type Store interface {
>        Manager
>        Provider
>        IngestManager
>        Ingester
>     }

clients := &proxyClients{}
    for name, pp := range config.ProxyPlugins {
    	var (
    		t plugin.Type
    		f func(*grpc.ClientConn) interface{}
    
    		address = pp.Address
    	)
    
    	switch pp.Type {
    	case string(plugin.SnapshotPlugin), "snapshot":
    		t = plugin.SnapshotPlugin
    		ssname := name
    		f = func(conn *grpc.ClientConn) interface{} {
    			return ssproxy.NewSnapshotter(ssapi.NewSnapshotsClient(conn), ssname)
    		}
    
    	case string(plugin.ContentPlugin), "content":
    		t = plugin.ContentPlugin
    		f = func(conn *grpc.ClientConn) interface{} {
    			return csproxy.NewContentStore(csapi.NewContentClient(conn))
    		}
    	default:
    		log.G(ctx).WithField("type", pp.Type).Warn("unknown proxy plugin type")
    	}
    
    	plugin.Register(&plugin.Registration{
    		Type: t,
    		ID:   name,
    		InitFn: func(ic *plugin.InitContext) (interface{}, error) {
    			ic.Meta.Exports["address"] = address
    			conn, err := clients.getClient(address)
    			if err != nil {
    				return nil, err
    			}
    			return f(conn), nil
    		},
    	})
    
    }

**     2.1.4 Graph 函数返回一个注册有序的插件列表**

// Graph returns an ordered list of registered plugins for initialization.
    // Plugins in disableList specified by id will be disabled.
    func Graph(filter DisableFilter) (ordered []*Registration) {
    	register.RLock()
    	defer register.RUnlock()
    
    	for _, r := range register.r {
    		if filter(r) {
    			r.Disable = true
    		}
    	}
    
    	added := map[*Registration]bool{}
    	for _, r := range register.r {
    		if r.Disable {
    			continue
    		}
    		children(r, added, &ordered)
    		if !added[r] {
    			ordered = append(ordered, r)
    			added[r] = true
    		}
    	}
    	return ordered
    }

**     2.1.5 启动 GRPC server**

func serve(ctx gocontext.Context, l net.Listener, serveFunc func(net.Listener) error) {
    	path := l.Addr().String()
    	log.G(ctx).WithField("address", path).Info("serving...")
    	go func() {
    		defer l.Close()
    		if err := serveFunc(l); err != nil {
    			log.G(ctx).WithError(err).WithField("address", path).Fatal("serve failure")
    		}
    	}()
    }

###     2.2 获取 GRPC server 配置参数 ###

>  \[grpc\]  
>   address = "/run/containerd/containerd.sock"  
>   tcp\_address = ""  
>   tcp\_tls\_cert = ""  
>   tcp\_tls\_key = ""  
>   uid = 0  
>   gid = 0  
>   max\_recv\_message\_size = 16777216  
>   max\_send\_message\_size = 16777216

serverOpts := []grpc.ServerOption{
    	grpc.UnaryInterceptor(grpc_prometheus.UnaryServerInterceptor),
    	grpc.StreamInterceptor(grpc_prometheus.StreamServerInterceptor),
    }
    if config.GRPC.MaxRecvMsgSize > 0 {
    	serverOpts = append(serverOpts, grpc.MaxRecvMsgSize(config.GRPC.MaxRecvMsgSize))
    }
    if config.GRPC.MaxSendMsgSize > 0 {
    	serverOpts = append(serverOpts, grpc.MaxSendMsgSize(config.GRPC.MaxSendMsgSize))
    }

### **    2.3 实例化 Server** ###

**      包括几种 server，配置等**

var (
    	grpcServer = grpc.NewServer(serverOpts...)
    	tcpServer  = grpc.NewServer(tcpServerOpts...)
    
    	grpcServices  []plugin.Service
    	tcpServices   []plugin.TCPService
    	ttrpcServices []plugin.TTRPCService
    
    	s = &Server{
    		grpcServer:  grpcServer,
    		tcpServer:   tcpServer,
    		ttrpcServer: ttrpcServer,
    		events:      exchange.NewExchange(),
    		config:      config,
    	}
    	initialized = plugin.NewPluginSet()
    	required    = make(map[string]struct{})
    )

###     2.4 需要开启设置的插件 ###

> required\_plugins = \[\]

for _, r := range config.RequiredPlugins {
    	required[r] = struct{}{}
    }

###     2.5 Init 调用各个注册的 InitFn 函数初始化，分别实现了不同的接口 ###

// Init the registered plugin
    func (r *Registration) Init(ic *InitContext) *Plugin {
    	p, err := r.InitFn(ic)
    	return &Plugin{
    		Registration: r,
    		Config:       ic.Config,
    		Meta:         ic.Meta,
    		instance:     p,
    		err:          err,
    	}
    }

###     2.6 对所有 grpc 服务分别调用 Register 方法注册      根据是否实现了 Register 接口，各个插件里面的实现的接口 Register ###

// register services after all plugins have been initialized
    for _, service := range grpcServices {
    	if err := service.Register(grpcServer); err != nil {
    		return nil, err
    	}
    }

Init 函数以及 Register 函数在后面在讲解，各个插件初始，实例化并注册服务

# 三. plugin #

### **     Registration 数据结构：** ###

type Registration struct {
           Type     PluginType
           ID       string
           Config   interface{}
           Requires []PluginType
           Init     func(*InitContext) (interface{}, error)
    
           added bool
    }

***    init 函数初始化类型包括***

# ctr plugins ls
    TYPE                            ID                    PLATFORMS      STATUS    
    io.containerd.content.v1        content               -              ok        
    io.containerd.snapshotter.v1    btrfs                 linux/amd64    error     
    io.containerd.snapshotter.v1    devmapper             linux/amd64    error     
    io.containerd.snapshotter.v1    aufs                  linux/amd64    error     
    io.containerd.snapshotter.v1    native                linux/amd64    ok        
    io.containerd.snapshotter.v1    overlayfs             linux/amd64    ok        
    io.containerd.snapshotter.v1    zfs                   linux/amd64    error     
    io.containerd.metadata.v1       bolt                  -              ok        
    io.containerd.differ.v1         walking               linux/amd64    ok        
    io.containerd.gc.v1             scheduler             -              ok        
    io.containerd.service.v1        containers-service    -              ok        
    io.containerd.service.v1        content-service       -              ok        
    io.containerd.service.v1        diff-service          -              ok        
    io.containerd.service.v1        images-service        -              ok        
    io.containerd.service.v1        leases-service        -              ok        
    io.containerd.service.v1        namespaces-service    -              ok        
    io.containerd.service.v1        snapshots-service     -              ok        
    io.containerd.runtime.v1        linux                 linux/amd64    ok        
    io.containerd.runtime.v2        task                  linux/amd64    ok        
    io.containerd.monitor.v1        cgroups               linux/amd64    ok        
    io.containerd.service.v1        tasks-service         -              ok        
    io.containerd.internal.v1       restart               -              ok        
    io.containerd.grpc.v1           containers            -              ok        
    io.containerd.grpc.v1           content               -              ok        
    io.containerd.grpc.v1           diff                  -              ok        
    io.containerd.grpc.v1           events                -              ok        
    io.containerd.grpc.v1           healthcheck           -              ok        
    io.containerd.grpc.v1           images                -              ok        
    io.containerd.grpc.v1           leases                -              ok        
    io.containerd.grpc.v1           namespaces            -              ok        
    io.containerd.internal.v1       opt                   -              ok        
    io.containerd.grpc.v1           snapshots             -              ok        
    io.containerd.grpc.v1           tasks                 -              ok        
    io.containerd.grpc.v1           version               -              ok        
    io.containerd.grpc.v1           cri                   linux/amd64    ok

这些可以根据目录大致位置看实现

# 四. plugin：io.containerd.grpc.v1.containers #

路径： services/containers/service.go，ID 为 containers，提供基础元素的存储

func init() {
           plugin.Register(&plugin.Registration{
                  Type: plugin.GRPCPlugin,
                  ID:   "containers",
                  Requires: []plugin.PluginType{
                         plugin.MetadataPlugin,
                  },
                  Init: func(ic *plugin.InitContext) (interface{}, error) {
                         m, err := ic.Get(plugin.MetadataPlugin)
                         if err != nil {
                                return nil, err
                         }
                         return NewService(m.(*bolt.DB), ic.Events), nil
                  },
           })
    }

实例化为 Service 结构体

type Service struct {
           db        *bolt.DB
           publisher events.Publisher
    }
    
    func NewService(db *bolt.DB, publisher events.Publisher) api.ContainersServer {
           return &Service{db: db, publisher: publisher}
    }

func (s *Service) Register(server *grpc.Server) error {
           api.RegisterContainersServer(server, s)
           return nil
    }

*    可以有如下 GRPC API 方法*

type ContainersServer interface {
           Get(context.Context, *GetContainerRequest) (*GetContainerResponse, error)
           List(context.Context, *ListContainersRequest) (*ListContainersResponse, error)
           Create(context.Context, *CreateContainerRequest) (*CreateContainerResponse, error)
           Update(context.Context, *UpdateContainerRequest) (*UpdateContainerResponse, error)
           Delete(context.Context, *DeleteContainerRequest) (*google_protobuf2.Empty, error)
    }

根据下面这个对应哪个 handler 处理

var _Containers_serviceDesc = grpc.ServiceDesc{
           ServiceName: "containerd.services.containers.v1.Containers",
           HandlerType: (*ContainersServer)(nil),
           Methods: []grpc.MethodDesc{
                  {
                         MethodName: "Get",
                         Handler:    _Containers_Get_Handler,
                  },
                  {
                         MethodName: "List",
                         Handler:    _Containers_List_Handler,
                  },
                  {
                         MethodName: "Create",
                         Handler:    _Containers_Create_Handler,
                  },
                  {
                         MethodName: "Update",
                         Handler:    _Containers_Update_Handler,
                  },
                  {
                         MethodName: "Delete",
                         Handler:    _Containers_Delete_Handler,
                  },
           },
           Streams:  []grpc.StreamDesc{},
           Metadata: "github.com/containerd/containerd/api/services/containers/v1/containers.proto",
    }

# 五. plugin：io.containerd.grpc.v1 tasks #

路径： services/tasks/service.go，ID 为 tasks，是现在 services/tasks/local.go

func init() {
    	plugin.Register(&plugin.Registration{
    		Type: plugin.GRPCPlugin,
    		ID:   "tasks",
    		Requires: []plugin.Type{
    			plugin.ServicePlugin,
    		},
    		InitFn: func(ic *plugin.InitContext) (interface{}, error) {
    			plugins, err := ic.GetByType(plugin.ServicePlugin)
    			if err != nil {
    				return nil, err
    			}
    			p, ok := plugins[services.TasksService]
    			if !ok {
    				return nil, errors.New("tasks service not found")
    			}
    			i, err := p.Instance()
    			if err != nil {
    				return nil, err
    			}
    			return &service{local: i.(api.TasksClient)}, nil
    		},
    	})
    }

###     5.1 local init 函数 ###

插件类型为 io.containerd.service.v1，ID 为 tasks-service，需求服务插件为：

*  io.containerd.runtime.v1
 *  io.containerd.runtime.v2
 *  io.containerd.metadata.v1
 *  io.containerd.monitor.v1

func init() {
    	plugin.Register(&plugin.Registration{
    		Type:     plugin.ServicePlugin,
    		ID:       services.TasksService,
    		Requires: tasksServiceRequires,
    		InitFn:   initFunc,
    	})
    
    	timeout.Set(stateTimeout, 2*time.Second)
    }

###     5.2 初始化函数 initFunc ###

func initFunc(ic *plugin.InitContext) (interface{}, error) {
    	runtimes, err := loadV1Runtimes(ic)
    	if err != nil {
    		return nil, err
    	}
    
    	v2r, err := ic.Get(plugin.RuntimePluginV2)
    	if err != nil {
    		return nil, err
    	}
    
    	m, err := ic.Get(plugin.MetadataPlugin)
    	if err != nil {
    		return nil, err
    	}
    
    	monitor, err := ic.Get(plugin.TaskMonitorPlugin)
    	if err != nil {
    		if !errdefs.IsNotFound(err) {
    			return nil, err
    		}
    		monitor = runtime.NewNoopMonitor()
    	}
    
    	db := m.(*metadata.DB)
    	l := &local{
    		runtimes:   runtimes,
    		containers: metadata.NewContainerStore(db),
    		store:      db.ContentStore(),
    		publisher:  ic.Events,
    		monitor:    monitor.(runtime.TaskMonitor),
    		v2Runtime:  v2r.(*v2.TaskManager),
    	}
    	for _, r := range runtimes {
    		tasks, err := r.Tasks(ic.Context, true)
    		if err != nil {
    			return nil, err
    		}
    		for _, t := range tasks {
    			l.monitor.Monitor(t)
    		}
    	}

**   **所有插件就不一一列举了，同样方法执行，至此建立 GRPC server 启动完毕。

**总结：**

**   **加载插件，根据插件建立 GRPC API 服务

[https_github.com_containerd_containerd]: https://github.com/containerd/containerd
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70]: /images/20220611/cc9120f571574120bcd3a6dc27f8fc72.png
[Center]: /images/20220611/e320724e8ac548bb8add3f8c33a6f604.png