s3: make callback of getS3Config
This commit is contained in:
Girish Ramakrishnan
@@ -57,11 +57,10 @@ function S3_NOT_FOUND(error) {
|
||||
return error.code === 'NoSuchKey' || error.code === 'NotFound' || error.code === 'ENOENT';
|
||||
}
|
||||
|
||||
function getS3Config(apiConfig, callback) {
|
||||
function getS3Config(apiConfig) {
|
||||
assert.strictEqual(typeof apiConfig, 'object');
|
||||
assert.strictEqual(typeof callback, 'function');
|
||||
|
||||
let credentials = {
|
||||
const credentials = {
|
||||
signatureVersion: apiConfig.signatureVersion || 'v4',
|
||||
s3ForcePathStyle: false, // Use vhost style instead of path style - https://forums.aws.amazon.com/ann.jspa?annID=6776
|
||||
accessKeyId: apiConfig.accessKeyId,
|
||||
@@ -88,7 +87,8 @@ function getS3Config(apiConfig, callback) {
|
||||
credentials.httpOptions.agent = new https.Agent({ rejectUnauthorized: false });
|
||||
}
|
||||
}
|
||||
callback(null, credentials);
|
||||
|
||||
return credentials;
|
||||
}
|
||||
|
||||
// storage api
|
||||
@@ -112,33 +112,31 @@ function upload(apiConfig, backupFilePath, sourceStream, callback) {
|
||||
assert.strictEqual(typeof sourceStream, 'object');
|
||||
assert.strictEqual(typeof callback, 'function');
|
||||
|
||||
getS3Config(apiConfig, function (error, credentials) {
|
||||
if (error) return callback(error);
|
||||
const credentials = getS3Config(apiConfig);
|
||||
|
||||
const params = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: backupFilePath,
|
||||
Body: sourceStream
|
||||
};
|
||||
const params = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: backupFilePath,
|
||||
Body: sourceStream
|
||||
};
|
||||
|
||||
const s3 = new aws.S3(credentials);
|
||||
const s3 = new aws.S3(credentials);
|
||||
|
||||
// s3.upload automatically does a multi-part upload. we set queueSize to 3 to reduce memory usage
|
||||
// uploader will buffer at most queueSize * partSize bytes into memory at any given time.
|
||||
// scaleway only supports 1000 parts per object (https://www.scaleway.com/en/docs/s3-multipart-upload/)
|
||||
// s3: https://docs.aws.amazon.com/AmazonS3/latest/dev/qfacts.html (max 10k parts and no size limit on the last part!)
|
||||
const partSize = apiConfig.uploadPartSize || (apiConfig.provider === 'scaleway-objectstorage' ? 100 * 1024 * 1024 : 10 * 1024 * 1024);
|
||||
// s3.upload automatically does a multi-part upload. we set queueSize to 3 to reduce memory usage
|
||||
// uploader will buffer at most queueSize * partSize bytes into memory at any given time.
|
||||
// scaleway only supports 1000 parts per object (https://www.scaleway.com/en/docs/s3-multipart-upload/)
|
||||
// s3: https://docs.aws.amazon.com/AmazonS3/latest/dev/qfacts.html (max 10k parts and no size limit on the last part!)
|
||||
const partSize = apiConfig.uploadPartSize || (apiConfig.provider === 'scaleway-objectstorage' ? 100 * 1024 * 1024 : 10 * 1024 * 1024);
|
||||
|
||||
s3.upload(params, { partSize, queueSize: 3 }, function (error, data) {
|
||||
if (error) {
|
||||
debug('Error uploading [%s]: s3 upload error.', backupFilePath, error);
|
||||
return callback(new BoxError(BoxError.EXTERNAL_ERROR, `Error uploading ${backupFilePath}. Message: ${error.message} HTTP Code: ${error.code}`));
|
||||
}
|
||||
s3.upload(params, { partSize, queueSize: 3 }, function (error, data) {
|
||||
if (error) {
|
||||
debug('Error uploading [%s]: s3 upload error.', backupFilePath, error);
|
||||
return callback(new BoxError(BoxError.EXTERNAL_ERROR, `Error uploading ${backupFilePath}. Message: ${error.message} HTTP Code: ${error.code}`));
|
||||
}
|
||||
|
||||
debug(`Uploaded ${backupFilePath} with partSize ${partSize}: ${JSON.stringify(data)}`);
|
||||
debug(`Uploaded ${backupFilePath} with partSize ${partSize}: ${JSON.stringify(data)}`);
|
||||
|
||||
callback(null);
|
||||
});
|
||||
callback(null);
|
||||
});
|
||||
}
|
||||
|
||||
@@ -147,38 +145,36 @@ function exists(apiConfig, backupFilePath, callback) {
|
||||
assert.strictEqual(typeof backupFilePath, 'string');
|
||||
assert.strictEqual(typeof callback, 'function');
|
||||
|
||||
getS3Config(apiConfig, function (error, credentials) {
|
||||
if (error) return callback(error);
|
||||
const credentials = getS3Config(apiConfig);
|
||||
|
||||
const s3 = new aws.S3(_.omit(credentials, 'retryDelayOptions', 'maxRetries'));
|
||||
const s3 = new aws.S3(_.omit(credentials, 'retryDelayOptions', 'maxRetries'));
|
||||
|
||||
if (!backupFilePath.endsWith('/')) { // check for file
|
||||
const params = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: backupFilePath
|
||||
};
|
||||
if (!backupFilePath.endsWith('/')) { // check for file
|
||||
const params = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: backupFilePath
|
||||
};
|
||||
|
||||
s3.headObject(params, function (error) {
|
||||
if (!Object.keys(this.httpResponse.headers).some(h => h.startsWith('x-amz'))) return callback(new BoxError(BoxError.EXTERNAL_ERROR, 'not a s3 endpoint'));
|
||||
if (error && S3_NOT_FOUND(error)) return callback(null, false);
|
||||
if (error) return callback(new BoxError(BoxError.EXTERNAL_ERROR, `Error headObject ${backupFilePath}. Message: ${error.message} HTTP Code: ${error.code}`));
|
||||
s3.headObject(params, function (error) {
|
||||
if (!Object.keys(this.httpResponse.headers).some(h => h.startsWith('x-amz'))) return callback(new BoxError(BoxError.EXTERNAL_ERROR, 'not a s3 endpoint'));
|
||||
if (error && S3_NOT_FOUND(error)) return callback(null, false);
|
||||
if (error) return callback(new BoxError(BoxError.EXTERNAL_ERROR, `Error headObject ${backupFilePath}. Message: ${error.message} HTTP Code: ${error.code}`));
|
||||
|
||||
callback(null, true);
|
||||
});
|
||||
} else { // list dir contents
|
||||
const listParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Prefix: backupFilePath,
|
||||
MaxKeys: 1
|
||||
};
|
||||
callback(null, true);
|
||||
});
|
||||
} else { // list dir contents
|
||||
const listParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Prefix: backupFilePath,
|
||||
MaxKeys: 1
|
||||
};
|
||||
|
||||
s3.listObjects(listParams, function (error, listData) {
|
||||
if (error) return callback(new BoxError(BoxError.EXTERNAL_ERROR, `Error listing objects ${backupFilePath}. Message: ${error.message} HTTP Code: ${error.code}`));
|
||||
s3.listObjects(listParams, function (error, listData) {
|
||||
if (error) return callback(new BoxError(BoxError.EXTERNAL_ERROR, `Error listing objects ${backupFilePath}. Message: ${error.message} HTTP Code: ${error.code}`));
|
||||
|
||||
callback(null, listData.Contents.length !== 0);
|
||||
});
|
||||
}
|
||||
});
|
||||
callback(null, listData.Contents.length !== 0);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
function download(apiConfig, backupFilePath, callback) {
|
||||
@@ -186,32 +182,30 @@ function download(apiConfig, backupFilePath, callback) {
|
||||
assert.strictEqual(typeof backupFilePath, 'string');
|
||||
assert.strictEqual(typeof callback, 'function');
|
||||
|
||||
getS3Config(apiConfig, function (error, credentials) {
|
||||
if (error) return callback(error);
|
||||
const credentials = getS3Config(apiConfig);
|
||||
|
||||
var params = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: backupFilePath
|
||||
};
|
||||
const params = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: backupFilePath
|
||||
};
|
||||
|
||||
var s3 = new aws.S3(credentials);
|
||||
const s3 = new aws.S3(credentials);
|
||||
|
||||
var ps = new PassThrough();
|
||||
var multipartDownload = new S3BlockReadStream(s3, params, { blockSize: 64 * 1024 * 1024 /*, logCallback: debug */ });
|
||||
const ps = new PassThrough();
|
||||
const multipartDownload = new S3BlockReadStream(s3, params, { blockSize: 64 * 1024 * 1024 /*, logCallback: debug */ });
|
||||
|
||||
multipartDownload.on('error', function (error) {
|
||||
if (S3_NOT_FOUND(error)) {
|
||||
ps.emit('error', new BoxError(BoxError.NOT_FOUND, `Backup not found: ${backupFilePath}`));
|
||||
} else {
|
||||
debug(`download: ${apiConfig.bucket}:${backupFilePath} s3 stream error.`, error);
|
||||
ps.emit('error', new BoxError(BoxError.EXTERNAL_ERROR, `Error multipartDownload ${backupFilePath}. Message: ${error.message} HTTP Code: ${error.code}`));
|
||||
}
|
||||
});
|
||||
|
||||
multipartDownload.pipe(ps);
|
||||
|
||||
callback(null, ps);
|
||||
multipartDownload.on('error', function (error) {
|
||||
if (S3_NOT_FOUND(error)) {
|
||||
ps.emit('error', new BoxError(BoxError.NOT_FOUND, `Backup not found: ${backupFilePath}`));
|
||||
} else {
|
||||
debug(`download: ${apiConfig.bucket}:${backupFilePath} s3 stream error.`, error);
|
||||
ps.emit('error', new BoxError(BoxError.EXTERNAL_ERROR, `Error multipartDownload ${backupFilePath}. Message: ${error.message} HTTP Code: ${error.code}`));
|
||||
}
|
||||
});
|
||||
|
||||
multipartDownload.pipe(ps);
|
||||
|
||||
callback(null, ps);
|
||||
}
|
||||
|
||||
function listDir(apiConfig, dir, batchSize, iteratorCallback, callback) {
|
||||
@@ -221,48 +215,44 @@ function listDir(apiConfig, dir, batchSize, iteratorCallback, callback) {
|
||||
assert.strictEqual(typeof iteratorCallback, 'function');
|
||||
assert.strictEqual(typeof callback, 'function');
|
||||
|
||||
getS3Config(apiConfig, function (error, credentials) {
|
||||
if (error) return callback(error);
|
||||
const credentials = getS3Config(apiConfig);
|
||||
|
||||
var s3 = new aws.S3(credentials);
|
||||
var listParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Prefix: dir,
|
||||
MaxKeys: batchSize
|
||||
};
|
||||
const s3 = new aws.S3(credentials);
|
||||
const listParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Prefix: dir,
|
||||
MaxKeys: batchSize
|
||||
};
|
||||
|
||||
let done = false;
|
||||
let done = false;
|
||||
|
||||
async.whilst((testDone) => testDone(null, !done), function listAndDownload(whilstCallback) {
|
||||
s3.listObjects(listParams, function (error, listData) {
|
||||
if (error) return whilstCallback(new BoxError(BoxError.EXTERNAL_ERROR, `Error listing objects in ${dir}. Message: ${error.message} HTTP Code: ${error.code}`));
|
||||
async.whilst((testDone) => testDone(null, !done), function listAndDownload(whilstCallback) {
|
||||
s3.listObjects(listParams, function (error, listData) {
|
||||
if (error) return whilstCallback(new BoxError(BoxError.EXTERNAL_ERROR, `Error listing objects in ${dir}. Message: ${error.message} HTTP Code: ${error.code}`));
|
||||
|
||||
if (listData.Contents.length === 0) { done = true; return whilstCallback(); }
|
||||
if (listData.Contents.length === 0) { done = true; return whilstCallback(); }
|
||||
|
||||
const entries = listData.Contents.map(function (c) { return { fullPath: c.Key, size: c.Size }; });
|
||||
const entries = listData.Contents.map(function (c) { return { fullPath: c.Key, size: c.Size }; });
|
||||
|
||||
iteratorCallback(entries, function (error) {
|
||||
if (error) return whilstCallback(error);
|
||||
iteratorCallback(entries, function (error) {
|
||||
if (error) return whilstCallback(error);
|
||||
|
||||
if (!listData.IsTruncated) { done = true; return whilstCallback(); }
|
||||
if (!listData.IsTruncated) { done = true; return whilstCallback(); }
|
||||
|
||||
listParams.Marker = listData.Contents[listData.Contents.length - 1].Key; // NextMarker is returned only with delimiter
|
||||
listParams.Marker = listData.Contents[listData.Contents.length - 1].Key; // NextMarker is returned only with delimiter
|
||||
|
||||
whilstCallback();
|
||||
});
|
||||
whilstCallback();
|
||||
});
|
||||
}, callback);
|
||||
});
|
||||
});
|
||||
}, callback);
|
||||
}
|
||||
|
||||
// https://github.com/aws/aws-sdk-js/blob/2b6bcbdec1f274fe931640c1b61ece999aae7a19/lib/util.js#L41
|
||||
// https://github.com/GeorgePhillips/node-s3-url-encode/blob/master/index.js
|
||||
// See aws-sdk-js/issues/1302
|
||||
function encodeCopySource(bucket, path) {
|
||||
var output = encodeURI(path);
|
||||
|
||||
// AWS percent-encodes some extra non-standard characters in a URI
|
||||
output = output.replace(/[+!"#$@&'()*+,:;=?@]/g, function(ch) {
|
||||
const output = encodeURI(path).replace(/[+!"#$@&'()*+,:;=?@]/g, function(ch) {
|
||||
return '%' + ch.charCodeAt(0).toString(16).toUpperCase();
|
||||
});
|
||||
|
||||
@@ -275,110 +265,108 @@ function copy(apiConfig, oldFilePath, newFilePath) {
|
||||
assert.strictEqual(typeof oldFilePath, 'string');
|
||||
assert.strictEqual(typeof newFilePath, 'string');
|
||||
|
||||
var events = new EventEmitter();
|
||||
const events = new EventEmitter();
|
||||
|
||||
function copyFile(entry, iteratorCallback) {
|
||||
getS3Config(apiConfig, function (error, credentials) {
|
||||
if (error) return iteratorCallback(error);
|
||||
const credentials = getS3Config(apiConfig);
|
||||
|
||||
var s3 = new aws.S3(credentials);
|
||||
var relativePath = path.relative(oldFilePath, entry.fullPath);
|
||||
const s3 = new aws.S3(credentials);
|
||||
const relativePath = path.relative(oldFilePath, entry.fullPath);
|
||||
|
||||
function done(error) {
|
||||
if (error) debug(`copy: s3 copy error when copying ${entry.fullPath}: ${error}`);
|
||||
function done(error) {
|
||||
if (error) debug(`copy: s3 copy error when copying ${entry.fullPath}: ${error}`);
|
||||
|
||||
if (error && S3_NOT_FOUND(error)) return iteratorCallback(new BoxError(BoxError.NOT_FOUND, `Old backup not found: ${entry.fullPath}`));
|
||||
if (error) return iteratorCallback(new BoxError(BoxError.EXTERNAL_ERROR, `Error copying ${entry.fullPath} (${entry.size} bytes): ${error.code || ''} ${error}`));
|
||||
if (error && S3_NOT_FOUND(error)) return iteratorCallback(new BoxError(BoxError.NOT_FOUND, `Old backup not found: ${entry.fullPath}`));
|
||||
if (error) return iteratorCallback(new BoxError(BoxError.EXTERNAL_ERROR, `Error copying ${entry.fullPath} (${entry.size} bytes): ${error.code || ''} ${error}`));
|
||||
|
||||
iteratorCallback(null);
|
||||
iteratorCallback(null);
|
||||
}
|
||||
|
||||
const copyParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: path.join(newFilePath, relativePath)
|
||||
};
|
||||
|
||||
// S3 copyObject has a file size limit of 5GB so if we have larger files, we do a multipart copy
|
||||
// Exoscale and B2 take too long to copy 5GB
|
||||
const largeFileLimit = (apiConfig.provider === 'exoscale-sos' || apiConfig.provider === 'backblaze-b2' || apiConfig.provider === 'digitalocean-spaces') ? 1024 * 1024 * 1024 : 5 * 1024 * 1024 * 1024;
|
||||
|
||||
if (entry.size < largeFileLimit) {
|
||||
events.emit('progress', `Copying ${relativePath || oldFilePath}`);
|
||||
|
||||
copyParams.CopySource = encodeCopySource(apiConfig.bucket, entry.fullPath);
|
||||
s3.copyObject(copyParams, done).on('retry', function (response) {
|
||||
events.emit('progress', `Retrying (${response.retryCount+1}) copy of ${relativePath || oldFilePath}. Error: ${response.error} ${response.httpResponse.statusCode}`);
|
||||
// on DO, we get a random 408. these are not retried by the SDK
|
||||
if (response.error) response.error.retryable = true; // https://github.com/aws/aws-sdk-js/issues/412
|
||||
});
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
events.emit('progress', `Copying (multipart) ${relativePath || oldFilePath}`);
|
||||
|
||||
s3.createMultipartUpload(copyParams, function (error, multipart) {
|
||||
if (error) return done(error);
|
||||
|
||||
// Exoscale (96M) was suggested by exoscale. 1GB - rather random size for others
|
||||
const chunkSize = apiConfig.provider === 'exoscale-sos' ? 96 * 1024 * 1024 : 1024 * 1024 * 1024;
|
||||
const uploadId = multipart.UploadId;
|
||||
let uploadedParts = [], ranges = [];
|
||||
|
||||
let cur = 0;
|
||||
while (cur + chunkSize < entry.size) {
|
||||
ranges.push({ startBytes: cur, endBytes: cur + chunkSize - 1 });
|
||||
cur += chunkSize;
|
||||
}
|
||||
ranges.push({ startBytes: cur, endBytes: entry.size-1 });
|
||||
|
||||
var copyParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: path.join(newFilePath, relativePath)
|
||||
};
|
||||
async.eachOfLimit(ranges, 3, function copyChunk(range, index, iteratorDone) {
|
||||
const partCopyParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: path.join(newFilePath, relativePath),
|
||||
CopySource: encodeCopySource(apiConfig.bucket, entry.fullPath), // See aws-sdk-js/issues/1302
|
||||
CopySourceRange: 'bytes=' + range.startBytes + '-' + range.endBytes,
|
||||
PartNumber: index+1,
|
||||
UploadId: uploadId
|
||||
};
|
||||
|
||||
// S3 copyObject has a file size limit of 5GB so if we have larger files, we do a multipart copy
|
||||
// Exoscale and B2 take too long to copy 5GB
|
||||
const largeFileLimit = (apiConfig.provider === 'exoscale-sos' || apiConfig.provider === 'backblaze-b2' || apiConfig.provider === 'digitalocean-spaces') ? 1024 * 1024 * 1024 : 5 * 1024 * 1024 * 1024;
|
||||
events.emit('progress', `Copying part ${partCopyParams.PartNumber} - ${partCopyParams.CopySource} ${partCopyParams.CopySourceRange}`);
|
||||
|
||||
if (entry.size < largeFileLimit) {
|
||||
events.emit('progress', `Copying ${relativePath || oldFilePath}`);
|
||||
s3.uploadPartCopy(partCopyParams, function (error, part) {
|
||||
if (error) return iteratorDone(error);
|
||||
|
||||
copyParams.CopySource = encodeCopySource(apiConfig.bucket, entry.fullPath);
|
||||
s3.copyObject(copyParams, done).on('retry', function (response) {
|
||||
events.emit('progress', `Retrying (${response.retryCount+1}) copy of ${relativePath || oldFilePath}. Error: ${response.error} ${response.httpResponse.statusCode}`);
|
||||
// on DO, we get a random 408. these are not retried by the SDK
|
||||
if (response.error) response.error.retryable = true; // https://github.com/aws/aws-sdk-js/issues/412
|
||||
events.emit('progress', `Copying part ${partCopyParams.PartNumber} - Etag: ${part.CopyPartResult.ETag}`);
|
||||
|
||||
if (!part.CopyPartResult.ETag) return iteratorDone(new Error('Multi-part copy is broken or not implemented by the S3 storage provider'));
|
||||
|
||||
uploadedParts[index] = { ETag: part.CopyPartResult.ETag, PartNumber: partCopyParams.PartNumber };
|
||||
|
||||
iteratorDone();
|
||||
}).on('retry', function (response) {
|
||||
events.emit('progress', `Retrying (${response.retryCount+1}) multipart copy of ${relativePath || oldFilePath}. Error: ${response.error} ${response.httpResponse.statusCode}`);
|
||||
});
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
events.emit('progress', `Copying (multipart) ${relativePath || oldFilePath}`);
|
||||
|
||||
s3.createMultipartUpload(copyParams, function (error, multipart) {
|
||||
if (error) return done(error);
|
||||
|
||||
// Exoscale (96M) was suggested by exoscale. 1GB - rather random size for others
|
||||
const chunkSize = apiConfig.provider === 'exoscale-sos' ? 96 * 1024 * 1024 : 1024 * 1024 * 1024;
|
||||
const uploadId = multipart.UploadId;
|
||||
let uploadedParts = [], ranges = [];
|
||||
|
||||
let cur = 0;
|
||||
while (cur + chunkSize < entry.size) {
|
||||
ranges.push({ startBytes: cur, endBytes: cur + chunkSize - 1 });
|
||||
cur += chunkSize;
|
||||
}, function chunksCopied(error) {
|
||||
if (error) { // we must still recommend the user to set a AbortIncompleteMultipartUpload lifecycle rule
|
||||
const abortParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: path.join(newFilePath, relativePath),
|
||||
UploadId: uploadId
|
||||
};
|
||||
events.emit('progress', `Aborting multipart copy of ${relativePath || oldFilePath}`);
|
||||
return s3.abortMultipartUpload(abortParams, () => done(error)); // ignore any abort errors
|
||||
}
|
||||
ranges.push({ startBytes: cur, endBytes: entry.size-1 });
|
||||
|
||||
async.eachOfLimit(ranges, 3, function copyChunk(range, index, iteratorDone) {
|
||||
const partCopyParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: path.join(newFilePath, relativePath),
|
||||
CopySource: encodeCopySource(apiConfig.bucket, entry.fullPath), // See aws-sdk-js/issues/1302
|
||||
CopySourceRange: 'bytes=' + range.startBytes + '-' + range.endBytes,
|
||||
PartNumber: index+1,
|
||||
UploadId: uploadId
|
||||
};
|
||||
const completeMultipartParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: path.join(newFilePath, relativePath),
|
||||
MultipartUpload: { Parts: uploadedParts },
|
||||
UploadId: uploadId
|
||||
};
|
||||
|
||||
events.emit('progress', `Copying part ${partCopyParams.PartNumber} - ${partCopyParams.CopySource} ${partCopyParams.CopySourceRange}`);
|
||||
events.emit('progress', `Finishing multipart copy - ${completeMultipartParams.Key}`);
|
||||
|
||||
s3.uploadPartCopy(partCopyParams, function (error, part) {
|
||||
if (error) return iteratorDone(error);
|
||||
|
||||
events.emit('progress', `Copying part ${partCopyParams.PartNumber} - Etag: ${part.CopyPartResult.ETag}`);
|
||||
|
||||
if (!part.CopyPartResult.ETag) return iteratorDone(new Error('Multi-part copy is broken or not implemented by the S3 storage provider'));
|
||||
|
||||
uploadedParts[index] = { ETag: part.CopyPartResult.ETag, PartNumber: partCopyParams.PartNumber };
|
||||
|
||||
iteratorDone();
|
||||
}).on('retry', function (response) {
|
||||
events.emit('progress', `Retrying (${response.retryCount+1}) multipart copy of ${relativePath || oldFilePath}. Error: ${response.error} ${response.httpResponse.statusCode}`);
|
||||
});
|
||||
}, function chunksCopied(error) {
|
||||
if (error) { // we must still recommend the user to set a AbortIncompleteMultipartUpload lifecycle rule
|
||||
const abortParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: path.join(newFilePath, relativePath),
|
||||
UploadId: uploadId
|
||||
};
|
||||
events.emit('progress', `Aborting multipart copy of ${relativePath || oldFilePath}`);
|
||||
return s3.abortMultipartUpload(abortParams, () => done(error)); // ignore any abort errors
|
||||
}
|
||||
|
||||
const completeMultipartParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: path.join(newFilePath, relativePath),
|
||||
MultipartUpload: { Parts: uploadedParts },
|
||||
UploadId: uploadId
|
||||
};
|
||||
|
||||
events.emit('progress', `Finishing multipart copy - ${completeMultipartParams.Key}`);
|
||||
|
||||
s3.completeMultipartUpload(completeMultipartParams, done);
|
||||
});
|
||||
s3.completeMultipartUpload(completeMultipartParams, done);
|
||||
});
|
||||
});
|
||||
}
|
||||
@@ -407,24 +395,22 @@ function remove(apiConfig, filename, callback) {
|
||||
assert.strictEqual(typeof filename, 'string');
|
||||
assert.strictEqual(typeof callback, 'function');
|
||||
|
||||
getS3Config(apiConfig, function (error, credentials) {
|
||||
if (error) return callback(error);
|
||||
const credentials = getS3Config(apiConfig);
|
||||
|
||||
var s3 = new aws.S3(credentials);
|
||||
const s3 = new aws.S3(credentials);
|
||||
|
||||
var deleteParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Delete: {
|
||||
Objects: [{ Key: filename }]
|
||||
}
|
||||
};
|
||||
const deleteParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Delete: {
|
||||
Objects: [{ Key: filename }]
|
||||
}
|
||||
};
|
||||
|
||||
// deleteObjects does not return error if key is not found
|
||||
s3.deleteObjects(deleteParams, function (error) {
|
||||
if (error) return callback(new BoxError(BoxError.EXTERNAL_ERROR, `Unable to remove ${deleteParams.Key}. error: ${error.message || error.code}`)); // DO sets 'code'
|
||||
// deleteObjects does not return error if key is not found
|
||||
s3.deleteObjects(deleteParams, function (error) {
|
||||
if (error) return callback(new BoxError(BoxError.EXTERNAL_ERROR, `Unable to remove ${deleteParams.Key}. error: ${error.message || error.code}`)); // DO sets 'code'
|
||||
|
||||
callback(null);
|
||||
});
|
||||
callback(null);
|
||||
});
|
||||
}
|
||||
|
||||
@@ -432,45 +418,43 @@ function removeDir(apiConfig, pathPrefix) {
|
||||
assert.strictEqual(typeof apiConfig, 'object');
|
||||
assert.strictEqual(typeof pathPrefix, 'string');
|
||||
|
||||
var events = new EventEmitter();
|
||||
var total = 0;
|
||||
const events = new EventEmitter();
|
||||
let total = 0;
|
||||
|
||||
getS3Config(apiConfig, function (error, credentials) {
|
||||
if (error) return process.nextTick(() => events.emit('done', error));
|
||||
const credentials = getS3Config(apiConfig);
|
||||
|
||||
var s3 = new aws.S3(credentials);
|
||||
const s3 = new aws.S3(credentials);
|
||||
|
||||
listDir(apiConfig, pathPrefix, 1000, function listDirIterator(entries, done) {
|
||||
total += entries.length;
|
||||
listDir(apiConfig, pathPrefix, 1000, function listDirIterator(entries, done) {
|
||||
total += entries.length;
|
||||
|
||||
const chunkSize = apiConfig.deleteConcurrency || (apiConfig.provider !== 'digitalocean-spaces' ? 1000 : 100); // throttle objects in each request
|
||||
var chunks = chunk(entries, chunkSize);
|
||||
const chunkSize = apiConfig.deleteConcurrency || (apiConfig.provider !== 'digitalocean-spaces' ? 1000 : 100); // throttle objects in each request
|
||||
var chunks = chunk(entries, chunkSize);
|
||||
|
||||
async.eachSeries(chunks, function deleteFiles(objects, iteratorCallback) {
|
||||
var deleteParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Delete: {
|
||||
Objects: objects.map(function (o) { return { Key: o.fullPath }; })
|
||||
}
|
||||
};
|
||||
async.eachSeries(chunks, function deleteFiles(objects, iteratorCallback) {
|
||||
var deleteParams = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Delete: {
|
||||
Objects: objects.map(function (o) { return { Key: o.fullPath }; })
|
||||
}
|
||||
};
|
||||
|
||||
events.emit('progress', `Removing ${objects.length} files from ${objects[0].fullPath} to ${objects[objects.length-1].fullPath}`);
|
||||
events.emit('progress', `Removing ${objects.length} files from ${objects[0].fullPath} to ${objects[objects.length-1].fullPath}`);
|
||||
|
||||
// deleteObjects does not return error if key is not found
|
||||
s3.deleteObjects(deleteParams, function (error /*, deleteData */) {
|
||||
if (error) {
|
||||
events.emit('progress', `Unable to remove ${deleteParams.Key} ${error.message || error.code}`);
|
||||
return iteratorCallback(new BoxError(BoxError.EXTERNAL_ERROR, `Unable to remove ${deleteParams.Key}. error: ${error.message || error.code}`)); // DO sets 'code'
|
||||
}
|
||||
// deleteObjects does not return error if key is not found
|
||||
s3.deleteObjects(deleteParams, function (error /*, deleteData */) {
|
||||
if (error) {
|
||||
events.emit('progress', `Unable to remove ${deleteParams.Key} ${error.message || error.code}`);
|
||||
return iteratorCallback(new BoxError(BoxError.EXTERNAL_ERROR, `Unable to remove ${deleteParams.Key}. error: ${error.message || error.code}`)); // DO sets 'code'
|
||||
}
|
||||
|
||||
iteratorCallback(null);
|
||||
});
|
||||
}, done);
|
||||
}, function (error) {
|
||||
events.emit('progress', `Removed ${total} files`);
|
||||
iteratorCallback(null);
|
||||
});
|
||||
}, done);
|
||||
}, function (error) {
|
||||
events.emit('progress', `Removed ${total} files`);
|
||||
|
||||
process.nextTick(() => events.emit('done', error));
|
||||
});
|
||||
process.nextTick(() => events.emit('done', error));
|
||||
});
|
||||
|
||||
return events;
|
||||
@@ -505,29 +489,27 @@ function testConfig(apiConfig, callback) {
|
||||
if ('s3ForcePathStyle' in apiConfig && typeof apiConfig.s3ForcePathStyle !== 'boolean') return callback(new BoxError(BoxError.BAD_FIELD, 's3ForcePathStyle must be a boolean'));
|
||||
|
||||
// attempt to upload and delete a file with new credentials
|
||||
getS3Config(apiConfig, function (error, credentials) {
|
||||
if (error) return callback(error);
|
||||
const credentials = getS3Config(apiConfig);
|
||||
|
||||
var params = {
|
||||
const params = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: path.join(apiConfig.prefix, 'cloudron-testfile'),
|
||||
Body: 'testcontent'
|
||||
};
|
||||
|
||||
const s3 = new aws.S3(_.omit(credentials, 'retryDelayOptions', 'maxRetries'));
|
||||
s3.putObject(params, function (error) {
|
||||
if (error) return callback(new BoxError(BoxError.EXTERNAL_ERROR, `Error put object cloudron-testfile. Message: ${error.message} HTTP Code: ${error.code}`));
|
||||
|
||||
const params = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: path.join(apiConfig.prefix, 'cloudron-testfile'),
|
||||
Body: 'testcontent'
|
||||
Key: path.join(apiConfig.prefix, 'cloudron-testfile')
|
||||
};
|
||||
|
||||
var s3 = new aws.S3(_.omit(credentials, 'retryDelayOptions', 'maxRetries'));
|
||||
s3.putObject(params, function (error) {
|
||||
if (error) return callback(new BoxError(BoxError.EXTERNAL_ERROR, `Error put object cloudron-testfile. Message: ${error.message} HTTP Code: ${error.code}`));
|
||||
s3.deleteObject(params, function (error) {
|
||||
if (error) return callback(new BoxError(BoxError.EXTERNAL_ERROR, `Error del object cloudron-testfile. Message: ${error.message} HTTP Code: ${error.code}`));
|
||||
|
||||
var params = {
|
||||
Bucket: apiConfig.bucket,
|
||||
Key: path.join(apiConfig.prefix, 'cloudron-testfile')
|
||||
};
|
||||
|
||||
s3.deleteObject(params, function (error) {
|
||||
if (error) return callback(new BoxError(BoxError.EXTERNAL_ERROR, `Error del object cloudron-testfile. Message: ${error.message} HTTP Code: ${error.code}`));
|
||||
|
||||
callback();
|
||||
});
|
||||
callback();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user