From d454ff9aa46a78e42071d318d8517562b5c8c59e Mon Sep 17 00:00:00 2001 From: Tom Coupland Date: Mon, 24 Sep 2018 12:20:30 +0100 Subject: [PATCH] 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 | 55 +----------- api/server/runner_fninvoke.go | 71 +++++++++------ api/server/runner_httptrigger.go | 130 ++++++--------------------- test/fn-system-tests/exec_fn_test.go | 3 + 4 files changed, 78 insertions(+), 181 deletions(-) diff --git a/api/agent/call.go b/api/agent/call.go index bb2f65cec..065799a69 100644 --- a/api/agent/call.go +++ b/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 == "" { - fn.Format = models.FormatDefault - } - - 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 + err := FromHTTPFnRequest(app, fn, req)(c) + c.Model().TriggerID = trigger.ID + return err } } diff --git a/api/server/runner_fninvoke.go b/api/server/runner_fninvoke.go index 2a51f5910..0c9ea984f 100644 --- a/api/server/runner_fninvoke.go +++ b/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) WriteHeader(code int) { s.status = code } +func (s *syncResponseWriter) Header() http.Header { return s.headers } + +// 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 - // 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, + writer := &syncResponseWriter{ headers: c.Writer.Header(), } - defer bufPool.Put(buf) // TODO need to ensure this is safe with Dispatch? - call, err := s.agent.GetCall( - agent.WithWriter(&writer), // XXX (reed): order matters [for now] - agent.FromHTTPFnRequest(app, fn, c.Request), - ) + call, err := s.agent.GetCall(agent.WithWriter(writer), // XXX (reed): order matters [for now] + 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) - if err != nil { + submitErr = s.agent.Submit(call) + 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 { - c.Writer.WriteHeader(writer.status) + if writer.Status() > 0 { + c.Writer.WriteHeader(writer.Status()) } - io.Copy(c.Writer, &writer) + + io.Copy(c.Writer, writer) return nil } diff --git a/api/server/runner_httptrigger.go b/api/server/runner_httptrigger.go index fbe930347..885cb3b59 100644 --- a/api/server/runner_httptrigger.go +++ b/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 - headers http.Header + syncResponseWriter 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 { - 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) + trw.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, - headers: make(http.Header), + syncResponseWriter{ + 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] - agent.FromHTTPTriggerRequest(app, fn, trigger, c.Request), - ) + + call, err := s.agent.GetCall(agent.WithWriter(triggerWriter), // XXX (reed): order matters [for now] + agent.FromHTTPTriggerRequest(app, fn, trigger, c.Request)) if err != nil { return err } - 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) - } - 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 + return s.fnInvoke(c, app, fn, triggerWriter, call) } diff --git a/test/fn-system-tests/exec_fn_test.go b/test/fn-system-tests/exec_fn_test.go index bc2f621c2..f040d175a 100644 --- a/test/fn-system-tests/exec_fn_test.go +++ b/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 {