diff --git a/extensions/mongoProcessor/MongoProcessorConfigValidator.js b/extensions/mongoProcessor/MongoProcessorConfigValidator.js index b5ae49183..e9b56aa11 100644 --- a/extensions/mongoProcessor/MongoProcessorConfigValidator.js +++ b/extensions/mongoProcessor/MongoProcessorConfigValidator.js @@ -1,5 +1,7 @@ const joi = require('joi'); -const { retryParamsJoi, probeServerJoi, logJoiOptional } = require('../../lib/config/configItems.joi'); +const { modes, defaultMode } = require('./modes'); +const { authJoi, retryParamsJoi, probeServerJoi, logJoiOptional, mongoJoi } = + require('../../lib/config/configItems.joi'); const { extensionConfigValidator } = require('../../lib/config/extensionConfigValidator'); const { MAX_QUEUED_DEFAULT } = require('../../lib/constants').backbeatConsumer; @@ -7,6 +9,9 @@ const { MAX_QUEUED_DEFAULT } = require('../../lib/constants').backbeatConsumer; const joiSchema = joi.object({ topic: joi.string().required(), groupId: joi.string().required(), + mode: joi.string().valid(...Object.keys(modes)).default(defaultMode), + mongodb: mongoJoi.when('mode', { is: 'dr', then: joi.required() }), + auth: authJoi.when('mode', { is: 'dr', then: joi.required() }), retry: retryParamsJoi, concurrency: joi.number().greater(0).default(1), maxQueued: joi.number().greater(0).default(MAX_QUEUED_DEFAULT), diff --git a/extensions/mongoProcessor/MongoQueueProcessor.js b/extensions/mongoProcessor/MongoQueueProcessor.js index d0b659354..c67370a58 100644 --- a/extensions/mongoProcessor/MongoQueueProcessor.js +++ b/extensions/mongoProcessor/MongoQueueProcessor.js @@ -4,11 +4,10 @@ const async = require('async'); const Logger = require('werelogs').Logger; const errors = require('arsenal').errors; -const { replicationBackends, emptyFileMd5 } = require('arsenal').constants; +const { replicationBackends } = require('arsenal').constants; const MongoClient = require('arsenal').storage .metadata.mongoclient.MongoClientInterface; const { ObjectMD, ReplicationConfiguration } = require('arsenal').models; -const { VersionID } = require('arsenal').versioning; const { extractVersionId } = require('../../lib/util/versioning'); const Config = require('../../lib/Config'); @@ -19,7 +18,7 @@ const ObjectQueueEntry = require('../../lib/models/ObjectQueueEntry'); const MetricsProducer = require('../../lib/MetricsProducer'); const { metricsExtension, metricsTypeCompleted, metricsTypePendingOnly } = require('../ingestion/constants'); -const getContentType = require('./utils/contentTypeHelper'); +const { modes, defaultMode } = require('./modes'); const BucketMemState = require('./utils/BucketMemState'); const MongoProcessorMetrics = require('./MongoProcessorMetrics'); @@ -75,6 +74,8 @@ class MongoQueueProcessor { this._bootstrapList = null; this.logger = new Logger('Backbeat:Ingestion:MongoProcessor'); this.mongoClientConfig.logger = this.logger; + this._mode = + new modes[mongoProcessorConfig.mode ?? defaultMode](); this._mongoClient = new MongoClient(this.mongoClientConfig); this._bucketMemState = new BucketMemState(Config); @@ -236,78 +237,6 @@ class MongoQueueProcessor { }); } - /** - * Update ingested entry metadata fields: owner-id, owner-display-name - * @param {ObjectQueueEntry} entry - object queue entry object - * @param {BucketInfo} bucketInfo - bucket info object - * @return {undefined} - */ - _updateOwnerMD(entry, bucketInfo) { - // zenko bucket owner information is being set on ingested md - entry.setOwnerDisplayName(bucketInfo.getOwnerDisplayName()); - entry.setOwnerId(bucketInfo.getOwner()); - } - - /** - * Update ingested entry metadata fields: dataStoreName - * @param {ObjectQueueEntry} entry - object queue entry object - * @param {string} location - owner details - * @return {undefined} - */ - _updateObjectDataStoreName(entry, location) { - entry.setDataStoreName(location); - } - - /** - * Update ingested entry metadata location field. Each location change - * includes: key, dataStoreName, dataStoreType, dataStoreVersionId - * @param {ObjectQueueEntry} entry - object queue entry object - * @param {string} zenkoLocation - zenko storage location name - * @return {undefined} - */ - _updateLocations(entry, zenkoLocation) { - const locations = entry.getLocation(); - // if version id is undefined, we have a single null object. - // To hold reference to this null object, we need to encode "null" - // as its dataStoreVersionId - const dataStoreVersionId = entry.getVersionId() ? - entry.getEncodedVersionId() : 'null'; - let zenkoDataLocations; - if (!locations || locations.length === 0) { - zenkoDataLocations = [{ - key: entry.getObjectKey(), - size: 0, - start: 0, - dataStoreName: zenkoLocation, - dataStoreType: 'aws_s3', - dataStoreETag: `1:${emptyFileMd5}`, - dataStoreVersionId, - }]; - } else { - zenkoDataLocations = [{ - key: entry.getObjectKey(), - size: entry.getContentLength(), - start: 0, - dataStoreName: zenkoLocation, - dataStoreType: 'aws_s3', - dataStoreETag: `1:${entry.getContentMd5()}`, - dataStoreVersionId, - }]; - } - entry.setLocation(zenkoDataLocations); - } - - /** - * Update acl info on ingested object MD - * @param {ObjectQueueEntry} entry - object queue entry object - * @return {undefined} - */ - _updateAcl(entry) { - // reset acl info - const objectMDModel = new ObjectMD(); - entry.setAcl(objectMDModel.getAcl()); - } - /** * Update replication info on ingested object MD to match Zenko defined * replication info. @@ -367,29 +296,17 @@ class MongoQueueProcessor { const key = sourceEntry.getObjectKey(); const entryVersionId = extractVersionId(sourceEntry.getObjectVersionedKey()); - // Use x-amz-meta-scal-version-id if provided, instead of the actual versionId of the object. - // This should happen only for restored objects : in all other situations, both the source - // and ingested objects should have the same version id (and no x-amz-meta-scal-version-id - // metadata). const scalVersionId = sourceEntry.getOverheadField('x-amz-meta-scal-version-id'); - const versionId = scalVersionId ? VersionID.decode(scalVersionId) : entryVersionId; + const versionId = + this._mode.resolveVersionId(scalVersionId, entryVersionId); this.logger.debug('processing object delete', { bucket, key, versionId }); async.waterfall([ cb => this._getZenkoObjectMetadata(log, sourceEntry, versionId, cb), (zenkoObjMd, cb) => { - // Skip if the object is in a different location, i.e. when the delete was caused - // by restored-object expiration or transition. It works because the dataStoreName - // is updated before actually sending the object to GC to effectively delete the - // data. - const encode = versionId => (versionId ? VersionID.encode(versionId) : 'null'); - if (zenkoObjMd.dataStoreName !== location || - zenkoObjMd.location?.length !== 1 || - zenkoObjMd.location[0].dataStoreName !== location || - zenkoObjMd.location[0].key !== key || - (zenkoObjMd.location[0].dataStoreVersionId || 'null') !== encode(entryVersionId) - ) { + if (!this._mode.shouldProcessDelete(zenkoObjMd, location, key, + entryVersionId)) { log.end().info('ignore delete entry, transitioned to another location', { entry: sourceEntry.getLogInfo(), location, @@ -467,19 +384,12 @@ class MongoQueueProcessor { this.logger.debug('processing object metadata', { bucket, key, scalVersionId }); const maybeGetZenkoObjectMetadata = cb => { - // NOTE: ZenkoObjMD is used for updating replication info, as well as validating the - // `x-amz-meta-scal-version-id` header of restored objects. If the Zenko bucket does - // not have repInfo set and the header is not set, then we can skip fetching. - const bucketRepInfo = bucketInfo.getReplicationConfiguration(); - if (!scalVersionId && !bucketRepInfo?.rules?.some(r => r.enabled)) { + if (!this._mode.needsExistingMetadata(sourceEntry, bucketInfo)) { return cb(); } - // Use x-amz-meta-scal-version-id if provided, instead of the actual versionId of the object. - // This should happen only for restored objects : in all other situations, both the source - // and ingested objects should have the same version id (and not x-amz-meta-scal-version-id - // metadata). - const versionId = scalVersionId ? VersionID.decode(scalVersionId) : sourceEntry.getVersionId(); + const versionId = this._mode.resolveVersionId(scalVersionId, + sourceEntry.getVersionId()); return this._getZenkoObjectMetadata(log, sourceEntry, versionId, cb); }; @@ -494,7 +404,8 @@ class MongoQueueProcessor { return done(err); } - const content = getContentType(sourceEntry, zenkoObjMd); + const content = + this._mode.getChangedContent(sourceEntry, zenkoObjMd); if (content.length === 0) { this._normalizePendingMetric(location); log.end().debug('skipping duplicate entry', { @@ -507,16 +418,10 @@ class MongoQueueProcessor { } if (zenkoObjMd) { - // Keep existing metadata fields, only need to update the tags - const tags = sourceEntry.getTags(); - sourceEntry._data = { ...zenkoObjMd }; // eslint-disable-line no-param-reassign - sourceEntry.setTags(tags); + this._mode.mergeExistingMetadata(sourceEntry, zenkoObjMd); } else { - // Update necessary metadata fields before saving to Zenko MongoDB - this._updateOwnerMD(sourceEntry, bucketInfo); - this._updateObjectDataStoreName(sourceEntry, location); - this._updateLocations(sourceEntry, location); - this._updateAcl(sourceEntry); + this._mode.applyNewObjectMetadata(sourceEntry, location, + bucketInfo); } // Try to update replication info, if applicable diff --git a/extensions/mongoProcessor/modes/DRMode.js b/extensions/mongoProcessor/modes/DRMode.js new file mode 100644 index 000000000..1c3d0742d --- /dev/null +++ b/extensions/mongoProcessor/modes/DRMode.js @@ -0,0 +1,133 @@ +'use strict'; + +const { ObjectMD } = require('arsenal').models; + +const locations = require('../../../lib/util/locations'); +const ProcessorMode = require('./ProcessorMode'); +const getContentType = require('../utils/contentTypeHelper'); + +class DRMode extends ProcessorMode { + /** + * The stored document tells a first write from an update: a source insert + * is redelivered on replay and overlaps the bootstrap dump. + * + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {BucketInfo} bucketInfo - bucket info object + * @return {boolean} true if the stored document is needed + */ + needsExistingMetadata(entry, bucketInfo) { // eslint-disable-line no-unused-vars + return true; + } + + /** + * The ingestion diff only covers tags; a replicated object can also change + * its object-lock state and, until localized, its placement. + * + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {Object|undefined} zenkoObjMd - metadata fetched from mongo + * @return {Array} array of ReplicationInfo Content Type + */ + getChangedContent(entry, zenkoObjMd) { + const content = getContentType(entry, zenkoObjMd); + if (!zenkoObjMd || content.length !== 0) { + return content; + } + + return this._hasMutableChange(entry, zenkoObjMd) ? ['METADATA'] : []; + } + + /** + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {Object} zenkoObjMd - metadata fetched from mongo + * @return {boolean} true if the entry changes mutable metadata + */ + _hasMutableChange(entry, zenkoObjMd) { + return entry.getRetentionMode() !== zenkoObjMd.retentionMode || + entry.getRetentionDate() !== zenkoObjMd.retentionDate || + entry.getLegalHold() !== !!zenkoObjMd.legalHold || + (this._isNotLocalized(zenkoObjMd) && + entry.getDataStoreName() !== zenkoObjMd.dataStoreName); + } + + /** + * A version whose data still lives on the remote site. + * + * @param {Object} zenkoObjMd - metadata fetched from mongo + * @return {boolean} true if the stored version is not localized + */ + _isNotLocalized(zenkoObjMd) { + return locations.isCRRLocation(zenkoObjMd.dataStoreName); + } + + /** + * The source-side pipeline shaped everything this object keeps. ACLs are + * not replicated, so they are reset as they are for an ingested object. + * + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {string} location - zenko storage location name + * @param {BucketInfo} bucketInfo - bucket info object + * @return {undefined} + */ + applyNewObjectMetadata(entry, location, bucketInfo) { // eslint-disable-line no-unused-vars + entry.setAcl(new ObjectMD().getAcl()); + } + + /** + * An update brings tags and object-lock state, cleared values included. A + * localized version keeps the placement the copy engine gave it, one still + * on the remote site takes the entry's. + * + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {Object} zenkoObjMd - metadata fetched from mongo + * @return {undefined} + */ + mergeExistingMetadata(entry, zenkoObjMd) { + const tags = entry.getTags(); + const retentionMode = entry.getRetentionMode(); + const retentionDate = entry.getRetentionDate(); + const legalHold = entry.getLegalHold(); + const notLocalized = this._isNotLocalized(zenkoObjMd); + const dataStoreName = entry.getDataStoreName(); + const location = entry.getLocation(); + + entry._data = { ...zenkoObjMd }; // eslint-disable-line no-param-reassign + + entry.setTags(tags); + entry.setRetentionMode(retentionMode); + entry.setRetentionDate(retentionDate); + entry.setLegalHold(legalHold); + + if (notLocalized) { + entry.setDataStoreName(dataStoreName); + entry.setLocation(location); + } + } + + /** + * Version ids are identical on both sides, so the entry's is authoritative: + * a scal version id names a version of another system entirely. + * + * @param {string|undefined} scalVersionId - encoded scal version id + * @param {string|undefined} versionId - version id the entry carries + * @return {string|undefined} version id to act on + */ + resolveVersionId(scalVersionId, versionId) { + return versionId; + } + + /** + * A replicated object's location legitimately differs, so the ingestion + * guard would ignore every deletion. + * + * @param {Object} zenkoObjMd - metadata fetched from mongo + * @param {string} location - zenko storage location name + * @param {string} key - object key + * @param {string|undefined} versionId - version id the entry carries + * @return {boolean} true if the object should be deleted + */ + shouldProcessDelete(zenkoObjMd, location, key, versionId) { // eslint-disable-line no-unused-vars + return true; + } +} + +module.exports = DRMode; diff --git a/extensions/mongoProcessor/modes/IngestionMode.js b/extensions/mongoProcessor/modes/IngestionMode.js new file mode 100644 index 000000000..dfce04034 --- /dev/null +++ b/extensions/mongoProcessor/modes/IngestionMode.js @@ -0,0 +1,172 @@ +'use strict'; + +const { emptyFileMd5 } = require('arsenal').constants; +const { ObjectMD } = require('arsenal').models; +const { VersionID } = require('arsenal').versioning; + +const ProcessorMode = require('./ProcessorMode'); +const getContentType = require('../utils/contentTypeHelper'); + +class IngestionMode extends ProcessorMode { + /** + * ZenkoObjMD is used for updating replication info, as well as validating + * the `x-amz-meta-scal-version-id` header of restored objects. If the Zenko + * bucket does not have repInfo set and the header is not set, then we can + * skip fetching. + * + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {BucketInfo} bucketInfo - bucket info object + * @return {boolean} true if the stored document is needed + */ + needsExistingMetadata(entry, bucketInfo) { + const scalVersionId = entry.getValue()['x-amz-meta-scal-version-id']; + const bucketRepInfo = bucketInfo.getReplicationConfiguration(); + + return !!scalVersionId || !!bucketRepInfo?.rules?.some(r => r.enabled); + } + + /** + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {Object|undefined} zenkoObjMd - metadata fetched from mongo + * @return {Array} array of ReplicationInfo Content Type + */ + getChangedContent(entry, zenkoObjMd) { + return getContentType(entry, zenkoObjMd); + } + + /** + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {string} location - zenko storage location name + * @param {BucketInfo} bucketInfo - bucket info object + * @return {undefined} + */ + applyNewObjectMetadata(entry, location, bucketInfo) { + this._updateOwnerMD(entry, bucketInfo); + this._updateObjectDataStoreName(entry, location); + this._updateLocations(entry, location); + this._updateAcl(entry); + } + + /** + * Update ingested entry metadata fields: owner-id, owner-display-name + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {BucketInfo} bucketInfo - bucket info object + * @return {undefined} + */ + _updateOwnerMD(entry, bucketInfo) { + // zenko bucket owner information is being set on ingested md + entry.setOwnerDisplayName(bucketInfo.getOwnerDisplayName()); + entry.setOwnerId(bucketInfo.getOwner()); + } + + /** + * Update ingested entry metadata fields: dataStoreName + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {string} location - owner details + * @return {undefined} + */ + _updateObjectDataStoreName(entry, location) { + entry.setDataStoreName(location); + } + + /** + * Update ingested entry metadata location field. Each location change + * includes: key, dataStoreName, dataStoreType, dataStoreVersionId + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {string} zenkoLocation - zenko storage location name + * @return {undefined} + */ + _updateLocations(entry, zenkoLocation) { + const locations = entry.getLocation(); + // if version id is undefined, we have a single null object. + // To hold reference to this null object, we need to encode "null" + // as its dataStoreVersionId + const dataStoreVersionId = entry.getVersionId() ? + entry.getEncodedVersionId() : 'null'; + let zenkoDataLocations; + if (!locations || locations.length === 0) { + zenkoDataLocations = [{ + key: entry.getObjectKey(), + size: 0, + start: 0, + dataStoreName: zenkoLocation, + dataStoreType: 'aws_s3', + dataStoreETag: `1:${emptyFileMd5}`, + dataStoreVersionId, + }]; + } else { + zenkoDataLocations = [{ + key: entry.getObjectKey(), + size: entry.getContentLength(), + start: 0, + dataStoreName: zenkoLocation, + dataStoreType: 'aws_s3', + dataStoreETag: `1:${entry.getContentMd5()}`, + dataStoreVersionId, + }]; + } + entry.setLocation(zenkoDataLocations); + } + + /** + * Update acl info on ingested object MD + * @param {ObjectQueueEntry} entry - object queue entry object + * @return {undefined} + */ + _updateAcl(entry) { + // reset acl info + const objectMDModel = new ObjectMD(); + entry.setAcl(objectMDModel.getAcl()); + } + + /** + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {Object} zenkoObjMd - metadata fetched from mongo + * @return {undefined} + */ + mergeExistingMetadata(entry, zenkoObjMd) { + // Keep existing metadata fields, only need to update the tags + const tags = entry.getTags(); + entry._data = { ...zenkoObjMd }; // eslint-disable-line no-param-reassign + entry.setTags(tags); + } + + /** + * Use x-amz-meta-scal-version-id if provided, instead of the actual + * versionId of the object. This should happen only for restored objects: + * in all other situations, both the source and ingested objects should have + * the same version id (and no x-amz-meta-scal-version-id metadata). + * + * @param {string|undefined} scalVersionId - encoded scal version id + * @param {string|undefined} versionId - version id the entry carries + * @return {string|undefined} version id to act on + */ + resolveVersionId(scalVersionId, versionId) { + return scalVersionId ? VersionID.decode(scalVersionId) : versionId; + } + + /** + * Skip if the object is in a different location, i.e. when the delete was + * caused by restored-object expiration or transition. It works because the + * dataStoreName is updated before actually sending the object to GC to + * effectively delete the data. + * + * @param {Object} zenkoObjMd - metadata fetched from mongo + * @param {string} location - zenko storage location name + * @param {string} key - object key + * @param {string|undefined} versionId - version id the entry carries + * @return {boolean} true if the object should be deleted + */ + shouldProcessDelete(zenkoObjMd, location, key, versionId) { + const encode = vid => (vid ? VersionID.encode(vid) : 'null'); + + return zenkoObjMd.dataStoreName === location && + zenkoObjMd.location?.length === 1 && + zenkoObjMd.location[0].dataStoreName === location && + zenkoObjMd.location[0].key === key && + (zenkoObjMd.location[0].dataStoreVersionId || 'null') === + encode(versionId); + } +} + +module.exports = IngestionMode; diff --git a/extensions/mongoProcessor/modes/ProcessorMode.js b/extensions/mongoProcessor/modes/ProcessorMode.js new file mode 100644 index 000000000..d17ea7570 --- /dev/null +++ b/extensions/mongoProcessor/modes/ProcessorMode.js @@ -0,0 +1,107 @@ +'use strict'; + +const assert = require('assert'); + +/** + * A processor mode holds the decisions that differ between the streams the + * mongo-processor writes: out-of-band ingestion, where identity and placement + * are rewritten to local values because the source system's accounts and + * locations do not exist here, and D/R, where they are replicated and so are + * applied as they arrive. + * + * Everything else about processing an entry is shared. + */ +class ProcessorMode { + /** + * Whether the entry's existing metadata has to be read before processing. + * + * This method must be implemented by subclasses of ProcessorMode + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {BucketInfo} bucketInfo - bucket info object + * @return {boolean} true if the stored document is needed + */ + needsExistingMetadata(entry, bucketInfo) { // eslint-disable-line no-unused-vars + assert(false, + 'sub-classes of ProcessorMode must implement ' + + 'the needsExistingMetadata() method'); + } + + /** + * What changed between the entry and the object already stored, as + * replicationInfo content values. An empty list means the entry carries no + * change and is not written. + * + * This method must be implemented by subclasses of ProcessorMode + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {Object|undefined} zenkoObjMd - metadata fetched from mongo + * @return {Array} array of ReplicationInfo Content Type + */ + getChangedContent(entry, zenkoObjMd) { // eslint-disable-line no-unused-vars + assert(false, + 'sub-classes of ProcessorMode must implement ' + + 'the getChangedContent() method'); + } + + /** + * Apply the metadata fields an object gets when it is first written here. + * + * This method must be implemented by subclasses of ProcessorMode + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {string} location - zenko storage location name + * @param {BucketInfo} bucketInfo - bucket info object + * @return {undefined} + */ + applyNewObjectMetadata(entry, location, bucketInfo) { // eslint-disable-line no-unused-vars + assert(false, + 'sub-classes of ProcessorMode must implement ' + + 'the applyNewObjectMetadata() method'); + } + + /** + * Merge the entry into the object already stored, deciding which fields the + * entry brings and which the stored document keeps. + * + * This method must be implemented by subclasses of ProcessorMode + * @param {ObjectQueueEntry} entry - object queue entry object + * @param {Object} zenkoObjMd - metadata fetched from mongo + * @return {undefined} + */ + mergeExistingMetadata(entry, zenkoObjMd) { // eslint-disable-line no-unused-vars + assert(false, + 'sub-classes of ProcessorMode must implement ' + + 'the mergeExistingMetadata() method'); + } + + /** + * Which version of the object the entry acts on, given the version id it + * carries and the `x-amz-meta-scal-version-id` it may carry alongside. + * + * This method must be implemented by subclasses of ProcessorMode + * @param {string|undefined} scalVersionId - encoded scal version id + * @param {string|undefined} versionId - version id the entry carries + * @return {string|undefined} version id to act on + */ + resolveVersionId(scalVersionId, versionId) { // eslint-disable-line no-unused-vars + assert(false, + 'sub-classes of ProcessorMode must implement ' + + 'the resolveVersionId() method'); + } + + /** + * Whether a delete entry still applies to the object as stored. + * + * This method must be implemented by subclasses of ProcessorMode + * @param {Object} zenkoObjMd - metadata fetched from mongo + * @param {string} location - zenko storage location name + * @param {string} key - object key + * @param {string|undefined} versionId - decoded version id of the entry + * @return {boolean} true if the object should be deleted + */ + shouldProcessDelete(zenkoObjMd, location, key, versionId) { // eslint-disable-line no-unused-vars + assert(false, + 'sub-classes of ProcessorMode must implement ' + + 'the shouldProcessDelete() method'); + } +} + +module.exports = ProcessorMode; diff --git a/extensions/mongoProcessor/modes/index.js b/extensions/mongoProcessor/modes/index.js new file mode 100644 index 000000000..31def379c --- /dev/null +++ b/extensions/mongoProcessor/modes/index.js @@ -0,0 +1,12 @@ +const DRMode = require('./DRMode'); +const IngestionMode = require('./IngestionMode'); + +const modes = { + ingestion: IngestionMode, + dr: DRMode, +}; + +module.exports = { + modes, + defaultMode: 'ingestion', +}; diff --git a/extensions/mongoProcessor/mongoProcessorTask.js b/extensions/mongoProcessor/mongoProcessorTask.js index e0ccf0258..da5d32e03 100644 --- a/extensions/mongoProcessor/mongoProcessorTask.js +++ b/extensions/mongoProcessor/mongoProcessorTask.js @@ -18,9 +18,10 @@ const { startProbeServer } = require('../../lib/util/probe'); const kafkaConfig = config.kafka; const mConfig = config.metrics; const mongoProcessorConfig = config.extensions.mongoProcessor; -// TODO: consider whether we would want a separate mongo config -// for the consumer side -const mongoClientConfig = config.queuePopulator.mongo; +// the queue populator's mongo config is the fallback for a deployment that +// runs one beside this process; a D/R sink runs none, so it configures its own +const mongoClientConfig = + mongoProcessorConfig.mongodb ?? config.queuePopulator?.mongo; const log = new werelogs.Logger('Backbeat:MongoProcessor:task'); const mongoProcessorLogConfig = mongoProcessorConfig.log ?? config.log; @@ -63,10 +64,13 @@ async function handleMetrics(res, log) { } function loadManagementDatabase() { - const ingestionServiceAuth = config.extensions.ingestion.auth; + // the ingestion extension's auth is the fallback for a deployment that + // configures it; a D/R sink enables no extension but this one + const serviceAuth = + mongoProcessorConfig.auth ?? config.extensions.ingestion?.auth; initManagement({ serviceName: 'md-ingestion', - serviceAccount: ingestionServiceAuth.account, + serviceAccount: serviceAuth.account, }, error => { if (error) { log.error('could not load management db', { error }); diff --git a/lib/Config.js b/lib/Config.js index 2a67c9398..9fdd2a655 100644 --- a/lib/Config.js +++ b/lib/Config.js @@ -119,9 +119,11 @@ class Config extends EventEmitter { // whitelist IP, CIDR for health checks const defaultHealthChecks = ['127.0.0.1/8', '::1']; - const healthChecks = parsedConfig.server.healthChecks; - healthChecks.allowFrom = - healthChecks.allowFrom.concat(defaultHealthChecks); + const healthChecks = parsedConfig.server?.healthChecks; + if (healthChecks) { + healthChecks.allowFrom = + healthChecks.allowFrom.concat(defaultHealthChecks); + } // additional certs checks if (parsedConfig.certFilePaths) { diff --git a/lib/util/locations.js b/lib/util/locations.js new file mode 100644 index 000000000..bb7c114e2 --- /dev/null +++ b/lib/util/locations.js @@ -0,0 +1,18 @@ +const locationsConfig = require('../../conf/locationConfig.json') || {}; + +/** + * Tell whether a location holds data owned by a remote site. + * + * Data stored on such a location is remote production data: we may read it + * (e.g. to copy it locally), but we must never delete it. + * + * @param {String} dataStoreName - location name + * @return {Boolean} true if the location is a CRR (remote) location + */ +function isCRRLocation(dataStoreName) { + return Boolean(locationsConfig[dataStoreName]?.isCRR); +} + +module.exports = { + isCRRLocation, +}; diff --git a/tests/functional/ingestion/MongoQueueProcessor.js b/tests/functional/ingestion/MongoQueueProcessor.js index 626b825a8..427ae529a 100644 --- a/tests/functional/ingestion/MongoQueueProcessor.js +++ b/tests/functional/ingestion/MongoQueueProcessor.js @@ -16,6 +16,7 @@ const authdata = require('../../../conf/authdata.json'); const ObjectQueueEntry = require('../../../lib/models/ObjectQueueEntry'); const DeleteOpQueueEntry = require('../../../lib/models/DeleteOpQueueEntry'); const fakeLogger = require('../../utils/fakeLogger'); +const locations = require('../../../lib/util/locations'); const { ObjectMDArchive, LifecycleConfiguration, NotificationConfiguration } = require('arsenal/build/lib/models'); const kafkaConfig = config.kafka; @@ -1068,3 +1069,239 @@ describe('MongoQueueProcessor', function mqp() { }); }); }); + +describe('MongoQueueProcessor in dr mode', function drMode() { + this.timeout(5000); + + let mqp; + let mongoClient; + + before(() => { + mqp = new MongoQueueProcessorMock(kafkaConfig, + { ...mongoProcessorConfig, mode: 'dr' }, mongoClientConfig, mConfig); + mqp.start(); + + mongoClient = mqp._mongoClient; + }); + + afterEach(() => { + mqp.reset(); + sinon.restore(); + }); + + function processEntry(entry, next) { + return async.waterfall([ + cb => mongoClient.getBucketAttributes(BUCKET, fakeLogger, cb), + (bucketInfo, cb) => mqp._processObjectQueueEntry(fakeLogger, entry, + LOCATION, bucketInfo, cb), + ], next); + } + + it('should apply a new object as the source describes it', done => { + const key = 'dr-new-key'; + const versionKey = `${key}${VID_SEP}${VERSION_ID}`; + const objmd = new ObjectMD() + .setKey(key) + .setVersionId(VERSION_ID) + .setOwnerId('source-owner-id') + .setOwnerDisplayName('source-owner') + .setDataStoreName('cold-location') + // ACLs are not replicated: the matrix in the design resets them + .setAcl({ Canned: '', FULL_CONTROL: ['source-grantee'], + WRITE_ACP: [], READ: [], READ_ACP: [] }); + const entry = new ObjectQueueEntry(BUCKET, versionKey, objmd); + + processEntry(entry, err => { + assert.ifError(err); + + const added = mqp.getAdded(); + assert.strictEqual(added.length, 1); + const { objVal } = added[0]; + assert.strictEqual(objVal['owner-id'], 'source-owner-id'); + assert.strictEqual(objVal['owner-display-name'], 'source-owner'); + assert.strictEqual(objVal.dataStoreName, 'cold-location'); + assert.deepStrictEqual(objVal.acl, new ObjectMD().getAcl()); + done(); + }); + }); + + it('should keep the stored placement when updating a localized object', + done => { + const versionKey = `${KEY}${VID_SEP}${VERSION_ID}`; + // the copy engine has since moved the object to a local location, so + // the entry's own placement must not be written back + const objmd = new ObjectMD() + .setKey(KEY) + .setVersionId(VERSION_ID) + .setTags({ mytag: 'mytags-value' }) + .setDataStoreName('source-site') + .setLocation([{ key: KEY, dataStoreName: 'source-site' }]) + .setLegalHold(true); + const entry = new ObjectQueueEntry(BUCKET, versionKey, objmd); + + processEntry(entry, err => { + assert.ifError(err); + + const added = mqp.getAdded(); + assert.strictEqual(added.length, 1); + const { objVal } = added[0]; + // placement from the stored document + assert.strictEqual(objVal.dataStoreName, LOCATION); + assert.strictEqual(objVal.location[0].dataStoreName, LOCATION); + // mutable metadata from the entry + assert.strictEqual(objVal.legalHold, true); + done(); + }); + }); + + it('should skip a replayed entry that changes nothing', done => { + // at-least-once delivery means the same entry arrives twice; the second + // must be a no-op rather than another write + const versionKey = `${KEY}${VID_SEP}${VERSION_ID}`; + const objmd = new ObjectMD() + .setKey(KEY) + .setVersionId(VERSION_ID) + .setTags({ mytag: 'mytags-value' }) + .setDataStoreName(LOCATION); + const entry = new ObjectQueueEntry(BUCKET, versionKey, objmd); + + processEntry(entry, err => { + assert.ifError(err); + assert.strictEqual(mqp.getAdded().length, 0); + done(); + }); + }); + + it('should take the entry placement for a version still on the source', + done => { + // not localized yet, so the source still describes where the data is: + // this is the production-side archive of an already-replicated version + sinon.stub(locations, 'isCRRLocation').returns(true); + const versionKey = `${KEY}${VID_SEP}${VERSION_ID}`; + const objmd = new ObjectMD() + .setKey(KEY) + .setVersionId(VERSION_ID) + .setTags({ mytag: 'mytags-value' }) + .setDataStoreName('cold-location'); + const entry = new ObjectQueueEntry(BUCKET, versionKey, objmd); + + processEntry(entry, err => { + assert.ifError(err); + + const added = mqp.getAdded(); + assert.strictEqual(added.length, 1); + assert.strictEqual(added[0].objVal.dataStoreName, 'cold-location'); + done(); + }); + }); + + it('should not skip an update that only changes the object lock', done => { + const versionKey = `${KEY}${VID_SEP}${VERSION_ID}`; + // same tags as the stored object: the ingestion diff would call this a + // duplicate and drop the retention with it + const objmd = new ObjectMD() + .setKey(KEY) + .setVersionId(VERSION_ID) + .setTags({ mytag: 'mytags-value' }) + .setRetentionMode('GOVERNANCE') + .setRetentionDate('2099-01-01T00:00:00.000Z'); + const entry = new ObjectQueueEntry(BUCKET, versionKey, objmd); + + processEntry(entry, err => { + assert.ifError(err); + + const added = mqp.getAdded(); + assert.strictEqual(added.length, 1); + assert.strictEqual(added[0].objVal.retentionMode, 'GOVERNANCE'); + assert.strictEqual(added[0].objVal.retentionDate, + '2099-01-01T00:00:00.000Z'); + done(); + }); + }); + + it('should read the stored object even when the bucket has no ' + + 'replication configuration', done => { + const getObject = sinon.spy(mongoClient, 'getObject'); + const versionKey = `${KEY}${VID_SEP}${VERSION_ID}`; + const objmd = new ObjectMD() + .setKey(KEY) + .setVersionId(VERSION_ID) + .setTags({ mytag: 'mytags-value' }) + .setLegalHold(true); + const entry = new ObjectQueueEntry(BUCKET, versionKey, objmd); + + async.waterfall([ + cb => mongoClient.getBucketAttributes(BUCKET, fakeLogger, cb), + (bucketInfo, cb) => { + sinon.stub(bucketInfo, 'getReplicationConfiguration') + .returns(null); + return mqp._processObjectQueueEntry(fakeLogger, entry, LOCATION, + bucketInfo, cb); + }, + ], err => { + assert.ifError(err); + + sinon.assert.called(getObject); + // the merge happened, so the entry was recognised as an update + assert.strictEqual(mqp.getAdded()[0].objVal.dataStoreName, + LOCATION); + done(); + }); + }); + + it('should ignore a scal version id the source object carries', done => { + // production may itself have ingested or cold-restored this object, so + // it can carry a scal version id of its own; it names a version of the + // system production ingested from, not anything here + const versionKey = `${KEY}${VID_SEP}${VERSION_ID}`; + const objmd = new ObjectMD() + .setKey(KEY) + .setVersionId(VERSION_ID) + .setTags({ mytag: 'mytags-value' }) + .setLegalHold(true) + .setUserMetadata({ + 'x-amz-meta-scal-version-id': encode(NEW_VERSION_ID), + }); + const entry = new ObjectQueueEntry(BUCKET, versionKey, objmd); + const getObject = sinon.spy(mongoClient, 'getObject'); + + processEntry(entry, err => { + assert.ifError(err); + + // read, and written back, on the version the entry names + assert.strictEqual(getObject.getCall(0).args[2].versionId, + VERSION_ID); + assert.strictEqual(mqp.getAdded()[0].key, + `${KEY}${VID_SEP}${VERSION_ID}`); + done(); + }); + }); + + it('should delete an object whose location differs from the bucket\'s', + done => { + const versionKey = `${KEY}${VID_SEP}${VERSION_ID}`; + const entry = new DeleteOpQueueEntry(BUCKET, versionKey); + sinon.stub(mongoClient, 'getObject').callsFake((b, k, p, l, cb) => + cb(null, new ObjectMD() + .setKey(KEY) + .setVersionId(VERSION_ID) + // cold, or still pointing at the source: never the bucket's + // own location constraint + .setDataStoreName('cold-location') + .setLocation(null) + ._data)); + + async.waterfall([ + cb => mongoClient.getBucketAttributes(BUCKET, fakeLogger, cb), + (bucketInfo, cb) => mqp._processDeleteOpQueueEntry(fakeLogger, + entry, LOCATION, bucketInfo, cb), + ], err => { + assert.ifError(err); + + const deleted = mqp.getDeleted(); + assert.strictEqual(deleted.length, 1); + assert.strictEqual(deleted[0].versionId, VERSION_ID); + done(); + }); + }); +}); diff --git a/tests/unit/lib/config/Config.spec.js b/tests/unit/lib/config/Config.spec.js index 67d6ba047..f8cbc6af2 100644 --- a/tests/unit/lib/config/Config.spec.js +++ b/tests/unit/lib/config/Config.spec.js @@ -31,6 +31,12 @@ describe('Config', () => { assert.doesNotThrow(() => config._parseConfig(testConfig)); }); + it('should accept a config serving no API, as a D/R sink does', () => { + delete testConfig.server; + testConfig.extensions = { mongoProcessor: testConfig.extensions.mongoProcessor }; + assert.doesNotThrow(() => config._parseConfig(testConfig)); + }); + it('should throw an error when dataMoverTopic is not provided and transition is supported', () => { delete testConfig.extensions.replication.dataMoverTopic; testConfig.extensions.lifecycle.supportedLifecycleRules = [ diff --git a/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js b/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js index 9723c20a4..f48a80067 100644 --- a/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js +++ b/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js @@ -9,6 +9,17 @@ const baseExtConfig = { probeServer: { port: 4000 }, }; +const serviceAuth = { type: 'service', account: 'service-md-ingestion' }; + +const mongodb = { + replicaSetHosts: 'mongo:27017', + database: 'datadb', + authCredentials: { username: 'u', password: 'p' }, +}; + +// what a D/R sink configures: no other extension supplies these +const drExtConfig = { ...baseExtConfig, mode: 'dr', auth: serviceAuth, mongodb }; + // the validated backbeat config, passed to every extension validator const globalConfig = { log: { logLevel: 'info', dumpLevel: 'trace' } }; @@ -22,7 +33,7 @@ describe('MongoProcessorConfigValidator log override', () => { }); it('should leave log undefined when not set, deferring to global config.log', () => { - const validated = configValidator(globalConfig, baseExtConfig); + const validated = configValidator(globalConfig, { ...baseExtConfig }); assert.strictEqual(validated.log, undefined); }); @@ -41,3 +52,118 @@ describe('MongoProcessorConfigValidator log override', () => { } }); }); + +describe('MongoProcessorConfigValidator mongodb', () => { + it('should accept a mongodb client config', () => { + const validated = configValidator(globalConfig, { ...baseExtConfig, mongodb }); + assert.strictEqual(validated.mongodb.replicaSetHosts, 'mongo:27017'); + assert.strictEqual(validated.mongodb.database, 'datadb'); + assert.deepStrictEqual(validated.mongodb.authCredentials, + { username: 'u', password: 'p' }); + }); + + it('should leave mongodb undefined when not set, deferring to the ' + + 'queue populator config', () => { + const validated = configValidator(globalConfig, { ...baseExtConfig }); + assert.strictEqual(validated.mongodb, undefined); + }); + + it('should require mongodb in dr mode, which runs no queue populator', () => { + let err; + try { + const noMongo = { ...drExtConfig }; + delete noMongo.mongodb; + configValidator(globalConfig, noMongo); + } catch (e) { + err = e; + } + assert(err, 'expected configValidator to throw on dr mode with no mongodb'); + assert.match(err.message, /mongodb/); + }); + + it('should reject credentials missing a password', () => { + let err; + try { + configValidator(globalConfig, { + ...baseExtConfig, + mongodb: { ...mongodb, authCredentials: { username: 'u' } }, + }); + } catch (e) { + err = e; + } + assert(err, 'expected configValidator to throw on partial credentials'); + }); +}); + +describe('MongoProcessorConfigValidator mode', () => { + it('should default to ingestion so an existing config is unchanged', () => { + const validated = configValidator(globalConfig, { ...baseExtConfig }); + assert.strictEqual(validated.mode, 'ingestion'); + }); + + it('should accept the dr mode', () => { + const validated = configValidator(globalConfig, { ...drExtConfig }); + assert.strictEqual(validated.mode, 'dr'); + }); + + it('should accept the mode from the environment', () => { + process.env.EXTENSIONS_MONGO_PROCESSOR_MODE = 'dr'; + try { + const validated = configValidator(globalConfig, + { ...baseExtConfig, auth: serviceAuth, mongodb }); + assert.strictEqual(validated.mode, 'dr'); + } finally { + delete process.env.EXTENSIONS_MONGO_PROCESSOR_MODE; + } + }); + + it('should reject a mode with no implementation', () => { + let err; + try { + configValidator(globalConfig, { ...baseExtConfig, mode: 'sideways' }); + } catch (e) { + err = e; + } + assert(err, 'expected configValidator to throw on an unknown mode'); + assert.match(err.message, /mode/); + }); +}); + +describe('MongoProcessorConfigValidator auth', () => { + it('should accept a service account', () => { + const validated = configValidator(globalConfig, + { ...baseExtConfig, auth: serviceAuth }); + assert.strictEqual(validated.auth.type, 'service'); + assert.strictEqual(validated.auth.account, 'service-md-ingestion'); + }); + + it('should leave auth undefined when not set, deferring to the ' + + 'ingestion extension config', () => { + const validated = configValidator(globalConfig, { ...baseExtConfig }); + assert.strictEqual(validated.auth, undefined); + }); + + it('should require auth in dr mode, which enables no other extension', () => { + let err; + try { + const noAuth = { ...drExtConfig }; + delete noAuth.auth; + configValidator(globalConfig, noAuth); + } catch (e) { + err = e; + } + assert(err, 'expected configValidator to throw on dr mode with no auth'); + assert.match(err.message, /auth/); + }); + + it('should reject a service auth missing the account', () => { + let err; + try { + configValidator(globalConfig, { ...baseExtConfig, auth: { type: 'service' } }); + } catch (e) { + err = e; + } + assert(err, 'expected configValidator to throw on a service auth with no account'); + assert.match(err.message, /account/); + }); +}); diff --git a/tests/unit/mongoProcessor/ProcessorMode.spec.js b/tests/unit/mongoProcessor/ProcessorMode.spec.js new file mode 100644 index 000000000..5e385260a --- /dev/null +++ b/tests/unit/mongoProcessor/ProcessorMode.spec.js @@ -0,0 +1,24 @@ +'use strict'; + +const assert = require('assert'); + +const ProcessorMode = + require('../../../extensions/mongoProcessor/modes/ProcessorMode'); + +describe('ProcessorMode', () => { + const mode = new ProcessorMode(); + + [ + 'needsExistingMetadata', + 'getChangedContent', + 'applyNewObjectMetadata', + 'mergeExistingMetadata', + 'resolveVersionId', + 'shouldProcessDelete', + ].forEach(method => it(`should refuse to ${method}() without an ` + + 'implementation', () => { + assert.throws(() => mode[method](), + new RegExp('sub-classes of ProcessorMode must implement the ' + + `${method}\\(\\) method`)); + })); +});