Add support for Account configuration (PBID-727, prebid#1395) (prebid#1426)

Author: laurb9

analytics/core.go

@@ -2,6 +2,7 @@ package analytics
 
 import (
 	"github.com/mxmCherry/openrtb"
+	"github.com/prebid/prebid-server/config"
 	"github.com/prebid/prebid-server/openrtb_ext"
 	"github.com/prebid/prebid-server/usersync"
 )
@@ -28,6 +29,7 @@ type AuctionObject struct {
 	Errors   []error
 	Request  *openrtb.BidRequest
 	Response *openrtb.BidResponse
+	Account  *config.Account
 }
 
 //Loggable object of a transaction at /openrtb2/amp endpoint

config/accounts.go

@@ -0,0 +1,8 @@
+package config
+
+// Account represents a publisher account configuration
+type Account struct {
+	ID       string      `mapstructure:"id" json:"id"`
+	Disabled bool        `mapstructure:"disabled" json:"disabled"`
+	CacheTTL DefaultTTLs `mapstructure:"cache_ttl" json:"cache_ttl"`
+}

config/config.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"bytes"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net/url"
@@ -40,6 +41,7 @@ type Configuration struct {
 	StoredRequests    StoredRequests  `mapstructure:"stored_requests"`
 	StoredRequestsAMP StoredRequests  `mapstructure:"stored_amp_req"`
 	CategoryMapping   StoredRequests  `mapstructure:"category_mapping"`
+	Accounts          StoredRequests  `mapstructure:"accounts"`
 	// Note that StoredVideo refers to stored video requests, and has nothing to do with caching video creatives.
 	StoredVideo StoredRequests `mapstructure:"stored_video_req"`
 
@@ -65,6 +67,11 @@ type Configuration struct {
 	BlacklistedAcctMap map[string]bool
 	// Is publisher/account ID required to be submitted in the OpenRTB2 request
 	AccountRequired bool `mapstructure:"account_required"`
+	// AccountDefaults defines default settings for valid accounts that are partially defined
+	// and provides a way to set global settings that can be overridden at account level.
+	AccountDefaults Account `mapstructure:"account_defaults"`
+	// accountDefaultsJSON is the internal serialized form of AccountDefaults used for json merge
+	accountDefaultsJSON json.RawMessage
 	// Local private file containing SSL certificates
 	PemCertsFile string `mapstructure:"certificates_file"`
 	// Custom headers to handle request timeouts from queueing infrastructure
@@ -106,10 +113,11 @@ func (c configErrors) Error() string {
 func (cfg *Configuration) validate() configErrors {
 	var errs configErrors
 	errs = cfg.AuctionTimeouts.validate(errs)
-	errs = cfg.StoredRequests.validate("stored_req", errs)
-	errs = cfg.StoredRequestsAMP.validate("stored_amp_req", errs)
-	errs = cfg.CategoryMapping.validate("categories", errs)
-	errs = cfg.StoredVideo.validate("stored_video_req", errs)
+	errs = cfg.StoredRequests.validate(errs)
+	errs = cfg.StoredRequestsAMP.validate(errs)
+	errs = cfg.Accounts.validate(errs)
+	errs = cfg.CategoryMapping.validate(errs)
+	errs = cfg.StoredVideo.validate(errs)
 	errs = cfg.Metrics.validate(errs)
 	if cfg.MaxRequestSize < 0 {
 		errs = append(errs, fmt.Errorf("cfg.max_request_size must be >= 0. Got %d", cfg.MaxRequestSize))
@@ -119,6 +127,9 @@ func (cfg *Configuration) validate() configErrors {
 	errs = validateAdapters(cfg.Adapters, errs)
 	errs = cfg.Debug.validate(errs)
 	errs = cfg.ExtCacheURL.validate(errs)
+	if cfg.AccountDefaults.Disabled {
+		glog.Warning(`With account_defaults.disabled=true, host-defined accounts must exist and have "disabled":false. All other requests will be rejected.`)
+	}
 	return errs
 }
 
@@ -589,6 +600,12 @@ func New(v *viper.Viper) (*Configuration, error) {
 		return nil, err
 	}
 
+	// Update account defaults and generate base json for patch
+	c.AccountDefaults.CacheTTL = c.CacheURL.DefaultTTLs // comment this out to set explicitly in config
+	if err := c.MarshalAccountDefaults(); err != nil {
+		return nil, err
+	}
+
 	// To look for a request's publisher_id in the NonStandardPublishers list in
 	// O(1) time, we fill this hash table located in the NonStandardPublisherMap field of GDPR
 	c.GDPR.NonStandardPublisherMap = make(map[string]int)
@@ -622,6 +639,20 @@ func New(v *viper.Viper) (*Configuration, error) {
 	return &c, nil
 }
 
+// MarshalAccountDefaults compiles AccountDefaults into the JSON format used for merge patch
+func (cfg *Configuration) MarshalAccountDefaults() error {
+	var err error
+	if cfg.accountDefaultsJSON, err = json.Marshal(cfg.AccountDefaults); err != nil {
+		glog.Warningf("converting %+v to json: %v", cfg.AccountDefaults, err)
+	}
+	return err
+}
+
+// AccountDefaultsJSON returns the precompiled JSON form of account_defaults
+func (cfg *Configuration) AccountDefaultsJSON() json.RawMessage {
+	return cfg.accountDefaultsJSON
+}
+
 //Allows for protocol relative URL if scheme is empty
 func (cfg *Cache) GetBaseURL() string {
 	cfg.Scheme = strings.ToLower(cfg.Scheme)
@@ -843,6 +874,10 @@ func SetupViper(v *viper.Viper, filename string) {
 	v.SetDefault("stored_video_req.http_events.refresh_rate_seconds", 0)
 	v.SetDefault("stored_video_req.http_events.timeout_ms", 0)
 
+	v.SetDefault("accounts.filesystem.enabled", false)
+	v.SetDefault("accounts.filesystem.directorypath", "./stored_requests/data/by_id")
+	v.SetDefault("accounts.in_memory_cache.type", "none")
+
 	for _, bidder := range openrtb_ext.BidderMap {
 		setBidderDefaults(v, strings.ToLower(string(bidder)))
 	}
@@ -976,6 +1011,7 @@ func SetupViper(v *viper.Viper, filename string) {
 	v.SetDefault("blacklisted_apps", []string{""})
 	v.SetDefault("blacklisted_accts", []string{""})
 	v.SetDefault("account_required", false)
+	v.SetDefault("account_defaults.disabled", false)
 	v.SetDefault("certificates_file", "")
 	v.SetDefault("auto_gen_source_tid", true)
 

config/config_test.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"bytes"
+	"errors"
 	"net"
 	"os"
 	"strings"
@@ -466,6 +467,10 @@ func TestValidConfig(t *testing.T) {
 		CategoryMapping: StoredRequests{
 			Files: FileFetcherConfig{Enabled: true},
 		},
+		Accounts: StoredRequests{
+			Files:         FileFetcherConfig{Enabled: true},
+			InMemoryCache: InMemoryCache{Type: "none"},
+		},
 	}
 
 	resolvedStoredRequestsConfig(&cfg)
@@ -640,6 +645,18 @@ func TestValidateDebug(t *testing.T) {
 	assert.NotNil(t, err, "cfg.debug.timeout_notification.sampling_rate should not be allowed to be greater than 1.0, but it was allowed")
 }
 
+func TestValidateAccountsConfigRestrictions(t *testing.T) {
+	cfg := newDefaultConfig(t)
+	cfg.Accounts.Files.Enabled = true
+	cfg.Accounts.HTTP.Endpoint = "http://localhost"
+	cfg.Accounts.Postgres.ConnectionInfo.Database = "accounts"
+
+	errs := cfg.validate()
+	assert.Len(t, errs, 2)
+	assert.Contains(t, errs, errors.New("accounts.http: retrieving accounts via http not available, use accounts.files"))
+	assert.Contains(t, errs, errors.New("accounts.postgres: retrieving accounts via postgres not available, use accounts.files"))
+}
+
 func newDefaultConfig(t *testing.T) *Configuration {
 	v := viper.New()
 	SetupViper(v, "")

config/stored_requests.go

@@ -18,8 +18,20 @@ const (
 	CategoryDataType   DataType = "Category"
 	VideoDataType      DataType = "Video"
 	AMPRequestDataType DataType = "AMP Request"
+	AccountDataType    DataType = "Account"
 )
 
+// Section returns the config section this type is defined in
+func (sr *StoredRequests) Section() string {
+	return map[DataType]string{
+		RequestDataType:    "stored_requests",
+		CategoryDataType:   "categories",
+		VideoDataType:      "stored_video_req",
+		AMPRequestDataType: "stored_amp_req",
+		AccountDataType:    "accounts",
+	}[sr.dataType]
+}
+
 func (sr *StoredRequests) DataType() DataType {
 	return sr.dataType
 }
@@ -109,34 +121,42 @@ func resolvedStoredRequestsConfig(cfg *Configuration) {
 	cfg.StoredRequestsAMP.dataType = AMPRequestDataType
 	cfg.StoredVideo.dataType = VideoDataType
 	cfg.CategoryMapping.dataType = CategoryDataType
+	cfg.Accounts.dataType = AccountDataType
 	return
 }
 
-func (cfg *StoredRequests) validate(section string, errs configErrors) configErrors {
-	errs = cfg.Postgres.validate(section, errs)
+func (cfg *StoredRequests) validate(errs configErrors) configErrors {
+	if cfg.DataType() == AccountDataType && cfg.HTTP.Endpoint != "" {
+		errs = append(errs, fmt.Errorf("%s.http: retrieving accounts via http not available, use accounts.files", cfg.Section()))
+	}
+	if cfg.DataType() == AccountDataType && cfg.Postgres.ConnectionInfo.Database != "" {
+		errs = append(errs, fmt.Errorf("%s.postgres: retrieving accounts via postgres not available, use accounts.files", cfg.Section()))
+	} else {
+		errs = cfg.Postgres.validate(cfg.Section(), errs)
+	}
 
 	// Categories do not use cache so none of the following checks apply
-	if cfg.dataType == CategoryDataType {
+	if cfg.DataType() == CategoryDataType {
 		return errs
 	}
 
 	if cfg.InMemoryCache.Type == "none" {
 		if cfg.CacheEvents.Enabled {
-			errs = append(errs, fmt.Errorf("%s: cache_events must be disabled if in_memory_cache=none", section))
+			errs = append(errs, fmt.Errorf("%s: cache_events must be disabled if in_memory_cache=none", cfg.Section()))
 		}
 
 		if cfg.HTTPEvents.RefreshRate != 0 {
-			errs = append(errs, fmt.Errorf("%s: http_events.refresh_rate_seconds must be 0 if in_memory_cache=none", section))
+			errs = append(errs, fmt.Errorf("%s: http_events.refresh_rate_seconds must be 0 if in_memory_cache=none", cfg.Section()))
 		}
 
 		if cfg.Postgres.PollUpdates.Query != "" {
-			errs = append(errs, fmt.Errorf("%s: postgres.poll_for_updates.query must be empty if in_memory_cache=none", section))
+			errs = append(errs, fmt.Errorf("%s: postgres.poll_for_updates.query must be empty if in_memory_cache=none", cfg.Section()))
 		}
 		if cfg.Postgres.CacheInitialization.Query != "" {
-			errs = append(errs, fmt.Errorf("%s: postgres.initialize_caches.query must be empty if in_memory_cache=none", section))
+			errs = append(errs, fmt.Errorf("%s: postgres.initialize_caches.query must be empty if in_memory_cache=none", cfg.Section()))
 		}
 	}
-	errs = cfg.InMemoryCache.validate(section, errs)
+	errs = cfg.InMemoryCache.validate(cfg.Section(), errs)
 	return errs
 }
 

endpoints/openrtb2/amp_auction.go

@@ -44,6 +44,7 @@ func NewAmpEndpoint(
 	ex exchange.Exchange,
 	validator openrtb_ext.BidderParamValidator,
 	requestsById stored_requests.Fetcher,
+	accounts stored_requests.AccountFetcher,
 	categories stored_requests.CategoryFetcher,
 	cfg *config.Configuration,
 	met pbsmetrics.MetricsEngine,
@@ -53,7 +54,7 @@ func NewAmpEndpoint(
 	bidderMap map[string]openrtb_ext.BidderName,
 ) (httprouter.Handle, error) {
 
-	if ex == nil || validator == nil || requestsById == nil || cfg == nil || met == nil {
+	if ex == nil || validator == nil || requestsById == nil || accounts == nil || cfg == nil || met == nil {
 		return nil, errors.New("NewAmpEndpoint requires non-nil arguments.")
 	}
 
@@ -69,6 +70,7 @@ func NewAmpEndpoint(
 		validator,
 		requestsById,
 		empty_fetcher.EmptyFetcher{},
+		accounts,
 		categories,
 		cfg,
 		met,
@@ -155,26 +157,30 @@ func (deps *endpointDeps) AmpAuction(w http.ResponseWriter, r *http.Request, _ h
 		labels.CookieFlag = pbsmetrics.CookieFlagYes
 	}
 	labels.PubID = getAccountID(req.Site.Publisher)
-
-	// Blacklist account now that we have resolved the value
-	if acctIdErr := validateAccount(deps.cfg, labels.PubID); acctIdErr != nil {
-		errL = append(errL, acctIdErr)
-		errCode := errortypes.ReadCode(acctIdErr)
-		if errCode == errortypes.BlacklistedAppErrorCode || errCode == errortypes.BlacklistedAcctErrorCode {
-			w.WriteHeader(http.StatusServiceUnavailable)
-			labels.RequestStatus = pbsmetrics.RequestStatusBlacklisted
-		} else {
-			w.WriteHeader(http.StatusBadRequest)
-			labels.RequestStatus = pbsmetrics.RequestStatusBadInput
+	// Look up account now that we have resolved the pubID value
+	account, acctIDErrs := deps.getAccount(ctx, labels.PubID)
+	if len(acctIDErrs) > 0 {
+		errL = append(errL, acctIDErrs...)
+		httpStatus := http.StatusBadRequest
+		metricsStatus := pbsmetrics.RequestStatusBadInput
+		for _, er := range errL {
+			errCode := errortypes.ReadCode(er)
+			if errCode == errortypes.BlacklistedAppErrorCode || errCode == errortypes.BlacklistedAcctErrorCode {
+				httpStatus = http.StatusServiceUnavailable
+				metricsStatus = pbsmetrics.RequestStatusBlacklisted
+				break
+			}
 		}
+		w.WriteHeader(httpStatus)
+		labels.RequestStatus = metricsStatus
 		for _, err := range errortypes.FatalOnly(errL) {
 			w.Write([]byte(fmt.Sprintf("Invalid request format: %s\n", err.Error())))
 		}
-		ao.Errors = append(ao.Errors, acctIdErr)
+		ao.Errors = append(ao.Errors, acctIDErrs...)
 		return
 	}
 
-	response, err := deps.ex.HoldAuction(ctx, req, usersyncs, labels, &deps.categories, nil)
+	response, err := deps.ex.HoldAuction(ctx, req, usersyncs, labels, account, &deps.categories, nil)
 	ao.AuctionResponse = response
 
 	if err != nil {