Connect My Computer: Join cluster

jest.config.js

@@ -2,6 +2,8 @@ const config = require('@gravitational/build/jest/config');
 
 process.env.TZ = 'UTC';
 
+const esModules = ['strip-ansi-stream', 'ansi-regex'].join('|');
+
 /** @type {import('@jest/types').Config.InitialOptions} */
 module.exports = {
   ...config,
@@ -13,6 +15,7 @@ module.exports = {
     // '**/packages/design/src/**/*.jsx',
     '**/packages/shared/components/**/*.jsx',
   ],
+  transformIgnorePatterns: [`/node_modules/(?!${esModules})`],
   coverageReporters: ['text-summary', 'lcov'],
   setupFilesAfterEnv: ['<rootDir>/web/packages/shared/setupTests.tsx'],
 };

web/packages/teleterm/package.json

@@ -31,6 +31,7 @@
     "@types/tar-fs": "^2.0.1",
     "emittery": "^1.0.1",
     "node-pty": "0.11.0-beta29",
+    "strip-ansi-stream": "^2.0.1",
     "tar-fs": "^3.0.3"
   },
   "devDependencies": {

web/packages/teleterm/src/mainProcess/agentDownloader/agentDownloader.ts

@@ -45,18 +45,16 @@ interface AgentBinary {
 /**
  * Downloads and unpacks the agent binary, if it has not already been downloaded.
  *
- * The agent version to download is taken from settings.appVersion if it is not a dev version (1.0.0-dev).
- * The settings.appVersion is set to a real version only for packaged apps that went through our CI build pipeline.
- * In local builds, both for the development version and for packaged apps, settings.appVersion is set to 1.0.0-dev.
- * In those cases, we fetch the latest available stable version of the agent.
+ * The agent version to download is taken from settings.appVersion if settings.isLocalBuild is false.
+ * If it isn't, we fetch the latest available stable version of the agent.
  * CONNECT_CMC_AGENT_VERSION is available as an escape hatch for cases where we want to fetch a different version.
  */
 export async function downloadAgent(
   fileDownloader: IFileDownloader,
   settings: RuntimeSettings,
   env: Record<string, any>
 ): Promise<void> {
-  const version = await calculateAgentVersion(settings.appVersion, env);
+  const version = await calculateAgentVersion(settings, env);
 
   if (
     await isCorrectAgentVersionAlreadyDownloaded(
@@ -87,11 +85,11 @@ export async function downloadAgent(
 }
 
 async function calculateAgentVersion(
-  appVersion: string,
+  settings: RuntimeSettings,
   env: Record<string, any>
 ): Promise<string> {
-  if (appVersion !== '1.0.0-dev') {
-    return appVersion;
+  if (!settings.isLocalBuild) {
+    return settings.appVersion;
   }
   if (env.CONNECT_CMC_AGENT_VERSION) {
     return env.CONNECT_CMC_AGENT_VERSION;

web/packages/teleterm/src/mainProcess/agentRunner/agentRunner.test.ts

@@ -0,0 +1,151 @@
+/*
+Copyright 2023 Gravitational, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+import path from 'node:path';
+
+import Logger, { NullService } from 'teleterm/logger';
+import { RootClusterUri } from 'teleterm/ui/uri';
+
+import { makeRuntimeSettings } from '../fixtures/mocks';
+import { AgentProcessState } from '../types';
+
+import { AgentRunner } from './agentRunner';
+
+beforeEach(() => {
+  Logger.init(new NullService());
+});
+
+const userDataDir = '/Users/test/Application Data/Teleport Connect';
+const agentBinaryPath = path.join(__dirname, 'agentTestProcess.mjs');
+const rootClusterUri: RootClusterUri = '/clusters/cluster.local';
+
+test('agent process starts with correct arguments', async () => {
+  const agentRunner = new AgentRunner(
+    makeRuntimeSettings({
+      agentBinaryPath,
+      userDataDir,
+    }),
+    () => {}
+  );
+
+  try {
+    const agentProcess = await agentRunner.start(rootClusterUri);
+
+    expect(agentProcess.spawnargs).toEqual([
+      agentBinaryPath,
+      'start',
+      `--config=${userDataDir}/agents/cluster.local/config.yaml`,
+    ]);
+  } finally {
+    await agentRunner.killAll();
+  }
+});
+
+test('previous agent process is killed when a new one is started', async () => {
+  const agentRunner = new AgentRunner(
+    makeRuntimeSettings({
+      agentBinaryPath,
+      userDataDir,
+    }),
+    () => {}
+  );
+
+  try {
+    const firstProcess = await agentRunner.start(rootClusterUri);
+    await agentRunner.start(rootClusterUri);
+
+    expect(firstProcess.killed).toBeTruthy();
+  } finally {
+    await agentRunner.killAll();
+  }
+});
+
+test('status updates are sent on a successful start', async () => {
+  const updateSender = jest.fn();
+  const agentRunner = new AgentRunner(
+    makeRuntimeSettings({
+      agentBinaryPath,
+      userDataDir,
+    }),
+    updateSender
+  );
+
+  try {
+    expect(agentRunner.getState(rootClusterUri)).toBeUndefined();
+    const agentProcess = await agentRunner.start(rootClusterUri);
+    expect(agentRunner.getState(rootClusterUri)).toStrictEqual({
+      status: 'not-started',
+    } as AgentProcessState);
+    await new Promise((resolve, reject) => {
+      const timeout = setTimeout(
+        () => reject('Process start timed out.'),
+        4_000
+      );
+      agentProcess.once('spawn', () => {
+        resolve(undefined);
+        clearTimeout(timeout);
+      });
+    });
+    const runningState: AgentProcessState = { status: 'running' };
+    expect(agentRunner.getState(rootClusterUri)).toStrictEqual(runningState);
+    expect(updateSender).toHaveBeenCalledWith(rootClusterUri, runningState);
+
+    await agentRunner.kill(rootClusterUri);
+    const exitedState: AgentProcessState = {
+      status: 'exited',
+      code: null,
+      stackTrace: undefined,
+      exitedSuccessfully: true,
+      signal: 'SIGTERM',
+    };
+    expect(agentRunner.getState(rootClusterUri)).toStrictEqual(exitedState);
+    expect(updateSender).toHaveBeenCalledWith(rootClusterUri, exitedState);
+
+    expect(updateSender).toHaveBeenCalledTimes(2);
+  } finally {
+    await agentRunner.killAll();
+  }
+});
+
+test('status updates are sent on a failed start', async () => {
+  const updateSender = jest.fn();
+  const nonExisingPath = path.join(
+    __dirname,
+    'agentTestProcess-nonExisting.mjs'
+  );
+  const agentRunner = new AgentRunner(
+    makeRuntimeSettings({
+      agentBinaryPath: nonExisingPath,
+      userDataDir,
+    }),
+    updateSender
+  );
+
+  try {
+    const agentProcess = await agentRunner.start(rootClusterUri);
+    await new Promise(resolve => agentProcess.on('error', resolve));
+
+    expect(updateSender).toHaveBeenCalledTimes(1);
+    const errorState: AgentProcessState = {
+      status: 'error',
+      message: expect.stringContaining('ENOENT'),
+    };
+    expect(agentRunner.getState(rootClusterUri)).toStrictEqual(errorState);
+    expect(updateSender).toHaveBeenCalledWith(rootClusterUri, errorState);
+  } finally {
+    await agentRunner.killAll();
+  }
+});

web/packages/teleterm/src/mainProcess/agentRunner/agentRunner.ts

@@ -0,0 +1,179 @@
+/**
+ * Copyright 2023 Gravitational, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { spawn, ChildProcess } from 'node:child_process';
+import os from 'node:os';
+
+import stripAnsiStream from 'strip-ansi-stream';
+
+import Logger from 'teleterm/logger';
+import { RootClusterUri } from 'teleterm/ui/uri';
+
+import { generateAgentConfigPaths } from '../createAgentConfigFile';
+import { AgentProcessState, RuntimeSettings } from '../types';
+import { terminateWithTimeout } from '../terminateWithTimeout';
+
+const MAX_STDERR_LINES = 10;
+
+export class AgentRunner {
+  private logger = new Logger('AgentRunner');
+  private agentProcesses = new Map<
+    RootClusterUri,
+    {
+      process: ChildProcess;
+      state: AgentProcessState;
+    }
+  >();
+
+  constructor(
+    private settings: RuntimeSettings,
+    private sendProcessState: (
+      rootClusterUri: RootClusterUri,
+      state: AgentProcessState
+    ) => void
+  ) {}
+
+  /**
+   * Starts a new agent process.
+   * If an existing process exists for the given root cluster, the old one will be killed.
+   */
+  async start(rootClusterUri: RootClusterUri): Promise<ChildProcess> {
+    if (this.agentProcesses.has(rootClusterUri)) {
+      await this.kill(rootClusterUri);
+    }
+
+    const { agentBinaryPath } = this.settings;
+    const { configFile } = generateAgentConfigPaths(
+      this.settings,
+      rootClusterUri
+    );
+
+    const args = [
+      'start',
+      `--config=${configFile}`,
+      this.settings.isLocalBuild && '--skip-version-check',
+    ].filter(Boolean);
+
+    this.logger.info(
+      `Starting agent for ${rootClusterUri} from ${agentBinaryPath} with arguments ${args.join(
+        ' '
+      )}`
+    );
+
+    const agentProcess = spawn(agentBinaryPath, args, {
+      windowsHide: true,
+    });
+
+    this.agentProcesses.set(rootClusterUri, {
+      process: agentProcess,
+      state: { status: 'not-started' },
+    });
+    this.addListeners(rootClusterUri, agentProcess);
+
+    return agentProcess;
+  }
+
+  getState(rootClusterUri: RootClusterUri): AgentProcessState | undefined {
+    return this.agentProcesses.get(rootClusterUri)?.state;
+  }
+
+  async kill(rootClusterUri: RootClusterUri): Promise<void> {
+    const agent = this.agentProcesses.get(rootClusterUri);
+    if (!agent) {
+      this.logger.warn(`Cannot get an agent to kill for ${rootClusterUri}`);
+      return;
+    }
+    await terminateWithTimeout(agent.process);
+    this.logger.info(`Killed agent for ${rootClusterUri}`);
+  }
+
+  async killAll(): Promise<void> {
+    const processes = Array.from(this.agentProcesses.values());
+    await Promise.all(
+      processes.map(async agent => {
+        await terminateWithTimeout(agent.process);
+      })
+    );
+  }
+
+  private addListeners(
+    rootClusterUri: RootClusterUri,
+    process: ChildProcess
+  ): void {
+    // Teleport logs output to stderr.
+    let stderrOutput = '';
+    process.stderr.setEncoding('utf-8');
+    process.stderr.pipe(stripAnsiStream()).on('data', (error: string) => {
+      stderrOutput += error;
+      stderrOutput = limitProcessOutputLines(stderrOutput);
+    });
+
+    const spawnHandler = () => {
+      this.updateProcessState(rootClusterUri, {
+        status: 'running',
+      });
+    };
+
+    const errorHandler = (error: Error) => {
+      process.off('spawn', spawnHandler);
+
+      this.updateProcessState(rootClusterUri, {
+        status: 'error',
+        message: `${error}`,
+      });
+    };
+
+    const exitHandler = (
+      code: number | null,
+      signal: NodeJS.Signals | null
+    ) => {
+      // Remove handlers when the process exits.
+      process.off('error', errorHandler);
+      process.off('spawn', spawnHandler);
+
+      const exitedSuccessfully = code === 0 || signal === 'SIGTERM';
+
+      this.updateProcessState(rootClusterUri, {
+        status: 'exited',
+        code,
+        signal,
+        exitedSuccessfully,
+        stackTrace: exitedSuccessfully ? undefined : stderrOutput,
+      });
+    };
+
+    process.once('spawn', spawnHandler);
+    process.once('error', errorHandler);
+    process.once('exit', exitHandler);
+  }
+
+  private updateProcessState(
+    rootClusterUri: RootClusterUri,
+    state: AgentProcessState
+  ): void {
+    this.logger.info(
+      `Updating agent state ${rootClusterUri}: ${JSON.stringify(state)}`
+    );
+
+    const agent = this.agentProcesses.get(rootClusterUri);
+    agent.state = state;
+    this.sendProcessState(rootClusterUri, state);
+  }
+}
+
+function limitProcessOutputLines(output: string): string {
+  return output.split(os.EOL).slice(-MAX_STDERR_LINES).join(os.EOL);
+}