diff --git a/Makefile b/Makefile index b04ed091..7680e1a7 100644 --- a/Makefile +++ b/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): diff --git a/api/dns.go b/api/dns.go index 750c34f2..36ec8630 100644 --- a/api/dns.go +++ b/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 } diff --git a/api/ipam.go b/api/ipam.go index 91a3053a..8be57b27 100644 --- a/api/ipam.go +++ b/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 { diff --git a/api/weaveapi.go b/api/weaveapi.go index d75394bd..8be6b334 100644 --- a/api/weaveapi.go +++ b/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 } diff --git a/ipam/allocate.go b/ipam/allocate.go index e8950e37..d6dd2f10 100644 --- a/ipam/allocate.go +++ b/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} diff --git a/ipam/http.go b/ipam/http.go index a86fbc50..eb51b87a 100644 --- a/ipam/http.go +++ b/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"] diff --git a/plugin/ipam/driver.go b/plugin/ipam/driver.go new file mode 100644 index 00000000..1fbdf91b --- /dev/null +++ b/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()) +} diff --git a/plugin/driver.go b/plugin/net/driver.go similarity index 95% rename from plugin/driver.go rename to plugin/net/driver.go index e5b1e413..5eb5c984 100644 --- a/plugin/driver.go +++ b/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 } diff --git a/plugin/watcher.go b/plugin/net/watcher.go similarity index 78% rename from plugin/watcher.go rename to plugin/net/watcher.go index fbf43014..05241d4f 100644 --- a/plugin/watcher.go +++ b/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) { diff --git a/plugin/skel/listener.go b/plugin/skel/listener.go index f86af91e..2b4deb71 100644 --- a/plugin/skel/listener.go +++ b/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) } diff --git a/prog/plugin/main.go b/prog/plugin/main.go index 659d5a8a..4ae75164 100644 --- a/prog/plugin/main.go +++ b/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 {