Repository: atlassian/gostatsd
Pull request #334 (2020-08-07 01:06:04)

Introduces sparse internal metrics
These are gauges which are only sent when the value changes. They're sent for several flush intervals, with the expectation that they are attached to error counters, so may need to survive a bursty failures.

Overall this will reduce the amount of data sent in the happy path when deployed in a forwarder configuration (2 metrics per host).


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 30f8df6..c3a3705 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,7 @@
+28.2.0
+------
+- Changes how `parser.bad_lines_seen` and `backend.retried` metrics are sent.  They are now only sent on change.
+
 28.1.0
 ------
 - The Datadog backend will discard values that result in NaN or ±Inf since they can not be used with our JSON processing.
diff --git a/METRICS.md b/METRICS.md
index 6f271ae..1e40b8a 100644
--- a/METRICS.md
+++ b/METRICS.md
@@ -9,6 +9,7 @@ Metric types:
 | gauge (flush)      | A value sent as a gauge with the value reset / calculated / sampled every flush interval
 | gauge (time)       | A single duration measured in milliseconds and sent as a gauge
 | gauge (cumulative) | An internal counter sent as a gauge with the value never resetting
+| gauge (sparse)     | The same as a cumulative gauge, but data is only sent on change
 | counter            | An internal counter, reset on flush
 
 
@@ -21,7 +22,7 @@ Metrics:
 |                                             |                     |                              | datapoints in this flush interval
 | aggregator.process_time                     | gauge (time)        | aggregator_id                | The time taken to process all synchronous flush actions
 | aggregator.reset_time                       | gauge (time)        | aggregator_id                | The time taken to reset the aggregator after flush
-| parser.bad_lines_seen                       | gauge (cumulative)  |                              | The number of unparseable lines
+| parser.bad_lines_seen                       | gauge (sparse)      |                              | The number of unparseable lines
 | parser.events_received                      | gauge (cumulative)  |                              | The number of events parsed
 | parser.metrics_received                     | gauge (cumulative)  |                              | The number of metrics parsed
 | receiver.datagrams_received                 | gauge (cumulative)  |                              | The number of datagrams received
@@ -33,12 +34,11 @@ Metrics:
 | channel.last                                | gauge (flush)       | channel                      | The last sample seen
 | channel.capacity                            | gauge (flush)       | channel                      | The capacity of the channel
 | channel.samples                             | gauge (flush)       | channel                      | The number of samples seen (guaranteed to be at least 1)
-| internal_dropped                            | gauge (cumulative)  |                              | The number of internal metrics which have been dropped
 | heartbeat                                   | gauge (flush)       | version, commit              | The value 1, tagged by the version (git tag) and short commit hash
 | flusher.total_time                          | gauge (time)        |                              | Time taken to flush all metrics to all backends for the flush interval
 | backend.created                             | gauge (cumulative)  | backend                      | Lifetime number of metric batches generated by the backend
 | backend.create.failed                       | gauge (cumulative)  | backend                      | Lifetime number of metric batches which failed to be serialized (DATALOSS!)
-| backend.retried                             | gauge (cumulative)  | backend                      | Lifetime number of metric batches retried by the backend
+| backend.retried                             | gauge (sparse)      | backend                      | Lifetime number of metric batches retried by the backend
 | backend.dropped                             | gauge (cumulative)  | backend                      | Lifetime number of metric batches dropped by the backend (DATALOSS!)
 | backend.sent                                | gauge (cumulative)  | backend                      | Lifetime number of metric batches successfully transmitted
 | backend.series.sent                         | gauge (cumulative)  | backend                      | Lifetime number of metric series successfully transmitted
@@ -52,7 +52,6 @@ Metrics:
 | cloudprovider.cache_refresh_positive        | gauge (cumulative)  |                              | The cumulative number of positive refreshes
 | cloudprovider.cache_refresh_negative        | gauge (cumulative)  |                              | The cumulative number of refreshes which had an error refreshing and used old data
 | cloudprovider.cache_hit                     | gauge (cumulative)  |                              | The cumulative number of cache hits (host was in the cache)
-|                                             |                     |                              | in progress which completed)
 | cloudprovider.cache_miss                    | gauge (cumulative)  |                              | The cumulative number of cache misses
 | cloudprovider.hosts_queued                  | gauge (flush)       | type                         | The absolute number of hosts waiting to be looked up
 | cloudprovider.items_queued                  | gauge (flush)       | type                         | The absolute number of metrics or events waiting for a host lookup to complete
diff --git a/pkg/backends/datadog/datadog.go b/pkg/backends/datadog/datadog.go
index f1256f4..acc48ad 100644
--- a/pkg/backends/datadog/datadog.go
+++ b/pkg/backends/datadog/datadog.go
@@ -56,11 +56,11 @@ var (
 
 // Client represents a Datadog client.
 type Client struct {
-	batchesCreated uint64 // Accumulated number of batches created
-	batchesRetried uint64 // Accumulated number of batches retried (first send is not a retry)
-	batchesDropped uint64 // Accumulated number of batches aborted (data loss)
-	batchesSent    uint64 // Accumulated number of batches successfully sent
-	seriesSent     uint64 // Accumulated number of series successfully sent
+	batchesCreated uint64            // Accumulated number of batches created
+	batchesDropped uint64            // Accumulated number of batches aborted (data loss)
+	batchesSent    uint64            // Accumulated number of batches successfully sent
+	seriesSent     uint64            // Accumulated number of series successfully sent
+	batchesRetried stats.ChangeGauge // Accumulated number of batches retried (first send is not a retry)
 
 	logger                logrus.FieldLogger
 	apiKey                string
@@ -148,7 +148,7 @@ func (d *Client) Run(ctx context.Context) {
 			return
 		case <-flushed:
 			statser.Gauge("backend.created", float64(atomic.LoadUint64(&d.batchesCreated)), nil)
-			statser.Gauge("backend.retried", float64(atomic.LoadUint64(&d.batchesRetried)), nil)
+			d.batchesRetried.SendIfChanged(statser, "backend.retried", nil)
 			statser.Gauge("backend.dropped", float64(atomic.LoadUint64(&d.batchesDropped)), nil)
 			statser.Gauge("backend.sent", float64(atomic.LoadUint64(&d.batchesSent)), nil)
 			statser.Gauge("backend.series.sent", float64(atomic.LoadUint64(&d.seriesSent)), nil)
@@ -307,7 +307,7 @@ func (d *Client) post(ctx context.Context, buffer *bytes.Buffer, path, typeOfPos
 		case <-timer.C:
 		}
 
-		atomic.AddUint64(&d.batchesRetried, 1)
+		atomic.AddUint64(&d.batchesRetried.Cur, 1)
 	}
 }
 
diff --git a/pkg/backends/influxdb/influxdb.go b/pkg/backends/influxdb/influxdb.go
index 43e112e..0f37a11 100644
--- a/pkg/backends/influxdb/influxdb.go
+++ b/pkg/backends/influxdb/influxdb.go
@@ -63,12 +63,12 @@ var (
 
 // Client represents an InfluxDB client.
 type Client struct {
-	batchesCreated      uint64 // Accumulated number of batches created
-	batchesRetried      uint64 // Accumulated number of batches retried (first send is not a retry)
-	batchesDropped      uint64 // Accumulated number of batches aborted (data loss)
-	batchesCreateFailed uint64 // Accumulated number of batches which failed to serialize (data loss, no retry is possible)
-	batchesSent         uint64 // Accumulated number of batches successfully sent
-	seriesSent          uint64 // Accumulated number of series successfully sent
+	batchesCreated      uint64            // Accumulated number of batches created
+	batchesDropped      uint64            // Accumulated number of batches aborted (data loss)
+	batchesCreateFailed uint64            // Accumulated number of batches which failed to serialize (data loss, no retry is possible)
+	batchesSent         uint64            // Accumulated number of batches successfully sent
+	seriesSent          uint64            // Accumulated number of series successfully sent
+	batchesRetried      stats.ChangeGauge // Accumulated number of batches retried (first send is not a retry)
 
 	logger logrus.FieldLogger
 
@@ -267,7 +267,7 @@ func (idb *Client) Run(ctx context.Context) {
 		case <-flushed:
 			statser.Gauge("backend.created", float64(atomic.LoadUint64(&idb.batchesCreated)), nil)
 			statser.Gauge("backend.create.failed", float64(atomic.LoadUint64(&idb.batchesCreateFailed)), nil)
-			statser.Gauge("backend.retried", float64(atomic.LoadUint64(&idb.batchesRetried)), nil)
+			idb.batchesRetried.SendIfChanged(statser, "backend.retried", nil)
 			statser.Gauge("backend.dropped", float64(atomic.LoadUint64(&idb.batchesDropped)), nil)
 			statser.Gauge("backend.sent", float64(atomic.LoadUint64(&idb.batchesSent)), nil)
 			statser.Gauge("backend.series.sent", float64(atomic.LoadUint64(&idb.seriesSent)), nil)
@@ -455,7 +455,7 @@ func (idb *Client) post(ctx context.Context, buffer *bytes.Buffer) error {
 		case <-timer.C:
 		}
 
-		atomic.AddUint64(&idb.batchesRetried, 1)
+		atomic.AddUint64(&idb.batchesRetried.Cur, 1)
 	}
 }
 
diff --git a/pkg/backends/newrelic/newrelic.go b/pkg/backends/newrelic/newrelic.go
index fb17f9b..c88bc00 100644
--- a/pkg/backends/newrelic/newrelic.go
+++ b/pkg/backends/newrelic/newrelic.go
@@ -77,11 +77,11 @@ type Client struct {
 	timerSum        string
 	timerSumSquares string
 
-	batchesCreated uint64 // Accumulated number of batches created
-	batchesRetried uint64 // Accumulated number of batches retried (first send is not a retry)
-	batchesDropped uint64 // Accumulated number of batches aborted (data loss)
-	batchesSent    uint64 // Accumulated number of batches successfully sent
-	seriesSent     uint64 // Accumulated number of series successfully sent
+	batchesCreated uint64            // Accumulated number of batches created
+	batchesDropped uint64            // Accumulated number of batches aborted (data loss)
+	batchesSent    uint64            // Accumulated number of batches successfully sent
+	seriesSent     uint64            // Accumulated number of series successfully sent
+	batchesRetried stats.ChangeGauge // Accumulated number of batches retried (first send is not a retry)
 
 	userAgent             string
 	maxRequestElapsedTime time.Duration
@@ -187,7 +187,7 @@ func (n *Client) RunMetrics(ctx context.Context, statser stats.Statser) {
 			return
 		case <-flushed:
 			statser.Gauge("backend.created", float64(atomic.LoadUint64(&n.batchesCreated)), nil)
-			statser.Gauge("backend.retried", float64(atomic.LoadUint64(&n.batchesRetried)), nil)
+			n.batchesRetried.SendIfChanged(statser, "backend.retried", nil)
 			statser.Gauge("backend.dropped", float64(atomic.LoadUint64(&n.batchesDropped)), nil)
 			statser.Gauge("backend.sent", float64(atomic.LoadUint64(&n.batchesSent)), nil)
 			statser.Gauge("backend.series.sent", float64(atomic.LoadUint64(&n.seriesSent)), nil)
@@ -329,7 +329,7 @@ func (n *Client) post(ctx context.Context, buffer *bytes.Buffer, data interface{
 		case <-timer.C:
 		}
 
-		atomic.AddUint64(&n.batchesRetried, 1)
+		atomic.AddUint64(&n.batchesRetried.Cur, 1)
 	}
 }
 
diff --git a/pkg/stats/change_gauge.go b/pkg/stats/change_gauge.go
new file mode 100644
index 0000000..06ff968
--- /dev/null
+++ b/pkg/stats/change_gauge.go
@@ -0,0 +1,39 @@
+package stats
+
+import (
+	"sync/atomic"
+
+	"github.com/atlassian/gostatsd"
+)
+
+// repeatCount is how many times to send the gauge once it changes.  22 is used because it will cover at
+// least 2 flush intervals on an aggregation server with a 10x longer interval than a forwarding server.
+const repeatCount = 22
+
+// ChangeGauge is a simpler wrapper to send a gauge for a rare event, multiple times.  It is primarily intended for
+// things that won't change in the majority of cases (such as failure to parse a statsd line), and things that will
+// be bursty (ie, transient http message failure).  It isn't suitable for things which are constantly changing such
+// as number of metrics or http requests.
+//
+type ChangeGauge struct {
+	// Cur is the last value expected to be sent.  If this is changed, SendIfChanged will send the value for
+	// 22 flush intervals in an attempt to retry any failures to send it.  Cur is assumed to be atomic by
+	// ChangeGauge.SendIfChanged, however if the owner has more specific knowledge, it can be written to
+	// with non-atomic operations.
+	Cur uint64 // atomic
+
+	prev    uint64
+	pending uint64 // number of times to re-send
+}
+
+func (cg *ChangeGauge) SendIfChanged(statser Statser, metricName string, tags gostatsd.Tags) {
+	v := atomic.LoadUint64(&cg.Cur)
+	if v != cg.prev {
+		cg.prev = v
+		cg.pending = repeatCount
+	}
+	if cg.pending > 0 {
+		cg.pending--
+		statser.Gauge(metricName, float64(v), tags)
+	}
+}
diff --git a/pkg/statsd/parser.go b/pkg/statsd/parser.go
index 177d39c..4b2ddf5 100644
--- a/pkg/statsd/parser.go
+++ b/pkg/statsd/parser.go
@@ -25,7 +25,7 @@ type DatagramParser struct {
 	// Counter fields below must be read/written only using atomic instructions.
 	// 64-bit fields must be the first fields in the struct to guarantee proper memory alignment.
 	// See https://golang.org/pkg/sync/atomic/#pkg-note-BUG
-	badLines        uint64
+	badLines        stats.ChangeGauge
 	metricsReceived uint64
 	eventsReceived  uint64
 
@@ -86,7 +86,7 @@ func (dp *DatagramParser) RunMetricsContext(ctx context.Context) {
 		case <-flushed:
 			statser.Gauge("parser.metrics_received", float64(atomic.LoadUint64(&dp.metricsReceived)), nil)
 			statser.Gauge("parser.events_received", float64(atomic.LoadUint64(&dp.eventsReceived)), nil)
-			statser.Gauge("parser.bad_lines_seen", float64(atomic.LoadUint64(&dp.badLines)), nil)
+			dp.badLines.SendIfChanged(statser, "parser.bad_lines_seen", nil)
 		}
 	}
 }
@@ -121,7 +121,7 @@ func (dp *DatagramParser) Run(ctx context.Context) {
 			}
 			atomic.AddUint64(&dp.metricsReceived, uint64(len(metrics)))
 			atomic.AddUint64(&dp.eventsReceived, accumE)
-			atomic.AddUint64(&dp.badLines, accumB)
+			atomic.AddUint64(&dp.badLines.Cur, accumB)
 		}
 	}
 }


Test patch:
diff --git a/pkg/stats/change_gauge_test.go b/pkg/stats/change_gauge_test.go
new file mode 100644
index 0000000..fb5db1d
--- /dev/null
+++ b/pkg/stats/change_gauge_test.go
@@ -0,0 +1,22 @@
+package stats
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestChangeGaugeSendIfChanged(t *testing.T) {
+	cg := &ChangeGauge{}
+	cs := &countingStatser{}
+
+	cg.SendIfChanged(cs, "change", nil)
+	require.Zero(t, cs.gauges)
+	cg.Cur = 5
+	for i := 1; i <= repeatCount; i++ {
+		cg.SendIfChanged(cs, "change", nil)
+		require.EqualValues(t, i, cs.gauges)
+	}
+	cg.SendIfChanged(cs, "change", nil) // should not send
+	require.EqualValues(t, repeatCount, cs.gauges)
+}
diff --git a/pkg/stats/fixtures_test.go b/pkg/stats/fixtures_test.go
new file mode 100644
index 0000000..c2c93a5
--- /dev/null
+++ b/pkg/stats/fixtures_test.go
@@ -0,0 +1,49 @@
+package stats
+
+import (
+	"context"
+	"sync/atomic"
+	"time"
+
+	"github.com/atlassian/gostatsd"
+)
+
+type countingStatser struct {
+	gauges   uint64
+	counters uint64
+	timers   uint64
+}
+
+func (cs *countingStatser) NotifyFlush(ctx context.Context, d time.Duration) {}
+
+func (cs *countingStatser) RegisterFlush() (ch <-chan time.Duration, unregister func()) {
+	return nil, func() {}
+}
+
+func (cs *countingStatser) Gauge(name string, value float64, tags gostatsd.Tags) {
+	atomic.AddUint64(&cs.gauges, 1)
+}
+
+func (cs *countingStatser) Count(name string, amount float64, tags gostatsd.Tags) {
+	atomic.AddUint64(&cs.counters, 1)
+}
+
+func (cs *countingStatser) Increment(name string, tags gostatsd.Tags) {
+	atomic.AddUint64(&cs.counters, 1)
+}
+
+func (cs *countingStatser) TimingMS(name string, ms float64, tags gostatsd.Tags) {
+	atomic.AddUint64(&cs.timers, 1)
+}
+
+func (cs *countingStatser) TimingDuration(name string, d time.Duration, tags gostatsd.Tags) {
+	atomic.AddUint64(&cs.timers, 1)
+}
+
+func (cs *countingStatser) NewTimer(name string, tags gostatsd.Tags) *Timer {
+	return newTimer(cs, name, tags)
+}
+
+func (cs *countingStatser) WithTags(tags gostatsd.Tags) Statser {
+	return cs
+}