From 758b535bcf776e98bd4d42782e5228bcd2a4fd96 Mon Sep 17 00:00:00 2001 From: Nick Knol Date: Wed, 17 Oct 2018 09:58:35 -0700 Subject: [PATCH 1/3] Created default parent handling --- lib/CollectionManager.js | 16 ++++++++++------ lib/elasticManager.js | 33 ++++++++++++++++++++------------- package.json | 2 +- 3 files changed, 31 insertions(+), 20 deletions(-) diff --git a/lib/CollectionManager.js b/lib/CollectionManager.js index 06c521d..1e322b2 100644 --- a/lib/CollectionManager.js +++ b/lib/CollectionManager.js @@ -82,16 +82,20 @@ class CollectionManager { break; } + const mapObj = CollectionManager.elasticManager.mappings[this.collectionName]; + // defaults to the parentId loaded into the config, but tries to guess what the field should be passed on the parent name + const parentIdFieldName = mapObj.parentId || (mapObj._parent && mapObj._parent.type && `${mapObj._parent.type}Id`); + this.dumpProgress.token = nextObject._id; delete nextObject._id; bulkOp.push({ index: { - _index: CollectionManager.elasticManager.mappings[this.collectionName].index, - _type: CollectionManager.elasticManager.mappings[this.collectionName].type, - _id: this.dumpProgress.token, - _parent: nextObject[CollectionManager.elasticManager.mappings[this.collectionName].parentId], - _versionType: CollectionManager.elasticManager.mappings[this.collectionName].versionType, - _version: versioning.getVersionAsInteger(nextObject[CollectionManager.elasticManager.mappings[this.collectionName].versionField]) + _index: mapObject.index, + _type: mapObject.type, + _id: this.dumpProgress.token, + _parent: nextObject[parentIdFieldName], + _versionType: mapObject.versionType, + _version: versioning.getVersionAsInteger(nextObject[mapObject.versionField]) } }); bulkOp.push(nextObject); diff --git a/lib/elasticManager.js b/lib/elasticManager.js index 7d6a228..ec48fe3 100644 --- a/lib/elasticManager.js +++ b/lib/elasticManager.js @@ -44,15 +44,19 @@ class ElasticManager { const esId = changeStreamObj.fullDocument._id.toString(); // convert mongo ObjectId to string delete changeStreamObj.fullDocument._id; const esReadyDoc = changeStreamObj.fullDocument; + const mapObj = this.mappings[changeStreamObj.ns.coll]; + + // defaults to the parentId loaded into the config, but tries to guess what the field should be passed on the parent name + const parentIdFieldName = mapObj.parentId || (mapObj._parent && mapObj._parent.type && `${mapObj._parent.type}Id`); this.bulkOp.push({ index: { - _index: this.mappings[changeStreamObj.ns.coll].index, - _type: this.mappings[changeStreamObj.ns.coll].type, + _index: mapObj.index, + _type: mapObj.type, _id: esId, - _parent: esReadyDoc[this.mappings[changeStreamObj.ns.coll].parentId], - _versionType: this.mappings[changeStreamObj.ns.coll].versionType, - _version: versioning.getVersionAsInteger(esReadyDoc[this.mappings[changeStreamObj.ns.coll].versionField]) + _parent: esReadyDoc[parentIdFieldName], + _versionType: mapObj.versionType, + _version: versioning.getVersionAsInteger(esReadyDoc[mapObj.versionField]) } }); @@ -68,7 +72,7 @@ class ElasticManager { return transformFunc(esReadyDoc); } else if(transformations) { - return jsonpatch.apply(esReadyDoc, transformations); + return jsonpatch.apply(esReadyDoc, transformations); } else { return esReadyDoc; @@ -77,18 +81,19 @@ class ElasticManager { async deleteDoc(changeStreamObj) { const esId = changeStreamObj.documentKey._id.toString(); // convert mongo ObjectId to string + const mapObj = this.mappings[changeStreamObj.ns.coll]; - const { parentId, version } = await this.getExistingDoc(this.mappings[changeStreamObj.ns.coll], esId).catch((err) => { + const { parentId, version } = await this.getExistingDoc(mapObj, esId).catch((err) => { logger.error(`error finding existing document in delete: ${err}`); }); this.bulkOp.push({ delete: { - _index: this.mappings[changeStreamObj.ns.coll].index, - _type: this.mappings[changeStreamObj.ns.coll].type, + _index: mapObj.index, + _type: mapObj.type, _id: esId, _parent: parentId, - _versionType: this.mappings[changeStreamObj.ns.coll].versionType, + _versionType: mapObj.versionType, _version: versioning.incrementVersionForDeletion(version), } }); @@ -136,14 +141,16 @@ class ElasticManager { for (let i = 0; i < Math.ceil(searchResponse.hits.total / this.bulkSize); i++) { const bulkDelete = []; const dumpDocs = searchResponse.hits.hits; + const mapObject = this.mappings[collectionName]; + for (let j = 0; j < dumpDocs.length; j++) { bulkDelete.push({ delete: { - _index: this.mappings[collectionName].index, - _type: this.mappings[collectionName].type, + _index: mapObject.index, + _type: mapObject.type, _id: dumpDocs[j]._id, _parent: dumpDocs[j]._parent, - _versionType: this.mappings[collectionName].versionType, + _versionType: mapObject.versionType, _version: versioning.incrementVersionForDeletion(dumpDocs[j]._version) } }); diff --git a/package.json b/package.json index 4461941..72e32cc 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "mongo-stream", - "version": "0.1.0", + "version": "1.4.3", "description": "Sync data from Mongodb to Elasticsearch", "main": "server.js", "scripts": { From 6bd72e45a6e7ef3f534f8c2f2529b6b9f9091265 Mon Sep 17 00:00:00 2001 From: Nick Knol Date: Fri, 19 Oct 2018 06:42:39 -0700 Subject: [PATCH 2/3] fixed race condition with async/await --- lib/CollectionManager.js | 10 +++--- lib/elasticManager.js | 71 +++++++++++++++++++++++++++++----------- package.json | 1 + 3 files changed, 58 insertions(+), 24 deletions(-) diff --git a/lib/CollectionManager.js b/lib/CollectionManager.js index 1e322b2..d5c701e 100644 --- a/lib/CollectionManager.js +++ b/lib/CollectionManager.js @@ -82,9 +82,11 @@ class CollectionManager { break; } - const mapObj = CollectionManager.elasticManager.mappings[this.collectionName]; - // defaults to the parentId loaded into the config, but tries to guess what the field should be passed on the parent name - const parentIdFieldName = mapObj.parentId || (mapObj._parent && mapObj._parent.type && `${mapObj._parent.type}Id`); + const mapObject = CollectionManager.elasticManager.mappings[this.collectionName]; + if (mapObject.parentId && !nextObject[mapObject.parentId]) { + logger.debug(`Skipping insert of ${this.collectionName}:${nextObject._id} -- parentId('${mapObject.parentId}') specified but could not find one in record`) + break; + } this.dumpProgress.token = nextObject._id; delete nextObject._id; @@ -93,7 +95,7 @@ class CollectionManager { _index: mapObject.index, _type: mapObject.type, _id: this.dumpProgress.token, - _parent: nextObject[parentIdFieldName], + _parent: nextObject[mapObject.parentId], _versionType: mapObject.versionType, _version: versioning.getVersionAsInteger(nextObject[mapObject.versionField]) } diff --git a/lib/elasticManager.js b/lib/elasticManager.js index ec48fe3..3c9760f 100644 --- a/lib/elasticManager.js +++ b/lib/elasticManager.js @@ -2,6 +2,7 @@ const elasticsearch = require('elasticsearch'); const logger = new (require('service-logger'))(__filename); const versioning = require('./versioning'); const jsonpatch = require('json-patch'); +const _ = require('lodash'); class ElasticManager { constructor(elasticOpts, mappings, bulkSize) { @@ -10,8 +11,27 @@ class ElasticManager { this.bulkSize = bulkSize; this.bulkOp = []; this.interval = null; + this.esMappings = {}; } + //Gets elasticsearch mappings for the purposes of initilizing any parentId's + getESMappings() { + if (!_.isEmpty(this.esMappings)) { + return Promise.resolve(this.esMappings); + } + + return this.esClient.indices.getMapping().then((rez) => { + this.esMappings = rez; + logger.info('ES mappings retrieved'); + return this.esMappings; + }).catch((err) => { + logger.error(`Error Connecting to ES using ${elasticOpts} to get mappings: `+ err); + this.esMappings = {}; + return this.esMappings; + }); + } + + // Calls the appropriate replication function based on the change object parsed from a change stream replicate(change) { if (!this.interval) { @@ -44,19 +64,21 @@ class ElasticManager { const esId = changeStreamObj.fullDocument._id.toString(); // convert mongo ObjectId to string delete changeStreamObj.fullDocument._id; const esReadyDoc = changeStreamObj.fullDocument; - const mapObj = this.mappings[changeStreamObj.ns.coll]; + const mapObject = this.mappings[changeStreamObj.ns.coll]; - // defaults to the parentId loaded into the config, but tries to guess what the field should be passed on the parent name - const parentIdFieldName = mapObj.parentId || (mapObj._parent && mapObj._parent.type && `${mapObj._parent.type}Id`); + if (mapObject.parentId && !esReadyDoc[mapObject.parentId]) { + logger.info(`Skipping insert of ${changeStreamObj.ns.coll}:${esId} -- parentId('${mapObject.parentId}') specified but could not find one in record`) + return; + } this.bulkOp.push({ index: { - _index: mapObj.index, - _type: mapObj.type, + _index: mapObject.index, + _type: mapObject.type, _id: esId, - _parent: esReadyDoc[parentIdFieldName], - _versionType: mapObj.versionType, - _version: versioning.getVersionAsInteger(esReadyDoc[mapObj.versionField]) + _parent: esReadyDoc[mapObject.parentId], + _versionType: mapObject.versionType, + _version: versioning.getVersionAsInteger(esReadyDoc[mapObject.versionField]) } }); @@ -81,19 +103,19 @@ class ElasticManager { async deleteDoc(changeStreamObj) { const esId = changeStreamObj.documentKey._id.toString(); // convert mongo ObjectId to string - const mapObj = this.mappings[changeStreamObj.ns.coll]; + const mapObject = this.mappings[changeStreamObj.ns.coll]; - const { parentId, version } = await this.getExistingDoc(mapObj, esId).catch((err) => { + const { parentId, version } = await this.getExistingDoc(mapObject, esId).catch((err) => { logger.error(`error finding existing document in delete: ${err}`); }); this.bulkOp.push({ delete: { - _index: mapObj.index, - _type: mapObj.type, + _index: mapObject.index, + _type: mapObject.type, _id: esId, _parent: parentId, - _versionType: mapObj.versionType, + _versionType: mapObject.versionType, _version: versioning.incrementVersionForDeletion(version), } }); @@ -166,24 +188,33 @@ class ElasticManager { return numDeleted; } - setMappings(collection) { + async setMappings(collection) { // set up mappings between mongo and elastic if they do not yet exist if (!this.mappings[collection]) { this.mappings[collection] = {}; } if (!this.mappings[collection].index) { - this.mappings[collection].index = this.mappings.default.index; - if (this.mappings[collection].index === "$self") - this.mappings[collection].index = collection; + this.mappings[collection].index = (this.mappings.default.index === "$self") ? collection : this.mappings.default.index; } if (!this.mappings[collection].type) { - this.mappings[collection].type = this.mappings.default.type; - if (this.mappings[collection].type === "$self") - this.mappings[collection].type = collection; + this.mappings[collection].type = (this.mappings.default.type === "$self") ? collection : this.mappings.default.type; } if (this.mappings[collection].transformations) { this.mappings[collection].transformFunc = jsonpatch.compile(this.mappings[collection].transformations); } + + // Tries to guess the parentId field name based on parent type, can be explicitly overriden by specifying 'parent' & 'parentId' in the config json loaded on startup + // Note: 'parentId' property name should really be 'parentIdField' but keeping it as is for backwards-compatability + if (!this.mappings[collection].parent) { + const esMappings = await this.getESMappings(); + const parent = _.get(esMappings, `${this.mappings[collection].index}.mappings.${this.mappings[collection].type}._parent.type`); + const parentProp = _.get(esMappings, `${this.mappings[collection].index}.mappings.${this.mappings[collection].type}.properties.${parent}Id`); + + if (parent && parentProp) { + this.mappings[collection].parent = parent; + this.mappings[collection].parentId = `${parent}Id`; + } + } } diff --git a/package.json b/package.json index 72e32cc..dab94df 100644 --- a/package.json +++ b/package.json @@ -25,6 +25,7 @@ "elasticsearch": "14.1.0", "express": "4.16.2", "json-patch": "0.7.0", + "lodash": "4.17.4", "mongodb": "3.1.0", "service-logger": "1.0.3" } From cfc6831c75e6ed56e4c51d8dea91db95795f5d1a Mon Sep 17 00:00:00 2001 From: Nick Knol Date: Fri, 27 Sep 2019 07:56:29 -0700 Subject: [PATCH 3/3] WIP: attempting to load ES mappings automatically and filter the properties of the inserted records based off of those mappings --- lib/CollectionManager.js | 7 +- lib/elasticManager.js | 134 +++++++++++++++++++++++++++++---------- lib/mongo-stream.js | 15 ++++- package.json | 1 + server.js | 38 ++++++++++- 5 files changed, 156 insertions(+), 39 deletions(-) diff --git a/lib/CollectionManager.js b/lib/CollectionManager.js index d5c701e..30ed80b 100644 --- a/lib/CollectionManager.js +++ b/lib/CollectionManager.js @@ -4,13 +4,15 @@ const DumpProgress = require('./dumpProgress'); const versioning = require('./versioning'); class CollectionManager { - constructor(collection) { + constructor(collection, loadESMappings) { this.collection = CollectionManager.db.collection(collection); this.collectionName = collection; this.resumeToken = new ResumeToken(collection); this.dumpProgress = new DumpProgress(collection); - CollectionManager.elasticManager.setMappings(collection); + if (loadESMappings) { + CollectionManager.elasticManager.setMappings(collection); + } } static initializeStaticVariables({db, elasticManager, dumpProgress, resumeToken}) { @@ -78,6 +80,7 @@ class CollectionManager { } nextObject = await cursor.next().catch(err => logger.error('next object error', err)); + nextObject = CollectionManager.elasticManager.transformDoc(this.collectionName, nextObject); if (nextObject === null) { break; } diff --git a/lib/elasticManager.js b/lib/elasticManager.js index 3c9760f..b646486 100644 --- a/lib/elasticManager.js +++ b/lib/elasticManager.js @@ -12,10 +12,64 @@ class ElasticManager { this.bulkOp = []; this.interval = null; this.esMappings = {}; + this.esFields = {}; } +/* +* Note to future Nick, I have no idea where I left off in this commit but I've got to save this and move on. +* TODO: Figure out why I wrote seperate "fields" and "mappings" functions, shouldn't they always be loaded at the same time? +* It looks like I'm calling pieces of them seperately but no idea why. Also loadESAll seems to be redundant with +* setMappings. One is probably a newer way of doing it that I was working on but I can't remember which. +*/ + + async loadESAll() { + this.esFields = {}; + this.esMappings = {}; + this.mappings = {}; + + await this.loadESFields(); + await this.loadESMappings(); + + _.forEach(this.esMappings, (val, index) => { + _.forEach(val.mappings, (mapObject, collection) => { + + const mapObject = _.get(this.esFields, `${index}.mappings.${collection}`, {}); + const fields = _.filter(_.keys(mapObject), (k) => {return !k.match(/(keyword|phonetic|raw)$/) && !k.match(/^_/)}); + + const parent = _.get(esMappings, `${index}.mappings.${collection}._parent.type`); + const parentProp = _.get(esMappings, `${index}.mappings.${collection}._parent.type.properties.${parent}Id`); + + this.mappings[collection] = { + index: index, + type: collection, + fields: fields, + parent: (parent && parentProp) ? parent : null, + parentId: (parent && parentProp) ? `${parent}Id` : null + }; + }); + }); + + return this.mappings; + } + + loadESFields() { + if (!_.isEmpty(this.esFields)) { + return Promise.resolve(this.esFields); + } + + return this.esClient.indices.getFieldMapping({fields: '*'}).then((rez) => { + this.esFields = rez; + logger.info('ES fields retrieved'); + return this.esFields; + }).catch((err) => { + logger.error(`Error Connecting to ES to get fields: `+ err); + return this.esFields; + }); + } + + //Gets elasticsearch mappings for the purposes of initilizing any parentId's - getESMappings() { + loadESMappings() { if (!_.isEmpty(this.esMappings)) { return Promise.resolve(this.esMappings); } @@ -26,11 +80,50 @@ class ElasticManager { return this.esMappings; }).catch((err) => { logger.error(`Error Connecting to ES using ${elasticOpts} to get mappings: `+ err); - this.esMappings = {}; return this.esMappings; }); } + getCollections() { + return _.keys(this.mappings); + } + + async setMappings(collection) { + // set up mappings between mongo and elastic if they do not yet exist + if (!this.mappings[collection]) { + this.mappings[collection] = {}; + } + if (!this.mappings[collection].index) { + this.mappings[collection].index = (this.mappings.default.index === "$self") ? collection : this.mappings.default.index; + } + if (!this.mappings[collection].type) { + this.mappings[collection].type = (this.mappings.default.type === "$self") ? collection : this.mappings.default.type; + } + if (this.mappings[collection].transformations) { + this.mappings[collection].transformFunc = jsonpatch.compile(this.mappings[collection].transformations); + } + + // Tries to guess the parentId field name based on parent type, can be explicitly overriden by specifying 'parent' & 'parentId' in the config json loaded on startup + // Note: 'parentId' property name should really be 'parentIdField' but keeping it as is for backwards-compatability + if (!this.mappings[collection].parent) { + const esMappings = await this.loadESMappings(); + const parent = _.get(esMappings, `${this.mappings[collection].index}.mappings.${this.mappings[collection].type}._parent.type`); + const parentProp = _.get(esMappings, `${this.mappings[collection].index}.mappings.${this.mappings[collection].type}.properties.${parent}Id`); + + if (parent && parentProp) { + this.mappings[collection].parent = parent; + this.mappings[collection].parentId = `${parent}Id`; + } + } + + if (!this.mappings[collection].fields) { + const esFields = await this.loadESFields(); + const mapObject = _.get(esFields, `${this.mappings[collection].index}.mappings.${this.mappings[collection].type}`, {}); + + this.mappings[collection].fields = _.filter(_.keys(mapObject), (k) => {return !k.match(/(keyword|phonetic|raw)$/) && !k.match(/^_/)}); + } + } + // Calls the appropriate replication function based on the change object parsed from a change stream replicate(change) { @@ -90,6 +183,13 @@ class ElasticManager { transformDoc(collName, esReadyDoc) { const transformFunc = this.mappings[collName].transformFunc; const transformations = this.mappings[collName].transformations; + + // Filters out any properties in the inserted object not in esMappings. + // TODO: This should probably be condfigurable + if (!_.isEmpty(this.mappings[collName].fields)) { + esReadyDoc = _.pick(esReadyDoc, this.mappings[collName].fields); + } + if(transformFunc) { return transformFunc(esReadyDoc); } @@ -188,36 +288,6 @@ class ElasticManager { return numDeleted; } - async setMappings(collection) { - // set up mappings between mongo and elastic if they do not yet exist - if (!this.mappings[collection]) { - this.mappings[collection] = {}; - } - if (!this.mappings[collection].index) { - this.mappings[collection].index = (this.mappings.default.index === "$self") ? collection : this.mappings.default.index; - } - if (!this.mappings[collection].type) { - this.mappings[collection].type = (this.mappings.default.type === "$self") ? collection : this.mappings.default.type; - } - if (this.mappings[collection].transformations) { - this.mappings[collection].transformFunc = jsonpatch.compile(this.mappings[collection].transformations); - } - - // Tries to guess the parentId field name based on parent type, can be explicitly overriden by specifying 'parent' & 'parentId' in the config json loaded on startup - // Note: 'parentId' property name should really be 'parentIdField' but keeping it as is for backwards-compatability - if (!this.mappings[collection].parent) { - const esMappings = await this.getESMappings(); - const parent = _.get(esMappings, `${this.mappings[collection].index}.mappings.${this.mappings[collection].type}._parent.type`); - const parentProp = _.get(esMappings, `${this.mappings[collection].index}.mappings.${this.mappings[collection].type}.properties.${parent}Id`); - - if (parent && parentProp) { - this.mappings[collection].parent = parent; - this.mappings[collection].parentId = `${parent}Id`; - } - } - } - - sendBulkRequest(bulkOp) { if (bulkOp.length === 0) { return; diff --git a/lib/mongo-stream.js b/lib/mongo-stream.js index 42ad6a5..05a265e 100644 --- a/lib/mongo-stream.js +++ b/lib/mongo-stream.js @@ -41,13 +41,15 @@ class MongoStream { await db.createCollection('init'); // workaround for "MongoError: cannot open $changeStream for non-existent database" await db.dropCollection('init'); - const elasticManager = new ElasticManager(options.elasticOpts, options.mappings, options.bulkSize, options.parentChildRelations); + // TODO: make the mappings and bulkSize parameters part of options.elasticOpts ?? + const elasticManager = new ElasticManager(options.elasticOpts, options.mappings, options.bulkSize); const resumeTokenInterval = options.resumeTokenInterval; const mongoStream = new MongoStream(elasticManager, db, resumeTokenInterval); const managerOptions = { dump: options.dumpOnStart, ignoreResumeTokens: options.ignoreResumeTokensOnStart, - watch: true + watch: true, // TODO: why is this hardcoded here? + loadESMappings: options.loadESMappings }; CollectionManager.initializeStaticVariables({ @@ -69,13 +71,20 @@ class MongoStream { }); } + async syncCollectionManagers(options) { + await this.elasticManager.loadESAll(); + await mongoStream.removeCollectionManager( Object.keys(this.collectionManagers) ); + await mongoStream.addCollectionManager( this.elasticManager.getCollections(), options); + return Object.keys(this.collectionManagers); + } + // accepts single collection or array async addCollectionManager(collections, options) { if (!Array.isArray(collections)) collections = [collections]; await this.removeCollectionManager(collections); for (const collection of collections) { - const collectionManager = new CollectionManager(collection); + const collectionManager = new CollectionManager(collection, options.loadESMappings); if (options.dump) { await collectionManager.dumpProgress.get(); diff --git a/package.json b/package.json index dab94df..6eacf9c 100644 --- a/package.json +++ b/package.json @@ -26,6 +26,7 @@ "express": "4.16.2", "json-patch": "0.7.0", "lodash": "4.17.4", + "lodash-deep": "2.0.0", "mongodb": "3.1.0", "service-logger": "1.0.3" } diff --git a/server.js b/server.js index 171595d..e0f2168 100755 --- a/server.js +++ b/server.js @@ -26,11 +26,44 @@ app.get('/', (request, response) => { response.send(responseBody); }); +// TODO: completely untested +app.post('/sync-managers', (request, response) => { + const collectionsSynced = mongoStream.syncCollectionManagers(request.body); + response.send(collectionsSynced); +}); + // returns the mappings of all collectionManagers currently running -app.get('/mappings', (request, response) => { +app.get('/es-mappings', (request, response) => { response.send(mongoStream.elasticManager.mappings); }); +app.get('/es-mappings/:collection', (request, response) => { + response.send(mongoStream.elasticManager.mappings[request.params.collection]); +}); + +app.get('/load-es-mappings', (request, response) => { + mongoStream.elasticManager.loadESFields().then((r) => response.send(r)); +}); + +// TODO: this probably is skipping some steps +app.post('/load-es-mappings', (request, response) => { + logger.info(request.body); + + _.forEach(request.body.mappings, (colMap, col) => { + // mongoStream.elasticManager.mappings[col].fields = []; + const fields = []; + + _.deepMapValues(colMap.properties, (val,path) => { + fields.push(path.replace(/properties\./g, '').replace(/\.\w+$/, '') ); + }); + + mongoStream.elasticManager.mappings[col].fields = _.filter(_.uniq(fields), (f) => !f.includes('_')).sort(); + }); + + response.send(_.mapValues(mongoStream.elasticManager.mappings, 'fields')); +}); + + app.post('/collection-manager?', (request, response) => { logger.info(request.body); const collections = request.body.collections; @@ -38,7 +71,8 @@ app.post('/collection-manager?', (request, response) => { dump: request.body.dump, ignoreResumeTokens: request.body.ignoreResumeTokens, ignoreDumpProgress: request.body.ignoreDumpProgress, - watch: request.body.watch + watch: request.body.watch, + loadESMappings: request.body.loadESMappings }; return mongoStream.addCollectionManager(collections, managerOptions)