feat: Added ProfilingAggregator that will be used to collect and send pprof_data telemetry (newrelic#3732)

Author: bizob2828

lib/aggregators/profiling-aggregator.js

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2026 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+
+const logger = require('../logger').child({ component: 'pprof_aggregator' })
+const BaseAggregator = require('./base-aggregator')
+
+/**
+ * Serves as a means for transmitting pprof data as it is being collected.
+ * This doesn't really work as an "aggregator".  It instead leverages the `BaseAggregator`
+ * to integrate with the `Harvester` and `Collector`
+ *
+ * The `ProfilingAggregator` aggregator overrides the base `start` method.
+ * It sets up the interval based on configuration but instead of calling `send`,
+ * it calls `collectData` which instructs the Profiler to collect profiling data
+ * for every register profiler(cpu, heap at the moment if enabled)
+ *
+ * @private
+ * @class
+ */
+class ProfilingAggregator extends BaseAggregator {
+  constructor(opts = {}, agent) {
+    const { collector, harvester } = agent
+    opts.method = opts.method || 'pprof_data'
+    super(opts, collector, harvester)
+    this.agent = agent
+    this.profiler = opts.profiler
+    this.pprofData = null
+  }
+
+  // simply returns what was given from the profiler collect method
+  _toPayloadSync() {
+    return this.pprofData
+  }
+
+  /**
+   * This overrides the default `start` method
+   * as we want to collect profiling data for `cpu` and/or `heap`
+   * and send the gzipped binary encoded data for each profiler
+   */
+  start() {
+    logger.trace(`${this.method} aggregator started.`)
+
+    if (!this.sendTimer) {
+      this.sendTimer = setInterval(this.collectData.bind(this), this.periodMs)
+      this.sendTimer.unref()
+    }
+  }
+
+  /**
+   * Called on an interval. Iterates over all registered profilers
+   * and collects data for the given time period. Then asynchronously
+   * calls send which takes care of sending the data to the collector
+   */
+  collectData() {
+    for (const profiler of this.profiler.profilers) {
+      this.pprofData = profiler.collect()
+      this.send()
+    }
+  }
+
+  // `pprof_data` is not retained/merged in any way to just return null
+  _getMergeData() {
+    return null
+  }
+
+  // this implies the transmission fails
+  // we log this error in `lib/collector/api.js#_handleResponseCode`
+  _merge() {
+    return null
+  }
+
+  // clears the `this.pprofData` for the next iteration of profiling collection
+  clear() {
+    this.pprofData = null
+  }
+}
+
+module.exports = ProfilingAggregator

lib/collector/api.js

@@ -103,7 +103,8 @@ function CollectorAPI(agent) {
     'sql_trace_data',
     'error_event_data',
     'span_event_data',
-    'log_event_data'
+    'log_event_data',
+    'pprof_data'
   ]) {
     const method = new RemoteMethod(name, agent, initialEndpoint)
     this._methods[name] = method

test/unit/aggregators/profiling-aggregator.test.js

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2024 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+
+const test = require('node:test')
+const assert = require('node:assert')
+const sinon = require('sinon')
+const ProfilingAggregator = require('#agentlib/aggregators/profiling-aggregator.js')
+const helper = require('#testlib/agent_helper.js')
+const RUN_ID = 1337
+
+test.beforeEach((ctx) => {
+  const agent = helper.loadMockedAgent()
+  const cpuProfiler = {
+    collect() {
+      return 'cpu profile data'
+    }
+  }
+
+  const clock = sinon.useFakeTimers()
+  const heapProfiler = {
+    collect() {
+      return 'heap profile data'
+    }
+  }
+  const profiler = {
+    profilers: [cpuProfiler, heapProfiler]
+  }
+  sinon.spy(agent.collector, 'send')
+  const profilingAggregator = new ProfilingAggregator({ runId: RUN_ID, periodMs: 100, profiler }, agent)
+  ctx.nr = {
+    agent,
+    clock,
+    profilingAggregator,
+    profiler
+  }
+})
+
+test.afterEach((ctx) => {
+  helper.unloadAgent(ctx.nr.agent)
+  ctx.nr.clock.restore()
+  ctx.nr.agent.collector.send.restore()
+})
+
+test('should set the correct default method', (t) => {
+  const { profilingAggregator } = t.nr
+  const method = profilingAggregator.method
+  assert.equal(method, 'pprof_data')
+})
+
+test('should intialize pprofData and profiler', (t) => {
+  const { profilingAggregator, profiler } = t.nr
+  assert.deepEqual(profilingAggregator.profiler, profiler)
+  assert.equal(profilingAggregator.pprofData, null)
+})
+
+test('should send 2 messages per interval', (t) => {
+  const { profilingAggregator, clock, agent } = t.nr
+  profilingAggregator.start()
+  assert.equal(agent.collector.send.callCount, 0)
+  clock.tick(100)
+  assert.equal(agent.collector.send.callCount, 2)
+  const [cpuCall, heapCall] = agent.collector.send.args
+  assert.equal(cpuCall[0], 'pprof_data')
+  assert.equal(cpuCall[1], 'cpu profile data')
+  assert.equal(heapCall[0], 'pprof_data')
+  assert.equal(heapCall[1], 'heap profile data')
+  assert.equal(profilingAggregator.pprofData, null)
+})