Initial Refactor (#1234)

* Inital Refactor

Removing the repeated logic exposed some problems with the reponse
writers.

Currently, the trigger writer was overlaid on part of the header
writing. The main invoke blog writing into the different levels of the
overlays at different points in the logic.

Instead, by extending the types and embedded structs, the writer is
more transparent. So, at the end of the flow it goes over all the
headers available and removes our prefixes. This lets the invoke logic
just write to the top level.

Going to continue after lunch to try and remove some of the layers and
param passing.

* Try and repeat concurrency failure

* Nested FromHTTPFnRequest inside FromHTTPTriggerRequest

* Consolidate buffer pooling logic

* go fmt yourself

* fix import

api/agent/call.go

@@ -67,7 +67,6 @@ var skipTriggerHeaders = map[string]bool{
 // Sets up a call from an http trigger request
 func FromHTTPTriggerRequest(app *models.App, fn *models.Fn, trigger *models.Trigger, req *http.Request) CallOpt {
 	return func(c *call) error {
-		contentType := req.Header.Get("Content-Type")
 		// transpose trigger headers into HTTP
 		headers := make(http.Header)
 		for k, vs := range req.Header {
@@ -84,61 +83,13 @@ func FromHTTPTriggerRequest(app *models.App, fn *models.Fn, trigger *models.Trig
 		requestUrl := reqURL(req)
 
 		headers.Set("Fn-Http-Method", req.Method)
-		if contentType != "" {
-			headers.Set("Content-Type", contentType)
-		}
 		headers.Set("Fn-Http-Request-Url", requestUrl)
 		headers.Set("Fn-Intent", "httprequest")
 		req.Header = headers
 
-		if fn.Format == "" {
+		err := FromHTTPFnRequest(app, fn, req)(c)
-			fn.Format = models.FormatDefault
+		c.Model().TriggerID = trigger.ID
-		}
+		return err
-
-		id := id.New().String()
-
-		// TODO this relies on ordering of opts, but tests make sure it works, probably re-plumb/destroy headers
-		// TODO async should probably supply an http.ResponseWriter that records the logs, to attach response headers to
-		if rw, ok := c.w.(http.ResponseWriter); ok {
-			// TODO deprecate after CLI is updated
-			rw.Header().Add("Fn-Call-ID", id)
-			rw.Header().Add("FN_CALL_ID", id)
-		}
-
-		var syslogURL string
-		if app.SyslogURL != nil {
-			syslogURL = *app.SyslogURL
-		}
-
-		c.Call = &models.Call{
-			ID:    id,
-			Image: fn.Image,
-			// Delay: 0,
-			Type:   "sync",
-			Format: fn.Format,
-			// Payload: TODO,
-			Priority:    new(int32), // TODO this is crucial, apparently
-			Timeout:     fn.Timeout,
-			IdleTimeout: fn.IdleTimeout,
-			TmpFsSize:   0, // TODO clean up this
-			Memory:      fn.Memory,
-			CPUs:        0, // TODO clean up this
-			Config:      buildConfig(app, fn, trigger.Source),
-			// TODO - this wasn't really the intention here (that annotations would naturally cascade
-			// but seems to be necessary for some runner behaviour
-			Annotations: app.Annotations.MergeChange(fn.Annotations).MergeChange(trigger.Annotations),
-			Headers:     req.Header,
-			CreatedAt:   common.DateTime(time.Now()),
-			URL:         requestUrl,
-			Method:      req.Method,
-			AppID:       app.ID,
-			AppName:     app.Name,
-			FnID:        fn.ID,
-			TriggerID:   trigger.ID,
-			SyslogURL:   syslogURL,
-		}
-		c.req = req
-		return nil
 	}
 }
 

api/server/runner_fninvoke.go

@@ -20,7 +20,15 @@ var (
 	bufPool = &sync.Pool{New: func() interface{} { return new(bytes.Buffer) }}
 )
 
-var _ http.ResponseWriter = new(syncResponseWriter)
+type ResponseBufferingWriter interface {
+	http.ResponseWriter
+	io.Reader
+	Status() int
+	GetBuffer() *bytes.Buffer
+	SetBuffer(*bytes.Buffer)
+}
+
+var _ ResponseBufferingWriter = new(syncResponseWriter)
 
 // implements http.ResponseWriter
 // this little guy buffers responses from user containers and lets them still
@@ -33,8 +41,13 @@ type syncResponseWriter struct {
 	*bytes.Buffer
 }
 
-func (s *syncResponseWriter) Header() http.Header  { return s.headers }
+func (s *syncResponseWriter) Header() http.Header { return s.headers }
-func (s *syncResponseWriter) WriteHeader(code int) { s.status = code }
+
+// By storing the status here, we effectively buffer the response
+func (s *syncResponseWriter) WriteHeader(code int)        { s.status = code }
+func (s *syncResponseWriter) Status() int                 { return s.status }
+func (s *syncResponseWriter) GetBuffer() *bytes.Buffer    { return s.Buffer }
+func (s *syncResponseWriter) SetBuffer(buf *bytes.Buffer) { s.Buffer = buf }
 
 // handleFnInvokeCall executes the function, for router handlers
 func (s *Server) handleFnInvokeCall(c *gin.Context) {
@@ -66,63 +79,71 @@ func (s *Server) handleFnInvokeCall2(c *gin.Context) error {
 }
 
 func (s *Server) ServeFnInvoke(c *gin.Context, app *models.App, fn *models.Fn) error {
-	// TODO: we should combine this logic with trigger, which just wraps this block with some headers wizardry
+	writer := &syncResponseWriter{
-	// TODO: we should get rid of the buffers, and stream back (saves memory (+splice), faster (splice), allows streaming, don't have to cap resp size)
-	buf := bufPool.Get().(*bytes.Buffer)
-	buf.Reset()
-	writer := syncResponseWriter{
-		Buffer:  buf,
 		headers: c.Writer.Header(),
 	}
-	defer bufPool.Put(buf) // TODO need to ensure this is safe with Dispatch?
 
-	call, err := s.agent.GetCall(
+	call, err := s.agent.GetCall(agent.WithWriter(writer), // XXX (reed): order matters [for now]
-		agent.WithWriter(&writer), // XXX (reed): order matters [for now]
+		agent.FromHTTPFnRequest(app, fn, c.Request))
-		agent.FromHTTPFnRequest(app, fn, c.Request),
+
-	)
 	if err != nil {
 		return err
 	}
 
+	return s.fnInvoke(c, app, fn, writer, call)
+}
+
+func (s *Server) fnInvoke(c *gin.Context, app *models.App, fn *models.Fn, writer ResponseBufferingWriter, call agent.Call) error {
+	// TODO: we should get rid of the buffers, and stream back (saves memory (+splice), faster (splice), allows streaming, don't have to cap resp size)
+	buf := bufPool.Get().(*bytes.Buffer)
+	buf.Reset()
+	var submitErr error
+	defer func() {
+		if buf.Len() == 0 && submitErr == nil {
+			bufPool.Put(buf) // TODO need to ensure this is safe with Dispatch?
+		}
+	}()
+	writer.SetBuffer(buf)
+
 	model := call.Model()
 	{ // scope this, to disallow ctx use outside of this scope. add id for handleV1ErrorResponse logger
 		ctx, _ := common.LoggerWithFields(c.Request.Context(), logrus.Fields{"id": model.ID})
 		c.Request = c.Request.WithContext(ctx)
 	}
 
-	err = s.agent.Submit(call)
+	submitErr = s.agent.Submit(call)
-	if err != nil {
+	if submitErr != nil {
 		// NOTE if they cancel the request then it will stop the call (kind of cool),
 		// we could filter that error out here too as right now it yells a little
-		if err == models.ErrCallTimeoutServerBusy || err == models.ErrCallTimeout {
+		if submitErr == models.ErrCallTimeoutServerBusy || submitErr == models.ErrCallTimeout {
 			// TODO maneuver
 			// add this, since it means that start may not have been called [and it's relevant]
 			c.Writer.Header().Add("XXX-FXLB-WAIT", time.Now().Sub(time.Time(model.CreatedAt)).String())
 		}
-		return err
+		return submitErr
 	}
-
 	// if they don't set a content-type - detect it
 	// TODO: remove this after removing all the formats (too many tests to scrub til then)
 	if writer.Header().Get("Content-Type") == "" {
 		// see http.DetectContentType, the go server is supposed to do this for us but doesn't appear to?
 		var contentType string
 		jsonPrefix := [1]byte{'{'} // stack allocated
-		if bytes.HasPrefix(buf.Bytes(), jsonPrefix[:]) {
+		if bytes.HasPrefix(writer.GetBuffer().Bytes(), jsonPrefix[:]) {
 			// try to detect json, since DetectContentType isn't a hipster.
 			contentType = "application/json; charset=utf-8"
 		} else {
-			contentType = http.DetectContentType(buf.Bytes())
+			contentType = http.DetectContentType(writer.GetBuffer().Bytes())
 		}
 		writer.Header().Set("Content-Type", contentType)
 	}
 
-	writer.Header().Set("Content-Length", strconv.Itoa(int(buf.Len())))
+	writer.Header().Set("Content-Length", strconv.Itoa(int(writer.GetBuffer().Len())))
 
-	if writer.status > 0 {
+	if writer.Status() > 0 {
-		c.Writer.WriteHeader(writer.status)
+		c.Writer.WriteHeader(writer.Status())
 	}
-	io.Copy(c.Writer, &writer)
+
+	io.Copy(c.Writer, writer)
 
 	return nil
 }

api/server/runner_httptrigger.go

@@ -1,20 +1,15 @@
 package server
 
 import (
-	"bytes"
-	"io"
 	"net/http"
 	"strconv"
-	"time"
 
 	"strings"
 
 	"github.com/fnproject/fn/api"
 	"github.com/fnproject/fn/api/agent"
-	"github.com/fnproject/fn/api/common"
 	"github.com/fnproject/fn/api/models"
 	"github.com/gin-gonic/gin"
-	"github.com/sirupsen/logrus"
 )
 
 // handleHTTPTriggerCall executes the function, for router handlers
@@ -65,12 +60,11 @@ func (s *Server) handleTriggerHTTPFunctionCall2(c *gin.Context) error {
 }
 
 type triggerResponseWriter struct {
-	w         http.ResponseWriter
+	syncResponseWriter
-	headers   http.Header
 	committed bool
 }
 
-var _ http.ResponseWriter = new(triggerResponseWriter)
+var _ ResponseBufferingWriter = new(triggerResponseWriter)
 
 func (trw *triggerResponseWriter) Header() http.Header {
 	return trw.headers
@@ -80,7 +74,7 @@ func (trw *triggerResponseWriter) Write(b []byte) (int, error) {
 	if !trw.committed {
 		trw.WriteHeader(http.StatusOK)
 	}
-	return trw.w.Write(b)
+	return trw.GetBuffer().Write(b)
 }
 
 func (trw *triggerResponseWriter) WriteHeader(statusCode int) {
@@ -88,13 +82,27 @@ func (trw *triggerResponseWriter) WriteHeader(statusCode int) {
 		return
 	}
 	trw.committed = true
+
+	for k, vs := range trw.Header() {
+		if strings.HasPrefix(k, "Fn-Http-H-") {
+			// TODO strip out content-length and stuff here.
+			realHeader := strings.TrimPrefix(k, "Fn-Http-H-")
+			if realHeader != "" { // case where header is exactly the prefix
+				for _, v := range vs {
+					trw.Header().Del(k)
+					trw.Header().Add(realHeader, v)
+				}
+			}
+		}
+	}
+
 	gatewayStatus := 200
 
 	if statusCode >= 400 {
 		gatewayStatus = 502
 	}
 
-	status := trw.headers.Get("Fn-Http-Status")
+	status := trw.Header().Get("Fn-Http-Status")
 	if status != "" {
 		statusInt, err := strconv.Atoi(status)
 		if err == nil {
@@ -102,39 +110,16 @@ func (trw *triggerResponseWriter) WriteHeader(statusCode int) {
 		}
 	}
 
-	for k, vs := range trw.headers {
+	trw.WriteHeader(gatewayStatus)
-		if strings.HasPrefix(k, "Fn-Http-H-") {
-			// TODO strip out content-length and stuff here.
-			realHeader := strings.TrimPrefix(k, "Fn-Http-H-")
-			if realHeader != "" { // case where header is exactly the prefix
-				for _, v := range vs {
-					trw.w.Header().Add(realHeader, v)
-				}
-			}
-		}
-	}
-
-	contentType := trw.headers.Get("Content-Type")
-	if contentType != "" {
-		trw.w.Header().Add("Content-Type", contentType)
-	}
-	trw.w.WriteHeader(gatewayStatus)
 }
 
 //ServeHTTPTr	igger serves an HTTP trigger for a given app/fn/trigger  based on the current request
 // This is exported to allow extensions to handle their own trigger naming and publishing
 func (s *Server) ServeHTTPTrigger(c *gin.Context, app *models.App, fn *models.Fn, trigger *models.Trigger) error {
-	buf := bufPool.Get().(*bytes.Buffer)
-	buf.Reset()
-	writer := &syncResponseWriter{
-		Buffer:  buf,
-		headers: c.Writer.Header(), // copy ref
-	}
-	defer bufPool.Put(buf) // TODO need to ensure this is safe with Dispatch?
-
 	triggerWriter := &triggerResponseWriter{
-		w:       writer,
+		syncResponseWriter{
-		headers: make(http.Header),
+			headers: c.Writer.Header()},
+		false,
 	}
 	// GetCall can mod headers, assign an id, look up the route/app (cached),
 	// strip params, etc.
@@ -142,75 +127,12 @@ func (s *Server) ServeHTTPTrigger(c *gin.Context, app *models.App, fn *models.Fn
 
 	// GetCall can mod headers, assign an id, look up the route/app (cached),
 	// strip params, etc.
-	call, err := s.agent.GetCall(
+
-		agent.WithWriter(triggerWriter), // XXX (reed): order matters [for now]
+	call, err := s.agent.GetCall(agent.WithWriter(triggerWriter), // XXX (reed): order matters [for now]
-		agent.FromHTTPTriggerRequest(app, fn, trigger, c.Request),
+		agent.FromHTTPTriggerRequest(app, fn, trigger, c.Request))
-	)
 
 	if err != nil {
 		return err
 	}
-	model := call.Model()
+	return s.fnInvoke(c, app, fn, triggerWriter, call)
-	{ // scope this, to disallow ctx use outside of this scope. add id for handleV1ErrorResponse logger
-		ctx, _ := common.LoggerWithFields(c.Request.Context(), logrus.Fields{"id": model.ID})
-		c.Request = c.Request.WithContext(ctx)
-	}
-	writer.Header().Add("Fn_call_id", model.ID)
-
-	// TODO TRIGGERWIP  not clear this makes sense here - but it works  so...
-	if model.Type == "async" {
-
-		// TODO we should push this into GetCall somehow (CallOpt maybe) or maybe agent.Queue(Call) ?
-		if c.Request.ContentLength > 0 {
-			buf.Grow(int(c.Request.ContentLength))
-		}
-		_, err := buf.ReadFrom(c.Request.Body)
-		if err != nil {
-			return models.ErrInvalidPayload
-		}
-		model.Payload = buf.String()
-
-		err = s.lbEnqueue.Enqueue(c.Request.Context(), model)
-		if err != nil {
-			return err
-		}
-
-		c.JSON(http.StatusAccepted, map[string]string{"call_id": model.ID})
-		return nil
-	}
-
-	err = s.agent.Submit(call)
-	if err != nil {
-		// NOTE if they cancel the request then it will stop the call (kind of cool),
-		// we could filter that error out here too as right now it yells a little
-		if err == models.ErrCallTimeoutServerBusy || err == models.ErrCallTimeout {
-			// TODO maneuver
-			// add this, since it means that start may not have been called [and it's relevant]
-			c.Writer.Header().Add("XXX-FXLB-WAIT", time.Now().Sub(time.Time(model.CreatedAt)).String())
-		}
-		return err
-	}
-
-	// if they don't set a content-type - detect it
-	if writer.Header().Get("Content-Type") == "" {
-		// see http.DetectContentType, the go server is supposed to do this for us but doesn't appear to?
-		var contentType string
-		jsonPrefix := [1]byte{'{'} // stack allocated
-		if bytes.HasPrefix(buf.Bytes(), jsonPrefix[:]) {
-			// try to detect json, since DetectContentType isn't a hipster.
-			contentType = "application/json; charset=utf-8"
-		} else {
-			contentType = http.DetectContentType(buf.Bytes())
-		}
-		writer.Header().Set("Content-Type", contentType)
-	}
-
-	writer.Header().Set("Content-Length", strconv.Itoa(int(buf.Len())))
-
-	if writer.status > 0 {
-		c.Writer.WriteHeader(writer.status)
-	}
-	io.Copy(c.Writer, writer)
-
-	return nil
 }

test/fn-system-tests/exec_fn_test.go

@@ -255,12 +255,14 @@ func TestBasicConcurrentExecution(t *testing.T) {
 	u.Path = path.Join(u.Path, "invoke", fn.ID)
 
 	results := make(chan error)
+	latch := make(chan struct{})
 	concurrentFuncs := 10
 	for i := 0; i < concurrentFuncs; i++ {
 		go func() {
 			body := `{"echoContent": "HelloWorld", "sleepTime": 0, "isDebug": true}`
 			content := bytes.NewBuffer([]byte(body))
 			output := &bytes.Buffer{}
+			<-latch
 			resp, err := callFN(ctx, u.String(), content, output)
 			if err != nil {
 				results <- fmt.Errorf("Got unexpected error: %v", err)
@@ -280,6 +282,7 @@ func TestBasicConcurrentExecution(t *testing.T) {
 			results <- nil
 		}()
 	}
+	close(latch)
 	for i := 0; i < concurrentFuncs; i++ {
 		err := <-results
 		if err != nil {