fix(metrics): do not report metrics from worker threads (#517)

refs 88823, refs #500

When instrumented via NODE_OPTIONS="--require ..." (which is inherited by
worker threads), each worker thread will announce independently and act like
the main thread. This can cause issues, in particular if the worker thread is
started from a dependency in node_ module and it identifies a different
package.json as its main package.json, hence reporting a different application
name etc.

To prevent this, we disable sending metrics and snapshot data from
worker threads for now.

packages/collector/src/announceCycle/agentready.js

@@ -5,6 +5,13 @@
 
 'use strict';
 
+let isMainThread = true;
+try {
+  isMainThread = require('worker_threads').isMainThread;
+} catch (err) {
+  // Worker threads are not available, so we know that this is the main thread.
+}
+
 const semver = require('semver');
 const { tracing } = require('@instana/core');
 
@@ -24,6 +31,9 @@ const TEN_MINUTES = 10 * 60 * 1000;
 /** @type {*} */
 let autoprofile;
 
+/** @type {*} */
+let profiler;
+
 /** @type {import('@instana/core/src/logger').GenericLogger} */
 let logger;
 logger = require('../logger').getLogger('announceCycle/agentready', newLogger => {
@@ -71,29 +81,34 @@ module.exports = exports = {
  */
 function enter(_ctx) {
   ctx = _ctx;
-  uncaught.activate();
-  metrics.activate();
+
   initializedTooLate.check();
-  transmissionCycle.activate(
-    metrics,
-    agentConnection,
-    /**
-     * @param {Array.<import('../agent/requestHandler').AnnounceRequest>} requests
-     */
-    function onSuccess(requests) {
-      requestHandler.handleRequests(requests);
-    },
-    function onError() {
-      ctx.transitionTo('unannounced');
-    }
-  );
+
+  if (isMainThread) {
+    uncaught.activate();
+    metrics.activate();
+    requestHandler.activate();
+    transmissionCycle.activate(
+      metrics,
+      agentConnection,
+      /**
+       * @param {Array.<import('../agent/requestHandler').AnnounceRequest>} requests
+       */
+      function onSuccess(requests) {
+        requestHandler.handleRequests(requests);
+      },
+      function onError() {
+        ctx.transitionTo('unannounced');
+      }
+    );
+    scheduleTracingMetrics();
+    detectEOLNodeVersion();
+  }
+
   tracing.activate();
-  requestHandler.activate();
-  scheduleTracingMetrics();
-  detectEOLNodeVersion();
 
   if (agentOpts.autoProfile && autoprofile) {
-    const profiler = autoprofile.start();
+    profiler = autoprofile.start();
     /**
      * @param {*} profiles
      * @param {(...args: *) => *} callback
@@ -110,14 +125,19 @@ function enter(_ctx) {
 }
 
 function leave() {
-  uncaught.deactivate();
-  metrics.deactivate();
-  transmissionCycle.deactivate();
+  if (isMainThread) {
+    uncaught.deactivate();
+    metrics.deactivate();
+    requestHandler.deactivate();
+    transmissionCycle.deactivate();
+    deScheduleTracingMetrics();
+  }
+
   tracing.deactivate();
-  requestHandler.deactivate();
-  if (tracingMetricsTimeout) {
-    clearTimeout(tracingMetricsTimeout);
-    tracingMetricsTimeout = null;
+
+  if (profiler) {
+    profiler.destroy();
+    profiler = null;
   }
 }
 
@@ -140,6 +160,13 @@ function scheduleTracingMetrics() {
   tracingMetricsTimeout.unref();
 }
 
+function deScheduleTracingMetrics() {
+  if (tracingMetricsTimeout) {
+    clearTimeout(tracingMetricsTimeout);
+    tracingMetricsTimeout = null;
+  }
+}
+
 function fireMonitoringEvent() {
   agentConnection.sendAgentMonitoringEvent('nodejs_collector_native_addon_autoprofile_missing', 'PROFILER', error => {
     if (error) {

packages/collector/src/index.js

@@ -5,6 +5,13 @@
 
 'use strict';
 
+let isMainThread = true;
+try {
+  isMainThread = require('worker_threads').isMainThread;
+} catch (err) {
+  // Worker threads are not available, so we know that this is the main thread.
+}
+
 const path = require('path');
 const instanaNodeJsCore = require('@instana/core');
 const instanaSharedMetrics = require('@instana/shared-metrics');
@@ -66,8 +73,11 @@ function init(_config) {
   agentOpts.init(config);
   instanaNodeJsCore.init(config, agentConnection, pidStore);
   instanaSharedMetrics.setLogger(logger);
-  uncaught.init(config, agentConnection, pidStore);
-  require('./metrics').init(config);
+
+  if (isMainThread) {
+    uncaught.init(config, agentConnection, pidStore);
+    require('./metrics').init(config);
+  }
 
   logger.info('@instana/collector module version:', require(path.join(__dirname, '..', 'package.json')).version);
   require('./announceCycle').start();

packages/collector/src/logger.js

@@ -7,6 +7,15 @@
 
 'use strict';
 
+/** @type number */
+let threadId = 0;
+try {
+  threadId = require('worker_threads').threadId;
+} catch (ignored) {
+  // We are apparently running in a Node.js version that does not have worker threads yet, thus we are on the main
+  // thread (0).
+}
+
 const bunyan = require('bunyan');
 const { logger } = require('@instana/core');
 
@@ -37,6 +46,7 @@ exports.init = function init(config, isReInit) {
     // we create later on.
     parentLogger = bunyan.createLogger({
       name: '@instana/collector',
+      thread: threadId,
       __in: 1
     });
   }

packages/collector/test/metrics/appWithWorkerThread/app.js

@@ -0,0 +1,35 @@
+/*
+ * (c) Copyright IBM Corp. 2022
+ */
+
+'use strict';
+
+if (!process.env.NODE_OPTIONS) {
+  // Either the test instruments this app via NODE_OTIONS=--require... or it expects the app to be instrumneted via
+  // an in-code require.
+  require('../../..')();
+}
+
+const express = require('express');
+const morgan = require('morgan');
+
+const { getLogger } = require('../../../../core/test/test_util');
+
+const app = express();
+const logPrefix = `Worker Thread App (${process.pid}):\t`;
+const log = getLogger(logPrefix);
+
+// starts the worker thread
+require('module-that-creates-a-worker-thread')();
+
+if (process.env.WITH_STDOUT) {
+  app.use(morgan(`${logPrefix}:method :url :status`));
+}
+
+app.get('/', (req, res) => {
+  res.sendStatus(200);
+});
+
+app.listen(process.env.APP_PORT, () => {
+  log(`Listening on port: ${process.env.APP_PORT}`);
+});

packages/collector/test/metrics/appWithWorkerThread/package.json

@@ -0,0 +1,16 @@
+{
+  "name": "worker-thread-test-app",
+  "private": true,
+  "version": "1.0.0",
+  "description": "This is a test application to test snapshot and metrics data.",
+  "repository": {
+    "type": "git",
+    "url": "git+https://github.com/instana/nodejs.git"
+  },
+  "main": "app.js",
+  "author": {
+    "name": "Bastian Krol",
+    "email": "[email protected]"
+  },
+  "license": "MIT"
+}