Integrate Pull Replication's clean read into cloudserver

.github/workflows/tests.yaml

@@ -376,6 +376,7 @@ jobs:
       S3KMS: file
       S3_LOCATION_FILE: /usr/src/app/tests/locationConfig/locationConfigTests.json
       DEFAULT_BUCKET_KEY_FORMAT: v0
+      S3_CLEAN_READ_ENABLED: 'true'
       MONGODB_IMAGE: ghcr.io/${{ github.repository }}/ci-mongodb:${{ github.sha }}
       CLOUDSERVER_IMAGE: ghcr.io/${{ github.repository }}:${{ github.sha }}-testcoverage
       JOB_NAME: ${{ github.job }}
@@ -436,6 +437,7 @@ jobs:
       S3_LOCATION_FILE: /usr/src/app/tests/locationConfig/locationConfigTests.json
       S3_VERSION_ID_ENCODING_TYPE: base62
       DEFAULT_BUCKET_KEY_FORMAT: v1
+      S3_CLEAN_READ_ENABLED: 'true'
       METADATA_MAX_CACHED_BUCKETS: 1
       MONGODB_IMAGE: ghcr.io/${{ github.repository }}/ci-mongodb:${{ github.sha }}
       CLOUDSERVER_IMAGE: ghcr.io/${{ github.repository }}:${{ github.sha }}-testcoverage

lib/Config.js

@@ -584,6 +584,32 @@ function parseServerAccessLogs(config) {
     return res;
 }
 
+/**
+ * Parse the clean-read activation.
+ *
+ * Clean read hides the non-localized object versions from the clients of a
+ * clean-room deployment: data still on the D/R source site, metadata replicated
+ * but not copied locally yet.
+ *
+ * - activated per deployment through S3_CLEAN_READ_ENABLED, defaults to false;
+ * - only on the user-facing CloudServer of a clean room, never on the internal
+ *   one Backbeat talks to, which must see every entry;
+ * - passed to the MetadataWrapper, which sets it on every read and listing call;
+ * - not implemented by Metadata: the flag is then ignored and logged, so leaving
+ *   it unset changes nothing anywhere else.
+ *
+ * @return {boolean} whether clean read is enabled
+ */
+function parseCleanRead() {
+    const envSchema = joi
+        .object({
+            S3_CLEAN_READ_ENABLED: joi.boolean().default(false),
+        })
+        .unknown(true);
+
+    return joi.attempt(process.env, envSchema, 'bad config').S3_CLEAN_READ_ENABLED;
+}
+
 /**
  * Parse the `integrityChecks` config section.
  *
@@ -1888,6 +1914,7 @@ class Config extends EventEmitter {
             }
         }
         this.integrityChecks = parseIntegrityChecks(config);
+        this.cleanRead = parseCleanRead();
         this.serverAccessLogs = parseServerAccessLogs(config);
         /**
          * S3C-10336: PutObject max size of 5GB is new in 9.5.1
@@ -2269,4 +2296,5 @@ module.exports = {
     azureGetLocationCredentials,
     parseSupportedLifecycleRules,
     parseIntegrityChecks,
+    parseCleanRead,
 };

lib/metadata/wrapper.js

@@ -32,6 +32,8 @@ if (clientName === 'mem') {
         replicationGroupId: config.replicationGroupId,
         instanceId: config.instanceId,
         config,
+        locations: config.locationConstraints,
+        hideNonLocalizedVersions: config.cleanRead,
     };
 } else if (clientName === 'cdmi') {
     params = {

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#baa462fc",
         "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,136 @@
+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 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' && process.env.S3_CLEAN_READ_ENABLED === 'true' ? describe : describe.skip;
+
+describeIfCleanRead('clean read', function testSuite() {
+    this.timeout(600000);
+
+    withV4(sigCfg => {
+        let s3;
+        let localizedVersionId;
+        let nonLocalizedVersionId;
+
+        // 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();
+            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 () => {
+            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/unit/Config.js

@@ -8,6 +8,7 @@ const {
     locationConstraintAssert,
     parseSupportedLifecycleRules,
     parseIntegrityChecks,
+    parseCleanRead,
     ConfigObject,
 } = require('../../lib/Config');
 
@@ -937,6 +938,28 @@ describe('Config', () => {
         });
     });
 
+    describe('parse clean read', () => {
+        beforeEach(() => {
+            deleteEnv('S3_CLEAN_READ_ENABLED');
+        });
+
+        it('should default to disabled', () => {
+            assert.strictEqual(parseCleanRead(), false);
+        });
+
+        it('should be activated by the environment variable', () => {
+            setEnv('S3_CLEAN_READ_ENABLED', 'true');
+            assert.strictEqual(parseCleanRead(), true);
+            setEnv('S3_CLEAN_READ_ENABLED', 'false');
+            assert.strictEqual(parseCleanRead(), false);
+        });
+
+        it('should throw if the environment variable is not a boolean', () => {
+            setEnv('S3_CLEAN_READ_ENABLED', 'yes please');
+            assert.throws(() => parseCleanRead(), /must be a boolean/);
+        });
+    });
+
     describe('parse integrity checks', () => {
         // CI exports S3_INTEGRITY_CHECKS_ENABLED for some jobs, so clear both
         // vars rather than assert around whatever the environment inherited.

yarn.lock

@@ -5838,9 +5838,9 @@ arraybuffer.prototype.slice@^1.0.4:
   optionalDependencies:
     ioctl "^2.0.2"
 
-"arsenal@git+https://github.com/scality/arsenal#8.5.15":
+"arsenal@git+https://github.com/scality/arsenal#baa462fc":
   version "8.5.15"
-  resolved "git+https://github.com/scality/arsenal#0bbe970dd72b2e235c47910474883af9b9c13eb1"
+  resolved "git+https://github.com/scality/arsenal#baa462fc4e81de3917f501582b66686305124e28"
   dependencies:
     "@aws-sdk/client-kms" "^3.975.0"
     "@aws-sdk/client-s3" "^3.975.0"