Repository: atlassian/gostatsd
Pull request #275 (2019-11-06 07:10:53)

Add support for New Relic's new dimensional metrics pipeline
Adds support for New Relic's dimensional metrics via the new metric api.

https://docs.newrelic.com/docs/data-ingest-apis/get-data-new-relic/metric-api/introduction-metric-api

BACKENDS.md has been updated with additional information as well as relevant tests.

Kav91: Have documented the flush type options now.

Also background for metric-name, metric-type and the other configurable attributes they were made as such to support the insights & infra flush-types, the reason for that was users were particular in the way they wanted to query their attributes coming from other StatsD solutions, so this made it easier for them to switch.

With the new metrics pipeline, this is obviously a lot more standardized and built for things like StatsD. Ultimately we would like all NR users to use the dimensional metrics `metrics` flush type, as the default solution going forward.

There is some background work we'd like to complete before that, so will plan another PR once ready and hopefully we can remove the other options and tidy it all up.

@tiedotguy thanks as always for the review, and appreciate your understanding 👍 
tiedotguy: Ahh, I was under the impression the data was more structured.  Since I'm not familiar with NR specifically I lack the context on why it needs to be configurable :)

And thank you for bringing this in - looking through the forks, there's a handful of alternate backends which have been added by other companies.  Because my refactoring tends to be quite invasive, it would probably be a nightmare to bring them up to date.  With NR in master, it's trivial for me to keep it up to date as I go.

Incidentally, you might be interested in having a look at the histograms PR (#245), I think it will fit https://github.com/newrelic/newrelic-exporter-specs/blob/master/Guidelines.md#histograms with a bit of gentle massaging.  We're still not certain if it's going to make it in, or remain in permanently, but it's a thing.
tiedotguy: @ash2k, if you've got the time, can I get another set of eyes over this sometime before I hit merge?
Kav91: @tiedotguy @ash2k just a friendly poke, or if you have any idea when someone can take a look roughly?
We are looking to make this our go to option for StatsD once merged, and want to use the latest on master to build some images and binaries for our users 😄so any timing you can provide would be appreciated. 
ash2k: I don't know anything specific about NewRelic but I'll have a look tomorrow. Sorry for the delay.
Kav91: hey @ash2k sorry to nudge again, any luck or are we looking at next week sometime? Thx
Kav91: No worries, & thanks for taking the time to review again @ash2k & @tiedotguy 😄 


Patch:
diff --git a/BACKENDS.md b/BACKENDS.md
index eea748a..0d3abee 100644
--- a/BACKENDS.md
+++ b/BACKENDS.md
@@ -63,28 +63,71 @@ When `mode` is `legacy`, the graphite backend will emit metrics with the followi
 
 New Relic Backend
 -----------------
-Supports two routes for flushing metrics to New Relic.
-- Directly to the Insights Collector - [Insights Event API](https://docs.newrelic.com/docs/insights/insights-data-sources/custom-data/send-custom-events-event-api)
-- Via the Infrastructure Agent's inbuilt HTTP Server
+Supports three flush-types for sending metrics to New Relic:
+- `infra` Via the Infrastructure Agent's inbuilt HTTP Server (default)
+- `insights` Directly to the Insights Collector - [Insights Event API](https://docs.newrelic.com/docs/insights/insights-data-sources/custom-data/send-custom-events-event-api)
+- `metrics` Directly to the Metric API - [Metric Event API](https://docs.newrelic.com/docs/data-ingest-apis/get-data-new-relic/metric-api/introduction-metric-api)
+
+An attempt will be made to automatically set the `flush-type` based on the address you have configured, [manual configuration](#manual-flush-type-configuration) is also available.
 
 ### [New Relic Insights Event API](https://docs.newrelic.com/docs/insights/insights-data-sources/custom-data/send-custom-events-event-api)
-Sending directly to the Event API alleviates the requirement of needing to have the New Relic Infrastructure Agent. Therefore you can run this from nearly anywhere for maximum flexibility. This also becomes a shorter data path with less resource requirements becoming a simpler setup.
+Sending directly to the Event API alleviates the requirement of needing to have the New Relic Infrastructure Agent.
+Therefore you can run this from nearly anywhere for maximum flexibility. This also becomes a shorter data path with less
+resource requirements becoming a simpler setup.
+
+To use this method, create an Insert API Key from here: https://insights.newrelic.com/accounts/YOUR_ACCOUNT_ID/manage/api_keys.
+More information on registering an Insert API Key can be found [here](https://docs.newrelic.com/docs/insights/insights-data-sources/custom-data/introduction-event-api#register).
+
+```
+#Example Insights Collector configuration
+
+[newrelic]
+    transport = "default"
+    address = "https://insights-collector.newrelic.com/v1/accounts/YOUR_ACCOUNT_ID/events"
+    api-key = "yourEventAPIInsertKey"
+```
+
+See [TRANSPORT.md](TRANSPORT.md) for information about `transport`.
+
+> Make sure that `address` is set to `https://insights-collector.eu01.nr-data.net/v1/accounts/YOUR_ACCOUNT_ID/events` if your account hosts data in the EU data center.
+
+### [New Relic's Metric API](https://docs.newrelic.com/docs/data-ingest-apis/get-data-new-relic/metric-api/introduction-metric-api)
+To enable sending metrics to the Metric API endpoint you need to set the config setting `address-metrics` to
+`https://metric-api.newrelic.com/metric/v1`. These events appear in [New Relic Insights](https://docs.newrelic.com/docs/insights/use-insights-ui/getting-started/introduction-new-relic-insights)
+as a new [event type](https://docs.newrelic.com/docs/insights/new-relic-insights/understanding-insights/new-relic-insights#event-type).
+These new events can then be used to leverage the power of [NRQL](https://docs.newrelic.com/docs/query-data/nrql-new-relic-query-language/getting-started/introduction-nrql)
+for [querying the metric data](https://docs.newrelic.com/docs/data-ingest-apis/get-data-new-relic/metric-api/introduction-metric-api#view-and-query).
 
 To use this method, create an Insert API Key from here: https://insights.newrelic.com/accounts/YOUR_ACCOUNT_ID/manage/api_keys
 
 ```
-#Example configuration
+#Example Metric API configuration
+
+[newrelic]
+    transport = "default"
+    address-metrics = "https://metric-api.newrelic.com/metric/v1"
+    api-key = "yourEventAPIInsertKey"
+```
+
+> Make sure that `address-metrics` is set to `https://metric-api.eu.newrelic.com/metric/v1` if your account hosts data in the EU data center.
+
+It is also recommended to set up the backend to either use the [Insight Collector](#new-relic-insights-event-api) or the
+[Infrastructure Agent](#new-relic-infrastructure-agent) to send none metric events back to New Relic.
+
+```
+#Example Metric API configuration with Insights Collector support
 
 [newrelic]
     transport = "default"
     address = "https://insights-collector.newrelic.com/v1/accounts/YOUR_ACCOUNT_ID/events"
+    address-metrics = "https://metric-api.newrelic.com/metric/v1"
     api-key = "yourEventAPIInsertKey"
 ```
 
-See [TRANSPORT.md] for information about `transport`.
 
 ### [New Relic Infrastructure Agent](https://newrelic.com/products/infrastructure)
-Sending via the Infrastructure Agent's inbuilt HTTP server provides additional features, such as automatically applying additional metadata to the event the host may have such as AWS tags, instance type, host information, labels etc.
+Sending via the Infrastructure Agent's inbuilt HTTP server provides additional features, such as automatically applying
+additional metadata to the event the host may have such as AWS tags, instance type, host information, labels etc.
 
 The payload structure required to be accepted by the agent can be viewed [here.](https://github.com/newrelic/infra-integrations-sdk/blob/master/docs/v2tov3.md#v2-json-full-sample)
 
@@ -95,6 +138,20 @@ http_server_host: 127.0.0.1 #(default host)
 http_server_port: 8001 #(default port)
 ```
 
+### Manual Flush Type Configuration
+The `flush-type` attribute can be configured with the following available  options - `insights`, `metrics` or `infra`.
+
+```
+[newrelic]
+    transport = "default"
+	flush-type = "insights" # <-- 
+    address = "https://another-collector.newrelic.com/v1/accounts/YOUR_ACCOUNT_ID/events"
+    api-key = "yourEventAPIInsertKey"
+```
+
+### Renaming Attributes
+This option is only available for `insights` and `infra` flush-types.
+
 Additional options are available to rename attributes if required.
 ```
 [newrelic]
diff --git a/pkg/backends/newrelic/flush.go b/pkg/backends/newrelic/flush.go
index 48a10d0..36c4a48 100644
--- a/pkg/backends/newrelic/flush.go
+++ b/pkg/backends/newrelic/flush.go
@@ -1,6 +1,10 @@
 package newrelic
 
 import (
+	"fmt"
+	"strconv"
+	"strings"
+
 	"github.com/atlassian/gostatsd"
 )
 
@@ -20,48 +24,107 @@ type timeSeries struct {
 
 // addMetric adds a metric to the series.
 func (f *flush) addMetric(n *Client, metricType string, value float64, persecond float64, hostname string, tags gostatsd.Tags, name string, timestamp gostatsd.Nanotime) {
-	standardMetric := newMetricSet(n, f, name, metricType, value, tags, timestamp)
-	if metricType == "counter" {
-		standardMetric[n.metricPerSecond] = persecond
+	if n.flushType == flushTypeMetrics {
+		metricName := name
+		if metricType == "counter" {
+			perSecondMetric := newDimensionalMetricSet(n, f, metricName+".per_second", "gauge", persecond, tags, timestamp)
+			f.ts.Metrics = append(f.ts.Metrics, perSecondMetric)
+		}
+		standardMetric := newDimensionalMetricSet(n, f, name, metricType, value, tags, timestamp)
+		f.ts.Metrics = append(f.ts.Metrics, standardMetric)
+	} else {
+		standardMetric := newMetricSet(n, f, name, metricType, value, tags, timestamp)
+		if metricType == "counter" {
+			standardMetric[n.metricPerSecond] = persecond
+		}
+		f.ts.Metrics = append(f.ts.Metrics, standardMetric)
 	}
-	f.ts.Metrics = append(f.ts.Metrics, standardMetric)
 }
 
 // addMetric adds a timer metric to the series.
 func (f *flush) addTimerMetric(n *Client, metricType string, timer gostatsd.Timer, tagsKey, name string) {
-	timerMetric := newMetricSet(n, f, name, metricType, float64(timer.Count), timer.Tags, timer.Timestamp)
+	if n.flushType == flushTypeMetrics {
+		timerMetric := newDimensionalMetricSet(n, f, name, metricType, float64(timer.Count), timer.Tags, timer.Timestamp)
 
-	if !n.disabledSubtypes.Lower {
-		timerMetric[n.timerMin] = timer.Min
-	}
-	if !n.disabledSubtypes.Upper {
-		timerMetric[n.timerMax] = timer.Max
-	}
-	if !n.disabledSubtypes.Count {
-		timerMetric[n.timerCount] = float64(timer.Count)
-	}
-	if !n.disabledSubtypes.CountPerSecond {
-		timerMetric[n.metricPerSecond] = timer.PerSecond
-	}
-	if !n.disabledSubtypes.Mean {
-		timerMetric[n.timerMean] = timer.Mean
-	}
-	if !n.disabledSubtypes.Median {
-		timerMetric[n.timerMedian] = timer.Median
-	}
-	if !n.disabledSubtypes.StdDev {
-		timerMetric[n.timerStdDev] = timer.StdDev
-	}
-	if !n.disabledSubtypes.Sum {
-		timerMetric[n.timerSum] = timer.Sum
-	}
-	if !n.disabledSubtypes.SumSquares {
-		timerMetric[n.timerSumSquares] = timer.SumSquares
-	}
-	for _, pct := range timer.Percentiles {
-		timerMetric[pct.Str] = pct.Float
+		timerMetric.Value = map[string]float64{
+			"count": float64(timer.Count),
+			"sum":   timer.Sum,
+			"min":   timer.Min,
+			"max":   timer.Max,
+		}
+
+		if !n.disabledSubtypes.CountPerSecond {
+			gaugeMetric := newDimensionalMetricSet(n, f, name+".per_second", "gauge", timer.PerSecond, timer.Tags, timer.Timestamp)
+			f.ts.Metrics = append(f.ts.Metrics, gaugeMetric)
+		}
+		if !n.disabledSubtypes.Mean {
+			gaugeMetric := newDimensionalMetricSet(n, f, name+".mean", "gauge", timer.Mean, timer.Tags, timer.Timestamp)
+			f.ts.Metrics = append(f.ts.Metrics, gaugeMetric)
+		}
+		if !n.disabledSubtypes.Median {
+			gaugeMetric := newDimensionalMetricSet(n, f, name+".median", "gauge", timer.Median, timer.Tags, timer.Timestamp)
+			f.ts.Metrics = append(f.ts.Metrics, gaugeMetric)
+		}
+		if !n.disabledSubtypes.StdDev {
+			gaugeMetric := newDimensionalMetricSet(n, f, name+".std_dev", "gauge", timer.StdDev, timer.Tags, timer.Timestamp)
+			f.ts.Metrics = append(f.ts.Metrics, gaugeMetric)
+		}
+		if !n.disabledSubtypes.SumSquares {
+			gaugeMetric := newDimensionalMetricSet(n, f, name+".sum_squares", "gauge", timer.SumSquares, timer.Tags, timer.Timestamp)
+			f.ts.Metrics = append(f.ts.Metrics, gaugeMetric)
+		}
+
+		// https://github.com/newrelic/newrelic-exporter-specs/blob/master/Guidelines.md#percentiles
+		// Percentiles MUST be implemented as a set of metrics, one metric for each percentile.
+		// Each metric MUST be a Gauge metric.
+		// Each metric value MUST represent a calculated percentile.
+		// The percentile attribute MUST be included with each metric. This attribute value MUST represent the percentile being measured as a floating-point number within the range [0.0, 100.0].
+		// Each metric name MUST have a ".percentiles" suffix.
+		for _, pct := range timer.Percentiles {
+			lastUnderscore := strings.LastIndex(pct.Str, "_")
+			gaugeMetric := newDimensionalMetricSet(n, f, fmt.Sprintf("%v.%v.percentiles", name, pct.Str[:lastUnderscore]), "gauge", pct.Float, timer.Tags, timer.Timestamp)
+			percentileResult, err := strconv.ParseFloat(pct.Str[lastUnderscore+1:], 64) // eg. for sum_squares_90 will return 90
+			if err == nil {
+				gaugeMetric.Attributes["percentile"] = percentileResult
+				f.ts.Metrics = append(f.ts.Metrics, gaugeMetric)
+			}
+		}
+		f.ts.Metrics = append(f.ts.Metrics, timerMetric)
+	} else {
+		timerMetric := newMetricSet(n, f, name, metricType, float64(timer.Count), timer.Tags, timer.Timestamp)
+
+		if !n.disabledSubtypes.Lower {
+			timerMetric[n.timerMin] = timer.Min
+		}
+		if !n.disabledSubtypes.Upper {
+			timerMetric[n.timerMax] = timer.Max
+		}
+		if !n.disabledSubtypes.Count {
+			timerMetric[n.timerCount] = float64(timer.Count)
+		}
+		if !n.disabledSubtypes.Sum {
+			timerMetric[n.timerSum] = timer.Sum
+		}
+		if !n.disabledSubtypes.CountPerSecond {
+			timerMetric[n.metricPerSecond] = timer.PerSecond
+		}
+		if !n.disabledSubtypes.Mean {
+			timerMetric[n.timerMean] = timer.Mean
+		}
+		if !n.disabledSubtypes.Median {
+			timerMetric[n.timerMedian] = timer.Median
+		}
+		if !n.disabledSubtypes.StdDev {
+			timerMetric[n.timerStdDev] = timer.StdDev
+		}
+		if !n.disabledSubtypes.SumSquares {
+			timerMetric[n.timerSumSquares] = timer.SumSquares
+		}
+		for _, pct := range timer.Percentiles {
+			timerMetric[pct.Str] = pct.Float
+		}
+		f.ts.Metrics = append(f.ts.Metrics, timerMetric)
 	}
-	f.ts.Metrics = append(f.ts.Metrics, timerMetric)
 }
 
 func (f *flush) maybeFlush() {
@@ -81,22 +144,23 @@ func (f *flush) finish() {
 
 func newMetricSet(n *Client, f *flush, metricName, Type string, Value float64, tags gostatsd.Tags, timestamp gostatsd.Nanotime) map[string]interface{} {
 	metricSet := map[string]interface{}{}
-	metricSet["interval"] = f.flushIntervalSec
-	metricSet["integration_version"] = integrationVersion
+
 	// GoStatsD provides the timestamp in Nanotime, New Relic requires seconds or milliseconds, see under "Limits and restricted characters"
 	// https://docs.newrelic.com/docs/insights/insights-data-sources/custom-data/send-custom-events-event-api#instrument
 	metricSet["timestamp"] = timestamp / 1e9
+	metricSet["interval"] = f.flushIntervalSec
+	metricSet["integration_version"] = integrationVersion
 
 	// New Relic Insights Event API, expects the "Event Type" to be in camel case format as opposed to an underscore with the Infrastructure Payload
 	// https://docs.newrelic.com/docs/insights/insights-data-sources/custom-data/send-custom-events-event-api#instrument
 	// https://github.com/newrelic/infra-integrations-sdk/blob/master/docs/v2tov3.md#v2-json-full-sample
+
 	switch n.flushType {
 	case flushTypeInsights:
 		metricSet["eventType"] = n.eventType
 	default:
 		metricSet["event_type"] = n.eventType
 	}
-
 	metricSet[n.metricType] = Type
 	metricSet[n.metricName] = metricName
 	metricSet[n.metricValue] = Value
@@ -104,3 +168,31 @@ func newMetricSet(n *Client, f *flush, metricName, Type string, Value float64, t
 
 	return metricSet
 }
+
+// create a new metric set for the dimensional metrics api
+func newDimensionalMetricSet(n *Client, f *flush, metricName, Type string, Value float64, tags gostatsd.Tags, timestamp gostatsd.Nanotime) NRMetric {
+	// GoStatsD provides the timestamp in Nanotime, New Relic requires seconds or milliseconds, see under "Limits and restricted characters"
+	// https://docs.newrelic.com/docs/insights/insights-data-sources/custom-data/send-custom-events-event-api#instrument
+	metricSet := NRMetric{
+		Name:      metricName,
+		Timestamp: int64(timestamp) / 1e9,
+		Attributes: map[string]interface{}{
+			"statsdType": Type,
+		},
+	}
+
+	n.setTags(tags, metricSet.Attributes)
+	switch Type {
+	case "timer":
+		metricSet.Type = "summary"
+		metricSet.Name = metricSet.Name + ".summary"
+	case "counter":
+		metricSet.Type = "count"
+		metricSet.Value = Value
+	case "gauge":
+		metricSet.Type = Type
+		metricSet.Value = Value
+	}
+
+	return metricSet
+}
diff --git a/pkg/backends/newrelic/newrelic.go b/pkg/backends/newrelic/newrelic.go
index d792031..468dd49 100644
--- a/pkg/backends/newrelic/newrelic.go
+++ b/pkg/backends/newrelic/newrelic.go
@@ -2,7 +2,7 @@ package newrelic
 
 import (
 	"bytes"
-	"compress/zlib"
+	"compress/gzip"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -29,7 +29,7 @@ const (
 	// BackendName is the name of this backend.
 	BackendName                  = "newrelic"
 	integrationName              = "com.newrelic.gostatsd"
-	integrationVersion           = "2.2.0"
+	integrationVersion           = "2.3.0"
 	protocolVersion              = "2"
 	defaultUserAgent             = "gostatsd"
 	defaultMaxRequestElapsedTime = 15 * time.Second
@@ -40,11 +40,12 @@ const (
 
 	flushTypeInsights = "insights"
 	flushTypeInfra    = "infra"
+	flushTypeMetrics  = "metrics"
 )
 
 var (
 	// Available flushTypes
-	flushTypes = []string{flushTypeInsights, flushTypeInfra}
+	flushTypes = []string{flushTypeInsights, flushTypeInfra, flushTypeMetrics}
 
 	// defaultMaxRequests is the number of parallel outgoing requests to New Relic.  As this mixes both
 	// CPU (JSON encoding, TLS) and network bound operations, balancing may require some experimentation.
@@ -53,11 +54,12 @@ var (
 
 // Client represents a New Relic client.
 type Client struct {
-	address   string
-	eventType string
-	flushType string
-	apiKey    string
-	tagPrefix string
+	address        string
+	addressMetrics string
+	eventType      string
+	flushType      string
+	apiKey         string
+	tagPrefix      string
 	// Options to define your own field names to support other StatsD implementations
 	metricName      string
 	metricType      string
@@ -88,14 +90,38 @@ type Client struct {
 	flushInterval    time.Duration
 }
 
-// NewRelicInfraPayload struct
-type NewRelicInfraPayload struct {
+// NRInfraPayload represents New Relic Infrastructure Payload format
+// https://github.com/newrelic/infra-integrations-sdk/blob/master/docs/v2tov3.md#v2-json-full-sample
+type NRInfraPayload struct {
 	Name               string        `json:"name"`
 	ProtocolVersion    string        `json:"protocol_version"`
 	IntegrationVersion string        `json:"integration_version"`
 	Data               []interface{} `json:"data"`
 }
 
+// NRMetricsPayload represents New Relic Metrics Payload format
+// https://docs.newrelic.com/docs/data-ingest-apis/get-data-new-relic/metric-api/report-metrics-metric-api#new-relic-guidelines
+type NRMetricsPayload struct {
+	Common  NRMetricsCommon `json:"common"`
+	Metrics []interface{}   `json:"metrics"`
+}
+
+// NRMetricsCommon common attributes to apply for New Relic Metrics Format
+type NRMetricsCommon struct {
+	Attributes map[string]interface{} `json:"attributes"`
+	IntervalMs float64                `json:"interval.ms"`
+}
+
+// NRMetric metric for New Relic Metrics Format
+type NRMetric struct {
+	Name       string                 `json:"name"`
+	Value      interface{}            `json:"value,omitempty"`
+	Type       string                 `json:"type,omitempty"`
+	Timestamp  int64                  `json:"timestamp,omitempty"`
+	Attributes map[string]interface{} `json:"attributes,omitempty"`
+	IntervalMs float64                `json:"interval.ms,omitempty"`
+}
+
 // SendMetricsAsync flushes the metrics to New Relic, preparing payload synchronously but doing the send asynchronously.
 func (n *Client) SendMetricsAsync(ctx context.Context, metrics *gostatsd.MetricMap, cb gostatsd.SendCallback) {
 
@@ -141,6 +167,7 @@ func (n *Client) SendMetricsAsync(ctx context.Context, metrics *gostatsd.MetricM
 	}()
 }
 
+// RunMetrics run metrics
 func (n *Client) RunMetrics(ctx context.Context, statser stats.Statser) {
 	statser = statser.WithTags(gostatsd.Tags{"backend:newrelic"})
 
@@ -196,19 +223,22 @@ 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 {
-	// Event format depends on flush type
-	data := n.EventFormatter(e)
-	b, err := json.Marshal(data)
-	if err != nil {
-		return err
-	}
+	if n.address != "" {
+		// 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
-	}
+		post, err := n.postWrapper(ctx, b, "events")
+		if err != nil {
+			return err
+		}
 
-	return post()
+		return post()
+	}
+	return nil
 }
 
 // EventFormatter formats gostatsd events
@@ -227,7 +257,7 @@ func (n *Client) EventFormatter(e *gostatsd.Event) interface{} {
 	n.setTags(e.Tags, event)
 
 	switch n.flushType {
-	case flushTypeInsights:
+	case flushTypeInsights, flushTypeMetrics:
 		event["eventType"] = n.eventType
 		return []interface{}{event}
 	default:
@@ -285,6 +315,9 @@ func (n *Client) constructPost(ctx context.Context, buffer *bytes.Buffer, data i
 	case flushTypeInsights:
 		NRPayload := data.(*timeSeries).Metrics
 		mJSON, mErr = json.Marshal(NRPayload)
+	case flushTypeMetrics:
+		NRPayload := n.newMetricsPayload(data.(*timeSeries).Metrics)
+		mJSON, mErr = json.Marshal([]interface{}{NRPayload})
 	default:
 		NRPayload := newInfraPayload(data)
 		mJSON, mErr = json.Marshal(NRPayload)
@@ -294,11 +327,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)
+	return n.postWrapper(ctx, mJSON, "metrics")
 }
 
 // postWrapper compresses JSON for Insights
-func (n *Client) postWrapper(ctx context.Context, json []byte) (func() error, error) {
+func (n *Client) postWrapper(ctx context.Context, json []byte, dataType string) (func() error, error) {
 	return func() error {
 		headers := map[string]string{
 			"Content-Type": "application/json",
@@ -306,12 +339,17 @@ func (n *Client) postWrapper(ctx context.Context, json []byte) (func() error, er
 		}
 
 		// Insights Event API requires gzip or deflate compression
-		if n.flushType == flushTypeInsights && n.apiKey != "" {
+		// https://docs.newrelic.com/docs/insights/insights-data-sources/custom-data/introduction-event-api#h2-basic-workflow
+		// Metrics API requires gzip or identity
+		// https://docs.newrelic.com/docs/data-ingest-apis/get-data-new-relic/metric-api/report-metrics-metric-api#headers-query-parameters
+		// Use GZIP as standard across both
+		if (n.flushType == flushTypeInsights || n.flushType == flushTypeMetrics) && n.apiKey != "" {
 			headers["X-Insert-Key"] = n.apiKey
-			headers["Content-Encoding"] = "deflate"
-			// zlib json
+			headers["Content-Encoding"] = "gzip"
+
+			// compress json
 			var buf bytes.Buffer
-			zw := zlib.NewWriter(&buf)
+			zw := gzip.NewWriter(&buf)
 			_, err := zw.Write([]byte(json))
 			if err != nil {
 				return err
@@ -324,7 +362,12 @@ func (n *Client) postWrapper(ctx context.Context, json []byte) (func() error, er
 			json = buf.Bytes()
 		}
 
-		req, err := http.NewRequest("POST", n.address, bytes.NewBuffer(json))
+		address := n.address
+		if n.flushType == flushTypeMetrics && dataType == "metrics" {
+			address = n.addressMetrics
+		}
+
+		req, err := http.NewRequest("POST", address, bytes.NewBuffer(json))
 		if err != nil {
 			return fmt.Errorf("unable to create http.Request: %v", err)
 		}
@@ -358,11 +401,15 @@ func NewClientFromViper(v *viper.Viper, pool *transport.TransportPool) (gostatsd
 	nr.SetDefault("transport", "default")
 	nr.SetDefault("address", "http://localhost:8001/v1/data")
 	nr.SetDefault("event-type", "GoStatsD")
-	if strings.Contains(nr.GetString("address"), "insights-collector.newrelic.com") {
+
+	if strings.Contains(nr.GetString("address-metrics"), "metric-api.newrelic.com") {
+		nr.SetDefault("flush-type", flushTypeMetrics)
+	} else if strings.Contains(nr.GetString("address"), "insights-collector.newrelic.com") {
 		nr.SetDefault("flush-type", flushTypeInsights)
 	} else {
 		nr.SetDefault("flush-type", flushTypeInfra)
 	}
+
 	nr.SetDefault("api-key", "")
 	nr.SetDefault("tag-prefix", "")
 	nr.SetDefault("metric-name", "name")
@@ -393,6 +440,7 @@ func NewClientFromViper(v *viper.Viper, pool *transport.TransportPool) (gostatsd
 	return NewClient(
 		nr.GetString("transport"),
 		nr.GetString("address"),
+		nr.GetString("address-metrics"),
 		nr.GetString("event-type"),
 		nr.GetString("flush-type"),
 		nr.GetString("api-key"),
@@ -420,7 +468,7 @@ func NewClientFromViper(v *viper.Viper, pool *transport.TransportPool) (gostatsd
 }
 
 // NewClient returns a new New Relic client.
-func NewClient(transport, address, eventType, flushType, apiKey, tagPrefix,
+func NewClient(transport, address, addressMetrics, eventType, flushType, apiKey, tagPrefix,
 	metricName, metricType, metricPerSecond, metricValue,
 	timerMin, timerMax, timerCount, timerMean, timerMedian, timerStdDev, timerSum, timerSumSquares,
 	userAgent string, metricsPerBatch int, maxRequests uint,
@@ -438,11 +486,11 @@ func NewClient(transport, address, eventType, flushType, apiKey, tagPrefix,
 	if !contains(flushTypes, flushType) && flushType != "" {
 		return nil, fmt.Errorf("[%s] flushType (%s) is not supported", BackendName, flushType)
 	}
-	if flushType == flushTypeInsights && apiKey == "" {
-		return nil, fmt.Errorf("[%s] api-key is required to flush to insights", BackendName)
+	if (flushType == flushTypeInsights || flushType == flushTypeMetrics) && apiKey == "" {
+		return nil, fmt.Errorf("[%s] api-key is required to flush to insights & metrics backends", BackendName)
 	}
-	if flushType != flushTypeInsights && apiKey != "" {
-		logger.Warnf("api-key is not required when not using insights")
+	if flushType != flushTypeInsights && flushType != flushTypeMetrics && apiKey != "" {
+		logger.Warnf("api-key is not required when not using insights or metrics")
 	}
 	if flushInterval.Seconds() < 10 {
 		logger.Warnf("flushInterval (%s) is recommended to be >= 10s", flushInterval)
@@ -468,6 +516,7 @@ func NewClient(transport, address, eventType, flushType, apiKey, tagPrefix,
 	}
 	return &Client{
 		address:               address,
+		addressMetrics:        addressMetrics,
 		eventType:             eventType,
 		flushType:             flushType,
 		apiKey:                apiKey,
@@ -495,8 +544,8 @@ func NewClient(transport, address, eventType, flushType, apiKey, tagPrefix,
 	}, nil
 }
 
-func newInfraPayload(data interface{}) NewRelicInfraPayload {
-	return NewRelicInfraPayload{
+func newInfraPayload(data interface{}) NRInfraPayload {
+	return NRInfraPayload{
 		Name:               integrationName,
 		ProtocolVersion:    protocolVersion,
 		IntegrationVersion: integrationVersion,
@@ -506,6 +555,19 @@ func newInfraPayload(data interface{}) NewRelicInfraPayload {
 	}
 }
 
+func (n *Client) newMetricsPayload(data []interface{}) NRMetricsPayload {
+	return NRMetricsPayload{
+		Common: NRMetricsCommon{
+			Attributes: map[string]interface{}{
+				"integration.version": integrationVersion,
+				"integration.name":    "GoStatsD",
+			},
+			IntervalMs: n.flushInterval.Seconds() * 1000,
+		},
+		Metrics: data,
+	}
+}
+
 func (n *Client) setTags(tags gostatsd.Tags, data map[string]interface{}) {
 	for _, tag := range tags {
 		if strings.Contains(tag, ":") {


Test patch:
diff --git a/pkg/backends/newrelic/newrelic_test.go b/pkg/backends/newrelic/newrelic_test.go
index 1201546..e2d7faf 100644
--- a/pkg/backends/newrelic/newrelic_test.go
+++ b/pkg/backends/newrelic/newrelic_test.go
@@ -1,6 +1,7 @@
 package newrelic
 
 import (
+	"compress/gzip"
 	"context"
 	"encoding/json"
 	"io/ioutil"
@@ -44,7 +45,7 @@ func TestRetries(t *testing.T) {
 	v.SetDefault("transport.default.client-timeout", 1*time.Second)
 	p := transport.NewTransportPool(logrus.New(), v)
 
-	client, err := NewClient("default", ts.URL+"/v1/data", "GoStatsD", "", "", "", "metric_name", "metric_type",
+	client, err := NewClient("default", ts.URL+"/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",
 		defaultMetricsPerBatch, defaultMaxRequests, 2*time.Second, 1*time.Second, gostatsd.TimerSubtypes{}, p)
@@ -81,7 +82,7 @@ func TestSendMetricsInMultipleBatches(t *testing.T) {
 	v.SetDefault("transport.default.client-timeout", 1*time.Second)
 	p := transport.NewTransportPool(logrus.New(), v)
 
-	client, err := NewClient("default", ts.URL+"/v1/data", "GoStatsD", "", "", "", "metric_name", "metric_type",
+	client, err := NewClient("default", ts.URL+"/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",
 		1, defaultMaxRequests, 2*time.Second, 1*time.Second, gostatsd.TimerSubtypes{}, p)
@@ -99,44 +100,119 @@ func TestSendMetricsInMultipleBatches(t *testing.T) {
 
 func TestSendMetrics(t *testing.T) {
 	t.Parallel()
-	mux := http.NewServeMux()
-	mux.HandleFunc("/v1/data", func(w http.ResponseWriter, r *http.Request) {
-		data, err := ioutil.ReadAll(r.Body)
-		if !assert.NoError(t, err) {
-			return
-		}
-		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))
-	})
-	ts := httptest.NewServer(mux)
-	defer ts.Close()
+	tests := []struct {
+		name      string
+		expected  string
+		flushType string
+		apiKey    string
+	}{
+		{
+			name:      "infra",
+			flushType: "infra",
+			apiKey:    "",
+			expected: `{"name":"com.newrelic.gostatsd","protocol_version":"2","integration_version":"2.3.0","data":[{"metrics":` +
+				`[{"event_type":"GoStatsD","integration_version":"2.3.0","interval":1,"metric_name":"g1","metric_type":"gauge","metric_value":3,"tag3":"true","timestamp":0},` +
+				`{"event_type":"GoStatsD","integration_version":"2.3.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.3.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.3.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}]}]}`,
+		},
+		{
+			name:      "insights",
+			flushType: "insights",
+			apiKey:    "some-api-key",
+			expected: `[{"eventType":"GoStatsD","integration_version":"2.3.0","interval":1,"metric_name":"g1","metric_type":"gauge","metric_value":3,"tag3":"true","timestamp":0},` +
+				`{"eventType":"GoStatsD","integration_version":"2.3.0","interval":1,"metric_name":"c1","metric_per_second":1.1,"metric_type":"counter","metric_value":5,"tag1":"true","timestamp":0},` +
+				`{"eventType":"GoStatsD","integration_version":"2.3.0","interval":1,"metric_name":"users","metric_type":"set","metric_value":3,"tag4":"true","timestamp":0},` +
+				`{"count_90":0.1,"eventType":"GoStatsD","integration_version":"2.3.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}]`,
+		},
+		{
+			name:      "metrics",
+			flushType: "metrics",
+			apiKey:    "some-api-key",
+			expected: `[{"common":{"attributes":{"integration.name":"GoStatsD","integration.version":"2.3.0"},"interval.ms":1000},` +
+				`"metrics":[{"name":"g1","value":3,"type":"gauge","attributes":{"statsdType":"gauge","tag3":"true"}},` +
+				`{"name":"c1.per_second","value":1.1,"type":"gauge","attributes":{"statsdType":"gauge","tag1":"true"}},` +
+				`{"name":"c1","value":5,"type":"count","attributes":{"statsdType":"counter","tag1":"true"}},` +
+				`{"name":"users","attributes":{"statsdType":"set","tag4":"true"}},{"name":"t1.per_second","value":1.1,"type":"gauge","attributes":{"statsdType":"gauge","tag2":"true"}},` +
+				`{"name":"t1.mean","value":0.5,"type":"gauge","attributes":{"statsdType":"gauge","tag2":"true"}},` +
+				`{"name":"t1.median","value":0.5,"type":"gauge","attributes":{"statsdType":"gauge","tag2":"true"}},` +
+				`{"name":"t1.std_dev","value":0.1,"type":"gauge","attributes":{"statsdType":"gauge","tag2":"true"}},` +
+				`{"name":"t1.sum_squares","value":1,"type":"gauge","attributes":{"statsdType":"gauge","tag2":"true"}},` +
+				`{"name":"t1.count.percentiles","value":0.1,"type":"gauge","attributes":{"percentile":90,"statsdType":"gauge","tag2":"true"}},` +
+				`{"name":"t1.summary","value":{"count":1,"max":1,"min":0,"sum":1},"type":"summary","attributes":{"statsdType":"timer","tag2":"true"}}]}]`,
+		},
+	}
 
-	v := viper.New()
-	v.SetDefault("transport.default.client-timeout", 1*time.Second)
-	p := transport.NewTransportPool(logrus.New(), v)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
 
-	client, err := NewClient("default", ts.URL+"/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",
-		defaultMetricsPerBatch, defaultMaxRequests, 2*time.Second, 1*time.Second, gostatsd.TimerSubtypes{}, p)
+			mux := http.NewServeMux()
+			path := "/v1/data"
+			if tt.flushType == "metrics" {
+				path = "/metric/v1"
+			}
+			mux.HandleFunc(path, func(w http.ResponseWriter, r *http.Request) {
+				enc := r.Header.Get("Content-Encoding")
+				var body string
+				body, done := decodeBody(enc, r, t)
+				if done {
+					return
+				}
+				assert.Equal(t, tt.expected, body)
+				assert.Equal(t, tt.apiKey, r.Header.Get("X-Insert-Key"))
+			})
+			ts := httptest.NewServer(mux)
+			defer ts.Close()
 
-	require.NoError(t, err)
-	client.now = func() time.Time {
-		return time.Unix(100, 0)
+			v := viper.New()
+			v.SetDefault("transport.default.client-timeout", 1*time.Second)
+			p := transport.NewTransportPool(logrus.New(), v)
+
+			client, err := NewClient("default", ts.URL+"/v1/data", ts.URL+"/metric/v1", "GoStatsD", tt.flushType, tt.apiKey, "", "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",
+				defaultMetricsPerBatch, defaultMaxRequests, 2*time.Second, 1*time.Second, gostatsd.TimerSubtypes{}, p)
+
+			require.NoError(t, err)
+			client.now = func() time.Time {
+				return time.Unix(100, 0)
+			}
+			res := make(chan []error, 1)
+			client.SendMetricsAsync(context.Background(), metricsOneOfEach(), func(errs []error) {
+				res <- errs
+			})
+			errs := <-res
+			for _, err := range errs {
+				assert.NoError(t, err)
+			}
+		})
 	}
-	res := make(chan []error, 1)
-	client.SendMetricsAsync(context.Background(), metricsOneOfEach(), func(errs []error) {
-		res <- errs
-	})
-	errs := <-res
-	for _, err := range errs {
-		assert.NoError(t, err)
+
+}
+
+func decodeBody(enc string, r *http.Request, t *testing.T) (string, bool) {
+	body := ""
+	if enc == "gzip" {
+		gr, err := gzip.NewReader(r.Body)
+		if !assert.NoError(t, err) {
+			t.FailNow()
+			return "", true
+		}
+		data, err := ioutil.ReadAll(gr)
+		if !assert.NoError(t, err) {
+			return "", true
+		}
+		body = string(data)
+	} else {
+		data, err := ioutil.ReadAll(r.Body)
+		if !assert.NoError(t, err) {
+			return "", true
+		}
+		body = string(data)
 	}
+	return body, false
 }
 
 // twoCounters returns two counters.
@@ -207,23 +283,42 @@ func metricsOneOfEach() *gostatsd.MetricMap {
 func TestEventFormatter(t *testing.T) {
 	t.Parallel()
 
-	v := viper.New()
-	v.SetDefault("transport.default.client-timeout", 1*time.Second)
-	p := transport.NewTransportPool(logrus.New(), v)
-
-	client, err := NewClient("default", "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",
-		defaultMetricsPerBatch, defaultMaxRequests, 2*time.Second, 1*time.Second, gostatsd.TimerSubtypes{}, p)
-	require.NoError(t, err)
+	tests := []struct {
+		name     string
+		expected string
+	}{
+		{
+			name: "infra",
+			expected: `{"name":"com.newrelic.gostatsd","protocol_version":"2","integration_version":"2.3.0","data":` +
+				`[{"metrics":[{"AggregationKey":"","AlertType":"","DateHappened":0,"Hostname":"blah","Priority":"low","SourceTypeName":"","Text":"hi","Title":"EventTitle","event_type":"GoStatsD","name":"event"}]}]}`,
+		},
+		{
+			name:     "insights",
+			expected: `[{"AggregationKey":"","AlertType":"","DateHappened":0,"Hostname":"blah","Priority":"low","SourceTypeName":"","Text":"hi","Title":"EventTitle","eventType":"GoStatsD","name":"event"}]`,
+		},
+		{
+			name:     "metrics",
+			expected: `[{"AggregationKey":"","AlertType":"","DateHappened":0,"Hostname":"blah","Priority":"low","SourceTypeName":"","Text":"hi","Title":"EventTitle","eventType":"GoStatsD","name":"event"}]`,
+		},
+	}
 
-	gostatsdEvent := gostatsd.Event{Title: "EventTitle", Text: "hi", Hostname: "blah", Priority: 1}
-	formattedEvent := client.EventFormatter(&gostatsdEvent)
-	fevent, err := json.Marshal(formattedEvent)
-	require.NoError(t, err)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			v := viper.New()
+			v.SetDefault("transport.default.client-timeout", 1*time.Second)
+			p := transport.NewTransportPool(logrus.New(), v)
 
-	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"}]}]}`
+			client, err := NewClient("default", "v1/data", "", "GoStatsD", tt.name, "api-key", "", "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",
+				defaultMetricsPerBatch, defaultMaxRequests, 2*time.Second, 1*time.Second, gostatsd.TimerSubtypes{}, p)
+			require.NoError(t, err)
 
-	require.Equal(t, expected, string(fevent))
+			gostatsdEvent := gostatsd.Event{Title: "EventTitle", Text: "hi", Hostname: "blah", Priority: 1}
+			formattedEvent := client.EventFormatter(&gostatsdEvent)
+			fevent, err := json.Marshal(formattedEvent)
+			require.NoError(t, err)
+			require.Equal(t, tt.expected, string(fevent))
+		})
+	}
 }