Merge branch 'JeffChien-discovery'

ROADMAP.md

@@ -24,6 +24,6 @@ Docker Swarm Roadmap
 * [ ] Discovery backends
   * [x]    etcd
   * [ ]    zookeeper
-  * [ ]    consul
+  * [x]    consul
   * [x]    hub 
   * [x]    file

discovery/README.md

@@ -78,6 +78,29 @@ $ swarm list --discovery etcd://<etcd_ip>/<path>
 http://<node_ip:2375>
 ```
 
+###### Using consul
+
+```bash
+# on each of your nodes, start the swarm agent
+#  <node_ip> doesn't have to be public (eg. 192.168.0.X),
+#  as long as the other nodes can reach it, it is fine.
+$ swarm join --discovery consul://<consul_addr>/<path> --addr=<node_ip:2375>
+
+# start the manager on any machine or your laptop
+$ swarm manage --discovery consul://<consul_addr>/<path> -H=<swarm_ip:swarm_port>
+
+# use the regular docker cli
+$ docker -H <swarm_ip:swarm_port> info
+$ docker -H <swarm_ip:swarm_port> run ...
+$ docker -H <swarm_ip:swarm_port> ps
+$ docker -H <swarm_ip:swarm_port> logs ...
+...
+
+# list nodes in your cluster
+$ swarm list --discovery consul://<consul_addr>/<path>
+http://<node_ip:2375>
+```
+
 ## Contributing
 
 Contributing a new discovery backend is easy,

discovery/consul/consul.go

@@ -0,0 +1,108 @@
+package consul
+
+import (
+	"errors"
+	"path"
+	"strings"
+	"time"
+
+	log "github.com/Sirupsen/logrus"
+	consul "github.com/armon/consul-api"
+	"github.com/docker/swarm/discovery"
+)
+
+type ConsulDiscoveryService struct {
+	heartbeat time.Duration
+	client    *consul.Client
+	prefix    string
+	lastIndex uint64
+}
+
+func init() {
+	discovery.Register("consul", &ConsulDiscoveryService{})
+}
+
+func (s *ConsulDiscoveryService) Initialize(uris string, heartbeat int) error {
+	parts := strings.SplitN(uris, "/", 2)
+	if len(parts) < 2 {
+		return errors.New("missing consul prefix")
+	}
+	addr := parts[0]
+	path := parts[1]
+
+	config := consul.DefaultConfig()
+	config.Address = addr
+
+	client, err := consul.NewClient(config)
+	if err != nil {
+		return err
+	}
+	s.client = client
+	s.heartbeat = time.Duration(heartbeat) * time.Second
+	s.prefix = path + "/"
+	kv := s.client.KV()
+	p := &consul.KVPair{Key: s.prefix, Value: nil}
+	if _, err = kv.Put(p, nil); err != nil {
+		return err
+	}
+	_, meta, err := kv.Get(s.prefix, nil)
+	if err != nil {
+		return err
+	}
+	s.lastIndex = meta.LastIndex
+	return nil
+}
+func (s *ConsulDiscoveryService) Fetch() ([]*discovery.Node, error) {
+	kv := s.client.KV()
+	pairs, _, err := kv.List(s.prefix, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	var nodes []*discovery.Node
+
+	for _, pair := range pairs {
+		if pair.Key == s.prefix {
+			continue
+		}
+		nodes = append(nodes, discovery.NewNode(string(pair.Value)))
+	}
+	return nodes, nil
+}
+
+func (s *ConsulDiscoveryService) Watch(callback discovery.WatchCallback) {
+	for _ = range s.waitForChange() {
+		nodes, err := s.Fetch()
+		if err == nil {
+			callback(nodes)
+		}
+	}
+}
+
+func (s *ConsulDiscoveryService) Register(addr string) error {
+	kv := s.client.KV()
+	p := &consul.KVPair{Key: path.Join(s.prefix, addr), Value: []byte(addr)}
+	_, err := kv.Put(p, nil)
+	return err
+}
+
+func (s *ConsulDiscoveryService) waitForChange() <-chan uint64 {
+	c := make(chan uint64)
+	go func() {
+		for {
+			kv := s.client.KV()
+			option := &consul.QueryOptions{
+				WaitIndex: s.lastIndex,
+				WaitTime:  s.heartbeat}
+			_, meta, err := kv.List(s.prefix, option)
+			if err != nil {
+				log.Errorln(err)
+				break
+			}
+			s.lastIndex = meta.LastIndex
+			c <- s.lastIndex
+		}
+		close(c)
+	}()
+	return c
+}

discovery/consul/consul_test.go

@@ -0,0 +1,13 @@
+package consul
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestInitialize(t *testing.T) {
+	discovery := &ConsulDiscoveryService{}
+	discovery.Initialize("127.0.0.1:8500/path", 0)
+	assert.Equal(t, discovery.prefix, "path/")
+}

flags.go

@@ -6,7 +6,7 @@ var (
 	flDiscovery = cli.StringFlag{
 		Name:   "discovery",
 		Value:  "",
-		Usage:  "DiscoveryService to use [token://<token>, etcd://<ip1>,<ip2>/<path>, file://path/to/file]",
+		Usage:  "DiscoveryService to use [token://<token>, etcd://<ip1>,<ip2>/<path>, file://path/to/file, consul://<addr>/<path>]",
 		EnvVar: "SWARM_DISCOVERY",
 	}
 	flAddr = cli.StringFlag{

main.go

@@ -8,6 +8,7 @@ import (
 	"github.com/codegangsta/cli"
 
 	"github.com/docker/swarm/discovery"
+	_ "github.com/docker/swarm/discovery/consul"
 	_ "github.com/docker/swarm/discovery/etcd"
 	_ "github.com/docker/swarm/discovery/file"
 	"github.com/docker/swarm/discovery/token"