[WIP] Memory investigation

example-apps/collector/package.json

@@ -8,7 +8,8 @@
   "dependencies": {
     "@instana/collector": "latest",
     "dotenv": "^5.0.1",
-    "express": "^4.16.3",
+    "express": "^4.21.2",
+    "express-status-monitor": "^1.3.4",
     "node-fetch": "^2.6.1"
   }
 }

example-apps/collector/src/index.js

@@ -17,6 +17,7 @@ const path = require('path');
 let packageToRequire = path.join(__dirname, '..', '..', '..', 'packages/collector');
 if (config.mode === 'npm') {
   packageToRequire = '@instana/collector';
+  console.log('Using NPM');
 }
 
 if (config.collectorEnabled) {

packages/collector/src/agentConnection.js

@@ -119,16 +119,18 @@ exports.announceNodeCollector = function announceNodeCollector(callback) {
 
   let payloadStr = JSON.stringify(payload);
   const contentLength = Buffer.from(payloadStr, 'utf8').length + paddingForInodeAndFileDescriptor;
+  let req;
 
   let wasCalled = false;
   const handleCallback = function () {
     if (!wasCalled) {
       wasCalled = true;
+      req = null; // free memory
       cb.apply(null, arguments);
     }
   };
 
-  const req = http.request(
+  req = http.request(
     {
       host: agentOpts.host,
       port: agentOpts.port,
@@ -148,11 +150,13 @@ exports.announceNodeCollector = function announceNodeCollector(callback) {
       }
 
       res.setEncoding('utf8');
-      let responseBody = '';
+      const chunks = [];
       res.on('data', chunk => {
-        responseBody += chunk;
+        chunks.push(chunk);
       });
       res.on('end', () => {
+        const responseBody = chunks.join('');
+        chunks.length = 0; // free memory
         handleCallback(null, responseBody);
       });
     }
@@ -201,7 +205,10 @@ exports.announceNodeCollector = function announceNodeCollector(callback) {
       }
     }
 
-    req.write(Buffer.from(JSON.stringify(payload), 'utf8'));
+    const finalBuffer = Buffer.from(JSON.stringify(payload), 'utf8');
+    payloadStr = null;
+
+    req.write(finalBuffer, 'utf8');
     req.end();
   });
 };
@@ -387,7 +394,7 @@ exports.sendTracingMetricsToAgent = function sendTracingMetricsToAgent(tracingMe
 function sendData(path, data, cb, ignore404 = false) {
   cb = util.atMostOnce(`callback for sendData: ${path}`, cb);
 
-  const payloadAsString = JSON.stringify(data, circularReferenceRemover());
+  let payloadAsString = JSON.stringify(data, circularReferenceRemover());
   if (typeof logger.trace === 'function') {
     logger.trace(`Sending data to ${path}.`);
   } else {
@@ -396,6 +403,8 @@ function sendData(path, data, cb, ignore404 = false) {
 
   // Convert payload to a buffer to correctly identify content-length ahead of time.
   const payload = Buffer.from(payloadAsString, 'utf8');
+  payloadAsString = null;
+
   if (payload.length > maxContentLength) {
     const error = new PayloadTooLargeError(`Request payload is too large. Will not send data to agent. (POST ${path})`);
     return setImmediate(cb.bind(null, error));
@@ -427,11 +436,13 @@ function sendData(path, data, cb, ignore404 = false) {
       }
 
       res.setEncoding('utf8');
-      let responseBody = '';
+      const chunks = [];
       res.on('data', chunk => {
-        responseBody += chunk;
+        chunks.push(chunk);
       });
       res.on('end', () => {
+        const responseBody = chunks.join('');
+        chunks.length = 0; // free memory
         cb(null, responseBody);
       });
     }

packages/collector/src/announceCycle/agentready.js

@@ -178,7 +178,7 @@ function leave() {
 }
 
 function sendTracingMetrics() {
-  const payload = tracing._getAndResetTracingMetrics();
+  let payload = tracing._getAndResetTracingMetrics();
 
   agentConnection.sendTracingMetricsToAgent(payload, error => {
     if (error) {
@@ -194,6 +194,8 @@ function sendTracingMetrics() {
         return;
       }
     }
+
+    payload = null;
     scheduleTracingMetrics();
   });
 }
@@ -222,30 +224,31 @@ function fireMonitoringEvent() {
 
 function sendEOLEvent() {
   const pid = pidStore.getEntityId();
-  agentConnection.sendEvent(
-    {
-      title: `Node.js version ${process.versions.node} reached its end of life`,
-      text:
-        'This version no longer receives updates or security fixes and might contain unfixed vulnerabilities.\n\n' +
-        'Please consider upgrading Node.js to an active version.\n\n' +
-        'For a list of active versions visit ' +
-        '[https://nodejs.org/en/about/releases/](https://nodejs.org/en/about/releases/)',
-      plugin: 'com.instana.forge.infrastructure.runtime.nodejs.NodeJsRuntimePlatform',
-      id: pid,
-      timestamp: Date.now(),
-      duration: EOL_EVENT_DURATION,
-      severity: agentConnection.AgentEventSeverity.WARNING,
-      path: `${agentOpts.agentUuid}/${pid}/nodejs-eol`
-    },
-    err => {
-      if (err) {
-        logger.debug(
-          `Sending a monitoring event for the Node.js version end-of-life check has failed. 
+  let eventData = {
+    title: `Node.js version ${process.versions.node} reached its end of life`,
+    text:
+      'This version no longer receives updates or security fixes and might contain unfixed vulnerabilities.\n\n' +
+      'Please consider upgrading Node.js to an active version.\n\n' +
+      'For a list of active versions visit ' +
+      '[https://nodejs.org/en/about/releases/](https://nodejs.org/en/about/releases/)',
+    plugin: 'com.instana.forge.infrastructure.runtime.nodejs.NodeJsRuntimePlatform',
+    id: pid,
+    timestamp: Date.now(),
+    duration: EOL_EVENT_DURATION,
+    severity: agentConnection.AgentEventSeverity.WARNING,
+    path: `${agentOpts.agentUuid}/${pid}/nodejs-eol`
+  };
+
+  agentConnection.sendEvent(eventData, err => {
+    if (err) {
+      logger.debug(
+        `Sending a monitoring event for the Node.js version end-of-life check has failed.
           ${err?.message} ${err?.stack}`
-        );
-      }
+      );
     }
-  );
+
+    eventData = null;
+  });
 }
 
 /**
@@ -256,6 +259,7 @@ function sendEOLEvent() {
  */
 function detectEOLNodeVersion() {
   if (isNodeVersionEOL()) {
+    // TODO: Does this need a deactivate method?
     setTimeout(() => {
       sendEOLEvent();
       setInterval(sendEOLEvent, EOL_EVENT_REFRESH_INTERVAL).unref();

packages/collector/src/metrics/transmissionCycle.js

@@ -90,7 +90,8 @@ function sendMetrics() {
   }
 
   // clone retrieved objects to allow mutations in metric retrievers
-  const newValueToTransmit = core.util.clone(metrics.gatherData());
+  // is there a memory problem with this line?
+  let newValueToTransmit = { ...metrics.gatherData() };
 
   /** @type {Object<string, *>} */
   let payload;
@@ -101,7 +102,10 @@ function sendMetrics() {
     payload = core.util.compression(previousTransmittedValue, newValueToTransmit);
   }
 
-  downstreamConnection.sendMetrics(payload, onMetricsHaveBeenSent.bind(null, isFullTransmission, newValueToTransmit));
+  downstreamConnection.sendMetrics(payload, (error, responsePayload) => {
+    onMetricsHaveBeenSent(isFullTransmission, payload, error, responsePayload);
+    newValueToTransmit = null;
+  });
 }
 
 /**
@@ -116,9 +120,14 @@ function onMetricsHaveBeenSent(isFullTransmission, transmittedValue, error, resp
       onError();
     }
 
+    transmittedValue = null;
+    responsePayload = null;
     return;
   }
+
+  previousTransmittedValue = null;
   previousTransmittedValue = transmittedValue;
+
   if (isFullTransmission) {
     transmissionsSinceLastFullDataEmit = 0;
   } else {
@@ -127,6 +136,9 @@ function onMetricsHaveBeenSent(isFullTransmission, transmittedValue, error, resp
   if (onSuccess) {
     onSuccess(responsePayload);
   }
+
+  transmittedValue = null;
+
   transmissionTimeoutHandle = setTimeout(sendMetrics, transmissionDelay);
   transmissionTimeoutHandle.unref();
 }

packages/core/src/tracing/instrumentation/process/process.js

@@ -85,6 +85,11 @@ function shimProcessSendForBullChildWorker(originalProcessSend) {
           clearInterval(keepAliveHandle);
         }
       }, 100);
+
+      // Safely clear the keep alive handle after a while, so that we do not leak memory.
+      setTimeout(() => {
+        clearInterval(keepAliveHandle);
+      }, bullKeepAlive + 1000).unref();
     }
     return originalProcessSend.apply(this, arguments);
   };

packages/core/src/tracing/spanBuffer.js

@@ -168,6 +168,12 @@ exports.deactivate = function deactivate() {
   spans = [];
   batchingBuckets.clear();
   clearTimeout(transmissionTimeoutHandle);
+
+  // CASE: If we have a pre-activation cleanup interval, we clear it.
+  if (preActivationCleanupIntervalHandle) {
+    clearInterval(preActivationCleanupIntervalHandle);
+    preActivationCleanupIntervalHandle = null;
+  }
 };
 
 /**
@@ -439,7 +445,7 @@ function transmitSpans() {
     return;
   }
 
-  const spansToSend = spans;
+  let spansToSend = spans;
   spans = [];
   batchingBuckets.clear();
 
@@ -455,6 +461,8 @@ function transmitSpans() {
       removeSpansIfNecessary();
     }
 
+    spansToSend = null;
+
     if (!isFaaS) {
       transmissionTimeoutHandle = setTimeout(transmitSpans, transmissionDelay);
       transmissionTimeoutHandle.unref();

packages/shared-metrics/src/heapSpaces.js

@@ -31,7 +31,7 @@ exports.activate = function activate() {
 };
 
 function gatherHeapSpaceStatistics() {
-  const rawStats = v8.getHeapSpaceStatistics();
+  let rawStats = v8.getHeapSpaceStatistics();
 
   // We are changing the native format to a format which can be more
   // efficiently compressed and processed in the backend.
@@ -48,6 +48,8 @@ function gatherHeapSpaceStatistics() {
     };
   }
 
+  rawStats = null;
+  exports.currentPayload = null;
   // @ts-ignore
   exports.currentPayload = processedStats;
 }

packages/shared-metrics/src/memory.js

@@ -19,6 +19,8 @@ exports.activate = function activate() {
 };
 
 function gatherMemoryUsageStatistics() {
+  exports.currentPayload = null;
+
   // @ts-ignore
   exports.currentPayload = process.memoryUsage();
 }