From 03f228763090eb80b4394fd1bc210b762c758077 Mon Sep 17 00:00:00 2001 From: sylvain senechal Date: Thu, 17 Sep 2026 16:52:32 +0200 Subject: [PATCH] Integrate Pull Replication's clean read into cloudserver Issue: CLDSRV-957 --- lib/metadata/metadataUtils.js | 14 +- lib/metadata/wrapper.js | 5 +- lib/routes/routeBackbeat.js | 3 + package.json | 2 +- .../aws-node-sdk/test/versioning/cleanRead.js | 154 ++++++++++++++++ .../backbeat/cleanReadLocalization.js | 170 ++++++++++++++++++ yarn.lock | 6 +- 7 files changed, 347 insertions(+), 7 deletions(-) create mode 100644 tests/functional/aws-node-sdk/test/versioning/cleanRead.js create mode 100644 tests/functional/backbeat/cleanReadLocalization.js diff --git a/lib/metadata/metadataUtils.js b/lib/metadata/metadataUtils.js index 35c6e01de6..b1c7fb53b3 100644 --- a/lib/metadata/metadataUtils.js +++ b/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, diff --git a/lib/metadata/wrapper.js b/lib/metadata/wrapper.js index 6bd800e60f..4ab80df427 100644 --- a/lib/metadata/wrapper.js +++ b/lib/metadata/wrapper.js @@ -32,6 +32,8 @@ if (clientName === 'mem') { replicationGroupId: config.replicationGroupId, instanceId: config.instanceId, config, + locations: config.locationConstraints, + hideNonLocalizedVersions: true, }; } else if (clientName === 'cdmi') { params = { @@ -39,6 +41,5 @@ if (clientName === 'mem') { }; } -const metadata = new MetadataWrapper(config.backends.metadata, params, - bucketclient, logger); +const metadata = new MetadataWrapper(config.backends.metadata, params, bucketclient, logger); module.exports = metadata; diff --git a/lib/routes/routeBackbeat.js b/lib/routes/routeBackbeat.js index 2da9e289a5..613768610b 100644 --- a/lib/routes/routeBackbeat.js +++ b/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, diff --git a/package.json b/package.json index 5fb4b68aea..082d23d139 100644 --- a/package.json +++ b/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", diff --git a/tests/functional/aws-node-sdk/test/versioning/cleanRead.js b/tests/functional/aws-node-sdk/test/versioning/cleanRead.js new file mode 100644 index 0000000000..aa1fe9fb02 --- /dev/null +++ b/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; + }, + ); + }); + }); +}); diff --git a/tests/functional/backbeat/cleanReadLocalization.js b/tests/functional/backbeat/cleanReadLocalization.js new file mode 100644 index 0000000000..523728d1ef --- /dev/null +++ b/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); + }); +}); diff --git a/yarn.lock b/yarn.lock index 854e838c90..a12b4b4cea 100644 --- a/yarn.lock +++ b/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": - version "8.5.15" - resolved "git+https://github.com/scality/arsenal#0bbe970dd72b2e235c47910474883af9b9c13eb1" +"arsenal@git+https://github.com/scality/arsenal#c6745fd3": + version "8.5.17" + resolved "git+https://github.com/scality/arsenal#c6745fd37694b1a8b248af9427251ef5dbaf3560" dependencies: "@aws-sdk/client-kms" "^3.975.0" "@aws-sdk/client-s3" "^3.975.0"