BB-786: trigger cold transition from oplog

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,

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}/${entry.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: encodeURIComponent(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;

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,
                         });

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);

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()) {

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 };