docker源码2-卷服务的初始化

作者: wangwDavid | 来源:发表于2018-12-23 11:42 被阅读0次

docker源码2-卷服务的初始化
Eureka源码剖析之四：服务续约
Eureka源码剖析之五：服务下线
Eureka源码剖析之七：总结
Eureka源码剖析之六：自我保护机制
Eureka源码剖析之三：服务拉取
使用Docker部署Yapi
quartz源码2－初始化
sentinal源码2－初始化
『中级篇』Docker 水平扩展和负载均衡（40）

谢绝转载

序言

上篇我们粗略过了一下docker 命令的调用流程,可知命令的最后运行部分是在服务端的daemon中完成.这一篇主要看一下docker卷服务的初始化和相关卷plugin的加载.

这部分代码我们需要带着两个疑问去看,第一就是plugin是怎么被加载的,第二就是创建卷的时候怎么与我们自己实现的plugin关联起来的.

注意事项:
1.本文共有四篇,每篇都有编号,编号类似1.2.1这种,其中1是文章编号,因为后面的调用关系需要去前面篇幅中找,所以我标注了这个方便寻找.

2.我是按调用过程列出代码,如果当前函数有多个地方需要讲解,比如函数1.2中有两个地方需要讲解,那么要展开的地方便是1.2.1,1.2.2这样排列.

3.链接:
第一篇:https://www.jianshu.com/p/9900ec52f2c1 (命令的调用流程)
第二篇:https://www.jianshu.com/p/db08b7d57721 (卷服务初始化)
第三篇:https://www.jianshu.com/p/bbc73f5687a2 (plugin的管理)
第四篇:https://www.jianshu.com/p/a92b1b11c8dd (卷相关命令的执行)

Daemon中关于volume服务的初始化

main函数是万物之源,我们继续从服务端的main函数入手,找到相关的代码(不知道怎么找的朋友请看上篇).我们要找的代码在NewDaemon函数中:

2.1 NewDaemon函数

NewDaemon是docker进程的初始化函数,找到NewDaemon中对卷服务的初始化代码:

path	function name	line number
components/engine/daemon/daemon.go	NewDaemon	635

func NewDaemon(ctx context.Context, config *config.Config, pluginStore *plugin.Store) (daemon *Daemon, err error) {
    setDefaultMtu(config)

    ...

    # 创建deamon
    d := &Daemon{
        configStore: config,
        PluginStore: pluginStore,
        startupDone: make(chan struct{}),
    }
    ...
    # d.pluginManager的初始化中加载了/var/lib/docker/plugins下的所有plugin
    # 此处为重点
    // Plugin system initialization should happen before restore. Do not change order.
    d.pluginManager, err = plugin.NewManager(plugin.ManagerConfig{
        Root:               filepath.Join(config.Root, "plugins"),
        ExecRoot:           getPluginExecRoot(config.Root),
        Store:              d.PluginStore,
        CreateExecutor:     createPluginExec,
        RegistryService:    registryService,
        LiveRestoreEnabled: config.LiveRestoreEnabled,
        LogPluginEvent:     d.LogPluginEvent, // todo: make private
        AuthzMiddleware:    config.AuthzMiddleware,
    })

    ...
    # d.volumes 具体执行对卷的管理
    # 此处为重点
    d.volumes, err = volumesservice.NewVolumeService(config.Root, d.PluginStore, rootIDs, d)
    if err != nil {
        return nil, err
    }

    ...
}

在上面NewDaemon函数中标注出来两个重要的环节,首先是pluginManager的初始化,其次是d.volumes的初始化.这块涉及到的接口比较多,我们先根据调用流程过一遍,最后再来总结.

2.2 plugin的加载过程

2.2.1 Store 结构体

这里有一个很重要的参数pluginStore, 在执行NewDaemon的时候传入的pluginStore是一个空的Store结构体,主要作用就是保存plugin:

path	struct name	line number
components/engine/plugin/defs.go	Store	12

// Store manages the plugin inventory in memory and on-disk
type Store struct {
    sync.RWMutex
    plugins  map[string]*v2.Plugin
    specOpts map[string][]SpecOpt
    /* handlers are necessary for transition path of legacy plugins
     * to the new model. Legacy plugins use Handle() for registering an
     * activation callback.*/
    handlers map[string][]func(string, *plugins.Client)
}

2.2.2 Manager结构体

先看d.pluginManager的定义,它是一个结构体,定义如下,用来管理plugin:

path	struct name	line number
components/engine/plugin/manager.go	Manager	70

// Manager controls the plugin subsystem.
type Manager struct {
    config    ManagerConfig
    mu        sync.RWMutex // protects cMap
    muGC      sync.RWMutex // protects blobstore deletions
    cMap      map[*v2.Plugin]*controller
    blobStore *basicBlobStore
    publisher *pubsub.Publisher
    executor  Executor
}

2.2.3 NewManager函数

再来看Manager的新建函数:

path	function name	line number
components/engine/plugin/manager.go	NewManager	102

// NewManager returns a new plugin manager.
func NewManager(config ManagerConfig) (*Manager, error) {
    ...
    manager := &Manager{
        config: config,
    }
    ...
    # 创建manager.executor
    # 此处为重点
    manager.executor, err = config.CreateExecutor(manager)
    if err != nil {
        return nil, err
    }
    ...
    manager.cMap = make(map[*v2.Plugin]*controller)
    # 此处为重点
    if err := manager.reload(); err != nil {
        return nil, errors.Wrap(err, "failed to restore plugins")
    }

    manager.publisher = pubsub.NewPublisher(0, 0)
    return manager, nil
}

首先,传入的唯一参数config中绑定了d.PluginStore,对应的是config.Store(这部分代码代码赋值在NewDaemon中完成,参考2.1章节的代码).这里重点讲一下manager.executor和manager.reload().

2.2.4 初始化manager.executor

2.2.4.1 Executor接口

在上面的函数中,manager.executor是实现了Executor接口的结构体,Executor的主要作用是用来启动和停止plugin(创建对应的容器,此容器不同于正常的docker容器,只能用docker-runc命令查看):

path	interface name	line number
components/engine/plugin/manager.go	Executor	38

// Executor is the interface that the plugin manager uses to interact with for starting/stopping plugins
type Executor interface {
    Create(id string, spec specs.Spec, stdout, stderr io.WriteCloser) error
    IsRunning(id string) (bool, error)
    Restore(id string, stdout, stderr io.WriteCloser) (alive bool, err error)
    Signal(id string, signal int) error
}

而config.CreateExecutor在创建的时候传入值是以下匿名函数:

2.2.4.2 createPluginExec对应的匿名函数

path	function name	line number
components/engine/daemon/daemon.go		811

createPluginExec := func(m *plugin.Manager) (plugin.Executor, error) {
        var pluginCli *containerd.Client

        // Windows is not currently using containerd, keep the
        // client as nil
        if config.ContainerdAddr != "" {
            pluginCli, err = containerd.New(config.ContainerdAddr, containerd.WithDefaultNamespace(pluginexec.PluginNamespace), containerd.WithDialOpts(gopts))
            if err != nil {
                return nil, errors.Wrapf(err, "failed to dial %q", config.ContainerdAddr)
            }
        }
        # getPluginExecRoot(config.Root) = "/var/lib/docker/plugin"
        return pluginexec.New(ctx, getPluginExecRoot(config.Root), pluginCli, m)
    }

该函数调用pluginexec.New函数,返回一个Executor结构体,如下:

2.2.4.3 New函数

path	function name	line number
components/engine/plugin/executor/containerd/containerd.go	New	42

func New(ctx context.Context, rootDir string, cli *containerd.Client, exitHandler ExitHandler) (*Executor, error) {
    e := &Executor{
        rootDir:     rootDir,
        exitHandler: exitHandler,
    }

    client, err := libcontainerd.NewClient(ctx, cli, rootDir, PluginNamespace, e)
    if err != nil {
        return nil, errors.Wrap(err, "error creating containerd exec client")
    }
    e.client = client
    return e, nil
}

这里创建的结构体Executor实现了上面的Executor接口(定义位置不同),然后初始化libcontainerd的客户端,这里就不深入看了,其实我也不懂太底层的:-).

2.2.4.4 Executor结构体

path	struct name	line number
components/engine/plugin/executor/containerd/containerd.go	Executor	57

// Executor is the containerd client implementation of a plugin executor
type Executor struct {
    rootDir     string
    client      Client
    exitHandler ExitHandler
}

看完manager.executor,让我们回到NewManager[2.2.3章节],继续执行manager.reload()方法

2.2.5 manager.reload()加载plugin

2.2.5.1 reload方法

path	struct name	line number
components/engine/plugin/manager.go	reload	182

func (pm *Manager) reload() error { // todo: restore
    # 从/var/lib/docker/plugins 路径下加载plugin
    dir, err := ioutil.ReadDir(pm.config.Root)
    ...
    plugins := make(map[string]*v2.Plugin)
    for _, v := range dir {
        if validFullID.MatchString(v.Name()) {
            # 从对应的plugin路径的json文件下加载plugin
            p, err := pm.loadPlugin(v.Name())
            if err != nil {
                handleLoadError(err, v.Name())
                continue
            }
            plugins[p.GetID()] = p
        }
    ...
    }

    pm.config.Store.SetAll(plugins)
    ...
    return nil
}

加载plugin调用的是pm.loadPlugin方法,这里不仔细看这个方法了,但是v2.Plugin这个结构体比较重要,要看一下:

2.2.5.2 Plugin结构体

path	struct name	line number
components/engine/plugin/v2/plugin.go	Plugin	19

// Plugin represents an individual plugin.
type Plugin struct {
    mu        sync.RWMutex
    PluginObj types.Plugin `json:"plugin"` // todo: embed struct
    pClient   *plugins.Client
    refCount  int
    Rootfs    string // TODO: make private

    Config   digest.Digest
    Blobsums []digest.Digest

    modifyRuntimeSpec func(*specs.Spec)

    SwarmServiceID string
    timeout        time.Duration
    addr           net.Addr
}

其中plugins.Client定义如下,是plugin的http客户端:

2.2.5.3 Client结构体

path	struct name	line number
components/engine/pkg/plugins/client.go	Client	82

// Client represents a plugin client.
type Client struct {
    http           *http.Client // http client to use
    requestFactory transport.RequestFactory
}

加载plugins后我们看到调用了SetAll方法,看一下SetAll的定义:

2.2.5.4 SetAll方法

path	struct name	line number
components/engine/plugin/store.go	SetAll	65

// SetAll initialized plugins during daemon restore.
func (ps *Store) SetAll(plugins map[string]*v2.Plugin) {
    ps.Lock()
    defer ps.Unlock()

    for _, p := range plugins {
        ps.setSpecOpts(p)
    }
    ps.plugins = plugins
}

上面的方法最后把plugins保存到config中的Store中,也就是之前绑定的d.PluginStore, 也就是说d.PluginStore.plugins = plugins

2.3 卷服务的创建

回到NewDaemon函数[2.1章节],继续看卷服务的创建,volumesservice也就是d.volumes是卷的相关方法的实际执行者,我们先看下它的初始化过程.

d.volumes, err = volumesservice.NewVolumeService(config.Root, d.PluginStore, rootIDs, d)

2.3.1 VolumesService结构体

先看VolumesService定义:

path	struct name	line number
components/engine/volume/service/service.go	VolumesService	30

// VolumesService manages access to volumes
type VolumesService struct {
    vs           *VolumeStore
    ds           ds
    pruneRunning int32
    eventLogger  volumeEventLogger
}

2.3.2 NewVolumeService方法

NewVolumeService方法, 实参是上面的d.PluginStore, 虚参是一个PluginGetter接口.注意在NewVolumeService函数中d.PluginStore变成了pg即plugingetter.PluginGetter.

path	func name	line number
components/engine/volume/service/service.go	NewVolumeService	38

// NewVolumeService creates a new volume service
func NewVolumeService(root string, pg plugingetter.PluginGetter, rootIDs idtools.Identity, logger volumeEventLogger) (*VolumesService, error) {
    # ds是一个接口,负责driver的存储
    ds := drivers.NewStore(pg)
    if err := setupDefaultDriver(ds, root, rootIDs); err != nil {
        return nil, err
    }

    # vs 用来存储卷的信息
    vs, err := NewStore(root, ds)
    if err != nil {
        return nil, err
    }
    return &VolumesService{vs: vs, ds: ds, eventLogger: logger}, nil
}

首先看一下PluginGetter接口

2.3.2.1 PluginGetter接口

path	interface name	line number
components/engine/pkg/plugingetter/getter.go	PluginGetter	47

// PluginGetter is the interface implemented by Store
type PluginGetter interface {
    Get(name, capability string, mode int) (CompatPlugin, error)
    GetAllByCap(capability string) ([]CompatPlugin, error)
    GetAllManagedPluginsByCap(capability string) []CompatPlugin
    Handle(capability string, callback func(string, *plugins.Client))
}

2.3.2.2 ds接口

再看ds, ds接口只定义了一个方法GetDriverList

path	interface name	line number
components/engine/volume/service/service.go	ds	21

type ds interface {
    GetDriverList() []string
}

实现这个接口的是Store结构体,此Store非彼Store,不同于之前的保存Plugin的那个Store,这个store用来保存driver,一定要注意:

2.3.2.3 Store结构体

path	struct name	line number
components/engine/volume/drivers/extpoint.go	Store	45

// Store is an in-memory store for volume drivers
type Store struct {
    extensions   map[string]volume.Driver
    mu           sync.Mutex
    driverLock   *locker.Locker
    pluginGetter getter.PluginGetter
}

在drivers.NewStore(pg)执行后d.pluginStore被绑定到ds的pluginGetter上,也就是ds.pluginGetter = d.pluginStore.而vs是一个管理卷的结构体,定义如下:

2.3.2.4 VolumeStore结构体

path	struct name	line number
components/engine/volume/service/store.go	VolumeStore	186

// VolumeStore is a struct that stores the list of volumes available and keeps track of their usage counts
type VolumeStore struct {
    // locks ensures that only one action is being performed on a particular volume at a time without locking the entire store
    // since actions on volumes can be quite slow, this ensures the store is free to handle requests for other volumes.
    locks   *locker.Locker
    drivers *drivers.Store
    // globalLock is used to protect access to mutable structures used by the store object
    globalLock sync.RWMutex
    // names stores the volume name -> volume relationship.
    // This is used for making lookups faster so we don't have to probe all drivers
    names map[string]volume.Volume
    // refs stores the volume name and the list of things referencing it
    refs map[string]map[string]struct{}
    // labels stores volume labels for each volume
    labels map[string]map[string]string
    // options stores volume options for each volume
    options map[string]map[string]string
    db      *bolt.DB
}

vs的drivers值为ds, vs.drivers.pluginGetter = d.pluginStore

总结

这部分代码就是知道在Daemon的初始化中如何加载plugin并初始化相关卷服务,在后续的代码中会调用这篇讲到的内容.

docker源码2-卷服务的初始化
谢绝转载序言上篇我们粗略过了一下docker 命令的调用流程,可知命令的最后运行部分是在服务端的daemon中...
Eureka源码剖析之四：服务续约
Eureka源码剖析之一：初始化-启动 Eureka源码剖析之二：服务注册 Eureka源码剖析之三：服务拉取这...
Eureka源码剖析之五：服务下线
Eureka源码剖析之一：初始化-启动 Eureka源码剖析之二：服务注册 Eureka源码剖析之三：服务拉取 E...
Eureka源码剖析之七：总结
Eureka源码剖析之一：初始化-启动 Eureka源码剖析之二：服务注册 Eureka源码剖析之三：服务拉取 E...
Eureka源码剖析之六：自我保护机制
Eureka源码剖析之一：初始化-启动 Eureka源码剖析之二：服务注册 Eureka源码剖析之三：服务拉取 E...
Eureka源码剖析之三：服务拉取
Eureka源码剖析之一：初始化-启动 Eureka源码剖析之二：服务注册前面已经剖析了Eureka初始化、Eu...
使用Docker部署Yapi
一、 Docker部署YApi: 创建MongoDB数据卷启动MongoDB 获取YApi镜像初始化YApi数...
quartz源码2－初始化
一 StdSchedulerFactory.getScheduler 1.1 配置获取 1.2 实例id生成器In...
sentinal源码2－初始化
包含很多static的初始化块，并通过ServiceLoader的方式加载各种类一 static初始化块二 C...
『中级篇』Docker 水平扩展和负载均衡（40）
Docker Compose 里面的scale 设置服务运行的容器数量。源码地址：https://github.c...

docker源码2-卷服务的初始化

序言

Daemon中关于volume服务的初始化

2.1 NewDaemon函数

2.2 plugin的加载过程

2.2.1 Store 结构体

2.2.2 Manager结构体

2.2.3 NewManager函数

2.2.4 初始化manager.executor

2.2.5 manager.reload()加载plugin

2.3 卷服务的创建

2.3.1 VolumesService结构体

2.3.2 NewVolumeService方法

总结

相关文章

docker源码2-卷服务的初始化

Eureka源码剖析之四：服务续约

Eureka源码剖析之五：服务下线

Eureka源码剖析之七：总结

Eureka源码剖析之六：自我保护机制

Eureka源码剖析之三：服务拉取

使用Docker部署Yapi

quartz源码2－初始化

sentinal源码2－初始化

『中级篇』Docker 水平扩展和负载均衡（40）

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Docker容器