BB-721: Properly get location in lifecycle metrics

extensions/lifecycle/bucketProcessor/policy.json

@@ -6,6 +6,7 @@
             "Effect": "Allow",
             "Action": [
                 "s3:GetLifecycleConfiguration",
+                "s3:GetBucketLocation",
                 "s3:GetBucketVersioning",
                 "s3:ListBucket",
                 "s3:ListBucketVersions",

extensions/lifecycle/tasks/LifecycleDeleteObjectTask.js

@@ -5,6 +5,7 @@ const { HeadObjectCommand, AbortMultipartUploadCommand, DeleteObjectCommand } =
 
 const BackbeatTask = require('../../../lib/tasks/BackbeatTask');
 const { LifecycleMetrics } = require('../LifecycleMetrics');
+const { resolveLifecycleMetricObjectLocation } = require('../util/lifecycleLocation');
 const { 
     DeleteObjectFromExpirationCommand,
     attachReqUids,
@@ -195,7 +196,10 @@ class LifecycleDeleteObjectTask extends BackbeatTask {
 
         const actionType = entry.getActionType();
         const transitionTime = entry.getAttribute('transitionTime');
-        const location = this.objectMD?.dataStoreName || entry.getAttribute('details.dataStoreName');
+        const location = resolveLifecycleMetricObjectLocation(
+            this.objectMD,
+            entry.getAttribute('details.dataStoreName')
+        );
 
         let reqMethod = 'deleteObject';
         let actionMethod = this._deleteObject.bind(this);

extensions/lifecycle/tasks/LifecycleTask.js

@@ -15,6 +15,7 @@ const {
     GetObjectTaggingCommand,
     HeadObjectCommand,
     GetBucketVersioningCommand,
+    GetBucketLocationCommand,
 } = require('@aws-sdk/client-s3');
 const { attachReqUids } = require('@scality/cloudserverclient');
 const config = require('../../../lib/Config');
@@ -24,6 +25,11 @@ const ReplicationAPI = require('../../replication/ReplicationAPI');
 const { LifecycleMetrics, LIFECYCLE_MARKER_METRICS_LOCATION } = require('../LifecycleMetrics');
 const locationsConfig = require('../../../conf/locationConfig.json') || {};
 const { rulesSupportTransition } = require('../util/rules');
+const {
+    isRealLocation,
+    shouldResolveLifecycleMetricLocation,
+    resolveLifecycleMetricObjectLocation,
+} = require('../util/lifecycleLocation');
 const { decode } = versioning.VersionID;
 
 const errorTransitionInProgress = errors.InternalError.
@@ -58,6 +64,19 @@ const MAX_RETRIES = 4;
 // parallel tasks, so the total delay of retries should about 1m30s.
 const MAX_RETRIES_TOTAL = CONCURRENCY_DEFAULT * MAX_RETRIES * 10;
 
+function attachArchiveInfoHeader(command) {
+    command.middlewareStack.add(next => async args => {
+        if (args.request && args.request.headers) {
+            // eslint-disable-next-line no-param-reassign
+            args.request.headers['x-amz-scal-archive-info'] = 'true';
+        }
+        return next(args);
+    }, {
+        step: 'build',
+        name: 'attachArchiveInfoHeader',
+    });
+}
+
 /**
  * compare 2 version by their stale dates returning:
  * - LT (-1) if v1 is less than v2
@@ -104,6 +123,7 @@ class LifecycleTask extends BackbeatTask {
 
         this.setSupportedRules(this.supportedRules);
         this._totalRetries = 0;
+        this._bucketLocationCache = new Map();
     }
 
     setSupportedRules(supportedRules) {
@@ -197,30 +217,176 @@ class LifecycleTask extends BackbeatTask {
      * @return {undefined}
      */
     _sendObjectAction(entry, cb) {
-        const location = entry.getAttribute('details.dataStoreName');
+        const initialLocation = entry.getAttribute('details.dataStoreName');
 
-        const shouldBreak = this.circuitBreakers.tripped(
-            'expiration',
-            location,
-            this.objectTasksTopic,
-        );
-        if (shouldBreak) {
-            process.nextTick(() => cb(errorCircuitBreakerTripped));
-            return;
+        this._resolveLifecycleMetricLocation(entry, initialLocation, this.log, (err, location) => {
+            if (err) {
+                return cb(err);
+            }
+            const metricLocation = location || '';
+            entry.setAttribute('details.dataStoreName', metricLocation);
+
+            const shouldBreak = this.circuitBreakers.tripped(
+                'expiration',
+                metricLocation,
+                this.objectTasksTopic,
+            );
+            if (shouldBreak) {
+                return process.nextTick(() => cb(errorCircuitBreakerTripped));
+            }
+
+            LifecycleMetrics.onLifecycleTriggered(this.log, 'bucket',
+                entry.getActionType() === 'deleteMPU' ? 'expiration:mpu' : 'expiration',
+                metricLocation,
+                Date.now() - entry.getAttribute('transitionTime'));
+
+            const entries = [{ message: entry.toKafkaMessage() }];
+            return this.producer.sendToTopic(this.objectTasksTopic, entries,  err => {
+                LifecycleMetrics.onKafkaPublish(null, 'ObjectTopic', 'bucket', err, 1);
+                return cb(err);
+            });
+        });
+    }
+
+    _getBucketLocationConstraint(bucket, log, cb) {
+        if (this._bucketLocationCache.has(bucket)) {
+            return process.nextTick(cb, null, this._bucketLocationCache.get(bucket));
         }
 
-        LifecycleMetrics.onLifecycleTriggered(this.log, 'bucket',
-            entry.getActionType() === 'deleteMPU' ? 'expiration:mpu' : 'expiration',
-            location,
-            Date.now() - entry.getAttribute('transitionTime'));
+        if (!this.s3target) {
+            return process.nextTick(cb);
+        }
 
-        const entries = [{ message: entry.toKafkaMessage() }];
-        this.producer.sendToTopic(this.objectTasksTopic, entries,  err => {
-            LifecycleMetrics.onKafkaPublish(null, 'ObjectTopic', 'bucket', err, 1);
-            return cb(err);
+        const command = new GetBucketLocationCommand({ Bucket: bucket });
+        attachReqUids(command, log.getSerializedUids());
+        return this.s3target.send(command)
+            .then(data => {
+                LifecycleMetrics.onS3Request(log, 'getBucketLocation', 'bucket', null);
+                const location = data && data.LocationConstraint;
+                this._bucketLocationCache.set(bucket, location);
+                return cb(null, location);
+            })
+            .catch(err => {
+                LifecycleMetrics.onS3Request(log, 'getBucketLocation', 'bucket', err);
+                log.debug('failed to get bucket location for lifecycle metrics', {
+                    method: 'LifecycleTask._getBucketLocationConstraint',
+                    bucket,
+                    error: err.message,
+                });
+                return cb();
+            });
+    }
+
+    _resolveLifecycleMetricLocationFromBucket(entry, fallbackLocation, log, cb) {
+        if (!shouldResolveLifecycleMetricLocation(fallbackLocation)) {
+            return process.nextTick(cb, null, fallbackLocation);
+        }
+
+        const bucket = entry.getAttribute('target.bucket');
+        return this._getBucketLocationConstraint(bucket, log, (err, bucketLocation) => {
+            if (err || !isRealLocation(bucketLocation)) {
+                return cb(null, fallbackLocation);
+            }
+            return cb(null, bucketLocation);
+        });
+    }
+
+    _getArchiveInfoLocation(entry, log, cb) {
+        if (!this.s3target) {
+            return process.nextTick(cb);
+        }
+
+        const params = {
+            Bucket: entry.getAttribute('target.bucket'),
+            Key: entry.getAttribute('target.key'),
+        };
+        const versionId = entry.getAttribute('target.version');
+        if (versionId) {
+            params.VersionId = versionId;
+        }
+
+        const command = new HeadObjectCommand(params);
+        attachArchiveInfoHeader(command);
+        attachReqUids(command, log.getSerializedUids());
+        return this.s3target.send(command)
+            .then(data => {
+                LifecycleMetrics.onS3Request(log, 'headObjectArchiveInfo', 'bucket', null);
+                return cb(null, data.StorageClass);
+            })
+            .catch(err => {
+                LifecycleMetrics.onS3Request(log, 'headObjectArchiveInfo', 'bucket', err);
+                log.debug('failed to get object archive info for lifecycle metric location', {
+                    method: 'LifecycleTask._getArchiveInfoLocation',
+                    bucket: params.Bucket,
+                    objectKey: params.Key,
+                    versionId,
+                    error: err.message,
+                });
+                return cb();
+            });
+    }
+
+    _resolveLifecycleMetricLocationFromArchiveInfo(entry, fallbackLocation, log, cb) {
+        if (!shouldResolveLifecycleMetricLocation(fallbackLocation)) {
+            return process.nextTick(cb, null, fallbackLocation);
+        }
+
+        return this._getArchiveInfoLocation(entry, log, (err, archiveInfoLocation) => {
+            if (err || !isRealLocation(archiveInfoLocation)) {
+                return this._resolveLifecycleMetricLocationFromBucket(entry, fallbackLocation, log, cb);
+            }
+            return cb(null, archiveInfoLocation);
         });
     }
 
+    _resolveLifecycleMetricLocationFromMetadata(entry, fallbackLocation, log, cb) {
+        if (!this.backbeatMetadataProxy) {
+            return this._resolveLifecycleMetricLocationFromArchiveInfo(entry, fallbackLocation, log, cb);
+        }
+
+        if (entry.getActionType() === 'deleteMPU') {
+            return this._resolveLifecycleMetricLocationFromBucket(entry, fallbackLocation, log, cb);
+        }
+
+        const params = {
+            bucket: entry.getAttribute('target.bucket'),
+            objectKey: entry.getAttribute('target.key'),
+            versionId: entry.getAttribute('target.version'),
+        };
+
+        return this._getObjectMD(params, log, (err, objectMD) => {
+            LifecycleMetrics.onS3Request(log, 'getMetadata', 'bucket', err);
+            if (err) {
+                log.debug('failed to get object metadata for lifecycle metric location', {
+                    method: 'LifecycleTask._resolveLifecycleMetricLocationFromMetadata',
+                    bucket: params.bucket,
+                    objectKey: params.objectKey,
+                    versionId: params.versionId,
+                    error: err.message,
+                });
+                return this._resolveLifecycleMetricLocationFromArchiveInfo(entry, fallbackLocation, log, cb);
+            }
+
+            const metadataLocation = resolveLifecycleMetricObjectLocation(objectMD, fallbackLocation);
+            if (!shouldResolveLifecycleMetricLocation(metadataLocation)) {
+                return cb(null, metadataLocation);
+            }
+            return this._resolveLifecycleMetricLocationFromArchiveInfo(entry, metadataLocation, log, cb);
+        }, false);
+    }
+
+    _resolveLifecycleMetricLocation(entry, fallbackLocation, log, cb) {
+        if (!shouldResolveLifecycleMetricLocation(fallbackLocation)) {
+            return process.nextTick(cb, null, fallbackLocation);
+        }
+
+        if (entry.getActionType() === 'deleteMPU') {
+            return this._resolveLifecycleMetricLocationFromBucket(entry, fallbackLocation, log, cb);
+        }
+
+        return this._resolveLifecycleMetricLocationFromMetadata(entry, fallbackLocation, log, cb);
+    }
+
     /**
      * Handles non-versioned objects
      * @param {object} bucketData - bucket data
@@ -1103,16 +1269,18 @@ class LifecycleTask extends BackbeatTask {
         return false;
     }
 
-    _getObjectMD(params, log, cb) {
+    _getObjectMD(params, log, cb, logError = true) {
         this.backbeatMetadataProxy.getMetadata(params, log, (err, blob) => {
             if (err) {
-                log.error('failed to get object metadata', {
-                    method: 'LifecycleTask._getObjectMD',
-                    error: err,
-                    bucket: params.bucket,
-                    objectKey: params.objectKey,
-                    versionId: params.versionId,
-                });
+                if (logError) {
+                    log.error('failed to get object metadata', {
+                        method: 'LifecycleTask._getObjectMD',
+                        error: err,
+                        bucket: params.bucket,
+                        objectKey: params.objectKey,
+                        versionId: params.versionId,
+                    });
+                }
                 return cb(err);
             }
             const { error, result } = ObjectMD.createFromBlob(blob.Body);
@@ -1580,12 +1748,16 @@ class LifecycleTask extends BackbeatTask {
             params.VersionId = obj.VersionId;
         }
         const command = new HeadObjectCommand(params);
+        attachArchiveInfoHeader(command);
         attachReqUids(command, log.getSerializedUids());
         return this.s3target.send(command)
             .then(data => {
                 LifecycleMetrics.onS3Request(log, 'headObject', 'bucket', null);
                 const object = Object.assign({}, obj,
-                    { LastModified: data.LastModified });
+                    {
+                        LastModified: data.LastModified,
+                        StorageClass: data.StorageClass || obj.StorageClass,
+                    });
 
                 // There is an order of importance in cases of conflicts
                 // Expiration and NoncurrentVersionExpiration should be priority

extensions/lifecycle/util/lifecycleLocation.js

@@ -0,0 +1,50 @@
+'use strict';
+
+const STANDARD_LOCATION = 'STANDARD';
+
+function _getObjectMDValue(objectMD, getterName, fieldName) {
+    if (!objectMD) {
+        return undefined;
+    }
+    if (typeof objectMD[getterName] === 'function') {
+        return objectMD[getterName]();
+    }
+    return objectMD[fieldName];
+}
+
+function isRealLocation(location) {
+    return !!location && location !== STANDARD_LOCATION;
+}
+
+function shouldResolveLifecycleMetricLocation(location) {
+    return !isRealLocation(location);
+}
+
+function resolveLifecycleMetricObjectLocation(objectMD, fallbackLocation) {
+    const archive = _getObjectMDValue(objectMD, 'getArchive', 'archive');
+    const amzStorageClass = _getObjectMDValue(objectMD, 'getAmzStorageClass', 'x-amz-storage-class');
+    if (archive && isRealLocation(amzStorageClass)) {
+        return amzStorageClass;
+    }
+
+    const dataStoreName = _getObjectMDValue(objectMD, 'getDataStoreName', 'dataStoreName');
+    if (isRealLocation(dataStoreName)) {
+        return dataStoreName;
+    }
+
+    const locations = _getObjectMDValue(objectMD, 'getLocation', 'location');
+    const locationDataStoreName = Array.isArray(locations) && locations[0]
+        && locations[0].dataStoreName;
+    if (isRealLocation(locationDataStoreName)) {
+        return locationDataStoreName;
+    }
+
+    return fallbackLocation;
+}
+
+module.exports = {
+    STANDARD_LOCATION,
+    isRealLocation,
+    shouldResolveLifecycleMetricLocation,
+    resolveLifecycleMetricObjectLocation,
+};

tests/unit/lifecycle/LifecycleBucketProcessorPolicy.spec.js

@@ -0,0 +1,13 @@
+const assert = require('assert');
+
+const bucketProcessorPolicy = require('../../../extensions/lifecycle/bucketProcessor/policy.json');
+
+describe('LifecycleBucketProcessor policy', () => {
+    it('should allow S3 actions required for lifecycle metric location resolution', () => {
+        const actions = bucketProcessorPolicy.Statement
+            .find(statement => statement.Sid === 'LifecycleExpirationBucketProcessor')
+            .Action;
+
+        assert(actions.includes('s3:GetBucketLocation'));
+    });
+});