Integrate Pull Replication's clean read into cloudserver

lib/metadata/metadataUtils.js

@@ -355,7 +355,16 @@ function checkRateLimitIfNeeded(request, authInfo, bucketMD, log, callback) {
  * @return {undefined} - and call callback with params err, bucket md
  */
 function standardMetadataValidateBucketAndObj(params, actionImplicitDenies, log, callback) {
-    const { authInfo, bucketName, objectKey, versionId, getDeleteMarker, request, withVersionId } = params;
+    const {
+        authInfo,
+        bucketName,
+        objectKey,
+        versionId,
+        getDeleteMarker,
+        request,
+        withVersionId,
+        hideNonLocalizedVersions,
+    } = params;
     let requestType = params.requestType;
     if (!Array.isArray(requestType)) {
         requestType = [requestType];
@@ -372,6 +381,9 @@ function standardMetadataValidateBucketAndObj(params, actionImplicitDenies, log,
                 if (getDeleteMarker) {
                     getOptions.getDeleteMarker = true;
                 }
+                if (hideNonLocalizedVersions !== undefined) {
+                    getOptions.hideNonLocalizedVersions = hideNonLocalizedVersions;
+                }
                 return metadata.getBucketAndObjectMD(
                     bucketName,
                     objectKey,

lib/metadata/wrapper.js

@@ -32,13 +32,14 @@ if (clientName === 'mem') {
         replicationGroupId: config.replicationGroupId,
         instanceId: config.instanceId,
         config,
+        locations: config.locationConstraints,
+        hideNonLocalizedVersions: true,
     };
 } else if (clientName === 'cdmi') {
     params = {
         cdmi: config.cdmi,
     };
 }
 
-const metadata = new MetadataWrapper(config.backends.metadata, params,
-    bucketclient, logger);
+const metadata = new MetadataWrapper(config.backends.metadata, params, bucketclient, logger);
 module.exports = metadata;

lib/routes/routeBackbeat.js

@@ -2056,6 +2056,9 @@ function routeBackbeat(clientIP, request, response, log) {
                     requestType: request.apiMethods || 'ReplicateObject',
                     request,
                 };
+                if (request.method === 'GET' && request.resourceType === 'metadata') {
+                    mdValParams.hideNonLocalizedVersions = false;
+                }
                 return standardMetadataValidateBucketAndObj(
                     mdValParams,
                     request.actionImplicitDenies,

package.json

@@ -35,7 +35,7 @@
         "@opentelemetry/instrumentation-ioredis": "~0.64.0",
         "@opentelemetry/instrumentation-mongodb": "~0.69.0",
         "@smithy/node-http-handler": "^3.0.0",
-        "arsenal": "git+https://github.com/scality/arsenal#8.5.15",
+        "arsenal": "git+https://github.com/scality/arsenal#c6745fd3",
         "async": "2.6.4",
         "aws-crt": "^1.24.0",
         "bucketclient": "scality/bucketclient#8.2.7",

tests/functional/aws-node-sdk/test/versioning/cleanRead.js

@@ -0,0 +1,154 @@
+const assert = require('assert');
+const crypto = require('crypto');
+const { versioning } = require('arsenal');
+const {
+    S3Client,
+    CreateBucketCommand,
+    DeleteBucketCommand,
+    PutBucketVersioningCommand,
+    PutObjectCommand,
+    GetObjectCommand,
+    HeadObjectCommand,
+    ListObjectsV2Command,
+    ListObjectVersionsCommand,
+} = require('@aws-sdk/client-s3');
+
+const withV4 = require('../support/withV4');
+const getConfig = require('../support/config');
+const { config } = require('../../../../../lib/Config');
+const metadata = require('../../../../../lib/metadata/wrapper');
+const { initMetadata, getMetadata } = require('../utils/init');
+const { DummyRequestLogger } = require('../../../../unit/helpers');
+const { removeAllVersions } = require('../../lib/utility/versioning-util');
+const { promisify } = require('util');
+
+const versionIdUtils = versioning.VersionID;
+const log = new DummyRequestLogger();
+const removeAllVersionsAsync = promisify(removeAllVersions);
+
+const bucket = `clean-read-bucket-${Date.now()}`;
+const objectKey = 'clean-read-object';
+const LOCAL_LOCATION = 'us-east-1';
+const LOCALIZED_BODY = 'localized';
+const NON_LOCALIZED_BODY = 'waiting for its data to be copied over';
+
+// clean read is implemented by the mongodb metadata backend only
+const describeIfCleanRead = process.env.S3METADATA === 'mongodb' ? describe : describe.skip;
+
+describeIfCleanRead('clean read', function testSuite() {
+    this.timeout(600000);
+
+    withV4(sigCfg => {
+        let s3;
+        let localizedVersionId;
+        let nonLocalizedVersionId;
+        let nonLocalizedMD;
+        let nonLocalizedDecodedVersionId;
+
+        // Replicates a version the way the clean-room mongo-processor does
+        async function replicateNonLocalizedVersion() {
+            const decodedVersionId = versionIdUtils.generateVersionId(`${process.pid}`, config.replicationGroupId);
+            const objMD = await getMetadata(bucket, objectKey, localizedVersionId);
+            objMD.versionId = decodedVersionId;
+            // the version carries a content of its own, written on the source site
+            objMD['content-length'] = NON_LOCALIZED_BODY.length;
+            objMD['content-md5'] = crypto.createHash('md5').update(NON_LOCALIZED_BODY).digest('hex');
+            // the only location flagged "isCRR" in tests/locationConfig/locationConfigTests.json
+            objMD.dataStoreName = 'location-crr-v1';
+            objMD['last-modified'] = new Date().toJSON();
+            nonLocalizedMD = objMD;
+            nonLocalizedDecodedVersionId = decodedVersionId;
+            await new Promise((resolve, reject) =>
+                metadata.putObjectMD(
+                    bucket,
+                    objectKey,
+                    objMD,
+                    { versionId: decodedVersionId, repairMaster: true },
+                    log,
+                    err => (err ? reject(err) : resolve()),
+                ),
+            );
+            return versionIdUtils.encode(decodedVersionId);
+        }
+
+        before(async () => {
+            s3 = new S3Client(getConfig('default', sigCfg));
+            await initMetadata();
+            await s3.send(new CreateBucketCommand({ Bucket: bucket }));
+            await s3.send(
+                new PutBucketVersioningCommand({
+                    Bucket: bucket,
+                    VersioningConfiguration: { Status: 'Enabled' },
+                }),
+            );
+            const localized = await s3.send(
+                new PutObjectCommand({ Bucket: bucket, Key: objectKey, Body: LOCALIZED_BODY }),
+            );
+            localizedVersionId = localized.VersionId;
+            nonLocalizedVersionId = await replicateNonLocalizedVersion();
+        });
+
+        after(async () => {
+            // removeAllVersions lists through the S3 API, which hides the non-localized
+            // version: localize it first, the way the data mover would, so that the
+            // teardown can see and delete it
+            nonLocalizedMD.dataStoreName = LOCAL_LOCATION;
+            await new Promise((resolve, reject) =>
+                metadata.putObjectMD(
+                    bucket,
+                    objectKey,
+                    nonLocalizedMD,
+                    { versionId: nonLocalizedDecodedVersionId, repairMaster: true },
+                    log,
+                    err => (err ? reject(err) : resolve()),
+                ),
+            );
+            await removeAllVersionsAsync({ Bucket: bucket });
+            await s3.send(new DeleteBucketCommand({ Bucket: bucket }));
+        });
+
+        it('should omit the non-localized version from the version listing', async () => {
+            const res = await s3.send(new ListObjectVersionsCommand({ Bucket: bucket }));
+            assert.deepStrictEqual(
+                (res.Versions || []).map(version => version.VersionId),
+                [localizedVersionId],
+            );
+        });
+
+        it('should list the object, carried by its newest localized version', async () => {
+            const res = await s3.send(new ListObjectsV2Command({ Bucket: bucket }));
+            assert.strictEqual(res.Contents.length, 1);
+            assert.strictEqual(res.Contents[0].Key, objectKey);
+            // the size tells the two versions apart, the master having kept the
+            // localized one rather than following the newer non-localized version
+            assert.strictEqual(res.Contents[0].Size, LOCALIZED_BODY.length);
+        });
+
+        it('should serve the newest localized version as the current object', async () => {
+            const res = await s3.send(new GetObjectCommand({ Bucket: bucket, Key: objectKey }));
+            assert.strictEqual(res.VersionId, localizedVersionId);
+            assert.strictEqual(res.ContentLength, LOCALIZED_BODY.length);
+            assert.strictEqual(await res.Body.transformToString(), LOCALIZED_BODY);
+        });
+
+        it('should reject a get on the non-localized version', async () => {
+            await assert.rejects(
+                s3.send(new GetObjectCommand({ Bucket: bucket, Key: objectKey, VersionId: nonLocalizedVersionId })),
+                err => {
+                    assert.strictEqual(err.name, 'NoSuchVersion');
+                    return true;
+                },
+            );
+        });
+
+        it('should reject a head on the non-localized version', async () => {
+            await assert.rejects(
+                s3.send(new HeadObjectCommand({ Bucket: bucket, Key: objectKey, VersionId: nonLocalizedVersionId })),
+                err => {
+                    assert.strictEqual(err.$metadata.httpStatusCode, 404);
+                    return true;
+                },
+            );
+        });
+    });
+});

tests/functional/backbeat/cleanReadLocalization.js

@@ -0,0 +1,170 @@
+'use strict';
+
+const assert = require('assert');
+const { createHash } = require('crypto');
+const { v4: uuidv4 } = require('uuid');
+const {
+    CreateBucketCommand,
+    PutBucketVersioningCommand,
+    PutObjectCommand,
+    GetObjectCommand,
+    ListObjectVersionsCommand,
+} = require('@aws-sdk/client-s3');
+
+const { versioning } = require('arsenal');
+const BucketUtility = require('../aws-node-sdk/lib/utility/bucket-util');
+
+const { BackbeatRoutesClient, GetMetadataCommand, PutMetadataCommand } = require('@scality/cloudserverclient');
+
+const { generateVersionId, encode: encodeVersionId } = versioning.VersionID;
+
+const TEST_BUCKET = `bucket-cleanread-${uuidv4().split('-')[0]}`;
+const LOCALIZED_BODY = 'localized';
+const CANONICAL_ID = '79a59df900b949e55d96a1e698fbacedfd6e09d98eacf8f8d5218e7cd47ef2be';
+// the only location flagged "isCRR" in tests/locationConfig/locationConfigTests.json
+const SOURCE_LOCATION = 'location-crr-v1';
+const LOCAL_LOCATION = 'us-east-1';
+const REPLICATED_BODY = 'waiting for its data to be copied over';
+
+const bucketUtil = new BucketUtility('default', {});
+const s3 = bucketUtil.s3;
+
+let backbeatClient;
+const replicatedVersions = [];
+
+function buildMetadataBody(versionId, dataStoreName) {
+    return JSON.stringify({
+        'content-length': Buffer.byteLength(REPLICATED_BODY),
+        'content-type': 'text/plain',
+        'last-modified': new Date().toISOString(),
+        'content-md5': createHash('md5').update(REPLICATED_BODY).digest('hex'),
+        'owner-id': CANONICAL_ID,
+        'owner-display-name': 'test',
+        versionId,
+        dataStoreName,
+        location: null,
+        replicationInfo: {
+            status: 'REPLICA',
+            isReplica: true,
+            backends: [],
+            content: [],
+            destination: '',
+            storageClass: '',
+            role: '',
+            storageType: '',
+            dataStoreVersionId: '',
+        },
+    });
+}
+
+// the mongo-processor replicating a version, then the data mover merging it once
+// the data has been copied: same version id, the location rewritten to the local one
+function writeVersion(key, versionId, dataStoreName) {
+    if (dataStoreName === SOURCE_LOCATION) {
+        replicatedVersions.push({ key, versionId });
+    }
+    return backbeatClient.send(
+        new PutMetadataCommand({
+            Bucket: TEST_BUCKET,
+            Key: key,
+            VersionId: encodeVersionId(versionId),
+            Body: buildMetadataBody(versionId, dataStoreName),
+        }),
+    );
+}
+
+async function currentVersion(key) {
+    const res = await s3.send(new GetObjectCommand({ Bucket: TEST_BUCKET, Key: key }));
+    return res.VersionId;
+}
+
+const describeIfCleanRead = process.env.S3METADATA === 'mongodb' ? describe : describe.skip;
+
+describeIfCleanRead('clean read: localizing a replicated version', function testSuite() {
+    this.timeout(120000);
+
+    before(async () => {
+        const creds = await s3.config.credentials();
+        backbeatClient = new BackbeatRoutesClient({
+            endpoint: `http://${process.env.IP || '127.0.0.1'}:8000`,
+            region: 'us-east-1',
+            credentials: {
+                accessKeyId: creds.accessKeyId,
+                secretAccessKey: creds.secretAccessKey,
+            },
+            forcePathStyle: true,
+        });
+        await s3.send(new CreateBucketCommand({ Bucket: TEST_BUCKET }));
+        await s3.send(
+            new PutBucketVersioningCommand({
+                Bucket: TEST_BUCKET,
+                VersioningConfiguration: { Status: 'Enabled' },
+            }),
+        );
+    });
+
+    after(async () => {
+        // the teardown lists through the S3 API, which hides the versions left
+        // non-localized: localize them first, the way the data mover would
+        await Promise.all(replicatedVersions.map(({ key, versionId }) => writeVersion(key, versionId, LOCAL_LOCATION)));
+        await bucketUtil.empty(TEST_BUCKET);
+        await bucketUtil.deleteOne(TEST_BUCKET);
+    });
+
+    it('should hide the replicated version, then serve it once localized', async () => {
+        const key = 'clean-read-localization';
+        const localized = await s3.send(new PutObjectCommand({ Bucket: TEST_BUCKET, Key: key, Body: LOCALIZED_BODY }));
+        const replicatedVersionId = generateVersionId(`${process.pid}`, 'RG001');
+
+        // the version is replicated, its data still on the source site
+        await writeVersion(key, replicatedVersionId, SOURCE_LOCATION);
+
+        // clean read hides it: the object still reads as the older localized version
+        assert.strictEqual(await currentVersion(key), localized.VersionId);
+        const versions = await s3.send(new ListObjectVersionsCommand({ Bucket: TEST_BUCKET, Prefix: key }));
+        assert.deepStrictEqual(
+            (versions.Versions || []).map(v => v.VersionId),
+            [localized.VersionId],
+        );
+
+        // the data mover copies the data and merges the metadata: same version,
+        // now pointing at the local location
+        await writeVersion(key, replicatedVersionId, LOCAL_LOCATION);
+
+        // it becomes the current version: the master was promoted
+        assert.strictEqual(await currentVersion(key), encodeVersionId(replicatedVersionId));
+    });
+
+    it('should let the data mover read the replicated version by its id', async () => {
+        const key = 'clean-read-by-version-id';
+        await s3.send(new PutObjectCommand({ Bucket: TEST_BUCKET, Key: key, Body: LOCALIZED_BODY }));
+        const replicatedVersionId = generateVersionId(`${process.pid}`, 'RG001');
+        await writeVersion(key, replicatedVersionId, SOURCE_LOCATION);
+
+        // hidden from the clients
+        await assert.rejects(
+            s3.send(
+                new GetObjectCommand({
+                    Bucket: TEST_BUCKET,
+                    Key: key,
+                    VersionId: encodeVersionId(replicatedVersionId),
+                }),
+            ),
+            err => {
+                assert.strictEqual(err.name, 'NoSuchVersion');
+                return true;
+            },
+        );
+
+        // but readable through the backbeat route, which is how it gets localized
+        const res = await backbeatClient.send(
+            new GetMetadataCommand({
+                Bucket: TEST_BUCKET,
+                Key: key,
+                VersionId: encodeVersionId(replicatedVersionId),
+            }),
+        );
+        const md = JSON.parse(res.Body);
+        assert.strictEqual(md.dataStoreName, SOURCE_LOCATION);
+    });
+});