Merge pull request #35 from justinfoote/add_events

Add custom Events; refactor Span Events and custom Events to use common struct

examples/server/main.go

@@ -97,7 +97,7 @@ func wrapHandler(path string, handler func(http.ResponseWriter, *http.Request))
 			},
 			Events: []telemetry.Event{
 				telemetry.Event{
-					Name:      "exception",
+					EventType: "exception",
 					Timestamp: before,
 					Attributes: map[string]interface{}{
 						"exception.message": "Everything is fine!",
@@ -106,6 +106,17 @@ func wrapHandler(path string, handler func(http.ResponseWriter, *http.Request))
 				},
 			},
 		})
+
+		h.RecordEvent(telemetry.Event{
+			EventType: "CustomEvent",
+			Timestamp: before,
+			Attributes: map[string]interface{}{
+				"path":        path,
+				"http.method": req.Method,
+				"isWeb":       true,
+				"isExample":   true,
+			},
+		})
 	}
 }
 
@@ -141,8 +152,9 @@ func main() {
 		telemetry.ConfigBasicErrorLogger(os.Stderr),
 		telemetry.ConfigBasicDebugLogger(os.Stdout),
 		func(cfg *telemetry.Config) {
-			cfg.MetricsURLOverride = os.Getenv("NEW_RELIC_METRICS_URL")
-			cfg.SpansURLOverride = os.Getenv("NEW_RELIC_SPANS_URL")
+			cfg.MetricsURLOverride = os.Getenv("NEW_RELIC_METRIC_URL")
+			cfg.SpansURLOverride = os.Getenv("NEW_RELIC_TRACE_URL")
+			cfg.EventsURLOverride = os.Getenv("NEW_RELIC_EVENT_URL")
 		},
 	)
 	if nil != err {

go.mod

@@ -1 +1,3 @@
 module github.com/newrelic/newrelic-telemetry-sdk-go
+
+go 1.13

telemetry/config.go

@@ -37,14 +37,16 @@ type Config struct {
 	// AuditLogger receives structured log messages that include the
 	// uncompressed data sent to New Relic.  Use this to log all data sent.
 	AuditLogger func(map[string]interface{})
-	// MetricsURLOverride overrides the metrics endpoint if not not empty.
+	// MetricsURLOverride overrides the metrics endpoint if not empty.
 	MetricsURLOverride string
-	// SpansURLOverride overrides the spans endpoint if not not empty.
+	// SpansURLOverride overrides the spans endpoint if not empty.
 	//
 	// To enable Infinite Tracing on the New Relic Edge, set this field to your
 	// Trace Observer URL.  See
 	// https://docs.newrelic.com/docs/understand-dependencies/distributed-tracing/enable-configure/enable-distributed-tracing
 	SpansURLOverride string
+	// EventsURLOverride overrides the events endpoint if not empty
+	EventsURLOverride string
 	// Product is added to the User-Agent header. eg. "NewRelic-Go-OpenCensus"
 	Product string
 	// ProductVersion is added to the User-Agent header. eg. "0.1.0".
@@ -113,13 +115,21 @@ func ConfigBasicAuditLogger(w io.Writer) func(*Config) {
 }
 
 // ConfigSpansURLOverride sets the Config's SpansURLOverride field which
-// overrides the spans endpoint if not not empty.
+// overrides the spans endpoint if not empty.
 func ConfigSpansURLOverride(url string) func(*Config) {
 	return func(cfg *Config) {
 		cfg.SpansURLOverride = url
 	}
 }
 
+// ConfigEventsURLOverride sets the Config's EventsURLOverride field which
+// overrides the events endpoint if not empty.
+func ConfigEventsURLOverride(url string) func(*Config) {
+	return func(cfg *Config) {
+		cfg.EventsURLOverride = url
+	}
+}
+
 // configTesting is the config function to be used when testing. It sets the
 // APIKey but disables the harvest goroutine.
 func configTesting(cfg *Config) {
@@ -155,6 +165,7 @@ func (cfg *Config) logAudit(fields map[string]interface{}) {
 const (
 	defaultSpanURL   = "https://trace-api.newrelic.com/trace/v1"
 	defaultMetricURL = "https://metric-api.newrelic.com/metric/v1"
+	defaultEventURL  = "https://insights-collector.newrelic.com/v1/accounts/events"
 )
 
 func (cfg *Config) spanURL() string {
@@ -171,6 +182,13 @@ func (cfg *Config) metricURL() string {
 	return defaultMetricURL
 }
 
+func (cfg *Config) eventURL() string {
+	if cfg.EventsURLOverride != "" {
+		return cfg.EventsURLOverride
+	}
+	return defaultEventURL
+}
+
 // userAgent creates the User-Agent header version according to the spec here:
 // https://github.com/newrelic/newrelic-telemetry-sdk-specs/blob/master/communication.md#user-agent
 func (cfg *Config) userAgent() string {

telemetry/config_test.go

@@ -119,17 +119,20 @@ func TestConfigMetricURL(t *testing.T) {
 	}
 }
 
-func TestConfigSpanURL(t *testing.T) {
+func TestConfigSpansURL(t *testing.T) {
+	t.Parallel()
+
+	// We get the default
 	h, err := NewHarvester(configTesting)
 	if nil == h || err != nil {
 		t.Fatal(h, err)
 	}
 	if u := h.config.spanURL(); u != defaultSpanURL {
 		t.Fatal(u)
 	}
-	h, err = NewHarvester(configTesting, func(cfg *Config) {
-		cfg.SpansURLOverride = "span-url-override"
-	})
+
+	// The override config option works
+	h, err = NewHarvester(configTesting, ConfigSpansURLOverride("span-url-override"))
 	if nil == h || err != nil {
 		t.Fatal(h, err)
 	}
@@ -138,6 +141,28 @@ func TestConfigSpanURL(t *testing.T) {
 	}
 }
 
+func TestConfigEventsURL(t *testing.T) {
+	t.Parallel()
+
+	// We get the default
+	h, err := NewHarvester(configTesting)
+	if nil == h || err != nil {
+		t.Fatal(h, err)
+	}
+	if u := h.config.eventURL(); u != defaultEventURL {
+		t.Fatal(u)
+	}
+
+	// The override config option works
+	h, err = NewHarvester(configTesting, ConfigEventsURLOverride("event-url-override"))
+	if nil == h || err != nil {
+		t.Fatal(h, err)
+	}
+	if u := h.config.eventURL(); u != "event-url-override" {
+		t.Fatal(u)
+	}
+}
+
 func TestConfigUserAgent(t *testing.T) {
 	testcases := []struct {
 		option func(*Config)

telemetry/events.go

@@ -0,0 +1,82 @@
+package telemetry
+
+import (
+	"bytes"
+	"encoding/json"
+	"time"
+
+	"github.com/newrelic/newrelic-telemetry-sdk-go/internal"
+)
+
+// Event is a unique set of data that happened at a specific point in time
+type Event struct {
+	// Required Fields:
+	//
+	// EventType is the name of the event
+	EventType string
+	// Timestamp is when this event happened.  If Timestamp is not set, it
+	// will be assigned to time.Now() in Harvester.RecordEvent.
+	Timestamp time.Time
+
+	// Recommended Fields:
+	//
+	// Attributes is a map of user specified data on this event.  The map
+	// values can be any of bool, number, or string.
+	Attributes map[string]interface{}
+	// AttributesJSON is a json.RawMessage of attributes for this metric. It
+	// will only be sent if Attributes is nil.
+	AttributesJSON json.RawMessage
+}
+
+func (e *Event) writeJSON(buf *bytes.Buffer) {
+	w := internal.JSONFieldsWriter{Buf: buf}
+	buf.WriteByte('{')
+
+	w.StringField("eventType", e.EventType)
+	w.IntField("timestamp", e.Timestamp.UnixNano()/(1000*1000))
+
+	internal.AddAttributes(&w, e.Attributes)
+
+	buf.WriteByte('}')
+}
+
+// eventBatch represents a single batch of events to report to New Relic.
+type eventBatch struct {
+	Events []Event
+}
+
+// split will split the eventBatch into 2 equally sized batches.
+// If the number of events in the original is 0 or 1 then nil is returned.
+func (batch *eventBatch) split() []requestsBuilder {
+	if len(batch.Events) < 2 {
+		return nil
+	}
+
+	half := len(batch.Events) / 2
+	b1 := *batch
+	b1.Events = batch.Events[:half]
+	b2 := *batch
+	b2.Events = batch.Events[half:]
+
+	return []requestsBuilder{
+		requestsBuilder(&b1),
+		requestsBuilder(&b2),
+	}
+}
+
+func (batch *eventBatch) writeJSON(buf *bytes.Buffer) {
+	buf.WriteByte('[')
+	for idx, s := range batch.Events {
+		if idx > 0 {
+			buf.WriteByte(',')
+		}
+		s.writeJSON(buf)
+	}
+	buf.WriteByte(']')
+}
+
+func (batch *eventBatch) makeBody() json.RawMessage {
+	buf := &bytes.Buffer{}
+	batch.writeJSON(buf)
+	return buf.Bytes()
+}

telemetry/events_batch_test.go

@@ -0,0 +1,110 @@
+// Copyright 2019 New Relic Corporation. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package telemetry
+
+import (
+	"io/ioutil"
+	"testing"
+	"time"
+
+	"github.com/newrelic/newrelic-telemetry-sdk-go/internal"
+)
+
+func testEventBatchJSON(t testing.TB, batch *eventBatch, expect string) {
+	if th, ok := t.(interface{ Helper() }); ok {
+		th.Helper()
+	}
+	reqs, err := newRequests(batch, "apiKey", defaultSpanURL, "userAgent")
+	if nil != err {
+		t.Fatal(err)
+	}
+	if len(reqs) != 1 {
+		t.Fatal(reqs)
+	}
+	req := reqs[0]
+	actual := string(req.UncompressedBody)
+	compact := compactJSONString(expect)
+	if actual != compact {
+		t.Errorf("\nexpect=%s\nactual=%s\n", compact, actual)
+	}
+
+	body, err := ioutil.ReadAll(req.Request.Body)
+	req.Request.Body.Close()
+	if err != nil {
+		t.Fatal("unable to read body", err)
+	}
+	if len(body) != req.compressedBodyLength {
+		t.Error("compressed body length mismatch",
+			len(body), req.compressedBodyLength)
+	}
+	uncompressed, err := internal.Uncompress(body)
+	if err != nil {
+		t.Fatal("unable to uncompress body", err)
+	}
+	if string(uncompressed) != string(req.UncompressedBody) {
+		t.Error("request JSON mismatch", string(uncompressed), string(req.UncompressedBody))
+	}
+}
+
+func TestEventsPayloadSplit(t *testing.T) {
+	t.Parallel()
+
+	// test len 0
+	ev := &eventBatch{}
+	split := ev.split()
+	if split != nil {
+		t.Error(split)
+	}
+
+	// test len 1
+	ev = &eventBatch{Events: []Event{{EventType: "a"}}}
+	split = ev.split()
+	if split != nil {
+		t.Error(split)
+	}
+
+	// test len 2
+	ev = &eventBatch{Events: []Event{{EventType: "a"}, {EventType: "b"}}}
+	split = ev.split()
+	if len(split) != 2 {
+		t.Error("split into incorrect number of slices", len(split))
+	}
+
+	testEventBatchJSON(t, split[0].(*eventBatch), `[{"eventType":"a","timestamp":-6795364578871}]`)
+	testEventBatchJSON(t, split[1].(*eventBatch), `[{"eventType":"b","timestamp":-6795364578871}]`)
+
+	// test len 3
+	ev = &eventBatch{Events: []Event{{EventType: "a"}, {EventType: "b"}, {EventType: "c"}}}
+	split = ev.split()
+	if len(split) != 2 {
+		t.Error("split into incorrect number of slices", len(split))
+	}
+	testEventBatchJSON(t, split[0].(*eventBatch), `[{"eventType":"a","timestamp":-6795364578871}]`)
+	testEventBatchJSON(t, split[1].(*eventBatch), `[{"eventType":"b","timestamp":-6795364578871},{"eventType":"c","timestamp":-6795364578871}]`)
+}
+
+func TestEventsJSON(t *testing.T) {
+	t.Parallel()
+
+	batch := &eventBatch{Events: []Event{
+		{}, // Empty
+		{ // with everything
+			EventType:  "testEvent",
+			Timestamp:  time.Date(2014, time.November, 28, 1, 1, 0, 0, time.UTC),
+			Attributes: map[string]interface{}{"zip": "zap"},
+		},
+	}}
+
+	testEventBatchJSON(t, batch, `[
+		{
+		  "eventType":"",
+			"timestamp":-6795364578871
+		},
+		{
+			"eventType":"testEvent",
+			"timestamp":1417136460000,
+			"zip":"zap"
+		}
+	]`)
+}