Repository: atlassian/gostatsd
Pull request #250 (2019-07-19 16:31:04)

Add support for events via New Relic backend
When the event is sent through, it is sent as a full New Relic event, so all context is maintained and queryable.

eg.
![image](https://user-images.githubusercontent.com/13043208/61550591-2f7c4380-aa96-11e9-8a3b-175fe2aa9a1d.png)


Patch:
diff --git a/pkg/backends/newrelic/flush.go b/pkg/backends/newrelic/flush.go
index d3fa49d..48a10d0 100644
--- a/pkg/backends/newrelic/flush.go
+++ b/pkg/backends/newrelic/flush.go
@@ -1,9 +1,6 @@
 package newrelic
 
 import (
-	"strconv"
-	"strings"
-
 	"github.com/atlassian/gostatsd"
 )
 
@@ -103,20 +100,7 @@ func newMetricSet(n *Client, f *flush, metricName, Type string, Value float64, t
 	metricSet[n.metricType] = Type
 	metricSet[n.metricName] = metricName
 	metricSet[n.metricValue] = Value
-
-	for _, tag := range tags {
-		if strings.Contains(tag, ":") {
-			keyvalpair := strings.SplitN(tag, ":", 2)
-			parsed, err := strconv.ParseFloat(keyvalpair[1], 64)
-			if err != nil || strings.EqualFold(keyvalpair[1], "infinity") {
-				metricSet[n.tagPrefix+keyvalpair[0]] = keyvalpair[1]
-			} else {
-				metricSet[n.tagPrefix+keyvalpair[0]] = parsed
-			}
-		} else {
-			metricSet[n.tagPrefix+tag] = "true"
-		}
-	}
+	n.setTags(tags, metricSet)
 
 	return metricSet
 }
diff --git a/pkg/backends/newrelic/newrelic.go b/pkg/backends/newrelic/newrelic.go
index 843a9a9..f5c9c70 100644
--- a/pkg/backends/newrelic/newrelic.go
+++ b/pkg/backends/newrelic/newrelic.go
@@ -12,6 +12,7 @@ import (
 	"net"
 	"net/http"
 	"runtime"
+	"strconv"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -28,7 +29,7 @@ const (
 	// BackendName is the name of this backend.
 	BackendName                  = "newrelic"
 	integrationName              = "com.newrelic.gostatsd"
-	integrationVersion           = "2.1.0"
+	integrationVersion           = "2.2.0"
 	protocolVersion              = "2"
 	defaultUserAgent             = "gostatsd"
 	defaultMaxRequestElapsedTime = 15 * time.Second
@@ -197,7 +198,46 @@ func (n *Client) processMetrics(metrics *gostatsd.MetricMap, cb func(*timeSeries
 
 // SendEvent sends an event to New Relic.
 func (n *Client) SendEvent(ctx context.Context, e *gostatsd.Event) error {
-	return nil
+	// Event format depends on flush type
+	data := n.EventFormatter(e)
+	b, err := json.Marshal(data)
+	if err != nil {
+		return err
+	}
+
+	post, err := n.postWrapper(ctx, b)
+	if err != nil {
+		return err
+	}
+
+	return post()
+}
+
+// EventFormatter formats gostatsd events
+func (n *Client) EventFormatter(e *gostatsd.Event) interface{} {
+	event := map[string]interface{}{
+		"name":           "event",
+		"Title":          e.Title,
+		"Text":           e.Text,
+		"DateHappened":   e.DateHappened,
+		"Hostname":       e.Hostname,
+		"AggregationKey": e.AggregationKey,
+		"SourceTypeName": e.SourceTypeName,
+		"Priority":       e.Priority.StringWithEmptyDefault(),
+		"AlertType":      e.AlertType.StringWithEmptyDefault(),
+	}
+	n.setTags(e.Tags, event)
+
+	switch n.flushType {
+	case flushTypeInsights:
+		event["eventType"] = n.eventType
+		return []interface{}{event}
+	default:
+		event["event_type"] = n.eventType
+		return newInfraPayload(map[string]interface{}{
+			"metrics": []interface{}{event},
+		})
+	}
 }
 
 // Name returns the name of the backend.
@@ -241,7 +281,6 @@ func (n *Client) post(ctx context.Context, buffer *bytes.Buffer, data interface{
 }
 
 func (n *Client) constructPost(ctx context.Context, buffer *bytes.Buffer, data interface{}) (func() error /*doPost*/, error) {
-
 	var mJSON []byte
 	var mErr error
 	switch n.flushType {
@@ -257,6 +296,11 @@ func (n *Client) constructPost(ctx context.Context, buffer *bytes.Buffer, data i
 		return nil, fmt.Errorf("[%s] unable to marshal: %v", BackendName, mErr)
 	}
 
+	return n.postWrapper(ctx, mJSON)
+}
+
+// postWrapper compresses JSON for Insights
+func (n *Client) postWrapper(ctx context.Context, json []byte) (func() error, error) {
 	return func() error {
 		headers := map[string]string{
 			"Content-Type": "application/json",
@@ -267,14 +311,22 @@ func (n *Client) constructPost(ctx context.Context, buffer *bytes.Buffer, data i
 		if n.flushType == flushTypeInsights && n.apiKey != "" {
 			headers["X-Insert-Key"] = n.apiKey
 			headers["Content-Encoding"] = "deflate"
-			var b bytes.Buffer
-			w := zlib.NewWriter(&b)
-			w.Write(mJSON) // nolint:errcheck
-			w.Close()
-			mJSON = b.Bytes()
+			// zlib json
+			var buf bytes.Buffer
+			zw := zlib.NewWriter(&buf)
+			_, err := zw.Write([]byte(json))
+			if err != nil {
+				return err
+			}
+
+			// Close to ensure a flush
+			if err := zw.Close(); err != nil {
+				return err
+			}
+			json = buf.Bytes()
 		}
 
-		req, err := http.NewRequest("POST", n.address, bytes.NewBuffer(mJSON))
+		req, err := http.NewRequest("POST", n.address, bytes.NewBuffer(json))
 		if err != nil {
 			return fmt.Errorf("unable to create http.Request: %v", err)
 		}
@@ -486,6 +538,22 @@ func newInfraPayload(data interface{}) NewRelicInfraPayload {
 	}
 }
 
+func (n *Client) setTags(tags gostatsd.Tags, data map[string]interface{}) {
+	for _, tag := range tags {
+		if strings.Contains(tag, ":") {
+			keyvalpair := strings.SplitN(tag, ":", 2)
+			parsed, err := strconv.ParseFloat(keyvalpair[1], 64)
+			if err != nil || strings.EqualFold(keyvalpair[1], "infinity") {
+				data[n.tagPrefix+keyvalpair[0]] = keyvalpair[1]
+			} else {
+				data[n.tagPrefix+keyvalpair[0]] = parsed
+			}
+		} else {
+			data[n.tagPrefix+tag] = "true"
+		}
+	}
+}
+
 // contains checks if item is within slice
 func contains(s []string, e string) bool {
 	for _, a := range s {


Test patch:
diff --git a/pkg/backends/newrelic/newrelic_test.go b/pkg/backends/newrelic/newrelic_test.go
index 6dbd28a..00b7256 100644
--- a/pkg/backends/newrelic/newrelic_test.go
+++ b/pkg/backends/newrelic/newrelic_test.go
@@ -2,6 +2,7 @@ package newrelic
 
 import (
 	"context"
+	"encoding/json"
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
@@ -92,11 +93,11 @@ func TestSendMetrics(t *testing.T) {
 		if !assert.NoError(t, err) {
 			return
 		}
-		expected := `{"name":"com.newrelic.gostatsd","protocol_version":"2","integration_version":"2.1.0","data":[{"metrics":` +
-			`[{"event_type":"GoStatsD","integration_version":"2.1.0","interval":1,"metric_name":"g1","metric_type":"gauge","metric_value":3,"tag3":"true","timestamp":0},` +
-			`{"event_type":"GoStatsD","integration_version":"2.1.0","interval":1,"metric_name":"c1","metric_per_second":1.1,"metric_type":"counter","metric_value":5,"tag1":"true","timestamp":0},` +
-			`{"event_type":"GoStatsD","integration_version":"2.1.0","interval":1,"metric_name":"users","metric_type":"set","metric_value":3,"tag4":"true","timestamp":0},` +
-			`{"count_90":0.1,"event_type":"GoStatsD","integration_version":"2.1.0","interval":1,"metric_name":"t1","metric_per_second":1.1,"metric_type":"timer","metric_value":1,` +
+		expected := `{"name":"com.newrelic.gostatsd","protocol_version":"2","integration_version":"2.2.0","data":[{"metrics":` +
+			`[{"event_type":"GoStatsD","integration_version":"2.2.0","interval":1,"metric_name":"g1","metric_type":"gauge","metric_value":3,"tag3":"true","timestamp":0},` +
+			`{"event_type":"GoStatsD","integration_version":"2.2.0","interval":1,"metric_name":"c1","metric_per_second":1.1,"metric_type":"counter","metric_value":5,"tag1":"true","timestamp":0},` +
+			`{"event_type":"GoStatsD","integration_version":"2.2.0","interval":1,"metric_name":"users","metric_type":"set","metric_value":3,"tag4":"true","timestamp":0},` +
+			`{"count_90":0.1,"event_type":"GoStatsD","integration_version":"2.2.0","interval":1,"metric_name":"t1","metric_per_second":1.1,"metric_type":"timer","metric_value":1,` +
 			`"samples_count":1,"samples_max":1,"samples_mean":0.5,"samples_median":0.5,"samples_min":0,"samples_std_dev":0.1,"samples_sum":1,"samples_sum_squares":1,"tag2":"true","timestamp":0}]}]}`
 		assert.Equal(t, expected, string(data))
 	})
@@ -186,3 +187,23 @@ func metricsOneOfEach() *gostatsd.MetricMap {
 		},
 	}
 }
+
+func TestEventFormatter(t *testing.T) {
+	t.Parallel()
+
+	client, err := NewClient("v1/data", "GoStatsD", "", "", "", "metric_name", "metric_type",
+		"metric_per_second", "metric_value", "samples_min", "samples_max", "samples_count",
+		"samples_mean", "samples_median", "samples_std_dev", "samples_sum", "samples_sum_squares", "agent", "tcp",
+		defaultMetricsPerBatch, defaultMaxRequests, false, 1*time.Second, 2*time.Second, 1*time.Second, gostatsd.TimerSubtypes{})
+	require.NoError(t, err)
+
+	gostatsdEvent := gostatsd.Event{Title: "EventTitle", Text: "hi", Hostname: "blah", Priority: 1}
+	formattedEvent := client.EventFormatter(&gostatsdEvent)
+	fevent, err := json.Marshal(formattedEvent)
+	require.NoError(t, err)
+
+	expected := `{"name":"com.newrelic.gostatsd","protocol_version":"2","integration_version":"2.2.0","data":` +
+		`[{"metrics":[{"AggregationKey":"","AlertType":"","DateHappened":0,"Hostname":"blah","Priority":"low","SourceTypeName":"","Text":"hi","Title":"EventTitle","event_type":"GoStatsD","name":"event"}]}]}`
+
+	require.Equal(t, expected, string(fevent))
+}