Proxy implementation

This adds a docker remote API proxy to the weaveexec image. The proxy
has hooks for (post-) container start and (pre-) container create; the
former attaches a weave interface to a container given a WEAVE_CIDR
environment entry, and the latter adjusts the entrypoint of a container
to wait for the ethwe interface.

Notes:

None of the hooks need to inspect/change large bodies. In order to avoid
a crash when handling images larger than memory we need to ensure that
the entire request body is never loaded into memory (i.e. No
ioutils.ReadAll).

When proxying a raw-stream response, we can't use http.WriteHeader. It
sets the Transfer-Encoding to chunked, which will throw off the clients.
Instead we have to hijack the socket, and write out http status code
manually.

For chunked transfer-encodings, clients (docker-py) depend on each http
chunk containing exactly one JSON object. Therefore, we cannot re-chunk
the body, and have to pass it across verbatim.

Proxy needs to pass multi-cidrs separately when calling weave. Otherwise
they get smashed together on the command line, and the call fails.

When creating the image, if no entrypoint is specified on the container,
it should use the entrypoint of the image. Code for this is based on
powerstrip-weave.

We handle WEAVE_CIDR='' by acting like it is not set. The reason being,
that because WEAVE_CIDR is inherited from images (along with env vars),
setting it to blank is equivalent to unsetting it.

.gitignore

@@ -39,6 +39,7 @@ releases
 weave.tar
 weavedns.tar
 weaveexec.tar
+weaveexec/proxy
 weaveexec/weave
 weaveexec/sigproxy
 weaveexec/docker*.tgz

Makefile

@@ -11,6 +11,7 @@ WEAVE_VERSION=git-$(shell git rev-parse --short=12 HEAD)
 WEAVER_EXE=weaver/weaver
 WEAVEDNS_EXE=weavedns/weavedns
 SIGPROXY_EXE=sigproxy/sigproxy
+PROXY_EXE=weaveexec/proxy
 WEAVER_IMAGE=$(DOCKERHUB_USER)/weave
 WEAVEDNS_IMAGE=$(DOCKERHUB_USER)/weavedns
 WEAVEEXEC_IMAGE=$(DOCKERHUB_USER)/weaveexec
@@ -27,9 +28,9 @@ all: $(WEAVER_EXPORT) $(WEAVEDNS_EXPORT) $(WEAVEEXEC_EXPORT)
 travis: $(WEAVER_EXE) $(WEAVEDNS_EXE)
 
 update:
-	go get -u -f -v -tags -netgo ./$(dir $(WEAVER_EXE)) ./$(dir $(WEAVEDNS_EXE)) ./$(dir $(SIGPROXY_EXE))
+	go get -u -f -v -tags -netgo ./$(dir $(WEAVER_EXE)) ./$(dir $(WEAVEDNS_EXE)) ./$(dir $(SIGPROXY_EXE)) ./$(dir $(PROXY_EXE))
 
-$(WEAVER_EXE) $(WEAVEDNS_EXE): common/*.go
+$(WEAVER_EXE) $(WEAVEDNS_EXE) $(PROXY_EXE): common/*.go
 	go get -tags netgo ./$(@D)
 	go build -ldflags "-extldflags \"-static\" -X main.version $(WEAVE_VERSION)" -tags netgo -o $@ ./$(@D)
 	@strings $@ | grep cgo_stub\\\.go >/dev/null || { \
@@ -43,6 +44,7 @@ $(WEAVER_EXE) $(WEAVEDNS_EXE): common/*.go
 
 $(WEAVER_EXE): router/*.go ipam/*.go ipam/*/*.go weaver/main.go
 $(WEAVEDNS_EXE): nameserver/*.go weavedns/main.go
+$(PROXY_EXE): proxy/*.go weaveexec/main.go
 
 # Sigproxy needs separate rule as it fails the netgo check in the main
 # build stanza due to not importing net package
@@ -57,7 +59,7 @@ $(WEAVEDNS_EXPORT): weavedns/Dockerfile $(WEAVEDNS_EXE)
 	$(SUDO) docker build -t $(WEAVEDNS_IMAGE) weavedns
 	$(SUDO) docker save $(WEAVEDNS_IMAGE):latest > $@
 
-$(WEAVEEXEC_EXPORT): weaveexec/Dockerfile $(DOCKER_DISTRIB) weave $(SIGPROXY_EXE)
+$(WEAVEEXEC_EXPORT): weaveexec/Dockerfile $(DOCKER_DISTRIB) weave $(SIGPROXY_EXE) $(PROXY_EXE)
 	cp weave weaveexec/weave
 	cp sigproxy/sigproxy weaveexec/sigproxy
 	cp $(DOCKER_DISTRIB) weaveexec/docker.tgz

proxy/client.go

@@ -0,0 +1,171 @@
+package proxy
+
+import (
+	"bufio"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"net/http/httputil"
+
+	. "github.com/weaveworks/weave/common"
+)
+
+type client struct {
+	interceptor
+	Dial func() (net.Conn, error)
+}
+
+func newClient(dial func() (net.Conn, error), i interceptor) *client {
+	return &client{
+		interceptor: i,
+		Dial:        dial,
+	}
+}
+
+func (c *client) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	req, err := c.InterceptRequest(r)
+	if err == docker.ErrNoSuchImage {
+		http.NotFound(w, r)
+		return
+	}
+	if err != nil {
+		http.Error(w, "Unable to create proxied request", http.StatusInternalServerError)
+		Warning.Print("Error intercepting request: ", err)
+		return
+	}
+
+	conn, err := c.Dial()
+	if err != nil {
+		http.Error(w, "Could not connect to target", http.StatusInternalServerError)
+		Warning.Print(err)
+		return
+	}
+	client := httputil.NewClientConn(conn, nil)
+	defer client.Close()
+
+	resp, err := client.Do(req)
+	if err != nil && err != httputil.ErrPersistEOF {
+		http.Error(w, fmt.Sprintf("Could not make request to target: %s", err.Error()), http.StatusInternalServerError)
+		Warning.Print("Error forwarding request: ", err)
+		return
+	}
+	resp, err = c.InterceptResponse(resp)
+	if err != nil {
+		http.Error(w, "Unable to intercept response", http.StatusInternalServerError)
+		Warning.Print("Error intercepting response: ", err)
+		return
+	}
+
+	hdr := w.Header()
+	for k, vs := range resp.Header {
+		for _, v := range vs {
+			hdr.Add(k, v)
+		}
+	}
+	Debug.Printf("Response from target: %s %v", resp.Status, w.Header())
+
+	if resp.Header.Get("Content-Type") == "application/vnd.docker.raw-stream" {
+		doRawStream(w, resp, client)
+	} else if resp.TransferEncoding != nil && resp.TransferEncoding[0] == "chunked" {
+		doChunkedResponse(w, resp, client)
+	} else {
+		w.WriteHeader(resp.StatusCode)
+		if _, err := io.Copy(w, resp.Body); err != nil {
+			Warning.Print(err)
+		}
+	}
+}
+
+func doRawStream(w http.ResponseWriter, resp *http.Response, client *httputil.ClientConn) {
+	down, downBuf, up, rem, err := hijack(w, client)
+	if err != nil {
+		Error.Fatal(w, "Unable to hijack connection for raw stream mode", http.StatusInternalServerError)
+		return
+	}
+	defer down.Close()
+	defer up.Close()
+
+	if _, err := down.Write([]byte("HTTP/1.1 200 OK\n")); err != nil {
+		Warning.Print(err)
+		return
+	}
+
+	if err := resp.Header.Write(down); err != nil {
+		Warning.Print(err)
+		return
+	}
+
+	if _, err := down.Write([]byte("\n")); err != nil {
+		Warning.Print(err)
+		return
+	}
+
+	end := make(chan bool)
+	go func() {
+		defer close(end)
+		if _, err := io.Copy(up, downBuf); err != nil {
+			Warning.Print(err)
+		}
+		err = up.(interface {
+			CloseWrite() error
+		}).CloseWrite()
+		if err != nil {
+			Debug.Printf("Error Closing upstream: %s", err)
+		}
+	}()
+
+	if _, err := io.Copy(down, io.MultiReader(rem, up)); err != nil {
+		Warning.Print(err)
+	}
+	err = down.(interface {
+		CloseWrite() error
+	}).CloseWrite()
+	if err != nil {
+		Debug.Printf("Error Closing downstream: %s", err)
+	}
+
+	<-end
+}
+
+func doChunkedResponse(w http.ResponseWriter, resp *http.Response, client *httputil.ClientConn) {
+	// Because we can't go back to request/response after we
+	// hijack the connection, we need to close it and make the
+	// client open another.
+	w.Header().Add("Connection", "close")
+	w.WriteHeader(resp.StatusCode)
+
+	down, _, up, rem, err := hijack(w, client)
+	if err != nil {
+		Error.Fatal("Unable to hijack response stream for chunked response", http.StatusInternalServerError)
+		return
+	}
+	defer up.Close()
+	defer down.Close()
+	// Copy the chunked response body to downstream,
+	// stopping at the end of the chunked section.
+	rawResponseBody := io.MultiReader(rem, up)
+	if _, err := io.Copy(ioutil.Discard, httputil.NewChunkedReader(io.TeeReader(rawResponseBody, down))); err != nil {
+		Error.Fatal("Error copying chunked response body", http.StatusInternalServerError)
+		return
+	}
+	resp.Trailer.Write(down)
+	// a chunked response ends with a CRLF
+	down.Write([]byte("\r\n"))
+}
+
+func hijack(w http.ResponseWriter, client *httputil.ClientConn) (down net.Conn, downBuf *bufio.ReadWriter, up net.Conn, rem io.Reader, err error) {
+	hj, ok := w.(http.Hijacker)
+	if !ok {
+		err = errors.New("Unable to cast to Hijack")
+		return
+	}
+	down, downBuf, err = hj.Hijack()
+	if err != nil {
+		return
+	}
+	up, rem = client.Hijack()
+	return
+}

proxy/create_container_interceptor.go

@@ -0,0 +1,88 @@
+package proxy
+
+import (
+	"bytes"
+	"encoding/json"
+	"io/ioutil"
+	"net/http"
+
+	"github.com/fsouza/go-dockerclient"
+)
+
+type createContainerInterceptor struct {
+	client *docker.Client
+}
+
+func CreateContainerInterceptor(client *docker.Client) *createContainerInterceptor {
+	return &createContainerInterceptor{
+		client: client,
+	}
+}
+
+type createContainerRequestBody struct {
+	*docker.Config
+	HostConfig *docker.HostConfig `json:"HostConfig,omitempty" yaml:"HostConfig,omitempty"`
+	MacAddress string             `json:"MacAddress,omitempty" yaml:"MacAddress,omitempty"`
+}
+
+func (i *createContainerInterceptor) InterceptRequest(r *http.Request) (*http.Request, error) {
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		return nil, err
+	}
+	r.Body.Close()
+
+	container := createContainerRequestBody{}
+	if err := json.Unmarshal(body, &container); err != nil {
+		return nil, err
+	}
+
+	if _, ok := weaveAddrFromConfig(container.Config); ok {
+		image, err := i.client.InspectImage(container.Config.Image)
+		if err != nil {
+			return nil, err
+		}
+
+		addToVolumesFrom(container.HostConfig, "weaveproxy")
+		setWeaveWaitEntrypoint(image.Config, container.Config)
+	}
+
+	newBody, err := json.Marshal(container)
+	if err != nil {
+		return nil, err
+	}
+	r.Body = ioutil.NopCloser(bytes.NewReader(newBody))
+	r.ContentLength = int64(len(newBody))
+
+	return r, nil
+}
+
+func addToVolumesFrom(config *docker.HostConfig, mounts ...string) error {
+	config.VolumesFrom = append(config.VolumesFrom, mounts...)
+	return nil
+}
+
+func setWeaveWaitEntrypoint(image, container *docker.Config) {
+	container.Cmd = combineEntrypoints(image, container)
+	container.Entrypoint = []string{"/home/weavewait/wait-for-weave"}
+}
+
+func combineEntrypoints(image, container *docker.Config) []string {
+	var entry, command []string
+	if len(container.Entrypoint) > 0 {
+		entry = container.Entrypoint
+	} else if len(image.Entrypoint) > 0 {
+		entry = image.Entrypoint
+	}
+	if len(container.Cmd) > 0 {
+		command = container.Cmd
+	} else if len(image.Cmd) > 0 {
+		command = image.Cmd
+	}
+
+	return append(entry, command...)
+}
+
+func (i *createContainerInterceptor) InterceptResponse(res *http.Response) (*http.Response, error) {
+	return res, nil
+}

proxy/interceptor.go

@@ -0,0 +1,19 @@
+package proxy
+
+import "net/http"
+
+type interceptor interface {
+	InterceptRequest(*http.Request) (*http.Request, error)
+	InterceptResponse(*http.Response) (*http.Response, error)
+}
+
+type nullInterceptor struct {
+}
+
+func (i nullInterceptor) InterceptRequest(r *http.Request) (*http.Request, error) {
+	return r, nil
+}
+
+func (i nullInterceptor) InterceptResponse(r *http.Response) (*http.Response, error) {
+	return r, nil
+}

proxy/proxy.go

@@ -0,0 +1,96 @@
+package proxy
+
+import (
+	"net"
+	"net/http"
+	"net/url"
+	"regexp"
+	"strings"
+
+	"github.com/fsouza/go-dockerclient"
+	. "github.com/weaveworks/weave/common"
+)
+
+const (
+	RAW_STREAM = "application/vnd.docker.raw-stream"
+)
+
+type proxy struct {
+	Dial   func() (net.Conn, error)
+	client *docker.Client
+}
+
+func targetNetwork(u *url.URL) string {
+	return u.Scheme
+}
+
+func targetAddress(u *url.URL) (addr string) {
+	switch u.Scheme {
+	case "tcp":
+		addr = u.Host
+	case "unix":
+		addr = u.Path
+	}
+	return
+}
+
+func NewProxy(targetUrl string) (*proxy, error) {
+	u, err := url.Parse(targetUrl)
+	if err != nil {
+		return nil, err
+	}
+	client, err := docker.NewClient(targetUrl)
+	if err != nil {
+		return nil, err
+	}
+	return &proxy{
+		Dial: func() (net.Conn, error) {
+			return net.Dial(targetNetwork(u), targetAddress(u))
+		},
+		client: client,
+	}, nil
+}
+
+func isWeaveStatus(r *http.Request) bool {
+	return strings.HasPrefix(r.URL.Path, "/weave")
+}
+
+func isCreateContainer(r *http.Request) bool {
+	ok, err := regexp.MatchString("/v[0-9\\.]*/containers/create", r.URL.Path)
+	return err == nil && ok
+}
+
+func isStartContainer(r *http.Request) bool {
+	ok, err := regexp.MatchString("^/v[0-9\\.]*/containers/[^/]*/start$", r.URL.Path)
+	return err == nil && ok
+}
+
+func (proxy *proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	Info.Printf("%s %s", r.Method, r.URL)
+	switch {
+	case isCreateContainer(r):
+		proxy.CreateContainer(w, r)
+	case isStartContainer(r):
+		proxy.StartContainer(w, r)
+	case isWeaveStatus(r):
+		proxy.WeaveStatus(w, r)
+	default:
+		proxy.ProxyRequest(w, r)
+	}
+}
+
+func (proxy *proxy) WeaveStatus(w http.ResponseWriter, r *http.Request) {
+	w.WriteHeader(http.StatusOK)
+}
+
+func (proxy *proxy) CreateContainer(w http.ResponseWriter, r *http.Request) {
+	NewClient(proxy.Dial, CreateContainerInterceptor(proxy.client)).ServeHTTP(w, r)
+}
+
+func (proxy *proxy) StartContainer(w http.ResponseWriter, r *http.Request) {
+	NewClient(proxy.Dial, StartContainerInterceptor(proxy.client)).ServeHTTP(w, r)
+}
+
+func (proxy *proxy) ProxyRequest(w http.ResponseWriter, r *http.Request) {
+	NewClient(proxy.Dial, NullInterceptor).ServeHTTP(w, r)
+}

proxy/start_container_interceptor.go

@@ -0,0 +1,75 @@
+package proxy
+
+import (
+	"net/http"
+	"os/exec"
+	"regexp"
+	"strings"
+
+	"github.com/fsouza/go-dockerclient"
+	. "github.com/weaveworks/weave/common"
+)
+
+type startContainerInterceptor struct {
+	client *docker.Client
+}
+
+func StartContainerInterceptor(client *docker.Client) Interceptor {
+	return &startContainerInterceptor{
+		client: client,
+	}
+}
+
+func (i *startContainerInterceptor) InterceptRequest(r *http.Request) (*http.Request, error) {
+	return r, nil
+}
+
+func (i *startContainerInterceptor) InterceptResponse(res *http.Response) (*http.Response, error) {
+	containerId := containerFromPath(res.Request.URL.Path)
+
+	container, err := i.client.InspectContainer(containerId)
+	if err != nil {
+		Warning.Print("Error Inspecting Container: ", err)
+		return res, nil
+	}
+
+	if cidr, ok := weaveAddrFromConfig(container.Config); ok {
+		Info.Printf("Container %s was started with CIDR \"%s\"", containerId, cidr)
+		args := []string{"attach"}
+		args = append(args, strings.Split(strings.TrimSpace(cidr), " ")...)
+		args = append(args, containerId)
+		if out, err := callWeave(args...); err != nil {
+			Warning.Print("Calling weave failed: ", err, string(out))
+		}
+	} else {
+		Debug.Print("No Weave CIDR, ignoring")
+	}
+
+	return res, nil
+}
+
+func containerFromPath(path string) string {
+	if subs := regexp.MustCompile("^/v[0-9\\.]*/containers/([^/]*)/.*").FindStringSubmatch(path); subs != nil {
+		return subs[1]
+	}
+	return ""
+}
+
+func callWeave(args ...string) ([]byte, error) {
+	args = append([]string{"--local"}, args...)
+	Debug.Print("Calling weave", args)
+	cmd := exec.Command("./weave", args...)
+	cmd.Env = []string{"PROCFS=/hostproc", "PATH=/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"}
+	out, err := cmd.CombinedOutput()
+	return out, err
+}
+
+func weaveAddrFromConfig(config *docker.Config) (string, bool) {
+	for _, e := range config.Env {
+		if strings.HasPrefix(e, "WEAVE_CIDR=") {
+			result := strings.Trim(e[11:], " ")
+			return result, result != ""
+		}
+	}
+	return "", false
+}

weaveexec/Dockerfile

@@ -15,6 +15,10 @@ WORKDIR /home/weave
 
 ADD ./weave /home/weave/
 ADD ./sigproxy /home/weave/
+ADD ./proxy /home/weave/
 ADD ./docker.tgz /
 
+ADD ./wait-for-weave /home/weavewait/wait-for-weave
+VOLUME /home/weavewait
+
 ENTRYPOINT ["/home/weave/sigproxy", "/home/weave/weave"]

weaveexec/main.go

@@ -0,0 +1,39 @@
+package main
+
+import (
+	"flag"
+	"net/http"
+
+	. "github.com/weaveworks/weave/common"
+	"github.com/weaveworks/weave/proxy"
+)
+
+func main() {
+	var target, listen string
+	var debug bool
+
+	flag.StringVar(&target, "H", "unix:///var/run/docker.sock", "docker daemon URL to proxy")
+	flag.StringVar(&listen, "L", ":12375", "address on which to listen")
+	flag.BoolVar(&debug, "debug", false, "log debugging information")
+	flag.Parse()
+
+	if debug {
+		InitDefaultLogging(true)
+	}
+
+	p, err := proxy.NewProxy(target)
+	if err != nil {
+		Error.Fatalf("Could not start proxy: %s", err)
+	}
+	s := &http.Server{
+		Addr:    listen,
+		Handler: p,
+	}
+
+	Info.Printf("Listening on %s", listen)
+	Info.Printf("Proxying %s", target)
+
+	if err := s.ListenAndServe(); err != nil {
+		Error.Fatalf("Could not listen on %s: %s", listen, err)
+	}
+}