Merge pull request #6009 from Countly/feature/ingestion

Feature/ingestion

Author: ar2rsawseen

api/aggregator/usage.js

@@ -147,6 +147,20 @@ usage.processSessionFromStream = function(token, currEvent, params) {
 };
 
 usage.processSessionMetricsFromStream = function(currEvent, uniqueLevelsZero, uniqueLevelsMonth, params) {
+
+    /**
+         * 
+         * @param {string} id - document id 
+         * @param {function} callback  - calback function
+         */
+    function fetchMeta(id, callback) {
+        common.readBatcher.getOne(metaToFetch[id].coll, {'_id': metaToFetch[id].id}, {meta_v2: 1}, (err, metaDoc) => {
+            var retObj = metaDoc || {};
+            retObj.coll = metaToFetch[id].coll;
+            callback(null, retObj);
+        });
+    }
+
     var isNewUser = true;
     var userProps = {};
     if (currEvent.sg && currEvent.sg.prev_session) {
@@ -160,6 +174,7 @@ usage.processSessionMetricsFromStream = function(currEvent, uniqueLevelsZero, un
     var dateIds = common.getDateIds(params);
     var metaToFetch = {};
     if (plugins.getConfig("api", params.app && params.app.plugins, true).metric_limit > 0) {
+        var postfix;
         for (let i = 0; i < predefinedMetrics.length; i++) {
             for (let j = 0; j < predefinedMetrics[i].metrics.length; j++) {
                 let tmpMetric = predefinedMetrics[i].metrics[j],
@@ -185,13 +200,7 @@ usage.processSessionMetricsFromStream = function(currEvent, uniqueLevelsZero, un
                 }
             }
         }
-        function fetchMeta(id, callback) {
-            common.readBatcher.getOne(metaToFetch[id].coll, {'_id': metaToFetch[id].id}, {meta_v2: 1}, (err, metaDoc) => {
-                var retObj = metaDoc || {};
-                retObj.coll = metaToFetch[id].coll;
-                callback(null, retObj);
-            });
-        }
+
         var metas = {};
         async.map(Object.keys(metaToFetch), fetchMeta, function(err, metaDocs) {
             for (let i = 0; i < metaDocs.length; i++) {
@@ -210,8 +219,9 @@ usage.processSessionMetricsFromStream = function(currEvent, uniqueLevelsZero, un
                         monthObjUpdate = [],
                         tmpMetric = predefinedMetrics[i].metrics[j],
                         recvMetricValue = "",
-                        escapedMetricVal = "",
-                        postfix = "";
+                        escapedMetricVal = "";
+
+                    postfix = "";
 
                     recvMetricValue = currEvent.up[tmpMetric.short_code];
 

api/api.js

@@ -14,7 +14,7 @@ const pack = require('../package.json');
 const versionInfo = require('../frontend/express/version.info.js');
 const moment = require("moment");
 
-var {MongoDbQueryRunner} = require('../plugins/drill/api/parts/data/MongoDbQueryRunner.js');
+var {MongoDbQueryRunner} = require('./utils/mongoDbQueryRunner.js');
 
 var t = ["countly:", "api"];
 common.processRequest = processRequest;

api/ingestor/requestProcessor.js

@@ -401,12 +401,18 @@ var processToDrill = async function(params, drill_updates, callback) {
                 "ts": events[i].timestamp || Date.now().valueOf(),
                 "uid": params.app_user.uid,
                 "_uid": params.app_user._id,
-                "did": params.app_user.did,
-                "ce": true,
+                "did": params.app_user.did
                 //d, w,m,h
             };
             if (currEvent.key.indexOf('[CLY]_') === 0) {
-                dbEventObject.ce = false;
+                dbEventObject.n = events[i].key;
+            }
+            else {
+                dbEventObject.n = events[i].key;
+                dbEventObject.e = "[CLY]_custom";
+            }
+            if (currEvent.name) {
+                dbEventObject.n = currEvent.name;
             }
 
             if (dbAppUser && dbAppUser[common.dbUserMap.user_id]) {
@@ -557,7 +563,7 @@ var processToDrill = async function(params, drill_updates, callback) {
             eventsToInsert.push({"insertOne": {"document": dbEventObject}});
             if (eventKey === "[CLY]_view") {
                 var view_id = crypto.createHash('md5').update(currEvent.segmentation.name).digest('hex');
-                viewUpdate[view_id] = {"lvid": dbEventObject._id, "ts": dbEventObject.ts};
+                viewUpdate[view_id] = {"lvid": dbEventObject._id, "ts": dbEventObject.ts, "a": params.app_id + ""};
                 if (currEvent.segmentation) {
                     var sgm = {};
                     var have_sgm = false;
@@ -580,25 +586,26 @@ var processToDrill = async function(params, drill_updates, callback) {
         for (var z4 = 0; z4 < drill_updates.length;z4++) {
             eventsToInsert.push(drill_updates[z4]);
         }
-
     }
     if (eventsToInsert.length > 0) {
         try {
             await common.drillDb.collection("drill_events").bulkWrite(eventsToInsert, {ordered: false});
+            callback(null);
             if (Object.keys(viewUpdate).length) {
                 //updates app_viewdata colelction.If delayed new incoming view updates will not have reference. (So can do in aggregator only if we can insure minimal delay)
                 try {
-                    await common.db.collection("app_userviews" + params.app_id).updateOne({_id: params.app_user.uid}, {$set: viewUpdate}, {upsert: true});
+                    await common.db.collection("app_userviews").updateOne({_id: params.app_id + "_" + params.app_user.uid}, {$set: viewUpdate}, {upsert: true});
                 }
                 catch (err) {
                     log.e(err);
                 }
             }
-            callback(null);
+
         }
         catch (errors) {
             var realError;
             if (errors && Array.isArray(errors)) {
+                log.e(JSON.stringify(errors));
                 for (let i = 0; i < errors.length; i++) {
                     if ([11000, 10334, 17419].indexOf(errors[i].code) === -1) {
                         realError = true;
@@ -609,16 +616,17 @@ var processToDrill = async function(params, drill_updates, callback) {
                     callback(realError);
                 }
                 else {
+                    callback(null);
                     if (Object.keys(viewUpdate).length) {
                         //updates app_viewdata colelction.If delayed new incoming view updates will not have reference. (So can do in aggregator only if we can insure minimal delay)
                         try {
-                            await common.db.collection("app_userviews" + params.app_id).updateOne({_id: params.app_user.uid}, {$set: viewUpdate}, {upsert: true});
+                            await common.db.collection("app_userviews").updateOne({_id: params.app_id + "_" + params.app_user.uid}, {$set: viewUpdate}, {upsert: true});
                         }
                         catch (err) {
                             log.e(err);
                         }
                     }
-                    callback(null);
+
                 }
             }
             else {
@@ -701,7 +709,14 @@ const processRequestData = (ob, done) => {
             update = common.mergeQuery(update, ob.updates[i]);
         }
     }
-    Promise.all([ common.updateAppUser(ob.params, update, false, true)]).then(function() {
+    //var SaveAppUser = Date.now().valueOf();
+
+    common.updateAppUser(ob.params, update, false, function() {
+
+        /*var AfterSaveAppUser = Date.now().valueOf();
+        if (AfterSaveAppUser - SaveAppUser > treshold) {
+            console.log("SaveAppUser time: " + (AfterSaveAppUser - SaveAppUser));
+        }*/
         processToDrill(ob.params, ob.drill_updates, function(error) {
             if (error) {
                 common.returnMessage(ob.params, 400, 'Could not record events:' + error);
@@ -712,10 +727,6 @@ const processRequestData = (ob, done) => {
             }
         });
 
-    }).catch(function(err) {
-        log.e(err);
-        common.returnMessage(ob.params, 400, 'Cannot process request');
-        done();
     });
 };
 
@@ -730,7 +741,6 @@ plugins.register("/sdk/process_request", async function(ob) {
  * 
  * @param {*} params  - request parameters
  * @param {*} done  - callback function
- * @returns {boolean} - returns false if request is cancelled
  * 
  * 
  * 1)Get App collection settings

api/parts/data/batcher.js

@@ -197,6 +197,11 @@ class WriteBatcher {
         });
     }
 
+    /**
+     *  Function to call once flushed
+     * @param {string} name - name of collection 
+     * @param {function} callback  - callback function
+     */
     addFlushCallback(name, callback) {
         this.flushCallbacks[name] = callback;
     }
@@ -226,7 +231,7 @@ class WriteBatcher {
      *  @param {string} db - name of the database for which to write data
      *  @param {string} collection - name of the collection for which to write data
      */
-    async flush(db, collection, callback) {
+    async flush(db, collection) {
         var no_fallback_errors = [10334, 17419, 14, 56];
         var notify_errors = [10334, 17419];
         if (this.data[db] && this.data[db][collection] && this.data[db][collection].data && Object.keys(this.data[db][collection].data).length) {
@@ -247,7 +252,6 @@ class WriteBatcher {
                     });
                 }
             }
-            var token0 = this.data[db][collection].t;
             this.data[db][collection] = {"data": {}};
             batcherStats.update_queued -= queries.length;
             batcherStats.update_processing += queries.length;
@@ -348,6 +352,7 @@ class WriteBatcher {
      *  @param {string} id - id of the document
      *  @param {object} operation - operation
      *  @param {string} db - name of the database for which to write data
+     * @param {object} options - options for the operation
      */
     add(collection, id, operation, db = "countly", options) {
         options = options || {};
@@ -451,9 +456,9 @@ class ReadBatcher {
         }
         else {
             try {
-                var res = await this.db.collection(collection).findOne(query, projection);
-                this.cache(collection, id, query, projection, res, false);
-                return res;
+                var res2 = await this.db.collection(collection).findOne(query, projection);
+                this.cache(collection, id, query, projection, res2, false);
+                return res2;
             }
             catch (err) {
                 if (this.data && this.data[collection] && this.data[collection][id] && this.data[collection][id].promise) {

api/parts/data/cacher.js

@@ -8,6 +8,7 @@ class Cacher {
 /**
      *  Create batcher instance
      *  @param {Db} db - database object
+     *  @param {object} options - options object
      */
     constructor(db, options) {
         this.db = db;

api/parts/data/changeStreamReader.js

@@ -2,9 +2,11 @@ const common = require("../../utils/common");
 const log = require('../../utils/log.js')("changeStreamReader");
 var Timestamp = require('mongodb').Timestamp;
 
-
+/**
+ * Class to ruse change streams to read from mongodb.
+ */
 class changeStreamReader {
-    /*
+    /** 
     * @param {Object} db - Database object
     * @param {Object} options - Options object
     * @param {function} onData - Finction to call when getting new data from stream
@@ -37,7 +39,7 @@ class changeStreamReader {
 
     }
 
-    /*
+    /** 
     * Check if stream is closed and restart if needed
     */
     checkState() {
@@ -52,16 +54,19 @@ class changeStreamReader {
         }
     }
 
-    /*
+    /** 
     * Process bad range((when token can't continue))
+    * @param {Object} options - Options object
+    * @param {Object} tokenInfo - Token info object
     */
     async processBadRange(options, tokenInfo) {
         console.log("Processing bad range");
         console.log(JSON.stringify({cd: {$gte: options.cd1, $lt: options.cd2}}));
         var gotTokenDoc = false;
+        var doc;
         var cursor = this.db.collection(this.collection).find({cd: {$gte: new Date(options.cd1), $lt: new Date(options.cd2)}}).sort({cd: 1});
         while (await cursor.hasNext() && !gotTokenDoc) {
-            var doc = await cursor.next();
+            doc = await cursor.next();
             if (JSON.stringify(doc._id) === JSON.stringify(tokenInfo._id) || doc.cd > tokenInfo.cd) {
                 gotTokenDoc = true;
             }
@@ -75,7 +80,7 @@ class changeStreamReader {
         }
 
         while (await cursor.hasNext()) {
-            var doc = await cursor.next();
+            doc = await cursor.next();
             console.log("Process:" + JSON.stringify(doc));
             tokenInfo.cd = doc.cd;
             tokenInfo._id = doc._id;
@@ -84,7 +89,10 @@ class changeStreamReader {
         console.log("done");
     }
 
-    /*Opening stream*/
+    /**
+     *  Sets up stream to read data from mongodb
+     *  @param {function} onData  - function to call on new data
+     */
     async setUp(onData) {
         var token;
         try {
@@ -149,14 +157,14 @@ class changeStreamReader {
             }
             else {
                 this.stream.on('change', (change) => {
-                    var token = {token: self.stream.resumeToken};
-                    token._id = change.__id;
+                    var my_token = {token: self.stream.resumeToken};
+                    my_token._id = change.__id;
                     if (change.cd) {
-                        token.cd = change.cd;
-                        onData(token, change);
+                        my_token.cd = change.cd;
+                        onData(my_token, change);
                     }
                     else {
-                        onData(token, change);
+                        onData(my_token, change);
                     }
                 });
 
@@ -184,12 +192,21 @@ class changeStreamReader {
                 console.log("Set Failed token", token);
                 this.failedToken = token;
             }
+            //Failed because of db does not support change streams. Run in "query mode";
+            else if (err.code === "look_for_right_code") {
+                //Call process bad range if there is any info about last token.
+                //Switch to query mode
+            }
             else {
                 log.e("Error on change stream", err);
             }
         }
     }
 
+    /**
+     * Acknowledges token as recorded
+     * @param {object} token  - token info
+     */
     async acknowledgeToken(token) {
         this.lastToken = token;
         //Update last processed token to database
@@ -207,7 +224,9 @@ class changeStreamReader {
         }
     }
 
-    //Close stream by command
+    /**
+     * Closes stream permanently
+     */
     close() {
         console.log("Closing permanently");
         if (this.intervalRunner) {

api/parts/data/events.js

@@ -13,25 +13,25 @@ var countlyEvents = {},
 
 
 
-countlyEvents.processEventFromStream = function(event) {
-    var forbiddenSegValues = [];
+countlyEvents.processEventFromStream = function() {
+    /*var forbiddenSegValues = [];
     for (let i = 1; i < 32; i++) {
         forbiddenSegValues.push(i + "");
     }
 
     //Write event totals for aggregated Data
 
     common.readBatcher.getOne("apps", {'_id': event.a}, function(err, app) {
-        /* common.readBatcher.getOne("events", {'_id': event.a}, {
+         common.readBatcher.getOne("events", {'_id': event.a}, {
             list: 1,
             segments: 1,
             omitted_segments: 1,
             whitelisted_segments: 1
         }, (err, eventColl) => {
 
-        });*/
+        });
 
-    });
+    });*/
 };
 /**
 * Process JSON decoded events data from request