fix(dogstatsd): stop re-emitting stale gauges (#8153)

This fixes Datadog "no data" alerting being silently masked for
custom metrics. The aggregator's flush loop never cleared the gauge
tree, so it kept rebroadcasting the last value of every gauge every
10 seconds for the lifetime of the process. Histograms had the same
masking effect once observed at least once: the count===0 fallback
emitted a row of zeros forever.

Runtime metrics re-set every gauge each interval, so dashboards using
them stay identical.

In addition, if being used with a high cardinality input, this would
be a memory leak.

Fixes: #7504

Author: BridgeAR

packages/dd-trace/src/dogstatsd.js

@@ -266,6 +266,8 @@ class MetricsAggregationClient {
     this._captureTree(this._gauges, (node, name, tags) => {
       this._client.gauge(name, node.value, tags)
     })
+
+    this._gauges.clear()
   }
 
   _captureCounters () {
@@ -278,12 +280,7 @@ class MetricsAggregationClient {
 
   _captureHistograms () {
     this._captureTree(this._histograms, (node, name, tags) => {
-      let stats = node.value
-
-      // Stats can contain garbage data when a value was never recorded.
-      if (stats.count === 0) {
-        stats = { max: 0, min: 0, sum: 0, avg: 0, median: 0, p95: 0, count: 0 }
-      }
+      const stats = node.value
 
       this._client.gauge(`${name}.min`, stats.min, tags)
       this._client.gauge(`${name}.max`, stats.max, tags)
@@ -293,9 +290,9 @@ class MetricsAggregationClient {
       this._client.increment(`${name}.count`, stats.count, tags)
       this._client.gauge(`${name}.median`, stats.median, tags)
       this._client.gauge(`${name}.95percentile`, stats.p95, tags)
-
-      node.value.reset()
     })
+
+    this._histograms.clear()
   }
 
   _captureTree (tree, fn) {

packages/dd-trace/test/dogstatsd.spec.js

@@ -15,6 +15,7 @@ describe('dogstatsd', () => {
   let client
   let DogStatsDClient
   let CustomMetrics
+  let MetricsAggregationClient
   let dgram
   let udp4
   let udp6
@@ -76,6 +77,7 @@ describe('dogstatsd', () => {
     })
     DogStatsDClient = dogstatsd.DogStatsDClient
     CustomMetrics = dogstatsd.CustomMetrics
+    MetricsAggregationClient = dogstatsd.MetricsAggregationClient
 
     httpData = []
     statusCode = 200
@@ -659,4 +661,75 @@ describe('dogstatsd', () => {
       assert.strictEqual(udp4.send.firstCall.args[0].toString(), 'test.avg:10|g|#foo:bar|c:ci-1234\n')
     })
   })
+
+  describe('MetricsAggregationClient', () => {
+    let aggregator
+    let gaugeCalls
+    let incrementCalls
+
+    beforeEach(() => {
+      gaugeCalls = []
+      incrementCalls = []
+      const inner = {
+        gauge: (name, value, tags) => gaugeCalls.push([name, value, tags?.slice()]),
+        increment: (name, value, tags) => incrementCalls.push([name, value, tags?.slice()]),
+        distribution: () => {},
+        histogram: () => {},
+        flush: () => {},
+      }
+      aggregator = new MetricsAggregationClient(inner)
+    })
+
+    it('emits a gauge once and then stays silent until it is set again', () => {
+      aggregator.gauge('test.avg', 5)
+      aggregator.flush()
+
+      assert.deepStrictEqual(gaugeCalls, [['test.avg', 5, []]])
+
+      gaugeCalls.length = 0
+      aggregator.flush()
+      aggregator.flush()
+
+      assert.deepStrictEqual(gaugeCalls, [])
+    })
+
+    it('re-emits a gauge on every flush when it is updated each cycle', () => {
+      for (let i = 1; i <= 3; i++) {
+        aggregator.gauge('test.avg', i)
+        aggregator.flush()
+      }
+
+      assert.deepStrictEqual(gaugeCalls, [
+        ['test.avg', 1, []],
+        ['test.avg', 2, []],
+        ['test.avg', 3, []],
+      ])
+    })
+
+    it('does not re-emit a histogram once observations stop', () => {
+      aggregator.histogram('test.hist', 10)
+      aggregator.flush()
+
+      assert(gaugeCalls.length > 0 && incrementCalls.length > 0)
+
+      gaugeCalls.length = 0
+      incrementCalls.length = 0
+      aggregator.flush()
+      aggregator.flush()
+
+      assert.deepStrictEqual(gaugeCalls, [])
+      assert.deepStrictEqual(incrementCalls, [])
+    })
+
+    it('drains all metric trees on flush so cardinality is bounded', () => {
+      aggregator.gauge('test.avg', 5, ['t:1'])
+      aggregator.histogram('test.hist', 10, ['t:1'])
+      aggregator.increment('test.count', 1, ['t:1'])
+      aggregator.flush()
+
+      assert.strictEqual(aggregator._gauges.size, 0)
+      assert.strictEqual(aggregator._histograms.size, 0)
+      assert.strictEqual(aggregator._counters.size, 0)
+    })
+  })
 })