Fix truncated body returned from router (#1420)
If the context of request is closed before ReverseProxy finishing writing a huge response body to the response writer, the client will only receive a truncated response body. To solve this, move the context cancel after ReverseProxy finished.
This commit is contained in:
@@ -86,9 +86,10 @@ type (
|
|||||||
|
|
||||||
// A layer on top of http.DefaultTransport, with retries.
|
// A layer on top of http.DefaultTransport, with retries.
|
||||||
RetryingRoundTripper struct {
|
RetryingRoundTripper struct {
|
||||||
logger *zap.Logger
|
logger *zap.Logger
|
||||||
funcHandler *functionHandler
|
funcHandler *functionHandler
|
||||||
timeout int
|
funcTimeout time.Duration
|
||||||
|
closeContextFunc *context.CancelFunc
|
||||||
}
|
}
|
||||||
|
|
||||||
// To keep the request body open during retries, we create an interface with Close operation being a no-op.
|
// To keep the request body open during retries, we create an interface with Close operation being a no-op.
|
||||||
@@ -145,7 +146,7 @@ func (w *fakeCloseReadCloser) RealClose() error {
|
|||||||
// inside ServeHttp function of the reverseProxy.
|
// inside ServeHttp function of the reverseProxy.
|
||||||
// Earlier, GetServiceForFunction was called inside handler function and fission explicitly set http status code to 500
|
// Earlier, GetServiceForFunction was called inside handler function and fission explicitly set http status code to 500
|
||||||
// if it returned an error.
|
// if it returned an error.
|
||||||
func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
|
func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
|
||||||
// Set forwarded host header if not exists
|
// Set forwarded host header if not exists
|
||||||
roundTripper.addForwardedHostHeader(req)
|
roundTripper.addForwardedHostHeader(req)
|
||||||
|
|
||||||
@@ -183,6 +184,8 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Res
|
|||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
roundTripper.logger.Debug("request headers", zap.Any("headers", req.Header))
|
||||||
|
|
||||||
// The reason for request failure may vary from case to case.
|
// The reason for request failure may vary from case to case.
|
||||||
// After some investigation, found most of the failure are due to
|
// After some investigation, found most of the failure are due to
|
||||||
// network timeout or target function is under heavy workload. In
|
// network timeout or target function is under heavy workload. In
|
||||||
@@ -265,30 +268,20 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Res
|
|||||||
|
|
||||||
overhead := time.Since(startTime)
|
overhead := time.Since(startTime)
|
||||||
|
|
||||||
roundTripper.logger.Debug("request headers", zap.Any("headers", req.Header))
|
// Do NOT assign returned request to "req"
|
||||||
|
// because the request used in the last round
|
||||||
// Creating context for client
|
// will be canceled when calling setContext.
|
||||||
if roundTripper.timeout <= 0 {
|
newReq := roundTripper.setContext(req)
|
||||||
roundTripper.timeout = fv1.DEFAULT_FUNCTION_TIMEOUT
|
|
||||||
}
|
|
||||||
|
|
||||||
roundTripper.logger.Debug("Creating context for request for ", zap.Any("time", roundTripper.timeout))
|
|
||||||
// pass request context as parent context for the case
|
|
||||||
// that user aborts connection before timeout. Otherwise,
|
|
||||||
// the request won't be canceled until the deadline exceeded
|
|
||||||
// which may be a potential security issue.
|
|
||||||
ctx, closeCtx := context.WithTimeout(req.Context(), time.Duration(roundTripper.timeout)*time.Second)
|
|
||||||
|
|
||||||
// forward the request to the function service
|
// forward the request to the function service
|
||||||
resp, err = ocRoundTripper.RoundTrip(req.WithContext(ctx))
|
resp, err = ocRoundTripper.RoundTrip(newReq)
|
||||||
closeCtx()
|
|
||||||
|
|
||||||
if err == nil {
|
if err == nil {
|
||||||
// Track metrics
|
// Track metrics
|
||||||
httpMetricLabels.code = resp.StatusCode
|
httpMetricLabels.code = resp.StatusCode
|
||||||
funcMetricLabels.cached = serviceUrlFromCache
|
funcMetricLabels.cached = serviceUrlFromCache
|
||||||
|
|
||||||
functionCallCompleted(funcMetricLabels, httpMetricLabels,
|
go functionCallCompleted(funcMetricLabels, httpMetricLabels,
|
||||||
overhead, time.Since(startTime), resp.ContentLength)
|
overhead, time.Since(startTime), resp.ContentLength)
|
||||||
|
|
||||||
// return response back to user
|
// return response back to user
|
||||||
@@ -373,6 +366,28 @@ func (roundTripper RetryingRoundTripper) getDefaultTransport() *http.Transport {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// setContext returns a shallow copy of request with a new timeout context.
|
||||||
|
func (roundTripper *RetryingRoundTripper) setContext(req *http.Request) *http.Request {
|
||||||
|
if roundTripper.closeContextFunc != nil {
|
||||||
|
(*roundTripper.closeContextFunc)()
|
||||||
|
}
|
||||||
|
// pass request context as parent context for the case
|
||||||
|
// that user aborts connection before timeout. Otherwise,
|
||||||
|
// the request won't be canceled until the deadline exceeded
|
||||||
|
// which may be a potential security issue.
|
||||||
|
ctx, closeCtx := context.WithTimeout(req.Context(), roundTripper.funcTimeout)
|
||||||
|
roundTripper.closeContextFunc = &closeCtx
|
||||||
|
|
||||||
|
return req.WithContext(ctx)
|
||||||
|
}
|
||||||
|
|
||||||
|
// closeContext closes the context to release resources.
|
||||||
|
func (roundTripper *RetryingRoundTripper) closeContext() {
|
||||||
|
if roundTripper.closeContextFunc != nil {
|
||||||
|
(*roundTripper.closeContextFunc)()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (fh *functionHandler) tapService(serviceUrl *url.URL) {
|
func (fh *functionHandler) tapService(serviceUrl *url.URL) {
|
||||||
if fh.executor == nil {
|
if fh.executor == nil {
|
||||||
return
|
return
|
||||||
@@ -408,21 +423,36 @@ func (fh functionHandler) handler(responseWriter http.ResponseWriter, request *h
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
var timeout int = fv1.DEFAULT_FUNCTION_TIMEOUT
|
fnTimeout := fh.functionTimeoutMap[fh.function.GetUID()]
|
||||||
if fh.functionTimeoutMap != nil {
|
if fnTimeout == 0 {
|
||||||
timeout = fh.functionTimeoutMap[fh.function.GetUID()]
|
fnTimeout = fv1.DEFAULT_FUNCTION_TIMEOUT
|
||||||
|
}
|
||||||
|
|
||||||
|
rrt := &RetryingRoundTripper{
|
||||||
|
logger: fh.logger.Named("roundtripper"),
|
||||||
|
funcHandler: &fh,
|
||||||
|
funcTimeout: time.Duration(fnTimeout) * time.Second,
|
||||||
}
|
}
|
||||||
|
|
||||||
proxy := &httputil.ReverseProxy{
|
proxy := &httputil.ReverseProxy{
|
||||||
Director: director,
|
Director: director,
|
||||||
Transport: &RetryingRoundTripper{
|
Transport: rrt,
|
||||||
logger: fh.logger.Named("roundtripper"),
|
|
||||||
funcHandler: &fh,
|
|
||||||
timeout: timeout,
|
|
||||||
},
|
|
||||||
ErrorHandler: getProxyErrorHandler(fh.logger, fh.function),
|
ErrorHandler: getProxyErrorHandler(fh.logger, fh.function),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
defer func() {
|
||||||
|
// If the context is closed when RoundTrip returns, client may receive
|
||||||
|
// truncated response body due to "context canceled" error. To avoid
|
||||||
|
// this, we need to close request context after proxy.ServeHTTP finished.
|
||||||
|
//
|
||||||
|
// NOTE: rrt.closeContext() must be put in the defer function; otherwise,
|
||||||
|
// reverseProxy may panic when failed to write response and the context
|
||||||
|
// will not be closed.
|
||||||
|
//
|
||||||
|
// ref: https://github.com/golang/go/issues/28239
|
||||||
|
rrt.closeContext()
|
||||||
|
}()
|
||||||
|
|
||||||
proxy.ServeHTTP(responseWriter, request)
|
proxy.ServeHTTP(responseWriter, request)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user