diff --git a/extensions/gc/tasks/GarbageCollectorTask.js b/extensions/gc/tasks/GarbageCollectorTask.js index dae27c580..8cb294f87 100644 --- a/extensions/gc/tasks/GarbageCollectorTask.js +++ b/extensions/gc/tasks/GarbageCollectorTask.js @@ -286,10 +286,14 @@ class GarbageCollectorTask extends BackbeatTask { version, }); + // The object already exposes the location it is transitioned to, so this is a + // direct-to-cold transition. + const isDirectToCold = objMD.getAmzStorageClass() === newLocation; + objMD.setLocation() .setDataStoreName(newLocation) .setAmzStorageClass(newLocation) - .setOriginOp('s3:LifecycleTransition') + .setOriginOp(isDirectToCold ? 's3:LifecycleTransition:Direct' : 's3:LifecycleTransition') .setTransitionInProgress(false) .setUserMetadata({ 'x-amz-meta-scal-s3-transition-attempt': undefined, diff --git a/extensions/lifecycle/LifecycleQueuePopulator.js b/extensions/lifecycle/LifecycleQueuePopulator.js index 756e5c3b7..bac52c43c 100644 --- a/extensions/lifecycle/LifecycleQueuePopulator.js +++ b/extensions/lifecycle/LifecycleQueuePopulator.js @@ -20,6 +20,7 @@ const { coldStorageRestoreAdjustTopicPrefix, coldStorageRestoreTopicPrefix, coldStorageGCTopicPrefix, + coldStorageArchiveTopicPrefix, } = config.extensions.lifecycle; const BackbeatProducer = require('../../lib/BackbeatProducer'); const locations = require('../../conf/locationConfig.json') || {}; @@ -100,6 +101,7 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { next => this._setupProducer(`${coldStorageRestoreAdjustTopicPrefix}${location}`, next), next => this._setupProducer(`${coldStorageRestoreTopicPrefix}${location}`, next), next => this._setupProducer(`${coldStorageGCTopicPrefix}${location}`, next), + next => this._setupProducer(`${coldStorageArchiveTopicPrefix}${location}`, next), ], done); }, cb); } else { @@ -239,23 +241,142 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { return new Date(date.$date || date); } - _handleRestoreOp(entry) { + _isColdLocation(locationName) { + return !!this.locationConfigs[locationName]?.isCold; + } + + /** + * Whether a put entry designates an object we may act upon: bucket entries carry no object + * key, mpu shadow bucket entries are internal, and the master entry of a versioned object + * duplicates the version entry, which is processed on its own. + * + * @param {Object} entry - The record log entry from metadata. + * @param {Object} value - The object metadata, already decoded by the caller. + * @return {boolean} true if the entry may be dispatched to an object handler. + */ + _isDistinctObjectEntry(entry, value) { + if (!entry.key || entry.key.startsWith(mpuBucketPrefix)) { + return false; + } + + if (this._isVersionedObject(value) && isMasterKey(entry.key)) { + this.log.trace('skip processing of object master entry'); + return false; + } + + return true; + } + + /** + * Check if object transition was initiated by direct-to-cold request instead of lifecycle rule. + * + * @param {Object} md - The object metadata. + * @return {boolean} true if a direct transition is pending for this object. + */ + _isDirectToCold(md) { + return md['x-amz-scal-transition-in-progress'] + && this._isColdLocation(md['x-amz-storage-class']) + && !this._isColdLocation(md.dataStoreName) + && !md.archive?.archiveInfo; + } + + /** + * Handle a "direct-to-cold" transition: cloudserver has written the object data to a hot + * location, but the user requested a cold storage class in the PUT request. Cloudserver + * flags the object as "transition in progress", and it is up to the queue populator to + * trigger the archival, as there is no lifecycle rule (nor lifecycle scan) involved. + * + * The message published here is strictly the same as the one the lifecycle bucket processor + * publishes (c.f. ReplicationAPI.sendDataMoverAction), so that the whole downstream pipeline + * (Sorbet, cold status processor, garbage collector) is reused unchanged. + * + * @param {Object} entry - The record log entry from metadata. + * @param {Object} [value] - The object metadata, already decoded by the caller. + * @return {undefined} + */ + _handleTransitionOp(entry, value) { if (!this.vaultClientWrapper) { return; } - if (entry.type !== 'put' || - entry.key.startsWith(mpuBucketPrefix)) { + if (!this._isDistinctObjectEntry(entry, value)) { return; } - const value = JSON.parse(entry.value); + if (!this._isDirectToCold(value)) { + return; + } + + const coldLocation = value['x-amz-storage-class']; + const attemptHeader = value['x-amz-meta-scal-s3-transition-attempt']; + const attempt = attemptHeader ? Number.parseInt(attemptHeader, 10) : undefined; + const ownerId = value['owner-id']; + this.vaultClientWrapper.getAccountId(ownerId, (err, accountId) => { + if (err) { + this.log.error('unable to get account', { + method: 'LifecycleQueuePopulator._handleTransitionOp', + ownerId, + err, + }); + return; + } + + this.log.trace( + 'publishing object transition entry', + { bucket: entry.bucket, key: entry.key, version: value.versionId, coldLocation }, + ); + + const topic = `${coldStorageArchiveTopicPrefix}${coldLocation}`; + const key = `${entry.bucket}/${value.key}`; + + let version; + if (value.versionId) { + version = encode(value.versionId); + } - const operation = value.originOp; - // supporting both 's3:ObjectRestore' and 's3:ObjectRestore:Post' to keep - // compatibility with older cloudserver versions, the switch to 's3:ObjectRestore:Post' - // was made to have the correct event type for bucket notifications - if (!['s3:ObjectRestore', 's3:ObjectRestore:Post', 's3:ObjectRestore:Retry'].includes(operation)) { + const transitionTime = this._parseDate( + value['x-amz-scal-transition-time'] || value['last-modified']); + const message = JSON.stringify({ + accountId, + bucketName: entry.bucket, + objectKey: value.key, + objectVersion: version, + requestId: uuid(), + size: value['content-length'], + eTag: `"${value['content-md5']}"`, + try: attempt, + transitionTime: transitionTime.toISOString(), + }); + + const producer = this._producers[topic]; + if (producer) { + LifecycleMetrics.onLifecycleTriggered(this.log, 'queuePopulator', 'archive', + coldLocation, Date.now() - transitionTime.getTime()); + + const kafkaEntry = { key, message }; + producer.send([kafkaEntry], err => { + LifecycleMetrics.onKafkaPublish(this.log, 'ColdStorageArchiveTopic', 'queuePopulator', err, 1); + if (err) { + this.log.error('error publishing object transition request entry', { + error: err, + method: 'LifecycleQueuePopulator._handleTransitionOp', + }); + } + }); + } else { + this.log.error(`producer not available for location ${coldLocation}`, { + method: 'LifecycleQueuePopulator._handleTransitionOp', + }); + } + }); + } + + _handleRestoreOp(entry, value) { + if (!this.vaultClientWrapper) { + return; + } + + if (!this._isDistinctObjectEntry(entry, value)) { return; } @@ -283,13 +404,6 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { return; } - // if entry is a versioned object and is the master entry, skip task as - // the non-master entry will be processed - if (this._isVersionedObject(value) && isMasterKey(entry.key)) { - this.log.trace('skip processing of object master entry'); - return; - } - // We would need to provide the object's bucket's account id as part of the kafka entry. // This account id would be used by Sorbet to assume the bucket's account role. // The assumed credentials will be sent and used by TLP server to put object version @@ -504,7 +618,42 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { return undefined; } - this._handleRestoreOp(entry); + // The entry is decoded once here, then dispatched to the single handler which may be + // interested in it: most entries are of no interest to any of them. + const { error, result: value } = safeJsonParse(entry.value); + if (error) { + this.log.error('could not parse log entry', { + method: 'LifecycleQueuePopulator.filter', + bucket: entry.bucket, + key: entry.key, + error, + }); + return undefined; + } + + switch (value.originOp) { + // supporting both 's3:ObjectRestore' and 's3:ObjectRestore:Post' to keep compatibility with + // older cloudserver versions, the switch to 's3:ObjectRestore:Post' was made to have the + // correct event type for bucket notifications + case 's3:ObjectRestore': + case 's3:ObjectRestore:Post': + case 's3:ObjectRestore:Retry': + this._handleRestoreOp(entry, value); + break; + + // Object creation is the only operation which may declare a cold storage class, and a retry + // asks for a new attempt: any other originOp comes from backbeat itself, and must not + // re-trigger a transition. + case 's3:ObjectCreated:Put': + case 's3:ObjectCreated:CompleteMultipartUpload': + case 's3:ObjectCreated:Copy': + case 's3:LifecycleTransition:Retry': + this._handleTransitionOp(entry, value); + break; + + default: + break; + } if (this.extConfig.conductor.bucketSource !== 'zookeeper') { this.log.debug('bucket source is not zookeeper, skipping entry', { @@ -515,15 +664,7 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { let bucketValue = {}; if (this._isBucketEntryFromBucketd(entry)) { - const parsedEntry = safeJsonParse(entry.value); - if (parsedEntry.error) { - this.log.error('could not parse raft log entry', { - value: entry.value, - error: parsedEntry.error, - }); - return undefined; - } - const parsedAttr = safeJsonParse(parsedEntry.result.attributes); + const parsedAttr = safeJsonParse(value.attributes); if (parsedAttr.error) { this.log.error('could not parse raft log entry attribute', { value: entry.value, @@ -533,13 +674,7 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { } bucketValue = parsedAttr.result; } else if (this._isBucketEntryFromFileMD(entry)) { - const { error, result } = safeJsonParse(entry.value); - if (error) { - this.log.error('could not parse file md log entry', - { value: entry.value, error }); - return undefined; - } - bucketValue = result; + bucketValue = value; } else { // not a bucket entry return undefined; diff --git a/extensions/lifecycle/tasks/LifecycleColdStatusArchiveTask.js b/extensions/lifecycle/tasks/LifecycleColdStatusArchiveTask.js index 8cea3c62b..9bf574a3d 100644 --- a/extensions/lifecycle/tasks/LifecycleColdStatusArchiveTask.js +++ b/extensions/lifecycle/tasks/LifecycleColdStatusArchiveTask.js @@ -112,10 +112,14 @@ class LifecycleColdStatusArchiveTask extends LifecycleUpdateTransitionTask { objectMD.setOriginOp('s3:LifecycleTransition:SetArchive'); if (skipLocationDeletion) { + // The object already exposes the location it is transitioned to, so this + // is a direct-to-cold transition. + const isDirectToCold = objectMD.getAmzStorageClass() === coldLocation; + objectMD.setDataStoreName(coldLocation) .setAmzStorageClass(coldLocation) .setTransitionInProgress(false) - .setOriginOp('s3:LifecycleTransition') + .setOriginOp(isDirectToCold ? 's3:LifecycleTransition:Direct' : 's3:LifecycleTransition') .setUserMetadata({ 'x-amz-meta-scal-s3-transition-attempt': undefined, }); diff --git a/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js b/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js index 8768d577f..2507e47b9 100644 --- a/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js +++ b/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js @@ -1,6 +1,7 @@ 'use strict'; const { LifecycleRequeueTask } = require('./LifecycleRequeueTask'); +const locationsConfig = require('../../../conf/locationConfig.json') || {}; class LifecycleResetTransitionInProgressTask extends LifecycleRequeueTask { /** @@ -18,13 +19,27 @@ class LifecycleResetTransitionInProgressTask extends LifecycleRequeueTask { return false; } md.setOriginOp('s3:LifecycleTransition:Retry'); - md.setTransitionInProgress(false); + if (!this._isDirectToCold(md)) { + // Keep the flag as the queue populator keys on it to trigger the next attempt + md.setTransitionInProgress(false); + } md.setUserMetadata({ 'x-amz-meta-scal-s3-transition-attempt': try_, }); return true; } + /** + * Check if object transition was initiated by direct-to-cold request instead of lifecycle rule. + * + * @param {ObjectMD} md - object metadata + * @return {boolean} true if this is a pending direct transition + */ + _isDirectToCold(md) { + return locationsConfig[md.getAmzStorageClass()]?.isCold + && !locationsConfig[md.getDataStoreName()]?.isCold; + } + shouldSkipObject(md, expectedEtag, log) { try { const etag = JSON.parse(expectedEtag); diff --git a/extensions/lifecycle/tasks/LifecycleTask.js b/extensions/lifecycle/tasks/LifecycleTask.js index 8a8311840..0160bbe4b 100644 --- a/extensions/lifecycle/tasks/LifecycleTask.js +++ b/extensions/lifecycle/tasks/LifecycleTask.js @@ -31,6 +31,8 @@ const errorTransitionInProgress = errors.InternalError. customizeDescription('transition is currently in progress'); const errorTransitionColdObject = errors.InternalError. customizeDescription('transitioning a cold object is forbidden'); +const errorTransitionDeclaredColdObject = errors.InternalError. + customizeDescription('transitioning an object declared as cold is forbidden'); const errorObjectTemporarilyRestored = errors.InternalError. customizeDescription('object temporarily restored'); const errorReplicationInProgress = errors.InternalError. @@ -1276,6 +1278,11 @@ class LifecycleTask extends BackbeatTask { if (isObjectCold) { return next(errorTransitionColdObject); } + // Skip direct-to-cold objects, whose transition is triggered by the queue + // populator and require the transition in progress flag to be set + if (locationsConfig[objectMD.getAmzStorageClass()]?.isCold) { + return next(errorTransitionDeclaredColdObject); + } // If transition is in progress, do not re-publish entry // to data-mover or cold-archive topic. if (objectMD.getTransitionInProgress()) { diff --git a/tests/unit/gc/GarbageCollectorTask.spec.js b/tests/unit/gc/GarbageCollectorTask.spec.js index 3cd4d7fdb..b39b2de9d 100644 --- a/tests/unit/gc/GarbageCollectorTask.spec.js +++ b/tests/unit/gc/GarbageCollectorTask.spec.js @@ -99,10 +99,47 @@ describe('GarbageCollectorTask', () => { assert.strictEqual(updatedMD.getDataStoreName(), 'new-location'); assert.strictEqual(updatedMD.getAmzStorageClass(), 'new-location'); assert.strictEqual(updatedMD.getTransitionInProgress(), false); + assert.strictEqual(updatedMD.getOriginOp(), 's3:LifecycleTransition'); done(); }); }); + it('should set the direct transition origin op if the new location was requested', done => { + backbeatClient.batchDeleteResponse = { error: null, res: null }; + + const entry = ActionQueueEntry.create('deleteArchivedSourceData') + .addContext({ + origin: 'lifecycle', + ruleType: 'archive', + bucketName: bucket, + objectKey: key, + versionId: version, + }) + .setAttribute('serviceName', 'lifecycle-transition') + .setAttribute('target.oldLocation', 'old-location') + .setAttribute('target.newLocation', 'new-location') + .setAttribute('target.bucket', bucket) + .setAttribute('target.key', version) + .setAttribute('target.version', key) + .setAttribute('target.accountId', accountId) + .setAttribute('target.owner', owner); + + mdObj.setLocation(loc) + .setDataStoreName('old-location') + .setAmzStorageClass('new-location') + .setTransitionInProgress(true); + backbeatMetadataProxyClient.setMdObj(mdObj); + + gcTask.processActionEntry(entry, err => { + assert.ifError(err); + + const updatedMD = backbeatMetadataProxyClient.mdObj; + assert.strictEqual(updatedMD.getDataStoreName(), 'new-location'); + assert.strictEqual(updatedMD.getTransitionInProgress(), false); + assert.strictEqual(updatedMD.getOriginOp(), 's3:LifecycleTransition:Direct'); + done(); + }); + }); it('should delete archived location info if gc failed with 404', done => { backbeatClient.batchDeleteResponse = { error: { statusCode: 404 }, res: null }; diff --git a/tests/unit/lifecycle/LifecycleColdStatusArchiveTask.spec.js b/tests/unit/lifecycle/LifecycleColdStatusArchiveTask.spec.js index cfc7127ce..0966cb8f2 100644 --- a/tests/unit/lifecycle/LifecycleColdStatusArchiveTask.spec.js +++ b/tests/unit/lifecycle/LifecycleColdStatusArchiveTask.spec.js @@ -117,6 +117,7 @@ describe('LifecycleColdStatusArchiveTask', () => { assert.strictEqual(gcEntry, null); assert.strictEqual(updatedMD.dataStoreName, 'cold'); assert.strictEqual(updatedMD['x-amz-storage-class'], 'cold'); + assert.strictEqual(updatedMD.originOp, 's3:LifecycleTransition'); assert.deepStrictEqual(updatedMD.archive.archiveInfo, { archiveId: 'da80b6dc-280d-4dce-83b5-d5b40276e321', archiveVersion: 5166759712787974, @@ -144,6 +145,26 @@ describe('LifecycleColdStatusArchiveTask', () => { }); }); + it('should set the direct transition origin op if the cold class was requested', done => { + backbeatClient.batchDeleteResponse = { error: { statusCode: 404 }, res: null }; + + const entry = ColdStorageStatusQueueEntry.createFromKafkaEntry({ value: message }); + mdObj.setLocation() + .setDataStoreName('us-east-1') + .setAmzStorageClass(coldLocation) + .setArchive(null); + backbeatMetadataProxyClient.setMdObj(mdObj); + + archiveTask.processEntry(coldLocation, entry, err => { + assert.ifError(err); + + const updatedMD = backbeatMetadataProxyClient.getReceivedMd(); + assert.strictEqual(updatedMD.dataStoreName, 'cold'); + assert.strictEqual(updatedMD.originOp, 's3:LifecycleTransition:Direct'); + done(); + }); + }); + it('should send kafka entry to delete orphan cold object when source object was deleted', done => { const entry = ColdStorageStatusQueueEntry.createFromKafkaEntry({ value: message }); diff --git a/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js b/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js index 8cd79202c..72bf49625 100644 --- a/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js +++ b/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js @@ -7,7 +7,8 @@ const config = require('../../../lib/Config'); const { coldStorageRestoreAdjustTopicPrefix, coldStorageRestoreTopicPrefix, - coldStorageGCTopicPrefix + coldStorageGCTopicPrefix, + coldStorageArchiveTopicPrefix } = config.extensions.lifecycle; const LifecycleQueuePopulator = require('../../../extensions/lifecycle/LifecycleQueuePopulator'); @@ -78,6 +79,10 @@ const templateEntry = { 'versionId': '98500086134471999999RG001 0', 'isNFS': true, 'archive': { + archiveInfo: { + archiveId: '04425717-a65c-4e8a-95e1-fa1d902d9d9f', + archiveVersion: 7504504064263669, + }, restoreRequestedAt: Date.now(), restoreRequestedDays: 1, }, @@ -131,16 +136,17 @@ describe('LifecycleQueuePopulator', () => { done(); }); }); - it('should have three producers per cold location', done => { + it('should have four producers per cold location', done => { lcqp.locationConfigs = Object.assign({}, locationConfigs, coldLocationConfigs); lcqp.setupProducers(() => { const producers = Object.keys(lcqp._producers); const coldLocations = Object.keys(coldLocationConfigs); - assert.strictEqual(producers.length, coldLocations.length * 3); + assert.strictEqual(producers.length, coldLocations.length * 4); coldLocations.forEach(loc => { assert(producers.includes(`${coldStorageRestoreAdjustTopicPrefix}${loc}`)); assert(producers.includes(`${coldStorageRestoreTopicPrefix}${loc}`)); assert(producers.includes(`${coldStorageGCTopicPrefix}${loc}`)); + assert(producers.includes(`${coldStorageArchiveTopicPrefix}${loc}`)); }); done(); }); @@ -149,6 +155,7 @@ describe('LifecycleQueuePopulator', () => { describe(':_handleRestoreOp', () => { let lcqp; + const handleRestoreOp = entry => lcqp._handleRestoreOp(entry, JSON.parse(entry.value)); const getAccountIdStub = sinon.stub().yields(null, '79a59df900b949e55d96a1e698fbacedfd6e09d98eacf8f8d5218e7cd47ef2be'); beforeEach(() => { @@ -161,37 +168,6 @@ describe('LifecycleQueuePopulator', () => { afterEach(() => { sinon.restore(); }); - [ - { - event: 's3:ObjectRestore', - ignore: false, - }, - { - event: 's3:ObjectRestore:Post', - ignore: false, - }, - { - event: 's3:ObjectRestore:Retry', - ignore: false, - }, - { - event: 's3:ObjectCreated:Put', - ignore: true, - }, - ].forEach(params => { - const outcome = params.ignore ? 'ignore' : 'consider'; - it(`should ${outcome} ${params.event} event`, () => { - const getAccountIdStub = sinon.stub().yields(null, - '79a59df900b949e55d96a1e698fbacedfd6e09d98eacf8f8d5218e7cd47ef2be'); - lcqp.vaultClientWrapper = { - getAccountId: getAccountIdStub, - }; - const entry = getKafkaEntry(params.event); - lcqp._handleRestoreOp(entry); - assert.strictEqual(getAccountIdStub.calledOnce, !params.ignore); - }); - }); - describe('restore requests', () => { const kafkaSendStub = sinon.stub().yields(); const kafkaAdjustSendStub = sinon.stub().yields(); @@ -262,7 +238,7 @@ describe('LifecycleQueuePopulator', () => { value: JSON.stringify(objMd), }; - lcqp._handleRestoreOp(entry); + handleRestoreOp(entry); assert(!kafkaAdjustSendStub.calledOnce); assert(kafkaSendStub.calledOnce); @@ -317,7 +293,7 @@ describe('LifecycleQueuePopulator', () => { value: JSON.stringify(objMd), }; - lcqp._handleRestoreOp(entry); + handleRestoreOp(entry); assert(kafkaAdjustSendStub.calledOnce); assert(!kafkaSendStub.calledOnce); @@ -369,11 +345,240 @@ describe('LifecycleQueuePopulator', () => { value: JSON.stringify(objMd), }; - lcqp._handleRestoreOp(entry); + handleRestoreOp(entry); assert(!kafkaAdjustSendStub.calledOnce); assert(!kafkaSendStub.calledOnce); }); + + it('should skip send duration-adjust for the master entry of a versioned object', () => { + const versionId = '98500086134471999999RG001 0'; + const objMd = { + 'md-model-version': 2, + 'owner-display-name': 'Bart', + 'owner-id': '79a59df900b949e55d96a1e698fbacedfd6e09d98eacf8f8d5218e7cd47ef2be', + 'x-amz-storage-class': 'dmf-v1', + 'content-length': 542, + 'content-type': 'text/plain', + 'last-modified': '2017-07-13T02:44:25.515Z', + 'content-md5': '01064f35c238bd2b785e34508c3d27f4', + 'key': 'object', + 'location': [], + 'isDeleteMarker': false, + 'isNull': false, + versionId, + 'archive': { + archiveInfo: { + archiveId: '04425717-a65c-4e8a-95e1-fa1d902d9d9f', + archiveVersion: 7504504064263669 + }, + restoreCompletedAt: '2017-07-13T02:44:25.519Z', + restoreWillExpireAt: '2017-07-15T02:44:25.519Z', + }, + 'dataStoreName': 'dmf-v1', + 'originOp': 's3:ObjectRestore:Post', + }; + // the version entry carries the same update, and is processed on its own + const entry = { + type: 'put', + bucket: 'lc-queue-populator-test-bucket', + key: 'object', + value: JSON.stringify(objMd), + }; + + handleRestoreOp(entry); + + assert(!kafkaAdjustSendStub.called); + assert(!kafkaSendStub.called); + + handleRestoreOp(Object.assign({}, entry, { key: `object\x00${versionId}` })); + + assert(kafkaAdjustSendStub.calledOnce); + assert(!kafkaSendStub.called); + }); + }); + }); + + describe(':_handleTransitionOp', () => { + const accountId = '79a59df900b949e55d96a1e698fbacedfd6e09d98eacf8f8d5218e7cd47ef2be'; + const versionId = '98500086134471999999RG001 0'; + const archiveTopic = `${coldStorageArchiveTopicPrefix}dmf-v1`; + + let lcqp; + let getAccountIdStub; + let kafkaSendStub; + + const handleTransitionOp = entry => lcqp._handleTransitionOp(entry, JSON.parse(entry.value)); + + function getTransitionEntry(overrides) { + const value = Object.assign({ + 'md-model-version': 2, + 'owner-display-name': 'Bart', + 'owner-id': accountId, + 'content-length': 542, + 'content-type': 'text/plain', + 'last-modified': '2017-07-13T02:44:25.519Z', + 'content-md5': '01064f35c238bd2b785e34508c3d27f4', + 'x-amz-storage-class': 'dmf-v1', + 'x-amz-scal-transition-in-progress': true, + 'x-amz-scal-transition-time': '2017-07-13T02:44:20.000Z', + 'key': 'hosts', + 'location': [], + 'isDeleteMarker': false, + 'isNull': false, + versionId, + 'dataStoreName': 'us-east-1', + 'originOp': 's3:ObjectCreated:Put', + }, overrides); + // allow overrides to remove a field by passing `undefined` + Object.keys(value).forEach(k => { + if (value[k] === undefined) { + delete value[k]; + } + }); + return { + type: 'put', + bucket: 'lc-queue-populator-test-bucket', + key: `hosts\x00${versionId}`, + value: JSON.stringify(value), + }; + } + + beforeEach(() => { + lcqp = new LifecycleQueuePopulator(params); + lcqp.locationConfigs = Object.assign({}, coldLocationConfigs, locationConfigs); + getAccountIdStub = sinon.stub().yields(null, accountId); + lcqp.vaultClientWrapper = { + getAccountId: getAccountIdStub, + }; + kafkaSendStub = sinon.stub().yields(); + lcqp._producers[archiveTopic] = { + send: kafkaSendStub, + }; + }); + + afterEach(() => { + sinon.restore(); + }); + + it('should publish an archive request matching the bucket processor message', () => { + handleTransitionOp(getTransitionEntry()); + + assert(kafkaSendStub.calledOnce); + const kafkaEntry = kafkaSendStub.args[0][0][0]; + assert.strictEqual(kafkaEntry.key, 'lc-queue-populator-test-bucket/hosts'); + + const message = JSON.parse(kafkaEntry.message); + assert.deepStrictEqual(message, { + accountId, + bucketName: 'lc-queue-populator-test-bucket', + objectKey: 'hosts', + objectVersion: encode(versionId), + requestId: message.requestId, + size: 542, + eTag: '"01064f35c238bd2b785e34508c3d27f4"', + transitionTime: '2017-07-13T02:44:20.000Z', + }); + assert(message.requestId); + }); + + it('should fall back on last-modified when no transition time is set', () => { + handleTransitionOp(getTransitionEntry({ + 'x-amz-scal-transition-time': undefined, + })); + + assert(kafkaSendStub.calledOnce); + const message = JSON.parse(kafkaSendStub.args[0][0][0].message); + assert.strictEqual(message.transitionTime, '2017-07-13T02:44:25.519Z'); + }); + + it('should publish the transition attempt count', () => { + handleTransitionOp(getTransitionEntry({ + 'originOp': 's3:LifecycleTransition:Retry', + 'x-amz-meta-scal-s3-transition-attempt': '3', + })); + + assert(kafkaSendStub.calledOnce); + const message = JSON.parse(kafkaSendStub.args[0][0][0].message); + assert.strictEqual(message.try, 3); + }); + + it('should not set objectVersion for a non-versioned object', () => { + const entry = getTransitionEntry({ versionId: undefined }); + entry.key = 'hosts'; + handleTransitionOp(entry); + + assert(kafkaSendStub.calledOnce); + const message = JSON.parse(kafkaSendStub.args[0][0][0].message); + assert.strictEqual(message.objectVersion, undefined); + }); + + [ + { + desc: 'transition is not in progress', + overrides: { 'x-amz-scal-transition-in-progress': undefined }, + }, + { + desc: 'the storage class is not cold', + overrides: { 'x-amz-storage-class': 'us-east-2' }, + }, + { + desc: 'the object has no storage class', + overrides: { 'x-amz-storage-class': undefined }, + }, + { + desc: 'the data is already in the cold location', + overrides: { dataStoreName: 'dmf-v1' }, + }, + { + desc: 'the object is already archived', + overrides: { + archive: { + archiveInfo: { + archiveId: '04425717-a65c-4e8a-95e1-fa1d902d9d9f', + archiveVersion: 7504504064263669, + }, + }, + }, + }, + ].forEach(({ desc, overrides }) => { + it(`should not publish when ${desc}`, () => { + handleTransitionOp(getTransitionEntry(overrides)); + assert(!getAccountIdStub.called); + assert(!kafkaSendStub.called); + }); + }); + + it('should skip the master key of a versioned object', () => { + const entry = getTransitionEntry(); + entry.key = 'hosts'; + handleTransitionOp(entry); + assert(!kafkaSendStub.called); + }); + + it('should skip mpu shadow bucket entries', () => { + const entry = getTransitionEntry(); + entry.key = `mpuShadowBucket${entry.key}`; + handleTransitionOp(entry); + assert(!kafkaSendStub.called); + }); + + it('should do nothing without a vault client', () => { + lcqp.vaultClientWrapper = null; + handleTransitionOp(getTransitionEntry()); + assert(!kafkaSendStub.called); + }); + + it('should not publish when the account cannot be resolved', () => { + getAccountIdStub.yields(errors.InternalError); + handleTransitionOp(getTransitionEntry()); + assert(!kafkaSendStub.called); + }); + + it('should not throw when no producer is available', () => { + delete lcqp._producers[archiveTopic]; + handleTransitionOp(getTransitionEntry()); + assert(getAccountIdStub.calledOnce); }); }); @@ -410,6 +615,30 @@ describe('LifecycleQueuePopulator', () => { assert(handleDeleteStub.calledOnce); }); + [ + { originOp: 's3:ObjectRestore', handler: '_handleRestoreOp' }, + { originOp: 's3:ObjectRestore:Post', handler: '_handleRestoreOp' }, + { originOp: 's3:ObjectRestore:Retry', handler: '_handleRestoreOp' }, + { originOp: 's3:ObjectCreated:Put', handler: '_handleTransitionOp' }, + { originOp: 's3:ObjectCreated:CompleteMultipartUpload', handler: '_handleTransitionOp' }, + { originOp: 's3:ObjectCreated:Copy', handler: '_handleTransitionOp' }, + { originOp: 's3:LifecycleTransition:Retry', handler: '_handleTransitionOp' }, + { originOp: 's3:LifecycleTransition:Start', handler: null }, + { originOp: 's3:LifecycleTransition:SetArchive', handler: null }, + { originOp: 's3:LifecycleTransition:Direct', handler: null }, + { originOp: 's3:LifecycleTransition', handler: null }, + ].forEach(({ originOp, handler }) => { + it(`should dispatch ${originOp} to ${handler || 'no handler'}`, () => { + const restoreStub = sinon.stub(lcqp, '_handleRestoreOp').returns(); + const transitionStub = sinon.stub(lcqp, '_handleTransitionOp').returns(); + + lcqp.filter(getKafkaEntry(originOp)); + + assert.strictEqual(restoreStub.calledOnce, handler === '_handleRestoreOp'); + assert.strictEqual(transitionStub.calledOnce, handler === '_handleTransitionOp'); + }); + }); + it('should not update zookeeper when bucketSource is mongodb (default)', () => { lcqp.extConfig.conductor.bucketSource = 'mongodb'; const putEntry = { diff --git a/tests/unit/lifecycle/LifecycleResetTransitionInProgressTask.spec.js b/tests/unit/lifecycle/LifecycleResetTransitionInProgressTask.spec.js index 4b466174f..af90dbb3f 100644 --- a/tests/unit/lifecycle/LifecycleResetTransitionInProgressTask.spec.js +++ b/tests/unit/lifecycle/LifecycleResetTransitionInProgressTask.spec.js @@ -38,6 +38,18 @@ describe('LifecycleResetTransitionInProgressTask', () => { .setUserMetadata({ 'x-amz-meta-scal-s3-transition-attempt': 11, }); + // "direct" transition: the cold storage class was requested in the PUT request, and the + // data still lies in the hot location + const objectDirectTransitioning = new ObjectMD() + .setContentMd5('etag1') + .setTransitionInProgress(true) + .setAmzStorageClass('location-dmf-v1') + .setDataStoreName('us-east-1'); + const objectDirectTransitioned = new ObjectMD() + .setContentMd5('etag1') + .setTransitionInProgress(true) + .setAmzStorageClass('location-dmf-v1') + .setDataStoreName('location-dmf-v1'); beforeEach(() => { backbeatMetadataProxyClient = new BackbeatMetadataProxyMock(); @@ -90,4 +102,31 @@ describe('LifecycleResetTransitionInProgressTask', () => { done(); }); }); + + it('should keep transition in progress flag for a direct transition', done => { + backbeatMetadataProxyClient.setMdObj(objectDirectTransitioning); + task.processActionEntry(actionEntry, err => { + assert.ifError(err); + + const md = backbeatMetadataProxyClient.mdObj; + assert.ok(md.getTransitionInProgress()); + assert.strictEqual(md.getOriginOp(), 's3:LifecycleTransition:Retry'); + const umd = JSON.parse(md.getUserMetadata()); + assert.strictEqual(umd['x-amz-meta-scal-s3-transition-attempt'], 12); + + done(); + }); + }); + + it('should reset transition in progress flag once the object is in the cold location', done => { + backbeatMetadataProxyClient.setMdObj(objectDirectTransitioned); + task.processActionEntry(actionEntry, err => { + assert.ifError(err); + + const md = backbeatMetadataProxyClient.mdObj; + assert.ok(!md.getTransitionInProgress()); + + done(); + }); + }); }); diff --git a/tests/unit/lifecycle/LifecycleTask.spec.js b/tests/unit/lifecycle/LifecycleTask.spec.js index 7150bbe14..4e9d0c5bc 100644 --- a/tests/unit/lifecycle/LifecycleTask.spec.js +++ b/tests/unit/lifecycle/LifecycleTask.spec.js @@ -12,6 +12,7 @@ const LifecycleTaskV2 = require( '../../../extensions/lifecycle/tasks/LifecycleTaskV2'); const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry'); const { LifecycleMetrics } = require('../../../extensions/lifecycle/LifecycleMetrics'); +const ReplicationAPI = require('../../../extensions/replication/ReplicationAPI'); const fakeLogger = require('../../utils/fakeLogger'); const { withActiveSpan } = require('../../utils/withActiveSpan'); const { timeOptions } = require('../../functional/lifecycle/configObjects'); @@ -2473,6 +2474,67 @@ describe('lifecycle task helper methods', () => { }); }); + describe('_applyTransitionRule', () => { + const testParams = { + bucket: 'test-bucket', + owner: 'test-owner', + objectKey: 'test-key', + site: 'us-east-2', + transitionTime: Date.now(), + }; + + let lifecycleTask; + + beforeEach(() => { + lifecycleTask = new LifecycleTask(lp); + lifecycleTask.pausedLocations = new Set(); + lifecycleTask.circuitBreakers = { tripped: () => false }; + }); + + afterEach(() => { + sinon.restore(); + }); + + function stubObjectMD(overrides) { + const objectMD = Object.assign({ + getReplicationStatus: () => 'COMPLETED', + getDataStoreName: () => 'us-east-1', + getAmzStorageClass: () => 'us-east-1', + getTransitionInProgress: () => false, + getArchive: () => undefined, + setTransitionInProgress: () => {}, + setOriginOp: () => {}, + getSerialized: () => '{}', + }, overrides); + sinon.stub(lifecycleTask, '_getObjectMD').yields(null, objectMD); + } + + it('should not transition an object declared as cold', done => { + stubObjectMD({ getAmzStorageClass: () => 'location-dmf-v1' }); + const getEntryStub = sinon.stub(lifecycleTask, '_getTransitionActionEntry'); + + lifecycleTask._applyTransitionRule(testParams, fakeLogger, err => { + assert.strictEqual(err.description, + 'transitioning an object declared as cold is forbidden'); + assert(!getEntryStub.called); + done(); + }); + }); + + it('should transition an object with a hot storage class', done => { + stubObjectMD(); + const getEntryStub = sinon.stub(lifecycleTask, '_getTransitionActionEntry').yields(null, {}); + sinon.stub(ReplicationAPI, 'sendDataMoverAction').yields(); + sinon.stub(lifecycleTask, '_putObjectMD').yields(); + + lifecycleTask._applyTransitionRule(testParams, fakeLogger, err => { + assert.ifError(err); + assert(getEntryStub.calledOnce); + done(); + }); + }); + }); + describe('_sendObjectAction', () => { it('should emit trigger metrics with the entry location', done => { const lifecycleTask = new LifecycleTask(lp);