Queued request timeout

aspects/request_timeout_handler.go

@@ -0,0 +1,41 @@
+package aspects
+
+import (
+	"github.com/julienschmidt/httprouter"
+	"github.com/prebid/prebid-server/config"
+	"net/http"
+	"strconv"
+)
+
+func QueuedRequestTimeout(f httprouter.Handle, custHeaders config.CustomHeaders) httprouter.Handle {
+
+	return func(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
+
+		reqTimeInQueue := r.Header.Get(custHeaders.RequestTimeInQueue)
+		reqTimeout := r.Header.Get(custHeaders.RequestTimeoutInQueue)
+
+		//If request timeout headers are not specified - process request as usual
+		if reqTimeInQueue == "" || reqTimeout == "" {
+			f(w, r, params)
+			return
+		}
+
+		reqTimeFloat, reqTimeFloatErr := strconv.ParseFloat(reqTimeInQueue, 64)
+		reqTimeoutFloat, reqTimeoutFloatErr := strconv.ParseFloat(reqTimeout, 64)
+
+		//Return HTTP 400 if request timeout headers are incorrect (wrong format)
+		if reqTimeFloatErr != nil || reqTimeoutFloatErr != nil {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+
+		//Return HTTP 408 if requests stays too long in queue
+		if reqTimeFloat >= reqTimeoutFloat {
+			w.WriteHeader(http.StatusRequestTimeout)
+			return
+		}
+
+		f(w, r, params)
+	}
+
+}

aspects/request_timeout_handler_test.go

@@ -0,0 +1,90 @@
+package aspects
+
+import (
+	"github.com/julienschmidt/httprouter"
+	"github.com/prebid/prebid-server/config"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+const reqTimeInQueueHeaderName = "X-Ngx-Request-Time"
+const reqTimeoutHeaderName = "X-Ngx-Request-Timeout"
+
+func TestQueuedRequestTimeoutWithTimeout(t *testing.T) {
+
+	rw := ExecuteAspectRequest(t, "6", "5", true)
+
+	assert.Equal(t, http.StatusRequestTimeout, rw.Code, "Http response code is incorrect, should be 408")
+	assert.Equal(t, "", string(rw.Body.Bytes()), "Body should not be present in response")
+
+}
+
+func TestQueuedRequestTimeoutNoTimeout(t *testing.T) {
+
+	rw := ExecuteAspectRequest(t, "0.9", "5", true)
+
+	assert.Equal(t, http.StatusOK, rw.Code, "Http response code is incorrect, should be 200")
+	assert.Equal(t, "Executed", string(rw.Body.Bytes()), "Body should be present in response")
+
+}
+
+func TestQueuedRequestNoHeaders(t *testing.T) {
+
+	rw := ExecuteAspectRequest(t, "", "", false)
+
+	assert.Equal(t, http.StatusOK, rw.Code, "Http response code is incorrect, should be 200")
+	assert.Equal(t, "Executed", string(rw.Body.Bytes()), "Body should be present in response")
+
+}
+
+func TestQueuedRequestAllHeadersIncorrect(t *testing.T) {
+
+	rw := ExecuteAspectRequest(t, "test1", "test2", true)
+
+	assert.Equal(t, http.StatusBadRequest, rw.Code, "Http response code is incorrect, should be 400")
+	assert.Equal(t, "", string(rw.Body.Bytes()), "Body should not be present in response")
+
+}
+
+func TestQueuedRequestSomeHeadersIncorrect(t *testing.T) {
+
+	rw := ExecuteAspectRequest(t, "test1", "123", true)
+
+	assert.Equal(t, http.StatusBadRequest, rw.Code, "Http response code is incorrect, should be 400")
+	assert.Equal(t, "", string(rw.Body.Bytes()), "Body should not be present in response")
+
+}
+
+func MockEndpoint() httprouter.Handle {
+	return httprouter.Handle(MockHandler)
+}
+
+func MockHandler(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
+	w.Write([]byte("Executed"))
+}
+
+func ExecuteAspectRequest(t *testing.T, timeInQueue string, reqTimeout string, setHeaders bool) *httptest.ResponseRecorder {
+	rw := httptest.NewRecorder()
+	req, err := http.NewRequest("POST", "/test", nil)
+	if err != nil {
+		assert.Fail(t, "Unable create mock http request")
+	}
+	if setHeaders {
+		req.Header.Set(reqTimeInQueueHeaderName, timeInQueue)
+		req.Header.Set(reqTimeoutHeaderName, reqTimeout)
+	}
+
+	customHeaders := config.CustomHeaders{reqTimeInQueueHeaderName, reqTimeoutHeaderName}
+
+	handler := QueuedRequestTimeout(MockEndpoint(), customHeaders)
+
+	r := httprouter.New()
+	r.POST("/test", handler)
+
+	r.ServeHTTP(rw, req)
+
+	return rw
+}

config/config.go

@@ -64,6 +64,8 @@ type Configuration struct {
 	AccountRequired bool `mapstructure:"account_required"`
 	// Local private file containing SSL certificates
 	PemCertsFile string `mapstructure:"certificates_file"`
+	// Custom headers set by Nginx to handle request timeout in queue or any other custom headers
+	CustomHeaders CustomHeaders `mapstructure:"custom_headers"`
 }
 
 const MIN_COOKIE_SIZE_BYTES = 500
@@ -199,6 +201,11 @@ type HostCookie struct {
 	TTL int64 `mapstructure:"ttl_days"`
 }
 
+type CustomHeaders struct {
+	RequestTimeInQueue    string `mapstructure:"request_time_in_queue"`
+	RequestTimeoutInQueue string `mapstructure:"request_timeout_in_queue"`
+}
+
 func (cfg *HostCookie) TTLDuration() time.Duration {
 	return time.Duration(cfg.TTL) * time.Hour * 24
 }

router/router.go

@@ -23,6 +23,7 @@ import (
 	"github.com/prebid/prebid-server/adapters/rubicon"
 	"github.com/prebid/prebid-server/adapters/sovrn"
 	analyticsConf "github.com/prebid/prebid-server/analytics/config"
+	"github.com/prebid/prebid-server/aspects"
 	"github.com/prebid/prebid-server/cache"
 	"github.com/prebid/prebid-server/cache/dummycache"
 	"github.com/prebid/prebid-server/cache/filecache"
@@ -255,6 +256,11 @@ func New(cfg *config.Configuration, rateConvertor *currencies.RateConverter) (r
 		glog.Fatalf("Failed to create the video endpoint handler. %v", err)
 	}
 
+	customHeaders := config.CustomHeaders{}
+	if cfg.CustomHeaders != customHeaders {
+		videoEndpoint = aspects.QueuedRequestTimeout(videoEndpoint, cfg.CustomHeaders)
+	}
+
 	r.POST("/auction", endpoints.Auction(cfg, syncers, gdprPerms, r.MetricsEngine, dataCache, exchanges))
 	r.POST("/openrtb2/auction", openrtbEndpoint)
 	r.POST("/openrtb2/video", videoEndpoint)