kpeters-cbsi
commented
4 years ago Outside of a Lambda context, it seems to upload the file successfully but it doesn't look as though the stream is ever being closed. This happens while SFTPing to both my local machine and the Salesforce Marketing server. Logs attached. Test script below.
The one thing that sticks out to me is that Salesforce Marketing looks to be running OpenSSH 7.9.0, but the machines that I tested against were on 7.6.2.
import SSH2Promise from 'ssh2-promise'
import { Stream, Writable } from 'stream'
import util from 'util'
import Path from 'path'
import SFTP from 'ssh2-promise/dist/sftp'
import fs from 'fs'
const pipeline = util.promisify(Stream.pipeline)
const logger = (...messages: any[]) => {
const timestamp = new Date().toISOString()
console.log(`${timestamp} `, ...messages)
}
// FIXME: Try without streams
const main = () => {
const file = process.argv[2]
const name = process.argv[3]
if (!file) {
process.stderr.write(`Usage: ${process.argv[1]} <file> [<remote name>]\n`)
process.exit(-1)
}
try {
doUpload(file, name)
} catch (e) {
if (e.code === 'ENOENT') {
process.stdout.write(`No such file or directory: ${file}`)
process.exit(-1)
} else {
logger(e, e.stack)
throw e
}
}
}
/**
* Upload an object from S3 to the configured SFTP host
*
* @param s3event S3EventRecord corresponding to an object uploaded to the bucket
*/
const doUpload = async (file: string, remoteName: string) => {
if (!remoteName) {
remoteName = file
}
logger(`Writing ${file} as ${remoteName}`)
const readStream = getReadStream(file)
logger('Got read stream')
logger(`Requesting write stream for ${file}`)
let writeStream: Writable
try {
writeStream = await getSftpWriteStream(remoteName)
logger('Got write stream')
} catch (e) {
logger(`Create write stream failed: ${e}`)
throw e
}
try {
logger('Pipeline read stream to write stream')
await pipeline(readStream, writeStream)
} catch (e) {
const message = `Failed to write ${file}: ${e}`
throw new Error(message)
}
logger(`${file} successfully written to SFTP as ${remoteName}`)
return file
}
/**
* Get a read stream of the object contents from the supplied S3 event record
*
* @param file S3 event record containing bucket and key name of the object
*/
const getReadStream = (file: string) => {
logger(`Request S3 read stream for ${file}`)
const rs = fs.createReadStream(file)
return rs
}
/**
* Get a write stream for the SFTP host
*
* @param key Path that data will be written to
*/
const getSftpWriteStream = async (key: string): Promise<Writable> => {
const sshConfig = {
host: 'ftp.s7.exacttarget.com',
username: 'REDACTED',
password: 'REDACTED',
port: 22,
readyTimeout: 30000,
debug: (message: any) => {
if (typeof message == 'object') {
logger(`SSH: `, { message })
} else {
logger(`SSH: ${message}`)
}
},
}
const ssh = new SSH2Promise(sshConfig)
console.debug('Requesting connection')
try {
await ssh.connect()
console.info(`Connected to ${sshConfig.host}`)
} catch (e) {
console.error(`Connection to ${sshConfig.host} failed: ${e}`)
throw e
}
logger(
`Connecting via SSH to ${sshConfig.host}:${sshConfig.port} as user ${sshConfig.username}`
)
logger(`Requesting write stream for path ${key} on ${sshConfig.host}`)
const dirname = Path.dirname(key)
const sftp = ssh.sftp()
if (dirname) {
await mkRemoteDir(sftp, dirname)
}
console.debug(`Create write stream for ${key}`)
return await sftp.createWriteStream(key)
}
/**
* Recursively create a remote directory if no such directory exists
*
* @param sftp SFTP connection
* @param path Directory to create
*/
const mkRemoteDir = async (sftp: SFTP, path: string): Promise<void> => {
logger(`dir: ${path}`)
// https://stackoverflow.com/a/60922162/132319
await path.split('/').reduce(async (promise, dir) => {
return promise.then(async (parent) => {
const ret = Path.join(parent, dir)
try {
await sftp.stat(ret)
} catch (e) {
if (e.code === 2) {
// path not found
try {
logger(`Attempt mkdir "${ret}"`)
await sftp.mkdir(ret)
logger(`Successfully created "${ret}"`)
} catch (e) {
logger(`Mkdir ${ret} failed: ${e}`)
throw e
}
} else {
throw e
}
}
return ret
})
}, Promise.resolve(''))
}
main()
jearles
I piggybacked onto #51 , but the solution to that problem didn't work for me. I'll reproduce the issue here.
I have a Lambda function that uses this module that has been timing out. Looking at the debug output, I see that it is connecting to the remote host but after
DEBUG: Parser: IN_PACKETDATAAFTER, packet: KEXDH_GEX_GROUPit seems to wait 30 seconds, disconnect, and then 5 seconds later try to connect again. Complete log is attached.log-events-viewer-result-2.zip