ProcessControls.js

/*
 * (c) Copyright IBM Corp. 2021
 * (c) Copyright Instana Inc. and contributors 2018
 */

/* eslint-disable max-len */

'use strict';

const _ = require('lodash');
const fork = require('child_process').fork;
const fs = require('fs');
const path = require('path');
const config = require('@_local/core/test/config');
const http2Promise = require('./http2Promise');
const testUtils = require('@_local/core/test/test_util');
const globalAgent = require('../globalAgent');
const portFinder = require('./portfinder');
const sslDir = path.join(__dirname, '..', 'apps', 'ssl');
const cert = fs.readFileSync(path.join(sslDir, 'cert'));

class ProcessControls {
  /**
   * @typedef {Object} ProcessControlsOptions
   * @property {string} [appPath]
   * @property {string} [appName]
   * @property {string} [cwd]
   * @property {number} [port]
   * @property {string} [dirname]
   * @property {boolean} [dontKillInAfterHook]
   * @property {boolean} [http2]
   * @property {Array.<string>} [args]
   * @property {Array.<string>} [execArgv]
   * @property {number} [minimalDelay]
   * @property {boolean} [usePreInit]
   * @property {boolean} [useGlobalAgent]
   * @property {boolean} [tracingEnabled]
   * @property {*} [agentControls]
   * @property {Object.<string, *} [env]
   */

  /**
   * @param {ProcessControlsOptions} opts
   */
  constructor(opts = {}) {
    if (!opts.dirname) {
      throw new Error('[ProcessControls] dirname is required');
    }

    if (!opts.cwd) {
      opts.cwd = opts.dirname;
    }

    if (process.env.RUN_ESM && !opts.execArgv) {
      const esmLoader = [
        `--import=${
          opts.esmLoaderPath
            ? opts.esmLoaderPath
            : path.join(opts.cwd, 'node_modules', '@instana', 'collector', 'esm-register.mjs')
        }`
      ];

      try {
        if (opts?.appName) {
          const appName = opts.appName.endsWith('.mjs') ? opts.appName : `${opts.appName}.mjs`;
          const appPath = path.join(opts.cwd, appName);
          const esmApp = testUtils.checkESMApp({ appPath });

          if (esmApp) {
            opts.execArgv = esmLoader;
            opts.appPath = appPath;
          }
        } else {
          const appPath = path.join(opts.cwd, 'app.mjs');
          const esmApp = testUtils.checkESMApp({ appPath });
          if (esmApp) {
            opts.execArgv = esmLoader;
            opts.appPath = appPath;
          }
        }
      } catch (err) {
        // eslint-disable-next-line no-console
        console.log('[ProcessControls] Unable to load the target app.mjs', err);
      }
    }

    if (!opts.appPath) {
      if (process.env.RUN_ESM) {
        console.log('[ProcessControls] No ESM app found.');
        this.noESMApp = true;
      }

      opts.appPath = path.join(opts.dirname, opts.appName ? opts.appName : 'app.js');
    }

    this.collectorUninitialized = opts.collectorUninitialized;

    this.processLogs = [];

    // absolute path to .js file that should be executed
    this.appPath = opts.appPath;
    // optional working directory for the child process
    this.cwd = opts.cwd;
    // for scenarios where the app under tests terminates on its own
    this.dontKillInAfterHook = opts.dontKillInAfterHook;
    // arguments for the app under test
    this.args = opts.args;
    // command line flags for the Node.js executable
    this.execArgv = opts.execArgv;
    // server http2
    this.http2 = opts.http2;
    this.appUsesHttps = 'appUsesHttps' in opts ? opts.appUsesHttps : false;

    if (this.appUsesHttps) {
      process.env.NODE_TLS_REJECT_UNAUTHORIZED = '0';
    }

    // By default, we test without OpenTelemetry instrumentation enabled
    // because the test setup is currently broken and not capturing OTEL spans.
    // TODO: INSTA-62539
    this.enableOtelIntegration = opts.enableOtelIntegration || false;
    // http/https/http2 port
    this.port = opts.port || process.env.APP_PORT || portFinder();

    this.tracingEnabled = opts.tracingEnabled !== false;
    this.usePreInit = opts.usePreInit === true;

    this.pipeSubprocessLogs = 'pipeSubprocessLogs' in opts ? opts.pipeSubprocessLogs : false;

    // Signals that this process intends to connect to the test suite's global agent stub on port 3211. Setting this to
    // true will result in a before/beforeEach call which ensures that the collector is successfully connected to that
    // agent.
    this.useGlobalAgent = opts.useGlobalAgent;

    // As an alternative to connecting to the global agent, process control instances can use an individual instance of
    // AgentStubControls (and consequently their own agent stub process). Passing an agent control instance will result
    // in a before/beforeEach call which ensures that the collector is successfully connected to that agent.
    this.useGlobalAgent = opts.useGlobalAgent;
    this.agentControls = opts.agentControls;
    if (!this.agentControls && this.useGlobalAgent) {
      this.agentControls = globalAgent.instance;
    }

    const agentPort = this.agentControls ? this.agentControls.agentPort : undefined;

    this.env = _.assign(
      {},
      process.env,
      {
        APP_PORT: this.port,
        APP_CWD: this.cwd,
        INSTANA_AGENT_PORT: agentPort,
        INSTANA_LOG_LEVEL: 'warn',
        INSTANA_FORCE_TRANSMISSION_STARTING_AT: '1',
        INSTANA_FULL_METRICS_INTERNAL_IN_S: 1,
        INSTANA_FIRE_MONITORING_EVENT_DURATION_IN_MS: 500,
        INSTANA_RETRY_AGENT_CONNECTION_IN_MS: 500,
        APP_USES_HTTPS: this.appUsesHttps ? 'true' : 'false',
        INSTANA_DISABLE_USE_OPENTELEMETRY: !this.enableOtelIntegration,
        LIBRARY_VERSION: process.env.LIBRARY_VERSION,
        LIBRARY_NAME: process.env.LIBRARY_NAME
      },
      opts.env
    );

    // Only set INSTANA_TRACING_DISABLE when tracing is actually disabled to avoid
    // overriding other disable environment variables (INSTANA_TRACING_DISABLE_INSTRUMENTATIONS, etc.)
    // See packages/core/src/config/configNormalizers/disable.js for precedence rules
    if (!this.tracingEnabled) {
      this.env.INSTANA_TRACING_DISABLE = 'true';
    }

    if (this.usePreInit) {
      this.env.INSTANA_EARLY_INSTRUMENTATION = 'true';
    }

    this.receivedIpcMessages = [];
  }

  getPort() {
    return this.port;
  }

  async start(retryTime, until, skipWaitUntilServerIsUp = false) {
    if (this.noESMApp) return;

    const that = this;
    this.receivedIpcMessages = [];

    // Will pipe stdout/stderr to the parent process.
    // We log & remember them in the event listener (process.stdout.on)
    const stdio = this.pipeSubprocessLogs ? ['pipe', 'pipe', 'pipe', 'ipc'] : config.getAppStdio();

    const forkConfig = {
      stdio,
      env: this.env
    };

    if (this.cwd) {
      forkConfig.cwd = this.cwd;
    }
    if (this.execArgv) {
      forkConfig.execArgv = this.execArgv;
    }

    if (!forkConfig.execArgv) {
      forkConfig.execArgv = [];
    }

    this.process = this.args ? fork(this.appPath, this.args || [], forkConfig) : fork(this.appPath, forkConfig);

    this.process.on('message', message => {
      if (message === 'instana.collector.initialized') {
        this.process.collectorInitialized = true;
      } else {
        that.receivedIpcMessages.push(message);
      }
    });

    this.process.stdout &&
      this.process.stdout.on('data', data => {
        // eslint-disable-next-line no-console
        console.log('Child Stdout:', data.toString());
        this.processLogs.push(data.toString());
      });

    this.process.stderr &&
      this.process.stderr.on('data', data => {
        // eslint-disable-next-line no-console
        console.log('Child Stderr:', data.toString());
        this.processLogs.push(data.toString());
      });

    if (skipWaitUntilServerIsUp) return;
    await this.waitUntilServerIsUp(retryTime, until);
  }

  async stop() {
    await this.kill();
  }

  async waitUntilServerIsUp(retryTime, until) {
    if (this.noESMApp) return;
    try {
      await testUtils.retry(
        async () => {
          await this.sendRequest({
            method: 'GET',
            suppressTracing: true,
            checkStatusCode: true
          });

          if (this.collectorUninitialized) return;
          if (!this.process.collectorInitialized) throw new Error('Collector not fullly initialized.');
        },
        retryTime,
        until
      );

      // eslint-disable-next-line no-console
      console.log('[ProcessControls] server is up.');
    } catch (err) {
      // eslint-disable-next-line no-console
      console.log(`[ProcessControls] error: ${err}${err.cause ? ` | cause: ${err.cause}` : ''}`);
      throw err;
    }
  }

  getProcessLogs() {
    return this.processLogs;
  }

  async startAndWaitForAgentConnection(retryTime, until) {
    if (this.noESMApp) return;

    // eslint-disable-next-line no-console
    console.log(
      `[ProcessControls] start with port: ${this.getPort()}, agentPort: ${this.agentControls.getPort()}, appPath: ${
        this.appPath
      }`
    );

    await this.clearIpcMessages();
    await this.start(retryTime, until);
    await this.agentControls.waitUntilAppIsCompletelyInitialized(this.getPid(), retryTime, until);

    // eslint-disable-next-line no-console
    console.log(
      `[ProcessControls] started with port: ${this.getPort()}, agentPort: ${this.agentControls.getPort()}, appPath: ${
        this.appPath
      }, pid: ${this.process.pid}`
    );
  }

  async waitForAgentConnection() {
    if (this.noESMApp) return;
    await this.agentControls.waitUntilAppIsCompletelyInitialized(this.getPid());
  }

  clearIpcMessages() {
    this.receivedIpcMessages = [];
  }

  getPid() {
    if (!this.process) {
      return false;
    }
    return this.process.pid;
  }

  /**
   * @param opts {{
   *   suppressTracing: boolean,
   *   url: string,
   *   json: boolean,
   *   ca: Buffer,
   *   headers: {
   *     'X-INSTANA-L': '0' | '1',
   *     [key:string]: any
   *   }
   * }} The request options
   */
  async sendRequest(opts = {}) {
    if (this.noESMApp) return Promise.resolve();
    const requestOpts = Object.assign({}, opts);
    const resolveWithFullResponse = requestOpts.resolveWithFullResponse;
    const checkStatusCode = requestOpts.checkStatusCode;

    // NOTE: http2Promise.request has an inbuild property called "resolveWithFullResponse"
    //       fetch does not have, manual implementation.
    delete requestOpts.resolveWithFullResponse;
    delete requestOpts.checkStatusCode;

    const baseUrl = this.getBaseUrl(opts);

    if (this.http2) {
      requestOpts.baseUrl = baseUrl;
      requestOpts.resolveWithFullResponse = resolveWithFullResponse;
      return http2Promise.request(requestOpts);
    } else {
      if (requestOpts.suppressTracing === true) {
        requestOpts.headers = requestOpts.headers || {};
        requestOpts.headers['X-INSTANA-L'] = '0';
      }

      requestOpts.url = baseUrl + (requestOpts.path || '');

      if (requestOpts.qs) {
        const queryParams = Object.entries(requestOpts.qs)
          .map(([key, value]) => {
            if (Array.isArray(value)) {
              return value.map(v => `${encodeURIComponent(key)}=${encodeURIComponent(v)}`).join('&');
            } else {
              return `${encodeURIComponent(key)}=${encodeURIComponent(value)}`;
            }
          })
          .join('&');

        requestOpts.url = requestOpts.url.includes('?')
          ? `${requestOpts.url}&${queryParams}`
          : `${requestOpts.url}?${queryParams}`;
      }

      requestOpts.json = true;
      requestOpts.ca = cert;

      // NOTE: The opts.body passed in must be serialized.
      // We need to JSON.stringify() the body and set Content-Type manually.
      // The "json: true" (requestOpts.json = true) option is ignored by native fetch.
      if (requestOpts.body && typeof requestOpts.body === 'object') {
        requestOpts.body = JSON.stringify(requestOpts.body);
        requestOpts.headers = requestOpts.headers || {};
        requestOpts.headers['Content-Type'] = 'application/json';
      }

      // Handle timeout using AbortController (native fetch doesn't support timeout option directly)
      let timeoutId;
      if (requestOpts.timeout) {
        const controller = new AbortController();
        requestOpts.signal = controller.signal;

        timeoutId = setTimeout(() => {
          controller.abort();
        }, requestOpts.timeout);
      }

      let response;
      try {
        const result = await fetch(requestOpts.url, requestOpts);

        // Clear timeout if request completes successfully
        if (timeoutId) {
          clearTimeout(timeoutId);
        }

        const contentType = result.headers.get('content-type');

        if (contentType && contentType.includes('application/json')) {
          response = await result.json();
        } else if (contentType && (contentType.includes('text/html') || contentType.includes('text/plain'))) {
          response = await result.text();
        } else {
          // CASE: Some tests do not use express and then the header is missing
          //       Some tests use `res.send`, which is not a JSON response.
          response = await result.text();
        }

        if (checkStatusCode) {
          if (result.status < 200 || result.status >= 300) {
            throw new Error(response);
          }
        }

        if (resolveWithFullResponse) {
          return {
            headers: result.headers,
            body: response
          };
        }

        return response;
      } catch (err) {
        if (timeoutId) {
          clearTimeout(timeoutId);
        }

        if (err.name === 'AbortError') {
          const timeoutError = new Error('Request timeout');
          timeoutError.type = 'request-timeout';
          timeoutError.error = { code: 'ETIMEDOUT' };
          throw timeoutError;
        }

        throw err;
      }
    }
  }

  getBaseUrl({ embedCredentialsInUrl }) {
    return `${this.appUsesHttps || this.http2 ? 'https' : 'http'}://${
      // eslint-disable-next-line no-unneeded-ternary
      embedCredentialsInUrl ? embedCredentialsInUrl : ''
    }localhost:${this.port}`;
  }

  sendViaIpc(message) {
    if (this.noESMApp) return;
    this.process.send(message);
  }

  getIpcMessages() {
    return this.receivedIpcMessages;
  }

  kill() {
    if (!this.process) {
      return Promise.resolve();
    }
    if (this.process.killed || this.dontKillInAfterHook) {
      return Promise.resolve();
    }

    // eslint-disable-next-line no-console
    console.log(
      `[ProcessControls] stopping with port: ${this.getPort()}, agentPort: ${
        this.agentControls && this.agentControls.getPort && this.agentControls.getPort()
      }, appPath: ${this.appPath}, pid: ${this.process.pid}`
    );

    return new Promise(resolve => {
      this.process.once('exit', () => {
        this.process.pid = null;

        // eslint-disable-next-line no-console
        console.log(
          `[ProcessControls] stopped with port: ${this.getPort()}, agentPort: ${
            this.agentControls && this.agentControls.getPort && this.agentControls.getPort()
          }, appPath: ${this.appPath}, pid: ${this.process.pid}`
        );

        resolve();
      });

      // Sends SIGTERM to the child process to terminate it gracefully.
      this.process.kill();
    });
  }

  toString() {
    return `${this.process && this.process.pid ? this.process.pid : '-'} (${this.appPath})`;
  }
}

module.exports = ProcessControls;