From 0dbc93725dc5ff30b4e0cd836f69e694f552c2b1 Mon Sep 17 00:00:00 2001 From: Qiu Jian Date: Sat, 20 Oct 2018 22:32:43 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=AD=A3=EF=BC=9A=E5=88=A9=E7=94=A8co?= =?UTF-8?q?ntext.Timeout=EF=BC=8Cappsrv=E6=94=AF=E6=8C=81processTimeout?= =?UTF-8?q?=EF=BC=8C=E9=81=BF=E5=85=8Dyunionapi=E7=94=B1=E4=BA=8E=E5=90=8E?= =?UTF-8?q?=E7=AB=AF=E5=8D=A1=E4=BD=8F=E5=AF=BC=E8=87=B4yunionapi=E5=8D=A1?= =?UTF-8?q?=E4=BD=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/appsrv/appsrv.go | 117 +++++++++-------------------------- pkg/appsrv/response.go | 87 ++++++++++++++++++++++++++ pkg/httperrors/errors.go | 5 ++ pkg/httperrors/httperrors.go | 4 ++ 4 files changed, 126 insertions(+), 87 deletions(-) create mode 100644 pkg/appsrv/response.go diff --git a/pkg/appsrv/appsrv.go b/pkg/appsrv/appsrv.go index 314bfc185f..f434bd44b4 100644 --- a/pkg/appsrv/appsrv.go +++ b/pkg/appsrv/appsrv.go @@ -10,79 +10,15 @@ import ( "time" "yunion.io/x/log" - "yunion.io/x/onecloud/pkg/appctx" - "yunion.io/x/onecloud/pkg/proxy" "yunion.io/x/pkg/trace" "yunion.io/x/pkg/utils" + + "yunion.io/x/onecloud/pkg/appctx" + "yunion.io/x/onecloud/pkg/httperrors" + "yunion.io/x/onecloud/pkg/proxy" + "yunion.io/x/onecloud/pkg/util/httputils" ) -type responseWriterResponse struct { - count int - err error -} - -type responseWriterChannel struct { - backend http.ResponseWriter - bodyChan chan []byte - bodyResp chan responseWriterResponse - statusChan chan int - statusResp chan bool -} - -func newResponseWriterChannel(backend http.ResponseWriter) responseWriterChannel { - return responseWriterChannel{backend: backend, - bodyChan: make(chan []byte), - bodyResp: make(chan responseWriterResponse), - statusChan: make(chan int), - statusResp: make(chan bool)} -} - -func (w *responseWriterChannel) Header() http.Header { - return w.backend.Header() -} - -func (w *responseWriterChannel) Write(bytes []byte) (int, error) { - w.bodyChan <- bytes - v := <-w.bodyResp - return v.count, v.err -} - -func (w *responseWriterChannel) WriteHeader(status int) { - w.statusChan <- status - <-w.statusResp -} - -func (w *responseWriterChannel) wait() { - stop := false - for !stop { - select { - case bytes, more := <-w.bodyChan: - // log.Print("Recive body ", len(bytes), " more ", more) - if more { - c, e := w.backend.Write(bytes) - w.bodyResp <- responseWriterResponse{count: c, err: e} - } else { - stop = true - } - case status, more := <-w.statusChan: - // log.Print("Recive status ", status, " more ", more) - if more { - w.backend.WriteHeader(status) - w.statusResp <- true - } else { - stop = true - } - } - } -} - -func (w *responseWriterChannel) closeChannels() { - close(w.bodyChan) - close(w.bodyResp) - close(w.statusChan) - close(w.statusResp) -} - type Application struct { name string context context.Context @@ -106,6 +42,7 @@ const ( DEFAULT_READ_TIMEOUT = 0 DEFAULT_READ_HEADER_TIMEOUT = 10 * time.Second DEFAULT_WRITE_TIMEOUT = 0 + DEFAULT_PROCESS_TIMEOUT = 15 * time.Second ) func NewApplication(name string, connMax int) *Application { @@ -118,7 +55,9 @@ func NewApplication(name string, connMax int) *Application { idleTimeout: DEFAULT_IDLE_TIMEOUT, readTimeout: DEFAULT_READ_TIMEOUT, readHeaderTimeout: DEFAULT_READ_HEADER_TIMEOUT, - writeTimeout: DEFAULT_WRITE_TIMEOUT} + writeTimeout: DEFAULT_WRITE_TIMEOUT, + processTimeout: DEFAULT_PROCESS_TIMEOUT, + } app.SetContext(appctx.APP_CONTEXT_KEY_APP, &app) app.SetContext(appctx.APP_CONTEXT_KEY_APPNAME, app.name) @@ -174,9 +113,6 @@ func (app *Application) AddHandler(method string, prefix string, handler func(co func (app *Application) AddHandler2(method string, prefix string, handler func(context.Context, http.ResponseWriter, *http.Request), metadata map[string]interface{}, name string, tags map[string]string) { log.Debugf("%s - %s", method, prefix) segs := SplitPath(prefix) - // for i := len(this.middlewares) - 1; i >= 0; i -= 1 { - // handler = this.middlewares[i](handler) - // } e := app.getRoot(method).Add(segs, newHandlerInfo(method, segs, handler, metadata, name, tags)) if e != nil { log.Fatalf("Fail to register %s %s: %s", method, prefix, e) @@ -254,9 +190,9 @@ func (app *Application) defaultHandle(w http.ResponseWriter, r *http.Request, ri if ok { fw := newResponseWriterChannel(w) errChan := make(chan interface{}) + ctx, cancel := context.WithTimeout(app.context, app.processTimeout) + defer cancel() app.session.Run(func() { - ctx, cancel := context.WithCancel(app.context) - defer cancel() defer fw.closeChannels() if ctx.Err() == nil { ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_REQUEST_ID, rid) @@ -266,25 +202,32 @@ func (app *Application) defaultHandle(w http.ResponseWriter, r *http.Request, ri if hand.metadata != nil { ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_METADATA, hand.metadata) } - span := trace.StartServerTrace(w, r, hand.GetName(params), app.GetName(), hand.GetTags()) - ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_TRACE, span) - hand.handler(ctx, &fw, r) - span.EndTrace() + func() { + span := trace.StartServerTrace(w, r, hand.GetName(params), app.GetName(), hand.GetTags()) + defer span.EndTrace() + ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_TRACE, span) + hand.handler(ctx, &fw, r) + }() } // otherwise, the task has been timeout }, errChan) - fw.wait() - runerr := WaitChannel(errChan) - if runerr != nil { - http.Error(w, fmt.Sprintf("Internal error: %s", runerr), http.StatusInternalServerError) + runErr := fw.wait(ctx, errChan) + if runErr != nil { + switch runErr.(type) { + case *httputils.JSONClientError: + je := runErr.(*httputils.JSONClientError) + httperrors.GeneralServerError(w, je) + default: + httperrors.InternalServerError(w, "Internal server error") + } } return hand } else { - log.Printf("Invalid handler for %s", r.URL) - http.Error(w, "Invalid handler", 500) + log.Errorf("Invalid handler for %s", r.URL) + httperrors.InternalServerError(w, "Invalid handler %s", r.URL) } } else if !isCors { - log.Printf("Handler not found") - http.NotFound(w, r) + log.Errorf("Handler not found") + httperrors.NotFoundError(w, "Handler not found") } return nil } diff --git a/pkg/appsrv/response.go b/pkg/appsrv/response.go new file mode 100644 index 0000000000..c2a4263700 --- /dev/null +++ b/pkg/appsrv/response.go @@ -0,0 +1,87 @@ +package appsrv + +import ( + "context" + "net/http" + + "yunion.io/x/onecloud/pkg/httperrors" +) + +type responseWriterResponse struct { + count int + err error +} + +type responseWriterChannel struct { + backend http.ResponseWriter + bodyChan chan []byte + bodyResp chan responseWriterResponse + statusChan chan int + statusResp chan bool +} + +func newResponseWriterChannel(backend http.ResponseWriter) responseWriterChannel { + return responseWriterChannel{backend: backend, + bodyChan: make(chan []byte), + bodyResp: make(chan responseWriterResponse), + statusChan: make(chan int), + statusResp: make(chan bool)} +} + +func (w *responseWriterChannel) Header() http.Header { + return w.backend.Header() +} + +func (w *responseWriterChannel) Write(bytes []byte) (int, error) { + w.bodyChan <- bytes + v := <-w.bodyResp + return v.count, v.err +} + +func (w *responseWriterChannel) WriteHeader(status int) { + w.statusChan <- status + <-w.statusResp +} + +func (w *responseWriterChannel) wait(ctx context.Context, errChan chan interface{}) interface{} { + var err interface{} + stop := false + for !stop { + select { + case <-ctx.Done(): + // ctx deadline reached, timeout + err = httperrors.NewTimeoutError("request process timeout") + stop = true + case e, more := <-errChan: + if more { + err = e + } else { + stop = true + } + case bytes, more := <-w.bodyChan: + // log.Print("Recive body ", len(bytes), " more ", more) + if more { + c, e := w.backend.Write(bytes) + w.bodyResp <- responseWriterResponse{count: c, err: e} + } else { + stop = true + } + case status, more := <-w.statusChan: + // log.Print("Recive status ", status, " more ", more) + if more { + w.backend.WriteHeader(status) + w.statusResp <- true + } else { + stop = true + } + } + } + return err +} + +func (w *responseWriterChannel) closeChannels() { + close(w.bodyChan) + close(w.bodyResp) + close(w.statusChan) + close(w.statusResp) +} diff --git a/pkg/httperrors/errors.go b/pkg/httperrors/errors.go index 4e4b226e03..90927a6a10 100644 --- a/pkg/httperrors/errors.go +++ b/pkg/httperrors/errors.go @@ -185,6 +185,11 @@ func NewRequireLicenseError(msg string, params ...interface{}) *httputils.JSONCl return NewJsonClientError(402, "RequireLicenseError", msg, err) } +func NewTimeoutError(msg string, params ...interface{}) *httputils.JSONClientError { + msg, err := errorMessage(msg, params...) + return NewJsonClientError(504, "TimeoutError", msg, err) +} + func NewGeneralError(err error) *httputils.JSONClientError { switch err.(type) { case *httputils.JSONClientError: diff --git a/pkg/httperrors/httperrors.go b/pkg/httperrors/httperrors.go index b9be4abbaf..e23e18c018 100644 --- a/pkg/httperrors/httperrors.go +++ b/pkg/httperrors/httperrors.go @@ -94,3 +94,7 @@ func TenantNotFoundError(w http.ResponseWriter, msg string, params ...interface{ func OutOfQuotaError(w http.ResponseWriter, msg string, params ...interface{}) { JsonClientError(w, NewOutOfQuotaError(msg, params...)) } + +func TimeoutError(w http.ResponseWriter, msg string, params ...interface{}) { + JsonClientError(w, NewTimeoutError(msg, params...)) +}