Connect My Computer: Join cluster

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,149 @@
+/*
+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,
+      signal: 'SIGTERM',
+    };
+    expect(agentRunner.getState(rootClusterUri)).toBeUndefined(); // the agent has been killed and removed
+    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,175 @@
+/**
+ * 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 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) {
+      await terminateWithTimeout(
+        this.agentProcesses.get(rootClusterUri).process
+      );
+      this.agentProcesses.delete(rootClusterUri);
+      this.logger.info(`Killed agent for ${rootClusterUri}`);
+    }
+    this.logger.warn(`Cannot get an agent to kill for ${rootClusterUri}`);
+  }
+
+  async killAll(): Promise<void> {
+    const processes = Array.from(this.agentProcesses.entries());
+    await Promise.all(
+      processes.map(async ([rootClusterUri, agent]) => {
+        await terminateWithTimeout(agent.process);
+        this.agentProcesses.delete(rootClusterUri);
+      })
+    );
+  }
+
+  private addListeners(
+    rootClusterUri: RootClusterUri,
+    process: ChildProcess
+  ): void {
+    // Teleport logs output to stderr.
+    let stderrOutput = '';
+    process.stderr.setEncoding('utf-8');
+    process.stderr.on('data', error => {
+      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);
+
+      this.updateProcessState(rootClusterUri, {
+        status: 'exited',
+        code,
+        signal,
+        stackTrace: signal !== 'SIGTERM' ? stderrOutput : undefined,
+      });
+    };
+
+    process.once('spawn', spawnHandler);
+    process.once('error', errorHandler);
+    process.once('exit', exitHandler);
+  }
+
+  private updateProcessState(
+    rootClusterUri: RootClusterUri,
+    state: AgentProcessState
+  ): void {
+    const agent = this.agentProcesses.get(rootClusterUri);
+    if (agent) {
+      agent.state = state;
+    }
+    this.sendProcessState(rootClusterUri, state);
+  }
+}
+
+function limitProcessOutputLines(output: string): string {
+  return output.split(os.EOL).slice(-MAX_STDERR_LINES).join(os.EOL);
+}

web/packages/teleterm/src/mainProcess/agentRunner/agentTestProcess.mjs

@@ -0,0 +1,21 @@
+#!/usr/bin/env node
+/**
+ * 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 { setTimeout } from 'node:timers/promises';
+
+await setTimeout(10_000);