feat: Add new logger module - Pubstack Analytics Module (prebid#1331)

* Pubstack Analytics V1 (#11)

* V1 Pubstack (#7)

* feat: Add Pubstack Logger (#6)

* first version of pubstack analytics

* bypass viperconfig

* commit #1

* gofmt

* update configuration and make the tests pass

* add readme on how to configure the adapter and update the network calls

* update logging and fix intake url definition

* feat: Pubstack Analytics Connector

* fixing go mod

* fix: bad behaviour on appending path to auction url

* add buffering

* support bootstyrap like configuration

* implement route for all the objects

* supports termination signal handling for goroutines

* move readme to the correct location

* wording

* enable configuration reload + add tests

* fix logs messages

* fix tests

* fix log line

* conclude merge

* merge

* update go mod

Co-authored-by: Amaury Ravanel <amaury.ravanel@gmail.com>

* fix duplicated channel keys

Co-authored-by: Amaury Ravanel <amaury.ravanel@gmail.com>

* first pass - PR reviews

* rename channel* -> eventChannel

* dead code

* Review (#10)

* use json.Decoder

* update documentation

* use nil instead []byte("")

* clean code

* do not use http.DefaultClient

* fix race condition (need validation)

* separate the sender and buffer logics

* refactor the default configuration

* remove error counter

* Review GP + AR

* updating default config

* add more logs

* remove alias fields in json

* fix json serializer

* close event channels

Co-authored-by: Amaury Ravanel <amaury.ravanel@gmail.com>

* fix race condition

* first pass (pr reviews)

* refactor: store enabled modules into a dedicated struct

* stop goroutine

* test: improve coverage

* PR Review

* Revert "refactor: store enabled modules into a dedicated struct"

This reverts commit f57d9d6.

# Conflicts:
#	analytics/config/config_test.go

Co-authored-by: Amaury Ravanel <amaury.ravanel@gmail.com>

analytics/clients/http.go

@@ -0,0 +1,12 @@
+package clients
+
+import (
+	"net/http"
+)
+
+var defaultHttpInstance = http.DefaultClient
+
+func GetDefaultHttpInstance() *http.Client {
+	// TODO 2020-06-22 @see https://github.com/prebid/prebid-server/pull/1331#discussion_r436110097
+	return defaultHttpInstance
+}

analytics/config/config.go

@@ -3,7 +3,9 @@ package config
 import (
 	"github.com/golang/glog"
 	"github.com/prebid/prebid-server/analytics"
+	"github.com/prebid/prebid-server/analytics/clients"
 	"github.com/prebid/prebid-server/analytics/filesystem"
+	"github.com/prebid/prebid-server/analytics/pubstack"
 	"github.com/prebid/prebid-server/config"
 )
 
@@ -17,6 +19,21 @@ func NewPBSAnalytics(analytics *config.Analytics) analytics.PBSAnalyticsModule {
 			glog.Fatalf("Could not initialize FileLogger for file %v :%v", analytics.File.Filename, err)
 		}
 	}
+	if analytics.Pubstack.Enabled {
+		pubstackModule, err := pubstack.NewPubstackModule(
+			clients.GetDefaultHttpInstance(),
+			analytics.Pubstack.ScopeId,
+			analytics.Pubstack.IntakeUrl,
+			analytics.Pubstack.ConfRefresh,
+			analytics.Pubstack.Buffers.EventCount,
+			analytics.Pubstack.Buffers.BufferSize,
+			analytics.Pubstack.Buffers.Timeout)
+		if err == nil {
+			modules = append(modules, pubstackModule)
+		} else {
+			glog.Errorf("Could not initialize PubstackModule: %v", err)
+		}
+	}
 	return modules
 }
 

analytics/config/config_test.go

@@ -1,6 +1,7 @@
 package config
 
 import (
+	"github.com/stretchr/testify/assert"
 	"net/http"
 	"os"
 	"testing"
@@ -73,6 +74,13 @@ func initAnalytics(count *int) analytics.PBSAnalyticsModule {
 }
 
 func TestNewPBSAnalytics(t *testing.T) {
+	pbsAnalytics := NewPBSAnalytics(&config.Analytics{})
+	instance := pbsAnalytics.(enabledAnalytics)
+
+	assert.Equal(t, len(instance), 0)
+}
+
+func TestNewPBSAnalytics_FileLogger(t *testing.T) {
 	if _, err := os.Stat(TEST_DIR); os.IsNotExist(err) {
 		if err = os.MkdirAll(TEST_DIR, 0755); err != nil {
 			t.Fatalf("Could not create test directory for FileLogger")
@@ -88,4 +96,37 @@ func TestNewPBSAnalytics(t *testing.T) {
 	default:
 		t.Fatalf("Failed to initialize analytics module")
 	}
+
+	pbsAnalytics := NewPBSAnalytics(&config.Analytics{File: config.FileLogs{Filename: TEST_DIR + "/test"}})
+	instance := pbsAnalytics.(enabledAnalytics)
+
+	assert.Equal(t, len(instance), 1)
+}
+
+func TestNewPBSAnalytics_Pubstack(t *testing.T) {
+
+	pbsAnalyticsWithoutError := NewPBSAnalytics(&config.Analytics{
+		Pubstack: config.Pubstack{
+			Enabled:   true,
+			ScopeId:   "scopeId",
+			IntakeUrl: "https://pubstack.io/intake",
+			Buffers: config.PubstackBuffer{
+				BufferSize: "100KB",
+				EventCount: 0,
+				Timeout:    "30s",
+			},
+			ConfRefresh: "2h",
+		},
+	})
+	instanceWithoutError := pbsAnalyticsWithoutError.(enabledAnalytics)
+
+	assert.Equal(t, len(instanceWithoutError), 1)
+
+	pbsAnalyticsWithError := NewPBSAnalytics(&config.Analytics{
+		Pubstack: config.Pubstack{
+			Enabled: true,
+		},
+	})
+	instanceWithError := pbsAnalyticsWithError.(enabledAnalytics)
+	assert.Equal(t, len(instanceWithError), 0)
 }

analytics/pubstack/README.md

@@ -0,0 +1,28 @@
+# Pubstack Analytics
+
+In order to use the pubstack analytics module, it needs to be configured by the host.
+
+You can configure the server using the following environment variables:
+
+```bash
+export PBS_ANALYTICS_PUBSTACK_ENABLED="true"
+export PBS_ANALYTICS_PUBSTACK_ENDPOINT="https://openrtb.preview.pubstack.io/v1/openrtb2"
+export PBS_ANALYTICS_PUBSTACK_SCOPEID=<your scopeId here> # should be an UUIDv4
+```
+
+Or using the pbs configuration file and by appending the following block:
+
+```yaml
+analytics:
+    pubstack:
+      # Required properties
+      enabled: true
+      endpoint: "https://openrtb.preview.pubstack.io/v1/openrtb2"
+      scopeid: "<scopeId>" # The scopeId provided by the Pubstack Support Team
+      # Optional properties (advanced configuration)
+      configuration_refresh_delay: "2h" # Dynamic configuration delay
+      buffers: # Flush events to Pubstack when (first condition reached)
+        size: "2MB" # greater than 2MB
+        count : 100 # greater than 100 events
+        timeout: "15m" # greater than 15 minutes
+```

analytics/pubstack/config.go

@@ -0,0 +1,51 @@
+package pubstack
+
+import (
+	"encoding/json"
+	"github.com/docker/go-units"
+	"net/http"
+	"net/url"
+	"time"
+)
+
+func fetchConfig(client *http.Client, endpoint *url.URL) (*Configuration, error) {
+
+	res, err := client.Get(endpoint.String())
+	if err != nil {
+		return nil, err
+	}
+
+	defer res.Body.Close()
+	c := Configuration{}
+	err = json.NewDecoder(res.Body).Decode(&c)
+	if err != nil {
+		return nil, err
+	}
+	return &c, nil
+}
+
+func newBufferConfig(count int, size, duration string) (*bufferConfig, error) {
+	pDuration, err := time.ParseDuration(duration)
+	if err != nil {
+		return nil, err
+	}
+	pSize, err := units.FromHumanSize(size)
+	if err != nil {
+		return nil, err
+	}
+	return &bufferConfig{
+		pDuration,
+		int64(count),
+		pSize,
+	}, nil
+}
+
+func (a *Configuration) isSameAs(b *Configuration) bool {
+	sameEndpoint := a.Endpoint == b.Endpoint
+	sameScopeID := a.ScopeID == b.ScopeID
+	sameFeature := len(a.Features) == len(b.Features)
+	for key := range a.Features {
+		sameFeature = sameFeature && a.Features[key] == b.Features[key]
+	}
+	return sameFeature && sameEndpoint && sameScopeID
+}

analytics/pubstack/config_test.go

@@ -0,0 +1,102 @@
+package pubstack
+
+import (
+	"github.com/stretchr/testify/assert"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+)
+
+func TestFetchConfig(t *testing.T) {
+	configResponse := `{
+		"scopeId":  "scopeId",
+		"endpoint": "https://pubstack.io",
+		"features": {
+			"auction":    true,
+			"cookiesync": true,
+			"amp":        true,
+			"setuid":     false,
+			"video":      false
+		}
+	}`
+
+	server := httptest.NewServer(http.HandlerFunc(func(res http.ResponseWriter, req *http.Request) {
+		defer req.Body.Close()
+		res.Write([]byte(configResponse))
+		res.WriteHeader(200)
+	}))
+
+	defer server.Close()
+
+	endpoint, _ := url.Parse(server.URL)
+	cfg, _ := fetchConfig(server.Client(), endpoint)
+
+	assert.Equal(t, cfg.ScopeID, "scopeId")
+	assert.Equal(t, cfg.Endpoint, "https://pubstack.io")
+	assert.Equal(t, cfg.Features[auction], true)
+	assert.Equal(t, cfg.Features[cookieSync], true)
+	assert.Equal(t, cfg.Features[amp], true)
+	assert.Equal(t, cfg.Features[setUID], false)
+	assert.Equal(t, cfg.Features[video], false)
+}
+
+func TestFetchConfig_Error(t *testing.T) {
+	configResponse := `{
+		"error":  "scopeId",
+	}`
+
+	server := httptest.NewServer(http.HandlerFunc(func(res http.ResponseWriter, req *http.Request) {
+		defer req.Body.Close()
+		res.Write([]byte(configResponse))
+		res.WriteHeader(200)
+	}))
+
+	defer server.Close()
+
+	endpoint, _ := url.Parse(server.URL)
+	cfg, err := fetchConfig(server.Client(), endpoint)
+
+	assert.Nil(t, cfg)
+	assert.NotNil(t, err)
+}
+
+func TestIsSameAs(t *testing.T) {
+	copyConfig := func(conf Configuration) *Configuration {
+		newConfig := conf
+		newConfig.Features = make(map[string]bool)
+		for k := range conf.Features {
+			newConfig.Features[k] = conf.Features[k]
+		}
+		return &newConfig
+	}
+
+	a := &Configuration{
+		ScopeID:  "scopeId",
+		Endpoint: "endpoint",
+		Features: map[string]bool{
+			"auction":    true,
+			"cookiesync": true,
+			"amp":        true,
+			"setuid":     false,
+			"video":      false,
+		},
+	}
+
+	assert.True(t, a.isSameAs(copyConfig(*a)))
+
+	b := copyConfig(*a)
+	b.ScopeID = "anotherId"
+	assert.False(t, a.isSameAs(b))
+
+	b = copyConfig(*a)
+	b.Endpoint = "anotherEndpoint"
+	assert.False(t, a.isSameAs(b))
+
+	b = copyConfig(*a)
+	b.Features["auction"] = true
+	assert.True(t, a.isSameAs(b))
+	b.Features["auction"] = false
+	assert.False(t, a.isSameAs(b))
+
+}