mirror of
https://github.com/yu-i-i/overleaf-cep.git
synced 2026-11-12 13:13:45 +01:00
Merge pull request #28554 from overleaf/msm-force-s3-lib-storage-uploads
[object-persistor] Use `@aws-sdk/lib-storage` for all uploads GitOrigin-RevId: ab8e54a7bae843f9e6b05ed9cf936130a36b8c2f
This commit is contained in:
@@ -46,7 +46,7 @@ Uploads a stream to the backend.
|
||||
- `key`: The key for the uploaded object
|
||||
- `readStream`: The data stream to upload
|
||||
- `opts` (optional):
|
||||
- `sourceMd5`: The md5 hash of the source data, if known. The uploaded data will be compared against this and the operation will fail if it does not match. If omitted, the md5 is calculated as the data is uploaded instead, and verified against the backend.
|
||||
- `sourceMd5`: The md5 hash of the source data, if known. The uploaded data will be compared against this and the operation will fail if it does not match. If omitted, the md5 is calculated as the data is uploaded instead, and verified against the backend. This is not supported in `S3Persistor` as it performs [its own integrity protections](https://aws.amazon.com/blogs/aws/introducing-default-data-integrity-protections-for-new-objects-in-amazon-s3/). Setting `sourceMd5` with `S3Persistor` will result in an error being thrown.
|
||||
- `contentType`: The content type to write in the object metadata
|
||||
- `contentEncoding`: The content encoding to write in the object metadata
|
||||
|
||||
@@ -266,7 +266,8 @@ For the `FS` persistor, the `bucketName` should be the full path to the folder o
|
||||
- `s3.partSize`: The part size for S3 uploads. Defaults to 100 megabytes.
|
||||
- `s3.httpOptions`: HTTP Options passed to the [`NodeHttpHandler` constructor](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/Package/-smithy-node-http-handler/Class/NodeHttpHandler/)
|
||||
- For backwards compatibility reasons, the `timeout` property that was passed to the [S3 constructor](https://docs.aws.amazon.com/AWSJavaScriptSDK/latest/AWS/S3.html#constructor-property) before migrating to AWS SDK v3 is now passed to the `NodeHttpHandler` constructor as `connectionTimeout`.
|
||||
- `s3.maxRetries`: The number of times the S3 client will retry in case of an error
|
||||
- `s3.maxRetries` (legacy): The number of times the S3 client will retry in case of an error
|
||||
- `s3.maxAttempts`: The number of times the S3 client will attempt to perform the operation in case there are errors. Default value is 3.
|
||||
- `s3.endpoint`: For testing - overrides the S3 endpoint to use a different service (e.g. a fake S3 server)
|
||||
- `s3.pathStyle`: For testing - use old path-style URLs, for services that do not support subdomain-based access
|
||||
|
||||
@@ -281,12 +282,6 @@ For the `FS` persistor, the `bucketName` should be the full path to the folder o
|
||||
}
|
||||
```
|
||||
|
||||
#### Notes
|
||||
|
||||
In order for server-side MD5 generation to work, uploads must be below the `partSize`. Otherwise a multipart upload will be used, and the S3 `eTag` which is used to retrieve the MD5 will not be the MD5 hash of the uploaded object. In these cases, we download the data and calculate the MD5 manually.
|
||||
|
||||
For verification during upload, we use S3's checksum mechanism to verify the integrity of the uploaded data, but when explicitly retrieving the md5 hash this will download the entire object if its size is above the part size.
|
||||
|
||||
### GCS-specific parameters
|
||||
|
||||
GCS authentication is configured automatically via the local service account, or the `GOOGLE_APPLICATION_CREDENTIALS` environment variable.
|
||||
|
||||
@@ -171,19 +171,7 @@ module.exports = class MigrationPersistor extends AbstractPersistor {
|
||||
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,
|
||||
})
|
||||
await this.primaryPersistor.sendStream(destBucket, destKey, stream)
|
||||
} catch (err) {
|
||||
const error = new WriteError(
|
||||
'unable to copy file to destination persistor',
|
||||
|
||||
@@ -433,7 +433,6 @@ class CachedPerProjectEncryptedS3Persistor {
|
||||
* @param {number} [opts.contentLength]
|
||||
* @param {'*'} [opts.ifNoneMatch]
|
||||
* @param {SSECOptions} [opts.ssecOptions]
|
||||
* @param {string} [opts.sourceMd5]
|
||||
* @return {Promise<void>}
|
||||
*/
|
||||
async sendStream(bucketName, path, sourceStream, opts = {}) {
|
||||
|
||||
@@ -81,9 +81,6 @@ class S3Persistor extends AbstractPersistor {
|
||||
/** @type {Map<string, S3Client>} */
|
||||
#clients = new Map()
|
||||
|
||||
/** @type {Map<string, S3Client>} */
|
||||
#noRetryClients = new Map()
|
||||
|
||||
constructor(settings = {}) {
|
||||
super()
|
||||
|
||||
@@ -111,7 +108,6 @@ class S3Persistor extends AbstractPersistor {
|
||||
* @param {number} [opts.contentLength]
|
||||
* @param {'*'} [opts.ifNoneMatch]
|
||||
* @param {SSECOptions} [opts.ssecOptions]
|
||||
* @param {string} [opts.sourceMd5]
|
||||
* @return {Promise<void>}
|
||||
*/
|
||||
async sendStream(bucketName, key, readStream, opts = {}) {
|
||||
@@ -136,6 +132,12 @@ class S3Persistor extends AbstractPersistor {
|
||||
uploadOptions.StorageClass = this.settings.storageClass[bucketName]
|
||||
}
|
||||
|
||||
if ('sourceMd5' in opts) {
|
||||
// we fail straight away to prevent the client from wasting CPU/IO precomputing the hash
|
||||
throw new Error(
|
||||
'sourceMd5 option is not supported, S3 provides its own integrity protection mechanism'
|
||||
)
|
||||
}
|
||||
if (opts.contentType) {
|
||||
uploadOptions.ContentType = opts.contentType
|
||||
}
|
||||
@@ -152,32 +154,12 @@ class S3Persistor extends AbstractPersistor {
|
||||
Object.assign(uploadOptions, opts.ssecOptions.getPutOptions())
|
||||
}
|
||||
|
||||
// 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
|
||||
let computeChecksums = false
|
||||
if (opts.sourceMd5) {
|
||||
uploadOptions.ContentMD5 = PersistorHelper.hexToBase64(opts.sourceMd5)
|
||||
} else {
|
||||
computeChecksums = true
|
||||
}
|
||||
|
||||
if (this.settings.disableMultiPartUpload) {
|
||||
// retries are disabled for `PutObjectCommand` when using streams
|
||||
// https://github.com/aws/aws-sdk-js-v3/issues/6770
|
||||
const noRetry = true
|
||||
await this._getClientForBucket(
|
||||
bucketName,
|
||||
computeChecksums,
|
||||
noRetry
|
||||
).send(new PutObjectCommand(uploadOptions))
|
||||
} else {
|
||||
const upload = new Upload({
|
||||
client: this._getClientForBucket(bucketName, computeChecksums),
|
||||
params: uploadOptions,
|
||||
partSize: this.settings.partSize,
|
||||
})
|
||||
await upload.done()
|
||||
}
|
||||
const upload = new Upload({
|
||||
client: this._getClientForBucket(bucketName),
|
||||
params: uploadOptions,
|
||||
partSize: this.settings.partSize,
|
||||
})
|
||||
await upload.done()
|
||||
} catch (err) {
|
||||
throw PersistorHelper.wrapError(
|
||||
err,
|
||||
@@ -555,34 +537,31 @@ class S3Persistor extends AbstractPersistor {
|
||||
|
||||
/**
|
||||
* @param {string} bucket
|
||||
* @param {boolean} computeChecksums
|
||||
* @param {boolean} noRetries
|
||||
* @return {S3Client}
|
||||
* @private
|
||||
*/
|
||||
_getClientForBucket(bucket, computeChecksums = false, noRetries = false) {
|
||||
_getClientForBucket(bucket) {
|
||||
/** @type {import('@aws-sdk/client-s3').S3ClientConfig} */
|
||||
const clientOptions = {}
|
||||
const cacheKey = `${bucket}:${computeChecksums}`
|
||||
if (computeChecksums) {
|
||||
clientOptions.requestChecksumCalculation = 'WHEN_SUPPORTED'
|
||||
clientOptions.responseChecksumValidation = 'WHEN_SUPPORTED'
|
||||
}
|
||||
const cacheKey = bucket
|
||||
|
||||
const clientMap = noRetries ? this.#noRetryClients : this.#clients
|
||||
let client = clientMap.get(cacheKey)
|
||||
let client = this.#clients.get(cacheKey)
|
||||
if (!client) {
|
||||
client = new S3Client(
|
||||
this._buildClientOptions(
|
||||
this.settings.bucketCreds?.[bucket],
|
||||
clientOptions,
|
||||
noRetries
|
||||
clientOptions
|
||||
)
|
||||
)
|
||||
clientMap.set(cacheKey, client)
|
||||
this.#clients.set(cacheKey, client)
|
||||
|
||||
// Some third-party S3-compatible services (as MinIO) do not support the default checksums
|
||||
// for object integrity in `DeleteObjectsCommand`. We add a fallback for those cases.
|
||||
// https://github.com/aws/aws-sdk-js-v3/blob/main/supplemental-docs/MD5_FALLBACK.md
|
||||
addMd5Middleware(client)
|
||||
const useMd5Fallback = process.env.DELETE_OBJECTS_MD5_FALLBACK === 'true'
|
||||
if (useMd5Fallback) {
|
||||
addMd5Middleware(client)
|
||||
}
|
||||
}
|
||||
|
||||
return client
|
||||
@@ -591,11 +570,10 @@ class S3Persistor extends AbstractPersistor {
|
||||
/**
|
||||
* @param {Object} bucketCredentials
|
||||
* @param {import('@aws-sdk/client-s3').S3ClientConfig} clientOptions
|
||||
* @param {boolean} noRetries
|
||||
* @return {import('@aws-sdk/client-s3').S3ClientConfig}
|
||||
* @private
|
||||
*/
|
||||
_buildClientOptions(bucketCredentials, clientOptions, noRetries = false) {
|
||||
_buildClientOptions(bucketCredentials, clientOptions) {
|
||||
const options = clientOptions || {}
|
||||
|
||||
if (bucketCredentials) {
|
||||
@@ -635,10 +613,6 @@ class S3Persistor extends AbstractPersistor {
|
||||
options.maxAttempts = this.settings.maxRetries + 1
|
||||
}
|
||||
|
||||
if (noRetries) {
|
||||
options.maxAttempts = 1
|
||||
}
|
||||
|
||||
const requestHandlerParams = this.settings.httpOptions || {}
|
||||
|
||||
if (sslEnabled && this.settings.ca) {
|
||||
|
||||
@@ -197,19 +197,11 @@ describe('MigrationPersistorTests', function () {
|
||||
expect(fallbackPersistor.getObjectStream).to.have.been.calledOnce
|
||||
})
|
||||
|
||||
it('should get the md5 hash from the source', function () {
|
||||
expect(fallbackPersistor.getObjectMd5Hash).to.have.been.calledWith(
|
||||
fallbackBucket,
|
||||
key
|
||||
)
|
||||
})
|
||||
|
||||
it('should send a stream to the primary', function () {
|
||||
expect(primaryPersistor.sendStream).to.have.been.calledWithExactly(
|
||||
bucket,
|
||||
key,
|
||||
sinon.match.instanceOf(Stream.PassThrough),
|
||||
{ sourceMd5: md5 }
|
||||
sinon.match.instanceOf(Stream.PassThrough)
|
||||
)
|
||||
})
|
||||
|
||||
@@ -476,19 +468,11 @@ describe('MigrationPersistorTests', function () {
|
||||
).not.to.have.been.calledWithExactly(fallbackBucket, key)
|
||||
})
|
||||
|
||||
it('should get the md5 hash from the source', function () {
|
||||
expect(fallbackPersistor.getObjectMd5Hash).to.have.been.calledWith(
|
||||
fallbackBucket,
|
||||
key
|
||||
)
|
||||
})
|
||||
|
||||
it('should send the file to the primary', function () {
|
||||
expect(primaryPersistor.sendStream).to.have.been.calledWithExactly(
|
||||
bucket,
|
||||
destKey,
|
||||
sinon.match.instanceOf(Stream.PassThrough),
|
||||
{ sourceMd5: md5 }
|
||||
sinon.match.instanceOf(Stream.PassThrough)
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -108,11 +108,5 @@ module.exports = function () {
|
||||
this.payload = payload
|
||||
}
|
||||
},
|
||||
PutObjectCommand: class PutObjectCommand {
|
||||
constructor(payload) {
|
||||
this.name = 'PutObjectCommand'
|
||||
this.payload = payload
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -549,27 +549,6 @@ describe('S3PersistorTests', function () {
|
||||
})
|
||||
})
|
||||
|
||||
describe('when a hash is supplied', function () {
|
||||
beforeEach(async function () {
|
||||
await S3Persistor.sendStream(bucket, key, ReadStream, {
|
||||
sourceMd5: 'aaaaaaaabbbbbbbbaaaaaaaabbbbbbbb',
|
||||
})
|
||||
})
|
||||
|
||||
it('sends the hash in base64', function () {
|
||||
expect(awsLibStorageUpload).to.have.been.calledWith({
|
||||
client: S3.s3ClientStub,
|
||||
params: {
|
||||
Bucket: bucket,
|
||||
Key: key,
|
||||
Body: sinon.match.instanceOf(Transform),
|
||||
ContentMD5: 'qqqqqru7u7uqqqqqu7u7uw==',
|
||||
},
|
||||
partSize: 100 * 1024 * 1024,
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
describe('when metadata is supplied', function () {
|
||||
const contentType = 'text/csv'
|
||||
const contentEncoding = 'gzip'
|
||||
@@ -596,23 +575,23 @@ describe('S3PersistorTests', function () {
|
||||
})
|
||||
})
|
||||
|
||||
describe('when multipart upload is disabled', function () {
|
||||
const contentType = 'text/csv'
|
||||
const contentEncoding = 'gzip'
|
||||
|
||||
describe('with sourceMd5 option', function () {
|
||||
let error
|
||||
beforeEach(async function () {
|
||||
S3.mockSend(S3.PutObjectCommand, {})
|
||||
settings.disableMultiPartUpload = true
|
||||
await S3Persistor.sendStream(bucket, key, ReadStream, {
|
||||
contentType,
|
||||
contentEncoding,
|
||||
})
|
||||
try {
|
||||
await S3Persistor.sendStream(bucket, key, ReadStream, {
|
||||
sourceMd5: 'ffffffff',
|
||||
})
|
||||
} catch (err) {
|
||||
error = err
|
||||
}
|
||||
})
|
||||
|
||||
it('configures the options to not to retry requests', function () {
|
||||
expect(S3.S3Client).to.have.been.calledWithMatch({
|
||||
maxAttempts: 1,
|
||||
})
|
||||
it('should throw an error', function () {
|
||||
expect(error.message).to.equal('upload to S3 failed')
|
||||
expect(error.cause.message).to.equal(
|
||||
'sourceMd5 option is not supported, S3 provides its own integrity protection mechanism'
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
Reference in New Issue
Block a user