[FAI-13688] Run source airbyte connector docker container (#101)

Author: jeniii

airbyte-local-cli-nodejs/package-lock.json

@@ -11,8 +11,8 @@
     "lint-fix": "npm run lint -- --fix",
     "clean": "rm -rf lib node_modules package-lock.json",
     "bundle": "npm run build && esbuild src/index.ts --bundle --platform=node --outfile=dist/index.js --external:*.node",
-    "pkg": "npm run bundle && pkg --output ./out/pkg/airbyte-local dist/index.js",
-    "pkg-linuxstatic": "npm run bundle && pkg --output ./out/pkg/airbyte-local -t linuxstatic dist/index.js",
+    "pkg": "npm run bundle && pkg --debug --output ./out/pkg/airbyte-local dist/index.js",
+    "pkg-linuxstatic": "npm run bundle && pkg --debug --output ./out/pkg/airbyte-local -t linuxstatic dist/index.js",
     "test": "jest",
     "shellspec": "npm run bundle && pkg --output ./test/exec/airbyte-local dist/index.js && cp -rf ./test/resources ./test/exec/resources && shellspec --chdir ./test/exec"
   },
@@ -35,16 +35,16 @@
     "@tsconfig/strictest": "^2.0.5",
     "@types/dockerode": "^3.3.32",
     "@types/jest": "^29.5.14",
-    "@types/lodash": "^4.17.13",
+    "@types/lodash": "^4.17.14",
     "@types/node": "^20.17.6",
-    "@typescript-eslint/eslint-plugin": "^8.13.0",
+    "@typescript-eslint/eslint-plugin": "^8.19.1",
     "@typescript-eslint/parser": "^8.13.0",
-    "@yao-pkg/pkg": "^6.1.1",
+    "@yao-pkg/pkg": "^6.2.0",
     "eslint": "^8.57.1",
     "eslint-config-faros": "^0.1.0",
     "eslint-config-prettier": "^9.1.0",
     "eslint-plugin-simple-import-sort": "^12.1.1",
-    "prettier": "^3.4.1",
+    "prettier": "^3.4.2",
     "ts-jest": "^29.2.5",
     "tsx": "^4.19.2",
     "typescript": "~5.6.3"
@@ -53,8 +53,8 @@
     "commander": "^12.1.0",
     "dockerode": "^4.0.2",
     "lodash": "^4.17.21",
-    "pino": "^9.5.0",
-    "pino-pretty": "^11.3.0"
+    "pino": "^9.6.0",
+    "pino-pretty": "^13.0.0"
   },
   "jest": {
     "silent": false,

airbyte-local-cli-nodejs/package.json

@@ -1,9 +1,13 @@
+import {writeFileSync} from 'node:fs';
 import {Writable} from 'node:stream';
 
 import Docker from 'dockerode';
 
-import {AirbyteConnectionStatus, AirbyteConnectionStatusMessage, AirbyteMessageType} from './types';
-import {logger, SRC_CONFIG_FILENAME} from './utils';
+import {AirbyteConnectionStatus, AirbyteConnectionStatusMessage, AirbyteMessageType, FarosConfig} from './types';
+import {DEFAULT_STATE_FILE, logger, SRC_CATALOG_FILENAME, SRC_CONFIG_FILENAME} from './utils';
+
+// Constants
+const DEFAULT_MAX_LOG_SIZE = '10m';
 
 // Create a new Docker instance
 let _docker = new Docker();
@@ -97,3 +101,114 @@ export async function checkSrcConnection(tmpDir: string, image: string, srcConfi
     throw new Error(`Failed to validate source connection: ${error.message ?? JSON.stringify(error)}.`);
   }
 }
+
+/**
+ * Spinning up a docker container to run source airbyte connector.
+ * Platform is set to 'linux/amd64' as we only publish airbyte connectors images for linux/amd64.
+ *
+ * Docker cli command:
+ *  docker run --name $src_container_name --init \
+ *    -v "$tempdir:/configs" \
+ *    $max_memory $max_cpus --log-opt max-size="$max_log_size" \
+ *    --env LOG_LEVEL="$log_level" \
+ *    $src_docker_options
+ *    --cidfile="$tempPrefix-src_cid" \
+ *    -a stdout -a stderr \
+ *    "$src_docker_image" \
+ *    read \
+ *    --config "/configs/$src_config_filename" \
+ *    --catalog "/configs/$src_catalog_filename" \
+ *    --state "/configs/$src_state_filename"
+ *
+ * @argument command - for testing purposes only
+ */
+export async function runSrcSync(tmpDir: string, config: FarosConfig): Promise<string> {
+  logger.info('Running source connector...');
+
+  if (!config.src?.image) {
+    throw new Error('Source image is missing.');
+  }
+
+  try {
+    const timestamp = Date.now();
+    const srcContainerName = `airbyte-local-src-${timestamp}`;
+    const cmd = [
+      'read',
+      '--config',
+      `/configs/${SRC_CONFIG_FILENAME}`,
+      '--catalog',
+      `/configs/${SRC_CATALOG_FILENAME}`,
+      '--state',
+      `/configs/${DEFAULT_STATE_FILE}`,
+    ];
+    const maxNanoCpus = config.src?.dockerOptions?.maxCpus;
+    const maxMemory = config.src?.dockerOptions?.maxMemory;
+    const createOptions: Docker.ContainerCreateOptions = {
+      // Default config: can be overridden by the docker options provided by users
+      name: srcContainerName,
+      Image: config.src.image,
+      ...config.src?.dockerOptions?.additionalOptions,
+
+      // Default options: cannot be overridden by users
+      Cmd: cmd,
+      AttachStdout: true,
+      AttachStderr: true,
+      platform: 'linux/amd64',
+      Env: [`LOG_LEVEL=${config.logLevel}`, ...(config.src?.dockerOptions?.additionalOptions?.Env || [])],
+      HostConfig: {
+        // Defautl host config: can be overridden by users
+        NanoCpus: maxNanoCpus, // 1e9 nano cpus = 1 cpu
+        Memory: maxMemory, // 1024 * 1024 bytes = 1MB
+        LogConfig: {
+          Type: 'json-file',
+          Config: {
+            'max-size': config.src?.dockerOptions?.maxLogSize ?? DEFAULT_MAX_LOG_SIZE,
+          },
+        },
+        ...config.src?.dockerOptions?.additionalOptions?.HostConfig,
+        // Default options: cannot be overridden by users
+        Binds: [`${tmpDir}:/configs`],
+        AutoRemove: true,
+        Init: true,
+      },
+    };
+
+    // Create the Docker container
+    const container = await _docker.createContainer(createOptions);
+
+    // Write the container ID to the cidfile
+    const cidfilePath = `tmp-${timestamp}-src_cid`;
+    writeFileSync(cidfilePath, container.id);
+
+    // create a writable stream to capture the stdout
+    let data = '';
+    const stdoutStream = new Writable({
+      write(chunk, _encoding, callback) {
+        data += chunk.toString();
+        callback();
+      },
+    });
+    // Attach the stderr to termincal stderr, and stdout to the output stream
+    const stream = await container.attach({stream: true, stdout: true, stderr: true});
+    container.modem.demuxStream(stream, stdoutStream, process.stderr);
+
+    // Start the container
+    await container.start();
+
+    // Wait for the container to finish
+    const res = await container.wait();
+    logger.debug(res);
+    logger.debug(data);
+
+    if (res.StatusCode === 0) {
+      logger.info('Source connector ran successfully.');
+    } else {
+      throw new Error('Failed to run source connector.');
+    }
+
+    // return the stdout data
+    return data;
+  } catch (error: any) {
+    throw new Error(`Failed to run source connector: ${error.message ?? JSON.stringify(error)}`);
+  }
+}

airbyte-local-cli-nodejs/src/docker.ts

@@ -1,5 +1,5 @@
 import {parseAndValidateInputs} from './command';
-import {checkDockerInstalled, checkSrcConnection, pullDockerImage} from './docker';
+import {checkDockerInstalled, checkSrcConnection, pullDockerImage, runSrcSync} from './docker';
 import {AirbyteCliContext} from './types';
 import {cleanUp, createTmpDir, loadStateFile, logger, writeConfig} from './utils';
 
@@ -23,6 +23,10 @@ async function main(): Promise<void> {
     if (cfg.srcCheckConnection && cfg.src?.image) {
       await checkSrcConnection(context.tmpDir, cfg.src.image);
     }
+    // Run airbyte source connector
+    if (!cfg.srcInputFile) {
+      await runSrcSync(context.tmpDir, cfg);
+    }
   } catch (error: any) {
     logger.error(error.message, 'Error');
     cleanUp(context);

airbyte-local-cli-nodejs/src/index.ts

@@ -40,7 +40,13 @@ export interface AirbyteConfig {
   image: string;
   config?: object;
   catalog?: object;
-  dockerOptions?: string;
+  dockerOptions?: AirbyteConfigDockerOptions;
+}
+export interface AirbyteConfigDockerOptions {
+  maxMemory?: number; // unit: MB
+  maxCpus?: number; // unit: CPU
+  maxLogSize?: string; // default: 10m (10MB)
+  additionalOptions?: any;
 }
 export enum AirbyteConfigInputType {
   FILE = 'file',

airbyte-local-cli-nodejs/src/types.ts

@@ -14,6 +14,7 @@ export const SRC_CONFIG_FILENAME = `${FILENAME_PREFIX}_src_config.json`;
 export const DST_CONFIG_FILENAME = `${FILENAME_PREFIX}_dst_config.json`;
 export const SRC_CATALOG_FILENAME = `${FILENAME_PREFIX}_src_catalog.json`;
 export const DST_CATALOG_FILENAME = `${FILENAME_PREFIX}_dst_catalog.json`;
+export const DEFAULT_STATE_FILE = 'state.json';
 
 // Create a pino logger instance
 export const logger = pino(pretty({colorize: true}));
@@ -82,16 +83,15 @@ export function createTmpDir(absTmpDir?: string): string {
 
 // Load the existing state file and write to the temporary folder
 export function loadStateFile(tempDir: string, filePath?: string, connectionName?: string): void {
-  const path = filePath ?? (connectionName ? `${connectionName}__state.json` : 'state.json');
-  const name = 'state.json';
+  const path = filePath ?? (connectionName ? `${connectionName}__state.json` : DEFAULT_STATE_FILE);
 
   // Read the state file and write to temp folder
   // Write an empty state file if the state file hasn't existed yet
   try {
     accessSync(path, constants.R_OK);
     const stateData = readFileSync(path, 'utf8');
-    logger.debug(`Writing state file to temporary directory: '${tempDir}/${name}'...`);
-    writeFileSync(`${tempDir}/${name}`, stateData);
+    logger.debug(`Writing state file to temporary directory: '${tempDir}/${DEFAULT_STATE_FILE}'...`);
+    writeFileSync(`${tempDir}/${DEFAULT_STATE_FILE}`, stateData);
   } catch (error: any) {
     if (error.code !== 'ENOENT') {
       throw new Error(`Failed to read state file '${path}' : ${error.message}`);
@@ -100,8 +100,8 @@ export function loadStateFile(tempDir: string, filePath?: string, connectionName
         `State file '${filePath}' not found. Please make sure the state file exists and have read access.`,
       );
     }
-    writeFileSync(`${tempDir}/${name}`, '{}');
-    logger.debug(`State file '${name}' not found. An empty state file is created.`);
+    writeFileSync(`${tempDir}/${DEFAULT_STATE_FILE}`, '{}');
+    logger.debug(`State file '${DEFAULT_STATE_FILE}' not found. An empty state file is created.`);
   }
 }
 

airbyte-local-cli-nodejs/src/utils.ts

@@ -12,7 +12,10 @@ exports[`parseConfigFile should pass 1`] = `
         "graph": "default",
       },
     },
-    "dockerOptions": "--memory 2048m --cpus 2",
+    "dockerOptions": {
+      "maxCpus": 2,
+      "maxMemory": 2048,
+    },
     "image": "farosai/airbyte-servicenow-destination",
   },
   "src": {
@@ -22,7 +25,10 @@ exports[`parseConfigFile should pass 1`] = `
       "url": "https://test-instance.service-now.com",
       "username": "test-username",
     },
-    "dockerOptions": "--memory 2048m --cpus 2",
+    "dockerOptions": {
+      "maxCpus": 2,
+      "maxMemory": 2048,
+    },
     "image": "farosai/airbyte-servicenow-source",
   },
 }

airbyte-local-cli-nodejs/test/__snapshots__/utils.it.test.ts.snap

@@ -1,7 +1,29 @@
-import {checkSrcConnection, pullDockerImage} from '../src/docker';
+import {readdirSync, unlinkSync} from 'node:fs';
+import path from 'node:path';
+import {Writable} from 'node:stream';
+
+import {checkSrcConnection, pullDockerImage, runSrcSync} from '../src/docker';
+import {FarosConfig} from '../src/types';
+
+const defaultConfig: FarosConfig = {
+  srcCheckConnection: false,
+  dstUseHostNetwork: false,
+  srcPull: false,
+  dstPull: false,
+  fullRefresh: false,
+  rawMessages: false,
+  keepContainers: false,
+  logLevel: 'info',
+  debug: false,
+  stateFile: undefined,
+  connectionName: undefined,
+  srcOutputFile: undefined,
+  srcInputFile: undefined,
+};
 
 beforeAll(async () => {
   await pullDockerImage('farosai/airbyte-example-source');
+  await pullDockerImage('farosai/airbyte-faros-graphql-source');
 });
 
 describe('checkSrcConnection', () => {
@@ -25,3 +47,55 @@ describe('checkSrcConnection', () => {
     ).rejects.toThrow('Failed to validate source connection: User is not chris.');
   });
 });
+
+describe('runSrcSync', () => {
+  const testCfg: FarosConfig = {
+    ...defaultConfig,
+    src: {
+      image: 'farosai/airbyte-example-source',
+    },
+  };
+
+  // Clean up files created by the test
+  afterAll(() => {
+    const pattern = /.*-src_cid$/;
+    const files = readdirSync(process.cwd());
+    const matchingFiles = files.filter((file) => pattern.test(file));
+
+    matchingFiles.forEach((file) => {
+      const filePath = path.join(process.cwd(), file);
+      unlinkSync(filePath);
+    });
+  });
+
+  it('should success', async () => {
+    await expect(runSrcSync(`${process.cwd()}/test/resources/dockerIt_runSrcSync`, testCfg)).resolves.not.toThrow();
+  });
+
+  // Check the error message is correctly redirect to process.stderr
+  it('should fail', async () => {
+    // Capture process.stderr
+    let stderrData = '';
+    const originalStderrWrite = process.stderr.write;
+    const stderrStream = new Writable({
+      write(chunk, _encoding, callback) {
+        stderrData += chunk.toString();
+        callback();
+      },
+    });
+    process.stderr.write = stderrStream.write.bind(stderrStream) as any;
+
+    try {
+      await expect(
+        runSrcSync(`${process.cwd()}/test/resources/dockerIt_runSrcSync`, {
+          ...testCfg,
+          src: {image: 'farosai/airbyte-faros-graphql-source'},
+        }),
+      ).rejects.toThrow();
+    } finally {
+      process.stderr.write = originalStderrWrite;
+    }
+
+    expect(stderrData).toContain(`Faros API key was not provided`);
+  });
+});

airbyte-local-cli-nodejs/test/docker.it.test.ts

@@ -109,3 +109,21 @@ Describe 'Check source connection'
     The status should equal 1
   End
 End
+
+Describe 'Run source sync'
+  It 'should fail if source sync fails auth'
+    airbyte_local_test() {
+      ./airbyte-local \
+        --src 'farosai/airbyte-faros-graphql-source' \
+        --src-only
+    }
+    When call airbyte_local_test
+    The stderr should include "Faros API key was not provided"
+    The output should include "Failed to run source connector: Failed to run source connector."
+    The status should equal 1
+  End
+End
+
+# Clean up temeporary test files
+find . -name 'faros_airbyte_cli_config.json' -delete
+find . -name '*-src_cid' -delete

airbyte-local-cli-nodejs/test/exec/faros_airbyte_cli_config.json

@@ -0,0 +1 @@
+{}

airbyte-local-cli-nodejs/test/exec/spec/index_spec.sh

@@ -0,0 +1,3 @@
+{
+  "user": "chris"
+}

airbyte-local-cli-nodejs/test/resources/dockerIt_runSrcSync/faros_airbyte_cli_src_catalog.json

@@ -0,0 +1 @@
+{}