Merge pull request #1728 from weaveworks/1676-ipam-plugin

Docker IPAM plugin

Makefile

@@ -83,7 +83,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 api/*.go common/docker/*.go
+$(DOCKERPLUGIN_EXE): prog/plugin/*.go plugin/net/*.go plugin/ipam/*.go plugin/skel/*.go api/*.go common/docker/*.go
 $(TEST_TLS_EXE): test/tls/*.go
 
 $(WEAVEWAIT_EXE) $(SIGPROXY_EXE) $(DOCKERTLSARGS_EXE) $(DOCKERPLUGIN_EXE) $(TEST_TLS_EXE):

api/dns.go

@@ -8,11 +8,11 @@ import (
 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)
+	_, err := client.httpVerb("PUT", fmt.Sprintf("/name/%s/%s", 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)
+	_, err := client.httpVerb("DELETE", fmt.Sprintf("/name/%s/%s", ID, ip), nil)
 	return err
 }

api/ipam.go

@@ -3,13 +3,10 @@ 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)
+	ip, err := client.httpVerb(op, fmt.Sprintf("/ip/%s", ID), nil)
 	if err != nil {
 		return nil, err
 	}
@@ -33,6 +30,15 @@ func (client *Client) ReleaseIP(ID string) error {
 	return err
 }
 
+func (client *Client) DefaultSubnet() (*net.IPNet, error) {
+	cidr, err := client.httpVerb("GET", fmt.Sprintf("/ipinfo/defaultsubnet"), nil)
+	if err != nil {
+		return nil, err
+	}
+	_, ipnet, err := net.ParseCIDR(cidr)
+	return ipnet, err
+}
+
 func parseIP(body string) (*net.IPNet, error) {
 	ip, ipnet, err := net.ParseCIDR(string(body))
 	if err != nil {

api/weaveapi.go

@@ -18,9 +18,19 @@ const (
 
 type Client struct {
 	baseUrl string
+	resolve func() (string, error)
 }
 
-func httpVerb(verb string, url string, values url.Values) (string, error) {
+func (client *Client) httpVerb(verb string, url string, values url.Values) (string, error) {
+	baseUrl := client.baseUrl
+	if client.resolve != nil {
+		addr, err := client.resolve()
+		if err != nil {
+			return "", err
+		}
+		baseUrl = fmt.Sprintf("http://%s:%d", addr, WeaveHttpPort)
+	}
+	url = baseUrl + url
 	Log.Debugf("weave %s to %s with %v", verb, url, values)
 	var body io.Reader
 	if values != nil {
@@ -53,7 +63,11 @@ func NewClient(addr string) *Client {
 	return &Client{baseUrl: fmt.Sprintf("http://%s:%d", addr, WeaveHttpPort)}
 }
 
+func NewClientWithResolver(resolver func() (string, error)) *Client {
+	return &Client{resolve: resolver}
+}
+
 func (client *Client) Connect(remote string) error {
-	_, err := httpVerb("POST", client.baseUrl+"/connect", url.Values{"peer": {remote}})
+	_, err := client.httpVerb("POST", "/connect", url.Values{"peer": {remote}})
 	return err
 }

ipam/allocate.go

@@ -38,6 +38,11 @@ func (g *allocate) Try(alloc *Allocator) bool {
 	alloc.establishRing()
 
 	if ok, addr := alloc.space.Allocate(g.r); ok {
+		// If caller hasn't supplied a unique ID, file it under the IP address
+		// which lets the caller then release the address using DELETE /ip/address
+		if g.ident == "_" {
+			g.ident = addr.String()
+		}
 		alloc.debugln("Allocated", addr, "for", g.ident, "in", g.r)
 		alloc.addOwned(g.ident, addr)
 		g.resultChan <- allocateResult{addr, nil}

ipam/http.go

@@ -57,6 +57,10 @@ func (alloc *Allocator) handleHTTPAllocate(dockerCli *docker.Client, w http.Resp
 
 // HandleHTTP wires up ipams HTTP endpoints to the provided mux.
 func (alloc *Allocator) HandleHTTP(router *mux.Router, defaultSubnet address.CIDR, dockerCli *docker.Client) {
+	router.Methods("GET").Path("/ipinfo/defaultsubnet").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintf(w, "%s", defaultSubnet)
+	})
+
 	router.Methods("PUT").Path("/ip/{id}/{ip}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		vars := mux.Vars(r)
 		ident := vars["id"]

plugin/ipam/driver.go

@@ -0,0 +1,54 @@
+package ipamplugin
+
+import (
+	"net"
+
+	"github.com/docker/libnetwork/ipamapi"
+	"github.com/weaveworks/weave/api"
+	. "github.com/weaveworks/weave/common"
+	"github.com/weaveworks/weave/common/docker"
+)
+
+const (
+	WeaveContainer = "weave"
+)
+
+type ipam struct {
+	client *docker.Client
+	weave  *api.Client
+}
+
+func NewIpam(client *docker.Client, version string) (ipamapi.Ipam, error) {
+	resolver := func() (string, error) { return client.GetContainerIP(WeaveContainer) }
+	return &ipam{client: client, weave: api.NewClientWithResolver(resolver)}, nil
+}
+
+func (i *ipam) GetDefaultAddressSpaces() (string, string, error) {
+	Log.Debugln("GetDefaultAddressSpaces")
+	return "weavelocal", "weaveglobal", nil
+}
+
+func (i *ipam) RequestPool(addressSpace, pool, subPool string, options map[string]string, v6 bool) (string, *net.IPNet, map[string]string, error) {
+	Log.Debugln("RequestPool", addressSpace, pool, subPool, options)
+	cidr, err := i.weave.DefaultSubnet()
+	Log.Debugln("RequestPool returning ", cidr, err)
+	return "weavepool", cidr, nil, err
+}
+
+func (i *ipam) ReleasePool(poolID string) error {
+	Log.Debugln("ReleasePool", poolID)
+	return nil
+}
+
+func (i *ipam) RequestAddress(poolID string, address net.IP, options map[string]string) (*net.IPNet, map[string]string, error) {
+	Log.Debugln("RequestAddress", poolID, address, options)
+	// Pass magic string to weave IPAM, which then stores the address under its own string
+	ip, err := i.weave.AllocateIP("_")
+	Log.Debugln("allocateIP returned", ip, err)
+	return ip, nil, err
+}
+
+func (i *ipam) ReleaseAddress(poolID string, address net.IP) error {
+	Log.Debugln("ReleaseAddress", poolID, address)
+	return i.weave.ReleaseIP(address.String())
+}

plugin/net/driver.go

@@ -27,19 +27,14 @@ type driver struct {
 	endpoints map[string]struct{}
 }
 
-func New(version string, nameserver string) (skel.Driver, error) {
-	client, err := docker.NewClient("unix:///var/run/docker.sock")
-	if err != nil {
-		return nil, errorf("could not connect to docker: %s", err)
-	}
-
+func New(client *docker.Client, version string, nameserver string) (skel.Driver, error) {
 	driver := &driver{
 		nameserver: nameserver,
 		version:    version,
 		endpoints:  make(map[string]struct{}),
 	}
 
-	_, err = NewWatcher(client, driver)
+	_, err := NewWatcher(client, driver)
 	if err != nil {
 		return nil, err
 	}

plugin/net/watcher.go

@@ -22,7 +22,9 @@ type Watcher interface {
 }
 
 func NewWatcher(client *docker.Client, driver *driver) (Watcher, error) {
-	w := &watcher{client: client, driver: driver}
+	resolver := func() (string, error) { return client.GetContainerIP(WeaveContainer) }
+	w := &watcher{client: client, driver: driver,
+		weave: weaveapi.NewClientWithResolver(resolver)}
 	err := client.AddObserver(w)
 	if err != nil {
 		return nil, err
@@ -31,18 +33,6 @@ func NewWatcher(client *docker.Client, driver *driver) (Watcher, error) {
 	return w, nil
 }
 
-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.client.InspectContainer(id)
@@ -50,9 +40,6 @@ func (w *watcher) ContainerStarted(id string) {
 		Log.Warningf("error inspecting container: %s", err)
 		return
 	}
-	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) {

plugin/skel/listener.go

@@ -9,10 +9,13 @@ import (
 	"github.com/gorilla/mux"
 
 	"github.com/docker/libnetwork/drivers/remote/api"
+	"github.com/docker/libnetwork/ipamapi"
+	iapi "github.com/docker/libnetwork/ipams/remote/api"
 )
 
 const (
-	MethodReceiver = "NetworkDriver"
+	networkReceiver = "NetworkDriver"
+	ipamReceiver    = ipamapi.PluginEndpointType
 )
 
 type Driver interface {
@@ -30,42 +33,67 @@ type Driver interface {
 
 type listener struct {
 	d Driver
+	i ipamapi.Ipam
 }
 
-func Listen(socket net.Listener, driver Driver) error {
+func Listen(socket net.Listener, driver Driver, ipamDriver ipamapi.Ipam) error {
 	router := mux.NewRouter()
 	router.NotFoundHandler = http.HandlerFunc(notFound)
 
-	router.Methods("POST").Path("/Plugin.Activate").HandlerFunc(handshake)
+	listener := &listener{driver, ipamDriver}
 
-	handleMethod := func(method string, h http.HandlerFunc) {
-		router.Methods("POST").Path(fmt.Sprintf("/%s.%s", MethodReceiver, method)).HandlerFunc(h)
+	router.Methods("POST").Path("/Plugin.Activate").HandlerFunc(listener.handshake)
+
+	handleMethod := func(receiver, method string, h http.HandlerFunc) {
+		router.Methods("POST").Path(fmt.Sprintf("/%s.%s", receiver, method)).HandlerFunc(h)
 	}
 
-	listener := &listener{driver}
+	handleMethod(networkReceiver, "GetCapabilities", listener.getCapabilities)
 
-	handleMethod("GetCapabilities", listener.getCapabilities)
-	handleMethod("CreateNetwork", listener.createNetwork)
-	handleMethod("DeleteNetwork", listener.deleteNetwork)
-	handleMethod("CreateEndpoint", listener.createEndpoint)
-	handleMethod("DeleteEndpoint", listener.deleteEndpoint)
-	handleMethod("EndpointOperInfo", listener.infoEndpoint)
-	handleMethod("Join", listener.joinEndpoint)
-	handleMethod("Leave", listener.leaveEndpoint)
+	if driver != nil {
+		handleMethod(networkReceiver, "CreateNetwork", listener.createNetwork)
+		handleMethod(networkReceiver, "DeleteNetwork", listener.deleteNetwork)
+		handleMethod(networkReceiver, "CreateEndpoint", listener.createEndpoint)
+		handleMethod(networkReceiver, "DeleteEndpoint", listener.deleteEndpoint)
+		handleMethod(networkReceiver, "EndpointOperInfo", listener.infoEndpoint)
+		handleMethod(networkReceiver, "Join", listener.joinEndpoint)
+		handleMethod(networkReceiver, "Leave", listener.leaveEndpoint)
+	}
+
+	if ipamDriver != nil {
+		handleMethod(ipamReceiver, "GetDefaultAddressSpaces", listener.getDefaultAddressSpaces)
+		handleMethod(ipamReceiver, "RequestPool", listener.requestPool)
+		handleMethod(ipamReceiver, "ReleasePool", listener.releasePool)
+		handleMethod(ipamReceiver, "RequestAddress", listener.requestAddress)
+		handleMethod(ipamReceiver, "ReleaseAddress", listener.releaseAddress)
+	}
 
 	return http.Serve(socket, router)
 }
 
+func decode(w http.ResponseWriter, r *http.Request, v interface{}) error {
+	err := json.NewDecoder(r.Body).Decode(v)
+	if err != nil {
+		sendError(w, "Unable to decode JSON payload: "+err.Error(), http.StatusBadRequest)
+	}
+	return err
+}
+
 // === protocol handlers
 
 type handshakeResp struct {
 	Implements []string
 }
 
-func handshake(w http.ResponseWriter, r *http.Request) {
-	err := json.NewEncoder(w).Encode(&handshakeResp{
-		[]string{"NetworkDriver"},
-	})
+func (listener *listener) handshake(w http.ResponseWriter, r *http.Request) {
+	var resp handshakeResp
+	if listener.d != nil {
+		resp.Implements = append(resp.Implements, "NetworkDriver")
+	}
+	if listener.i != nil {
+		resp.Implements = append(resp.Implements, "IpamDriver")
+	}
+	err := json.NewEncoder(w).Encode(&resp)
 	if err != nil {
 		sendError(w, "encode error", http.StatusInternalServerError)
 		return
@@ -164,6 +192,70 @@ func (listener *listener) discoverDelete(w http.ResponseWriter, r *http.Request)
 
 // ===
 
+func (listener *listener) getDefaultAddressSpaces(w http.ResponseWriter, r *http.Request) {
+	local, global, err := listener.i.GetDefaultAddressSpaces()
+	response := &iapi.GetAddressSpacesResponse{
+		LocalDefaultAddressSpace:  local,
+		GlobalDefaultAddressSpace: global,
+	}
+	objectOrErrorResponse(w, response, err)
+}
+
+func (listener *listener) requestPool(w http.ResponseWriter, r *http.Request) {
+	var rq iapi.RequestPoolRequest
+	if err := decode(w, r, &rq); err != nil {
+		return
+	}
+	poolID, pool, data, err := listener.i.RequestPool(rq.AddressSpace, rq.Pool, rq.SubPool, rq.Options, rq.V6)
+	if err != nil {
+		errorResponse(w, err.Error())
+		return
+	}
+	response := &iapi.RequestPoolResponse{
+		PoolID: poolID,
+		Pool:   pool.String(),
+		Data:   data,
+	}
+	objectResponse(w, response)
+}
+
+func (listener *listener) releasePool(w http.ResponseWriter, r *http.Request) {
+	var rq iapi.ReleasePoolRequest
+	if err := decode(w, r, &rq); err != nil {
+		return
+	}
+	err := listener.i.ReleasePool(rq.PoolID)
+	emptyOrErrorResponse(w, err)
+}
+
+func (listener *listener) requestAddress(w http.ResponseWriter, r *http.Request) {
+	var rq iapi.RequestAddressRequest
+	if err := decode(w, r, &rq); err != nil {
+		return
+	}
+	address, data, err := listener.i.RequestAddress(rq.PoolID, net.ParseIP(rq.Address), rq.Options)
+	if err != nil {
+		errorResponse(w, err.Error())
+		return
+	}
+	response := &iapi.RequestAddressResponse{
+		Address: address.String(),
+		Data:    data,
+	}
+	objectResponse(w, response)
+}
+
+func (listener *listener) releaseAddress(w http.ResponseWriter, r *http.Request) {
+	var rq iapi.ReleaseAddressRequest
+	if err := decode(w, r, &rq); err != nil {
+		return
+	}
+	err := listener.i.ReleaseAddress(rq.PoolID, net.ParseIP(rq.Address))
+	emptyOrErrorResponse(w, err)
+}
+
+// ===
+
 func notFound(w http.ResponseWriter, r *http.Request) {
 	http.NotFound(w, r)
 }

prog/plugin/main.go

@@ -9,7 +9,9 @@ import (
 	"syscall"
 
 	. "github.com/weaveworks/weave/common"
-	"github.com/weaveworks/weave/plugin"
+	"github.com/weaveworks/weave/common/docker"
+	ipamplugin "github.com/weaveworks/weave/plugin/ipam"
+	netplugin "github.com/weaveworks/weave/plugin/net"
 	"github.com/weaveworks/weave/plugin/skel"
 )
 
@@ -39,8 +41,17 @@ func main() {
 
 	Log.Println("Weave plugin", version, "Command line options:", os.Args)
 
-	var d skel.Driver
-	d, err := plugin.New(version, nameserver)
+	dockerClient, err := docker.NewClient("unix:///var/run/docker.sock")
+	if err != nil {
+		Log.Fatalf("unable to connect to docker: %s", err)
+	}
+
+	d, err := netplugin.New(dockerClient, version, nameserver)
+	if err != nil {
+		Log.Fatalf("unable to create driver: %s", err)
+	}
+
+	i, err := ipamplugin.NewIpam(dockerClient, version)
 	if err != nil {
 		Log.Fatalf("unable to create driver: %s", err)
 	}
@@ -62,7 +73,7 @@ func main() {
 
 	endChan := make(chan error, 1)
 	go func() {
-		endChan <- skel.Listen(listener, d)
+		endChan <- skel.Listen(listener, d, i)
 	}()
 
 	select {