謝絕轉(zhuǎn)載

序言

終于看完了volumeservice的初始化,鋪墊了這么多,就是為了看一下卷相關(guān)方法的調(diào)用,那就快進(jìn)入正題吧:

注意事項:
1.本文共有四篇,每篇都有編號,編號類似1.2.1這種,其中1是文章編號,因為后面的調(diào)用關(guān)系需要去前面篇幅中找,所以我標(biāo)注了這個方便尋找.

2.我是按調(diào)用過程列出代碼,如果當(dāng)前函數(shù)有多個地方需要講解,比如函數(shù)1.2中有兩個地方需要講解,那么要展開的地方便是1.2.1,1.2.2這樣排列.

3.鏈接:
第一篇:http://www.reibang.com/p/9900ec52f2c1 (命令的調(diào)用流程)
第二篇:http://www.reibang.com/p/db08b7d57721 (卷服務(wù)初始化)
第三篇:http://www.reibang.com/p/bbc73f5687a2 (plugin的管理)
第四篇:http://www.reibang.com/p/a92b1b11c8dd (卷相關(guān)命令的執(zhí)行)

卷相關(guān)命令的調(diào)用

4.1 initRouter函數(shù)

func initRouter(opts routerOptions) {
    decoder := runconfig.ContainerDecoder{}

    routers := []router.Router{
        ...
        volume.NewRouter(opts.daemon.VolumesService()),
        ...
    }
...
}

而volume.NewRouter方法的定義如下:

4.2 NewRouter函數(shù)

// NewRouter initializes a new volume router
func NewRouter(b Backend) router.Router {
    r := &volumeRouter{
        backend: b,
    }
    r.initRoutes()
    return r
}

繼續(xù)看路由的初始化:

4.3 initRoutes方法

func (r *volumeRouter) initRoutes() {
    r.routes = []router.Route{
        // GET
        router.NewGetRoute("/volumes", r.getVolumesList),
        router.NewGetRoute("/volumes/{name:.*}", r.getVolumeByName),
        // POST
        router.NewPostRoute("/volumes/create", r.postVolumesCreate),
        router.NewPostRoute("/volumes/prune", r.postVolumesPrune, router.WithCancel),
        // DELETE
        router.NewDeleteRoute("/volumes/{name:.*}", r.deleteVolumes),
    }
}

而具體的方法是通過傳入的volumeservice來實現(xiàn)的,以List方法為例,即getVolumesList實際執(zhí)行的是傳入的volumeservice的List方法:

4.4 getVolumesList方法

func (v *volumeRouter) getVolumesList(ctx context.Context, w http.ResponseWriter, r *http.Request, vars map[string]string) error {
    ...
    volumes, warnings, err := v.backend.List(ctx, filters)
    ...
}

4.5 List方法

我們就按最常規(guī)的方式,不指定具體的filter,看一下這個方法如何執(zhí)行.

func (s *VolumesService) List(ctx context.Context, filter filters.Args) (volumesOut []*types.Volume, warnings []string, err error) {
    by, err := filtersToBy(filter, acceptedListFilters)
    if err != nil {
        return nil, nil, err
    }

    # 調(diào)用vs的find方法,vs是前面提到的VolumeStore
    volumes, warnings, err := s.vs.Find(ctx, by)
    if err != nil {
        return nil, nil, err
    }

    return s.volumesToAPI(ctx, volumes, useCachedPath(true)), warnings, nil
}

我們按filter為空的情況繼續(xù)執(zhí)行s.vs.Find方法,Find方法中調(diào)用到了前面提到的ds[],也就是存儲driver的Store:

4.6 Find方法

// Find lists volumes filtered by the past in filter.
// If a driver returns a volume that has name which conflicts with another volume from a different driver,
// the first volume is chosen and the conflicting volume is dropped.
func (s *VolumeStore) Find(ctx context.Context, by By) (vols []volume.Volume, warnings []string, err error) {
    logrus.WithField("ByType", fmt.Sprintf("%T", by)).WithField("ByValue", fmt.Sprintf("%+v", by)).Debug("VolumeStore.Find")
    switch f := by.(type) {
    case nil, orCombinator, andCombinator, byDriver, ByReferenced, CustomFilter:
        # 在filter中加載卷
        warnings, err = s.filter(ctx, &vols, by)
    ...
    }
    if err != nil {
        return nil, nil, &OpErr{Err: err, Op: "list"}
    }

    var out []volume.Volume

    for _, v := range vols {
        name := normalizeVolumeName(v.Name())

        s.locks.Lock(name)
        # 從內(nèi)存中加載卷的名稱
        storedV, exists := s.getNamed(name)
        // Note: it's not safe to populate the cache here because the volume may have been
        // deleted before we acquire a lock on its name
        # 如果存在并且driver不同,則保留第一個同名稱的卷
        if exists && storedV.DriverName() != v.DriverName() {
            logrus.Warnf("Volume name %s already exists for driver %s, not including volume returned by %s", v.Name(), storedV.DriverName(), v.DriverName())
            s.locks.Unlock(v.Name())
            continue
        }

        out = append(out, v)
        s.locks.Unlock(v.Name())
    }
    return out, warnings, nil
}

上面的方法先加載卷,然后再跟緩存中的進(jìn)行對比,如果名稱相同而driver不同,則跳過第二個同名稱的卷. 加載卷的過程如下:

4.7 filter方法

func (s *VolumeStore) filter(ctx context.Context, vols *[]volume.Volume, by By) (warnings []string, err error) {
    // note that this specifically does not support the `FromList` By type.
    switch f := by.(type) {
    case nil:
        if *vols == nil {
            var ls []volume.Volume
            ls, warnings, err = s.list(ctx)
            if err != nil {
                return warnings, err
            }
            *vols = ls
        }
    ...
    }
    return warnings, nil
}

繼續(xù)看s.list方法,這個方法就會調(diào)用到我們在自定義的卷plugin中實現(xiàn)的接口方法:

4.8 list方法

// list goes through each volume driver and asks for its list of volumes.
// TODO(@cpuguy83): plumb context through
func (s *VolumeStore) list(ctx context.Context, driverNames ...string) ([]volume.Volume, []string, error) {
    var (
        ls       = []volume.Volume{} // do not return a nil value as this affects filtering
        warnings []string
    )

    # volume.Driver是個接口
    var dls []volume.Driver
    # GetAllDrivers加載所有的VolumeDriver,返回值其實是一個driver的適配器
    # s.drivers也就是之前講到的ds[2.3.2.2 章節(jié)]
    # 此處是重點
    all, err := s.drivers.GetAllDrivers()
    ...

    type vols struct {
        vols       []volume.Volume
        err        error
        driverName string
    }
    chVols := make(chan vols, len(dls))

    for _, vd := range dls {
        go func(d volume.Driver) {
            # 調(diào)用plugin中實現(xiàn)的List方法獲取卷
            # 此處是重點
            vs, err := d.List()
            if err != nil {
                chVols <- vols{driverName: d.Name(), err: &OpErr{Err: err, Name: d.Name(), Op: "list"}}
                return
            }
            for i, v := range vs {
                s.globalLock.RLock()
                vs[i] = volumeWrapper{v, s.labels[v.Name()], d.Scope(), s.options[v.Name()]}
                s.globalLock.RUnlock()
            }

            chVols <- vols{vols: vs}
        }(vd)
    }

    ...
    return ls, warnings, nil
}

上面的s.drivers.GetAllDrivers方法加載所有注冊過的drivers:

4.8.1 GetAllDrivers方法

// GetAllDrivers lists all the registered drivers
func (s *Store) GetAllDrivers() ([]volume.Driver, error) {
    var plugins []getter.CompatPlugin
    if s.pluginGetter != nil {
        var err error
        # 這里的store就是前面講過的ds[2.3.2.3章節(jié)], ds的pluginGetter就是前面講到的d.pluginStore
        # 過濾卷driver的名稱 extName = "VolumeDriver"
        plugins, err = s.pluginGetter.GetAllByCap(extName)
        if err != nil {
            return nil, fmt.Errorf("error listing plugins: %v", err)
        }
    }
    # 此處的volume.Driver是個接口
    var ds []volume.Driver

    s.mu.Lock()
    defer s.mu.Unlock()

    #先加緩存過的drivers
    for _, d := range s.extensions {
        ds = append(ds, d)
    }
    #加載沒有緩存過的driver
    for _, p := range plugins {
        name := p.Name()

        if _, ok := s.extensions[name]; ok {
            continue
        }

        # 用plugin創(chuàng)建對應(yīng)的driver適配器,適配器中的方法調(diào)用實際driver的方法
        # 此處是重點
        ext, err := makePluginAdapter(p)
        if err != nil {
            return nil, errors.Wrap(err, "error making plugin client")
        }
        if p.IsV1() {
            s.extensions[name] = ext
        }
        ds = append(ds, ext)
    }
    return ds, nil
}

• 4.8.1.1 Driver接口

先看volume.Driver接口:

// Driver is for creating and removing volumes.
type Driver interface {
    // Name returns the name of the volume driver.
    Name() string
    // Create makes a new volume with the given name.
    Create(name string, opts map[string]string) (Volume, error)
    // Remove deletes the volume.
    Remove(vol Volume) (err error)
    // List lists all the volumes the driver has
    List() ([]Volume, error)
    // Get retrieves the volume with the requested name
    Get(name string) (Volume, error)
    // Scope returns the scope of the driver (e.g. `global` or `local`).
    // Scope determines how the driver is handled at a cluster level
    Scope() string
}

然后繼續(xù)看makePluginAdapter方法

• 4.8.1.2 makePluginAdapter方法

可知返回值是volumeDriverAdapter,而volumeDriverAdapter結(jié)構(gòu)體實現(xiàn)了volume.Driver接口:

func makePluginAdapter(p getter.CompatPlugin) (*volumeDriverAdapter, error) {
    # 如果是v1/http plugin client方式的plugin
    if pc, ok := p.(getter.PluginWithV1Client); ok {
        return &volumeDriverAdapter{name: p.Name(), scopePath: p.ScopedPath, proxy: &volumeDriverProxy{pc.Client()}}, nil
    }

    # 如果是PluginAddr方式的plugin
    pa, ok := p.(getter.PluginAddr)
    ...

    addr := pa.Addr()
    # 此處的client與上面的pc.Client()都是同一個類型,即*plugins.client
    client, err := plugins.NewClientWithTimeout(addr.Network()+"://"+addr.String(), nil, pa.Timeout())
    if err != nil {
        return nil, errors.Wrap(err, "error creating plugin client")
    }

    return &volumeDriverAdapter{name: p.Name(), scopePath: p.ScopedPath, proxy: &volumeDriverProxy{client}}, nil    ...
}

不管是哪種方式的plugin,都要構(gòu)造一個volumeDriverAdapter,其成員變量proxy作為plugin的客戶端.

• 4.8.1.3 volumeDriverAdapter結(jié)構(gòu)體

volumeDriverAdapter結(jié)構(gòu)體定義如下:

type volumeDriverAdapter struct {
    name         string
    scopePath    func(s string) string
    capabilities *volume.Capability
    proxy        volumeDriver
}

其中,proxy是volumeDriver接口類型

• 4.8.1.4 volumeDriver接口

type volumeDriver interface {
    // Create a volume with the given name
    Create(name string, opts map[string]string) (err error)
    // Remove the volume with the given name
    Remove(name string) (err error)
    // Get the mountpoint of the given volume
    Path(name string) (mountpoint string, err error)
    // Mount the given volume and return the mountpoint
    Mount(name, id string) (mountpoint string, err error)
    // Unmount the given volume
    Unmount(name, id string) (err error)
    // List lists all the volumes known to the driver
    List() (volumes []*proxyVolume, err error)
    // Get retrieves the volume with the requested name
    Get(name string) (volume *proxyVolume, err error)
    // Capabilities gets the list of capabilities of the driver
    Capabilities() (capabilities volume.Capability, err error)
}

• 4.8.1.5 volumeDriverProxy結(jié)構(gòu)體

新建volumeDriverAdapter的時候用volumeDriverProxy給成員變量proxy賦值,所以volumeDriverProxy中要實現(xiàn)volumeDriver的各種接口.

看下volumeDriverProxy的定義:

type volumeDriverProxy struct {
    client
}

匿名成員client是一個接口:

• 4.8.1.6 client接口

type client interface {
    CallWithOptions(string, interface{}, interface{}, ...func(*plugins.RequestOpts)) error
}

• 4.8.1.7 CallWithOptions方法
  這個接口的實現(xiàn)即是2.2.5.3中的Client結(jié)構(gòu)體,再看Client結(jié)構(gòu)體的CallWithOptions方法的定義:

// CallWithOptions is just like call except it takes options
func (c *Client) CallWithOptions(serviceMethod string, args interface{}, ret interface{}, opts ...func(*RequestOpts)) error {
    var buf bytes.Buffer
    if args != nil {
        if err := json.NewEncoder(&buf).Encode(args); err != nil {
            return err
        }
    }
    body, err := c.callWithRetry(serviceMethod, &buf, true, opts...)
    if err != nil {
        return err
    }
    defer body.Close()
    if ret != nil {
        if err := json.NewDecoder(body).Decode(&ret); err != nil {
            logrus.Errorf("%s: error reading plugin resp: %v", serviceMethod, err)
            return err
        }
    }
    return nil
}

主要作用就是根據(jù)傳入的serviceMethod調(diào)用對應(yīng)的plugin中的方法.

4.8.2 List 方法

再回到4.8章節(jié),去看d.List()方法調(diào)用,實際上就是volumeDriverAdapter的List方法

func (a *volumeDriverAdapter) List() ([]volume.Volume, error) {
    ls, err := a.proxy.List()
    if err != nil {
        return nil, err
    }

    var out []volume.Volume
    for _, vp := range ls {
        out = append(out, &volumeAdapter{
            proxy:      a.proxy,
            name:       vp.Name,
            scopePath:  a.scopePath,
            driverName: a.name,
            eMount:     a.scopePath(vp.Mountpoint),
        })
    }
    return out, nil
}

a.proxy.List()就調(diào)用到具體的你實現(xiàn)的plugin的List方法了.

• 4.8.2.1 List方法

func (pp *volumeDriverProxy) List() (volumes []*proxyVolume, err error) {
    var (
        req volumeDriverProxyListRequest
        ret volumeDriverProxyListResponse
    )

    if err = pp.CallWithOptions("VolumeDriver.List", req, &ret, plugins.WithRequestTimeout(shortTimeout)); err != nil {
        return
    }

    volumes = ret.Volumes

    if ret.Err != "" {
        err = errors.New(ret.Err)
    }

    return
}

如果你使用過go-plugins-helpers的話,你應(yīng)該熟悉下面的Handle方法:

const (
    // DefaultDockerRootDirectory is the default directory where volumes will be created.
    DefaultDockerRootDirectory = "/var/lib/docker-volumes"

    manifest         = `{"Implements": ["VolumeDriver"]}`
    listPath         = "/VolumeDriver.List"
    ...
)

func (h *Handler) initMux() {
    ...
    h.HandleFunc(listPath, func(w http.ResponseWriter, r *http.Request) {
        log.Println("Entering go-plugins-helpers listPath")
        res, err := h.driver.List()
        if err != nil {
            sdk.EncodeResponse(w, NewErrorResponse(err.Error()), true)
            return
        }
        sdk.EncodeResponse(w, res, false)
    })
    ...
}

總結(jié)

以上就是volumeService的初始化以及加載對應(yīng)docker volume plugin的過程,要弄懂的話根據(jù)這個流程看一遍,基本就明白了.所有的東西都是為了把源碼中卷相關(guān)的操作和我們自己實現(xiàn)的go-plugins-helpers中的接口方法連接起來.