[WIP][MINOR]DefaultParentMapping

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);

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;

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();
 

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"
   }