simplify away http proxy; fix forever-deferred body close bug

Author: tylerharter

src/lambda/lambdaInstance.go

@@ -4,9 +4,7 @@ import (
 	"io"
 	"log"
 	"net/http"
-	"net/http/httputil"
 	"strings"
-	"time"
 
 	"github.com/open-lambda/open-lambda/ol/common"
 	"github.com/open-lambda/open-lambda/ol/sandbox"
@@ -40,8 +38,6 @@ func (linst *LambdaInstance) Task() {
 	f := linst.lfunc
 
 	var sb sandbox.Sandbox = nil
-	//var client *http.Client = nil // whenever we create a Sandbox, we init this too
-	var proxy *httputil.ReverseProxy = nil // whenever we create a Sandbox, we init this too
 	var err error
 
 	for {
@@ -129,27 +125,11 @@ func (linst *LambdaInstance) Task() {
 				f.doneChan <- req
 				continue // wait for another request before retrying
 			}
-
-			log.Printf("Connecting to sandbox")
-
-			proxy, err = sb.HTTPProxy()
-			if err != nil {
-				linst.TrySendError(req, http.StatusBadGateway, "could not connect to Sandbox: "+err.Error()+"\n", sb)
-				f.doneChan <- req
-				f.printf("discard sandbox %s due to Channel error: %v", sb.ID(), err)
-				sb = nil
-				continue // wait for another request before retrying
-			}
 		}
 		t.T1()
 
 		// below here, we're guaranteed (1) sb != nil, (2) proxy != nil, (3) sb is unpaused
 
-		client := http.Client{
-			Transport: proxy.Transport,
-			Timeout: time.Second * time.Duration(common.Conf.Limits.Max_runtime_default),
-		}
-
 		// serve until we incoming queue is empty
 		t = common.T0("LambdaInstance-ServeRequests")
 		for req != nil {
@@ -163,7 +143,7 @@ func (linst *LambdaInstance) Task() {
 			if err != nil {
 				linst.TrySendError(req, http.StatusInternalServerError, "Could not create NewRequest: "+err.Error(), sb)
 			} else {
-				resp, err := client.Do(httpReq)
+				resp, err := sb.Client().Do(httpReq)
 
 				// copy response out
 				if err != nil {
@@ -179,13 +159,14 @@ func (linst *LambdaInstance) Task() {
 					req.w.WriteHeader(resp.StatusCode)
 
 					// copy body
-					defer resp.Body.Close()
 					if _, err := io.Copy(req.w, resp.Body); err != nil {
 						// already used WriteHeader, so can't use that to surface on error anymore
 						msg := "reading lambda response failed: "+err.Error()+"\n"
 						f.printf("error: "+msg)
 						linst.TrySendError(req, 0, msg, sb)
 					}
+
+					resp.Body.Close()
 				}
 			}
 

src/lambda/packagePuller.go

@@ -275,11 +275,6 @@ func (pp *PackagePuller) sandboxInstall(p *Package) (err error) {
 	}
 	defer sb.Destroy("package installation complete")
 
-	proxy, err := sb.HTTPProxy()
-	if err != nil {
-		return err
-	}
-
 	// we still need to run a Sandbox to parse the dependencies, even if it is already installed
 	msg := fmt.Sprintf(`{"pkg": "%s", "alreadyInstalled": %v}`, p.name, alreadyInstalled)
 	reqBody := bytes.NewReader([]byte(msg))
@@ -289,7 +284,7 @@ func (pp *PackagePuller) sandboxInstall(p *Package) (err error) {
 	if err != nil {
 		return err
 	}
-	resp, err := proxy.Transport.RoundTrip(req)
+	resp, err := sb.Client().Do(req)
 	if err != nil {
 		return err
 	}

src/sandbox/api.go

@@ -2,7 +2,7 @@ package sandbox
 
 import (
 	"github.com/open-lambda/open-lambda/ol/common"
-	"net/http/httputil"
+	"net/http"
 )
 
 type SandboxPool interface {
@@ -60,7 +60,7 @@ type Sandbox interface {
 	Unpause() error
 
 	// Communication channel to forward requests.
-	HTTPProxy() (*httputil.ReverseProxy, error)
+	Client() (*http.Client)
 
 	// Lookup metadata that Sandbox was initialized with (static over time)
 	Meta() *SandboxMeta
@@ -81,7 +81,7 @@ type Sandbox interface {
 	// Child calls this on parent to notify of child Destroy
 	childExit(child Sandbox)
 
-	GetRuntimeType() common.RuntimeType
+	GetRuntimeType() common.RuntimeType // TODO: make it part of SandboxMeta?
 }
 
 type SandboxMeta struct {

src/sandbox/docker.go

@@ -15,11 +15,8 @@ import (
 	"fmt"
 	"io/ioutil"
 	"log"
-	"net"
-	"net/http"
-	"net/http/httputil"
-	"net/url"
 	"os"
+	"net/http"
 	"path/filepath"
 	"time"
 
@@ -37,6 +34,7 @@ type DockerContainer struct {
 	installed map[string]bool
 	meta	  *SandboxMeta
 	rtType   common.RuntimeType
+	httpClient *http.Client
 }
 
 type HandlerState int
@@ -111,26 +109,8 @@ func (c *DockerContainer) State() (hstate HandlerState, err error) {
 	return hstate, nil
 }
 
-// HTTPProxy returns a file socket channel for direct communication with the sandbox.
-func (c *DockerContainer) HTTPProxy() (p *httputil.ReverseProxy, err error) {
-	sockPath := filepath.Join(c.hostDir, "ol.sock")
-	if len(sockPath) > 108 {
-		return nil, fmt.Errorf("socket path length cannot exceed 108 characters (try moving cluster closer to the root directory")
-	}
-
-	dial := func(proto, addr string) (net.Conn, error) {
-		return net.Dial("unix", sockPath)
-	}
-
-	tr := &http.Transport{Dial: dial}
-	u, err := url.Parse("http://sock-container")
-	if err != nil {
-		panic(err)
-	}
-
-	proxy := httputil.NewSingleHostReverseProxy(u)
-	proxy.Transport = tr
-	return proxy, nil
+func (c *DockerContainer) Client() (*http.Client) {
+	return c.httpClient
 }
 
 // Start starts the container.
@@ -165,6 +145,9 @@ func (c *DockerContainer) Pause() error {
 		return c.dockerError(err)
 	}
 
+	// idle connections use a LOT of memory in the OL process
+	c.httpClient.CloseIdleConnections()
+
 	return nil
 }
 

src/sandbox/dockerPool.go

@@ -6,6 +6,9 @@ import (
 	"strings"
 	"sync/atomic"
 	"syscall"
+	"net"
+	"net/http"
+	"time"
 
 	docker "github.com/fsouza/go-dockerclient"
 
@@ -135,6 +138,21 @@ func (pool *DockerPool) Create(parent Sandbox, isLeaf bool, codeDir, scratchDir
 		return nil, err
 	}
 
+	// start HTTP client
+	sockPath := filepath.Join(c.hostDir, "ol.sock")
+	if len(sockPath) > 108 {
+		return nil, fmt.Errorf("socket path length cannot exceed 108 characters (try moving cluster closer to the root directory")
+	}
+
+	dial := func(proto, addr string) (net.Conn, error) {
+		return net.Dial("unix", sockPath)
+	}
+
+	c.httpClient = &http.Client{
+		Transport: &http.Transport{Dial: dial},
+		Timeout: time.Second * time.Duration(common.Conf.Limits.Max_runtime_default),
+	}
+
 	// wrap to make thread-safe and handle container death
 	safe := newSafeSandbox(c)
 	safe.startNotifyingListeners(pool.eventHandlers)

src/sandbox/safeSandbox.go

@@ -10,7 +10,7 @@ package sandbox
 import (
 	"fmt"
 	"log"
-	"net/http/httputil"
+	"net/http"
 	"strings"
 	"sync"
 
@@ -167,22 +167,16 @@ func (sb *safeSandbox) Unpause() (err error) {
 	return nil
 }
 
-func (sb *safeSandbox) HttpProxy() (p *httputil.ReverseProxy, err error) {
-	sb.printf("Channel()")
-	t := common.T0("Channel()")
-	defer t.T1()
-	sb.Mutex.Lock()
-	defer sb.Mutex.Unlock()
-
-	if sb.dead != nil {
-		return nil, sb.dead
-	}
-
-	p, err = sb.Sandbox.HTTPProxy()
-	if err != nil {
-		sb.destroyOnErr("HttpProxy", err) // TODO: rename if/when HttpProxy is replaced
-	}
-	return p, err
+func (sb *safeSandbox) Client() (*http.Client) {
+	// According to the docs, "Clients and Transports are safe for
+	// concurrent use by multiple goroutines and for efficiency
+	// should only be created once and re-used."
+	//
+	// So we don't use any locking around this one.  This also
+	// can't fail because the client should have been created
+	// during sandbox initialization (and if there were any error,
+	// it should have occured then).
+	return sb.Sandbox.Client()
 }
 
 // fork (as a private method) doesn't cleanup parent sb if fork fails

src/sandbox/sock.go

@@ -5,10 +5,7 @@ import (
 	"io/ioutil"
 	"log"
 	"sync/atomic"
-	"net"
 	"net/http"
-	"net/http/httputil"
-	"net/url"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -29,6 +26,7 @@ type SOCKContainer struct {
 	scratchDir       string
 	cg               *Cgroup
 	rtType           common.RuntimeType
+	client *http.Client
 
 	// 1 for self, plus 1 for each child (we can't release memory
 	// until all descendants are dead, because they share the
@@ -58,32 +56,6 @@ func (c *SOCKContainer) GetRuntimeType() common.RuntimeType {
 	return c.rtType
 }
 
-func (container *SOCKContainer) HTTPProxy() (p *httputil.ReverseProxy, err error) {
-	// note, for debugging, you can directly contact the sock file like this:
-	// curl -XPOST --unix-socket ./ol.sock http:/test -d '{"some": "data"}'
-
-	sockPath := filepath.Join(container.scratchDir, "ol.sock")
-	if len(sockPath) > 108 {
-		return nil, fmt.Errorf("socket path length cannot exceed 108 characters (try moving cluster closer to the root directory")
-	}
-
-	log.Printf("Connecting to container at '%s'", sockPath)
-
-	dial := func(proto, addr string) (net.Conn, error) {
-		return net.Dial("unix", sockPath)
-	}
-
-	tr := &http.Transport{Dial: dial}
-	u, err := url.Parse("http://sock-container")
-	if err != nil {
-		panic(err)
-	}
-
-	proxy := httputil.NewSingleHostReverseProxy(u)
-	proxy.Transport = tr
-	return proxy, nil
-}
-
 func (container *SOCKContainer) freshProc() (err error) {
 	// get FD to cgroup
 	cgFiles := make([]*os.File, 1)
@@ -267,6 +239,9 @@ func (container *SOCKContainer) Pause() (err error) {
 		}
 	}
 
+	// save a little memory
+	container.client.CloseIdleConnections()
+
 	return nil
 }
 
@@ -436,6 +411,10 @@ func (container *SOCKContainer) Meta() *SandboxMeta {
 	return container.meta
 }
 
+func (c *SOCKContainer) Client() (*http.Client) {
+	return c.client
+}
+
 // GetRuntimeLog returns the log of the runtime
 func (container *SOCKContainer) GetRuntimeLog() string {
 	data, err := ioutil.ReadFile(filepath.Join(container.scratchDir, "ol-runtime.log"))

src/sandbox/sockPool.go

@@ -7,6 +7,9 @@ import (
 	"path/filepath"
 	"strings"
 	"sync/atomic"
+	"net"
+	"net/http"
+	"time"
 
 	"github.com/open-lambda/open-lambda/ol/common"
 )
@@ -170,6 +173,23 @@ func (pool *SOCKPool) Create(parent Sandbox, isLeaf bool, codeDir, scratchDir st
 		t2.T1()
 	}
 
+	// start HTTP client
+	sockPath := filepath.Join(cSock.scratchDir, "ol.sock")
+	if len(sockPath) > 108 {
+		return nil, fmt.Errorf("socket path length cannot exceed 108 characters (try moving cluster closer to the root directory")
+	}
+
+	log.Printf("Connecting to container at '%s'", sockPath)
+	dial := func(proto, addr string) (net.Conn, error) {
+		return net.Dial("unix", sockPath)
+	}
+
+	cSock.client = &http.Client{
+		Transport: &http.Transport{Dial: dial},
+		Timeout: time.Second * time.Duration(common.Conf.Limits.Max_runtime_default),
+	}
+
+	// event handling
 	safe.startNotifyingListeners(pool.eventHandlers)
 	return c, nil
 }