Metrics: Record incoming request sizes (#4495)

Author: guscarreon

endpoints/openrtb2/amp_auction.go

@@ -160,7 +160,7 @@ func (deps *endpointDeps) AmpAuction(w http.ResponseWriter, r *http.Request, _ h
 
 	// There is no body for AMP requests, so we pass a nil body and ignore the return value.
 	_, rejectErr := hookExecutor.ExecuteEntrypointStage(r, nilBody)
-	reqWrapper, storedAuctionResponses, storedBidResponses, bidderImpReplaceImp, errL := deps.parseAmpRequest(r)
+	reqWrapper, storedAuctionResponses, storedBidResponses, bidderImpReplaceImp, errL := deps.parseAmpRequest(r, labels)
 	ao.Errors = append(ao.Errors, errL...)
 	// Process reject after parsing amp request, so we can use reqWrapper.
 	// There is no body for AMP requests, so we pass a nil body and ignore the return value.
@@ -503,9 +503,9 @@ func getExtBidResponse(
 // possible, it will return errors with messages that suggest improvements.
 //
 // If the errors list has at least one element, then no guarantees are made about the returned request.
-func (deps *endpointDeps) parseAmpRequest(httpRequest *http.Request) (req *openrtb_ext.RequestWrapper, storedAuctionResponses stored_responses.ImpsWithBidResponses, storedBidResponses stored_responses.ImpBidderStoredResp, bidderImpReplaceImp stored_responses.BidderImpReplaceImpID, errs []error) {
+func (deps *endpointDeps) parseAmpRequest(httpRequest *http.Request, labels metrics.Labels) (req *openrtb_ext.RequestWrapper, storedAuctionResponses stored_responses.ImpsWithBidResponses, storedBidResponses stored_responses.ImpBidderStoredResp, bidderImpReplaceImp stored_responses.BidderImpReplaceImpID, errs []error) {
 	// Load the stored request for the AMP ID.
-	reqNormal, storedAuctionResponses, storedBidResponses, bidderImpReplaceImp, e := deps.loadRequestJSONForAmp(httpRequest)
+	reqNormal, storedAuctionResponses, storedBidResponses, bidderImpReplaceImp, e := deps.loadRequestJSONForAmp(httpRequest, labels)
 	if errs = append(errs, e...); errortypes.ContainsFatalError(errs) {
 		return
 	}
@@ -536,7 +536,7 @@ func (deps *endpointDeps) parseAmpRequest(httpRequest *http.Request) (req *openr
 }
 
 // Load the stored OpenRTB request for an incoming AMP request, or return the errors found.
-func (deps *endpointDeps) loadRequestJSONForAmp(httpRequest *http.Request) (req *openrtb2.BidRequest, storedAuctionResponses stored_responses.ImpsWithBidResponses, storedBidResponses stored_responses.ImpBidderStoredResp, bidderImpReplaceImp stored_responses.BidderImpReplaceImpID, errs []error) {
+func (deps *endpointDeps) loadRequestJSONForAmp(httpRequest *http.Request, labels metrics.Labels) (req *openrtb2.BidRequest, storedAuctionResponses stored_responses.ImpsWithBidResponses, storedBidResponses stored_responses.ImpBidderStoredResp, bidderImpReplaceImp stored_responses.BidderImpReplaceImpID, errs []error) {
 	req = &openrtb2.BidRequest{}
 	errs = nil
 
@@ -559,6 +559,7 @@ func (deps *endpointDeps) loadRequestJSONForAmp(httpRequest *http.Request) (req
 
 	// The fetched config becomes the entire OpenRTB request
 	requestJSON := storedRequests[ampParams.StoredRequestID]
+	labels.RequestSize = len(requestJSON)
 	if err := jsonutil.UnmarshalValid(requestJSON, req); err != nil {
 		errs = []error{err}
 		return

endpoints/openrtb2/auction.go

@@ -442,6 +442,7 @@ func (deps *endpointDeps) parseRequest(httpRequest *http.Request, labels *metric
 		errs = []error{err}
 		return
 	}
+	labels.RequestSize = len(requestJson)
 
 	if limitedReqReader.N <= 0 {
 		// Limited Reader returns 0 if the request was exactly at the max size or over the limit.

endpoints/openrtb2/video_auction.go

@@ -176,6 +176,7 @@ func (deps *endpointDeps) VideoAuctionEndpoint(w http.ResponseWriter, r *http.Re
 		handleError(&labels, w, []error{err}, &vo, &debugLog)
 		return
 	}
+	labels.RequestSize = len(requestJson)
 
 	resolvedRequest := requestJson
 	if debugLog.DebugEnabledOrOverridden {

metrics/config/metrics_test.go

@@ -53,6 +53,7 @@ func TestMultiMetricsEngine(t *testing.T) {
 		PubID:         "test1",
 		CookieFlag:    metrics.CookieFlagYes,
 		RequestStatus: metrics.RequestStatusOK,
+		RequestSize:   1024,
 	}
 	apnLabels := metrics.AdapterLabels{
 		Source:      metrics.DemandWeb,
@@ -157,6 +158,9 @@ func TestMultiMetricsEngine(t *testing.T) {
 	VerifyMetrics(t, "RequestStatuses.OpenRTB2.BadInput", goEngine.RequestStatuses[metrics.ReqTypeORTB2Web][metrics.RequestStatusBadInput].Count(), 0)
 	VerifyMetrics(t, "RequestStatuses.OpenRTB2.BlockedApp", goEngine.RequestStatuses[metrics.ReqTypeORTB2Web][metrics.RequestStatusBlockedApp].Count(), 0)
 
+	VerifyMetrics(t, "RequestSizeByEndpoint.Auction", goEngine.RequestSizeByEndpoint[metrics.EndpointAuction].Count(), 5)
+	VerifyMetrics(t, "RequestSizeByEndpoint.Auction", goEngine.RequestSizeByEndpoint[metrics.EndpointAuction].Max(), 1024)
+
 	VerifyMetrics(t, "ImpsTypeBanner", goEngine.ImpsTypeBanner.Count(), 5)
 	VerifyMetrics(t, "ImpsTypeVideo", goEngine.ImpsTypeVideo.Count(), 3)
 	VerifyMetrics(t, "ImpsTypeAudio", goEngine.ImpsTypeAudio.Count(), 5)

metrics/go_metrics.go

@@ -41,6 +41,7 @@ type Metrics struct {
 
 	// Metrics for OpenRTB requests specifically
 	RequestStatuses       map[RequestType]map[RequestStatus]metrics.Meter
+	RequestSizeByEndpoint map[EndpointType]metrics.Histogram
 	AmpNoCookieMeter      metrics.Meter
 	CookieSyncMeter       metrics.Meter
 	CookieSyncStatusMeter map[CookieSyncStatus]metrics.Meter
@@ -159,6 +160,7 @@ func NewBlankMetrics(registry metrics.Registry, exchanges []string, disabledMetr
 	newMetrics := &Metrics{
 		MetricsRegistry:                registry,
 		RequestStatuses:                make(map[RequestType]map[RequestStatus]metrics.Meter),
+		RequestSizeByEndpoint:          make(map[EndpointType]metrics.Histogram),
 		ConnectionCounter:              metrics.NilCounter{},
 		ConnectionAcceptErrorMeter:     blankMeter,
 		ConnectionCloseErrorMeter:      blankMeter,
@@ -235,6 +237,12 @@ func NewBlankMetrics(registry metrics.Registry, exchanges []string, disabledMetr
 		}
 	}
 
+	for _, t := range EndpointTypes() {
+		if t != EndpointAmp {
+			newMetrics.RequestSizeByEndpoint[t] = &metrics.NilHistogram{}
+		}
+	}
+
 	for _, c := range CacheResults() {
 		newMetrics.StoredReqCacheMeter[c] = blankMeter
 		newMetrics.StoredImpCacheMeter[c] = blankMeter
@@ -358,6 +366,10 @@ func NewMetrics(registry metrics.Registry, exchanges []openrtb_ext.BidderName, d
 		}
 	}
 
+	for _, endpoint := range EndpointTypes() {
+		newMetrics.RequestSizeByEndpoint[endpoint] = metrics.GetOrRegisterHistogram("requests.size."+string(endpoint), registry, metrics.NewUniformSample(1024))
+	}
+
 	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)
@@ -612,6 +624,11 @@ func (me *Metrics) RecordRequest(labels Labels) {
 		}
 	}
 
+	// Request size by endpoint
+	if labels.RequestSize > 0 && labels.RType != ReqTypeAMP {
+		me.RequestSizeByEndpoint[GetEndpointFromRequestType(labels.RType)].Update(int64(labels.RequestSize))
+	}
+
 	// Handle the account metrics now.
 	am := me.getAccountMetrics(labels.PubID)
 	am.requestMeter.Mark(1)

metrics/go_metrics_test.go

@@ -57,6 +57,8 @@ 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, "requests.size.auction", m.RequestSizeByEndpoint[EndpointAuction])
+	ensureContains(t, registry, "requests.size.video", m.RequestSizeByEndpoint[EndpointVideo])
 
 	ensureContains(t, registry, "queued_requests.video.rejected", m.RequestsQueueTimer[ReqTypeVideo][false])
 	ensureContains(t, registry, "queued_requests.video.accepted", m.RequestsQueueTimer[ReqTypeVideo][true])

metrics/metrics.go

@@ -13,6 +13,7 @@ type Labels struct {
 	PubID         string // exchange specific ID, so we cannot compile in values
 	CookieFlag    CookieFlag
 	RequestStatus RequestStatus
+	RequestSize   int
 }
 
 // AdapterLabels defines the labels that can be attached to the adapter metrics.
@@ -139,6 +140,8 @@ type DemandSource string
 // ImpMediaType : Media type described in the "imp" JSON object  TODO is this still needed?
 type ImpMediaType string
 
+type EndpointType string
+
 // RequestType : Request type enumeration
 type RequestType string
 
@@ -196,6 +199,40 @@ func RequestTypes() []RequestType {
 	}
 }
 
+// Endpoint type constants
+const (
+	EndpointAuction EndpointType = "auction"
+	EndpointVideo   EndpointType = "video"
+	EndpointAmp     EndpointType = "amp"
+)
+
+func EndpointTypes() []EndpointType {
+	return []EndpointType{
+		EndpointAuction,
+		EndpointVideo,
+		EndpointAmp,
+	}
+}
+
+func GetEndpointFromRequestType(requestType RequestType) EndpointType {
+	var requestEndpoint EndpointType
+
+	switch requestType {
+	case ReqTypeORTB2Web:
+		fallthrough
+	case ReqTypeORTB2App:
+		fallthrough
+	case ReqTypeORTB2DOOH:
+		requestEndpoint = EndpointAuction
+	case ReqTypeAMP:
+		requestEndpoint = EndpointAmp
+	case ReqTypeVideo:
+		requestEndpoint = EndpointVideo
+	}
+
+	return requestEndpoint
+}
+
 // The media types described in the "imp" json objects
 const (
 	ImpTypeBanner ImpMediaType = "banner"

metrics/metrics_test.go

@@ -0,0 +1,49 @@
+package metrics
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetEndpointFromRequestType(t *testing.T) {
+	testCases := []struct {
+		name          string
+		inRequestType RequestType
+		expected      EndpointType
+	}{
+		{
+			name:          "request-type-openrtb2-web",
+			inRequestType: ReqTypeORTB2Web,
+			expected:      EndpointAuction,
+		},
+		{
+			name:          "request-type-openrtb2-app",
+			inRequestType: ReqTypeORTB2App,
+			expected:      EndpointAuction,
+		},
+		{
+			name:          "request-type-openrtb2-dooh",
+			inRequestType: ReqTypeORTB2DOOH,
+			expected:      EndpointAuction,
+		},
+		{
+			name:          "request-type-amp",
+			inRequestType: ReqTypeAMP,
+			expected:      EndpointAmp,
+		},
+		{
+			name:          "request-type-video",
+			inRequestType: ReqTypeVideo,
+			expected:      EndpointVideo,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			out := GetEndpointFromRequestType(tc.inRequestType)
+
+			assert.Equal(t, tc.expected, out)
+		})
+	}
+}

metrics/prometheus/prometheus.go

@@ -30,6 +30,7 @@ type Metrics struct {
 	impressions                  *prometheus.CounterVec
 	prebidCacheWriteTimer        *prometheus.HistogramVec
 	requests                     *prometheus.CounterVec
+	requestsSize                 *prometheus.HistogramVec
 	debugRequests                prometheus.Counter
 	requestsTimer                *prometheus.HistogramVec
 	requestsQueueTimer           *prometheus.HistogramVec
@@ -127,6 +128,7 @@ const (
 	privacyBlockedLabel  = "privacy_blocked"
 	requestStatusLabel   = "request_status"
 	requestTypeLabel     = "request_type"
+	requestEndpointLabel = "request_size"
 	stageLabel           = "stage"
 	statusLabel          = "status"
 	successLabel         = "success"
@@ -171,6 +173,7 @@ func NewMetrics(cfg config.PrometheusMetrics, disabledMetrics config.DisabledMet
 	priceBuckets := []float64{250, 500, 750, 1000, 1500, 2000, 2500, 3000, 3500, 4000}
 	queuedRequestTimeBuckets := []float64{0, 1, 5, 30, 60, 120, 180, 240, 300}
 	overheadTimeBuckets := []float64{0.05, 0.06, 0.07, 0.08, 0.09, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1}
+	requestSizeBuckets := []float64{100, 500, 750, 1000, 2000, 4000, 7000, 10000, 15000, 20000, 50000, 75000}
 
 	metrics := Metrics{}
 	reg := prometheus.NewRegistry()
@@ -227,6 +230,11 @@ func NewMetrics(cfg config.PrometheusMetrics, disabledMetrics config.DisabledMet
 		"Count of total requests to Prebid Server labeled by type and status.",
 		[]string{requestTypeLabel, requestStatusLabel})
 
+	metrics.requestsSize = newHistogramVec(cfg, reg,
+		"request_size_bytes",
+		"Count that keeps track of incoming request size in bytes labeled by endpoint.",
+		[]string{requestEndpointLabel}, requestSizeBuckets)
+
 	metrics.debugRequests = newCounterWithoutLabels(cfg, reg,
 		"debug_requests",
 		"Count of total requests to Prebid Server that have debug enabled")
@@ -682,6 +690,13 @@ func (m *Metrics) RecordRequest(labels metrics.Labels) {
 		requestStatusLabel: string(labels.RequestStatus),
 	}).Inc()
 
+	if labels.RequestSize > 0 && labels.RType != metrics.ReqTypeAMP {
+		endpoint := metrics.GetEndpointFromRequestType(labels.RType)
+		m.requestsSize.With(prometheus.Labels{
+			requestEndpointLabel: string(endpoint),
+		}).Observe(float64(labels.RequestSize))
+	}
+
 	if labels.CookieFlag == metrics.CookieFlagNo {
 		m.requestsWithoutCookie.With(prometheus.Labels{
 			requestTypeLabel: string(labels.RType),

metrics/prometheus/prometheus_test.go

@@ -166,6 +166,7 @@ func TestRequestMetric(t *testing.T) {
 	m.RecordRequest(metrics.Labels{
 		RType:         requestType,
 		RequestStatus: requestStatus,
+		RequestSize:   1024,
 	})
 
 	expectedCount := float64(1)
@@ -175,6 +176,9 @@ func TestRequestMetric(t *testing.T) {
 			requestTypeLabel:   string(requestType),
 			requestStatusLabel: string(requestStatus),
 		})
+
+	histogram := getHistogramFromHistogramVec(m.requestsSize, requestEndpointLabel, string(metrics.EndpointAuction))
+	assertHistogram(t, "requests_size_auction", histogram, 1, 1024)
 }
 
 func TestDebugRequestMetric(t *testing.T) {