mirror of
https://github.com/yu-i-i/overleaf-cep.git
synced 2026-11-12 13:13:45 +01:00
Migrate to new object-persistor module (#122)
* Migrate to new object-persistor module * Support updated persistor module using ES6 classes * Update object-persistor * Upgrade new persistor code to OError3 * Update to latest minor release for outdated packages * Update package hashes to sha512 * Point object-persistor at master branch
This commit is contained in:
Simon Detheridge
@@ -1,12 +1,9 @@
|
||||
const OError = require('@overleaf/o-error')
|
||||
const { Errors } = require('@overleaf/object-persistor')
|
||||
|
||||
class NotFoundError extends OError {}
|
||||
class WriteError extends OError {}
|
||||
class ReadError extends OError {}
|
||||
class HealthCheckError extends OError {}
|
||||
class ConversionsDisabledError extends OError {}
|
||||
class ConversionError extends OError {}
|
||||
class SettingsError extends OError {}
|
||||
class TimeoutError extends OError {}
|
||||
class InvalidParametersError extends OError {}
|
||||
|
||||
@@ -23,14 +20,11 @@ class FailedCommandError extends OError {
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
NotFoundError,
|
||||
FailedCommandError,
|
||||
ConversionsDisabledError,
|
||||
WriteError,
|
||||
ReadError,
|
||||
ConversionError,
|
||||
HealthCheckError,
|
||||
SettingsError,
|
||||
TimeoutError,
|
||||
InvalidParametersError
|
||||
InvalidParametersError,
|
||||
...Errors
|
||||
}
|
||||
|
||||
@@ -1,247 +0,0 @@
|
||||
const fs = require('fs')
|
||||
const glob = require('glob')
|
||||
const path = require('path')
|
||||
const Stream = require('stream')
|
||||
const { promisify, callbackify } = require('util')
|
||||
|
||||
const LocalFileWriter = require('./LocalFileWriter').promises
|
||||
const { NotFoundError, ReadError, WriteError } = require('./Errors')
|
||||
const PersistorHelper = require('./PersistorHelper')
|
||||
|
||||
const pipeline = promisify(Stream.pipeline)
|
||||
const fsUnlink = promisify(fs.unlink)
|
||||
const fsOpen = promisify(fs.open)
|
||||
const fsStat = promisify(fs.stat)
|
||||
const fsGlob = promisify(glob)
|
||||
|
||||
const filterName = key => key.replace(/\//g, '_')
|
||||
|
||||
async function sendFile(location, target, source) {
|
||||
const filteredTarget = filterName(target)
|
||||
|
||||
// actually copy the file (instead of moving it) to maintain consistent behaviour
|
||||
// between the different implementations
|
||||
try {
|
||||
const sourceStream = fs.createReadStream(source)
|
||||
const targetStream = fs.createWriteStream(`${location}/${filteredTarget}`)
|
||||
await pipeline(sourceStream, targetStream)
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to copy the specified file',
|
||||
{ location, target, source },
|
||||
WriteError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function sendStream(location, target, sourceStream, sourceMd5) {
|
||||
const fsPath = await LocalFileWriter.writeStream(sourceStream)
|
||||
if (!sourceMd5) {
|
||||
sourceMd5 = await _getFileMd5HashForPath(fsPath)
|
||||
}
|
||||
|
||||
try {
|
||||
await sendFile(location, target, fsPath)
|
||||
const destMd5 = await getFileMd5Hash(location, target)
|
||||
if (sourceMd5 !== destMd5) {
|
||||
await LocalFileWriter.deleteFile(`${location}/${filterName(target)}`)
|
||||
throw new WriteError('md5 hash mismatch', {
|
||||
sourceMd5,
|
||||
destMd5,
|
||||
location,
|
||||
target
|
||||
})
|
||||
}
|
||||
} finally {
|
||||
await LocalFileWriter.deleteFile(fsPath)
|
||||
}
|
||||
}
|
||||
|
||||
// opts may be {start: Number, end: Number}
|
||||
async function getFileStream(location, name, opts) {
|
||||
const filteredName = filterName(name)
|
||||
|
||||
try {
|
||||
opts.fd = await fsOpen(`${location}/${filteredName}`, 'r')
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to open file for streaming',
|
||||
{ location, filteredName, opts },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
|
||||
return fs.createReadStream(null, opts)
|
||||
}
|
||||
|
||||
async function getRedirectUrl() {
|
||||
// not implemented
|
||||
return null
|
||||
}
|
||||
|
||||
async function getFileSize(location, filename) {
|
||||
const fullPath = path.join(location, filterName(filename))
|
||||
|
||||
try {
|
||||
const stat = await fsStat(fullPath)
|
||||
return stat.size
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to stat file',
|
||||
{ location, filename },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function getFileMd5Hash(location, filename) {
|
||||
const fullPath = path.join(location, filterName(filename))
|
||||
try {
|
||||
return await _getFileMd5HashForPath(fullPath)
|
||||
} catch (err) {
|
||||
throw new ReadError('unable to get md5 hash from file', {
|
||||
location,
|
||||
filename
|
||||
}).withCause(err)
|
||||
}
|
||||
}
|
||||
|
||||
async function copyFile(location, fromName, toName) {
|
||||
const filteredFromName = filterName(fromName)
|
||||
const filteredToName = filterName(toName)
|
||||
|
||||
try {
|
||||
const sourceStream = fs.createReadStream(`${location}/${filteredFromName}`)
|
||||
const targetStream = fs.createWriteStream(`${location}/${filteredToName}`)
|
||||
await pipeline(sourceStream, targetStream)
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to copy file',
|
||||
{ location, filteredFromName, filteredToName },
|
||||
WriteError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function deleteFile(location, name) {
|
||||
const filteredName = filterName(name)
|
||||
try {
|
||||
await fsUnlink(`${location}/${filteredName}`)
|
||||
} catch (err) {
|
||||
const wrappedError = PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to delete file',
|
||||
{ location, filteredName },
|
||||
WriteError
|
||||
)
|
||||
if (!(wrappedError instanceof NotFoundError)) {
|
||||
// S3 doesn't give us a 404 when a file wasn't there to be deleted, so we
|
||||
// should be consistent here as well
|
||||
throw wrappedError
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// this is only called internally for clean-up by `FileHandler` and isn't part of the external API
|
||||
async function deleteDirectory(location, name) {
|
||||
const filteredName = filterName(name.replace(/\/$/, ''))
|
||||
|
||||
try {
|
||||
await Promise.all(
|
||||
(await fsGlob(`${location}/${filteredName}*`)).map(file => fsUnlink(file))
|
||||
)
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to delete directory',
|
||||
{ location, filteredName },
|
||||
WriteError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function checkIfFileExists(location, name) {
|
||||
const filteredName = filterName(name)
|
||||
try {
|
||||
const stat = await fsStat(`${location}/${filteredName}`)
|
||||
return !!stat
|
||||
} catch (err) {
|
||||
if (err.code === 'ENOENT') {
|
||||
return false
|
||||
}
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to stat file',
|
||||
{ location, filteredName },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// note, does not recurse into subdirectories, as we use a flattened directory structure
|
||||
async function directorySize(location, name) {
|
||||
const filteredName = filterName(name.replace(/\/$/, ''))
|
||||
let size = 0
|
||||
|
||||
try {
|
||||
const files = await fsGlob(`${location}/${filteredName}_*`)
|
||||
for (const file of files) {
|
||||
try {
|
||||
const stat = await fsStat(file)
|
||||
if (stat.isFile()) {
|
||||
size += stat.size
|
||||
}
|
||||
} catch (err) {
|
||||
// ignore files that may have just been deleted
|
||||
if (err.code !== 'ENOENT') {
|
||||
throw err
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to get directory size',
|
||||
{ location, name },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
|
||||
return size
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
sendFile: callbackify(sendFile),
|
||||
sendStream: callbackify(sendStream),
|
||||
getFileStream: callbackify(getFileStream),
|
||||
getRedirectUrl: callbackify(getRedirectUrl),
|
||||
getFileSize: callbackify(getFileSize),
|
||||
getFileMd5Hash: callbackify(getFileMd5Hash),
|
||||
copyFile: callbackify(copyFile),
|
||||
deleteFile: callbackify(deleteFile),
|
||||
deleteDirectory: callbackify(deleteDirectory),
|
||||
checkIfFileExists: callbackify(checkIfFileExists),
|
||||
directorySize: callbackify(directorySize),
|
||||
promises: {
|
||||
sendFile,
|
||||
sendStream,
|
||||
getFileStream,
|
||||
getRedirectUrl,
|
||||
getFileSize,
|
||||
getFileMd5Hash,
|
||||
copyFile,
|
||||
deleteFile,
|
||||
deleteDirectory,
|
||||
checkIfFileExists,
|
||||
directorySize
|
||||
}
|
||||
}
|
||||
|
||||
async function _getFileMd5HashForPath(fullPath) {
|
||||
const stream = fs.createReadStream(fullPath)
|
||||
return PersistorHelper.calculateStreamMd5(stream)
|
||||
}
|
||||
@@ -75,12 +75,11 @@ function getFile(req, res, next) {
|
||||
res.end()
|
||||
} else if (err) {
|
||||
next(
|
||||
new Errors.ReadError('error transferring stream', {
|
||||
bucket,
|
||||
key,
|
||||
format,
|
||||
style
|
||||
}).withCause(err)
|
||||
new Errors.ReadError(
|
||||
'error transferring stream',
|
||||
{ bucket, key, format, style },
|
||||
err
|
||||
)
|
||||
)
|
||||
}
|
||||
})
|
||||
@@ -139,23 +138,17 @@ function copyFile(req, res, next) {
|
||||
})
|
||||
req.requestLogger.setMessage('copying file')
|
||||
|
||||
PersistorManager.copyFile(
|
||||
bucket,
|
||||
`${oldProjectId}/${oldFileId}`,
|
||||
key,
|
||||
function(err) {
|
||||
PersistorManager.copyObject(bucket, `${oldProjectId}/${oldFileId}`, key)
|
||||
.then(() => res.sendStatus(200))
|
||||
.catch(err => {
|
||||
if (err) {
|
||||
if (err instanceof Errors.NotFoundError) {
|
||||
res.sendStatus(404)
|
||||
} else {
|
||||
next(err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
res.sendStatus(200)
|
||||
}
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
function deleteFile(req, res, next) {
|
||||
|
||||
@@ -86,12 +86,11 @@ async function _convert(sourcePath, requestedFormat, command) {
|
||||
timeout: FOURTY_SECONDS
|
||||
})
|
||||
} catch (err) {
|
||||
throw new ConversionError('something went wrong converting file', {
|
||||
stderr: err.stderr,
|
||||
sourcePath,
|
||||
requestedFormat,
|
||||
destPath
|
||||
}).withCause(err)
|
||||
throw new ConversionError(
|
||||
'something went wrong converting file',
|
||||
{ stderr: err.stderr, sourcePath, requestedFormat, destPath },
|
||||
err
|
||||
)
|
||||
}
|
||||
|
||||
timer.done()
|
||||
|
||||
@@ -37,9 +37,9 @@ async function insertFile(bucket, key, stream) {
|
||||
})
|
||||
}
|
||||
if (Settings.enableConversions) {
|
||||
await PersistorManager.promises.deleteDirectory(bucket, convertedKey)
|
||||
await PersistorManager.deleteDirectory(bucket, convertedKey)
|
||||
}
|
||||
await PersistorManager.promises.sendStream(bucket, key, stream)
|
||||
await PersistorManager.sendStream(bucket, key, stream)
|
||||
}
|
||||
|
||||
async function deleteFile(bucket, key) {
|
||||
@@ -51,9 +51,9 @@ async function deleteFile(bucket, key) {
|
||||
convertedKey
|
||||
})
|
||||
}
|
||||
const jobs = [PersistorManager.promises.deleteFile(bucket, key)]
|
||||
const jobs = [PersistorManager.deleteObject(bucket, key)]
|
||||
if (Settings.enableConversions) {
|
||||
jobs.push(PersistorManager.promises.deleteDirectory(bucket, convertedKey))
|
||||
jobs.push(PersistorManager.deleteDirectory(bucket, convertedKey))
|
||||
}
|
||||
await Promise.all(jobs)
|
||||
}
|
||||
@@ -65,13 +65,13 @@ async function deleteProject(bucket, key) {
|
||||
key
|
||||
})
|
||||
}
|
||||
await PersistorManager.promises.deleteDirectory(bucket, key)
|
||||
await PersistorManager.deleteDirectory(bucket, key)
|
||||
}
|
||||
|
||||
async function getFile(bucket, key, opts) {
|
||||
opts = opts || {}
|
||||
if (!opts.format && !opts.style) {
|
||||
return PersistorManager.promises.getFileStream(bucket, key, opts)
|
||||
return PersistorManager.getObjectStream(bucket, key, opts)
|
||||
} else {
|
||||
return _getConvertedFile(bucket, key, opts)
|
||||
}
|
||||
@@ -89,28 +89,28 @@ async function getRedirectUrl(bucket, key, opts) {
|
||||
Object.values(Settings.filestore.stores).includes(bucket) &&
|
||||
Settings.filestore.allowRedirects
|
||||
) {
|
||||
return PersistorManager.promises.getRedirectUrl(bucket, key)
|
||||
return PersistorManager.getRedirectUrl(bucket, key)
|
||||
}
|
||||
|
||||
return null
|
||||
}
|
||||
|
||||
async function getFileSize(bucket, key) {
|
||||
return PersistorManager.promises.getFileSize(bucket, key)
|
||||
return PersistorManager.getObjectSize(bucket, key)
|
||||
}
|
||||
|
||||
async function getDirectorySize(bucket, projectId) {
|
||||
return PersistorManager.promises.directorySize(bucket, projectId)
|
||||
return PersistorManager.directorySize(bucket, projectId)
|
||||
}
|
||||
|
||||
async function _getConvertedFile(bucket, key, opts) {
|
||||
const convertedKey = KeyBuilder.addCachingToKey(key, opts)
|
||||
const exists = await PersistorManager.promises.checkIfFileExists(
|
||||
const exists = await PersistorManager.checkIfObjectExists(
|
||||
bucket,
|
||||
convertedKey
|
||||
)
|
||||
if (exists) {
|
||||
return PersistorManager.promises.getFileStream(bucket, convertedKey, opts)
|
||||
return PersistorManager.getObjectStream(bucket, convertedKey, opts)
|
||||
} else {
|
||||
return _getConvertedFileAndCache(bucket, key, convertedKey, opts)
|
||||
}
|
||||
@@ -121,19 +121,14 @@ async function _getConvertedFileAndCache(bucket, key, convertedKey, opts) {
|
||||
try {
|
||||
convertedFsPath = await _convertFile(bucket, key, opts)
|
||||
await ImageOptimiser.promises.compressPng(convertedFsPath)
|
||||
await PersistorManager.promises.sendFile(
|
||||
bucket,
|
||||
convertedKey,
|
||||
convertedFsPath
|
||||
)
|
||||
await PersistorManager.sendFile(bucket, convertedKey, convertedFsPath)
|
||||
} catch (err) {
|
||||
LocalFileWriter.deleteFile(convertedFsPath, () => {})
|
||||
throw new ConversionError('failed to convert file', {
|
||||
opts,
|
||||
bucket,
|
||||
key,
|
||||
convertedKey
|
||||
}).withCause(err)
|
||||
throw new ConversionError(
|
||||
'failed to convert file',
|
||||
{ opts, bucket, key, convertedKey },
|
||||
err
|
||||
)
|
||||
}
|
||||
// Send back the converted file from the local copy to avoid problems
|
||||
// with the file not being present in S3 yet. As described in the
|
||||
@@ -159,11 +154,11 @@ async function _convertFile(bucket, originalKey, opts) {
|
||||
try {
|
||||
originalFsPath = await _writeFileToDisk(bucket, originalKey, opts)
|
||||
} catch (err) {
|
||||
throw new ConversionError('unable to write file to disk', {
|
||||
bucket,
|
||||
originalKey,
|
||||
opts
|
||||
}).withCause(err)
|
||||
throw new ConversionError(
|
||||
'unable to write file to disk',
|
||||
{ bucket, originalKey, opts },
|
||||
err
|
||||
)
|
||||
}
|
||||
|
||||
let promise
|
||||
@@ -184,21 +179,17 @@ async function _convertFile(bucket, originalKey, opts) {
|
||||
try {
|
||||
destPath = await promise
|
||||
} catch (err) {
|
||||
throw new ConversionError('error converting file', {
|
||||
bucket,
|
||||
originalKey,
|
||||
opts
|
||||
}).withCause(err)
|
||||
throw new ConversionError(
|
||||
'error converting file',
|
||||
{ bucket, originalKey, opts },
|
||||
err
|
||||
)
|
||||
}
|
||||
LocalFileWriter.deleteFile(originalFsPath, function() {})
|
||||
return destPath
|
||||
}
|
||||
|
||||
async function _writeFileToDisk(bucket, key, opts) {
|
||||
const fileStream = await PersistorManager.promises.getFileStream(
|
||||
bucket,
|
||||
key,
|
||||
opts
|
||||
)
|
||||
const fileStream = await PersistorManager.getObjectStream(bucket, key, opts)
|
||||
return LocalFileWriter.promises.writeStream(fileStream, key)
|
||||
}
|
||||
|
||||
@@ -1,305 +0,0 @@
|
||||
const settings = require('settings-sharelatex')
|
||||
const fs = require('fs')
|
||||
const { promisify } = require('util')
|
||||
const Stream = require('stream')
|
||||
const { Storage } = require('@google-cloud/storage')
|
||||
const { callbackify } = require('util')
|
||||
const { WriteError, ReadError, NotFoundError } = require('./Errors')
|
||||
const asyncPool = require('tiny-async-pool')
|
||||
const PersistorHelper = require('./PersistorHelper')
|
||||
|
||||
const pipeline = promisify(Stream.pipeline)
|
||||
|
||||
// endpoint settings will be null by default except for tests
|
||||
// that's OK - GCS uses the locally-configured service account by default
|
||||
const storage = new Storage(settings.filestore.gcs.endpoint)
|
||||
// workaround for broken uploads with custom endpoints:
|
||||
// https://github.com/googleapis/nodejs-storage/issues/898
|
||||
if (
|
||||
settings.filestore.gcs.endpoint &&
|
||||
settings.filestore.gcs.endpoint.apiEndpoint
|
||||
) {
|
||||
storage.interceptors.push({
|
||||
request: function(reqOpts) {
|
||||
const url = new URL(reqOpts.uri)
|
||||
url.host = settings.filestore.gcs.endpoint.apiEndpoint
|
||||
if (settings.filestore.gcs.endpoint.apiScheme) {
|
||||
url.protocol = settings.filestore.gcs.endpoint.apiScheme
|
||||
}
|
||||
reqOpts.uri = url.toString()
|
||||
return reqOpts
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
const GcsPersistor = {
|
||||
sendFile: callbackify(sendFile),
|
||||
sendStream: callbackify(sendStream),
|
||||
getFileStream: callbackify(getFileStream),
|
||||
getRedirectUrl: callbackify(getRedirectUrl),
|
||||
getFileMd5Hash: callbackify(getFileMd5Hash),
|
||||
deleteDirectory: callbackify(deleteDirectory),
|
||||
getFileSize: callbackify(getFileSize),
|
||||
deleteFile: callbackify(deleteFile),
|
||||
copyFile: callbackify(copyFile),
|
||||
checkIfFileExists: callbackify(checkIfFileExists),
|
||||
directorySize: callbackify(directorySize),
|
||||
promises: {
|
||||
sendFile,
|
||||
sendStream,
|
||||
getFileStream,
|
||||
getRedirectUrl,
|
||||
getFileMd5Hash,
|
||||
deleteDirectory,
|
||||
getFileSize,
|
||||
deleteFile,
|
||||
copyFile,
|
||||
checkIfFileExists,
|
||||
directorySize
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = GcsPersistor
|
||||
|
||||
async function sendFile(bucketName, key, fsPath) {
|
||||
return sendStream(bucketName, key, fs.createReadStream(fsPath))
|
||||
}
|
||||
|
||||
async function sendStream(bucketName, key, readStream, sourceMd5) {
|
||||
try {
|
||||
// egress from us to gcs
|
||||
const observeOptions = { metric: 'gcs.egress' }
|
||||
|
||||
if (!sourceMd5) {
|
||||
// if there is no supplied md5 hash, we calculate the hash as the data passes through
|
||||
observeOptions.hash = 'md5'
|
||||
}
|
||||
|
||||
const observer = new PersistorHelper.ObserverStream(observeOptions)
|
||||
|
||||
const writeOptions = {
|
||||
// disabling of resumable uploads is recommended by Google:
|
||||
resumable: false
|
||||
}
|
||||
|
||||
if (sourceMd5) {
|
||||
writeOptions.validation = 'md5'
|
||||
writeOptions.metadata = {
|
||||
md5Hash: PersistorHelper.hexToBase64(sourceMd5)
|
||||
}
|
||||
}
|
||||
|
||||
const uploadStream = storage
|
||||
.bucket(bucketName)
|
||||
.file(key)
|
||||
.createWriteStream(writeOptions)
|
||||
|
||||
await pipeline(readStream, observer, uploadStream)
|
||||
|
||||
// if we didn't have an md5 hash, we should compare our computed one with Google's
|
||||
// as we couldn't tell GCS about it beforehand
|
||||
if (!sourceMd5) {
|
||||
sourceMd5 = observer.getHash()
|
||||
// throws on mismatch
|
||||
await PersistorHelper.verifyMd5(GcsPersistor, bucketName, key, sourceMd5)
|
||||
}
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'upload to GCS failed',
|
||||
{ bucketName, key },
|
||||
WriteError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function getFileStream(bucketName, key, _opts = {}) {
|
||||
const opts = Object.assign({}, _opts)
|
||||
const stream = storage
|
||||
.bucket(bucketName)
|
||||
.file(key)
|
||||
.createReadStream(opts)
|
||||
|
||||
// ingress to us from gcs
|
||||
const observer = new PersistorHelper.ObserverStream({
|
||||
metric: 'gcs.ingress'
|
||||
})
|
||||
|
||||
try {
|
||||
// wait for the pipeline to be ready, to catch non-200s
|
||||
await PersistorHelper.getReadyPipeline(stream, observer)
|
||||
return observer
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'error reading file from GCS',
|
||||
{ bucketName, key, opts },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function getRedirectUrl(bucketName, key) {
|
||||
try {
|
||||
const [url] = await storage
|
||||
.bucket(bucketName)
|
||||
.file(key)
|
||||
.getSignedUrl({
|
||||
action: 'read',
|
||||
expires: new Date().getTime() + settings.filestore.signedUrlExpiryInMs
|
||||
})
|
||||
return url
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'error generating signed url for GCS file',
|
||||
{ bucketName, key },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function getFileSize(bucketName, key) {
|
||||
try {
|
||||
const [metadata] = await storage
|
||||
.bucket(bucketName)
|
||||
.file(key)
|
||||
.getMetadata()
|
||||
return metadata.size
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'error getting size of GCS object',
|
||||
{ bucketName, key },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function getFileMd5Hash(bucketName, key) {
|
||||
try {
|
||||
const [metadata] = await storage
|
||||
.bucket(bucketName)
|
||||
.file(key)
|
||||
.getMetadata()
|
||||
return PersistorHelper.base64ToHex(metadata.md5Hash)
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'error getting hash of GCS object',
|
||||
{ bucketName, key },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function deleteFile(bucketName, key) {
|
||||
try {
|
||||
const file = storage.bucket(bucketName).file(key)
|
||||
|
||||
if (settings.filestore.gcs.deletedBucketSuffix) {
|
||||
await file.copy(
|
||||
storage
|
||||
.bucket(`${bucketName}${settings.filestore.gcs.deletedBucketSuffix}`)
|
||||
.file(`${key}-${new Date().toISOString()}`)
|
||||
)
|
||||
}
|
||||
if (settings.filestore.gcs.unlockBeforeDelete) {
|
||||
await file.setMetadata({ eventBasedHold: false })
|
||||
}
|
||||
await file.delete()
|
||||
} catch (err) {
|
||||
const error = PersistorHelper.wrapError(
|
||||
err,
|
||||
'error deleting GCS object',
|
||||
{ bucketName, key },
|
||||
WriteError
|
||||
)
|
||||
if (!(error instanceof NotFoundError)) {
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function deleteDirectory(bucketName, key) {
|
||||
try {
|
||||
const [files] = await storage
|
||||
.bucket(bucketName)
|
||||
.getFiles({ directory: key })
|
||||
|
||||
await asyncPool(
|
||||
settings.filestore.gcs.deleteConcurrency,
|
||||
files,
|
||||
async file => {
|
||||
await deleteFile(bucketName, file.name)
|
||||
}
|
||||
)
|
||||
} catch (err) {
|
||||
const error = PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to delete directory in GCS',
|
||||
{ bucketName, key },
|
||||
WriteError
|
||||
)
|
||||
if (error instanceof NotFoundError) {
|
||||
return
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
async function directorySize(bucketName, key) {
|
||||
let files
|
||||
|
||||
try {
|
||||
const [response] = await storage
|
||||
.bucket(bucketName)
|
||||
.getFiles({ directory: key })
|
||||
files = response
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to list objects in GCS',
|
||||
{ bucketName, key },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
|
||||
return files.reduce((acc, file) => Number(file.metadata.size) + acc, 0)
|
||||
}
|
||||
|
||||
async function checkIfFileExists(bucketName, key) {
|
||||
try {
|
||||
const [response] = await storage
|
||||
.bucket(bucketName)
|
||||
.file(key)
|
||||
.exists()
|
||||
return response
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'error checking if file exists in GCS',
|
||||
{ bucketName, key },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function copyFile(bucketName, sourceKey, destKey) {
|
||||
try {
|
||||
const src = storage.bucket(bucketName).file(sourceKey)
|
||||
const dest = storage.bucket(bucketName).file(destKey)
|
||||
await src.copy(dest)
|
||||
} catch (err) {
|
||||
// fake-gcs-server has a bug that returns an invalid response when the file does not exist
|
||||
if (err.message === 'Cannot parse response as JSON: not found\n') {
|
||||
err.code = 404
|
||||
}
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to copy file in GCS',
|
||||
{ bucketName, sourceKey, destKey },
|
||||
WriteError
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -31,7 +31,7 @@ async function checkCanGetFiles() {
|
||||
try {
|
||||
await pipeline(sourceStream, buffer)
|
||||
} catch (err) {
|
||||
throw new HealthCheckError('failed to get health-check file').withCause(err)
|
||||
throw new HealthCheckError('failed to get health-check file', {}, err)
|
||||
}
|
||||
|
||||
if (!buffer.size()) {
|
||||
|
||||
@@ -30,9 +30,7 @@ async function writeStream(stream, key) {
|
||||
} catch (err) {
|
||||
await deleteFile(fsPath)
|
||||
|
||||
throw new WriteError('problem writing file locally', {
|
||||
fsPath
|
||||
}).withCause(err)
|
||||
throw new WriteError('problem writing file locally', { fsPath }, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -44,7 +42,7 @@ async function deleteFile(fsPath) {
|
||||
await promisify(fs.unlink)(fsPath)
|
||||
} catch (err) {
|
||||
if (err.code !== 'ENOENT') {
|
||||
throw new WriteError('failed to delete file', { fsPath }).withCause(err)
|
||||
throw new WriteError('failed to delete file', { fsPath }, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,230 +0,0 @@
|
||||
const metrics = require('metrics-sharelatex')
|
||||
const Settings = require('settings-sharelatex')
|
||||
const logger = require('logger-sharelatex')
|
||||
const Stream = require('stream')
|
||||
const { callbackify, promisify } = require('util')
|
||||
const { NotFoundError, WriteError } = require('./Errors')
|
||||
|
||||
const pipeline = promisify(Stream.pipeline)
|
||||
|
||||
// Persistor that wraps two other persistors. Talks to the 'primary' by default,
|
||||
// but will fall back to an older persistor in the case of a not-found error.
|
||||
// If `Settings.filestore.fallback.copyOnMiss` is set, this will copy files from the fallback
|
||||
// to the primary, in the event that they are missing.
|
||||
//
|
||||
// It is unlikely that the bucket/location name will be the same on the fallback
|
||||
// as the primary. The bucket names should be overridden in `Settings.filestore.fallback.buckets`
|
||||
// e.g.
|
||||
// Settings.filestore.fallback.buckets = {
|
||||
// myBucketOnS3: 'myBucketOnGCS'
|
||||
// }
|
||||
|
||||
module.exports = function(primary, fallback) {
|
||||
function _wrapMethodOnBothPersistors(method) {
|
||||
return async function(bucket, key, ...moreArgs) {
|
||||
const fallbackBucket = _getFallbackBucket(bucket)
|
||||
|
||||
await Promise.all([
|
||||
primary.promises[method](bucket, key, ...moreArgs),
|
||||
fallback.promises[method](fallbackBucket, key, ...moreArgs)
|
||||
])
|
||||
}
|
||||
}
|
||||
|
||||
async function getFileStreamWithFallback(bucket, key, opts) {
|
||||
const shouldCopy =
|
||||
Settings.filestore.fallback.copyOnMiss && !opts.start && !opts.end
|
||||
|
||||
try {
|
||||
return await primary.promises.getFileStream(bucket, key, opts)
|
||||
} catch (err) {
|
||||
if (err instanceof NotFoundError) {
|
||||
const fallbackBucket = _getFallbackBucket(bucket)
|
||||
const fallbackStream = await fallback.promises.getFileStream(
|
||||
fallbackBucket,
|
||||
key,
|
||||
opts
|
||||
)
|
||||
// tee the stream to the client, and as a copy to the primary (if necessary)
|
||||
// start listening on both straight away so that we don't consume bytes
|
||||
// in one place before the other
|
||||
const returnStream = new Stream.PassThrough()
|
||||
pipeline(fallbackStream, returnStream)
|
||||
|
||||
if (shouldCopy) {
|
||||
const copyStream = new Stream.PassThrough()
|
||||
pipeline(fallbackStream, copyStream)
|
||||
|
||||
_copyStreamFromFallbackAndVerify(
|
||||
copyStream,
|
||||
fallbackBucket,
|
||||
bucket,
|
||||
key,
|
||||
key
|
||||
).catch(() => {
|
||||
// swallow errors, as this runs in the background and will log a warning
|
||||
})
|
||||
}
|
||||
return returnStream
|
||||
}
|
||||
throw err
|
||||
}
|
||||
}
|
||||
|
||||
async function copyFileWithFallback(bucket, sourceKey, destKey) {
|
||||
try {
|
||||
return await primary.promises.copyFile(bucket, sourceKey, destKey)
|
||||
} catch (err) {
|
||||
if (err instanceof NotFoundError) {
|
||||
const fallbackBucket = _getFallbackBucket(bucket)
|
||||
const fallbackStream = await fallback.promises.getFileStream(
|
||||
fallbackBucket,
|
||||
sourceKey,
|
||||
{}
|
||||
)
|
||||
|
||||
const copyStream = new Stream.PassThrough()
|
||||
pipeline(fallbackStream, copyStream)
|
||||
|
||||
if (Settings.filestore.fallback.copyOnMiss) {
|
||||
const missStream = new Stream.PassThrough()
|
||||
pipeline(fallbackStream, missStream)
|
||||
|
||||
// copy from sourceKey -> sourceKey
|
||||
_copyStreamFromFallbackAndVerify(
|
||||
missStream,
|
||||
fallbackBucket,
|
||||
bucket,
|
||||
sourceKey,
|
||||
sourceKey
|
||||
).then(() => {
|
||||
// swallow errors, as this runs in the background and will log a warning
|
||||
})
|
||||
}
|
||||
// copy from sourceKey -> destKey
|
||||
return _copyStreamFromFallbackAndVerify(
|
||||
copyStream,
|
||||
fallbackBucket,
|
||||
bucket,
|
||||
sourceKey,
|
||||
destKey
|
||||
)
|
||||
}
|
||||
throw err
|
||||
}
|
||||
}
|
||||
|
||||
function _getFallbackBucket(bucket) {
|
||||
return Settings.filestore.fallback.buckets[bucket] || bucket
|
||||
}
|
||||
|
||||
function _wrapFallbackMethod(method) {
|
||||
return async function(bucket, key, ...moreArgs) {
|
||||
try {
|
||||
return await primary.promises[method](bucket, key, ...moreArgs)
|
||||
} catch (err) {
|
||||
if (err instanceof NotFoundError) {
|
||||
const fallbackBucket = _getFallbackBucket(bucket)
|
||||
if (Settings.filestore.fallback.copyOnMiss) {
|
||||
const fallbackStream = await fallback.promises.getFileStream(
|
||||
fallbackBucket,
|
||||
key,
|
||||
{}
|
||||
)
|
||||
// run in background
|
||||
_copyStreamFromFallbackAndVerify(
|
||||
fallbackStream,
|
||||
fallbackBucket,
|
||||
bucket,
|
||||
key,
|
||||
key
|
||||
).catch(err => {
|
||||
logger.warn({ err }, 'failed to copy file from fallback')
|
||||
})
|
||||
}
|
||||
return fallback.promises[method](fallbackBucket, key, ...moreArgs)
|
||||
}
|
||||
throw err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function _copyStreamFromFallbackAndVerify(
|
||||
stream,
|
||||
sourceBucket,
|
||||
destBucket,
|
||||
sourceKey,
|
||||
destKey
|
||||
) {
|
||||
try {
|
||||
let sourceMd5
|
||||
try {
|
||||
sourceMd5 = await fallback.promises.getFileMd5Hash(
|
||||
sourceBucket,
|
||||
sourceKey
|
||||
)
|
||||
} catch (err) {
|
||||
logger.warn(err, 'error getting md5 hash from fallback persistor')
|
||||
}
|
||||
|
||||
await primary.promises.sendStream(destBucket, destKey, stream, sourceMd5)
|
||||
} catch (err) {
|
||||
const error = new WriteError(
|
||||
'unable to copy file to destination persistor',
|
||||
{
|
||||
sourceBucket,
|
||||
destBucket,
|
||||
sourceKey,
|
||||
destKey
|
||||
}
|
||||
).withCause(err)
|
||||
metrics.inc('fallback.copy.failure')
|
||||
|
||||
try {
|
||||
await primary.promises.deleteFile(destBucket, destKey)
|
||||
} catch (err) {
|
||||
error.info.cleanupError = new WriteError(
|
||||
'unable to clean up destination copy artifact',
|
||||
{
|
||||
destBucket,
|
||||
destKey
|
||||
}
|
||||
).withCause(err)
|
||||
}
|
||||
|
||||
logger.warn({ error }, 'failed to copy file from fallback')
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
primaryPersistor: primary,
|
||||
fallbackPersistor: fallback,
|
||||
sendFile: primary.sendFile,
|
||||
sendStream: primary.sendStream,
|
||||
getFileStream: callbackify(getFileStreamWithFallback),
|
||||
getRedirectUrl: primary.getRedirectUrl,
|
||||
getFileMd5Hash: callbackify(_wrapFallbackMethod('getFileMd5Hash')),
|
||||
deleteDirectory: callbackify(
|
||||
_wrapMethodOnBothPersistors('deleteDirectory')
|
||||
),
|
||||
getFileSize: callbackify(_wrapFallbackMethod('getFileSize')),
|
||||
deleteFile: callbackify(_wrapMethodOnBothPersistors('deleteFile')),
|
||||
copyFile: callbackify(copyFileWithFallback),
|
||||
checkIfFileExists: callbackify(_wrapFallbackMethod('checkIfFileExists')),
|
||||
directorySize: callbackify(_wrapFallbackMethod('directorySize')),
|
||||
promises: {
|
||||
sendFile: primary.promises.sendFile,
|
||||
sendStream: primary.promises.sendStream,
|
||||
getFileStream: getFileStreamWithFallback,
|
||||
getRedirectUrl: primary.promises.getRedirectUrl,
|
||||
getFileMd5Hash: _wrapFallbackMethod('getFileMd5Hash'),
|
||||
deleteDirectory: _wrapMethodOnBothPersistors('deleteDirectory'),
|
||||
getFileSize: _wrapFallbackMethod('getFileSize'),
|
||||
deleteFile: _wrapMethodOnBothPersistors('deleteFile'),
|
||||
copyFile: copyFileWithFallback,
|
||||
checkIfFileExists: _wrapFallbackMethod('checkIfFileExists'),
|
||||
directorySize: _wrapFallbackMethod('directorySize')
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,177 +0,0 @@
|
||||
const crypto = require('crypto')
|
||||
const Stream = require('stream')
|
||||
const logger = require('logger-sharelatex')
|
||||
const metrics = require('metrics-sharelatex')
|
||||
const { WriteError, ReadError, NotFoundError } = require('./Errors')
|
||||
const { promisify } = require('util')
|
||||
|
||||
const pipeline = promisify(Stream.pipeline)
|
||||
|
||||
// Observes data that passes through and computes some metadata for it
|
||||
// - specifically, it computes the number of bytes transferred, and optionally
|
||||
// computes a cryptographic hash based on the 'hash' option. e.g., pass
|
||||
// { hash: 'md5' } to compute the md5 hash of the stream
|
||||
// - if 'metric' is supplied as an option, this metric will be incremented by
|
||||
// the number of bytes transferred
|
||||
class ObserverStream extends Stream.Transform {
|
||||
constructor(options) {
|
||||
options.autoDestroy = true
|
||||
super(options)
|
||||
|
||||
this.bytes = 0
|
||||
|
||||
if (options.hash) {
|
||||
this.hash = crypto.createHash(options.hash)
|
||||
}
|
||||
|
||||
if (options.metric) {
|
||||
const onEnd = () => {
|
||||
metrics.count(options.metric, this.bytes)
|
||||
}
|
||||
this.once('error', onEnd)
|
||||
this.once('end', onEnd)
|
||||
}
|
||||
}
|
||||
|
||||
_transform(chunk, encoding, done) {
|
||||
if (this.hash) {
|
||||
this.hash.update(chunk)
|
||||
}
|
||||
this.bytes += chunk.length
|
||||
this.push(chunk)
|
||||
done()
|
||||
}
|
||||
|
||||
getHash() {
|
||||
return this.hash && this.hash.digest('hex')
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
ObserverStream,
|
||||
calculateStreamMd5,
|
||||
verifyMd5,
|
||||
getReadyPipeline,
|
||||
wrapError,
|
||||
hexToBase64,
|
||||
base64ToHex
|
||||
}
|
||||
|
||||
// returns a promise which resolves with the md5 hash of the stream
|
||||
// - consumes the stream
|
||||
function calculateStreamMd5(stream) {
|
||||
const hash = crypto.createHash('md5')
|
||||
hash.setEncoding('hex')
|
||||
|
||||
return pipeline(stream, hash).then(() => hash.read())
|
||||
}
|
||||
|
||||
// verifies the md5 hash of a file against the supplied md5 or the one stored in
|
||||
// storage if not supplied - deletes the new file if the md5 does not match and
|
||||
// throws an error
|
||||
async function verifyMd5(persistor, bucket, key, sourceMd5, destMd5 = null) {
|
||||
if (!destMd5) {
|
||||
destMd5 = await persistor.promises.getFileMd5Hash(bucket, key)
|
||||
}
|
||||
|
||||
if (sourceMd5 !== destMd5) {
|
||||
try {
|
||||
await persistor.promises.deleteFile(bucket, key)
|
||||
} catch (err) {
|
||||
logger.warn(err, 'error deleting file for invalid upload')
|
||||
}
|
||||
|
||||
throw new WriteError('source and destination hashes do not match', {
|
||||
sourceMd5,
|
||||
destMd5,
|
||||
bucket,
|
||||
key
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// resolves when a stream is 'readable', or rejects if the stream throws an error
|
||||
// before that happens - this lets us handle protocol-level errors before trying
|
||||
// to read them
|
||||
function getReadyPipeline(...streams) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const lastStream = streams.slice(-1)[0]
|
||||
|
||||
// in case of error or stream close, we must ensure that we drain the
|
||||
// previous stream so that it can clean up its socket (if it has one)
|
||||
const drainPreviousStream = function(previousStream) {
|
||||
// this stream is no longer reliable, so don't pipe anything more into it
|
||||
previousStream.unpipe(this)
|
||||
previousStream.resume()
|
||||
}
|
||||
|
||||
// handler to resolve when either:
|
||||
// - an error happens, or
|
||||
// - the last stream in the chain is readable
|
||||
// for example, in the case of a 4xx error an error will occur and the
|
||||
// streams will not become readable
|
||||
const handler = function(err) {
|
||||
// remove handler from all streams because we don't want to do this on
|
||||
// later errors
|
||||
lastStream.removeListener('readable', handler)
|
||||
for (const stream of streams) {
|
||||
stream.removeListener('error', handler)
|
||||
}
|
||||
|
||||
// return control to the caller
|
||||
if (err) {
|
||||
reject(
|
||||
wrapError(err, 'error before stream became ready', {}, ReadError)
|
||||
)
|
||||
} else {
|
||||
resolve(lastStream)
|
||||
}
|
||||
}
|
||||
|
||||
// ensure the handler fires when the last strem becomes readable
|
||||
lastStream.on('readable', handler)
|
||||
|
||||
for (const stream of streams) {
|
||||
// when a stream receives a pipe, set up the drain handler to drain the
|
||||
// connection if an error occurs or the stream is closed
|
||||
stream.on('pipe', previousStream => {
|
||||
stream.on('error', x => {
|
||||
drainPreviousStream(previousStream)
|
||||
})
|
||||
stream.on('close', () => {
|
||||
drainPreviousStream(previousStream)
|
||||
})
|
||||
})
|
||||
// add the handler function to resolve this method on error if we can't
|
||||
// set up the pipeline
|
||||
stream.on('error', handler)
|
||||
}
|
||||
|
||||
// begin the pipeline
|
||||
for (let index = 0; index < streams.length - 1; index++) {
|
||||
streams[index].pipe(streams[index + 1])
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
function wrapError(error, message, params, ErrorType) {
|
||||
if (
|
||||
error instanceof NotFoundError ||
|
||||
['NoSuchKey', 'NotFound', 404, 'AccessDenied', 'ENOENT'].includes(
|
||||
error.code
|
||||
) ||
|
||||
(error.response && error.response.statusCode === 404)
|
||||
) {
|
||||
return new NotFoundError('no such file', params).withCause(error)
|
||||
} else {
|
||||
return new ErrorType(message, params).withCause(error)
|
||||
}
|
||||
}
|
||||
|
||||
function base64ToHex(base64) {
|
||||
return Buffer.from(base64, 'base64').toString('hex')
|
||||
}
|
||||
|
||||
function hexToBase64(hex) {
|
||||
return Buffer.from(hex, 'hex').toString('base64')
|
||||
}
|
||||
@@ -1,39 +1,10 @@
|
||||
const settings = require('settings-sharelatex')
|
||||
const logger = require('logger-sharelatex')
|
||||
|
||||
logger.log(
|
||||
{
|
||||
backend: settings.filestore.backend,
|
||||
fallback: settings.filestore.fallback && settings.filestore.fallback.backend
|
||||
},
|
||||
'Loading backend'
|
||||
)
|
||||
if (!settings.filestore.backend) {
|
||||
throw new Error('no backend specified - config incomplete')
|
||||
}
|
||||
const persistorSettings = settings.filestore
|
||||
persistorSettings.Metrics = require('metrics-sharelatex')
|
||||
persistorSettings.paths = settings.path
|
||||
|
||||
function getPersistor(backend) {
|
||||
switch (backend) {
|
||||
case 'aws-sdk':
|
||||
case 's3':
|
||||
return require('./S3Persistor')
|
||||
case 'fs':
|
||||
return require('./FSPersistor')
|
||||
case 'gcs':
|
||||
return require('./GcsPersistor')
|
||||
default:
|
||||
throw new Error(`unknown filestore backend: ${backend}`)
|
||||
}
|
||||
}
|
||||
|
||||
let persistor = getPersistor(settings.filestore.backend)
|
||||
|
||||
if (settings.filestore.fallback && settings.filestore.fallback.backend) {
|
||||
const migrationPersistor = require('./MigrationPersistor')
|
||||
persistor = migrationPersistor(
|
||||
persistor,
|
||||
getPersistor(settings.filestore.fallback.backend)
|
||||
)
|
||||
}
|
||||
const ObjectPersistor = require('@overleaf/object-persistor')
|
||||
const persistor = ObjectPersistor(persistorSettings)
|
||||
|
||||
module.exports = persistor
|
||||
|
||||
@@ -1,367 +0,0 @@
|
||||
const http = require('http')
|
||||
const https = require('https')
|
||||
http.globalAgent.maxSockets = 300
|
||||
https.globalAgent.maxSockets = 300
|
||||
|
||||
const settings = require('settings-sharelatex')
|
||||
|
||||
const PersistorHelper = require('./PersistorHelper')
|
||||
|
||||
const fs = require('fs')
|
||||
const S3 = require('aws-sdk/clients/s3')
|
||||
const { URL } = require('url')
|
||||
const Stream = require('stream')
|
||||
const { promisify, callbackify } = require('util')
|
||||
const {
|
||||
WriteError,
|
||||
ReadError,
|
||||
NotFoundError,
|
||||
SettingsError
|
||||
} = require('./Errors')
|
||||
const pipeline = promisify(Stream.pipeline)
|
||||
|
||||
const S3Persistor = {
|
||||
sendFile: callbackify(sendFile),
|
||||
sendStream: callbackify(sendStream),
|
||||
getFileStream: callbackify(getFileStream),
|
||||
getRedirectUrl: callbackify(getRedirectUrl),
|
||||
getFileMd5Hash: callbackify(getFileMd5Hash),
|
||||
deleteDirectory: callbackify(deleteDirectory),
|
||||
getFileSize: callbackify(getFileSize),
|
||||
deleteFile: callbackify(deleteFile),
|
||||
copyFile: callbackify(copyFile),
|
||||
checkIfFileExists: callbackify(checkIfFileExists),
|
||||
directorySize: callbackify(directorySize),
|
||||
promises: {
|
||||
sendFile,
|
||||
sendStream,
|
||||
getFileStream,
|
||||
getRedirectUrl,
|
||||
getFileMd5Hash,
|
||||
deleteDirectory,
|
||||
getFileSize,
|
||||
deleteFile,
|
||||
copyFile,
|
||||
checkIfFileExists,
|
||||
directorySize
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = S3Persistor
|
||||
|
||||
async function sendFile(bucketName, key, fsPath) {
|
||||
return sendStream(bucketName, key, fs.createReadStream(fsPath))
|
||||
}
|
||||
|
||||
async function sendStream(bucketName, key, readStream, sourceMd5) {
|
||||
try {
|
||||
// egress from us to S3
|
||||
const observeOptions = { metric: 's3.egress' }
|
||||
let b64Hash
|
||||
|
||||
if (sourceMd5) {
|
||||
b64Hash = PersistorHelper.hexToBase64(sourceMd5)
|
||||
} else {
|
||||
// if there is no supplied md5 hash, we calculate the hash as the data passes through
|
||||
observeOptions.hash = 'md5'
|
||||
}
|
||||
|
||||
const observer = new PersistorHelper.ObserverStream(observeOptions)
|
||||
pipeline(readStream, observer)
|
||||
|
||||
// if we have an md5 hash, pass this to S3 to verify the upload
|
||||
const uploadOptions = {
|
||||
Bucket: bucketName,
|
||||
Key: key,
|
||||
Body: observer
|
||||
}
|
||||
if (b64Hash) {
|
||||
uploadOptions.ContentMD5 = b64Hash
|
||||
}
|
||||
|
||||
const response = await _getClientForBucket(bucketName)
|
||||
.upload(uploadOptions, { partSize: settings.filestore.s3.partSize })
|
||||
.promise()
|
||||
let destMd5 = _md5FromResponse(response)
|
||||
if (!destMd5) {
|
||||
// the eTag isn't in md5 format so we need to calculate it ourselves
|
||||
const verifyStream = await getFileStream(
|
||||
response.Bucket,
|
||||
response.Key,
|
||||
{}
|
||||
)
|
||||
destMd5 = await PersistorHelper.calculateStreamMd5(verifyStream)
|
||||
}
|
||||
|
||||
// if we didn't have an md5 hash, we should compare our computed one with S3's
|
||||
// as we couldn't tell S3 about it beforehand
|
||||
if (!sourceMd5) {
|
||||
sourceMd5 = observer.getHash()
|
||||
// throws on mismatch
|
||||
await PersistorHelper.verifyMd5(
|
||||
S3Persistor,
|
||||
bucketName,
|
||||
key,
|
||||
sourceMd5,
|
||||
destMd5
|
||||
)
|
||||
}
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'upload to S3 failed',
|
||||
{ bucketName, key },
|
||||
WriteError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function getFileStream(bucketName, key, opts) {
|
||||
opts = opts || {}
|
||||
|
||||
const params = {
|
||||
Bucket: bucketName,
|
||||
Key: key
|
||||
}
|
||||
if (opts.start != null && opts.end != null) {
|
||||
params.Range = `bytes=${opts.start}-${opts.end}`
|
||||
}
|
||||
|
||||
const stream = _getClientForBucket(bucketName)
|
||||
.getObject(params)
|
||||
.createReadStream()
|
||||
|
||||
// ingress from S3 to us
|
||||
const observer = new PersistorHelper.ObserverStream({ metric: 's3.ingress' })
|
||||
|
||||
try {
|
||||
// wait for the pipeline to be ready, to catch non-200s
|
||||
await PersistorHelper.getReadyPipeline(stream, observer)
|
||||
return observer
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'error reading file from S3',
|
||||
{ bucketName, key, opts },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function getRedirectUrl() {
|
||||
// not implemented
|
||||
return null
|
||||
}
|
||||
|
||||
async function deleteDirectory(bucketName, key) {
|
||||
let response
|
||||
|
||||
try {
|
||||
response = await _getClientForBucket(bucketName)
|
||||
.listObjects({ Bucket: bucketName, Prefix: key })
|
||||
.promise()
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to list objects in S3',
|
||||
{ bucketName, key },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
|
||||
const objects = response.Contents.map(item => ({ Key: item.Key }))
|
||||
if (objects.length) {
|
||||
try {
|
||||
await _getClientForBucket(bucketName)
|
||||
.deleteObjects({
|
||||
Bucket: bucketName,
|
||||
Delete: {
|
||||
Objects: objects,
|
||||
Quiet: true
|
||||
}
|
||||
})
|
||||
.promise()
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to delete objects in S3',
|
||||
{ bucketName, key },
|
||||
WriteError
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function getFileSize(bucketName, key) {
|
||||
try {
|
||||
const response = await _getClientForBucket(bucketName)
|
||||
.headObject({ Bucket: bucketName, Key: key })
|
||||
.promise()
|
||||
return response.ContentLength
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'error getting size of s3 object',
|
||||
{ bucketName, key },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function getFileMd5Hash(bucketName, key) {
|
||||
try {
|
||||
const response = await _getClientForBucket(bucketName)
|
||||
.headObject({ Bucket: bucketName, Key: key })
|
||||
.promise()
|
||||
return _md5FromResponse(response)
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'error getting hash of s3 object',
|
||||
{ bucketName, key },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function deleteFile(bucketName, key) {
|
||||
try {
|
||||
await _getClientForBucket(bucketName)
|
||||
.deleteObject({ Bucket: bucketName, Key: key })
|
||||
.promise()
|
||||
} catch (err) {
|
||||
// s3 does not give us a NotFoundError here
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to delete file in S3',
|
||||
{ bucketName, key },
|
||||
WriteError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function copyFile(bucketName, sourceKey, destKey) {
|
||||
const params = {
|
||||
Bucket: bucketName,
|
||||
Key: destKey,
|
||||
CopySource: `${bucketName}/${sourceKey}`
|
||||
}
|
||||
try {
|
||||
await _getClientForBucket(bucketName)
|
||||
.copyObject(params)
|
||||
.promise()
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to copy file in S3',
|
||||
params,
|
||||
WriteError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function checkIfFileExists(bucketName, key) {
|
||||
try {
|
||||
await getFileSize(bucketName, key)
|
||||
return true
|
||||
} catch (err) {
|
||||
if (err instanceof NotFoundError) {
|
||||
return false
|
||||
}
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'error checking whether S3 object exists',
|
||||
{ bucketName, key },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function directorySize(bucketName, key) {
|
||||
try {
|
||||
const response = await _getClientForBucket(bucketName)
|
||||
.listObjects({ Bucket: bucketName, Prefix: key })
|
||||
.promise()
|
||||
|
||||
return response.Contents.reduce((acc, item) => item.Size + acc, 0)
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'error getting directory size in S3',
|
||||
{ bucketName, key },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
const _clients = new Map()
|
||||
let _defaultClient
|
||||
|
||||
function _getClientForBucket(bucket) {
|
||||
if (_clients[bucket]) {
|
||||
return _clients[bucket]
|
||||
}
|
||||
|
||||
if (
|
||||
settings.filestore.s3BucketCreds &&
|
||||
settings.filestore.s3BucketCreds[bucket]
|
||||
) {
|
||||
_clients[bucket] = new S3(
|
||||
_buildClientOptions(settings.filestore.s3BucketCreds[bucket])
|
||||
)
|
||||
return _clients[bucket]
|
||||
}
|
||||
|
||||
// no specific credentials for the bucket
|
||||
if (_defaultClient) {
|
||||
return _defaultClient
|
||||
}
|
||||
|
||||
if (settings.filestore.s3.key) {
|
||||
_defaultClient = new S3(_buildClientOptions())
|
||||
return _defaultClient
|
||||
}
|
||||
|
||||
throw new SettingsError(
|
||||
'no bucket-specific or default credentials provided',
|
||||
{ bucket }
|
||||
)
|
||||
}
|
||||
|
||||
function _buildClientOptions(bucketCredentials) {
|
||||
const options = {}
|
||||
|
||||
if (bucketCredentials) {
|
||||
options.credentials = {
|
||||
accessKeyId: bucketCredentials.auth_key,
|
||||
secretAccessKey: bucketCredentials.auth_secret
|
||||
}
|
||||
} else {
|
||||
options.credentials = {
|
||||
accessKeyId: settings.filestore.s3.key,
|
||||
secretAccessKey: settings.filestore.s3.secret
|
||||
}
|
||||
}
|
||||
|
||||
if (settings.filestore.s3.endpoint) {
|
||||
const endpoint = new URL(settings.filestore.s3.endpoint)
|
||||
options.endpoint = settings.filestore.s3.endpoint
|
||||
options.sslEnabled = endpoint.protocol === 'https'
|
||||
}
|
||||
|
||||
// path-style access is only used for acceptance tests
|
||||
if (settings.filestore.s3.pathStyle) {
|
||||
options.s3ForcePathStyle = true
|
||||
}
|
||||
|
||||
return options
|
||||
}
|
||||
|
||||
function _md5FromResponse(response) {
|
||||
const md5 = (response.ETag || '').replace(/[ "]/g, '')
|
||||
if (!md5.match(/^[a-f0-9]{32}$/)) {
|
||||
return null
|
||||
}
|
||||
|
||||
return md5
|
||||
}
|
||||
Reference in New Issue
Block a user