support putting metadata of null versions in backbeat routes

Added support for handling the "null" version id in the putMetadata
backbeat route.

Possible cases:
- null version issued from a non versioned object in a versioned or versioning
suspended bucket. In this case we should only update the master object in MongoDB
- null version issued from a versioning suspended bucket. In this case we should also
only update the master object as versioning suspended objects don't have version objects
- null version issued from a non versioned object in a versioned bucket with a version put
on top of it. In this case the null version gets transformed into a version, we need to specify
the version id to modify it.

Issue: CLDSRV-392

lib/routes/routeBackbeat.js

@@ -494,12 +494,19 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
         const versionId = decodeVersionId(request.query);
         const options = {};
         if (versionId || omVal.replicationInfo.isNFS) {
-            // specify both 'versioning' and 'versionId' to create a "new"
+            if (versionId === 'null') {
-            // version (updating master as well) but with specified
+                const versioningStatus = bucketInfo.getVersioningConfiguration();
-            // versionId
+                const isVersioningSuspended = versioningStatus.Status === 'Suspended';
+                // in versioning suspended mode, objects have an internal version id that
+                // shouldn't be used to get the object as they only have a master key
+                if (objMd.versionId !== undefined && !isVersioningSuspended) {
+                    options.versionId = objMd.versionId;
+                }
+            } else {
                 options.versioning = bucketInfo.isVersioningEnabled();
                 options.versionId = versionId;
             }
+        }
 
         // If the object is from a source bucket without versioning (i.e. NFS),
         // then we want to create a version for the replica object even though

tests/functional/backbeat/putMetadata.js

@@ -0,0 +1,223 @@
+const assert = require('assert');
+const async = require('async');
+const uuid = require('uuid');
+const BucketUtility = require('../aws-node-sdk/lib/utility/bucket-util');
+const { removeAllVersions } = require('../aws-node-sdk/lib/utility/versioning-util');
+const { updateMetadata, runIfMongo } = require('./utils');
+
+const credentials = {
+    accessKey: 'accessKey1',
+    secretKey: 'verySecretKey1',
+};
+
+const destLocation = 'us-east-2';
+
+const objectName = 'key';
+
+runIfMongo('backbeat routes: putMetadata', () => {
+    let bucketUtil;
+    let s3;
+    let bucketName;
+
+    before(() => {
+        bucketUtil = new BucketUtility('account1', { signatureVersion: 'v4' });
+        s3 = bucketUtil.s3;
+    });
+
+    beforeEach(done => {
+        bucketName = `put-metadata-bucket-${uuid.v4()}`;
+        s3.createBucket({ Bucket: bucketName }, done);
+    });
+
+    afterEach(done => async.series([
+        next => removeAllVersions({ Bucket: bucketName }, next),
+        next => s3.deleteBucket({ Bucket: bucketName }, next),
+    ], done));
+
+    function updateMetadataAndAssertState(versionId, expectedObjectCount, cb) {
+        async.series([
+            next => updateMetadata(
+                { bucket: bucketName, objectKey: objectName, versionId, authCredentials: credentials },
+                { storageClass: destLocation },
+                next),
+            next => s3.headObject({ Bucket: bucketName, Key: objectName, VersionId: versionId }, (err, data) => {
+                assert.ifError(err);
+                assert(data.StorageClass, destLocation);
+                return next();
+            }),
+            next => s3.listObjectVersions({ Bucket: bucketName }, (err, data) => {
+                assert.ifError(err);
+                assert.strictEqual(data.Versions.length, expectedObjectCount);
+                assert.strictEqual(data.DeleteMarkers.length, 0);
+                return next();
+            }),
+        ], cb);
+    }
+
+    it('should update the storage class of a non versioned object', done => {
+        async.series([
+            next => s3.putObject({ Bucket: bucketName, Key: objectName }, err => {
+                assert.ifError(err);
+                return next();
+            }),
+            next => updateMetadataAndAssertState(undefined, 1, next),
+        ], done);
+    });
+
+    it('should update the storage class of a versioned object', done => {
+        let versionId;
+        async.series([
+            next => s3.putBucketVersioning({
+                Bucket: bucketName,
+                VersioningConfiguration: { Status: 'Enabled' },
+            }, next),
+            next => s3.putObject({ Bucket: bucketName, Key: objectName }, (err, data) => {
+                assert.ifError(err);
+                versionId = data.VersionId;
+                return next();
+            }),
+            next => updateMetadataAndAssertState(versionId, 1, next),
+        ], done);
+    });
+
+    it('should update the storage class of a non last version object', done => {
+        let versionId;
+        async.series([
+            next => s3.putBucketVersioning({
+                Bucket: bucketName,
+                VersioningConfiguration: { Status: 'Enabled' },
+            }, next),
+            next => s3.putObject({ Bucket: bucketName, Key: objectName }, (err, data) => {
+                assert.ifError(err);
+                versionId = data.VersionId;
+                return next();
+            }),
+            next => s3.putObject({ Bucket: bucketName, Key: objectName }, err => {
+                assert.ifError(err);
+                return next();
+            }),
+            next => updateMetadataAndAssertState(versionId, 2, next),
+        ], done);
+    });
+
+    it('should update the storage class of a non versioned object in a versioned bucket', done => {
+        async.series([
+            next => s3.putObject({ Bucket: bucketName, Key: objectName }, err => {
+                assert.ifError(err);
+                return next();
+            }),
+            next => s3.putBucketVersioning({
+                Bucket: bucketName,
+                VersioningConfiguration: { Status: 'Enabled' },
+            }, next),
+            next => updateMetadataAndAssertState('null', 1, next),
+        ], done);
+    });
+
+    it('should update the storage class of a null version created from non versioned object '
+        + 'in a versioned bucket', done => {
+        async.series([
+            next => s3.putObject({ Bucket: bucketName, Key: objectName }, err => {
+                assert.ifError(err);
+                return next();
+            }),
+            next => s3.putBucketVersioning({
+                Bucket: bucketName,
+                VersioningConfiguration: { Status: 'Enabled' },
+            }, next),
+            next => s3.putObject({ Bucket: bucketName, Key: objectName }, err => {
+                assert.ifError(err);
+                return next();
+            }),
+            next => updateMetadataAndAssertState('null', 2, next),
+        ], done);
+    });
+
+    it('should update the storage class of a null version in a versioned bucket', done => {
+        async.series([
+            next => s3.putBucketVersioning({
+                Bucket: bucketName,
+                VersioningConfiguration: { Status: 'Suspended' },
+            }, next),
+            next => s3.putObject({ Bucket: bucketName, Key: objectName }, err => {
+                assert.ifError(err);
+                return next();
+            }),
+            next => s3.putBucketVersioning({
+                Bucket: bucketName,
+                VersioningConfiguration: { Status: 'Enabled' },
+            }, next),
+            next => updateMetadataAndAssertState('null', 1, next),
+        ], done);
+    });
+
+    it('should update the storage class of a null non last version in a versioned bucket', done => {
+        async.series([
+            next => s3.putBucketVersioning({
+                Bucket: bucketName,
+                VersioningConfiguration: { Status: 'Suspended' },
+            }, next),
+            next => s3.putObject({ Bucket: bucketName, Key: objectName }, err => {
+                assert.ifError(err);
+                return next();
+            }),
+            next => s3.putBucketVersioning({
+                Bucket: bucketName,
+                VersioningConfiguration: { Status: 'Enabled' },
+            }, next),
+            next => s3.putObject({ Bucket: bucketName, Key: objectName }, err => {
+                assert.ifError(err);
+                return next();
+            }),
+            next => updateMetadataAndAssertState('null', 2, next),
+        ], done);
+    });
+
+    it('should update the storage class of a null version in a versioning suspended bucket', done => {
+        async.series([
+            next => s3.putBucketVersioning({
+                Bucket: bucketName,
+                VersioningConfiguration: { Status: 'Suspended' },
+            }, next),
+            next => s3.putObject({ Bucket: bucketName, Key: objectName }, err => {
+                assert.ifError(err);
+                return next();
+            }),
+            next => updateMetadataAndAssertState('null', 1, next),
+        ], done);
+    });
+
+    it('should update the storage class of a non versioned object in a versioning suspended bucket', done => {
+        async.series([
+            next => s3.putObject({ Bucket: bucketName, Key: objectName }, err => {
+                assert.ifError(err);
+                return next();
+            }),
+            next => s3.putBucketVersioning({
+                Bucket: bucketName,
+                VersioningConfiguration: { Status: 'Suspended' },
+            }, next),
+            next => updateMetadataAndAssertState('null', 1, next),
+        ], done);
+    });
+
+    it('should update the storage class of a version in a versioning suspended bucket', done => {
+        let versionId;
+        async.series([
+            next => s3.putBucketVersioning({
+                Bucket: bucketName,
+                VersioningConfiguration: { Status: 'Enabled' },
+            }, next),
+            next => s3.putObject({ Bucket: bucketName, Key: objectName }, (err, data) => {
+                assert.ifError(err);
+                versionId = data.VersionId;
+                return next();
+            }),
+            next => s3.putBucketVersioning({
+                Bucket: bucketName,
+                VersioningConfiguration: { Status: 'Suspended' },
+            }, next),
+            next => updateMetadataAndAssertState(versionId, 1, next),
+        ], done);
+    });
+});

tests/functional/backbeat/utils.js

@@ -8,7 +8,7 @@ const { models: { ObjectMD } } = require('arsenal');
 // as it allows the tests to be compatible with S3C metadata.
 function updateMetadata(params, toUpdate, cb) {
     const { bucket, objectKey, versionId, authCredentials } = params;
-    const { dataStoreName } = toUpdate;
+    const { dataStoreName, storageClass } = toUpdate;
     const options = {
         authCredentials,
         hostname: ipAddress,
@@ -38,6 +38,11 @@ function updateMetadata(params, toUpdate, cb) {
         if (dataStoreName) {
             result.setDataStoreName(dataStoreName);
         }
+
+        if (storageClass) {
+            result.setAmzStorageClass(storageClass);
+        }
+
         const options = {
             authCredentials,
             hostname: ipAddress,