diff --git a/.gitignore b/.gitignore index 94b5d700..e685a1b6 100644 --- a/.gitignore +++ b/.gitignore @@ -39,6 +39,7 @@ releases weave.tar weavedns.tar weaveexec.tar +weaveexec/proxy weaveexec/weave weaveexec/sigproxy weaveexec/docker*.tgz diff --git a/Makefile b/Makefile index 601e6430..36fef509 100644 --- a/Makefile +++ b/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 diff --git a/proxy/client.go b/proxy/client.go new file mode 100644 index 00000000..adfac43e --- /dev/null +++ b/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 +} diff --git a/proxy/create_container_interceptor.go b/proxy/create_container_interceptor.go new file mode 100644 index 00000000..c96431c4 --- /dev/null +++ b/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 +} diff --git a/proxy/interceptor.go b/proxy/interceptor.go new file mode 100644 index 00000000..203a0d03 --- /dev/null +++ b/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 +} diff --git a/proxy/proxy.go b/proxy/proxy.go new file mode 100644 index 00000000..d36e6347 --- /dev/null +++ b/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) +} diff --git a/proxy/start_container_interceptor.go b/proxy/start_container_interceptor.go new file mode 100644 index 00000000..5cf20fe6 --- /dev/null +++ b/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 +} diff --git a/weaveexec/Dockerfile b/weaveexec/Dockerfile index e7134926..6295b8a6 100644 --- a/weaveexec/Dockerfile +++ b/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"] diff --git a/weaveexec/main.go b/weaveexec/main.go new file mode 100644 index 00000000..548aafb9 --- /dev/null +++ b/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) + } +}