Skip to content

Commit c8acd08

Browse files
SylvainSenechalSylvainSenechal
committed
Implement CRR Cascaded feature
Issue: CLDSRV-897
1 parent 8af6bfe commit c8acd08

6 files changed

Lines changed: 552 additions & 26 deletions

File tree

‎lib/api/apiUtils/object/createAndStoreObject.js‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -157,6 +157,7 @@ function createAndStoreObject(bucketName, bucketMD, objectKey, objMD, authInfo,
157157
replicationInfo: getReplicationInfo(config,
158158
objectKey, bucketMD, false, size, null, null, authInfo),
159159
overheadField,
160+
updateMicroVersionId: true,
160161
log,
161162
};
162163

‎lib/routes/routeBackbeat.js‎

Lines changed: 130 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
const { constants: { HTTP_STATUS_CONFLICT } } = require('http2');
12
const url = require('url');
23
const async = require('async');
34
const httpProxy = require('http-proxy');
@@ -7,7 +8,13 @@ const joi = require('@hapi/joi');
78
const backbeatProxy = httpProxy.createProxyServer({
89
ignorePath: true,
910
});
10-
const { auth, errors, errorInstances, s3middleware, s3routes, models, storage } = require('arsenal');
11+
const { auth, errors, errorInstances, s3middleware, s3routes, models, storage, versioning } = require('arsenal');
12+
const { decode, encode, compare: compareMicroVersionId, Ordering } = versioning.VersionID;
13+
const {
14+
VersionIdCollisionException,
15+
StaleMicroVersionIdException,
16+
MicroVersionIdAlreadyStoredException,
17+
} = require('@scality/cloudserverclient');
1118

1219
const { responseJSONBody } = s3routes.routesUtils;
1320
const { getSubPartIds } = s3middleware.azureHelper.mpuUtils;
@@ -21,6 +28,7 @@ const locationStorageCheck = require('../api/apiUtils/object/locationStorageChec
2128
const { dataStore } = require('../api/apiUtils/object/storeObject');
2229
const prepareRequestContexts = require('../api/apiUtils/authorization/prepareRequestContexts');
2330
const { decodeVersionId } = require('../api/apiUtils/object/versioning');
31+
const getReplicationInfo = require('../api/apiUtils/object/getReplicationInfo');
2432
const locationKeysHaveChanged = require('../api/apiUtils/object/locationKeysHaveChanged');
2533
const { standardMetadataValidateBucketAndObj, metadataGetObject } = require('../metadata/metadataUtils');
2634
const { config } = require('../Config');
@@ -32,6 +40,7 @@ const {
3240
} = require('../api/apiUtils/integrity/validateChecksums');
3341
const { BackendInfo } = models;
3442
const { pushReplicationMetric } = require('./utilities/pushReplicationMetric');
43+
const writeContinue = require('../utilities/writeContinue');
3544
const kms = require('../kms/wrapper');
3645
const { listLifecycleCurrents } = require('../api/backbeat/listLifecycleCurrents');
3746
const { listLifecycleNonCurrents } = require('../api/backbeat/listLifecycleNonCurrents');
@@ -93,7 +102,7 @@ function _isObjectRequest(req) {
93102
return ['data', 'metadata', 'multiplebackenddata', 'multiplebackendmetadata'].includes(req.resourceType);
94103
}
95104

96-
function _respondWithHeaders(response, payload, extraHeaders, log, callback) {
105+
function _respondWithHeaders(response, payload, extraHeaders, log, callback, statusCode = 200) {
97106
let body = '';
98107
if (typeof payload === 'string') {
99108
body = payload;
@@ -115,10 +124,10 @@ function _respondWithHeaders(response, payload, extraHeaders, log, callback) {
115124
// eslint-disable-next-line no-param-reassign
116125
response.serverAccessLog.endTurnAroundTime = process.hrtime.bigint();
117126
}
118-
response.writeHead(200, httpHeaders);
127+
response.writeHead(statusCode, httpHeaders);
119128
response.end(body, 'utf8', () => {
120129
log.end().info('responded with payload', {
121-
httpCode: 200,
130+
httpCode: statusCode,
122131
contentLength: Buffer.byteLength(body),
123132
});
124133
callback();
@@ -129,6 +138,15 @@ function _respond(response, payload, log, callback) {
129138
_respondWithHeaders(response, payload, {}, log, callback);
130139
}
131140

141+
function _respondWithHeaderCrrConflict(response, log, callback, code, message, mvId) {
142+
return _respondWithHeaders(
143+
response,
144+
{ code, message },
145+
mvId ? { 'x-scal-micro-version-id': encode(mvId) } : {},
146+
log, callback, HTTP_STATUS_CONFLICT,
147+
);
148+
}
149+
132150
function _getRequestPayload(req, cb) {
133151
const payload = [];
134152
let payloadLen = 0;
@@ -414,6 +432,30 @@ function putData(request, response, bucketInfo, objMd, log, callback) {
414432
log.error(errMessage);
415433
return callback(errorInstances.BadRequest.customizeDescription(errMessage));
416434
}
435+
436+
const incomingVersionIdEncoded = request.headers['x-scal-version-id'];
437+
const decoded = incomingVersionIdEncoded ? decode(incomingVersionIdEncoded) : null;
438+
const incomingVersionIdDecoded = decoded instanceof Error ? null : decoded;
439+
if (incomingVersionIdDecoded && objMd && objMd.versionId === incomingVersionIdDecoded) {
440+
// Skip the write if data is already at destination for this version id
441+
// Return 409 with the existing microVersionId so backbeat can
442+
// decide if putMetadata is still needed
443+
log.debug('crr cascade putData: version already at destination', {
444+
method: 'putData',
445+
bucketName: request.bucketName,
446+
objectKey: request.objectKey,
447+
hasMicroVersionId: !!objMd.microVersionId,
448+
});
449+
request.resume();
450+
return _respondWithHeaderCrrConflict(
451+
response, log, callback,
452+
VersionIdCollisionException.name,
453+
'version id already at destination',
454+
objMd.microVersionId,
455+
);
456+
}
457+
458+
writeContinue(request, response);
417459
const context = {
418460
bucketName: request.bucketName,
419461
owner: canonicalID,
@@ -539,6 +581,42 @@ function getCanonicalIdsByAccountId(accountId, log, cb) {
539581
}
540582

541583
function putMetadata(request, response, bucketInfo, objMd, log, callback) {
584+
const { bucketName, objectKey } = request;
585+
586+
const encodedMicroVersionId = request.headers['x-scal-micro-version-id'];
587+
const decoded = encodedMicroVersionId ? decode(encodedMicroVersionId) : null;
588+
const incomingMicroVersionId = decoded instanceof Error ? null : decoded;
589+
if (incomingMicroVersionId) {
590+
const cmp = compareMicroVersionId(incomingMicroVersionId, objMd?.microVersionId);
591+
if (cmp === Ordering.EQUAL) {
592+
log.debug('crr cascade putMetadata: loop detected, skipping write', {
593+
method: 'putMetadata',
594+
bucketName,
595+
objectKey,
596+
});
597+
request.resume();
598+
return _respondWithHeaderCrrConflict(
599+
response, log, callback,
600+
MicroVersionIdAlreadyStoredException.name,
601+
'incoming microVersionId already at destination',
602+
);
603+
}
604+
if (cmp === Ordering.OLDER) {
605+
log.debug('crr cascade putMetadata: stale event, rejecting', {
606+
method: 'putMetadata',
607+
bucketName,
608+
objectKey,
609+
});
610+
request.resume();
611+
return _respondWithHeaderCrrConflict(
612+
response, log, callback,
613+
StaleMicroVersionIdException.name,
614+
'incoming revision is older than destination',
615+
objMd?.microVersionId,
616+
);
617+
}
618+
}
619+
542620
return _getRequestPayload(request, (err, payload) => {
543621
if (err) {
544622
return callback(err);
@@ -552,15 +630,15 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
552630
return callback(errors.MalformedPOSTRequest);
553631
}
554632

555-
const { headers, bucketName, objectKey } = request;
633+
const { headers } = request;
556634

557635
// Destination-side delete-marker replication.
558636
// We need the REPLICA status to distinguish from
559637
// source-side replication status updates that also carry isDeleteMarker=true.
560638
if (
561639
omVal.isDeleteMarker &&
562640
omVal.replicationInfo &&
563-
omVal.replicationInfo.status === 'REPLICA' &&
641+
(omVal.replicationInfo.isReplica === true || omVal.replicationInfo.status === 'REPLICA') &&
564642
request.serverAccessLog
565643
) {
566644
// eslint-disable-next-line no-param-reassign
@@ -576,7 +654,7 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
576654
// The REPLICA status excludes source-side replication-status updates.
577655
if (
578656
omVal.replicationInfo &&
579-
omVal.replicationInfo.status === 'REPLICA' &&
657+
(omVal.replicationInfo.isReplica === true || omVal.replicationInfo.status === 'REPLICA') &&
580658
(omVal.originOp === 's3:ObjectTagging:Put' || omVal.originOp === 's3:ObjectTagging:Delete') &&
581659
request.serverAccessLog
582660
) {
@@ -593,7 +671,7 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
593671
// The REPLICA status excludes source-side replication-status updates.
594672
if (
595673
omVal.replicationInfo &&
596-
omVal.replicationInfo.status === 'REPLICA' &&
674+
(omVal.replicationInfo.isReplica === true || omVal.replicationInfo.status === 'REPLICA') &&
597675
omVal.originOp === 's3:ObjectAcl:Put' &&
598676
request.serverAccessLog
599677
) {
@@ -672,7 +750,8 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
672750
// then we want to create a version for the replica object even though
673751
// none was provided in the object metadata value.
674752
if (omVal.replicationInfo.isNFS) {
675-
const { isReplica } = omVal.replicationInfo;
753+
const isReplica = omVal.replicationInfo.isReplica === true
754+
|| omVal.replicationInfo.status === 'REPLICA';
676755
versioning = isReplica;
677756
omVal.replicationInfo.isNFS = !isReplica;
678757
}
@@ -724,6 +803,48 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
724803
options.isNull = isNull;
725804
}
726805

806+
// Cascade triggering
807+
// If the bucket receiving this replica has its own CRR rules, set
808+
// status to PENDING so the queue populator here picks it up for the
809+
// next hop. If not, clear the source-side replicationInfo fields
810+
// Always mark isReplica=true.
811+
if (incomingMicroVersionId) {
812+
const isMDOnly = headers['x-scal-replication-content'] === 'METADATA';
813+
const objSize = omVal['content-length'] || 0;
814+
815+
// These S3-compatible Scality locations are excluded
816+
// as cascade targets because they use the MultiBackend S3 path which
817+
// bypasses the putData/putMetadata routes, so loop detection cannot fire
818+
// on those destinations.
819+
const BLOCKED_LOCATION_TYPES = ['location-scality-ring-s3-v1', 'location-scality-artesca-s3-v1'];
820+
821+
const nextReplInfo = getReplicationInfo(config, objectKey, bucketInfo, isMDOnly, objSize, null, null, null);
822+
823+
if (nextReplInfo) {
824+
nextReplInfo.backends = nextReplInfo.backends.filter(b => {
825+
const loc = config.locationConstraints[b.site];
826+
return !loc || !BLOCKED_LOCATION_TYPES.includes(loc.type);
827+
});
828+
}
829+
830+
if (nextReplInfo && nextReplInfo.backends.length > 0) {
831+
omVal.replicationInfo = nextReplInfo;
832+
} else {
833+
omVal.replicationInfo = {
834+
status: '',
835+
backends: [],
836+
content: [],
837+
destination: '',
838+
storageClass: '',
839+
role: '',
840+
storageType: '',
841+
dataStoreVersionId: '',
842+
};
843+
}
844+
845+
omVal.replicationInfo.isReplica = true;
846+
}
847+
727848
return async.series(
728849
[
729850
// Zenko's CRR delegates replacing the account

‎package.json‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -19,17 +19,17 @@
1919
},
2020
"homepage": "https://github.com/scality/S3#readme",
2121
"dependencies": {
22+
"@aws-crypto/crc32": "^5.2.0",
23+
"@aws-crypto/crc32c": "^5.2.0",
2224
"@aws-sdk/client-iam": "^3.930.0",
2325
"@aws-sdk/client-s3": "^3.1013.0",
2426
"@aws-sdk/client-sts": "^3.930.0",
27+
"@aws-sdk/crc64-nvme-crt": "^3.989.0",
2528
"@aws-sdk/credential-providers": "^3.864.0",
2629
"@aws-sdk/middleware-retry": "^3.374.0",
2730
"@aws-sdk/protocol-http": "^3.374.0",
2831
"@aws-sdk/s3-request-presigner": "^3.901.0",
2932
"@aws-sdk/signature-v4": "^3.374.0",
30-
"@aws-crypto/crc32": "^5.2.0",
31-
"@aws-crypto/crc32c": "^5.2.0",
32-
"@aws-sdk/crc64-nvme-crt": "^3.989.0",
3333
"@azure/storage-blob": "^12.28.0",
3434
"@hapi/joi": "^17.1.1",
3535
"@opentelemetry/api": "^1.9.0",
@@ -65,11 +65,11 @@
6565
"vaultclient": "scality/vaultclient#8.5.3",
6666
"werelogs": "scality/werelogs#8.2.2",
6767
"ws": "^8.18.0",
68+
"@scality/cloudserverclient": "1.0.9",
6869
"xml2js": "^0.6.2"
6970
},
7071
"devDependencies": {
7172
"@eslint/compat": "^1.2.2",
72-
"@scality/cloudserverclient": "1.0.7",
7373
"@scality/eslint-config-scality": "scality/Guidelines#8.3.1",
7474
"eslint": "^9.14.0",
7575
"eslint-plugin-import": "^2.31.0",
@@ -88,10 +88,10 @@
8888
"nodemon": "^3.1.10",
8989
"nyc": "^15.1.0",
9090
"pino-pretty": "^13.1.3",
91+
"prettier": "^3.4.2",
9192
"sinon": "^13.0.1",
9293
"ts-morph": "^28.0.0",
93-
"tv4": "^1.3.0",
94-
"prettier": "^3.4.2"
94+
"tv4": "^1.3.0"
9595
},
9696
"resolutions": {
9797
"jsonwebtoken": "^9.0.0",

0 commit comments

Comments
 (0)