Merge branch 'master' into next

# Conflicts:
#	frontend/express/public/javascripts/countly/countly.template.js

Author: ar2rsawseen

.travis.yml

@@ -20,7 +20,9 @@ before_script:
 - nodejs --version
 - ${PWD}/mongodb-linux-x86_64-${MONGODB}/bin/mongod --version
 - cat ${PWD}/mongod.log
-script: sudo npm test
+script:
+- countly shellcheck
+- sudo npm test
 after_script:
 - cat log/countly-api.log
 - cat log/countly-dashboard.log

README.md

@@ -3,7 +3,7 @@
 
 <p align="right">
 
-[![Build Status](https://api.travis-ci.org/Countly/countly-server.png?branch=master)](https://travis-ci.org/Countly/countly-server) [![Codacy Badge](https://api.codacy.com/project/badge/Grade/98c2726f2d734697a5f1ac0d453f0a06)](https://app.codacy.com/app/ar2rsawseen/countly-server?utm_source=github.com&utm_medium=referral&utm_content=Countly/countly-server&utm_campaign=Badge_Grade_Dashboard) [![Install Countly on DigitalOcean](https://count.ly/github/install-on-digital-ocean.svg)](http://do.count.ly)
+[![Build Status](https://api.travis-ci.org/Countly/countly-server.png?branch=master)](https://travis-ci.org/Countly/countly-server) [![Install Countly on DigitalOcean](https://count.ly/github/install-on-digital-ocean.svg)](http://do.count.ly)
 
 </p>
 

api/parts/data/cache.js

@@ -722,6 +722,11 @@ class CacheMaster {
 function createCollection(db, name, size = 1e7) {
     return new Promise((resolve, reject) => {
         db.onOpened(() => {
+            if (!db._native) {
+                return setTimeout(() => {
+                    createCollection(db, name, size).then(resolve, reject);
+                }, 1000);
+            }
             db._native.createCollection(name, {capped: true, size: size}, (e, col) => {
                 if (e) {
                     log.e(`Error while creating capped collection ${name}:`, e);

api/parts/data/exports.js

@@ -95,7 +95,7 @@ exports.convertData = function(data, type) {
         return JSON.stringify(data);
     case "csv":
         obj = flattenArray(data);
-        return json2csv.parse(obj.data, {fields: obj.fields});
+        return json2csv.parse(obj.data, {fields: obj.fields, excelStrings: true});
     case "xls":
         obj = flattenArray(data);
         return json2xls(obj.data, {fields: obj.fields});

api/parts/jobs/job.js

@@ -308,7 +308,7 @@ class Job extends EventEmitter {
                     reject(err);
                 }
                 else if (!doc || !doc.ok || !doc.value) {
-                    reject('Not found');
+                    reject('Job not found');
                 }
                 else {
                     resolve(doc.value);
@@ -438,7 +438,7 @@ class Job extends EventEmitter {
             log.i('Replacing job %s (%j) with date %d: %j', this.name, this.data, next, query);
 
             Job.updateAtomically(this.db(), query, {$set: {next: next}}).then(resolve, err => {
-                if (err === 'Not found') {
+                if (err === 'Job not found') {
                     log.i('Replacing job %s (%j) with date %d: no future jobs to move', this.name, this.data, next);
                     query.next = {$lt: next};
                     Job.findMany(this.db(), query).then(existing => {
@@ -511,7 +511,14 @@ class Job extends EventEmitter {
                 await Job.updateMany(this.db(), {_id: {$in: others.map(o => o._id)}}, {$set: {status: STATUS.CANCELLED}});
             }
 
-            let neo = await Job.updateAtomically(this.db(), query, {$set: this._json});
+            let neo;
+            try {
+                neo = await Job.updateAtomically(this.db(), query, {$set: this._json});
+            }
+            catch (e) {
+                log.w('Job was modified while rescheduling, skipping', e);
+            }
+
             if (!neo) {
                 log.w('Job was modified while rescheduling, skipping');
                 return null;

api/parts/jobs/manager.js

@@ -150,7 +150,10 @@ class Manager {
                                             name: j.name
                                         };
                                     }));
-                                    this.process(array.filter(j => this.types.indexOf(j.name) !== -1));
+                                    this.process(array.filter(j => this.types.indexOf(j.name) !== -1)).catch(e => {
+                                        log.e('Error during job resuming', e);
+                                        resume();
+                                    });
                                 }
                                 else {
                                     this.checkAfterDelay(DELAY_BETWEEN_CHECKS * 5);
@@ -224,7 +227,10 @@ class Manager {
                 this.checkAfterDelay();
             }
             else {
-                this.process(jobs);
+                this.process(jobs).catch(e => {
+                    log.e('Error during job processing', e);
+                    this.checkAfterDelay();
+                });
             }
         });
     }
@@ -256,12 +262,13 @@ class Manager {
 
                 if (!this.canRun(job)) {
                     jobs = jobs.filter(j => j.name !== job.name);
+                    log.d('Cannot run %s:%s:%s, skipping for now', json.name, json._id, new Date(json.next));
                     continue;
                 }
 
                 if (job instanceof JOB.IPCJob) {
                     if (!this.hasResources(job)) {
-                        log.i('All resources are busy for %j, skipping for now', json);
+                        log.i('All resources are busy for %s:%s:%s, skipping for now', json.name, json._id, new Date(json.next));
                         continue;
                     }
                 }
@@ -287,15 +294,33 @@ class Manager {
                     continue;
                 }
 
-                let old = await JOB.Job.updateAtomically(this.db, {
-                    _id: job._id,
-                    status: {$in: [STATUS.RUNNING, STATUS.SCHEDULED, STATUS.PAUSED]}
-                }, update, false);
+                let old;
+                try {
+                    old = await JOB.Job.updateAtomically(this.db, {
+                        _id: job._id,
+                        status: {$in: [STATUS.RUNNING, STATUS.SCHEDULED, STATUS.PAUSED]}
+                    }, update, false);
+                }
+                catch (e) {
+                    log.i('Job %s wasn\'t found', job._id, e);
+                }
+
                 if (old) {
                     if (old.status === STATUS.RUNNING) {
                         log.i('Job %s is running on another server, won\'t start it here', job.id);
                     }
                     else if (old.status === STATUS.SCHEDULED || old.status === STATUS.PAUSED) {
+                        if (job instanceof JOB.IPCJob) {
+                            if (!this.hasResources(job)) {
+                                log.i('Started the job, but all resources are busy for %j, putting it back to SCHEDULED', json);
+                                await JOB.Job.updateAtomically(this.db, {
+                                    _id: job._id,
+                                    status: STATUS.RUNNING
+                                }, {$set: {status: STATUS.SCHEDULED}}, false);
+                                return;
+                            }
+                        }
+
                         let p = this.start(job);
                         if (!p) {
                             await JOB.Job.updateAtomically(this.db, {

api/parts/jobs/resource.js

@@ -317,7 +317,7 @@ class ResourceFaçade extends ResourceInterface {
     * Run job
     * @param {Job} job to run
     * @returns {Promise} promise
-    **/ 
+    **/
     run(job) {
         if (this.isBusy) {
             log.w('[façade]: Resource façade %j is busy in %d: %j', this.name, this._worker.pid, this.id);
@@ -340,7 +340,7 @@ class ResourceFaçade extends ResourceInterface {
     /**
     * Close resource
     * @returns {Promise} promise
-    **/ 
+    **/
     close() {
         if (this.isOpen) {
             log.w('Closing underlying resource %s from façade', this.id);
@@ -362,7 +362,7 @@ class ResourceFaçade extends ResourceInterface {
     /**
     * Kill resource
     * @returns {Promise} promise
-    **/ 
+    **/
     kill() {
         return new Promise((resolve) => {
             this._worker.kill();
@@ -375,7 +375,7 @@ class ResourceFaçade extends ResourceInterface {
     /**
     * Open resource
     * @returns {Promise} promise
-    **/ 
+    **/
     open() {
         if (this.isOpen) {
             return Promise.resolve();
@@ -404,7 +404,7 @@ class ResourceFaçade extends ResourceInterface {
     * Abort job
     * @param {Job} job to abort
     * @returns {Promise} promise
-    **/ 
+    **/
     abort(job) {
         if (!this.job) {
             log.w('[façade]: Resource façade %j is not open in %d: %j', this.name, this._worker.pid, this.id);
@@ -421,7 +421,7 @@ class ResourceFaçade extends ResourceInterface {
 
     /**
     * Resolve job
-    **/ 
+    **/
     resolve() {
         if (this._resolve) {
             log.w('[façade]: Resolving %s', this.job.channel);
@@ -443,7 +443,7 @@ class ResourceFaçade extends ResourceInterface {
     /**
     * Reject job
     * @param {Error} error with which to reject
-    **/ 
+    **/
     reject(error) {
         if (this._reject) {
             log.w('[façade]: Rejecting %s', this.job.channel);

api/tcp_example.js

@@ -75,7 +75,7 @@ net.createServer(function(socket) {
             * Accepting req data in format {"url":"endpoint", "body":"data"}
             * Example: {"url":"/o/ping"}
             * Example: {"url":"/i", "body":{"device_id":"test","app_key":"APP_KEY","begin_session":1,"metrics":{}}}
-            **/            
+            **/
             //creating request context
             var params = {
                 //providing data in request object

api/utils/authorizer.js

@@ -102,7 +102,7 @@ authorizer.read = function(options) {
 * @param {object} options.db - database connection
 * @param {string} options.token - token to rvalidate
 * @param {function} options.callback - function called when reading was completed or errored, providing error object as first param, true or false if expired as second, seconds till expiration as third.(-1 if never expires, 0 - if expired) 
-*/ 
+*/
 authorizer.check_if_expired = function(options) {
     options.db = options.db || common.db;
     options.token = options.token + "";

api/utils/common.js

@@ -2200,7 +2200,7 @@ common.checkDatabaseConfigMatch = (apiConfig, frontendConfig) => {
              *      ]
              *  }
              * }
-             */ 
+             */
             else if (Object.prototype.hasOwnProperty.call(apiConfig, 'replSetServers') && Object.prototype.hasOwnProperty.call(frontendConfig, 'replSetServers')) {
                 if (apiConfig.replSetServers.length === frontendConfig.replSetServers.length && apiConfig.db === frontendConfig.db) {
                     let isCorrect = true;

bin/backup/run.sh

@@ -14,10 +14,11 @@
 
 
 #  make relative paths work.
-cd $(dirname $0)
+cd "$(dirname "$0")"
 
 function check_connectivity_mongo() {
-	local MONGO_OK=$(mongo \
+    local MONGO_OK;
+	MONGO_OK=$(mongo \
 		--quiet \
 		--eval "db.serverStatus().ok == true")
 
@@ -32,16 +33,16 @@ function check_connectivity_mongo() {
 # generic check connectivity function
 function check_connectivity() {
 	retries=600
-	until eval "check_connectivity_$@"; do
+	until eval "check_connectivity_$*"; do
 		sleep 1
 		let retries--
 		if [ $retries == 0 ]; then
-			echo "time out while waiting for $@ is ready"
+			echo "time out while waiting for $* is ready"
 			exit 1
 		fi
-		echo "$@ is not reachable yet, trying again..."
+		echo "$* is not reachable yet, trying again..."
 	done
-	echo "$@ is up and running..."
+	echo "$* is up and running..."
 }
 
 # wait till mongo becomes online.