Merge pull request #1730 from weaveworks/1725-plugin-dns

Fix DNS integration in Weave docker plugin

Makefile

@@ -82,7 +82,7 @@ $(NETCHECK_EXE): prog/netcheck/netcheck.go
 $(SIGPROXY_EXE): prog/sigproxy/main.go
 $(WEAVEWAIT_EXE): prog/weavewait/*.go net/*.go
 $(DOCKERTLSARGS_EXE): prog/docker_tls_args/*.go
-$(DOCKERPLUGIN_EXE): prog/plugin/*.go plugin/*.go plugin/skel/*.go
+$(DOCKERPLUGIN_EXE): prog/plugin/*.go plugin/*.go plugin/skel/*.go api/*.go common/docker/*.go
 
 $(WEAVEWAIT_EXE) $(SIGPROXY_EXE) $(DOCKERTLSARGS_EXE) $(DOCKERPLUGIN_EXE):
 	go get -tags netgo ./$(@D)

api/dns.go

@@ -0,0 +1,18 @@
+package api
+
+import (
+	"fmt"
+	"net/url"
+)
+
+func (client *Client) RegisterWithDNS(ID string, fqdn string, ip string) error {
+	data := url.Values{}
+	data.Add("fqdn", fqdn)
+	_, err := httpVerb("PUT", fmt.Sprintf("%s/name/%s/%s", client.baseUrl, ID, ip), data)
+	return err
+}
+
+func (client *Client) DeregisterWithDNS(ID string, ip string) error {
+	_, err := httpVerb("DELETE", fmt.Sprintf("%s/name/%s/%s", client.baseUrl, ID, ip), nil)
+	return err
+}

api/ipam.go

@@ -0,0 +1,43 @@
+package api
+
+import (
+	"fmt"
+	"net"
+
+	. "github.com/weaveworks/weave/common"
+)
+
+func (client *Client) ipamOp(ID string, op string) (*net.IPNet, error) {
+	Log.Debugf("IPAM operation %s for %s", op, ID)
+	ip, err := httpVerb(op, fmt.Sprintf("%s/ip/%s", client.baseUrl, ID), nil)
+	if err != nil {
+		return nil, err
+	}
+	return parseIP(ip)
+}
+
+// returns an IP for the ID given, allocating a fresh one if necessary
+func (client *Client) AllocateIP(ID string) (*net.IPNet, error) {
+	return client.ipamOp(ID, "POST")
+}
+
+// returns an IP for the ID given, or nil if one has not been
+// allocated
+func (client *Client) LookupIP(ID string) (*net.IPNet, error) {
+	return client.ipamOp(ID, "GET")
+}
+
+// release an IP which is no longer needed
+func (client *Client) ReleaseIP(ID string) error {
+	_, err := client.ipamOp(ID, "DELETE")
+	return err
+}
+
+func parseIP(body string) (*net.IPNet, error) {
+	ip, ipnet, err := net.ParseCIDR(string(body))
+	if err != nil {
+		return nil, err
+	}
+	ipnet.IP = ip
+	return ipnet, nil
+}

api/weaveapi.go

@@ -0,0 +1,59 @@
+package api
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"strings"
+
+	. "github.com/weaveworks/weave/common"
+)
+
+const (
+	WeaveHttpPort = 6784
+)
+
+type Client struct {
+	baseUrl string
+}
+
+func httpVerb(verb string, url string, values url.Values) (string, error) {
+	Log.Debugf("weave %s to %s with %v", verb, url, values)
+	var body io.Reader
+	if values != nil {
+		body = strings.NewReader(values.Encode())
+	}
+	req, err := http.NewRequest(verb, url, body)
+	if err != nil {
+		return "", err
+	}
+	if values != nil {
+		req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	}
+	if resp, err := http.DefaultClient.Do(req); err != nil {
+		return "", err
+	} else {
+		defer resp.Body.Close()
+		body, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			return "", err
+		}
+		if resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusNoContent {
+			return string(body), nil
+		} else {
+			return "", errors.New(resp.Status + ": " + string(body))
+		}
+	}
+}
+
+func NewClient(addr string) *Client {
+	return &Client{baseUrl: fmt.Sprintf("http://%s:%d", addr, WeaveHttpPort)}
+}
+
+func (client *Client) Connect(remote string) error {
+	_, err := httpVerb("POST", client.baseUrl+"/connect", url.Values{"peer": {remote}})
+	return err
+}

common/docker/client.go

@@ -1,7 +1,9 @@
 package docker
 
 import (
+	"errors"
 	"github.com/fsouza/go-dockerclient"
+
 	. "github.com/weaveworks/weave/common"
 )
 
@@ -80,3 +82,28 @@ func (c *Client) IsContainerNotRunning(idStr string) bool {
 	Log.Errorf("[docker] Could not check container status: %s", err)
 	return false
 }
+
+// This is intended to find an IP address that we can reach the container on;
+// if it is on the Docker bridge network then that address; if on the host network
+// then localhost
+func (c *Client) GetContainerIP(nameOrID string) (string, error) {
+	Log.Debugf("Getting IP for container %s", nameOrID)
+	info, err := c.InspectContainer(nameOrID)
+	if err != nil {
+		return "", err
+	}
+	if info.NetworkSettings.Networks != nil {
+		Log.Debugln("Networks: ", info.NetworkSettings.Networks)
+		if bridgeNetwork, ok := info.NetworkSettings.Networks["bridge"]; ok {
+			return bridgeNetwork.IPAddress, nil
+		} else if _, ok := info.NetworkSettings.Networks["host"]; ok {
+			return "127.0.0.1", nil
+		}
+	} else if info.HostConfig.NetworkMode == "host" {
+		return "127.0.0.1", nil
+	}
+	if info.NetworkSettings.IPAddress == "" {
+		return "", errors.New("No IP address found for container " + nameOrID)
+	}
+	return info.NetworkSettings.IPAddress, nil
+}

plugin/dns.go

@@ -1,55 +0,0 @@
-package plugin
-
-import (
-	"fmt"
-	"net/http"
-	"net/url"
-	"strings"
-)
-
-func (d *dockerer) registerWithDNS(ID string, fqdn string, ip string) error {
-	dnsip, err := d.getContainerBridgeIP(WeaveDNSContainer)
-	if err != nil {
-		return fmt.Errorf("nameserver not available: %s", err)
-	}
-	data := url.Values{}
-	data.Add("fqdn", fqdn)
-
-	req, err := http.NewRequest("PUT", fmt.Sprintf("http://%s:6785/name/%s/%s", dnsip, ID, ip), strings.NewReader(data.Encode()))
-	if err != nil {
-		return err
-	}
-	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
-
-	cl := &http.Client{}
-	res, err := cl.Do(req)
-	if err != nil {
-		return err
-	}
-	if res.StatusCode != http.StatusOK {
-		return fmt.Errorf("non-OK status from nameserver: %d", res.StatusCode)
-	}
-	return nil
-}
-
-func (d *dockerer) deregisterWithDNS(ID string, ip string) error {
-	dnsip, err := d.getContainerBridgeIP(WeaveDNSContainer)
-	if err != nil {
-		return fmt.Errorf("nameserver not available: %s", err)
-	}
-
-	req, err := http.NewRequest("DELETE", fmt.Sprintf("http://%s:6785/name/%s/%s", dnsip, ID, ip), nil)
-	if err != nil {
-		return err
-	}
-
-	cl := &http.Client{}
-	res, err := cl.Do(req)
-	if err != nil {
-		return err
-	}
-	if res.StatusCode != http.StatusOK {
-		return fmt.Errorf("non-OK status from nameserver: %d", res.StatusCode)
-	}
-	return nil
-}

plugin/docker.go

@@ -1,23 +0,0 @@
-package plugin
-
-import (
-	"github.com/fsouza/go-dockerclient"
-	. "github.com/weaveworks/weave/common"
-)
-
-type dockerer struct {
-	client *docker.Client
-}
-
-func (d *dockerer) getContainerBridgeIP(nameOrID string) (string, error) {
-	Log.Debugf("Getting IP for container %s", nameOrID)
-	info, err := d.InspectContainer(nameOrID)
-	if err != nil {
-		return "", err
-	}
-	return info.NetworkSettings.IPAddress, nil
-}
-
-func (d *dockerer) InspectContainer(nameOrId string) (*docker.Container, error) {
-	return d.client.InspectContainer(nameOrId)
-}

plugin/driver.go

@@ -2,15 +2,16 @@ package plugin
 
 import (
 	"fmt"
+	"sync"
 
 	"github.com/docker/libnetwork/drivers/remote/api"
 	"github.com/docker/libnetwork/types"
 
 	. "github.com/weaveworks/weave/common"
+	"github.com/weaveworks/weave/common/docker"
 	"github.com/weaveworks/weave/common/odp"
 	"github.com/weaveworks/weave/plugin/skel"
 
-	docker "github.com/fsouza/go-dockerclient"
 	"github.com/vishvananda/netlink"
 )
 
@@ -22,6 +23,8 @@ const (
 type driver struct {
 	version    string
 	nameserver string
+	sync.RWMutex
+	endpoints map[string]struct{}
 }
 
 func New(version string, nameserver string) (skel.Driver, error) {
@@ -30,15 +33,17 @@ func New(version string, nameserver string) (skel.Driver, error) {
 		return nil, errorf("could not connect to docker: %s", err)
 	}
 
-	_, err = NewWatcher(client)
+	driver := &driver{
+		nameserver: nameserver,
+		version:    version,
+		endpoints:  make(map[string]struct{}),
+	}
+
+	_, err = NewWatcher(client, driver)
 	if err != nil {
 		return nil, err
 	}
-
-	return &driver{
-		nameserver: nameserver,
-		version:    version,
-	}, nil
+	return driver, nil
 }
 
 func errorf(format string, a ...interface{}) error {
@@ -76,18 +81,31 @@ func (driver *driver) CreateEndpoint(create *api.CreateEndpointRequest) (*api.Cr
 	if create.Interface == nil {
 		return nil, fmt.Errorf("Not supported: creating an interface from within CreateEndpoint")
 	}
+	driver.Lock()
+	driver.endpoints[endID] = struct{}{}
+	driver.Unlock()
 	resp := &api.CreateEndpointResponse{}
 
 	Log.Infof("Create endpoint %s %+v", endID, resp)
 	return resp, nil
 }
 
-func (driver *driver) DeleteEndpoint(delete *api.DeleteEndpointRequest) error {
-	Log.Debugf("Delete endpoint request: %+v", delete)
-	Log.Infof("Delete endpoint %s", delete.EndpointID)
+func (driver *driver) DeleteEndpoint(deleteReq *api.DeleteEndpointRequest) error {
+	Log.Debugf("Delete endpoint request: %+v", deleteReq)
+	Log.Infof("Delete endpoint %s", deleteReq.EndpointID)
+	driver.Lock()
+	delete(driver.endpoints, deleteReq.EndpointID)
+	driver.Unlock()
 	return nil
 }
 
+func (driver *driver) HasEndpoint(endpointID string) bool {
+	driver.Lock()
+	_, found := driver.endpoints[endpointID]
+	driver.Unlock()
+	return found
+}
+
 func (driver *driver) EndpointInfo(req *api.EndpointInfoRequest) (*api.EndpointInfoResponse, error) {
 	Log.Debugf("Endpoint info request: %+v", req)
 	Log.Infof("Endpoint info %s", req.EndpointID)

plugin/ipam.go

@@ -1,89 +0,0 @@
-package plugin
-
-import (
-	"fmt"
-	"io/ioutil"
-	"net"
-	"net/http"
-
-	. "github.com/weaveworks/weave/common"
-)
-
-func (d *dockerer) ipamOp(ID string, op string) (*net.IPNet, error) {
-	weaveip, err := d.getContainerBridgeIP(WeaveContainer)
-	Log.Debugf("IPAM operation %s for %s", op, ID)
-	if err != nil {
-		return nil, err
-	}
-
-	var res *http.Response
-
-	url := fmt.Sprintf("http://%s:6784/ip/%s", weaveip, ipamID(ID))
-	Log.Debugf("Attempting to %s to %s", op, url)
-	if op == "POST" {
-		res, err = http.Post(url, "", nil)
-	} else if op == "GET" {
-		res, err = http.Get(url)
-	}
-
-	if err != nil {
-		return nil, err
-	}
-	if res.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("received status %d from IPAM", res.StatusCode)
-	}
-	body, err := ioutil.ReadAll(res.Body)
-	if err != nil {
-		return nil, err
-	}
-	return parseIP(string(body))
-}
-
-// returns an IP for the ID given, allocating a fresh one if necessary
-func (d *dockerer) allocateIP(ID string) (*net.IPNet, error) {
-	return d.ipamOp(ID, "POST")
-}
-
-// returns an IP for the ID given, or nil if one has not been
-// allocated
-func (d *dockerer) lookupIP(ID string) (*net.IPNet, error) {
-	return d.ipamOp(ID, "GET")
-}
-
-// release an IP which is no longer needed
-func (d *dockerer) releaseIP(ID string) error {
-	weaveip, err := d.getContainerBridgeIP(WeaveContainer)
-	if err != nil {
-		return err
-	}
-	req, err := http.NewRequest("DELETE", fmt.Sprintf("http://%s:6784/ip/%s", weaveip, ipamID(ID)), nil)
-	if err != nil {
-		return err
-	}
-	cl := &http.Client{}
-	res, err := cl.Do(req)
-	if err != nil {
-		return err
-	}
-	if res.StatusCode != http.StatusOK && res.StatusCode != http.StatusNoContent {
-		return fmt.Errorf("unexpected HTTP status code from IP release: %d", res.StatusCode)
-	}
-	return nil
-}
-
-func parseIP(body string) (*net.IPNet, error) {
-	ip, ipnet, err := net.ParseCIDR(string(body))
-	if err != nil {
-		return nil, err
-	}
-	ipnet.IP = ip
-	return ipnet, nil
-}
-
-// If something looking like a container ID is supplied to IPAM, it
-// will try to check that it's running, and return nothing (and empty
-// string) if it's not. So I make sure it doesn't look like a
-// container ID by including non-hex characters.
-func ipamID(ID string) string {
-	return "endpoint:" + ID
-}

plugin/watcher.go

@@ -2,88 +2,69 @@ package plugin
 
 import (
 	"fmt"
-	"strings"
 
-	"github.com/fsouza/go-dockerclient"
+	weaveapi "github.com/weaveworks/weave/api"
 	. "github.com/weaveworks/weave/common"
+	"github.com/weaveworks/weave/common/docker"
 )
 
 const (
-	WeaveDNSContainer = "weavedns"
-	WeaveDomain       = "weave.local"
+	WeaveDomain = "weave.local"
 )
 
 type watcher struct {
-	dockerer
-	networks map[string]bool
-	events   chan *docker.APIEvents
+	client *docker.Client
+	weave  *weaveapi.Client
+	driver *driver
 }
 
 type Watcher interface {
 }
 
-func NewWatcher(client *docker.Client) (Watcher, error) {
-	w := &watcher{
-		dockerer: dockerer{
-			client: client,
-		},
-		networks: make(map[string]bool),
-		events:   make(chan *docker.APIEvents),
-	}
-	err := client.AddEventListener(w.events)
+func NewWatcher(client *docker.Client, driver *driver) (Watcher, error) {
+	w := &watcher{client: client, driver: driver}
+	err := client.AddObserver(w)
 	if err != nil {
 		return nil, err
 	}
 
-	go func() {
-		for event := range w.events {
-			switch event.Status {
-			case "start":
-				w.ContainerStart(event.ID)
-			case "die":
-				w.ContainerDied(event.ID)
-			}
-		}
-	}()
-
 	return w, nil
 }
 
-func (w *watcher) ContainerStart(id string) {
+func (w *watcher) haveWeaveClient() bool {
+	if w.weave == nil {
+		dnsip, err := w.client.GetContainerIP(WeaveContainer)
+		if err != nil {
+			Log.Warningf("nameserver not available: %s", err)
+			return false
+		}
+		w.weave = weaveapi.NewClient(dnsip)
+	}
+	return true
+}
+
+func (w *watcher) ContainerStarted(id string) {
 	Log.Debugf("Container started %s", id)
-	info, err := w.InspectContainer(id)
+	info, err := w.client.InspectContainer(id)
 	if err != nil {
 		Log.Warningf("error inspecting container: %s", err)
 		return
 	}
-	// FIXME: check that it's on our network; but, the docker client lib doesn't know about .NetworkID
-	if isSubdomain(info.Config.Domainname, WeaveDomain) {
-		// one of ours
-		ip := info.NetworkSettings.IPAddress
-		fqdn := fmt.Sprintf("%s.%s", info.Config.Hostname, info.Config.Domainname)
-		if err := w.registerWithDNS(id, fqdn, ip); err != nil {
-			Log.Warningf("unable to register with weaveDNS: %s", err)
+	if !w.haveWeaveClient() {
+		return
+	}
+	// check that it's on our network, via the endpointID
+	for _, net := range info.NetworkSettings.Networks {
+		if w.driver.HasEndpoint(net.EndpointID) {
+			fqdn := fmt.Sprintf("%s.%s", info.Config.Hostname, info.Config.Domainname)
+			if err := w.weave.RegisterWithDNS(id, fqdn, net.IPAddress); err != nil {
+				Log.Warningf("unable to register with weaveDNS: %s", err)
+			}
 		}
 	}
 }
 
 func (w *watcher) ContainerDied(id string) {
-	Log.Debugf("Container died %s", id)
-	info, err := w.InspectContainer(id)
-	if err != nil {
-		Log.Warningf("error inspecting container: %s", err)
-		return
-	}
-	if isSubdomain(info.Config.Domainname, WeaveDomain) {
-		ip := info.NetworkSettings.IPAddress
-		if err := w.deregisterWithDNS(id, ip); err != nil {
-			Log.Warningf("unable to deregister with weaveDNS: %s", err)
-		}
-	}
-}
-
-// Cheap and cheerful way to check x is, or is a subdomain, of
-// y. Neither are expected to start with a '.'.
-func isSubdomain(x string, y string) bool {
-	return x == y || strings.HasSuffix(x, "."+y)
+	// don't need to do this as WeaveDNS removes names on container died anyway
+	// (note by the time we get this event we can't see the EndpointID)
 }