Repository: rollbar/rollbar-go
Pull request #82 (2021-04-07 23:22:07)

added telemetry support
## Description of the change

- Added telemetry support 
- To initialize Telemetry with different options, a user can set it as following: 
```go
client := &http.Client{}
rollbar.SetTelemetry(rollbar.EnableNetworkTelemetry(client), rollbar.EnableLoggerTelemetry(),
	        rollbar.EnableNetworkTelemetryRequestHeaders(),
		rollbar.SetCustomQueueSize(100),
	)
```

## Type of change
- [ ] Bug fix (non-breaking change that fixes an issue)
- [x] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)

## Related issues

> Fix [#82032](https://app.clubhouse.io/rollbar/story/82032/spike-go-telemetry-support) 
## Checklists

### Development

- [x] Lint rules pass locally
- [x] The code changed/added as part of this pull request has been covered with tests
- [x] All tests related to the changed code pass in development

### Code review 

- [x]  This pull request has a descriptive title and information useful to a reviewer. There may be a screenshot or screencast attached
- [x] "Ready for review" label attached to the PR and reviewers mentioned in a comment
- [x] Changes have been reviewed by at least one other engineer
- [x] Issue from task tracker has a link to this pull request 



Patch:
diff --git a/client.go b/client.go
index f9d966e..6793c41 100644
--- a/client.go
+++ b/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(),
 	}
 }
 
diff --git a/go.mod b/go.mod
index ac50a32..06412f5 100644
--- a/go.mod
+++ b/go.mod
@@ -1,3 +1,5 @@
 module github.com/rollbar/rollbar-go
 
 go 1.13
+
+require github.com/stretchr/testify v1.7.0
diff --git a/go.sum b/go.sum
index e69de29..b380ae4 100644
--- a/go.sum
+++ b/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=
diff --git a/queue.go b/queue.go
new file mode 100644
index 0000000..9cc0923
--- /dev/null
+++ b/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]
+}
diff --git a/rollbar.go b/rollbar.go
index 860fc47..33776be 100644
--- a/rollbar.go
+++ b/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.
diff --git a/telemetry.go b/telemetry.go
new file mode 100644
index 0000000..cb589d1
--- /dev/null
+++ b/telemetry.go
@@ -0,0 +1,172 @@
+package rollbar
+
+import (
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"os"
+	"regexp"
+	"time"
+)
+
+const TelemetryQueueSize = 50
+
+// Telemetry struct contains writer (for logs) and round tripper (for http client) and enables to queue the events
+type Telemetry struct {
+	Logger struct {
+		Writer io.Writer
+	}
+
+	Network struct {
+		Proxied      http.RoundTripper
+		ScrubHeaders *regexp.Regexp
+
+		enableReqHeaders bool
+		enableResHeaders bool
+	}
+	Queue *Queue
+}
+
+// Write is the writer for telemetry logs
+func (t *Telemetry) Write(p []byte) (int, error) {
+	telemetryData := t.populateLoggerBody(p)
+	t.Queue.Push(telemetryData)
+	return t.Logger.Writer.Write(p)
+}
+
+// RoundTrip implements RoundTrip in http.RoundTripper
+func (t *Telemetry) RoundTrip(req *http.Request) (res *http.Response, e error) {
+
+	// Send the request, get the response (or the error)
+	res, e = t.Network.Proxied.RoundTrip(req)
+	if e != nil {
+		fmt.Printf("Error: %v", e)
+	}
+	telemetryData := t.populateTransporterBody(req, res)
+	t.Queue.Push(telemetryData)
+	return
+}
+
+func (t *Telemetry) populateLoggerBody(p []byte) map[string]interface{} {
+	var data = map[string]interface{}{}
+	message := map[string]interface{}{"message": string(p)}
+	data["body"] = message
+	data["source"] = "client"
+	data["timestamp_ms"] = time.Now().UnixNano() / int64(time.Millisecond)
+	data["type"] = "log"
+	data["level"] = "log"
+	return data
+}
+
+func (t *Telemetry) populateTransporterBody(req *http.Request, res *http.Response) map[string]interface{} {
+	var data = map[string]interface{}{}
+	var dataBody = map[string]interface{}{}
+	dataBody["status_code"] = nil
+	data["level"] = "info"
+	if res != nil {
+		dataBody["status_code"] = res.StatusCode
+		if res.StatusCode >= http.StatusInternalServerError {
+			data["level"] = "critical"
+		} else if res.StatusCode >= http.StatusBadRequest {
+			data["level"] = "error"
+		}
+
+		if t.Network.enableResHeaders {
+			var dataHeaders = map[string][]string{}
+			for k, v := range res.Header {
+				dataHeaders[k] = v
+			}
+			filteredDataHeaders := filterFlatten(t.Network.ScrubHeaders, dataHeaders, nil)
+			response := map[string]interface{}{"headers": filteredDataHeaders}
+			dataBody["response"] = response
+		}
+	}
+	dataBody["url"] = req.URL.Scheme + "://" + req.Host + req.URL.Path
+	dataBody["method"] = req.Method
+	dataBody["subtype"] = "http"
+
+	if t.Network.enableReqHeaders {
+		var dataHeaders = map[string][]string{}
+		for k, v := range req.Header {
+			dataHeaders[k] = v
+		}
+		filteredDataHeaders := filterFlatten(t.Network.ScrubHeaders, dataHeaders, nil)
+		dataBody["request_headers"] = filteredDataHeaders
+	}
+	data["body"] = dataBody
+	data["source"] = "client"
+	data["timestamp_ms"] = time.Now().UnixNano() / int64(time.Millisecond)
+	data["type"] = "network"
+	return data
+}
+
+// GetQueueItems gets all the items from the queue
+func (t *Telemetry) GetQueueItems() []interface{} {
+	return t.Queue.Items()
+}
+
+// OptionFunc is the pointer to the optional parameter function
+type OptionFunc func(*Telemetry)
+
+// EnableNetworkTelemetry enables the network telemetry
+// it wraps up the client for telemetry
+// http.DefaultClient can also be passed by the reference
+func EnableNetworkTelemetry(httpClient *http.Client) OptionFunc {
+	return func(f *Telemetry) {
+		if httpClient.Transport == nil {
+			f.Network.Proxied = http.DefaultTransport
+		} else {
+			f.Network.Proxied = httpClient.Transport
+		}
+		httpClient.Transport = f
+	}
+}
+
+// EnableNetworkTelemetryRequestHeaders enables telemetry request headers
+func EnableNetworkTelemetryRequestHeaders() OptionFunc {
+	return func(f *Telemetry) {
+		f.Network.enableReqHeaders = true
+	}
+}
+
+// EnableNetworkTelemetryResponseHeaders enables telemetry response headers
+func EnableNetworkTelemetryResponseHeaders() OptionFunc {
+	return func(f *Telemetry) {
+		f.Network.enableResHeaders = true
+	}
+}
+
+// SetCustomQueueSize initializes the queue with a custom size
+func SetCustomQueueSize(size int) OptionFunc {
+	return func(f *Telemetry) {
+		f.Queue = NewQueue(size)
+	}
+}
+
+// EnableLoggerTelemetry enables logger telemetry
+func EnableLoggerTelemetry() OptionFunc {
+	return func(f *Telemetry) {
+		f.Logger.Writer = os.Stdout
+		log.SetOutput(f)
+	}
+}
+
+// NewTelemetry initializes telemetry object with scrubheader
+func NewTelemetry(scrubHeaders *regexp.Regexp, options ...OptionFunc) *Telemetry {
+	res := &Telemetry{
+		Queue: NewQueue(TelemetryQueueSize),
+	}
+
+	for _, opt := range options {
+		opt(res)
+	}
+
+	if scrubHeaders == nil {
+		res.Network.ScrubHeaders = regexp.MustCompile("Authorization")
+	} else {
+		res.Network.ScrubHeaders = scrubHeaders
+	}
+
+	return res
+}
diff --git a/transforms.go b/transforms.go
index e5339ff..11278e5 100644
--- a/transforms.go
+++ b/transforms.go
@@ -34,7 +34,7 @@ func buildBody(ctx context.Context, configuration configuration, diagnostic diag
 			"name":    NAME,
 			"version": VERSION,
 			"diagnostic": map[string]interface{}{
-				"languageVersion": diagnostic.languageVersion,
+				"languageVersion":   diagnostic.languageVersion,
 				"configuredOptions": buildConfiguredOptions(configuration),
 			},
 		},
@@ -101,10 +101,12 @@ func buildConfiguredOptions(configuration configuration) map[string]interface{}
 	}
 }
 
-func addErrorToBody(configuration configuration, body map[string]interface{}, err error, skip int) map[string]interface{} {
+func addErrorToBody(configuration configuration, body map[string]interface{}, err error, skip int, telemetry []interface{}) map[string]interface{} {
 	data := body["data"].(map[string]interface{})
 	errBody, fingerprint := errorBody(configuration, err, skip)
-	data["body"] = errBody
+	dataBody := errBody
+	dataBody["telemetry"] = telemetry
+	data["body"] = dataBody
 	if configuration.fingerprint {
 		data["fingerprint"] = fingerprint
 	}


Test patch:
diff --git a/client_test.go b/client_test.go
index 7f9a770..e213bdd 100644
--- a/client_test.go
+++ b/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{})
diff --git a/telemetry_test.go b/telemetry_test.go
new file mode 100644
index 0000000..dea4cbf
--- /dev/null
+++ b/telemetry_test.go
@@ -0,0 +1,128 @@
+package rollbar
+
+import (
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"regexp"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewTelemetryDefault(t *testing.T) {
+	telemetry := NewTelemetry(nil)
+	assert.NotNil(t, telemetry)
+	expectedTelemetry := &Telemetry{Queue: NewQueue(TelemetryQueueSize)}
+	expectedTelemetry.Network.ScrubHeaders = regexp.MustCompile("Authorization")
+
+	assert.Equal(t, expectedTelemetry, telemetry)
+
+}
+
+func TestNewTelemetryWithOptions(t *testing.T) {
+	client := http.Client{}
+	telemetry := NewTelemetry(nil, SetCustomQueueSize(100), EnableNetworkTelemetry(&client),
+		EnableNetworkTelemetryRequestHeaders(), EnableNetworkTelemetryResponseHeaders(), EnableLoggerTelemetry())
+	expectedTelemetry := &Telemetry{Queue: telemetry.Queue}
+	expectedTelemetry.Network.ScrubHeaders = regexp.MustCompile("Authorization")
+	expectedTelemetry.Network.enableReqHeaders = true
+	expectedTelemetry.Network.enableResHeaders = true
+	expectedTelemetry.Network.Proxied = http.DefaultTransport
+	expectedTelemetry.Logger.Writer = os.Stdout
+
+	assert.Equal(t, expectedTelemetry, telemetry)
+	assert.Equal(t, client.Transport, expectedTelemetry)
+}
+func TestPopulateBody(t *testing.T) {
+	req := httptest.NewRequest("GET", "/some_url", nil)
+	req.Header.Set("Some_name", "some_value")
+	rec := httptest.NewRecorder()
+
+	telemetry := NewTelemetry(nil)
+	EnableNetworkTelemetryRequestHeaders()(telemetry)
+	EnableNetworkTelemetryResponseHeaders()(telemetry)
+	data := telemetry.populateTransporterBody(req, rec.Result())
+	assert.NotNil(t, data)
+	assert.True(t, data["timestamp_ms"].(int64) > 0)
+	delete(data, "timestamp_ms")
+
+	expectedBodyData := map[string]interface{}{"method": "GET", "status_code": 200, "subtype": "http",
+		"url": "://example.com/some_url", "request_headers": map[string]interface{}{"Some_name": "some_value"},
+		"response": map[string]interface{}{"headers": map[string]interface{}{}}}
+
+	expectedData := map[string]interface{}{"body": expectedBodyData, "level": "info", "source": "client", "type": "network"}
+
+	assert.Equal(t, expectedData, data)
+}
+
+func TestPopulateLoggerBody(t *testing.T) {
+
+	message := "some message"
+	telemetry := NewTelemetry(nil)
+
+	data := telemetry.populateLoggerBody([]byte(message))
+
+	assert.NotNil(t, data)
+	assert.True(t, data["timestamp_ms"].(int64) > 0)
+	delete(data, "timestamp_ms")
+	expectedData := map[string]interface{}{"body": map[string]interface{}{"message": message}, "level": "log",
+		"source": "client", "type": "log"}
+
+	assert.Equal(t, expectedData, data)
+}
+
+func TestRoundTrip(t *testing.T) {
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		log.Print(">request: ", r)
+		if r.URL.String() == "/good" {
+			fmt.Fprintln(w, "Hello, client")
+		}
+	}))
+	defer ts.Close()
+
+	client := http.Client{}
+	telemetry := NewTelemetry(nil, EnableNetworkTelemetry(&client))
+
+	req := httptest.NewRequest("GET", ts.URL+"/good", nil)
+	res, err := telemetry.RoundTrip(req)
+	assert.Nil(t, err)
+	assert.NotNil(t, res)
+
+	body, err := ioutil.ReadAll(res.Body)
+	assert.Nil(t, err)
+	assert.Equal(t, "Hello, client\n", string(body))
+
+	items := telemetry.GetQueueItems()
+	assert.NotNil(t, items)
+
+	item := items[0].(map[string]interface{})
+	delete(item, "timestamp_ms")
+
+	expectedData := telemetry.populateTransporterBody(req, res)
+	delete(expectedData, "timestamp_ms")
+	assert.Equal(t, item, expectedData)
+}
+
+func TestWrite(t *testing.T) {
+	telemetry := NewTelemetry(nil, EnableLoggerTelemetry())
+	message := "some message"
+	count, err := telemetry.Write([]byte(message))
+
+	assert.Nil(t, err)
+	assert.Equal(t, count, len(message))
+
+	items := telemetry.GetQueueItems()
+	assert.NotNil(t, items)
+
+	item := items[0].(map[string]interface{})
+	delete(item, "timestamp_ms")
+
+	expectedData := telemetry.populateLoggerBody([]byte(message))
+	delete(expectedData, "timestamp_ms")
+	assert.Equal(t, item, expectedData)
+}