-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[FAI-13686] Spin up docker in cli and check source connection (#99)
- Loading branch information
Showing
17 changed files
with
311 additions
and
37 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -13,3 +13,4 @@ out/pkg | |
| sample_command.sh | ||
| *airbyte-local | ||
| test/exec/resources | ||
| tmp* | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1 +1 @@ | ||
| 20.17 | ||
| 20.18 |
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| import {Writable} from 'node:stream'; | ||
|
|
||
| import Docker from 'dockerode'; | ||
|
|
||
| import {AirbyteConnectionStatus, AirbyteConnectionStatusMessage, AirbyteMessageType} from './types'; | ||
| import {logger, SRC_CONFIG_FILENAME} from './utils'; | ||
|
|
||
| // Create a new Docker instance | ||
| let _docker = new Docker(); | ||
|
|
||
| // For testing purposes | ||
| export function setDocker(docker: Docker): void { | ||
| _docker = docker; | ||
| } | ||
|
|
||
| export async function checkDockerInstalled(): Promise<void> { | ||
| try { | ||
| await _docker.version(); | ||
| logger.debug('Docker is installed and running.'); | ||
| } catch (error: any) { | ||
| logger.error('Docker is not installed or running.'); | ||
| throw error; | ||
| } | ||
| } | ||
|
|
||
| export async function pullDockerImage(image: string): Promise<void> { | ||
| logger.info(`Pulling docker image: ${image}`); | ||
|
|
||
| try { | ||
| const stream = await _docker.pull(image); | ||
| await new Promise((resolve, reject) => { | ||
| _docker.modem.followProgress(stream, (err, res) => (err ? reject(err) : resolve(res))); | ||
| }); | ||
| logger.info(`Docker image pulled: ${image}`); | ||
| } catch (error: any) { | ||
| logger.error(`Failed to pull docker image: ${image}`); | ||
| throw error; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Spinning up a docker container to check the source connection. | ||
| * `docker run --rm -v "$tempdir:/configs" $src_docker_options "$src_docker_image" | ||
| * check --config "/configs/$src_config_filename"` | ||
| * | ||
| * Sample output from the docker container: | ||
| * {"connectionStatus":{"status":"SUCCEEDED"},"type":"CONNECTION_STATUS"} | ||
| * {"connectionStatus":{"status":"FAILED","message":"Faros API key was not provided"},"type":"CONNECTION_STATUS"} | ||
| */ | ||
| export async function checkSrcConnection(tmpDir: string, image: string, srcConfigFile?: string): Promise<void> { | ||
| logger.info('Validating connection to source...'); | ||
|
|
||
| if (!image) { | ||
| throw new Error('Source image is missing.'); | ||
| } | ||
|
|
||
| try { | ||
| const cfgFile = srcConfigFile ?? SRC_CONFIG_FILENAME; | ||
| const command = ['check', '--config', `/configs/${cfgFile}`]; | ||
| const createOptions: Docker.ContainerCreateOptions = { | ||
| HostConfig: { | ||
| Binds: [`${tmpDir}:/configs`], | ||
| AutoRemove: true, | ||
| }, | ||
| platform: 'linux/amd64', | ||
| }; | ||
|
|
||
| // create a writable stream to capture the output | ||
| let data = ''; | ||
| const outputStream = new Writable({ | ||
| write(chunk, _encoding, callback) { | ||
| data += chunk.toString(); | ||
| callback(); | ||
| }, | ||
| }); | ||
|
|
||
| // docker run | ||
| const res = await _docker.run(image, command, outputStream, createOptions); | ||
|
|
||
| // capture connection status from the output | ||
| let status: AirbyteConnectionStatusMessage | undefined; | ||
| data.split('\n').forEach((line) => { | ||
| if (line.includes(AirbyteMessageType.CONNECTION_STATUS)) { | ||
| status = JSON.parse(line) as AirbyteConnectionStatusMessage; | ||
| } | ||
| }); | ||
| if ( | ||
| status?.type === AirbyteMessageType.CONNECTION_STATUS && | ||
| status?.connectionStatus.status === AirbyteConnectionStatus.SUCCEEDED && | ||
| res[0].StatusCode === 0 | ||
| ) { | ||
| logger.info('Source connection is valid.'); | ||
| } else { | ||
| throw new Error(status?.connectionStatus.message); | ||
| } | ||
| } catch (error: any) { | ||
| throw new Error(`Failed to validate source connection: ${error.message ?? JSON.stringify(error)}.`); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,21 +1,36 @@ | ||
| import {parseAndValidateInputs} from './command'; | ||
| import {checkDockerInstalled, checkSrcConnection, pullDockerImage} from './docker'; | ||
| import {AirbyteCliContext} from './types'; | ||
| import {checkDockerInstalled, cleanUp, createTmpDir, loadStateFile, logger, writeConfig} from './utils'; | ||
| import {cleanUp, createTmpDir, loadStateFile, logger, writeConfig} from './utils'; | ||
|
|
||
| function main(): void { | ||
| async function main(): Promise<void> { | ||
| const context: AirbyteCliContext = {}; | ||
| try { | ||
| // Parse and validate cli arguments | ||
| const cfg = parseAndValidateInputs(process.argv); | ||
| checkDockerInstalled(); | ||
| await checkDockerInstalled(); | ||
|
|
||
| // Create temporary directory, load state file, write config to files | ||
| context.tmpDir = createTmpDir(); | ||
| loadStateFile(context.tmpDir, cfg?.stateFile, cfg?.connectionName); | ||
| writeConfig(context.tmpDir, cfg); | ||
|
|
||
| // Pull source docker image | ||
| if (cfg.srcPull && cfg.src?.image) { | ||
| await pullDockerImage(cfg.src.image); | ||
| } | ||
| // Check source connection | ||
| if (cfg.srcCheckConnection && cfg.src?.image) { | ||
| await checkSrcConnection(context.tmpDir, cfg.src.image); | ||
| } | ||
| } catch (error: any) { | ||
| logger.error(error.message, 'Error'); | ||
| cleanUp(context); | ||
| logger.error('Exit Airbyte CLI with errors.'); | ||
| process.exit(1); | ||
| throw error; | ||
| } | ||
| } | ||
|
|
||
| main(); | ||
| main().catch((_error) => { | ||
| logger.error('Exit Airbyte CLI with errors.'); | ||
| process.exit(1); | ||
| }); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,27 @@ | ||
| import {checkSrcConnection, pullDockerImage} from '../src/docker'; | ||
|
|
||
| beforeAll(async () => { | ||
| await pullDockerImage('farosai/airbyte-example-source'); | ||
| }); | ||
|
|
||
| describe('checkSrcConnection', () => { | ||
| it('should success', async () => { | ||
| await expect( | ||
| checkSrcConnection( | ||
| `${process.cwd()}/test/resources`, | ||
| 'farosai/airbyte-example-source', | ||
| 'faros_airbyte_cli_src_config_chris.json', | ||
| ), | ||
| ).resolves.not.toThrow(); | ||
| }); | ||
|
|
||
| it('should fail with', async () => { | ||
| await expect( | ||
| checkSrcConnection( | ||
| `${process.cwd()}/test/resources`, | ||
| 'farosai/airbyte-example-source', | ||
| 'faros_airbyte_cli_src_config_jennie.json', | ||
| ), | ||
| ).rejects.toThrow('Failed to validate source connection: User is not chris.'); | ||
| }); | ||
| }); |
Oops, something went wrong.