support msp ab (#6)

* support msp ab

* rename ab buckets name

endpoints/openrtb2/auction.go

@@ -182,7 +182,7 @@ func (deps *endpointDeps) Auction(w http.ResponseWriter, r *http.Request, _ http
 		return
 	}
 
-	mspUpdateByImpInfo(&labels, mspLogInfo, mspExtOutput.StoredImp)
+	mspUpdateLogLabelsByExtOutput(&labels, mspLogInfo, mspExtOutput)
 	mspUpdateLogMsgByReq(mspLogInfo, req)
 
 	if rejectErr := hookexecution.FindFirstRejectOrNil(errL); rejectErr != nil {
@@ -235,6 +235,7 @@ func (deps *endpointDeps) Auction(w http.ResponseWriter, r *http.Request, _ http
 		PubID:                      labels.PubID,
 		HookExecutor:               deps.hookExecutor,
 		MspStoredImp:               mspExtOutput.StoredImp,
+		MspAbList:                  mspExtOutput.MspLayerAbBuckets,
 	}
 	response, err := deps.ex.HoldAuction(ctx, auctionRequest, nil)
 	ao.Request = req.BidRequest
@@ -435,6 +436,10 @@ func (deps *endpointDeps) parseRequest(httpRequest *http.Request, labels *metric
 		return
 	}
 
+	if requestJson, errs = mspExtOutput.processStoredAb(requestJson, deps.storedReqFetcher, ctx); len(errs) > 0 {
+		return
+	}
+
 	//Stored auction responses should be processed after stored requests due to possible impression modification
 	storedAuctionResponses, storedBidResponses, bidderImpReplaceImpId, errs = stored_responses.ProcessStoredResponses(ctx, requestJson, deps.storedRespFetcher, deps.bidderMap)
 	if len(errs) > 0 {

endpoints/openrtb2/auction_test.go

@@ -5293,6 +5293,10 @@ func (cf *mockStoredResponseFetcher) FetchRequests(ctx context.Context, requestI
 	return nil, nil, nil
 }
 
+func (cf *mockStoredResponseFetcher) FetchABs(ctx context.Context, bucketList []string) (bucketData map[string]json.RawMessage, errs []error) {
+	return nil, nil
+}
+
 func (cf *mockStoredResponseFetcher) FetchResponses(ctx context.Context, ids []string) (data map[string]json.RawMessage, errs []error) {
 	return cf.data, nil
 }

endpoints/openrtb2/msp_ext.go

@@ -1,20 +1,28 @@
 package openrtb2
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 
 	"github.com/prebid/openrtb/v17/openrtb2"
 	"github.com/prebid/prebid-server/metrics"
+	"github.com/prebid/prebid-server/msp/ab"
 	"github.com/prebid/prebid-server/openrtb_ext"
+	"github.com/prebid/prebid-server/stored_requests"
 	u "github.com/rjNemo/underscore"
+	jsonpatch "gopkg.in/evanphx/json-patch.v4"
 )
 
 const Unknown = "unknown"
 
 // This struct holds the extra output parsed out from request and needed by MSP
 type mspExtOutput struct {
 	StoredImp string
+	// ab buckets configured at ads layer
+	AdsLayerAbBuckets []string
+	// ab buckets configured at msp layer(server side)
+	MspLayerAbBuckets []string
 }
 
 func newMspExtOutput() *mspExtOutput {
@@ -32,6 +40,41 @@ func (m *mspExtOutput) extractStoredImp(impInfo []ImpExtPrebidData) {
 	}
 }
 
+func (m *mspExtOutput) processStoredAb(rawReq []byte, storedReqFetcher stored_requests.Fetcher, ctx context.Context) ([]byte, []error) {
+	resolvedReq := rawReq
+	req := &openrtb2.BidRequest{}
+	if err := json.Unmarshal(rawReq, req); err != nil {
+		return nil, []error{err}
+	}
+
+	if ab.Init && req.Device != nil && req.App != nil {
+		bucketList, err := ab.GetMspAB().GetBucket(req.User.ID, req.Device.OS, req.App.Ver)
+		if err != nil {
+			return nil, []error{err}
+		}
+
+		m.AdsLayerAbBuckets = bucketList
+		bucketList = append(bucketList, "all")
+		storedABs, _ := storedReqFetcher.FetchABs(ctx, bucketList)
+
+		for _, bucketName := range bucketList {
+			// only use configed experiments on server
+			val, ok := storedABs[bucketName]
+			if ok {
+				if len(m.MspLayerAbBuckets) > 0 && bucketName == "all" {
+					continue
+				}
+				resolvedReq, err = jsonpatch.MergePatch(resolvedReq, val)
+				m.MspLayerAbBuckets = append(m.MspLayerAbBuckets, bucketName)
+				if err != nil {
+					return nil, []error{err}
+				}
+			}
+		}
+	}
+	return resolvedReq, nil
+}
+
 type mspLogInfo struct {
 	Os         string
 	Idfa       string
@@ -50,6 +93,7 @@ type mspLogInfo struct {
 	HasBuyerId bool
 	Seat       string
 	Adomain    []string
+	AbBuckets  []string
 }
 
 func newMSPLogInfo() *mspLogInfo {
@@ -71,16 +115,20 @@ func newMSPLogInfo() *mspLogInfo {
 		HasBuyerId: false,
 		Seat:       Unknown,
 		Adomain:    []string{Unknown},
+		AbBuckets:  []string{Unknown},
 	}
 }
 
-func mspUpdateByImpInfo(labels *metrics.Labels, logInfo *mspLogInfo, storedImp string) {
-	if storedImp != "" {
-		logInfo.AdUnit = storedImp
-		labels.MSPStoredImp = storedImp
+func mspUpdateLogLabelsByExtOutput(labels *metrics.Labels, logInfo *mspLogInfo, extOut *mspExtOutput) {
+	if extOut.StoredImp != "" {
+		logInfo.AdUnit = extOut.StoredImp
+		labels.MSPStoredImp = extOut.StoredImp
 	} else {
 		labels.MSPStoredImp = Unknown
 	}
+
+	logInfo.AbBuckets = extOut.AdsLayerAbBuckets
+	labels.MspAbBuckets = extOut.MspLayerAbBuckets
 }
 
 func mspUpdateLogMsgByResponse(logInfo *mspLogInfo, response *openrtb2.BidResponse) {

endpoints/openrtb2/test_utils.go

@@ -843,6 +843,10 @@ func (cf mockStoredReqFetcher) FetchResponses(ctx context.Context, ids []string)
 	return nil, nil
 }
 
+func (cf *mockStoredReqFetcher) FetchABs(ctx context.Context, bucketList []string) (bucketData map[string]json.RawMessage, errs []error) {
+	return nil, nil
+}
+
 // mockExchange implements the Exchange interface
 type mockExchange struct {
 	lastRequest *openrtb2.BidRequest
@@ -1360,6 +1364,14 @@ func (cf *mockAmpStoredReqFetcher) FetchResponses(ctx context.Context, ids []str
 	return nil, nil
 }
 
+func (cf *mockAmpStoredReqFetcher) FetchABs(ctx context.Context, bucketList []string) (bucketData map[string]json.RawMessage, errs []error) {
+	return nil, nil
+}
+
+func (cf *mockAmpStoredResponseFetcher) FetchABs(ctx context.Context, bucketList []string) (bucketData map[string]json.RawMessage, errs []error) {
+	return nil, nil
+}
+
 type mockAmpStoredResponseFetcher struct {
 	data map[string]json.RawMessage
 }

endpoints/openrtb2/video_auction_test.go

@@ -1349,6 +1349,10 @@ func (cf mockVideoStoredReqFetcher) FetchResponses(ctx context.Context, ids []st
 	return nil, nil
 }
 
+func (cf mockVideoStoredReqFetcher) FetchABs(ctx context.Context, bucketList []string) (data map[string]json.RawMessage, errs []error) {
+	return nil, nil
+}
+
 type mockExchangeVideo struct {
 	lastRequest *openrtb2.BidRequest
 	cache       *mockCacheClient

exchange/exchange.go

@@ -205,6 +205,8 @@ type AuctionRequest struct {
 
 	// msp stored Imp
 	MspStoredImp string
+	// msp configed abs
+	MspAbList []string
 }
 
 // BidderRequest holds the bidder specific request and all other

exchange/msp_exchange.go

@@ -8,6 +8,7 @@ import (
 func addMspInfoToBidderRequest(bidderRequest []BidderRequest, r AuctionRequest) []BidderRequest {
 	return u.Map(bidderRequest, func(req BidderRequest) BidderRequest {
 		req.BidderLabels.MSPStoredImp = r.MspStoredImp
+		req.BidderLabels.MspAbBuckets = r.MspAbList
 		return req
 	})
 }
@@ -19,6 +20,7 @@ func (e *exchange) recordAdapterWinningInfo(auc *auction, r AuctionRequest) {
 				topBidderLabel := metrics.AdapterLabels{
 					Adapter:      bidderName,
 					MSPStoredImp: r.MspStoredImp,
+					MspAbBuckets: r.MspAbList,
 				}
 
 				if len(topBidPerBidder) > 0 {

metrics/metrics.go

@@ -16,6 +16,7 @@ type Labels struct {
 
 	// Following Fields are for MSP only.
 	MSPStoredImp string
+	MspAbBuckets []string
 }
 
 // AdapterLabels defines the labels that can be attached to the adapter metrics.
@@ -30,6 +31,7 @@ type AdapterLabels struct {
 
 	// Following Fields are for MSP only.
 	MSPStoredImp string
+	MspAbBuckets []string
 }
 
 // ImpLabels defines metric labels describing the impression type.

metrics/prometheus/msp_prometheus.go

@@ -9,6 +9,8 @@ import (
 const (
 	mspStoredImpUnknown = "unknown"
 	mspStoredImpLabel   = "stored_imp"
+	mspAbBucketUnknown  = "unknown"
+	mspAbLabel          = "ab_bucket"
 )
 
 func (m *Metrics) RecordAdapterWinningBidReceived(labels metrics.AdapterLabels, bidType openrtb_ext.BidType, hasAdm bool) {
@@ -17,16 +19,22 @@ func (m *Metrics) RecordAdapterWinningBidReceived(labels metrics.AdapterLabels,
 		markupDelivery = markupDeliveryAdm
 	}
 
-	m.mspAdapterWinningBids.With(prometheus.Labels{
-		adapterLabel:        string(labels.Adapter),
-		markupDeliveryLabel: markupDelivery,
-		mspStoredImpLabel:   labels.MSPStoredImp,
-	}).Inc()
+	for _, bucket := range labels.MspAbBuckets {
+		m.mspAdapterWinningBids.With(prometheus.Labels{
+			adapterLabel:        string(labels.Adapter),
+			markupDeliveryLabel: markupDelivery,
+			mspStoredImpLabel:   labels.MSPStoredImp,
+			mspAbLabel:          bucket,
+		}).Inc()
+	}
 }
 
 func (m *Metrics) RecordAdapterWinningPrice(labels metrics.AdapterLabels, cpm float64) {
-	m.mspAdapterWinningPrices.With(prometheus.Labels{
-		adapterLabel:      string(labels.Adapter),
-		mspStoredImpLabel: labels.MSPStoredImp,
-	}).Observe(cpm)
+	for _, bucket := range labels.MspAbBuckets {
+		m.mspAdapterWinningPrices.With(prometheus.Labels{
+			adapterLabel:      string(labels.Adapter),
+			mspStoredImpLabel: labels.MSPStoredImp,
+			mspAbLabel:        bucket,
+		}).Observe(cpm)
+	}
 }

metrics/prometheus/preload.go

@@ -24,6 +24,7 @@ func preloadLabelValues(m *Metrics, syncerKeys []string, moduleStageNames map[st
 		syncerSetsStatusValues    = syncerSetStatusesAsString()
 		sourceValues              = []string{sourceRequest}
 		mspStoredImpValues        = []string{mspStoredImpUnknown}
+		mspAbBucketValues         = []string{mspAbBucketUnknown}
 	)
 
 	preloadLabelValuesForCounter(m.connectionsError, map[string][]string{
@@ -53,11 +54,13 @@ func preloadLabelValues(m *Metrics, syncerKeys []string, moduleStageNames map[st
 		requestTypeLabel:   requestTypeValues,
 		requestStatusLabel: requestStatusValues,
 		mspStoredImpLabel:  mspStoredImpValues,
+		mspAbLabel:         mspAbBucketValues,
 	})
 
 	preloadLabelValuesForHistogram(m.requestsTimer, map[string][]string{
 		requestTypeLabel:  requestTypeValues,
 		mspStoredImpLabel: mspStoredImpValues,
+		mspAbLabel:        mspAbBucketValues,
 	})
 
 	preloadLabelValuesForHistogram(m.storedAccountFetchTimer, map[string][]string{
@@ -128,6 +131,7 @@ func preloadLabelValues(m *Metrics, syncerKeys []string, moduleStageNames map[st
 		adapterLabel:        adapterValues,
 		markupDeliveryLabel: bidTypeValues,
 		mspStoredImpLabel:   mspStoredImpValues,
+		mspAbLabel:          mspAbBucketValues,
 	})
 
 	preloadLabelValuesForCounter(m.adapterErrors, map[string][]string{
@@ -162,13 +166,15 @@ func preloadLabelValues(m *Metrics, syncerKeys []string, moduleStageNames map[st
 	preloadLabelValuesForHistogram(m.adapterPrices, map[string][]string{
 		adapterLabel:      adapterValues,
 		mspStoredImpLabel: mspStoredImpValues,
+		mspAbLabel:        mspAbBucketValues,
 	})
 
 	preloadLabelValuesForCounter(m.adapterRequests, map[string][]string{
 		adapterLabel:      adapterValues,
 		cookieLabel:       cookieValues,
 		hasBidsLabel:      boolValues,
 		mspStoredImpLabel: mspStoredImpValues,
+		mspAbLabel:        mspAbBucketValues,
 	})
 
 	preloadLabelValuesForCounter(m.adsCertRequests, map[string][]string{
@@ -273,11 +279,13 @@ func preloadLabelValues(m *Metrics, syncerKeys []string, moduleStageNames map[st
 		adapterLabel:        adapterValues,
 		markupDeliveryLabel: bidTypeValues,
 		mspStoredImpLabel:   mspStoredImpValues,
+		mspAbLabel:          mspAbBucketValues,
 	})
 
 	preloadLabelValuesForHistogram(m.mspAdapterWinningPrices, map[string][]string{
 		adapterLabel:      adapterValues,
 		mspStoredImpLabel: mspStoredImpValues,
+		mspAbLabel:        mspAbBucketValues,
 	})
 }