From 350b8efb0b374ad3a53babceb9300f3f983f7dda Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 26 Nov 2015 13:20:11 +0000 Subject: [PATCH] Move Weave docker plugin into this repo --- .gitignore | 2 + Makefile | 18 +++- plugin/dns.go | 55 +++++++++++ plugin/docker.go | 23 +++++ plugin/driver.go | 183 +++++++++++++++++++++++++++++++++++ plugin/ipam.go | 89 +++++++++++++++++ plugin/skel/listener.go | 206 ++++++++++++++++++++++++++++++++++++++++ plugin/watcher.go | 89 +++++++++++++++++ prog/plugin/Dockerfile | 3 + prog/plugin/main.go | 78 +++++++++++++++ 10 files changed, 741 insertions(+), 5 deletions(-) create mode 100644 plugin/dns.go create mode 100644 plugin/docker.go create mode 100644 plugin/driver.go create mode 100644 plugin/ipam.go create mode 100644 plugin/skel/listener.go create mode 100644 plugin/watcher.go create mode 100644 prog/plugin/Dockerfile create mode 100644 prog/plugin/main.go diff --git a/.gitignore b/.gitignore index 4506fcc5..1934d8f6 100644 --- a/.gitignore +++ b/.gitignore @@ -42,12 +42,14 @@ prog/weavewait/weavewait prog/weavewait/weavewait_noop prog/netcheck/netcheck prog/docker_tls_args/docker_tls_args +prog/plugin/plugin testing/cover/cover testing/runner/runner releases .weaver.uptodate .weavedns.uptodate .weaveexec.uptodate +.dockerplugin.uptodate weave.tar.gz prog/weaveexec/weave prog/weaveexec/weavehosts diff --git a/Makefile b/Makefile index b4878f0a..d802efc0 100644 --- a/Makefile +++ b/Makefile @@ -16,19 +16,22 @@ WEAVEWAIT_EXE=prog/weavewait/weavewait WEAVEWAIT_NOOP_EXE=prog/weavewait/weavewait_noop NETCHECK_EXE=prog/netcheck/netcheck DOCKERTLSARGS_EXE=prog/docker_tls_args/docker_tls_args +DOCKERPLUGIN_EXE=prog/plugin/plugin RUNNER_EXE=tools/runner/runner -EXES=$(WEAVER_EXE) $(SIGPROXY_EXE) $(WEAVEPROXY_EXE) $(WEAVEWAIT_EXE) $(WEAVEWAIT_NOOP_EXE) $(NETCHECK_EXE) $(DOCKERTLSARGS_EXE) +EXES=$(WEAVER_EXE) $(SIGPROXY_EXE) $(WEAVEPROXY_EXE) $(WEAVEWAIT_EXE) $(WEAVEWAIT_NOOP_EXE) $(NETCHECK_EXE) $(DOCKERTLSARGS_EXE) $(DOCKERPLUGIN_EXE) WEAVER_UPTODATE=.weaver.uptodate WEAVEEXEC_UPTODATE=.weaveexec.uptodate +DOCKERPLUGIN_UPTODATE=.dockerplugin.uptodate -IMAGES_UPTODATE=$(WEAVER_UPTODATE) $(WEAVEEXEC_UPTODATE) +IMAGES_UPTODATE=$(WEAVER_UPTODATE) $(WEAVEEXEC_UPTODATE) $(DOCKERPLUGIN_UPTODATE) WEAVER_IMAGE=$(DOCKERHUB_USER)/weave WEAVEEXEC_IMAGE=$(DOCKERHUB_USER)/weaveexec +DOCKERPLUGIN_IMAGE=$(DOCKERHUB_USER)/plugin -IMAGES=$(WEAVER_IMAGE) $(WEAVEEXEC_IMAGE) +IMAGES=$(WEAVER_IMAGE) $(WEAVEEXEC_IMAGE) $(DOCKERPLUGIN_IMAGE) WEAVE_EXPORT=weave.tar.gz @@ -74,13 +77,14 @@ $(WEAVER_EXE): router/*.go mesh/*.go ipam/*.go ipam/*/*.go nameserver/*.go prog/ $(WEAVEPROXY_EXE): proxy/*.go prog/weaveproxy/main.go $(NETCHECK_EXE): prog/netcheck/netcheck.go -# Sigproxy and weavewait need separate rules as they fail the netgo check in +# These next programs need separate rules as they fail the netgo check in # the main build stanza due to not importing net package $(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 -$(WEAVEWAIT_EXE) $(SIGPROXY_EXE) $(DOCKERTLSARGS_EXE): +$(WEAVEWAIT_EXE) $(SIGPROXY_EXE) $(DOCKERTLSARGS_EXE) $(DOCKERPLUGIN_EXE): go get -tags netgo ./$(@D) go build $(BUILD_FLAGS) -o $@ ./$(@D) @@ -104,6 +108,10 @@ $(WEAVEEXEC_UPTODATE): prog/weaveexec/Dockerfile $(DOCKER_DISTRIB) weave $(SIGPR $(SUDO) docker build -t $(WEAVEEXEC_IMAGE) prog/weaveexec touch $@ +$(DOCKERPLUGIN_UPTODATE): prog/plugin/Dockerfile $(DOCKERPLUGIN_EXE) + $(SUDO) docker build -t $(DOCKERPLUGIN_IMAGE) prog/plugin + touch $@ + $(WEAVE_EXPORT): $(IMAGES_UPTODATE) $(SUDO) docker save $(addsuffix :latest,$(IMAGES)) | gzip > $@ diff --git a/plugin/dns.go b/plugin/dns.go new file mode 100644 index 00000000..27652cfe --- /dev/null +++ b/plugin/dns.go @@ -0,0 +1,55 @@ +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 +} diff --git a/plugin/docker.go b/plugin/docker.go new file mode 100644 index 00000000..d7b4b337 --- /dev/null +++ b/plugin/docker.go @@ -0,0 +1,23 @@ +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) +} diff --git a/plugin/driver.go b/plugin/driver.go new file mode 100644 index 00000000..8467c58f --- /dev/null +++ b/plugin/driver.go @@ -0,0 +1,183 @@ +package plugin + +import ( + "fmt" + + "github.com/docker/libnetwork/drivers/remote/api" + "github.com/docker/libnetwork/types" + + . "github.com/weaveworks/weave/common" + "github.com/weaveworks/weave/common/odp" + "github.com/weaveworks/weave/plugin/skel" + + docker "github.com/fsouza/go-dockerclient" + "github.com/vishvananda/netlink" +) + +const ( + WeaveContainer = "weave" + WeaveBridge = "weave" +) + +type driver struct { + version string + nameserver string +} + +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) + } + + _, err = NewWatcher(client) + if err != nil { + return nil, err + } + + return &driver{ + nameserver: nameserver, + version: version, + }, nil +} + +func errorf(format string, a ...interface{}) error { + Log.Errorf(format, a...) + return fmt.Errorf(format, a...) +} + +// === protocol handlers + +var caps = &api.GetCapabilityResponse{ + Scope: "global", +} + +func (driver *driver) GetCapabilities() (*api.GetCapabilityResponse, error) { + Log.Debugf("Get capabilities: responded with %+v", caps) + return caps, nil +} + +func (driver *driver) CreateNetwork(create *api.CreateNetworkRequest) error { + Log.Debugf("Create network request %+v", create) + Log.Infof("Create network %s", create.NetworkID) + return nil +} + +func (driver *driver) DeleteNetwork(delete *api.DeleteNetworkRequest) error { + Log.Debugf("Delete network request: %+v", delete) + Log.Infof("Destroy network %s", delete.NetworkID) + return nil +} + +func (driver *driver) CreateEndpoint(create *api.CreateEndpointRequest) (*api.CreateEndpointResponse, error) { + Log.Debugf("Create endpoint request %+v", create) + endID := create.EndpointID + + if create.Interface == nil { + return nil, fmt.Errorf("Not supported: creating an interface from within CreateEndpoint") + } + 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) + return nil +} + +func (driver *driver) EndpointInfo(req *api.EndpointInfoRequest) (*api.EndpointInfoResponse, error) { + Log.Debugf("Endpoint info request: %+v", req) + Log.Infof("Endpoint info %s", req.EndpointID) + return &api.EndpointInfoResponse{Value: map[string]interface{}{}}, nil +} + +func (driver *driver) JoinEndpoint(j *api.JoinRequest) (*api.JoinResponse, error) { + endID := j.EndpointID + + maybeBridge, err := netlink.LinkByName(WeaveBridge) + if err != nil { + return nil, errorf(`bridge "%s" not present; did you launch weave?`, WeaveBridge) + } + + // create and attach local name to the bridge + local := vethPair(endID[:5]) + local.Attrs().MTU = maybeBridge.Attrs().MTU + if err := netlink.LinkAdd(local); err != nil { + return nil, errorf("could not create veth pair: %s", err) + } + + switch maybeBridge.(type) { + case *netlink.Bridge: + if err := netlink.LinkSetMasterByIndex(local, maybeBridge.Attrs().Index); err != nil { + return nil, errorf(`unable to set master: %s`, err) + } + case *netlink.GenericLink: + if maybeBridge.Type() != "openvswitch" { + Log.Errorf("device %s is %+v", WeaveBridge, maybeBridge) + return nil, errorf(`device "%s" is of type "%s"`, WeaveBridge, maybeBridge.Type()) + } + odp.AddDatapathInterface(WeaveBridge, local.Name) + case *netlink.Device: + Log.Warnf("kernel does not report what kind of device %s is, just %+v", WeaveBridge, maybeBridge) + // Assume it's our openvswitch device, and the kernel has not been updated to report the kind. + odp.AddDatapathInterface(WeaveBridge, local.Name) + default: + Log.Errorf("device %s is %+v", WeaveBridge, maybeBridge) + return nil, errorf(`device "%s" not a bridge`, WeaveBridge) + } + if err := netlink.LinkSetUp(local); err != nil { + return nil, errorf(`unable to bring veth up: %s`, err) + } + + ifname := &api.InterfaceName{ + SrcName: local.PeerName, + DstPrefix: "ethwe", + } + + response := &api.JoinResponse{ + InterfaceName: ifname, + } + if driver.nameserver != "" { + routeToDNS := api.StaticRoute{ + Destination: driver.nameserver + "/32", + RouteType: types.CONNECTED, + NextHop: "", + } + response.StaticRoutes = []api.StaticRoute{routeToDNS} + } + Log.Infof("Join endpoint %s:%s to %s", j.NetworkID, j.EndpointID, j.SandboxKey) + return response, nil +} + +func (driver *driver) LeaveEndpoint(leave *api.LeaveRequest) error { + Log.Debugf("Leave request: %+v", leave) + + local := vethPair(leave.EndpointID[:5]) + if err := netlink.LinkDel(local); err != nil { + Log.Warningf("unable to delete veth on leave: %s", err) + } + Log.Infof("Leave %s:%s", leave.NetworkID, leave.EndpointID) + return nil +} + +func (driver *driver) DiscoverNew(disco *api.DiscoveryNotification) error { + Log.Debugf("Dicovery new notification: %+v", disco) + return nil +} + +func (driver *driver) DiscoverDelete(disco *api.DiscoveryNotification) error { + Log.Debugf("Dicovery delete notification: %+v", disco) + return nil +} + +// === + +func vethPair(suffix string) *netlink.Veth { + return &netlink.Veth{ + LinkAttrs: netlink.LinkAttrs{Name: "vethwl" + suffix}, + PeerName: "vethwg" + suffix, + } +} diff --git a/plugin/ipam.go b/plugin/ipam.go new file mode 100644 index 00000000..927f7fff --- /dev/null +++ b/plugin/ipam.go @@ -0,0 +1,89 @@ +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 +} diff --git a/plugin/skel/listener.go b/plugin/skel/listener.go new file mode 100644 index 00000000..f86af91e --- /dev/null +++ b/plugin/skel/listener.go @@ -0,0 +1,206 @@ +package skel + +import ( + "encoding/json" + "fmt" + "net" + "net/http" + + "github.com/gorilla/mux" + + "github.com/docker/libnetwork/drivers/remote/api" +) + +const ( + MethodReceiver = "NetworkDriver" +) + +type Driver interface { + GetCapabilities() (*api.GetCapabilityResponse, error) + CreateNetwork(create *api.CreateNetworkRequest) error + DeleteNetwork(delete *api.DeleteNetworkRequest) error + CreateEndpoint(create *api.CreateEndpointRequest) (*api.CreateEndpointResponse, error) + DeleteEndpoint(delete *api.DeleteEndpointRequest) error + EndpointInfo(req *api.EndpointInfoRequest) (*api.EndpointInfoResponse, error) + JoinEndpoint(j *api.JoinRequest) (response *api.JoinResponse, error error) + LeaveEndpoint(leave *api.LeaveRequest) error + DiscoverNew(discover *api.DiscoveryNotification) error + DiscoverDelete(delete *api.DiscoveryNotification) error +} + +type listener struct { + d Driver +} + +func Listen(socket net.Listener, driver Driver) error { + router := mux.NewRouter() + router.NotFoundHandler = http.HandlerFunc(notFound) + + router.Methods("POST").Path("/Plugin.Activate").HandlerFunc(handshake) + + handleMethod := func(method string, h http.HandlerFunc) { + router.Methods("POST").Path(fmt.Sprintf("/%s.%s", MethodReceiver, method)).HandlerFunc(h) + } + + listener := &listener{driver} + + 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) + + return http.Serve(socket, router) +} + +// === protocol handlers + +type handshakeResp struct { + Implements []string +} + +func handshake(w http.ResponseWriter, r *http.Request) { + err := json.NewEncoder(w).Encode(&handshakeResp{ + []string{"NetworkDriver"}, + }) + if err != nil { + sendError(w, "encode error", http.StatusInternalServerError) + return + } +} + +func (listener *listener) getCapabilities(w http.ResponseWriter, r *http.Request) { + caps, err := listener.d.GetCapabilities() + objectOrErrorResponse(w, caps, err) +} + +func (listener *listener) createNetwork(w http.ResponseWriter, r *http.Request) { + var create api.CreateNetworkRequest + err := json.NewDecoder(r.Body).Decode(&create) + if err != nil { + sendError(w, "Unable to decode JSON payload: "+err.Error(), http.StatusBadRequest) + return + } + emptyOrErrorResponse(w, listener.d.CreateNetwork(&create)) +} + +func (listener *listener) deleteNetwork(w http.ResponseWriter, r *http.Request) { + var delete api.DeleteNetworkRequest + if err := json.NewDecoder(r.Body).Decode(&delete); err != nil { + sendError(w, "Unable to decode JSON payload: "+err.Error(), http.StatusBadRequest) + return + } + emptyOrErrorResponse(w, listener.d.DeleteNetwork(&delete)) +} + +func (listener *listener) createEndpoint(w http.ResponseWriter, r *http.Request) { + var create api.CreateEndpointRequest + if err := json.NewDecoder(r.Body).Decode(&create); err != nil { + sendError(w, "unable to decode JSON payload: "+err.Error(), http.StatusBadRequest) + return + } + res, err := listener.d.CreateEndpoint(&create) + objectOrErrorResponse(w, res, err) +} + +func (listener *listener) deleteEndpoint(w http.ResponseWriter, r *http.Request) { + var delete api.DeleteEndpointRequest + if err := json.NewDecoder(r.Body).Decode(&delete); err != nil { + sendError(w, "Could not decode JSON encode payload", http.StatusBadRequest) + return + } + emptyOrErrorResponse(w, listener.d.DeleteEndpoint(&delete)) +} + +func (listener *listener) infoEndpoint(w http.ResponseWriter, r *http.Request) { + var req api.EndpointInfoRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + sendError(w, "Could not decode JSON encode payload", http.StatusBadRequest) + return + } + info, err := listener.d.EndpointInfo(&req) + objectOrErrorResponse(w, info, err) +} + +func (listener *listener) joinEndpoint(w http.ResponseWriter, r *http.Request) { + var join api.JoinRequest + if err := json.NewDecoder(r.Body).Decode(&join); err != nil { + sendError(w, "Could not decode JSON encode payload", http.StatusBadRequest) + return + } + res, err := listener.d.JoinEndpoint(&join) + objectOrErrorResponse(w, res, err) +} + +func (listener *listener) leaveEndpoint(w http.ResponseWriter, r *http.Request) { + var l api.LeaveRequest + if err := json.NewDecoder(r.Body).Decode(&l); err != nil { + sendError(w, "Could not decode JSON encode payload", http.StatusBadRequest) + return + } + emptyOrErrorResponse(w, listener.d.LeaveEndpoint(&l)) +} + +func (listener *listener) discoverNew(w http.ResponseWriter, r *http.Request) { + var disco api.DiscoveryNotification + if err := json.NewDecoder(r.Body).Decode(&disco); err != nil { + sendError(w, "Could not decode JSON encode payload", http.StatusBadRequest) + return + } + emptyOrErrorResponse(w, listener.d.DiscoverNew(&disco)) +} + +func (listener *listener) discoverDelete(w http.ResponseWriter, r *http.Request) { + var disco api.DiscoveryNotification + if err := json.NewDecoder(r.Body).Decode(&disco); err != nil { + sendError(w, "Could not decode JSON encode payload", http.StatusBadRequest) + return + } + emptyOrErrorResponse(w, listener.d.DiscoverDelete(&disco)) +} + +// === + +func notFound(w http.ResponseWriter, r *http.Request) { + http.NotFound(w, r) +} + +func sendError(w http.ResponseWriter, msg string, code int) { + http.Error(w, msg, code) +} + +func errorResponse(w http.ResponseWriter, fmtString string, item ...interface{}) { + json.NewEncoder(w).Encode(map[string]string{ + "Err": fmt.Sprintf(fmtString, item...), + }) +} + +func objectResponse(w http.ResponseWriter, obj interface{}) { + if err := json.NewEncoder(w).Encode(obj); err != nil { + sendError(w, "Could not JSON encode response", http.StatusInternalServerError) + return + } +} + +func emptyResponse(w http.ResponseWriter) { + json.NewEncoder(w).Encode(map[string]string{}) +} + +func objectOrErrorResponse(w http.ResponseWriter, obj interface{}, err error) { + if err != nil { + errorResponse(w, err.Error()) + return + } + objectResponse(w, obj) +} + +func emptyOrErrorResponse(w http.ResponseWriter, err error) { + if err != nil { + errorResponse(w, err.Error()) + return + } + emptyResponse(w) +} diff --git a/plugin/watcher.go b/plugin/watcher.go new file mode 100644 index 00000000..50f87dd5 --- /dev/null +++ b/plugin/watcher.go @@ -0,0 +1,89 @@ +package plugin + +import ( + "fmt" + "strings" + + "github.com/fsouza/go-dockerclient" + . "github.com/weaveworks/weave/common" +) + +const ( + WeaveDNSContainer = "weavedns" + WeaveDomain = "weave.local" +) + +type watcher struct { + dockerer + networks map[string]bool + events chan *docker.APIEvents +} + +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) + 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) { + Log.Debugf("Container started %s", id) + info, err := w.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) + } + } +} + +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) +} diff --git a/prog/plugin/Dockerfile b/prog/plugin/Dockerfile new file mode 100644 index 00000000..946f0b28 --- /dev/null +++ b/prog/plugin/Dockerfile @@ -0,0 +1,3 @@ +FROM weaveworks/weaveexec +ADD ./plugin /home/weave/ +ENTRYPOINT ["/home/weave/plugin"] diff --git a/prog/plugin/main.go b/prog/plugin/main.go new file mode 100644 index 00000000..77438cd4 --- /dev/null +++ b/prog/plugin/main.go @@ -0,0 +1,78 @@ +package main + +import ( + "flag" + "fmt" + "net" + "os" + "os/signal" + "syscall" + + . "github.com/weaveworks/weave/common" + "github.com/weaveworks/weave/plugin" + "github.com/weaveworks/weave/plugin/skel" +) + +var version = "(unreleased version)" + +func main() { + var ( + justVersion bool + address string + nameserver string + logLevel string + ) + + flag.BoolVar(&justVersion, "version", false, "print version and exit") + flag.StringVar(&logLevel, "log-level", "info", "logging level (debug, info, warning, error)") + flag.StringVar(&address, "socket", "/run/docker/plugins/weave.sock", "socket on which to listen") + flag.StringVar(&nameserver, "nameserver", "", "nameserver to provide to containers") + + flag.Parse() + + if justVersion { + fmt.Printf("weave plugin %s\n", version) + os.Exit(0) + } + + SetLogLevel(logLevel) + + Log.Println("Weave plugin", version, "Command line options:", os.Args) + + var d skel.Driver + d, err := plugin.New(version, nameserver) + if err != nil { + Log.Fatalf("unable to create driver: %s", err) + } + + var listener net.Listener + + // remove socket from last invocation + if err := os.Remove(address); err != nil && !os.IsNotExist(err) { + Log.Fatal(err) + } + listener, err = net.Listen("unix", address) + if err != nil { + Log.Fatal(err) + } + defer listener.Close() + + sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, os.Interrupt, os.Kill, syscall.SIGTERM) + + endChan := make(chan error, 1) + go func() { + endChan <- skel.Listen(listener, d) + }() + + select { + case sig := <-sigChan: + Log.Debugf("Caught signal %s; shutting down", sig) + case err := <-endChan: + if err != nil { + Log.Errorf("Error from listener: ", err) + listener.Close() + os.Exit(1) + } + } +}