diff --git a/CRR/ReplicationSiteRemover.js b/CRR/ReplicationSiteRemover.js new file mode 100644 index 00000000..59ab14ae --- /dev/null +++ b/CRR/ReplicationSiteRemover.js @@ -0,0 +1,351 @@ +const { promisify } = require('util'); +const { eachLimit } = require('async'); +const { ObjectMD } = require('arsenal').models; +const { ListObjectVersionsCommand, GetBucketReplicationCommand } = require('@aws-sdk/client-s3'); +const CloudserverClient = require('../Clients/CloudserverClient'); +const createS3Client = require('../Clients/s3Client'); + +const LOG_PROGRESS_INTERVAL_MS = 10000; + +/** + * Returns the replication info of an object never set up for replication. + * Taken from arsenal ObjectMD defaults, so it follows the installed arsenal format. + * @returns {Object} Empty replication info (fresh copy). + */ +function getEmptyReplicationInfo() { + return new ObjectMD().getReplicationInfo(); +} + +/** + * Computes the global replication status from the remaining sites. + * A REPLICA backend means the object is a replica on this site. Its other + * backends are a frozen copy of the source (BB-875), so it stays REPLICA. + * On S3C 9.5 the REPLICA status is the only thing that keeps backbeat from + * replicating it again. + * Otherwise same order as backbeat: FAILED, then PENDING, then COMPLETED. + * One difference: a PENDING site gives PENDING, not PROCESSING, so the object + * is requeued for that site. + * @param {Array} backends - Remaining replication backends. + * @returns {string} Global replication status. + */ +function getGlobalStatus(backends) { + const statuses = backends.map(b => b.status); + if (statuses.includes('REPLICA')) { + return 'REPLICA'; + } + if (statuses.includes('FAILED')) { + return 'FAILED'; + } + if (statuses.includes('PENDING')) { + return 'PENDING'; + } + return 'COMPLETED'; +} + +/** + * Removes the site from the comma-separated site list. + * Also removes the item at the same position in the aligned list (storageType) + * when both lists have the same length. + * @param {string} storageClass - Comma-separated site list. + * @param {string} storageType - Comma-separated list aligned with the site list. + * @param {string} site - Site to remove. + * @returns {Object} Updated storageClass and storageType. + */ +function removeSiteFromLists(storageClass, storageType, site) { + const sites = storageClass ? storageClass.split(',') : []; + const types = storageType ? storageType.split(',') : []; + const idx = sites.findIndex(s => s.split(':preferred_read')[0] === site); + if (idx === -1) { + return { storageClass, storageType }; + } + sites.splice(idx, 1); + if (types.length === sites.length + 1) { + types.splice(idx, 1); + } + return { storageClass: sites.join(','), storageType: types.join(',') }; +} + +/** + * Returns the sites set in the StorageClass of the bucket replication rules. + * @param {Array} rules - Bucket replication rules. + * @returns {Array} Site names. + */ +function getRuleSites(rules) { + return (rules || []) + .map(rule => rule.Destination && rule.Destination.StorageClass) + .filter(Boolean) + .flatMap(storageClass => storageClass.split(',')) + .map(site => site.split(':preferred_read')[0]); +} + +class ReplicationSiteRemover { + /** + * @param {Object} params - Configuration. + * @param {Array} params.buckets - Buckets to process. + * @param {string} params.siteToRemove - Replication site to remove from object metadata. + * @param {string} params.accessKey - Access key. + * @param {string} params.secretKey - Secret key. + * @param {string} params.endpoint - S3 endpoint. + * @param {boolean} [params.dryRun=true] - Log what would change, write nothing. + * @param {number} [params.workers=10] - Parallel metadata updates. + * @param {number} [params.listingLimit=1000] - Listing page size. + * @param {string} [params.targetPrefix] - Only process keys with this prefix. + * @param {number} [params.maxUpdates] - Stop after this many updates (resume with the markers). + * @param {number} [params.maxScanned] - Stop after this many scanned versions (resume with the markers). + * @param {string} [params.keyMarker] - Resume listing from this key (first bucket only). + * @param {string} [params.versionIdMarker] - Resume listing from this version (first bucket only). + * @param {boolean} [params.forceRemoveConfiguredSite=false] - Remove the site even if a bucket + * replication rule names it in StorageClass. + * @param {Object} log - Logger. + */ + constructor(params, log) { + this.buckets = params.buckets; + this.siteToRemove = params.siteToRemove; + this.dryRun = params.dryRun !== false; + this.workers = params.workers || 10; + this.listingLimit = params.listingLimit || 1000; + this.targetPrefix = params.targetPrefix; + this.maxUpdates = params.maxUpdates; + this.maxScanned = params.maxScanned; + this.inputKeyMarker = params.keyMarker; + this.inputVersionIdMarker = params.versionIdMarker; + this.forceRemoveConfiguredSite = params.forceRemoveConfiguredSite === true; + this.log = log; + + this.s3 = createS3Client({ + accessKey: params.accessKey, + secretKey: params.secretKey, + endpoint: params.endpoint, + }, log); + this.cloudserverClient = new CloudserverClient(params.endpoint, params.accessKey, params.secretKey); + + this.stats = { + scanned: 0, updated: 0, reset: 0, skipped: 0, errors: 0, + }; + this._bucketInProgress = null; + this._keyMarker = null; + this._versionIdMarker = null; + } + + _logProgress(message = 'progress update') { + this.log.info(message, { + dryRun: this.dryRun, + site: this.siteToRemove, + bucket: this._bucketInProgress, + keyMarker: this._keyMarker, + versionIdMarker: this._versionIdMarker, + ...this.stats, + }); + } + + /** + * Returns the limit reached (MAX_UPDATES or MAX_SCANNED), if any. + * Checked at the end of each listing page. + * @returns {string|null} Limit name and value, or null. + */ + _limitReached() { + if (this.maxUpdates !== undefined && this.stats.updated >= this.maxUpdates) { + return `MAX_UPDATES (${this.maxUpdates})`; + } + if (this.maxScanned !== undefined && this.stats.scanned >= this.maxScanned) { + return `MAX_SCANNED (${this.maxScanned})`; + } + return null; + } + + async _getMetadata(params) { + return promisify(this.cloudserverClient.getMetadata.bind(this.cloudserverClient))(params); + } + + async _putMetadata(params) { + return promisify(this.cloudserverClient.putMetadata.bind(this.cloudserverClient))(params); + } + + /** + * Removes the site from one object version. Never throws: errors are + * counted and logged, so one bad version does not stop the bucket. + * @param {string} bucket - Bucket name. + * @param {string} key - Object key. + * @param {string} versionId - Version ID. + * @returns {Promise} Resolves when done. + */ + async _processVersion(bucket, key, versionId) { + const params = { Bucket: bucket, Key: key, VersionId: versionId }; + const logFields = { bucket, key, versionId }; + this.stats.scanned += 1; + try { + const mdRes = await this._getMetadata(params); + const originalMD = JSON.parse(mdRes.Body); + const objMD = new ObjectMD(originalMD); + if (objMD.getModelVersion() < originalMD['md-model-version']) { + this.stats.errors += 1; + this.log.error('model version regression, refusing to overwrite newer metadata', logFields); + return; + } + const repInfo = objMD.getReplicationInfo(); + const backends = (repInfo && repInfo.backends) || []; + if (!backends.some(b => b.site === this.siteToRemove)) { + this.stats.skipped += 1; + return; + } + const remaining = backends.filter(b => b.site !== this.siteToRemove); + const { storageClass, storageType } = removeSiteFromLists( + objMD.getReplicationStorageClass(), objMD.getReplicationStorageType(), this.siteToRemove); + const reset = remaining.length === 0; + const newStatus = reset ? '' : getGlobalStatus(remaining); + this.log.info(this.dryRun ? 'would update object (dry run)' : 'updating object', { + ...logFields, + action: reset ? 'reset' : 'remove-site', + isDeleteMarker: objMD.getIsDeleteMarker(), + oldStatus: repInfo.status, + newStatus, + oldStorageClass: objMD.getReplicationStorageClass(), + newStorageClass: storageClass, + }); + if (!this.dryRun) { + if (reset) { + // the site was the only destination: back to the never-replicated state, + // so a crrExistingObjects run with the right SITE_NAME picks the object up + objMD.setReplicationInfo({ + ...getEmptyReplicationInfo(), + isNFS: repInfo.isNFS, + }); + // a replica stays a replica (S3C 10.1: isReplica). setReplicationInfo + // only keeps known fields, so set it back after. + if (repInfo.isReplica !== undefined) { + objMD.getReplicationInfo().isReplica = repInfo.isReplica; + } + } else { + objMD.setReplicationBackends(remaining) + .setReplicationStorageClass(storageClass) + .setReplicationStorageType(storageType) + .setReplicationStatus(newStatus); + } + objMD.updateMicroVersionId(); + await this._putMetadata({ ...params, Body: objMD.getSerialized() }); + } + this.stats.updated += 1; + if (reset) { + this.stats.reset += 1; + } + } catch (err) { + this.stats.errors += 1; + this.log.error('error updating object', { ...logFields, error: err.message }); + } + } + + /** + * Refuses to remove a site that a bucket replication rule names in StorageClass: + * that site is a real destination. Removing it would reset objects that replicate + * to it, and a real run can't be undone. + * A bucket without replication rules has nothing to protect. + * @param {string} bucket - Bucket name. + * @returns {Promise} Resolves if the site can be removed, rejects otherwise. + */ + async _checkSiteNotInRules(bucket) { + if (this.forceRemoveConfiguredSite) { + return; + } + let rules; + try { + const res = await this.s3.send(new GetBucketReplicationCommand({ Bucket: bucket })); + rules = res.ReplicationConfiguration.Rules; + } catch (err) { + if (err.name === 'ReplicationConfigurationNotFoundError') { + return; + } + throw err; + } + const ruleSites = getRuleSites(rules); + if (ruleSites.includes(this.siteToRemove)) { + const message = `SITE_TO_REMOVE "${this.siteToRemove}" is a destination of bucket ${bucket} ` + + '(StorageClass of its replication rule). Removing it would reset objects that ' + + 'replicate to it. Check SITE_TO_REMOVE, or set FORCE_REMOVE_CONFIGURED_SITE=true'; + this.log.error(message, { bucket, site: this.siteToRemove, ruleSites }); + throw new Error(message); + } + } + + /** + * Processes one bucket, page by page. + * @param {string} bucket - Bucket name. + * @returns {Promise} True if the bucket was fully processed, + * false if stopped by MAX_UPDATES or MAX_SCANNED. + */ + async _processBucket(bucket) { + this._bucketInProgress = bucket; + await this._checkSiteNotInRules(bucket); + this._keyMarker = this.inputKeyMarker || null; + this._versionIdMarker = this.inputVersionIdMarker || null; + // markers only apply to the first bucket processed + this.inputKeyMarker = undefined; + this.inputVersionIdMarker = undefined; + this.log.info('starting bucket', { + bucket, site: this.siteToRemove, dryRun: this.dryRun, + keyMarker: this._keyMarker, versionIdMarker: this._versionIdMarker, + }); + do { + const data = await this.s3.send(new ListObjectVersionsCommand({ + Bucket: bucket, + MaxKeys: this.listingLimit, + Prefix: this.targetPrefix, + KeyMarker: this._keyMarker || undefined, + VersionIdMarker: this._versionIdMarker || undefined, + })); + const versions = (data.Versions || []).concat(data.DeleteMarkers || []); + await eachLimit(versions, this.workers, + async v => this._processVersion(bucket, v.Key, v.VersionId)); + this._keyMarker = data.NextKeyMarker || null; + this._versionIdMarker = data.NextVersionIdMarker || null; + if (this._limitReached()) { + return !(this._keyMarker || this._versionIdMarker); + } + } while (this._keyMarker || this._versionIdMarker); + this._logProgress('completed bucket'); + return true; + } + + /** + * Runs on all buckets, in order. + * Stops on a listing error, or when MAX_UPDATES or MAX_SCANNED is reached: then + * it logs the bucket list and markers to resume. + * @returns {Promise} Final stats. + */ + async run() { + const progressInterval = setInterval(() => this._logProgress(), LOG_PROGRESS_INTERVAL_MS); + try { + for (let i = 0; i < this.buckets.length; i++) { + const bucket = this.buckets[i].trim(); + const bucketDone = await this._processBucket(bucket); + const limit = this._limitReached(); + if (limit) { + const remaining = this.buckets.slice(bucketDone ? i + 1 : i); + if (remaining.length > 0) { + let message = `reached ${limit}, resume by re-running ` + + `with the bucket list "${remaining.join(',')}"`; + if (!bucketDone) { + message += ` and KEY_MARKER=${this._keyMarker} VERSION_ID_MARKER=${this._versionIdMarker}`; + } + this.log.info(message); + } + break; + } + } + this._bucketInProgress = null; + this._logProgress('completed task'); + return this.stats; + } catch (err) { + // log what was already changed before the stop + this._logProgress('stopped on error'); + throw err; + } finally { + clearInterval(progressInterval); + } + } +} + +module.exports = ReplicationSiteRemover; +module.exports.getGlobalStatus = getGlobalStatus; +module.exports.removeSiteFromLists = removeSiteFromLists; +module.exports.getEmptyReplicationInfo = getEmptyReplicationInfo; +module.exports.getRuleSites = getRuleSites; diff --git a/CRR/ReplicationStatusUpdater.js b/CRR/ReplicationStatusUpdater.js index eafd9083..a3d67b5e 100644 --- a/CRR/ReplicationStatusUpdater.js +++ b/CRR/ReplicationStatusUpdater.js @@ -32,6 +32,8 @@ class ReplicationStatusUpdater { * @param {string} [params.versionIdMarker] - (Optional) Version ID marker for resuming object listing. * @param {boolean} [params.currentVersionOnly] - (Optional) Whether to process only the current version of objects. * @param {boolean} [params.forceUsingConfiguration] - (Optional) Force reset replication target to bucket's configuration. + * @param {boolean} [params.allowNewSite] - (Optional) Accept a site that is not in the bucket + * replication rules and not in the object replication info. */ constructor(params, log) { const { @@ -51,6 +53,7 @@ class ReplicationStatusUpdater { versionIdMarker, currentVersionOnly, forceUsingConfiguration, + allowNewSite, } = params; // inputs @@ -70,6 +73,7 @@ class ReplicationStatusUpdater { this.inputVersionIdMarker = versionIdMarker; this.currentVersionOnly = currentVersionOnly; this.forceUsingConfiguration = forceUsingConfiguration; + this.allowNewSite = allowNewSite; this.log = log; this._setupClients(); @@ -82,6 +86,8 @@ class ReplicationStatusUpdater { this._nUpdated = 0; this._nErrors = 0; this._bucketInProgress = null; + // object that stopped the run on an unknown site, if any + this._unknownSite = null; this._VersionIdMarker = null; this._KeyMarker = null; } @@ -137,6 +143,89 @@ class ReplicationStatusUpdater { }); } + /** + * Returns the sites set in the StorageClass of the bucket replication rules. + * Empty when the rules use the default replication endpoint (no StorageClass). + * @private + * @param {Array} rules - Bucket replication rules. + * @returns {Array} Site names. + */ + _getConfiguredSites(rules) { + return (rules || []) + .map(rule => rule.Destination && rule.Destination.StorageClass) + .filter(Boolean) + .flatMap(storageClass => storageClass.split(',')) + .map(site => site.split(':preferred_read')[0]); + } + + /** + * Returns the sites in the object replication info: backend sites plus + * the top-level storageClass. + * Newer arsenal (multi-destination CRR) deprecates storageClass, so new objects + * may not have it. Objects written before 2018 have no backends. Reading both + * covers every format. + * Empty when the object was never set up for replication. + * @private + * @param {ObjectMD} objMD - Object metadata. + * @returns {Array} Site names. + */ + _getObjectSites(objMD) { + const repInfo = objMD.getReplicationInfo(); + if (!repInfo) { + return []; + } + const backendSites = (repInfo.backends || []).map(backend => backend.site); + const storageClassSites = repInfo.storageClass + ? repInfo.storageClass.split(',').map(site => site.split(':preferred_read')[0]) + : []; + return backendSites.concat(storageClassSites); + } + + /** + * Checks that the site is a known replication destination. + * An unknown site creates a backend no replication processor handles -> the + * object stays PENDING forever. + * If there is something reliable to compare against, the site must be in it: + * - the object already replicates: its sites plus the rule sites + * - every rule names its site (StorageClass): the rule sites + * Otherwise accept. A rule without StorageClass uses the S3C default site from + * the federation config (env_replication_endpoints). This script can't read it. + * + * "names X": a rule has StorageClass=X. + * + * | Rules | Object sites | SITE_NAME | Result | + * |------------------------|------------------|-----------|----------------------------------------| + * | no StorageClass | none | any value | processed: can't check | + * | all name lab-9512 | none | lab-9512 | processed | + * | all name lab-9512 | none | foo | run stopped | + * | mixed (one without) | none | any value | processed: can't check | + * | no StorageClass | lab-9512 | lab-9512 | processed: requeue to its own site | + * | no StorageClass | lab-9512 | foo | run stopped | + * | one names lab-94 | lab-9512 | lab-94 | processed: site from the rule | + * | one names lab-94 | lab-9512 | lab-9512 | processed: requeue to its own site | + * | one names lab-94 | lab-9512 | foo | run stopped | + * | one names foo | lab-9512 | foo | processed: site from the rule | + * | one names lab-9512 | none | unset | processed with lab-9512 (_markPending) | + * | no StorageClass | none | unset | error "missing SITE_NAME" (_markPending) | + * | no StorageClass | 9.5 replica (dr) | primary | run stopped: needs ALLOW_NEW_SITE=true | + * + * ALLOW_NEW_SITE=true skips the check. It's needed when the object must get a + * site it doesn't know and that is the S3C default site: DR failback with + * replicas written by S3C 9.5, or a renamed default DR site. + * @private + * @param {ObjectMD} objMD - Object metadata. + * @param {string} site - Destination site name. + * @param {Array} rules - Bucket replication rules. + * @returns {boolean} True if the site is known, or nothing to compare against. + */ + _isKnownSite(objMD, site, rules) { + const ruleSites = this._getConfiguredSites(rules); + const objectSites = this._getObjectSites(objMD); + const canCheck = objectSites.length > 0 + || rules.every(rule => rule.Destination && rule.Destination.StorageClass); + return !canCheck || ruleSites.concat(objectSites).includes(site); + } + /** * Marks an object as pending for replication. * @private @@ -188,6 +277,33 @@ class ReplicationStatusUpdater { skip = true; return process.nextTick(next); } + // unknown site -> write nothing and stop the run: SITE_NAME is likely + // wrong for every bucket (see _isKnownSite) + if (!this.allowNewSite + && !this._isKnownSite(objMD, storageClass, repConfig.Rules)) { + if (!this._unknownSite) { + this._unknownSite = { bucket, key, versionId }; + this.log.error('unknown replication site, run stopped. SITE_NAME is not a site ' + + 'of this object or of the bucket rules. Expected for the S3C default site ' + + '(rules without StorageClass), e.g. DR failback or renamed DR site. ' + + 'Objects processed before this point may already have this SITE_NAME: ' + + 'if SITE_NAME is wrong, run removeReplicationSite (dry run) with ' + + `SITE_TO_REMOVE=${storageClass} on these buckets to find and remove it. ` + + 'If it reset objects, re-run with the right SITE_NAME from the start of ' + + 'this run (same bucket list and markers): the reset objects are before ' + + 'the resume point. Otherwise re-run from the resume point with the right ' + + 'SITE_NAME, or with ALLOW_NEW_SITE=true', { + bucket, + key, + versionId, + site: storageClass, + ruleSites: this._getConfiguredSites(repConfig.Rules), + objectSites: this._getObjectSites(objMD), + }); + } + skip = true; + return process.nextTick(next); + } // Initialize replication info, if missing // This is particularly important if the object was created before // enabling replication on the bucket. @@ -324,6 +440,11 @@ class ReplicationStatusUpdater { } return eachLimit(versions, this.workers, (i, apply) => { const { Key, VersionId, IsLatest } = i; + if (this._unknownSite) { + // run stopped on an unknown site: don't start new objects + apply(); + return; + } if (this.currentVersionOnly && !IsLatest) { ++this._nSkipped; apply(); @@ -370,6 +491,10 @@ class ReplicationStatusUpdater { if (err) { return done(err); } + if (this._unknownSite) { + // keep the markers of this page: the resume point + return done(); + } this._VersionIdMarker = data.NextVersionIdMarker; this._KeyMarker = data.NextKeyMarker; return done(); @@ -377,6 +502,9 @@ class ReplicationStatusUpdater { }, ), async () => { + if (this._unknownSite) { + return false; + } if (this._nUpdated >= this.maxUpdates || this._nProcessed >= this.maxScanned) { this._logProgress(); let remainingBuckets; @@ -417,6 +545,19 @@ class ReplicationStatusUpdater { return; } this._logProgress(); + if (this._unknownSite) { + let message = 'stopped on an unknown replication site, resume by re-running ' + + 'the script with the bucket list ' + + `"${this.buckets.slice(this.buckets.indexOf(bucketName)).join(',')}"`; + if (this._VersionIdMarker || this._KeyMarker) { + message += ' and the following environment variables set: ' + + `KEY_MARKER=${this._KeyMarker} ` + + `VERSION_ID_MARKER=${this._VersionIdMarker}`; + } + this.log.error(message); + cb(new Error('stopped on an unknown replication site')); + return; + } this.log.info(`completed task for bucket: ${bucket}`); cb(); }, diff --git a/README.md b/README.md index 6af573de..2e4396fc 100644 --- a/README.md +++ b/README.md @@ -164,6 +164,69 @@ Example: `VERSION_ID_MARKER="123456789"` +#### ALLOW_NEW_SITE + +A backend for a site that no replication processor handles leaves the +object PENDING forever. So the script checks `SITE_NAME` when it has +something reliable to compare against: + +* the object already replicates: `SITE_NAME` must be one of its sites + (backend sites, or `storageClass` on older objects), or a site named in + the `StorageClass` of the bucket replication rules +* the object was never replicated and every bucket rule names its site in + `StorageClass`: `SITE_NAME` must be one of those sites + +Otherwise the object is accepted: a rule without `StorageClass` uses the +S3C default replication site, set in the federation config +(`env_replication_endpoints`), which the script can't read. + +At the first object with an unknown site, the script writes nothing for +it, stops, logs the object and the sites it compared against, logs the +bucket list, `KEY_MARKER` and `VERSION_ID_MARKER` to resume from that +point, and exits with code 1. Check `SITE_NAME`, then re-run from the +resume point with the right `SITE_NAME`, or with `ALLOW_NEW_SITE=true`. + +The check can't run on objects with nothing to compare against, so objects +processed before the stop may already have the wrong `SITE_NAME`. Before +re-running with the right `SITE_NAME`, run +[removeReplicationSite](#remove-a-replication-site-from-object-metadata) in dry run with +`SITE_TO_REMOVE=` on the same buckets: it lists the +objects that got the wrong site. If there are any, run it again with +`DRY_RUN=false`. Skip the cleanup if `SITE_NAME` was in fact right (see +below). + +If the cleanup reset objects (the wrong site was their only site), they +are back to "never replicated" and they are before the resume point. So +re-run crrExistingObjects with the right `SITE_NAME` from the start of the +stopped run: same bucket list, same `KEY_MARKER` and `VERSION_ID_MARKER` +as that run, if it had any. With the default +`TARGET_REPLICATION_STATUS=NEW` this is safe: objects that already have the +right site are skipped. + +Set `ALLOW_NEW_SITE=true` to skip the check when the object must get a +site it doesn't know yet and that is the S3C default site: + +* a DR failback (re-sync the replicas of the DR site back to the primary + site with `SITE_NAME=`), for replicas written by S3C 9.5, + also on clusters later upgraded: a replica keeps the replication info of + its source, so it only knows the DR site itself +* a new DR site with a new name, when the bucket rules use the default + replication endpoint (no `StorageClass`) + +Example: + +`ALLOW_NEW_SITE=true` + +#### Notes on S3C versions + +* S3C 9.5.2 / 9.5.4: Backbeat routes on the replication `storageClass`. + Re-triggering one site with `SITE_NAME` also re-sends the objects under + the `sourceCheckIfSizeGreaterThanMB` limit of the Backbeat config + (100 MB by default) to the other sites of the object. +* S3C 10.1: requeueing objects that already exist at the destination can + leave them PENDING, because of the microVersionId format written by + s3utils 1.18 (S3C-11591). + ### Example use cases @@ -220,6 +283,13 @@ export TARGET_REPLICATION_STATUS=NEW,PENDING,COMPLETED,FAILED export MAX_UPDATES=10000 ``` +If the new DR site has a different name and the bucket replication rules +don't set it in `StorageClass`, also set: + +``` +export ALLOW_NEW_SITE=true +``` + #### Re-sync a DR site back to the primary site When objects have been lost from the primary site you can re-sync @@ -233,6 +303,79 @@ export TARGET_REPLICATION_STATUS=REPLICA export MAX_UPDATES=10000 ``` +# Remove a replication site from object metadata + +Clean up objects left PENDING forever by a `crrExistingObjects.js` run +with a wrong `SITE_NAME`. The script removes the site from the object +replication info (backends, storageClass, storageType) and recomputes +the global replication status from the remaining sites: + +* all remaining sites COMPLETED -> COMPLETED, nothing is replicated again +* a remaining site PENDING -> PENDING, the object is requeued for that site +* a remaining site FAILED -> FAILED +* a replica site left -> REPLICA, the object stays a replica +* the site was the only site -> replication info reset to empty (never + replicated), a replica keeps `isReplica`. Run `crrExistingObjects.js` + with the right `SITE_NAME` to replicate the object. + +Object versions and delete markers are both processed. + +**Warning**: the script reads the object metadata, then writes it back. +A tag, ACL, retention or legal hold change on the same version in between +is lost. Run it when no such change happens on the target buckets. + +``` +node removeReplicationSite.js bucket1[,bucket2...] +``` + +## Mandatory environment variables + +* **ENDPOINT**, **ACCESS_KEY**, **SECRET_KEY** +* **SITE_TO_REMOVE**: the wrong site name to remove + +## Optional environment variables + +* **DRY_RUN**: `true` by default, only logs what would change. Set + `DRY_RUN=false` to write the changes. +* **TARGET_PREFIX**: only process keys with this prefix +* **WORKERS**: parallel metadata updates (default 10) +* **LISTING_LIMIT**: listing page size (default 1000) +* **MAX_UPDATES**: stop once this many updates are reached, at the end of + the current listing page (so up to `LISTING_LIMIT` more), and log the + bucket list, `KEY_MARKER` and `VERSION_ID_MARKER` to resume. Use it to + limit the number of objects requeued at once. +* **MAX_SCANNED**: same, on the number of scanned versions. Use it to split + a large bucket into batches: the script reads the metadata of every + version, even when few of them need a change. +* **KEY_MARKER**, **VERSION_ID_MARKER**: resume where an earlier run stopped +* **FORCE_REMOVE_CONFIGURED_SITE**: by default, the script refuses to remove + a site that a bucket replication rule names in `StorageClass`, since it is + a real destination, and stops. Set it to `true` to remove it anyway. +* **DEBUG**: set to 1 for debug logs + +## Procedure + +1. Dry run on the source buckets, check the logs, then run with + `DRY_RUN=false`. +2. On S3C 9.5, do the same on the destination buckets (see notes). +3. Dry run again on both: it must report `updated: 0`. +4. Objects reset to "never replicated": run `crrExistingObjects.js` with + the right `SITE_NAME` to replicate them. + +## Notes + +* S3C 9.5.2 / 9.5.4: Backbeat routes on the replication `storageClass`. + After a run with a wrong `SITE_NAME`, the real site re-sends the objects + under the `sourceCheckIfSizeGreaterThanMB` limit of the Backbeat config + (100 MB by default), and their replicas on the destination carry the + wrong site (`dr=REPLICA, foo=PENDING`). It is harmless while they are + replicas, but run the script on the destination buckets too, at least + before using them as a source (DR failback). Replicas stay REPLICA. +* S3C 10.1: an object left PENDING by the script is requeued. If it + already exists at the destination, it can stay PENDING because of the + microVersionId format written by s3utils 1.18 (S3C-11591). Check those + objects after the run. + # Empty a versioned bucket This script deletes all versions of objects in the bucket, including delete markers, diff --git a/crrExistingObjects.js b/crrExistingObjects.js index 1a4aae00..e5d13ac3 100644 --- a/crrExistingObjects.js +++ b/crrExistingObjects.js @@ -25,6 +25,7 @@ const MAX_SCANNED = (process.env.MAX_SCANNED const { KEY_MARKER } = process.env; const { VERSION_ID_MARKER } = process.env; const CURRENT_VERSION_ONLY = process.env.CURRENT_VERSION_ONLY === 'true' || process.env.CURRENT_VERSION_ONLY === '1' || false; +const ALLOW_NEW_SITE = process.env.ALLOW_NEW_SITE === 'true' || process.env.ALLOW_NEW_SITE === '1' || false; const { ACCESS_KEY, @@ -94,16 +95,18 @@ const replicationStatusUpdater = new ReplicationStatusUpdater({ versionIdMarker: VERSION_ID_MARKER, currentVersionOnly: CURRENT_VERSION_ONLY, forceUsingConfiguration: FORCE_USING_CONFIGURATION, + allowNewSite: ALLOW_NEW_SITE, }, log); replicationStatusUpdater.run(err => { if (err) { - return log.error('error during task execution', { error: err }); + process.exitCode = 1; + return log.error('error during task execution', { error: err.message }); } return log.info('completed task for all buckets'); }); -const stopCrr = replicationStatusUpdater.stop; +const stopCrr = replicationStatusUpdater.stop.bind(replicationStatusUpdater); process.on('SIGINT', stopCrr); process.on('SIGHUP', stopCrr); process.on('SIGQUIT', stopCrr); diff --git a/removeReplicationSite.js b/removeReplicationSite.js new file mode 100644 index 00000000..b85aab5e --- /dev/null +++ b/removeReplicationSite.js @@ -0,0 +1,80 @@ +const werelogs = require('werelogs'); +const ReplicationSiteRemover = require('./CRR/ReplicationSiteRemover'); + +const logLevel = Number.parseInt(process.env.DEBUG, 10) === 1 + ? 'debug' : 'info'; +werelogs.configure({ level: logLevel, dump: 'error' }); +const log = new werelogs.Logger('s3utils::removeReplicationSite'); + +const BUCKETS = process.argv[2] ? process.argv[2].split(',') : null; +const { + ACCESS_KEY, + SECRET_KEY, + ENDPOINT, + SITE_TO_REMOVE, + TARGET_PREFIX, + KEY_MARKER, + VERSION_ID_MARKER, +} = process.env; +// dry run unless explicitly disabled +const DRY_RUN = process.env.DRY_RUN !== 'false' && process.env.DRY_RUN !== '0'; +const WORKERS = (process.env.WORKERS + && Number.parseInt(process.env.WORKERS, 10)) || 10; +const LISTING_LIMIT = (process.env.LISTING_LIMIT + && Number.parseInt(process.env.LISTING_LIMIT, 10)) || 1000; +const MAX_UPDATES = (process.env.MAX_UPDATES + && Number.parseInt(process.env.MAX_UPDATES, 10)) || undefined; +const MAX_SCANNED = (process.env.MAX_SCANNED + && Number.parseInt(process.env.MAX_SCANNED, 10)) || undefined; +const FORCE_REMOVE_CONFIGURED_SITE = process.env.FORCE_REMOVE_CONFIGURED_SITE === 'true' + || process.env.FORCE_REMOVE_CONFIGURED_SITE === '1'; + +if (!BUCKETS || BUCKETS.length === 0) { + log.fatal('No buckets given as input! Please provide ' + + 'a comma-separated list of buckets'); + process.exit(1); +} +['ENDPOINT', 'ACCESS_KEY', 'SECRET_KEY', 'SITE_TO_REMOVE'].forEach(name => { + if (!process.env[name]) { + log.fatal(`${name} not defined`); + process.exit(1); + } +}); + +const remover = new ReplicationSiteRemover({ + buckets: BUCKETS, + siteToRemove: SITE_TO_REMOVE, + accessKey: ACCESS_KEY, + secretKey: SECRET_KEY, + endpoint: ENDPOINT, + dryRun: DRY_RUN, + workers: WORKERS, + listingLimit: LISTING_LIMIT, + targetPrefix: TARGET_PREFIX, + maxUpdates: MAX_UPDATES, + maxScanned: MAX_SCANNED, + forceRemoveConfiguredSite: FORCE_REMOVE_CONFIGURED_SITE, + keyMarker: KEY_MARKER, + versionIdMarker: VERSION_ID_MARKER, +}, log); + +function stop() { + log.warn('stopping execution'); + remover._logProgress(); + process.exit(1); +} +process.on('SIGINT', stop); +process.on('SIGHUP', stop); +process.on('SIGQUIT', stop); +process.on('SIGTERM', stop); + +remover.run() + .then(stats => { + if (stats.errors > 0) { + process.exitCode = 1; + } + }) + .catch(err => { + log.error('error during task execution', { error: err.message }); + process.exitCode = 1; + }); diff --git a/tests/functional/crrExistingObjects.js b/tests/functional/crrExistingObjects.js index 874f0ab5..99d22fb0 100644 --- a/tests/functional/crrExistingObjects.js +++ b/tests/functional/crrExistingObjects.js @@ -1,7 +1,7 @@ const { promisify } = require('util'); const vaultclient = require('vaultclient'); const { Logger } = require('werelogs'); -const { PutObjectCommand, HeadObjectCommand } = require('@aws-sdk/client-s3'); +const { HeadObjectCommand } = require('@aws-sdk/client-s3'); const ReplicationStatusUpdater = require('../../CRR/ReplicationStatusUpdater'); const { @@ -15,10 +15,14 @@ const { deleteTestAccount, configureCrr, removeCrrConfiguration, + putObject, + getReplicationInfo, } = require('../utils/S3Setup'); const log = new Logger('crrExistingObjects:test'); const SITE_NAME = 'test-site'; +// default replication site of the cloudserver under test (workbench: "sf") +const DEFAULT_SITE = process.env.CRR_DEFAULT_SITE || 'sf'; function makeCaptureLogger() { const captured = []; @@ -79,11 +83,7 @@ describe('crrExistingObjects', () => { async function runTest(testObjects) { for (const obj of testObjects) { - await accountSource.s3Client.send(new PutObjectCommand({ - Bucket: accountSource.bucketName, - Key: obj.Key, - Body: obj.Body, - })); + await putObject(accountSource, obj.Key, obj.Body); log.info('Uploaded object', { key: obj.Key }); } @@ -100,6 +100,9 @@ describe('crrExistingObjects', () => { listingLimit: 1000, currentVersionOnly: false, forceUsingConfiguration: true, + // these tests check key encoding: SITE_NAME is not the bucket's + // default replication site, so allow it explicitly + allowNewSite: true, }, logger); await promisify(updater.run.bind(updater))(); @@ -149,4 +152,24 @@ describe('crrExistingObjects', () => { { Key: 'ascii-test-2.txt', Body: 'ascii data 2' }, ]); }, 60000); + it('should stop the run when SITE_NAME is not a known replication site', async () => { + const versionId = await putObject(accountSource, 'obj-1'); + + const updater = new ReplicationStatusUpdater({ + buckets: [accountSource.bucketName], + replicationStatusToProcess: ['NEW', 'PENDING', 'COMPLETED', 'FAILED'], + workers: 10, + accessKey: accountSource.accountAccessKey, + secretKey: accountSource.accountSecretKey, + endpoint, + siteName: 'wrong-site', + }, log); + await expect(promisify(updater.run.bind(updater))()) + .rejects.toThrow('stopped on an unknown replication site'); + + expect(updater._nUpdated).toBe(0); + expect(updater._unknownSite).toEqual(expect.objectContaining({ key: 'obj-1' })); + const repInfo = await getReplicationInfo(accountSource, 'obj-1', versionId); + expect(repInfo.backends.map(b => b.site)).toEqual([DEFAULT_SITE]); + }, 60000); }); diff --git a/tests/functional/removeReplicationSite.js b/tests/functional/removeReplicationSite.js new file mode 100644 index 00000000..75daf07a --- /dev/null +++ b/tests/functional/removeReplicationSite.js @@ -0,0 +1,174 @@ +const vaultclient = require('vaultclient'); +const { Logger } = require('werelogs'); +const { + DeleteObjectCommand, + GetBucketReplicationCommand, + PutBucketReplicationCommand, +} = require('@aws-sdk/client-s3'); +const ReplicationSiteRemover = require('../../CRR/ReplicationSiteRemover'); + +const { + iamHost, + iamPort, + s3Host, + s3Port, + adminAccessKeyId, + adminSecretAccessKey, + createTestAccount, + deleteTestAccount, + configureCrr, + removeCrrConfiguration, + putObject, + getReplicationInfo, + setReplicationInfo, +} = require('../utils/S3Setup'); + +const log = new Logger('removeReplicationSite:test'); +// default replication site of the cloudserver under test (workbench: "sf") +const DEFAULT_SITE = process.env.CRR_DEFAULT_SITE || 'sf'; +const WRONG_SITE = 'wrong-site'; + +/** + * Same metadata as a crrExistingObjects run with SITE_NAME=wrong-site: + * a PENDING backend appended for a site nobody processes. + * @param {Object} replicationInfo - current replication info + * @returns {Object} broken replication info + */ +function withWrongSite(replicationInfo) { + return { + ...replicationInfo, + status: 'PENDING', + backends: replicationInfo.backends.concat({ site: WRONG_SITE, status: 'PENDING', dataStoreVersionId: '' }), + storageClass: [replicationInfo.storageClass, WRONG_SITE].filter(Boolean).join(','), + }; +} + +function sitesOf(replicationInfo) { + return replicationInfo.backends.map(b => b.site); +} + +function runRemover(account, siteToRemove = WRONG_SITE) { + const remover = new ReplicationSiteRemover({ + buckets: [account.bucketName], + siteToRemove, + accessKey: account.accountAccessKey, + secretKey: account.accountSecretKey, + endpoint: `http://${s3Host}:${s3Port}`, + dryRun: false, + }, log); + return remover.run(); +} + +describe('removeReplicationSite', () => { + let vaultClient; + let accountSource; + let accountDest; + + beforeEach(async () => { + vaultClient = new vaultclient.Client( + iamHost, iamPort, false, undefined, undefined, undefined, undefined, + adminAccessKeyId, adminSecretAccessKey, + ); + accountSource = await createTestAccount(vaultClient); + accountDest = await createTestAccount(vaultClient); + // no StorageClass: objects get the default replication site + await configureCrr(accountSource, accountDest); + }); + + afterEach(async () => { + await removeCrrConfiguration(accountSource); + await removeCrrConfiguration(accountDest); + await deleteTestAccount(vaultClient, accountSource); + await deleteTestAccount(vaultClient, accountDest); + }); + + it('should remove the wrong site and recompute the status from the remaining site', async () => { + const versionId = await putObject(accountSource, 'obj-1'); + const repInfo = await getReplicationInfo(accountSource, 'obj-1', versionId); + await setReplicationInfo(accountSource, 'obj-1', versionId, withWrongSite(repInfo)); + + const stats = await runRemover(accountSource); + + expect(stats).toEqual(expect.objectContaining({ updated: 1, reset: 0, errors: 0 })); + const cleaned = await getReplicationInfo(accountSource, 'obj-1', versionId); + expect(sitesOf(cleaned)).toEqual([DEFAULT_SITE]); + expect(cleaned.storageClass).toBe(DEFAULT_SITE); + expect(cleaned.status).toBe(ReplicationSiteRemover.getGlobalStatus(cleaned.backends)); + }, 60000); + + it('should clean delete markers', async () => { + const versionId = await putObject(accountSource, 'obj-1'); + const { VersionId: markerVersionId } = await accountSource.s3Client.send(new DeleteObjectCommand({ + Bucket: accountSource.bucketName, Key: 'obj-1', + })); + for (const vId of [versionId, markerVersionId]) { + const repInfo = await getReplicationInfo(accountSource, 'obj-1', vId); + await setReplicationInfo(accountSource, 'obj-1', vId, withWrongSite(repInfo)); + } + + const stats = await runRemover(accountSource); + + expect(stats).toEqual(expect.objectContaining({ updated: 2, errors: 0 })); + expect(sitesOf(await getReplicationInfo(accountSource, 'obj-1', markerVersionId))).toEqual([DEFAULT_SITE]); + }, 60000); + + it('should keep a replica as REPLICA', async () => { + const versionId = await putObject(accountSource, 'obj-1'); + const repInfo = await getReplicationInfo(accountSource, 'obj-1', versionId); + // S3C 9.5 replica of a multi-destination source (frozen PENDING site), + // then a run with SITE_NAME=wrong-site + await setReplicationInfo(accountSource, 'obj-1', versionId, { + ...repInfo, + status: 'PENDING', + backends: [ + { site: DEFAULT_SITE, status: 'PENDING', dataStoreVersionId: '' }, + { site: 'primary', status: 'REPLICA', dataStoreVersionId: '' }, + { site: WRONG_SITE, status: 'PENDING', dataStoreVersionId: '' }, + ], + storageClass: `${DEFAULT_SITE},primary,${WRONG_SITE}`, + }); + + const stats = await runRemover(accountSource); + + expect(stats).toEqual(expect.objectContaining({ updated: 1, errors: 0 })); + const cleaned = await getReplicationInfo(accountSource, 'obj-1', versionId); + expect(cleaned.status).toBe('REPLICA'); + expect(sitesOf(cleaned)).toEqual([DEFAULT_SITE, 'primary']); + }, 60000); + + it('should reset the replication info when the wrong site is the only one', async () => { + const versionId = await putObject(accountSource, 'obj-1'); + const repInfo = await getReplicationInfo(accountSource, 'obj-1', versionId); + // object never replicated, then crrExistingObjects with SITE_NAME=wrong-site + await setReplicationInfo(accountSource, 'obj-1', versionId, { + ...repInfo, + status: 'PENDING', + backends: [{ site: WRONG_SITE, status: 'PENDING', dataStoreVersionId: '' }], + storageClass: WRONG_SITE, + }); + + const stats = await runRemover(accountSource); + + expect(stats).toEqual(expect.objectContaining({ updated: 1, reset: 1, errors: 0 })); + const cleaned = await getReplicationInfo(accountSource, 'obj-1', versionId); + expect(cleaned.status).toBe(''); + expect(cleaned.backends).toEqual([]); + // empty: '' with arsenal 8.2, absent with arsenal 8.5 (new format) + expect(cleaned.storageClass || '').toBe(''); + }, 60000); + it('should refuse to remove a site named in the bucket rule', async () => { + const versionId = await putObject(accountSource, 'obj-1'); + // name the real site in the rule StorageClass + const { ReplicationConfiguration } = await accountSource.s3Client.send( + new GetBucketReplicationCommand({ Bucket: accountSource.bucketName })); + ReplicationConfiguration.Rules[0].Destination.StorageClass = DEFAULT_SITE; + await accountSource.s3Client.send(new PutBucketReplicationCommand({ + Bucket: accountSource.bucketName, ReplicationConfiguration, + })); + + await expect(runRemover(accountSource, DEFAULT_SITE)) + .rejects.toThrow(`SITE_TO_REMOVE "${DEFAULT_SITE}" is a destination`); + + expect(sitesOf(await getReplicationInfo(accountSource, 'obj-1', versionId))).toEqual([DEFAULT_SITE]); + }, 60000); +}); diff --git a/tests/unit/CRR/ReplicationSiteRemover.js b/tests/unit/CRR/ReplicationSiteRemover.js new file mode 100644 index 00000000..38a231d8 --- /dev/null +++ b/tests/unit/CRR/ReplicationSiteRemover.js @@ -0,0 +1,435 @@ +const werelogs = require('werelogs'); + +const ReplicationSiteRemover = require('../../../CRR/ReplicationSiteRemover'); +const { objectMd } = require('../../utils/crr'); + +const { getGlobalStatus, removeSiteFromLists, getEmptyReplicationInfo } = ReplicationSiteRemover; + +const logger = new werelogs.Logger('ReplicationSiteRemover::tests', 'debug', 'debug'); + +function makeMd(replicationInfo, extra = {}) { + return JSON.stringify({ ...objectMd, ...extra, replicationInfo }); +} + +// object broken by crrExistingObjects with SITE_NAME=foo +const brokenReplicationInfo = { + status: 'PENDING', + backends: [ + { site: 'destination', status: 'COMPLETED', dataStoreVersionId: '' }, + { site: 'foo', status: 'PENDING', dataStoreVersionId: '' }, + ], + content: ['DATA', 'METADATA'], + destination: 'arn:aws:s3:::destination', + storageClass: 'destination,foo', + role: 'arn:aws:iam::root:role/s3-replication-role', + storageType: '', + dataStoreVersionId: '', +}; + +/** + * Builds a remover with mocked S3 and metadata clients. + * @param {Object} params - Extra constructor params. + * @param {Object} opts - Mock setup. + * @param {Array} opts.pages - ListObjectVersions responses, in order. + * @param {Object} opts.mds - Metadata blob per "key/versionId". + * @param {Set} [opts.failGet] - "key/versionId" whose getMetadata fails. + * @param {Array} [opts.rules] - Bucket replication rules (default: one rule without + * StorageClass). null: the bucket has no replication configuration. + * @returns {ReplicationSiteRemover} Remover with mocks. + */ +function initRemover(params, opts) { + const remover = new ReplicationSiteRemover({ + buckets: ['bucket0'], + siteToRemove: 'foo', + accessKey: 'ak', + secretKey: 'sk', + endpoint: 'http://dummyEndpoint:8000', + dryRun: false, + ...params, + }, logger); + const pages = [...opts.pages]; + const rules = opts.rules === undefined + ? [{ Prefix: '', Status: 'Enabled', Destination: { Bucket: 'arn:aws:s3:::destination' } }] + : opts.rules; + remover.s3.send = jest.fn(command => { + if (command.constructor.name === 'GetBucketReplicationCommand') { + if (rules === null) { + const err = new Error('The replication configuration was not found'); + err.name = 'ReplicationConfigurationNotFoundError'; + return Promise.reject(err); + } + return Promise.resolve({ ReplicationConfiguration: { Role: 'role', Rules: rules } }); + } + return Promise.resolve(pages.shift()); + }); + remover.cloudserverClient.getMetadata = jest.fn((p, cb) => { + const id = `${p.Key}/${p.VersionId}`; + if (opts.failGet && opts.failGet.has(id)) { + return cb(new Error('getMetadata failed')); + } + return cb(null, { Body: opts.mds[id] }); + }); + remover.cloudserverClient.putMetadata = jest.fn((p, cb) => cb(null, {})); + return remover; +} + +function page(versions, deleteMarkers = [], next = {}) { + return { + Versions: versions.map(([Key, VersionId]) => ({ Key, VersionId })), + DeleteMarkers: deleteMarkers.map(([Key, VersionId]) => ({ Key, VersionId })), + NextKeyMarker: next.key, + NextVersionIdMarker: next.versionId, + }; +} + +function listCalls(remover) { + return remover.s3.send.mock.calls + .filter(([command]) => command.constructor.name === 'ListObjectVersionsCommand'); +} + +function writtenReplicationInfo(remover, callIndex = 0) { + return JSON.parse(remover.cloudserverClient.putMetadata.mock.calls[callIndex][0].Body).replicationInfo; +} + +describe('ReplicationSiteRemover helpers', () => { + it('getGlobalStatus: FAILED, then PENDING, then COMPLETED', () => { + expect(getGlobalStatus([{ status: 'COMPLETED' }])).toBe('COMPLETED'); + expect(getGlobalStatus([{ status: 'COMPLETED' }, { status: 'PENDING' }])).toBe('PENDING'); + expect(getGlobalStatus([{ status: 'PENDING' }, { status: 'FAILED' }])).toBe('FAILED'); + }); + + it('getGlobalStatus: a replica stays REPLICA', () => { + expect(getGlobalStatus([{ status: 'REPLICA' }])).toBe('REPLICA'); + // the other backends are a frozen copy of the source (BB-875) + expect(getGlobalStatus([{ status: 'PENDING' }, { status: 'REPLICA' }])).toBe('REPLICA'); + expect(getGlobalStatus([{ status: 'COMPLETED' }, { status: 'REPLICA' }])).toBe('REPLICA'); + expect(getGlobalStatus([{ status: 'FAILED' }, { status: 'REPLICA' }])).toBe('REPLICA'); + }); + + it('removeSiteFromLists: removes the site and its aligned storage type', () => { + expect(removeSiteFromLists('aws-location,foo', 'aws_s3,azure', 'foo')) + .toEqual({ storageClass: 'aws-location', storageType: 'aws_s3' }); + }); + + it('removeSiteFromLists: removes a site tagged preferred_read', () => { + expect(removeSiteFromLists('foo:preferred_read,destination', '', 'foo')) + .toEqual({ storageClass: 'destination', storageType: '' }); + }); + + it('removeSiteFromLists: keeps storageType when lists are not aligned', () => { + expect(removeSiteFromLists('destination,foo', '', 'foo')) + .toEqual({ storageClass: 'destination', storageType: '' }); + }); + + it('getEmptyReplicationInfo: returns a fresh empty copy each time', () => { + const first = getEmptyReplicationInfo(); + first.backends.push({ site: 'foo' }); + const second = getEmptyReplicationInfo(); + expect(second.status).toBe(''); + expect(second.backends).toEqual([]); + }); + + it('removeSiteFromLists: no change when the site is absent', () => { + expect(removeSiteFromLists('destination', '', 'foo')) + .toEqual({ storageClass: 'destination', storageType: '' }); + }); +}); + +describe('ReplicationSiteRemover', () => { + it('should remove the site and set COMPLETED when the other sites are COMPLETED', async () => { + const remover = initRemover({}, { + pages: [page([['key0', 'v0']])], + mds: { 'key0/v0': makeMd(brokenReplicationInfo) }, + }); + const stats = await remover.run(); + expect(stats).toEqual({ + scanned: 1, updated: 1, reset: 0, skipped: 0, errors: 0, + }); + const repInfo = writtenReplicationInfo(remover); + expect(repInfo.status).toBe('COMPLETED'); + expect(repInfo.storageClass).toBe('destination'); + expect(repInfo.backends).toEqual([ + { site: 'destination', status: 'COMPLETED', dataStoreVersionId: '' }, + ]); + }); + + it('should keep REPLICA on a replica', async () => { + // replica on the DR site, broken by a run with SITE_NAME=foo + const brokenReplica = { + ...brokenReplicationInfo, + backends: [ + { site: 'destination', status: 'REPLICA', dataStoreVersionId: '' }, + brokenReplicationInfo.backends[1], + ], + }; + const remover = initRemover({}, { + pages: [page([['key0', 'v0']])], + mds: { 'key0/v0': makeMd(brokenReplica) }, + }); + await remover.run(); + const repInfo = writtenReplicationInfo(remover); + expect(repInfo.status).toBe('REPLICA'); + expect(repInfo.backends).toEqual([{ site: 'destination', status: 'REPLICA', dataStoreVersionId: '' }]); + }); + + it('should keep REPLICA on a replica of a multi-destination source', async () => { + // S3C 9.5 replica on "primary" with a frozen dr=PENDING, then a run with SITE_NAME=foo + const brokenReplica = { + ...brokenReplicationInfo, + status: 'PENDING', + backends: [ + { site: 'dr', status: 'PENDING', dataStoreVersionId: '' }, + { site: 'primary', status: 'REPLICA', dataStoreVersionId: '' }, + { site: 'foo', status: 'PENDING', dataStoreVersionId: '' }, + ], + storageClass: 'dr,primary,foo', + }; + const remover = initRemover({}, { + pages: [page([['key0', 'v0']])], + mds: { 'key0/v0': makeMd(brokenReplica) }, + }); + await remover.run(); + const repInfo = writtenReplicationInfo(remover); + expect(repInfo.status).toBe('REPLICA'); + expect(repInfo.storageClass).toBe('dr,primary'); + expect(repInfo.backends).toEqual([ + { site: 'dr', status: 'PENDING', dataStoreVersionId: '' }, + { site: 'primary', status: 'REPLICA', dataStoreVersionId: '' }, + ]); + }); + + it('should keep PENDING when another site is still PENDING', async () => { + const repInfo = JSON.parse(JSON.stringify(brokenReplicationInfo)); + repInfo.backends[0].status = 'PENDING'; + const remover = initRemover({}, { + pages: [page([['key0', 'v0']])], + mds: { 'key0/v0': makeMd(repInfo) }, + }); + await remover.run(); + expect(writtenReplicationInfo(remover).status).toBe('PENDING'); + }); + + it('should write nothing in dry run', async () => { + const remover = initRemover({ dryRun: true }, { + pages: [page([['key0', 'v0']])], + mds: { 'key0/v0': makeMd(brokenReplicationInfo) }, + }); + const stats = await remover.run(); + expect(remover.cloudserverClient.putMetadata).not.toHaveBeenCalled(); + expect(stats.updated).toBe(1); + }); + + it('should default to dry run', async () => { + const remover = initRemover({ dryRun: undefined }, { + pages: [page([['key0', 'v0']])], + mds: { 'key0/v0': makeMd(brokenReplicationInfo) }, + }); + await remover.run(); + expect(remover.cloudserverClient.putMetadata).not.toHaveBeenCalled(); + }); + + it('should skip objects without the site', async () => { + const clean = { + ...brokenReplicationInfo, + status: 'COMPLETED', + backends: [brokenReplicationInfo.backends[0]], + storageClass: 'destination', + }; + const remover = initRemover({}, { + pages: [page([['key0', 'v0']])], + mds: { 'key0/v0': makeMd(clean) }, + }); + const stats = await remover.run(); + expect(remover.cloudserverClient.putMetadata).not.toHaveBeenCalled(); + expect(stats.skipped).toBe(1); + }); + + it('should reset the replication info when the site is the only one', async () => { + // crrExistingObjects with SITE_NAME=foo on an object never replicated + const onlyFoo = { + ...brokenReplicationInfo, + backends: [brokenReplicationInfo.backends[1]], + storageClass: 'foo', + }; + const remover = initRemover({}, { + pages: [page([['key0', 'v0']])], + mds: { 'key0/v0': makeMd(onlyFoo) }, + }); + const stats = await remover.run(); + expect(stats).toEqual(expect.objectContaining({ updated: 1, reset: 1 })); + // same as an object never set up for replication, in the installed arsenal format + expect(writtenReplicationInfo(remover)).toEqual(JSON.parse(JSON.stringify(getEmptyReplicationInfo()))); + }); + + it('should not reset in dry run', async () => { + const onlyFoo = { + ...brokenReplicationInfo, + backends: [brokenReplicationInfo.backends[1]], + storageClass: 'foo', + }; + const remover = initRemover({ dryRun: true }, { + pages: [page([['key0', 'v0']])], + mds: { 'key0/v0': makeMd(onlyFoo) }, + }); + const stats = await remover.run(); + expect(remover.cloudserverClient.putMetadata).not.toHaveBeenCalled(); + expect(stats).toEqual(expect.objectContaining({ updated: 1, reset: 1 })); + }); + + it('should process delete markers', async () => { + const remover = initRemover({}, { + pages: [page([], [['key0', 'dm0']])], + mds: { 'key0/dm0': makeMd(brokenReplicationInfo, { isDeleteMarker: true }) }, + }); + const stats = await remover.run(); + expect(stats.updated).toBe(1); + expect(writtenReplicationInfo(remover).status).toBe('COMPLETED'); + }); + + it('should count an error and keep going', async () => { + const remover = initRemover({}, { + pages: [page([['key0', 'v0'], ['key1', 'v1']])], + mds: { 'key1/v1': makeMd(brokenReplicationInfo) }, + failGet: new Set(['key0/v0']), + }); + const stats = await remover.run(); + expect(stats.errors).toBe(1); + expect(stats.updated).toBe(1); + }); + + it('should list all pages', async () => { + const remover = initRemover({}, { + pages: [ + page([['key0', 'v0']], [], { key: 'key0', versionId: 'v0' }), + page([['key1', 'v1']]), + ], + mds: { + 'key0/v0': makeMd(brokenReplicationInfo), + 'key1/v1': makeMd(brokenReplicationInfo), + }, + }); + const stats = await remover.run(); + expect(listCalls(remover)).toHaveLength(2); + expect(listCalls(remover)[1][0].input).toEqual(expect.objectContaining({ + KeyMarker: 'key0', VersionIdMarker: 'v0', + })); + expect(stats.updated).toBe(2); + }); + + it('should stop at MAX_UPDATES and keep the resume markers', async () => { + const remover = initRemover({ buckets: ['bucket0', 'bucket1'], maxUpdates: 1 }, { + pages: [ + page([['key0', 'v0']], [], { key: 'key0', versionId: 'v0' }), + page([['key1', 'v1']]), + ], + mds: { + 'key0/v0': makeMd(brokenReplicationInfo), + 'key1/v1': makeMd(brokenReplicationInfo), + }, + }); + const stats = await remover.run(); + expect(listCalls(remover)).toHaveLength(1); + expect(stats.updated).toBe(1); + expect(remover._keyMarker).toBe('key0'); + expect(remover._versionIdMarker).toBe('v0'); + }); + + it('should stop at MAX_SCANNED even when nothing is updated', async () => { + const clean = { + ...brokenReplicationInfo, + status: 'COMPLETED', + backends: [brokenReplicationInfo.backends[0]], + storageClass: 'destination', + }; + const remover = initRemover({ buckets: ['bucket0', 'bucket1'], maxScanned: 1 }, { + pages: [ + page([['key0', 'v0']], [], { key: 'key0', versionId: 'v0' }), + page([['key1', 'v1']]), + ], + mds: { + 'key0/v0': makeMd(clean), + 'key1/v1': makeMd(clean), + }, + }); + const stats = await remover.run(); + expect(listCalls(remover)).toHaveLength(1); + expect(stats).toEqual(expect.objectContaining({ scanned: 1, updated: 0 })); + expect(remover._keyMarker).toBe('key0'); + expect(remover._versionIdMarker).toBe('v0'); + }); + + it('should resume from the input markers on the first bucket only', async () => { + const remover = initRemover({ + buckets: ['bucket0', 'bucket1'], keyMarker: 'key0', versionIdMarker: 'v0', + }, { + pages: [page([]), page([])], + mds: {}, + }); + await remover.run(); + expect(listCalls(remover)[0][0].input).toEqual(expect.objectContaining({ + Bucket: 'bucket0', KeyMarker: 'key0', VersionIdMarker: 'v0', + })); + expect(listCalls(remover)[1][0].input).toEqual(expect.objectContaining({ + Bucket: 'bucket1', KeyMarker: undefined, VersionIdMarker: undefined, + })); + }); + + it('should stop the run on a listing error', async () => { + const remover = initRemover({}, { pages: [], mds: {} }); + remover.s3.send = jest.fn(() => Promise.reject(new Error('listing failed'))); + await expect(remover.run()).rejects.toThrow('listing failed'); + }); + it('should keep isReplica when resetting a replica', async () => { + // S3C 10.1 replica, then a run with SITE_NAME=foo: foo is its only site + const replicaWithOnlyFoo = { + ...brokenReplicationInfo, + backends: [brokenReplicationInfo.backends[1]], + storageClass: 'foo', + isReplica: true, + }; + const remover = initRemover({}, { + pages: [page([['key0', 'v0']])], + mds: { 'key0/v0': makeMd(replicaWithOnlyFoo) }, + }); + const stats = await remover.run(); + expect(stats.reset).toBe(1); + const repInfo = writtenReplicationInfo(remover); + expect(repInfo.isReplica).toBe(true); + expect(repInfo.backends).toEqual([]); + expect(repInfo.status).toBe(''); + }); + + it('should refuse to remove a site named in the bucket rule', async () => { + const remover = initRemover({ siteToRemove: 'destination' }, { + pages: [page([['key0', 'v0']])], + mds: { 'key0/v0': makeMd(brokenReplicationInfo) }, + rules: [{ Prefix: '', Status: 'Enabled', + Destination: { Bucket: 'arn:aws:s3:::destination', StorageClass: 'destination' } }], + }); + await expect(remover.run()).rejects.toThrow('SITE_TO_REMOVE "destination" is a destination'); + expect(listCalls(remover)).toHaveLength(0); + expect(remover.cloudserverClient.putMetadata).not.toHaveBeenCalled(); + }); + + it('should remove a site named in the bucket rule with forceRemoveConfiguredSite', async () => { + const remover = initRemover({ siteToRemove: 'destination', forceRemoveConfiguredSite: true }, { + pages: [page([['key0', 'v0']])], + mds: { 'key0/v0': makeMd(brokenReplicationInfo) }, + rules: [{ Prefix: '', Status: 'Enabled', + Destination: { Bucket: 'arn:aws:s3:::destination', StorageClass: 'destination' } }], + }); + const stats = await remover.run(); + expect(stats.updated).toBe(1); + expect(writtenReplicationInfo(remover).backends.map(b => b.site)).toEqual(['foo']); + }); + + it('should process a bucket without replication configuration', async () => { + const remover = initRemover({}, { + pages: [page([['key0', 'v0']])], + mds: { 'key0/v0': makeMd(brokenReplicationInfo) }, + rules: null, + }); + const stats = await remover.run(); + expect(stats.updated).toBe(1); + }); +}); diff --git a/tests/unit/CRR/ReplicationStatusUpdater.js b/tests/unit/CRR/ReplicationStatusUpdater.js index a03141dc..ac07580e 100644 --- a/tests/unit/CRR/ReplicationStatusUpdater.js +++ b/tests/unit/CRR/ReplicationStatusUpdater.js @@ -905,3 +905,243 @@ describe('ReplicationStatusUpdater model version guard', () => { }); }); }); + +describe('ReplicationStatusUpdater site validation', () => { + // bucket rule relying on the default replication endpoint (no StorageClass), as on S3C + const defaultEndpointReplicationRes = { + ReplicationConfiguration: { + Role: 'arn:aws:iam::root:role/s3-replication-role', + Rules: [{ + ID: 'r0', + Prefix: '', + Status: 'Enabled', + Destination: { Bucket: 'arn:aws:s3:::destination' }, + }], + }, + }; + const replicatedReplicationInfo = { + status: 'COMPLETED', + backends: [{ site: 'destination', status: 'COMPLETED', dataStoreVersionId: '' }], + content: ['DATA', 'METADATA'], + destination: 'arn:aws:s3:::destination', + storageClass: 'destination', + role: 'arn:aws:iam::root:role/s3-replication-role', + storageType: '', + dataStoreVersionId: '', + }; + + function initCrr(params, replicationRes, replicationInfo) { + const crr = initializeCrrWithMocks({ + buckets: ['bucket0'], + workers: 10, + replicationStatusToProcess: ['NEW'], + ...params, + }, logger, replicationRes ? { GetBucketReplicationCommand: replicationRes } : {}); + if (replicationInfo) { + crr.cloudserverclient.getMetadata = jest.fn((p, cb) => { + const md = JSON.parse(getMetadataRes.Body); + md.replicationInfo = replicationInfo; + cb(null, { Body: JSON.stringify(md) }); + }); + } + return crr; + } + + it('should stop the run when every rule names its site and SITE_NAME is not one of them', done => { + // default fixture: rule with StorageClass=aws-location, object never replicated + const crr = initCrr({ siteName: 'foo' }); + crr.run(err => { + assert(err); + expect(crr.cloudserverclient.putMetadata).not.toHaveBeenCalled(); + assert.strictEqual(crr._nUpdated, 0); + assert.strictEqual(crr._unknownSite.bucket, 'bucket0'); + done(); + }); + }); + + it('should accept a never-replicated object when a rule has no StorageClass (mixed rules)', done => { + // R6: the default site of the rule without StorageClass can't be checked + const res = JSON.parse(JSON.stringify(defaultEndpointReplicationRes)); + res.ReplicationConfiguration.Rules.push({ + ID: 'r1', + Prefix: 'other/', + Status: 'Enabled', + Destination: { Bucket: 'arn:aws:s3:::destination', StorageClass: 'primary' }, + }); + const crr = initCrr({ siteName: 'destination' }, res); + crr.run(err => { + assert.ifError(err); + expect(crr.cloudserverclient.putMetadata).toHaveBeenCalledTimes(1); + assert.strictEqual(crr._nUpdated, 1); + done(); + }); + }); + + it('should accept a site known by the object but not in the bucket rules', done => { + // rule names aws-location, object already replicates to destination + const crr = initCrr({ siteName: 'destination', replicationStatusToProcess: ['COMPLETED'] }, + null, replicatedReplicationInfo); + crr.run(err => { + assert.ifError(err); + expect(crr.cloudserverclient.putMetadata).toHaveBeenCalledTimes(1); + assert.strictEqual(crr._nUpdated, 1); + done(); + }); + }); + + it('should accept the default endpoint site when rules mix StorageClass and no StorageClass', done => { + const res = JSON.parse(JSON.stringify(defaultEndpointReplicationRes)); + res.ReplicationConfiguration.Rules.push({ + ID: 'r1', + Prefix: 'other/', + Status: 'Enabled', + Destination: { Bucket: 'arn:aws:s3:::destination', StorageClass: 'aws-location' }, + }); + const crr = initCrr({ siteName: 'destination', replicationStatusToProcess: ['COMPLETED'] }, + res, replicatedReplicationInfo); + crr.run(err => { + assert.ifError(err); + expect(crr.cloudserverclient.putMetadata).toHaveBeenCalledTimes(1); + assert.strictEqual(crr._nUpdated, 1); + done(); + }); + }); + + it('should accept a SITE_NAME listed with preferred_read in the bucket rules', done => { + const res = JSON.parse(JSON.stringify(defaultEndpointReplicationRes)); + res.ReplicationConfiguration.Rules[0].Destination.StorageClass = 'aws-location:preferred_read,foo'; + const crr = initCrr({ siteName: 'aws-location' }, res); + crr.run(err => { + assert.ifError(err); + expect(crr.cloudserverclient.putMetadata).toHaveBeenCalledTimes(1); + done(); + }); + }); + + it('should stop the run when the object replicates to another site (CERSAI case)', done => { + const crr = initCrr({ siteName: 'foo' }, defaultEndpointReplicationRes, replicatedReplicationInfo); + crr.run(err => { + assert(err); + expect(crr.cloudserverclient.putMetadata).not.toHaveBeenCalled(); + assert.strictEqual(crr._nUpdated, 0); + done(); + }); + }); + + it('should mark an object with no replication info when rules have no StorageClass', done => { + // no site in the rule nor in the object: nothing to compare, the site is accepted + const crr = initCrr({ siteName: 'destination' }, defaultEndpointReplicationRes); + crr.run(err => { + assert.ifError(err); + expect(crr.cloudserverclient.putMetadata).toHaveBeenCalledTimes(1); + assert.strictEqual(crr._nUpdated, 1); + done(); + }); + }); + + it('should stop the run at the first unknown site and keep the resume point', done => { + // listing always returns next markers: an unstopped run would loop forever + const endlessListing = { + ...listVersionsRes, + IsTruncated: true, + NextKeyMarker: 'next-key', + NextVersionIdMarker: 'next-version', + }; + const crr = initializeCrrWithMocks({ + buckets: ['bucket0', 'bucket1'], + workers: 1, + replicationStatusToProcess: ['NEW'], + siteName: 'foo', + keyMarker: 'resume-key', + versionIdMarker: 'resume-version', + }, logger, { + ListObjectVersionsCommand: endlessListing, + GetBucketReplicationCommand: defaultEndpointReplicationRes, + }); + crr.cloudserverclient.getMetadata = jest.fn((p, cb) => { + const md = JSON.parse(getMetadataRes.Body); + md.replicationInfo = replicatedReplicationInfo; + cb(null, { Body: JSON.stringify(md) }); + }); + const errorSpy = jest.spyOn(crr.log, 'error'); + crr.run(err => { + assert(err); + // bucket0 only: one listing + one GetBucketReplication, first object only + expect(crr.s3.send).toHaveBeenCalledTimes(2); + expect(crr.cloudserverclient.getMetadata).toHaveBeenCalledTimes(1); + expect(crr.cloudserverclient.putMetadata).not.toHaveBeenCalled(); + // markers of the stopped page, to resume from there + assert.strictEqual(crr._KeyMarker, 'resume-key'); + assert.strictEqual(crr._VersionIdMarker, 'resume-version'); + expect(errorSpy).toHaveBeenCalledWith(expect.stringContaining( + 'bucket list "bucket0,bucket1" and the following environment variables set: ' + + 'KEY_MARKER=resume-key VERSION_ID_MARKER=resume-version')); + errorSpy.mockRestore(); + done(); + }); + }); + + it('should not stop on unknown sites when allowNewSite is set', done => { + const crr = initializeCrrWithMocks({ + buckets: ['bucket0'], + workers: 1, + replicationStatusToProcess: ['NEW'], + siteName: 'foo', + allowNewSite: true, + }, logger, { + ListObjectVersionsCommand: listVersionsRes, + GetBucketReplicationCommand: defaultEndpointReplicationRes, + }); + crr.cloudserverclient.getMetadata = jest.fn((p, cb) => { + const md = JSON.parse(getMetadataRes.Body); + md.replicationInfo = replicatedReplicationInfo; + cb(null, { Body: JSON.stringify(md) }); + }); + crr.run(err => { + assert.ifError(err); + expect(crr.cloudserverclient.putMetadata).toHaveBeenCalledTimes(2); + assert.strictEqual(crr._unknownSite, null); + done(); + }); + }); + + // newer arsenal (multi-destination CRR): sites in backends, no top-level storageClass + const backendsOnlyReplicationInfo = { + status: 'COMPLETED', + backends: [{ site: 'destination', status: 'COMPLETED', dataStoreVersionId: '' }], + content: ['DATA', 'METADATA'], + }; + + it('should read the object sites from backends when storageClass is missing', done => { + const crr = initCrr({ siteName: 'foo' }, defaultEndpointReplicationRes, backendsOnlyReplicationInfo); + crr.run(err => { + assert(err); + expect(crr.cloudserverclient.putMetadata).not.toHaveBeenCalled(); + done(); + }); + }); + + it('should accept a site known only from backends when storageClass is missing', done => { + const crr = initCrr({ siteName: 'destination', replicationStatusToProcess: ['COMPLETED'] }, + defaultEndpointReplicationRes, backendsOnlyReplicationInfo); + crr.run(err => { + assert.ifError(err); + expect(crr.cloudserverclient.putMetadata).toHaveBeenCalledTimes(1); + assert.strictEqual(crr._nUpdated, 1); + done(); + }); + }); + + it('should add a new site when allowNewSite is set', done => { + const crr = initCrr({ siteName: 'foo', allowNewSite: true }, + defaultEndpointReplicationRes, replicatedReplicationInfo); + crr.run(err => { + assert.ifError(err); + expect(crr.cloudserverclient.putMetadata).toHaveBeenCalledTimes(1); + const { replicationInfo } = JSON.parse(crr.cloudserverclient.putMetadata.mock.calls[0][0].Body); + assert.strictEqual(replicationInfo.storageClass, 'destination,foo'); + assert.strictEqual(crr._nUpdated, 1); + done(); + }); + }); +}); diff --git a/tests/unit/CRR/crrExistingObjects.js b/tests/unit/CRR/crrExistingObjects.js index 9cf5c1e4..d200fa62 100644 --- a/tests/unit/CRR/crrExistingObjects.js +++ b/tests/unit/CRR/crrExistingObjects.js @@ -100,6 +100,7 @@ describe('crrExistingObjects', () => { versionIdMarker: 'testVersionIdMarker', currentVersionOnly: true, forceUsingConfiguration: true, + allowNewSite: false, }, expect.anything()); expect(mockFatal).not.toHaveBeenCalled(); @@ -137,6 +138,7 @@ describe('crrExistingObjects', () => { versionIdMarker: undefined, currentVersionOnly: false, forceUsingConfiguration: false, + allowNewSite: false, }, expect.anything()); expect(mockFatal).not.toHaveBeenCalled(); diff --git a/tests/utils/S3Setup.js b/tests/utils/S3Setup.js index 5f939cbb..cf86f7a0 100644 --- a/tests/utils/S3Setup.js +++ b/tests/utils/S3Setup.js @@ -1,9 +1,10 @@ -const { S3Client, CreateBucketCommand, ListBucketsCommand, ListObjectVersionsCommand, DeleteObjectsCommand, DeleteBucketCommand, PutBucketVersioningCommand, PutBucketReplicationCommand, DeleteBucketReplicationCommand } = require('@aws-sdk/client-s3'); +const { S3Client, CreateBucketCommand, ListBucketsCommand, ListObjectVersionsCommand, DeleteObjectsCommand, DeleteBucketCommand, PutBucketVersioningCommand, PutBucketReplicationCommand, DeleteBucketReplicationCommand, PutObjectCommand } = require('@aws-sdk/client-s3'); const { IAMClient, CreateUserCommand, CreatePolicyCommand, CreateRoleCommand, AttachRolePolicyCommand, DetachRolePolicyCommand, DeleteRoleCommand, DeletePolicyCommand, DeleteUserCommand, ListRolesCommand, ListAttachedRolePoliciesCommand, ListUsersCommand, ListAttachedUserPoliciesCommand, DetachUserPolicyCommand } = require('@aws-sdk/client-iam'); const { promisify } = require('util'); const { Logger } = require('werelogs'); const admincredentials = require('vaultclient/tests/utils/admincredentials.json'); const crypto = require('crypto'); +const CloudserverClient = require('../../Clients/CloudserverClient'); const log = new Logger('S3Setup'); @@ -244,7 +245,7 @@ async function configureCrr(accountSource, accountDest, opts = {}) { { Effect: 'Allow', Action: ['s3:ListBucket', 's3:GetReplicationConfiguration'], - Resource: ['arn:aws:s3:::source'], + Resource: [`arn:aws:s3:::${accountSource.bucketName}`], }, { Effect: 'Allow', @@ -358,11 +359,65 @@ async function removeCrrConfiguration(account) { } } +/** + * Put an object in the account bucket. + * @param {Object} account - object returned by createTestAccount + * @param {string} key - object key + * @param {string} [body] - object data + * @returns {Promise} version ID + */ +async function putObject(account, key, body = `data of ${key}`) { + const res = await account.s3Client.send(new PutObjectCommand({ + Bucket: account.bucketName, Key: key, Body: body, + })); + return res.VersionId; +} + +function cloudserverClientFor(account) { + return new CloudserverClient(`http://${s3Host}:${s3Port}`, + account.accountAccessKey, account.accountSecretKey); +} + +/** + * Read the replication info of an object version, through the backbeat metadata route. + * @param {Object} account - object returned by createTestAccount + * @param {string} key - object key + * @param {string} versionId - version ID + * @returns {Promise} replicationInfo + */ +async function getReplicationInfo(account, key, versionId) { + const client = cloudserverClientFor(account); + const res = await promisify(client.getMetadata.bind(client))({ + Bucket: account.bucketName, Key: key, VersionId: versionId, + }); + return JSON.parse(res.Body).replicationInfo; +} + +/** + * Overwrite the replication info of an object version, through the backbeat metadata route. + * @param {Object} account - object returned by createTestAccount + * @param {string} key - object key + * @param {string} versionId - version ID + * @param {Object} replicationInfo - new replication info + * @returns {Promise} resolves when written + */ +async function setReplicationInfo(account, key, versionId, replicationInfo) { + const client = cloudserverClientFor(account); + const params = { Bucket: account.bucketName, Key: key, VersionId: versionId }; + const res = await promisify(client.getMetadata.bind(client))(params); + const md = JSON.parse(res.Body); + md.replicationInfo = replicationInfo; + await promisify(client.putMetadata.bind(client))({ ...params, Body: JSON.stringify(md) }); +} + module.exports = { createTestAccount, deleteTestAccount, configureCrr, removeCrrConfiguration, + putObject, + getReplicationInfo, + setReplicationInfo, iamHost, iamPort, s3Host,