作者：nanxiao

Swarm driver的NewCluster函数的核心功能如下（cluster/swarm/cluster.go）：

discoveryCh, errCh := cluster.discovery.Watch(nil)
go cluster.monitorDiscovery(discoveryCh, errCh)
go cluster.monitorPendingEngines()

cluster.discovery.Watch的定义如下：

Watch(stopCh <-chan struct{}) (<-chan Entries, <-chan error)

返回两个channel：第一个channel类型是Entries（type Entries []*Entry），用来传输cluster所包含的主机信息，第二个channel用来通知是否有错误发生。

Cluster.monitorDiscovery功能是整理Cluster.engines：

select {
    case entries := <-ch:
        added, removed := currentEntries.Diff(entries)
        currentEntries = entries

        // Remove engines first. `addEngine` will refuse to add an engine
        // if there's already an engine with the same ID.  If an engine
        // changes address, we have to first remove it then add it back.
        for _, entry := range removed {
            c.removeEngine(entry.String())
        }

        for _, entry := range added {
            c.addEngine(entry.String())
        }
    case err := <-errCh:
        log.Errorf("Discovery error: %v", err)
    }

而Cluster.monitorPendingEngines则是验证处于pending状态的Cluster.engines现在是否能够连接上：

for _, e := range pEngines {
        if e.TimeToValidate() {
            go c.validatePendingEngine(e)
        }
    }

cluster\cluster.go文件定义了Cluster interface：

// Cluster is exported
type Cluster interface {
    // Create a container
    CreateContainer(config *ContainerConfig, name string, authConfig *types.AuthConfig) (*Container, error)

    // Remove a container
    RemoveContainer(container *Container, force, volumes bool) error

    // Return all images
    Images() Images
    ......
}

目前实现了mesos（cluster/mesos/cluster.go，目前仍然处于试验阶段）和swarm（cluster/swarm/cluster.go）两种driver。如果你想实现自己的driver，就要实现上面Cluster interface的所有函数。

Docker Swarm的join命令的定义：

    {
        Name:      "join",
        ShortName: "j",
        Usage:     "Join a docker cluster",
        Flags:     []cli.Flag{flJoinAdvertise, flHeartBeat, flTTL, flJoinRandomDelay, flDiscoveryOpt},
        Action:    join,
    },

flHeartBeat的默认值是60s，而flTTL默认值是180s:

flHeartBeat = cli.StringFlag{
    Name:  "heartbeat",
    Value: "60s",
    Usage: "period between each heartbeat",
}
flTTL = cli.StringFlag{
    Name:  "ttl",
    Value: "180s",
    Usage: "set the expiration of an ephemeral node",
}

join函数的核心代码：

......
for {
    log.WithFields(log.Fields{"addr": addr, "discovery": dflag}).Infof("Registering on the discovery service every %s...", hb)
    if err := d.Register(addr); err != nil {
        log.Error(err)
    }
    time.Sleep(hb)
}
......

token.Register函数实现：

func (s *Discovery) Register(addr string) error {
    buf := strings.NewReader(addr)

    resp, err := http.Post(fmt.Sprintf("%s/%s/%s?ttl=%d", s.url,
        "clusters", s.token, uint64(s.ttl.Seconds())), "application/json", buf)

    if err != nil {
        return err
    }

    resp.Body.Close()
    return nil
}

join命令其实就是每隔heartbeat时间（例如，60s），向https://discovery.hub.docker.com/v1/clusters/token/ttl=180（ttl取默认值），注册一下当前Docker的地址（IP:PORT）。

Docker Swarm的create命令代码很简单：

func create(c *cli.Context) {
    if len(c.Args()) != 0 {
        log.Fatalf("the `create` command takes no arguments. See '%s create --help'.", c.App.Name)
    }
    discovery := &token.Discovery{}
    discovery.Initialize("", 0, 0, nil)
    token, err := discovery.CreateCluster()
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(token)
}

而token.CreateCluster()函数的实现：

// CreateCluster returns a unique cluster token
func (s *Discovery) CreateCluster() (string, error) {
    resp, err := http.Post(fmt.Sprintf("%s/%s", s.url, "clusters"), "", nil)
    if err != nil {
        return "", err
    }

    defer resp.Body.Close()
    token, err := ioutil.ReadAll(resp.Body)
    return string(token), err
}

其实就是向https://discovery.hub.docker.com/v1/clusters发送一个https post请求，然后得到唯一的一个token，利用这个token来创建和管理cluster。

Docker的discovery package提供了实现新的backend定义的interface（discovey.go）：

package discovery

import (
    "errors"
    "time"
)

var (
    // ErrNotSupported is returned when a discovery service is not supported.
    ErrNotSupported = errors.New("discovery service not supported")

    // ErrNotImplemented is returned when discovery feature is not implemented
    // by discovery backend.
    ErrNotImplemented = errors.New("not implemented in this discovery service")
)

// Watcher provides watching over a cluster for nodes joining and leaving.
type Watcher interface {
    // Watch the discovery for entry changes.
    // Returns a channel that will receive changes or an error.
    // Providing a non-nil stopCh can be used to stop watching.
    Watch(stopCh <-chan struct{}) (<-chan Entries, <-chan error)
}

// Backend is implemented by discovery backends which manage cluster entries.
type Backend interface {
    // Watcher must be provided by every backend.
    Watcher

    // Initialize the discovery with URIs, a heartbeat, a ttl and optional settings.
    Initialize(string, time.Duration, time.Duration, map[string]string) error

    // Register to the discovery.
    Register(string) error
}

Watcher interface用于node加入或移除出cluster时的回调函数。Initialize用于backend的初始化，而Register用于将node加入cluster。