Added metrics support to endpoint aspect (prebid#1226)

Co-authored-by: Veronika Solovei <veronika.solovei@xandr.com>

pbsmetrics/config/metrics.go

@@ -181,6 +181,13 @@ func (me *MultiMetricsEngine) RecordPrebidCacheRequestTime(success bool, length
 	}
 }
 
+// RecordRequestQueueTime across all engines
+func (me *MultiMetricsEngine) RecordRequestQueueTime(success bool, requestType pbsmetrics.RequestType, length time.Duration) {
+	for _, thisME := range *me {
+		thisME.RecordRequestQueueTime(success, requestType, length)
+	}
+}
+
 // DummyMetricsEngine is a Noop metrics engine in case no metrics are configured. (may also be useful for tests)
 type DummyMetricsEngine struct{}
 
@@ -251,3 +258,7 @@ func (me *DummyMetricsEngine) RecordStoredImpCacheResult(cacheResult pbsmetrics.
 // RecordPrebidCacheRequestTime as a noop
 func (me *DummyMetricsEngine) RecordPrebidCacheRequestTime(success bool, length time.Duration) {
 }
+
+// RecordRequestQueueTime as a noop
+func (me *DummyMetricsEngine) RecordRequestQueueTime(success bool, requestType pbsmetrics.RequestType, length time.Duration) {
+}

pbsmetrics/config/metrics_test.go

@@ -115,6 +115,9 @@ func TestMultiMetricsEngine(t *testing.T) {
 	for i := 0; i < 3; i++ {
 		metricsEngine.RecordImps(impTypeLabels)
 	}
+
+	metricsEngine.RecordRequestQueueTime(false, pbsmetrics.ReqTypeVideo, time.Duration(1))
+
 	//Make the metrics engine, instantiated here with goEngine, fill its RequestStatuses[RequestType][pbsmetrics.RequestStatusXX] with the new boolean values added to pbsmetrics.Labels
 	VerifyMetrics(t, "RequestStatuses.OpenRTB2.OK", goEngine.RequestStatuses[pbsmetrics.ReqTypeORTB2Web][pbsmetrics.RequestStatusOK].Count(), 5)
 	VerifyMetrics(t, "RequestStatuses.Legacy.OK", goEngine.RequestStatuses[pbsmetrics.ReqTypeLegacy][pbsmetrics.RequestStatusOK].Count(), 0)
@@ -148,6 +151,9 @@ func TestMultiMetricsEngine(t *testing.T) {
 	}
 	VerifyMetrics(t, "AdapterMetrics.AppNexus.GotBidsMeter", goEngine.AdapterMetrics[openrtb_ext.BidderAppnexus].GotBidsMeter.Count(), 0)
 	VerifyMetrics(t, "AdapterMetrics.AppNexus.NoBidMeter", goEngine.AdapterMetrics[openrtb_ext.BidderAppnexus].NoBidMeter.Count(), 5)
+
+	VerifyMetrics(t, "RecordRequestQueueTime.Video.Rejected", goEngine.RequestsQueueTimer[pbsmetrics.ReqTypeVideo][false].Count(), 1)
+	VerifyMetrics(t, "RecordRequestQueueTime.Video.Accepted", goEngine.RequestsQueueTimer[pbsmetrics.ReqTypeVideo][true].Count(), 0)
 }
 
 func VerifyMetrics(t *testing.T, name string, actual int64, expected int64) {

pbsmetrics/go_metrics.go

@@ -24,6 +24,7 @@ type Metrics struct {
 	SafariRequestMeter             metrics.Meter
 	SafariNoCookieMeter            metrics.Meter
 	RequestTimer                   metrics.Timer
+	RequestsQueueTimer             map[RequestType]map[bool]metrics.Timer
 	PrebidCacheRequestTimerSuccess metrics.Timer
 	PrebidCacheRequestTimerError   metrics.Timer
 	StoredReqCacheMeter            map[CacheResult]metrics.Meter
@@ -111,6 +112,7 @@ func NewBlankMetrics(registry metrics.Registry, exchanges []openrtb_ext.BidderNa
 		SafariRequestMeter:             blankMeter,
 		SafariNoCookieMeter:            blankMeter,
 		RequestTimer:                   blankTimer,
+		RequestsQueueTimer:             make(map[RequestType]map[bool]metrics.Timer),
 		PrebidCacheRequestTimerSuccess: blankTimer,
 		PrebidCacheRequestTimerError:   blankTimer,
 		StoredReqCacheMeter:            make(map[CacheResult]metrics.Meter),
@@ -146,6 +148,11 @@ func NewBlankMetrics(registry metrics.Registry, exchanges []openrtb_ext.BidderNa
 		}
 	}
 
+	//to minimize memory usage, queuedTimeout metric is now supported for video endpoint only
+	//boolean value represents 2 general request statuses: accepted and rejected
+	newMetrics.RequestsQueueTimer["video"] = make(map[bool]metrics.Timer)
+	newMetrics.RequestsQueueTimer["video"][true] = blankTimer
+	newMetrics.RequestsQueueTimer["video"][false] = blankTimer
 	return newMetrics
 }
 
@@ -191,11 +198,15 @@ func NewMetrics(registry metrics.Registry, exchanges []openrtb_ext.BidderName, d
 			statusMap[stat] = metrics.GetOrRegisterMeter("requests."+string(stat)+"."+string(typ), registry)
 		}
 	}
+
 	for _, cacheRes := range CacheResults() {
 		newMetrics.StoredReqCacheMeter[cacheRes] = metrics.GetOrRegisterMeter(fmt.Sprintf("stored_request_cache_%s", string(cacheRes)), registry)
 		newMetrics.StoredImpCacheMeter[cacheRes] = metrics.GetOrRegisterMeter(fmt.Sprintf("stored_imp_cache_%s", string(cacheRes)), registry)
 	}
 
+	newMetrics.RequestsQueueTimer["video"][true] = metrics.GetOrRegisterTimer("queued_requests.video.accepted", registry)
+	newMetrics.RequestsQueueTimer["video"][false] = metrics.GetOrRegisterTimer("queued_requests.video.rejected", registry)
+
 	newMetrics.userSyncSet[unknownBidder] = metrics.GetOrRegisterMeter("usersync.unknown.sets", registry)
 	newMetrics.userSyncGDPRPrevent[unknownBidder] = metrics.GetOrRegisterMeter("usersync.unknown.gdpr_prevent", registry)
 	return newMetrics
@@ -526,6 +537,13 @@ func (me *Metrics) RecordPrebidCacheRequestTime(success bool, length time.Durati
 	}
 }
 
+func (me *Metrics) RecordRequestQueueTime(success bool, requestType RequestType, length time.Duration) {
+	if requestType == ReqTypeVideo { //remove this check when other request types are supported
+		me.RequestsQueueTimer[requestType][success].Update(length)
+	}
+
+}
+
 func doMark(bidder openrtb_ext.BidderName, meters map[openrtb_ext.BidderName]metrics.Meter) {
 	met, ok := meters[bidder]
 	if ok {

pbsmetrics/go_metrics_test.go

@@ -50,6 +50,9 @@ func TestNewMetrics(t *testing.T) {
 	ensureContains(t, registry, "requests.badinput.video", m.RequestStatuses[ReqTypeVideo][RequestStatusBadInput])
 	ensureContains(t, registry, "requests.err.video", m.RequestStatuses[ReqTypeVideo][RequestStatusErr])
 	ensureContains(t, registry, "requests.networkerr.video", m.RequestStatuses[ReqTypeVideo][RequestStatusNetworkErr])
+
+	ensureContains(t, registry, "queued_requests.video.rejected", m.RequestsQueueTimer[ReqTypeVideo][false])
+	ensureContains(t, registry, "queued_requests.video.accepted", m.RequestsQueueTimer[ReqTypeVideo][true])
 }
 
 func TestRecordBidType(t *testing.T) {

pbsmetrics/metrics.go

@@ -154,11 +154,12 @@ func CookieTypes() []CookieFlag {
 
 // Request/return status
 const (
-	RequestStatusOK          RequestStatus = "ok"
-	RequestStatusBadInput    RequestStatus = "badinput"
-	RequestStatusErr         RequestStatus = "err"
-	RequestStatusNetworkErr  RequestStatus = "networkerr"
-	RequestStatusBlacklisted RequestStatus = "blacklistedacctorapp"
+	RequestStatusOK           RequestStatus = "ok"
+	RequestStatusBadInput     RequestStatus = "badinput"
+	RequestStatusErr          RequestStatus = "err"
+	RequestStatusNetworkErr   RequestStatus = "networkerr"
+	RequestStatusBlacklisted  RequestStatus = "blacklistedacctorapp"
+	RequestStatusQueueTimeout RequestStatus = "queuetimeout"
 )
 
 func RequestStatuses() []RequestStatus {
@@ -168,6 +169,7 @@ func RequestStatuses() []RequestStatus {
 		RequestStatusErr,
 		RequestStatusNetworkErr,
 		RequestStatusBlacklisted,
+		RequestStatusQueueTimeout,
 	}
 }
 
@@ -272,4 +274,5 @@ type MetricsEngine interface {
 	RecordStoredReqCacheResult(cacheResult CacheResult, inc int)
 	RecordStoredImpCacheResult(cacheResult CacheResult, inc int)
 	RecordPrebidCacheRequestTime(success bool, length time.Duration)
+	RecordRequestQueueTime(success bool, requestType RequestType, length time.Duration)
 }

pbsmetrics/metrics_mock.go

@@ -96,3 +96,8 @@ func (me *MetricsEngineMock) RecordStoredImpCacheResult(cacheResult CacheResult,
 func (me *MetricsEngineMock) RecordPrebidCacheRequestTime(success bool, length time.Duration) {
 	me.Called(success, length)
 }
+
+// RecordRequestQueueTime mock
+func (me *MetricsEngineMock) RecordRequestQueueTime(success bool, requestType RequestType, length time.Duration) {
+	me.Called(success, requestType, length)
+}

pbsmetrics/prometheus/preload.go

@@ -1,6 +1,7 @@
 package prometheusmetrics
 
 import (
+	"github.com/prebid/prebid-server/pbsmetrics"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
@@ -91,6 +92,13 @@ func preloadLabelValues(m *Metrics) {
 		adapterLabel: adapterValues,
 		actionLabel:  actionValues,
 	})
+
+	//to minimize memory usage, queuedTimeout metric is now supported for video endpoint only
+	//boolean value represents 2 general request statuses: accepted and rejected
+	preloadLabelValuesForHistogram(m.requestsQueueTimer, map[string][]string{
+		requestTypeLabel:   {string(pbsmetrics.ReqTypeVideo)},
+		requestStatusLabel: {requestSuccessLabel, requestRejectLabel},
+	})
 }
 
 func preloadLabelValuesForCounter(counter *prometheus.CounterVec, labelsWithValues map[string][]string) {

pbsmetrics/prometheus/prometheus.go

@@ -24,6 +24,7 @@ type Metrics struct {
 	prebidCacheWriteTimer        *prometheus.HistogramVec
 	requests                     *prometheus.CounterVec
 	requestsTimer                *prometheus.HistogramVec
+	requestsQueueTimer           *prometheus.HistogramVec
 	requestsWithoutCookie        *prometheus.CounterVec
 	storedImpressionsCacheResult *prometheus.CounterVec
 	storedRequestCacheResult     *prometheus.CounterVec
@@ -73,11 +74,17 @@ const (
 	markupDeliveryNurl = "nurl"
 )
 
+const (
+	requestSuccessLabel = "requestAcceptedLabel"
+	requestRejectLabel  = "requestRejectedLabel"
+)
+
 // NewMetrics initializes a new Prometheus metrics instance with preloaded label values.
 func NewMetrics(cfg config.PrometheusMetrics) *Metrics {
 	requestTimeBuckets := []float64{0.05, 0.1, 0.15, 0.20, 0.25, 0.3, 0.4, 0.5, 0.75, 1}
 	cacheWriteTimeBuckets := []float64{0.001, 0.002, 0.005, 0.01, 0.025, 0.05, 0.1, 0.2, 0.3, 0.4, 0.5, 1}
 	priceBuckets := []float64{250, 500, 750, 1000, 1500, 2000, 2500, 3000, 3500, 4000}
+	queuedRequestTimeBuckets := []float64{0, 1, 5, 30, 60, 120, 180, 240, 300}
 
 	metrics := Metrics{}
 	metrics.Registry = prometheus.NewRegistry()
@@ -187,6 +194,12 @@ func NewMetrics(cfg config.PrometheusMetrics) *Metrics {
 		"Count of total requests to Prebid Server labeled by account.",
 		[]string{accountLabel})
 
+	metrics.requestsQueueTimer = newHistogram(cfg, metrics.Registry,
+		"request_queue_time",
+		"Seconds request was waiting in queue",
+		[]string{requestTypeLabel, requestStatusLabel},
+		queuedRequestTimeBuckets)
+
 	preloadLabelValues(&metrics)
 
 	return &metrics
@@ -374,3 +387,14 @@ func (m *Metrics) RecordPrebidCacheRequestTime(success bool, length time.Duratio
 		successLabel: strconv.FormatBool(success),
 	}).Observe(length.Seconds())
 }
+
+func (m *Metrics) RecordRequestQueueTime(success bool, requestType pbsmetrics.RequestType, length time.Duration) {
+	successLabelFormatted := requestRejectLabel
+	if success {
+		successLabelFormatted = requestSuccessLabel
+	}
+	m.requestsQueueTimer.With(prometheus.Labels{
+		requestTypeLabel:   string(requestType),
+		requestStatusLabel: successLabelFormatted,
+	}).Observe(length.Seconds())
+}

pbsmetrics/prometheus/prometheus_test.go

@@ -881,6 +881,48 @@ func TestMetricAccumulationSpotCheck(t *testing.T) {
 		expectedValue)
 }
 
+func TestRecordRequestQueueTimeMetric(t *testing.T) {
+	performTest := func(m *Metrics, requestStatus bool, requestType pbsmetrics.RequestType, timeInSec float64) {
+		m.RecordRequestQueueTime(requestStatus, requestType, time.Duration(timeInSec*float64(time.Second)))
+	}
+
+	testCases := []struct {
+		description   string
+		status        string
+		testCase      func(m *Metrics)
+		expectedCount uint64
+		expectedSum   float64
+	}{
+		{
+			description: "Success",
+			status:      requestSuccessLabel,
+			testCase: func(m *Metrics) {
+				performTest(m, true, pbsmetrics.ReqTypeVideo, 2)
+			},
+			expectedCount: 1,
+			expectedSum:   2,
+		},
+		{
+			description: "TimeoutError",
+			status:      requestRejectLabel,
+			testCase: func(m *Metrics) {
+				performTest(m, false, pbsmetrics.ReqTypeVideo, 50)
+			},
+			expectedCount: 1,
+			expectedSum:   50,
+		},
+	}
+
+	m := createMetricsForTesting()
+	for _, test := range testCases {
+
+		test.testCase(m)
+
+		result := getHistogramFromHistogramVecByTwoKeys(m.requestsQueueTimer, requestTypeLabel, "video", requestStatusLabel, test.status)
+		assertHistogram(t, test.description, result, test.expectedCount, test.expectedSum)
+	}
+}
+
 func assertCounterValue(t *testing.T, description, name string, counter prometheus.Counter, expected float64) {
 	m := dto.Metric{}
 	counter.Write(&m)
@@ -906,6 +948,24 @@ func getHistogramFromHistogramVec(histogram *prometheus.HistogramVec, labelKey,
 	return result
 }
 
+func getHistogramFromHistogramVecByTwoKeys(histogram *prometheus.HistogramVec, label1Key, label1Value, label2Key, label2Value string) dto.Histogram {
+	var result dto.Histogram
+	processMetrics(histogram, func(m dto.Metric) {
+		for ind, label := range m.GetLabel() {
+			if label.GetName() == label1Key && label.GetValue() == label1Value {
+				valInd := ind
+				if ind == 1 {
+					valInd = 0
+				}
+				if m.Label[valInd].GetName() == label2Key && m.Label[valInd].GetValue() == label2Value {
+					result = *m.GetHistogram()
+				}
+			}
+		}
+	})
+	return result
+}
+
 func processMetrics(collector prometheus.Collector, handler func(m dto.Metric)) {
 	collectorChan := make(chan prometheus.Metric)
 	go func() {

router/aspects/request_timeout_handler.go

@@ -3,11 +3,13 @@ package aspects
 import (
 	"github.com/julienschmidt/httprouter"
 	"github.com/prebid/prebid-server/config"
+	"github.com/prebid/prebid-server/pbsmetrics"
 	"net/http"
 	"strconv"
+	"time"
 )
 
-func QueuedRequestTimeout(f httprouter.Handle, reqTimeoutHeaders config.RequestTimeoutHeaders) httprouter.Handle {
+func QueuedRequestTimeout(f httprouter.Handle, reqTimeoutHeaders config.RequestTimeoutHeaders, metricsEngine pbsmetrics.MetricsEngine, requestType pbsmetrics.RequestType) httprouter.Handle {
 
 	return func(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
 
@@ -30,13 +32,17 @@ func QueuedRequestTimeout(f httprouter.Handle, reqTimeoutHeaders config.RequestT
 			return
 		}
 
+		reqTimeDuration := time.Duration(reqTimeFloat * float64(time.Second))
+
 		//Return HTTP 408 if requests stays too long in queue
 		if reqTimeFloat >= reqTimeoutFloat {
 			w.WriteHeader(http.StatusRequestTimeout)
 			w.Write([]byte("Queued request processing time exceeded maximum"))
+			metricsEngine.RecordRequestQueueTime(false, requestType, reqTimeDuration)
 			return
 		}
 
+		metricsEngine.RecordRequestQueueTime(true, requestType, reqTimeDuration)
 		f(w, r, params)
 	}