[FAI-14746] Run dst sync #106
Conversation
| const lastState = states.pop(); | ||
| if (lastState) { | ||
| writeFileSync(`${config.stateFile}`, lastState); | ||
| logger.debug(`New state is udpated in '${config.stateFile}'.`); |
There was a problem hiding this comment.
get the last line of the state and write it to the state file
| // Create a readable stream from the src output file and pipe it to the container stdin | ||
| const inputStream = createReadStream(`${tmpDir}/${SRC_OUTPUT_DATA_FILE}`); | ||
| const stdinStream = await container.attach({stream: true, hijack: true, stdin: true}); | ||
| inputStream.pipe(stdinStream); |
There was a problem hiding this comment.
somehow i wasn't able to use the same stream for stdout and stdin. one would work and the other would break.
I have to separate them into two streams.
| AttachStderr: true, | ||
| AttachStdin: true, | ||
| OpenStdin: true, | ||
| StdinOnce: true, |
There was a problem hiding this comment.
the configuration here is followed by inspecting the existing dst connector container settings.
| @@ -24,7 +24,7 @@ async function main(): Promise<void> { | |||
| // Create temporary directory, load state file, write config and catalog to files | |||
| context.tmpDir = createTmpDir(); | |||
| generateDstStreamPrefix(cfg); | |||
| loadStateFile(context.tmpDir, cfg?.stateFile, cfg?.connectionName); | |||
| cfg.stateFile = loadStateFile(context.tmpDir, cfg?.stateFile, cfg?.connectionName); | |||
There was a problem hiding this comment.
return the state file location.
note: this is not the state file in tmp folder.
if the user provides one already, this function will return the same value.
i might refactor this later - separate the internal config usage from the user input config. in place update it's a big confusing. feel like it would be harder to maintain
| process.stdout.write(`${line}\n`); | ||
| } else { | ||
| logger.info(formatDstMsg(data)); | ||
| } |
There was a problem hiding this comment.
output non-state data to stdout.
| @@ -0,0 +1,51 @@ | |||
| {"state":{"data":{"format":"base64/gzip","data":"H4sIAAAAAAAAA6uuBQBDv6ajAgAAAA=="}},"type":"STATE","redactedConfig":{"api_key":"REDACTED","models_filter":["org_Team"],"graphql_api":"v2","graph":"faros","result_model":"Flat","api_url":"https://dev.api.faros.ai"},"sourceType":"faros-graphql","sourceVersion":"0.12.5","record":{"stream":"myfarosgraphqlsrc__faros_graphql__"}} | |||
There was a problem hiding this comment.
i assume this doesn't have sensitive data?
| "graph": "jennie-test", | ||
| "graphql_api": "v2", | ||
| "api_url": "https://dev.api.faros.ai", | ||
| "api_key": "test" |
There was a problem hiding this comment.
this is a template that will be copied during it test. the api_key will be replaced in some tests
| }); | ||
|
|
||
| try { | ||
| await pipelineAsync(rl, transform, outputStream); |
There was a problem hiding this comment.
I think closing output stream when readline closes actually causing some race condition -> the output isn't able to fully written out.
update to use pipeline and transform. this should avoid the race condition. confirmed it works by adding sleep 5s in function processSrcDataByLine to make the writing out much slower than reading.
|
Description
Not included (will cover in later PRs)
Type of change