Merge pull request #82 from rollbar/pawel/telemetry

added telemetry support

client.go

@@ -10,6 +10,7 @@ import (
 	"reflect"
 	"regexp"
 	"runtime"
+	"time"
 )
 
 // A Client can be used to interact with Rollbar via the configured Transport.
@@ -23,6 +24,7 @@ type Client struct {
 	// Transport used to send data to the Rollbar API. By default an asynchronous
 	// implementation of the Transport interface is used.
 	Transport     Transport
+	Telemetry     *Telemetry
 	configuration configuration
 	diagnostic    diagnostic
 }
@@ -40,6 +42,7 @@ func NewAsync(token, environment, codeVersion, serverHost, serverRoot string) *C
 	diagnostic := createDiagnostic()
 	return &Client{
 		Transport:     transport,
+		Telemetry:     NewTelemetry(nil),
 		configuration: configuration,
 		diagnostic:    diagnostic,
 	}
@@ -52,11 +55,29 @@ func NewSync(token, environment, codeVersion, serverHost, serverRoot string) *Cl
 	diagnostic := createDiagnostic()
 	return &Client{
 		Transport:     transport,
+		Telemetry:     NewTelemetry(nil),
 		configuration: configuration,
 		diagnostic:    diagnostic,
 	}
 }
 
+// CaptureTelemetryEvent sets the user-specified telemetry event
+func (c *Client) CaptureTelemetryEvent(eventType, eventlevel string, eventData map[string]interface{}) {
+	data := map[string]interface{}{}
+	data["body"] = eventData
+	data["type"] = eventType
+	data["level"] = eventlevel
+	data["source"] = "client"
+	data["timestamp_ms"] = time.Now().UnixNano() / int64(time.Millisecond)
+
+	c.Telemetry.Queue.Push(data)
+}
+
+// SetTelemetry sets the telemetry
+func (c *Client) SetTelemetry(options ...OptionFunc) {
+	c.Telemetry = NewTelemetry(c.configuration.scrubHeaders, options...)
+}
+
 // SetEnabled sets whether or not Rollbar is enabled.
 // If this is true then this library works as normal.
 // If this is false then no calls will be made to the network.
@@ -147,6 +168,7 @@ func (c *Client) SetLogger(logger ClientLogger) {
 // The default value is regexp.MustCompile("Authorization")
 func (c *Client) SetScrubHeaders(headers *regexp.Regexp) {
 	c.configuration.scrubHeaders = headers
+	c.Telemetry.Network.ScrubHeaders = headers
 }
 
 // SetScrubFields sets the regular expression to match keys in the item payload for scrubbing.
@@ -361,7 +383,8 @@ func (c *Client) ErrorWithStackSkipWithExtrasAndContext(ctx context.Context, lev
 		return
 	}
 	body := c.buildBody(ctx, level, err.Error(), extras)
-	addErrorToBody(c.configuration, body, err, skip)
+	telemetry := c.Telemetry.GetQueueItems()
+	addErrorToBody(c.configuration, body, err, skip, telemetry)
 	c.push(body)
 }
 
@@ -389,7 +412,8 @@ func (c *Client) RequestErrorWithStackSkipWithExtrasAndContext(ctx context.Conte
 		return
 	}
 	body := c.buildBody(ctx, level, err.Error(), extras)
-	data := addErrorToBody(c.configuration, body, err, skip)
+	telemetry := c.Telemetry.GetQueueItems()
+	data := addErrorToBody(c.configuration, body, err, skip, telemetry)
 	data["request"] = c.requestDetails(r)
 	c.push(body)
 }
@@ -415,7 +439,10 @@ func (c *Client) MessageWithExtrasAndContext(ctx context.Context, level string,
 	}
 	body := c.buildBody(ctx, level, msg, extras)
 	data := body["data"].(map[string]interface{})
-	data["body"] = messageBody(msg)
+	dataBody := messageBody(msg)
+	telemetry := c.Telemetry.GetQueueItems()
+	dataBody["telemetry"] = telemetry
+	data["body"] = dataBody
 	c.push(body)
 }
 
@@ -440,7 +467,10 @@ func (c *Client) RequestMessageWithExtrasAndContext(ctx context.Context, level s
 	}
 	body := c.buildBody(ctx, level, msg, extras)
 	data := body["data"].(map[string]interface{})
-	data["body"] = messageBody(msg)
+	dataBody := messageBody(msg)
+	telemetry := c.Telemetry.GetQueueItems()
+	dataBody["telemetry"] = telemetry
+	data["body"] = dataBody
 	data["request"] = c.requestDetails(r)
 	c.push(body)
 }
@@ -669,12 +699,12 @@ func createConfiguration(token, environment, codeVersion, serverHost, serverRoot
 }
 
 type diagnostic struct {
-	languageVersion   string
+	languageVersion string
 }
 
 func createDiagnostic() diagnostic {
 	return diagnostic{
-		languageVersion:   runtime.Version(),
+		languageVersion: runtime.Version(),
 	}
 }
 

client_test.go

@@ -4,12 +4,13 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/rollbar/rollbar-go"
 	"net/http"
 	"reflect"
 	"regexp"
 	"strings"
 	"testing"
+
+	"github.com/rollbar/rollbar-go"
 )
 
 type TestTransport struct {
@@ -435,6 +436,7 @@ func testGettersAndSetters(client *rollbar.Client, t *testing.T) {
 	errorIfEqual(fingerprint, client.Fingerprint(), t)
 	errorIfEqual(captureIP, client.CaptureIp(), t)
 	errorIfEqual(scrubHeaders, client.ScrubHeaders(), t)
+	errorIfEqual(scrubHeaders, client.Telemetry.Network.ScrubHeaders, t)
 	errorIfEqual(scrubFields, client.ScrubFields(), t)
 
 	if client.Fingerprint() {
@@ -463,6 +465,7 @@ func testGettersAndSetters(client *rollbar.Client, t *testing.T) {
 	client.SetScrubHeaders(scrubHeaders)
 	client.SetScrubFields(scrubFields)
 	client.SetCaptureIp(captureIP)
+	client.SetTelemetry()
 
 	client.SetEnabled(true)
 
@@ -476,6 +479,7 @@ func testGettersAndSetters(client *rollbar.Client, t *testing.T) {
 	errorIfNotEqual(fingerprint, client.Fingerprint(), t)
 	errorIfNotEqual(captureIP, client.CaptureIp(), t)
 	errorIfNotEqual(scrubHeaders, client.ScrubHeaders(), t)
+	errorIfNotEqual(scrubHeaders, client.Telemetry.Network.ScrubHeaders, t)
 	errorIfNotEqual(scrubFields, client.ScrubFields(), t)
 
 	if !client.Fingerprint() {
@@ -677,6 +681,23 @@ func TestEnabled(t *testing.T) {
 	}
 }
 
+func TestCaptureTelemetryEvent(t *testing.T) {
+	client := testClient()
+	data := map[string]interface{}{"message": "some message"}
+	client.CaptureTelemetryEvent("eventType", "eventLevel", data)
+	items := client.Telemetry.GetQueueItems()
+	if len(items) < 1 {
+		t.Error("Queue should not be empty")
+	}
+	item := items[0].(map[string]interface{})
+	delete(item, "timestamp_ms")
+	expectedData := map[string]interface{}{"body": data, "type": "eventType", "level": "eventLevel", "source": "client"}
+	eq := reflect.DeepEqual(item, expectedData)
+	if !eq {
+		t.Error("Maps are different")
+	}
+}
+
 func configuredOptionsFromData(data map[string]interface{}) map[string]interface{} {
 	notifier := data["notifier"].(map[string]interface{})
 	diagnostic := notifier["diagnostic"].(map[string]interface{})

go.mod

@@ -1,3 +1,5 @@
 module github.com/rollbar/rollbar-go
 
 go 1.13
+
+require github.com/stretchr/testify v1.7.0

go.sum

@@ -0,0 +1,10 @@
+github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

queue.go

@@ -0,0 +1,59 @@
+package rollbar
+
+import "sync"
+
+// NewQueue returns a new queue with the given initial size.
+func NewQueue(size int) *Queue {
+	return &Queue{
+		nodes: make([]interface{}, size),
+		size:  size,
+	}
+}
+
+// Queue is a basic FIFO queue based on a circular list that resizes as needed.
+type Queue struct {
+	nodes []interface{}
+	size  int
+	head  int
+	tail  int
+	count int
+
+	lock sync.RWMutex
+}
+
+// Push adds a node to the queue.
+func (q *Queue) Push(n interface{}) {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+	if q.head == q.tail && q.count > 0 {
+		nodes := make([]interface{}, len(q.nodes)+q.size)
+		copy(nodes, q.nodes[q.head:])
+		copy(nodes[len(q.nodes)-q.head:], q.nodes[:q.head])
+		q.head = 0
+		q.tail = len(q.nodes)
+		q.nodes = nodes
+	}
+	q.nodes[q.tail] = n
+	q.tail = (q.tail + 1) % len(q.nodes)
+	q.count++
+}
+
+// Pop removes and returns a node from the queue in first to last order.
+func (q *Queue) Pop() interface{} {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+	if q.count == 0 {
+		return nil
+	}
+	node := q.nodes[q.head]
+	q.head = (q.head + 1) % len(q.nodes)
+	q.count--
+	return node
+}
+
+// Items returns all populated (non nil) items
+func (q *Queue) Items() []interface{} {
+	q.lock.RLock()
+	defer q.lock.RUnlock()
+	return q.nodes[:q.count]
+}

rollbar.go

@@ -90,6 +90,16 @@ var DefaultStackTracer StackTracerFunc = func(err error) ([]runtime.Frame, bool)
 	return nil, false
 }
 
+// SetTelemetry sets the telemetry
+func SetTelemetry(options ...OptionFunc) {
+	std.SetTelemetry(options...)
+}
+
+// CaptureTelemetryEvent sets the user-specified telemetry event
+func CaptureTelemetryEvent(eventType, eventlevel string, eventData map[string]interface{}) {
+	std.CaptureTelemetryEvent(eventType, eventlevel, eventData)
+}
+
 // SetEnabled sets whether or not the managed Client instance is enabled.
 // If this is true then this library works as normal.
 // If this is false then no calls will be made to the network.