mirror of
https://github.com/yu-i-i/overleaf-cep.git
synced 2026-11-13 21:53:43 +01:00
Create new module from overleaf/filestore persistors (#1)
* Create new module from overleaf/filestore persistors * Convert persistors to ES6 classes with local settings * Update README.md Co-authored-by: Eric Mc Sween <eric.mcsween@overleaf.com> * Update README.md Co-authored-by: Eric Mc Sween <eric.mcsween@overleaf.com> * Update .gitignore Co-authored-by: Eric Mc Sween <eric.mcsween@overleaf.com> * Switch to AGPL license * Paginate S3 list-object results * Remove S3 client caching * Clean up S3 md5-verification mechanism * Update README for recent changes * Update README.md Co-authored-by: Eric Mc Sween <eric.mcsween@overleaf.com> * Remove package-lock * Remove comment about FileHandler * Add directory marker to FSPersistor.deleteDirectory * Don't copy opts in GcsPersistor.getObjectStream * Use Date.now instead of getTime * Catch errors in migration persistor * Check that settings.buckets exists * Don't mutate options in ObserverStream constructor * Update src/PersistorHelper.js Co-authored-by: Eric Mc Sween <eric.mcsween@overleaf.com> * Lint and format fixes Co-authored-by: Eric Mc Sween <eric.mcsween@overleaf.com>
This commit is contained in:
Simon Detheridge
committed by
GitHub
GitHub
co-authored by
Eric Mc Sween
Eric Mc Sween
parent
b83ee27f83
commit
e92b75a2f8
@@ -0,0 +1,81 @@
|
||||
const { NotImplementedError } = require('./Errors')
|
||||
|
||||
module.exports = class AbstractPersistor {
|
||||
async sendFile(location, target, source) {
|
||||
throw new NotImplementedError({
|
||||
message: 'method not implemented in persistor',
|
||||
info: { method: 'sendFile', location, target, source }
|
||||
})
|
||||
}
|
||||
|
||||
async sendStream(location, target, sourceStream, sourceMd5) {
|
||||
throw new NotImplementedError({
|
||||
message: 'method not implemented in persistor',
|
||||
info: { method: 'sendStream', location, target, sourceMd5 }
|
||||
})
|
||||
}
|
||||
|
||||
// opts may be {start: Number, end: Number}
|
||||
async getObjectStream(location, name, opts) {
|
||||
throw new NotImplementedError({
|
||||
message: 'method not implemented in persistor',
|
||||
info: { method: 'getObjectStream', location, name, opts }
|
||||
})
|
||||
}
|
||||
|
||||
async getRedirectUrl(location, name) {
|
||||
throw new NotImplementedError({
|
||||
message: 'method not implemented in persistor',
|
||||
info: { method: 'getRedirectUrl', location, name }
|
||||
})
|
||||
}
|
||||
|
||||
async getObjectSize(location, name) {
|
||||
throw new NotImplementedError({
|
||||
message: 'method not implemented in persistor',
|
||||
info: { method: 'getObjectSize', location, name }
|
||||
})
|
||||
}
|
||||
|
||||
async getObjectMd5Hash(location, name) {
|
||||
throw new NotImplementedError({
|
||||
message: 'method not implemented in persistor',
|
||||
info: { method: 'getObjectMd5Hash', location, name }
|
||||
})
|
||||
}
|
||||
|
||||
async copyObject(location, fromName, toName) {
|
||||
throw new NotImplementedError({
|
||||
message: 'method not implemented in persistor',
|
||||
info: { method: 'copyObject', location, fromName, toName }
|
||||
})
|
||||
}
|
||||
|
||||
async deleteObject(location, name) {
|
||||
throw new NotImplementedError({
|
||||
message: 'method not implemented in persistor',
|
||||
info: { method: 'deleteObject', location, name }
|
||||
})
|
||||
}
|
||||
|
||||
async deleteDirectory(location, name) {
|
||||
throw new NotImplementedError({
|
||||
message: 'method not implemented in persistor',
|
||||
info: { method: 'deleteDirectory', location, name }
|
||||
})
|
||||
}
|
||||
|
||||
async checkIfObjectExists(location, name) {
|
||||
throw new NotImplementedError({
|
||||
message: 'method not implemented in persistor',
|
||||
info: { method: 'checkIfObjectExists', location, name }
|
||||
})
|
||||
}
|
||||
|
||||
async directorySize(location, name) {
|
||||
throw new NotImplementedError({
|
||||
message: 'method not implemented in persistor',
|
||||
info: { method: 'directorySize', location, name }
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
const OError = require('@overleaf/o-error')
|
||||
|
||||
class NotFoundError extends OError {}
|
||||
class WriteError extends OError {}
|
||||
class ReadError extends OError {}
|
||||
class SettingsError extends OError {}
|
||||
class NotImplementedError extends OError {}
|
||||
|
||||
module.exports = {
|
||||
NotFoundError,
|
||||
WriteError,
|
||||
ReadError,
|
||||
SettingsError,
|
||||
NotImplementedError
|
||||
}
|
||||
@@ -0,0 +1,278 @@
|
||||
const fs = require('fs')
|
||||
const glob = require('glob')
|
||||
const uuid = require('node-uuid')
|
||||
const path = require('path')
|
||||
const Stream = require('stream')
|
||||
const { promisify } = require('util')
|
||||
|
||||
const AbstractPersistor = require('./AbstractPersistor')
|
||||
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, '_')
|
||||
|
||||
module.exports = class FSPersistor extends AbstractPersistor {
|
||||
constructor(settings) {
|
||||
super()
|
||||
|
||||
this.settings = settings
|
||||
}
|
||||
|
||||
async 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 sendStream(location, target, sourceStream, sourceMd5) {
|
||||
const fsPath = await this._writeStream(sourceStream)
|
||||
if (!sourceMd5) {
|
||||
sourceMd5 = await FSPersistor._getFileMd5HashForPath(fsPath)
|
||||
}
|
||||
|
||||
try {
|
||||
await this.sendFile(location, target, fsPath)
|
||||
const destMd5 = await this.getObjectMd5Hash(location, target)
|
||||
if (sourceMd5 !== destMd5) {
|
||||
await this._deleteFile(`${location}/${filterName(target)}`)
|
||||
throw new WriteError({
|
||||
message: 'md5 hash mismatch',
|
||||
info: { sourceMd5, destMd5, location, target }
|
||||
})
|
||||
}
|
||||
} finally {
|
||||
await this._deleteFile(fsPath)
|
||||
}
|
||||
}
|
||||
|
||||
// opts may be {start: Number, end: Number}
|
||||
async getObjectStream(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 getRedirectUrl() {
|
||||
// not implemented
|
||||
return null
|
||||
}
|
||||
|
||||
async getObjectSize(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 getObjectMd5Hash(location, filename) {
|
||||
const fullPath = path.join(location, filterName(filename))
|
||||
try {
|
||||
return await FSPersistor._getFileMd5HashForPath(fullPath)
|
||||
} catch (err) {
|
||||
throw new ReadError({
|
||||
message: 'unable to get md5 hash from file',
|
||||
info: { location, filename }
|
||||
}).withCause(err)
|
||||
}
|
||||
}
|
||||
|
||||
async copyObject(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 deleteObject(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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async 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 checkIfObjectExists(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 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
|
||||
}
|
||||
|
||||
_getPath(key) {
|
||||
if (key == null) {
|
||||
key = uuid.v1()
|
||||
}
|
||||
key = key.replace(/\//g, '-')
|
||||
return path.join(this.settings.paths.uploadFolder, key)
|
||||
}
|
||||
|
||||
async _writeStream(stream, key) {
|
||||
let timer
|
||||
if (this.settings.Metrics) {
|
||||
timer = new this.settings.Metrics.Timer('writingFile')
|
||||
}
|
||||
const fsPath = this._getPath(key)
|
||||
|
||||
const writeStream = fs.createWriteStream(fsPath)
|
||||
try {
|
||||
await pipeline(stream, writeStream)
|
||||
if (timer) {
|
||||
timer.done()
|
||||
}
|
||||
return fsPath
|
||||
} catch (err) {
|
||||
await this._deleteFile(fsPath)
|
||||
|
||||
throw new WriteError({
|
||||
message: 'problem writing file locally',
|
||||
info: { err, fsPath }
|
||||
}).withCause(err)
|
||||
}
|
||||
}
|
||||
|
||||
async _deleteFile(fsPath) {
|
||||
if (!fsPath) {
|
||||
return
|
||||
}
|
||||
try {
|
||||
await fsUnlink(fsPath)
|
||||
} catch (err) {
|
||||
if (err.code !== 'ENOENT') {
|
||||
throw new WriteError({
|
||||
message: 'failed to delete file',
|
||||
info: { fsPath }
|
||||
}).withCause(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static async _getFileMd5HashForPath(fullPath) {
|
||||
const stream = fs.createReadStream(fullPath)
|
||||
return PersistorHelper.calculateStreamMd5(stream)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,279 @@
|
||||
const fs = require('fs')
|
||||
const { promisify } = require('util')
|
||||
const Stream = require('stream')
|
||||
const { Storage } = require('@google-cloud/storage')
|
||||
const { WriteError, ReadError, NotFoundError } = require('./Errors')
|
||||
const asyncPool = require('tiny-async-pool')
|
||||
const AbstractPersistor = require('./AbstractPersistor')
|
||||
const PersistorHelper = require('./PersistorHelper')
|
||||
|
||||
const pipeline = promisify(Stream.pipeline)
|
||||
|
||||
module.exports = class GcsPersistor extends AbstractPersistor {
|
||||
constructor(settings) {
|
||||
super()
|
||||
|
||||
this.settings = settings
|
||||
|
||||
// endpoint settings will be null by default except for tests
|
||||
// that's OK - GCS uses the locally-configured service account by default
|
||||
this.storage = new Storage(this.settings.endpoint)
|
||||
// workaround for broken uploads with custom endpoints:
|
||||
// https://github.com/googleapis/nodejs-storage/issues/898
|
||||
if (this.settings.endpoint && this.settings.endpoint.apiEndpoint) {
|
||||
this.storage.interceptors.push({
|
||||
request: (reqOpts) => {
|
||||
const url = new URL(reqOpts.uri)
|
||||
url.host = this.settings.endpoint.apiEndpoint
|
||||
if (this.settings.endpoint.apiScheme) {
|
||||
url.protocol = this.settings.endpoint.apiScheme
|
||||
}
|
||||
reqOpts.uri = url.toString()
|
||||
return reqOpts
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
async sendFile(bucketName, key, fsPath) {
|
||||
return this.sendStream(bucketName, key, fs.createReadStream(fsPath))
|
||||
}
|
||||
|
||||
async sendStream(bucketName, key, readStream, sourceMd5) {
|
||||
try {
|
||||
// egress from us to gcs
|
||||
const observeOptions = {
|
||||
metric: 'gcs.egress',
|
||||
Metrics: this.settings.Metrics
|
||||
}
|
||||
|
||||
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 = this.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(this, bucketName, key, sourceMd5)
|
||||
}
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'upload to GCS failed',
|
||||
{ bucketName, key },
|
||||
WriteError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async getObjectStream(bucketName, key, opts = {}) {
|
||||
const stream = this.storage
|
||||
.bucket(bucketName)
|
||||
.file(key)
|
||||
.createReadStream(opts)
|
||||
|
||||
// ingress to us from gcs
|
||||
const observer = new PersistorHelper.ObserverStream({
|
||||
metric: 'gcs.ingress',
|
||||
Metrics: this.settings.Metrics
|
||||
})
|
||||
|
||||
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 getRedirectUrl(bucketName, key) {
|
||||
try {
|
||||
const [url] = await this.storage
|
||||
.bucket(bucketName)
|
||||
.file(key)
|
||||
.getSignedUrl({
|
||||
action: 'read',
|
||||
expires: Date.now() + this.settings.signedUrlExpiryInMs
|
||||
})
|
||||
return url
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'error generating signed url for GCS file',
|
||||
{ bucketName, key },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async getObjectSize(bucketName, key) {
|
||||
try {
|
||||
const [metadata] = await this.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 getObjectMd5Hash(bucketName, key) {
|
||||
try {
|
||||
const [metadata] = await this.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 deleteObject(bucketName, key) {
|
||||
try {
|
||||
const file = this.storage.bucket(bucketName).file(key)
|
||||
|
||||
if (this.settings.deletedBucketSuffix) {
|
||||
await file.copy(
|
||||
this.storage
|
||||
.bucket(`${bucketName}${this.settings.deletedBucketSuffix}`)
|
||||
.file(`${key}-${new Date().toISOString()}`)
|
||||
)
|
||||
}
|
||||
if (this.settings.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 deleteDirectory(bucketName, key) {
|
||||
try {
|
||||
const [files] = await this.storage
|
||||
.bucket(bucketName)
|
||||
.getFiles({ directory: key })
|
||||
|
||||
await asyncPool(this.settings.deleteConcurrency, files, async (file) => {
|
||||
await this.deleteObject(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 directorySize(bucketName, key) {
|
||||
let files
|
||||
|
||||
try {
|
||||
const [response] = await this.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 checkIfObjectExists(bucketName, key) {
|
||||
try {
|
||||
const [response] = await this.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 copyObject(bucketName, sourceKey, destKey) {
|
||||
try {
|
||||
const src = this.storage.bucket(bucketName).file(sourceKey)
|
||||
const dest = this.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
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,256 @@
|
||||
const AbstractPersistor = require('./AbstractPersistor')
|
||||
const Logger = require('logger-sharelatex')
|
||||
const Stream = require('stream')
|
||||
const { 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.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.fallback.buckets`
|
||||
// e.g.
|
||||
// Settings.fallback.buckets = {
|
||||
// myBucketOnS3: 'myBucketOnGCS'
|
||||
// }
|
||||
|
||||
module.exports = class MigrationPersistor extends AbstractPersistor {
|
||||
constructor(primaryPersistor, fallbackPersistor, settings) {
|
||||
super()
|
||||
|
||||
this.primaryPersistor = primaryPersistor
|
||||
this.fallbackPersistor = fallbackPersistor
|
||||
this.settings = settings
|
||||
}
|
||||
|
||||
async sendFile(...args) {
|
||||
return this.primaryPersistor.sendFile(...args)
|
||||
}
|
||||
|
||||
async sendStream(...args) {
|
||||
return this.primaryPersistor.sendStream(...args)
|
||||
}
|
||||
|
||||
async getRedirectUrl(...args) {
|
||||
return this.primaryPersistor.getRedirectUrl(...args)
|
||||
}
|
||||
|
||||
async getObjectMd5Hash(...args) {
|
||||
return this._runWithFallback('getObjectMd5Hash', ...args)
|
||||
}
|
||||
|
||||
async checkIfObjectExists(...args) {
|
||||
return this._runWithFallback('checkIfObjectExists', ...args)
|
||||
}
|
||||
|
||||
async getObjectSize(...args) {
|
||||
return this._runWithFallback('getObjectSize', ...args)
|
||||
}
|
||||
|
||||
async directorySize(...args) {
|
||||
return this._runWithFallback('directorySize', ...args)
|
||||
}
|
||||
|
||||
async deleteObject(...args) {
|
||||
return this._runOnBoth('deleteObject', ...args)
|
||||
}
|
||||
|
||||
async deleteDirectory(...args) {
|
||||
return this._runOnBoth('deleteDirectory', ...args)
|
||||
}
|
||||
|
||||
async getObjectStream(bucket, key, opts) {
|
||||
const shouldCopy = this.settings.copyOnMiss && !opts.start && !opts.end
|
||||
|
||||
try {
|
||||
// 'return await' so we catch NotFoundError before returning
|
||||
return await this.primaryPersistor.getObjectStream(bucket, key, opts)
|
||||
} catch (err) {
|
||||
if (err instanceof NotFoundError) {
|
||||
const fallbackBucket = this._getFallbackBucket(bucket)
|
||||
const fallbackStream = await this.fallbackPersistor.getObjectStream(
|
||||
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).catch((error) => {
|
||||
Logger.warn({ error }, 'failed to copy object from fallback')
|
||||
})
|
||||
|
||||
if (shouldCopy) {
|
||||
const copyStream = new Stream.PassThrough()
|
||||
pipeline(fallbackStream, copyStream).catch((error) => {
|
||||
Logger.warn({ error }, 'failed to copy object from fallback')
|
||||
})
|
||||
|
||||
this._copyStreamFromFallbackAndVerify(
|
||||
copyStream,
|
||||
fallbackBucket,
|
||||
bucket,
|
||||
key,
|
||||
key
|
||||
).catch((error) => {
|
||||
Logger.warn({ error }, 'failed to copy file from fallback')
|
||||
})
|
||||
}
|
||||
return returnStream
|
||||
}
|
||||
throw err
|
||||
}
|
||||
}
|
||||
|
||||
async copyObject(bucket, sourceKey, destKey) {
|
||||
try {
|
||||
return await this.primaryPersistor.copyObject(bucket, sourceKey, destKey)
|
||||
} catch (err) {
|
||||
if (err instanceof NotFoundError) {
|
||||
const fallbackBucket = this._getFallbackBucket(bucket)
|
||||
const fallbackStream = await this.fallbackPersistor.getObjectStream(
|
||||
fallbackBucket,
|
||||
sourceKey,
|
||||
{}
|
||||
)
|
||||
|
||||
const copyStream = new Stream.PassThrough()
|
||||
pipeline(fallbackStream, copyStream).catch((error) => {
|
||||
Logger.warn({ error }, 'failed to copy object from fallback')
|
||||
})
|
||||
|
||||
if (this.settings.copyOnMiss) {
|
||||
const missStream = new Stream.PassThrough()
|
||||
pipeline(fallbackStream, missStream).catch((error) => {
|
||||
Logger.warn({ error }, 'failed to copy object from fallback')
|
||||
})
|
||||
|
||||
// copy from sourceKey -> sourceKey
|
||||
this._copyStreamFromFallbackAndVerify(
|
||||
missStream,
|
||||
fallbackBucket,
|
||||
bucket,
|
||||
sourceKey,
|
||||
sourceKey
|
||||
).catch(() => {
|
||||
// swallow errors, as this runs in the background and will log a warning
|
||||
})
|
||||
}
|
||||
// copy from sourceKey -> destKey
|
||||
return this._copyStreamFromFallbackAndVerify(
|
||||
copyStream,
|
||||
fallbackBucket,
|
||||
bucket,
|
||||
sourceKey,
|
||||
destKey
|
||||
)
|
||||
}
|
||||
throw err
|
||||
}
|
||||
}
|
||||
|
||||
async _copyStreamFromFallbackAndVerify(
|
||||
stream,
|
||||
sourceBucket,
|
||||
destBucket,
|
||||
sourceKey,
|
||||
destKey
|
||||
) {
|
||||
try {
|
||||
let sourceMd5
|
||||
try {
|
||||
sourceMd5 = await this.fallbackPersistor.getObjectMd5Hash(
|
||||
sourceBucket,
|
||||
sourceKey
|
||||
)
|
||||
} catch (err) {
|
||||
Logger.warn(err, 'error getting md5 hash from fallback persistor')
|
||||
}
|
||||
|
||||
await this.primaryPersistor.sendStream(
|
||||
destBucket,
|
||||
destKey,
|
||||
stream,
|
||||
sourceMd5
|
||||
)
|
||||
} catch (err) {
|
||||
const error = new WriteError({
|
||||
message: 'unable to copy file to destination persistor',
|
||||
info: {
|
||||
sourceBucket,
|
||||
destBucket,
|
||||
sourceKey,
|
||||
destKey
|
||||
}
|
||||
}).withCause(err)
|
||||
if (this.settings.Metrics) {
|
||||
this.settings.Metrics.inc('fallback.copy.failure')
|
||||
}
|
||||
|
||||
try {
|
||||
await this.primaryPersistor.deleteObject(destBucket, destKey)
|
||||
} catch (err) {
|
||||
error.info.cleanupError = new WriteError({
|
||||
message: 'unable to clean up destination copy artifact',
|
||||
info: {
|
||||
destBucket,
|
||||
destKey
|
||||
}
|
||||
}).withCause(err)
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
_getFallbackBucket(bucket) {
|
||||
return (this.settings.buckets && this.settings.buckets[bucket]) || bucket
|
||||
}
|
||||
|
||||
async _runOnBoth(methodName, bucket, ...moreArgs) {
|
||||
const fallbackBucket = this._getFallbackBucket(bucket)
|
||||
|
||||
await Promise.all([
|
||||
this.primaryPersistor[methodName](bucket, ...moreArgs),
|
||||
this.fallbackPersistor[methodName](fallbackBucket, ...moreArgs)
|
||||
])
|
||||
}
|
||||
|
||||
async _runWithFallback(methodName, bucket, key, ...moreArgs) {
|
||||
try {
|
||||
// 'return await' so we catch NotFoundError before returning
|
||||
return await this.primaryPersistor[methodName](bucket, key, ...moreArgs)
|
||||
} catch (err) {
|
||||
if (err instanceof NotFoundError) {
|
||||
const fallbackBucket = this._getFallbackBucket(bucket)
|
||||
if (this.settings.copyOnMiss) {
|
||||
const fallbackStream = await this.fallbackPersistor.getObjectStream(
|
||||
fallbackBucket,
|
||||
key,
|
||||
{}
|
||||
)
|
||||
// run in background
|
||||
this._copyStreamFromFallbackAndVerify(
|
||||
fallbackStream,
|
||||
fallbackBucket,
|
||||
bucket,
|
||||
key,
|
||||
key
|
||||
).catch((err) => {
|
||||
Logger.warn({ err }, 'failed to copy file from fallback')
|
||||
})
|
||||
}
|
||||
return this.fallbackPersistor[methodName](
|
||||
fallbackBucket,
|
||||
key,
|
||||
...moreArgs
|
||||
)
|
||||
}
|
||||
throw err
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
const Logger = require('logger-sharelatex')
|
||||
const { SettingsError } = require('./Errors')
|
||||
const GcsPersistor = require('./GcsPersistor')
|
||||
const S3Persistor = require('./S3Persistor')
|
||||
const FSPersistor = require('./FSPersistor')
|
||||
const MigrationPersistor = require('./MigrationPersistor')
|
||||
|
||||
function getPersistor(backend, settings) {
|
||||
switch (backend) {
|
||||
case 'aws-sdk':
|
||||
case 's3':
|
||||
return new S3Persistor(
|
||||
Object.assign({}, settings.s3, { Metrics: settings.Metrics })
|
||||
)
|
||||
case 'fs':
|
||||
return new FSPersistor({
|
||||
paths: settings.paths,
|
||||
Metrics: settings.Metrics
|
||||
})
|
||||
case 'gcs':
|
||||
return new GcsPersistor(
|
||||
Object.assign({}, settings.gcs, { Metrics: settings.Metrics })
|
||||
)
|
||||
default:
|
||||
throw new SettingsError({ message: 'unknown backend', info: { backend } })
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = function create(settings) {
|
||||
Logger.info(
|
||||
{
|
||||
backend: settings.backend,
|
||||
fallback: settings.fallback && settings.fallback.backend
|
||||
},
|
||||
'Loading backend'
|
||||
)
|
||||
if (!settings.backend) {
|
||||
throw new SettingsError({
|
||||
message: 'no backend specified - config incomplete',
|
||||
info: {}
|
||||
})
|
||||
}
|
||||
|
||||
let persistor = getPersistor(settings.backend, settings)
|
||||
|
||||
if (settings.fallback && settings.fallback.backend) {
|
||||
const primary = persistor
|
||||
const fallback = getPersistor(settings.fallback.backend, settings)
|
||||
persistor = new MigrationPersistor(
|
||||
primary,
|
||||
fallback,
|
||||
Object.assign({}, settings.fallback, { Metrics: settings.Metrics })
|
||||
)
|
||||
}
|
||||
|
||||
return persistor
|
||||
}
|
||||
@@ -0,0 +1,185 @@
|
||||
const Crypto = require('crypto')
|
||||
const Stream = require('stream')
|
||||
const Logger = require('logger-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) {
|
||||
super({ autoDestroy: true, ...options })
|
||||
|
||||
this.bytes = 0
|
||||
|
||||
if (options.hash) {
|
||||
this.hash = Crypto.createHash(options.hash)
|
||||
}
|
||||
|
||||
if (options.metric && options.Metrics) {
|
||||
const onEnd = () => {
|
||||
options.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
|
||||
async function calculateStreamMd5(stream) {
|
||||
const hash = Crypto.createHash('md5')
|
||||
hash.setEncoding('hex')
|
||||
|
||||
await pipeline(stream, hash)
|
||||
return 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.getObjectMd5Hash(bucket, key)
|
||||
}
|
||||
|
||||
if (sourceMd5 !== destMd5) {
|
||||
try {
|
||||
await persistor.deleteObject(bucket, key)
|
||||
} catch (err) {
|
||||
Logger.warn(err, 'error deleting file for invalid upload')
|
||||
}
|
||||
|
||||
throw new WriteError({
|
||||
message: 'source and destination hashes do not match',
|
||||
info: {
|
||||
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({
|
||||
message: 'no such file',
|
||||
info: params
|
||||
}).withCause(error)
|
||||
} else {
|
||||
return new ErrorType({
|
||||
message: message,
|
||||
info: params
|
||||
}).withCause(error)
|
||||
}
|
||||
}
|
||||
|
||||
function base64ToHex(base64) {
|
||||
return Buffer.from(base64, 'base64').toString('hex')
|
||||
}
|
||||
|
||||
function hexToBase64(hex) {
|
||||
return Buffer.from(hex, 'hex').toString('base64')
|
||||
}
|
||||
@@ -0,0 +1,352 @@
|
||||
const http = require('http')
|
||||
const https = require('https')
|
||||
if (http.globalAgent.maxSockets < 300) {
|
||||
http.globalAgent.maxSockets = 300
|
||||
}
|
||||
if (https.globalAgent.maxSockets < 300) {
|
||||
https.globalAgent.maxSockets = 300
|
||||
}
|
||||
|
||||
const AbstractPersistor = require('./AbstractPersistor')
|
||||
const PersistorHelper = require('./PersistorHelper')
|
||||
|
||||
const fs = require('fs')
|
||||
const S3 = require('aws-sdk/clients/s3')
|
||||
const { URL } = require('url')
|
||||
const {
|
||||
WriteError,
|
||||
ReadError,
|
||||
NotFoundError,
|
||||
SettingsError
|
||||
} = require('./Errors')
|
||||
|
||||
module.exports = class S3Persistor extends AbstractPersistor {
|
||||
constructor(settings = {}) {
|
||||
super()
|
||||
|
||||
this.settings = settings
|
||||
}
|
||||
|
||||
async sendFile(bucketName, key, fsPath) {
|
||||
return this.sendStream(bucketName, key, fs.createReadStream(fsPath))
|
||||
}
|
||||
|
||||
async sendStream(bucketName, key, readStream, sourceMd5) {
|
||||
try {
|
||||
// egress from us to S3
|
||||
const observeOptions = {
|
||||
metric: 's3.egress',
|
||||
Metrics: this.settings.Metrics
|
||||
}
|
||||
|
||||
const observer = new PersistorHelper.ObserverStream(observeOptions)
|
||||
// observer will catch errors, clean up and log a warning
|
||||
readStream.pipe(observer)
|
||||
|
||||
// if we have an md5 hash, pass this to S3 to verify the upload
|
||||
const uploadOptions = {
|
||||
Bucket: bucketName,
|
||||
Key: key,
|
||||
Body: observer
|
||||
}
|
||||
|
||||
// if we have an md5 hash, pass this to S3 to verify the upload - otherwise
|
||||
// we rely on the S3 client's checksum calculation to validate the upload
|
||||
const clientOptions = {}
|
||||
if (sourceMd5) {
|
||||
uploadOptions.ContentMD5 = PersistorHelper.hexToBase64(sourceMd5)
|
||||
} else {
|
||||
clientOptions.computeChecksums = true
|
||||
}
|
||||
|
||||
await this._getClientForBucket(bucketName, clientOptions)
|
||||
.upload(uploadOptions, { partSize: this.settings.partSize })
|
||||
.promise()
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'upload to S3 failed',
|
||||
{ bucketName, key },
|
||||
WriteError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async getObjectStream(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 = this._getClientForBucket(bucketName)
|
||||
.getObject(params)
|
||||
.createReadStream()
|
||||
|
||||
// ingress from S3 to us
|
||||
const observer = new PersistorHelper.ObserverStream({
|
||||
metric: 's3.ingress',
|
||||
Metrics: this.settings.Metrics
|
||||
})
|
||||
|
||||
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 getRedirectUrl() {
|
||||
// not implemented
|
||||
return null
|
||||
}
|
||||
|
||||
async deleteDirectory(bucketName, key, continuationToken) {
|
||||
let response
|
||||
const options = { Bucket: bucketName, Prefix: key }
|
||||
if (continuationToken) {
|
||||
options.ContinuationToken = continuationToken
|
||||
}
|
||||
|
||||
try {
|
||||
response = await this._getClientForBucket(bucketName)
|
||||
.listObjectsV2(options)
|
||||
.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 this._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
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
if (response.IsTruncated) {
|
||||
await this.deleteDirectory(
|
||||
bucketName,
|
||||
key,
|
||||
response.NextContinuationToken
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async getObjectSize(bucketName, key) {
|
||||
try {
|
||||
const response = await this._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 getObjectMd5Hash(bucketName, key) {
|
||||
try {
|
||||
const response = await this._getClientForBucket(bucketName)
|
||||
.headObject({ Bucket: bucketName, Key: key })
|
||||
.promise()
|
||||
const md5 = S3Persistor._md5FromResponse(response)
|
||||
if (md5) {
|
||||
return md5
|
||||
}
|
||||
// etag is not in md5 format
|
||||
if (this.settings.Metrics) {
|
||||
this.settings.Metrics.inc('s3.md5Download')
|
||||
}
|
||||
return PersistorHelper.calculateStreamMd5(
|
||||
await this.getObjectStream(bucketName, key)
|
||||
)
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'error getting hash of s3 object',
|
||||
{ bucketName, key },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async deleteObject(bucketName, key) {
|
||||
try {
|
||||
await this._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 copyObject(bucketName, sourceKey, destKey) {
|
||||
const params = {
|
||||
Bucket: bucketName,
|
||||
Key: destKey,
|
||||
CopySource: `${bucketName}/${sourceKey}`
|
||||
}
|
||||
try {
|
||||
await this._getClientForBucket(bucketName).copyObject(params).promise()
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'failed to copy file in S3',
|
||||
params,
|
||||
WriteError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async checkIfObjectExists(bucketName, key) {
|
||||
try {
|
||||
await this.getObjectSize(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 directorySize(bucketName, key, continuationToken) {
|
||||
try {
|
||||
const options = {
|
||||
Bucket: bucketName,
|
||||
Prefix: key
|
||||
}
|
||||
if (continuationToken) {
|
||||
options.ContinuationToken = continuationToken
|
||||
}
|
||||
const response = await this._getClientForBucket(bucketName)
|
||||
.listObjectsV2(options)
|
||||
.promise()
|
||||
|
||||
const size = response.Contents.reduce((acc, item) => item.Size + acc, 0)
|
||||
if (response.IsTruncated) {
|
||||
return (
|
||||
size +
|
||||
(await this.directorySize(
|
||||
bucketName,
|
||||
key,
|
||||
response.NextContinuationToken
|
||||
))
|
||||
)
|
||||
}
|
||||
return size
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
'error getting directory size in S3',
|
||||
{ bucketName, key },
|
||||
ReadError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
_getClientForBucket(bucket, clientOptions) {
|
||||
if (this.settings.bucketCreds && this.settings.bucketCreds[bucket]) {
|
||||
return new S3(
|
||||
this._buildClientOptions(
|
||||
this.settings.bucketCreds[bucket],
|
||||
clientOptions
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
// no specific credentials for the bucket
|
||||
if (this.settings.key) {
|
||||
return new S3(this._buildClientOptions(null, clientOptions))
|
||||
}
|
||||
|
||||
throw new SettingsError({
|
||||
message: 'no bucket-specific or default credentials provided',
|
||||
info: { bucket }
|
||||
})
|
||||
}
|
||||
|
||||
_buildClientOptions(bucketCredentials, clientOptions) {
|
||||
const options = clientOptions || {}
|
||||
|
||||
if (bucketCredentials) {
|
||||
options.credentials = {
|
||||
accessKeyId: bucketCredentials.auth_key,
|
||||
secretAccessKey: bucketCredentials.auth_secret
|
||||
}
|
||||
} else {
|
||||
options.credentials = {
|
||||
accessKeyId: this.settings.key,
|
||||
secretAccessKey: this.settings.secret
|
||||
}
|
||||
}
|
||||
|
||||
if (this.settings.endpoint) {
|
||||
const endpoint = new URL(this.settings.endpoint)
|
||||
options.endpoint = this.settings.endpoint
|
||||
options.sslEnabled = endpoint.protocol === 'https'
|
||||
}
|
||||
|
||||
// path-style access is only used for acceptance tests
|
||||
if (this.settings.pathStyle) {
|
||||
options.s3ForcePathStyle = true
|
||||
}
|
||||
|
||||
return options
|
||||
}
|
||||
|
||||
static _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