fix memory leak in runtime.node.heap.* metrics (#5476)

* fix memory leak in runtime.node.heap metrics

* switch to promise-based timers for test

* add benchmark

Author: rochdev

benchmark/sirun/runtime-metrics/README.md

@@ -0,0 +1,5 @@
+This benchmark runs the with runtime metrics and an accelerated flush. While
+this can catch code regressions, it's mostly meant to catch things like memory
+leaks where metrics would start piling up. This can be hard to catch in tests,
+but it would cause the app to become slower and slower over time which would
+be visible in the benchmark.

benchmark/sirun/runtime-metrics/index.js

@@ -0,0 +1,5 @@
+'use strict'
+
+require('../../..').init()
+
+setTimeout(() => {}, 1000)

benchmark/sirun/runtime-metrics/meta.json

@@ -0,0 +1,22 @@
+{
+  "name": "runtime-metrics",
+  "run": "node index.js",
+  "run_with_affinity": "bash -c \"taskset -c $CPU_AFFINITY node index.js\"",
+  "cachegrind": false,
+  "iterations": 5,
+  "instructions": true,
+  "variants": {
+    "control": {
+      "env": {
+        "DD_RUNTIME_METRICS_ENABLED": "false"
+      }
+    },
+    "with-runtime-metrics": {
+      "baseline": "control",
+      "env": {
+        "DD_RUNTIME_METRICS_ENABLED": "true",
+        "DD_RUNTIME_METRICS_FLUSH_INTERVAL": "20"
+      }
+    }
+  }
+}

packages/dd-trace/src/runtime_metrics/runtime_metrics.js

@@ -10,7 +10,8 @@ const Histogram = require('../histogram')
 const { performance, PerformanceObserver } = require('perf_hooks')
 
 const { NODE_MAJOR, NODE_MINOR } = require('../../../../version')
-const INTERVAL = 10 * 1000
+const { DD_RUNTIME_METRICS_FLUSH_INTERVAL = '10000' } = process.env
+const INTERVAL = parseInt(DD_RUNTIME_METRICS_FLUSH_INTERVAL, 10)
 
 // Node >=16 has PerformanceObserver with `gc` type, but <16.7 had a critical bug.
 // See: https://github.com/nodejs/node/issues/39548
@@ -270,12 +271,12 @@ function captureNativeMetrics () {
   })
 
   for (let i = 0, l = spaces.length; i < l; i++) {
-    const tags = [`heap_space:${spaces[i].space_name}`]
+    const tag = `heap_space:${spaces[i].space_name}`
 
-    client.gauge('runtime.node.heap.size.by.space', spaces[i].space_size, tags)
-    client.gauge('runtime.node.heap.used_size.by.space', spaces[i].space_used_size, tags)
-    client.gauge('runtime.node.heap.available_size.by.space', spaces[i].space_available_size, tags)
-    client.gauge('runtime.node.heap.physical_size.by.space', spaces[i].physical_space_size, tags)
+    client.gauge('runtime.node.heap.size.by.space', spaces[i].space_size, tag)
+    client.gauge('runtime.node.heap.used_size.by.space', spaces[i].space_used_size, tag)
+    client.gauge('runtime.node.heap.available_size.by.space', spaces[i].space_available_size, tag)
+    client.gauge('runtime.node.heap.physical_size.by.space', spaces[i].physical_space_size, tag)
   }
 }
 

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

@@ -146,7 +146,7 @@ suiteDescribe('runtimeMetrics', () => {
       }
     }
 
-    setImmediate = globalThis.setImmediate
+    setImmediate = require('timers/promises').setImmediate
     clock = sinon.useFakeTimers()
 
     runtimeMetrics.start(config)
@@ -188,79 +188,97 @@ suiteDescribe('runtimeMetrics', () => {
       })
     })
 
-    it('should start collecting runtimeMetrics every 10 seconds', (done) => {
+    it('should start collecting runtimeMetrics every 10 seconds', async () => {
       runtimeMetrics.stop()
       runtimeMetrics.start(config)
 
       global.gc()
 
-      setImmediate(() => setImmediate(() => { // Wait for GC observer to trigger.
-        clock.tick(10000)
+      // Wait for GC observer to trigger.
+      await setImmediate()
+      await setImmediate()
+
+      clock.tick(10000)
+
+      expect(client.gauge).to.have.been.calledWith('runtime.node.cpu.user')
+      expect(client.gauge).to.have.been.calledWith('runtime.node.cpu.system')
+      expect(client.gauge).to.have.been.calledWith('runtime.node.cpu.total')
+
+      expect(client.gauge).to.have.been.calledWith('runtime.node.mem.rss')
+      expect(client.gauge).to.have.been.calledWith('runtime.node.mem.heap_total')
+      expect(client.gauge).to.have.been.calledWith('runtime.node.mem.heap_used')
+
+      expect(client.gauge).to.have.been.calledWith('runtime.node.process.uptime')
+
+      expect(client.gauge).to.have.been.calledWith('runtime.node.heap.total_heap_size')
+      expect(client.gauge).to.have.been.calledWith('runtime.node.heap.total_heap_size_executable')
+      expect(client.gauge).to.have.been.calledWith('runtime.node.heap.total_physical_size')
+      expect(client.gauge).to.have.been.calledWith('runtime.node.heap.total_available_size')
+      expect(client.gauge).to.have.been.calledWith('runtime.node.heap.total_heap_size')
+      expect(client.gauge).to.have.been.calledWith('runtime.node.heap.heap_size_limit')
+
+      expect(client.gauge).to.have.been.calledWith('runtime.node.heap.malloced_memory')
+      expect(client.gauge).to.have.been.calledWith('runtime.node.heap.peak_malloced_memory')
+
+      expect(client.gauge).to.have.been.calledWith('runtime.node.event_loop.delay.max', sinon.match.number)
+      expect(client.gauge).to.have.been.calledWith('runtime.node.event_loop.delay.min', sinon.match.number)
+      expect(client.increment).to.have.been.calledWith('runtime.node.event_loop.delay.sum', sinon.match.number)
+      expect(client.gauge).to.have.been.calledWith('runtime.node.event_loop.delay.avg', sinon.match.number)
+      expect(client.gauge).to.have.been.calledWith('runtime.node.event_loop.delay.median', sinon.match.number)
+      expect(client.gauge).to.have.been.calledWith('runtime.node.event_loop.delay.95percentile', sinon.match.number)
+      expect(client.increment).to.have.been.calledWith('runtime.node.event_loop.delay.count', sinon.match.number)
+
+      expect(client.gauge).to.have.been.calledWith('runtime.node.event_loop.utilization')
+
+      expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.max', sinon.match.number)
+      expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.min', sinon.match.number)
+      expect(client.increment).to.have.been.calledWith('runtime.node.gc.pause.sum', sinon.match.number)
+      expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.avg', sinon.match.number)
+      expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.median', sinon.match.number)
+      expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.95percentile', sinon.match.number)
+      expect(client.increment).to.have.been.calledWith('runtime.node.gc.pause.count', sinon.match.number)
+
+      expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.by.type.max', sinon.match.number)
+      expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.by.type.min', sinon.match.number)
+      expect(client.increment).to.have.been.calledWith('runtime.node.gc.pause.by.type.sum', sinon.match.number)
+      expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.by.type.avg', sinon.match.number)
+      expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.by.type.median', sinon.match.number)
+      expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.by.type.95percentile', sinon.match.number)
+      expect(client.increment).to.have.been.calledWith('runtime.node.gc.pause.by.type.count', sinon.match.number)
+      expect(client.increment).to.have.been.calledWith(
+        'runtime.node.gc.pause.by.type.count', sinon.match.any, sinon.match(val => {
+          return val && /^gc_type:[a-z_]+$/.test(val[0])
+        })
+      )
+
+      expect(client.gauge).to.have.been.calledWith('runtime.node.heap.size.by.space')
+      expect(client.gauge).to.have.been.calledWith('runtime.node.heap.used_size.by.space')
+      expect(client.gauge).to.have.been.calledWith('runtime.node.heap.available_size.by.space')
+      expect(client.gauge).to.have.been.calledWith('runtime.node.heap.physical_size.by.space')
+
+      expect(client.flush).to.have.been.called
+    })
+
+    it('should collect individual metrics only once every 10 seconds', async () => {
+      runtimeMetrics.stop()
+      runtimeMetrics.start(config)
+
+      global.gc()
+
+      // Wait for GC observer to trigger.
+      await setImmediate()
+      await setImmediate()
+
+      clock.tick(60 * 60 * 1000)
 
-        try {
-          expect(client.gauge).to.have.been.calledWith('runtime.node.cpu.user')
-          expect(client.gauge).to.have.been.calledWith('runtime.node.cpu.system')
-          expect(client.gauge).to.have.been.calledWith('runtime.node.cpu.total')
-
-          expect(client.gauge).to.have.been.calledWith('runtime.node.mem.rss')
-          expect(client.gauge).to.have.been.calledWith('runtime.node.mem.heap_total')
-          expect(client.gauge).to.have.been.calledWith('runtime.node.mem.heap_used')
-
-          expect(client.gauge).to.have.been.calledWith('runtime.node.process.uptime')
-
-          expect(client.gauge).to.have.been.calledWith('runtime.node.heap.total_heap_size')
-          expect(client.gauge).to.have.been.calledWith('runtime.node.heap.total_heap_size_executable')
-          expect(client.gauge).to.have.been.calledWith('runtime.node.heap.total_physical_size')
-          expect(client.gauge).to.have.been.calledWith('runtime.node.heap.total_available_size')
-          expect(client.gauge).to.have.been.calledWith('runtime.node.heap.total_heap_size')
-          expect(client.gauge).to.have.been.calledWith('runtime.node.heap.heap_size_limit')
-
-          expect(client.gauge).to.have.been.calledWith('runtime.node.heap.malloced_memory')
-          expect(client.gauge).to.have.been.calledWith('runtime.node.heap.peak_malloced_memory')
-
-          expect(client.gauge).to.have.been.calledWith('runtime.node.event_loop.delay.max', sinon.match.number)
-          expect(client.gauge).to.have.been.calledWith('runtime.node.event_loop.delay.min', sinon.match.number)
-          expect(client.increment).to.have.been.calledWith('runtime.node.event_loop.delay.sum', sinon.match.number)
-          expect(client.gauge).to.have.been.calledWith('runtime.node.event_loop.delay.avg', sinon.match.number)
-          expect(client.gauge).to.have.been.calledWith('runtime.node.event_loop.delay.median', sinon.match.number)
-          expect(client.gauge).to.have.been.calledWith('runtime.node.event_loop.delay.95percentile', sinon.match.number)
-          expect(client.increment).to.have.been.calledWith('runtime.node.event_loop.delay.count', sinon.match.number)
-
-          expect(client.gauge).to.have.been.calledWith('runtime.node.event_loop.utilization')
-
-          expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.max', sinon.match.number)
-          expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.min', sinon.match.number)
-          expect(client.increment).to.have.been.calledWith('runtime.node.gc.pause.sum', sinon.match.number)
-          expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.avg', sinon.match.number)
-          expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.median', sinon.match.number)
-          expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.95percentile', sinon.match.number)
-          expect(client.increment).to.have.been.calledWith('runtime.node.gc.pause.count', sinon.match.number)
-
-          expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.by.type.max', sinon.match.number)
-          expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.by.type.min', sinon.match.number)
-          expect(client.increment).to.have.been.calledWith('runtime.node.gc.pause.by.type.sum', sinon.match.number)
-          expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.by.type.avg', sinon.match.number)
-          expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.by.type.median', sinon.match.number)
-          expect(client.gauge).to.have.been.calledWith('runtime.node.gc.pause.by.type.95percentile', sinon.match.number)
-          expect(client.increment).to.have.been.calledWith('runtime.node.gc.pause.by.type.count', sinon.match.number)
-          expect(client.increment).to.have.been.calledWith(
-            'runtime.node.gc.pause.by.type.count', sinon.match.any, sinon.match(val => {
-              return val && /^gc_type:[a-z_]+$/.test(val[0])
-            })
-          )
-
-          expect(client.gauge).to.have.been.calledWith('runtime.node.heap.size.by.space')
-          expect(client.gauge).to.have.been.calledWith('runtime.node.heap.used_size.by.space')
-          expect(client.gauge).to.have.been.calledWith('runtime.node.heap.available_size.by.space')
-          expect(client.gauge).to.have.been.calledWith('runtime.node.heap.physical_size.by.space')
-
-          expect(client.flush).to.have.been.called
-
-          done()
-        } catch (e) {
-          done(e)
-        }
-      }))
+      // If a metric is leaking, it will leak expontentially because it will
+      // be sent one more time each flush, in addition to the previous
+      // flushes that also had the metric multiple times in them, so after
+      // 1 hour even if a single metric is leaking it would get over
+      // 64980 calls on its own without any other metric. A slightly lower
+      // value is used here to be on the safer side.
+      expect(client.gauge.callCount).to.be.lt(60000)
+      expect(client.increment.callCount).to.be.lt(60000)
     })
   })