diff --git a/lib/CollectionManager.js b/lib/CollectionManager.js index 06c521d..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,20 +80,27 @@ 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; } + 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; 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[mapObject.parentId], + _versionType: mapObject.versionType, + _version: versioning.getVersionAsInteger(nextObject[mapObject.versionField]) } }); bulkOp.push(nextObject); diff --git a/lib/elasticManager.js b/lib/elasticManager.js index 7d6a228..b646486 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,120 @@ class ElasticManager { this.bulkSize = bulkSize; 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 + loadESMappings() { + 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); + 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) { if (!this.interval) { @@ -44,15 +157,21 @@ class ElasticManager { const esId = changeStreamObj.fullDocument._id.toString(); // convert mongo ObjectId to string delete changeStreamObj.fullDocument._id; const esReadyDoc = changeStreamObj.fullDocument; + const mapObject = this.mappings[changeStreamObj.ns.coll]; + + 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: this.mappings[changeStreamObj.ns.coll].index, - _type: this.mappings[changeStreamObj.ns.coll].type, + _index: mapObject.index, + _type: mapObject.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[mapObject.parentId], + _versionType: mapObject.versionType, + _version: versioning.getVersionAsInteger(esReadyDoc[mapObject.versionField]) } }); @@ -64,11 +183,18 @@ 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); } else if(transformations) { - return jsonpatch.apply(esReadyDoc, transformations); + return jsonpatch.apply(esReadyDoc, transformations); } else { return esReadyDoc; @@ -77,18 +203,19 @@ class ElasticManager { async deleteDoc(changeStreamObj) { const esId = changeStreamObj.documentKey._id.toString(); // convert mongo ObjectId to string + const mapObject = 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(mapObject, 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: mapObject.index, + _type: mapObject.type, _id: esId, _parent: parentId, - _versionType: this.mappings[changeStreamObj.ns.coll].versionType, + _versionType: mapObject.versionType, _version: versioning.incrementVersionForDeletion(version), } }); @@ -136,14 +263,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) } }); @@ -159,27 +288,6 @@ class ElasticManager { return numDeleted; } - 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; - } - if (!this.mappings[collection].type) { - this.mappings[collection].type = this.mappings.default.type; - if (this.mappings[collection].type === "$self") - this.mappings[collection].type = collection; - } - if (this.mappings[collection].transformations) { - this.mappings[collection].transformFunc = jsonpatch.compile(this.mappings[collection].transformations); - } - } - - 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 4461941..6eacf9c 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": { @@ -25,6 +25,8 @@ "elasticsearch": "14.1.0", "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)