Issue 515

.travis.yml

@@ -8,54 +8,105 @@ services:
 branches:
   only:
   - master
-  - demo
+  - develop
+  - staging
+  - QA
 
 jobs:
   include:
-    - stage: Seneca-Jobqueue-Flowz
+#    - stage: Seneca-Jobqueue-Flowz
+#      script:
+#      - 'if [ ${TRAVIS_BRANCH} = "master" ]; then
+#            docker login -u="$DOCKER_USERNAME_FLOWZ" -p="$DOCKER_PASSWORD_FLOWZ";
+#            docker build -t seneca_jobqueue_flowz:latest ./JobQue/;
+#            docker images;
+#            docker tag seneca_jobqueue_flowz:latest $DOCKER_USERNAME_FLOWZ/seneca_jobqueue_flowz:latest;
+#            docker push $DOCKER_USERNAME_FLOWZ/seneca_jobqueue_flowz:latest;
+#        elif [ ${TRAVIS_BRANCH} = "develop" ]; then
+#            docker login -u="$DOCKER_USERNAME" -p="$DOCKER_PASSWORD";
+#            docker build -t seneca_jobqueue_flowz:dev ./JobQue/;
+#            docker images;
+#            docker tag seneca_jobqueue_flowz:dev $DOCKER_USERNAME/seneca_jobqueue_flowz:dev;
+#            docker push $DOCKER_USERNAME/seneca_jobqueue_flowz:dev;
+#        elif [ ${TRAVIS_BRANCH} = "staging" ]; then
+#            docker login -u="$DOCKER_USERNAME" -p="$DOCKER_PASSWORD";
+#            docker build -t seneca_jobqueue_flowz:$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER ./JobQue/;
+#            docker tag seneca_jobqueue_flowz:$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER $DOCKER_USERNAME/seneca_jobqueue_flowz:$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER;
+#            docker tag seneca_jobqueue_flowz:$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER $DOCKER_USERNAME/seneca_jobqueue_flowz:staging;
+#            docker images;
+#            docker push $DOCKER_USERNAME/seneca_jobqueue_flowz:$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER;
+#            docker push $DOCKER_USERNAME/seneca_jobqueue_flowz:staging;
+#        else
+#            docker login -u="$DOCKER_USERNAME" -p="$DOCKER_PASSWORD";
+#            docker build -t seneca_jobqueue_flowz:qa ./JobQue/;
+#            docker images;
+#            docker tag seneca_jobqueue_flowz:qa $DOCKER_USERNAME/seneca_jobqueue_flowz:qa;
+#            docker push $DOCKER_USERNAME/seneca_jobqueue_flowz:qa;
+#        fi'
+    - stage: Flowz-Frontend-Flowz
       script:
-      - 'if [ ${TRAVIS_BRANCH} = "demo" ]; then
-            echo "not build seneca_jobqueue_flowz image";
+      - 'if [ ${TRAVIS_BRANCH} = "master" ]; then
+            docker login -u="$DOCKER_USERNAME_FLOWZ" -p="$DOCKER_PASSWORD_FLOWZ";
+            docker build -t flowz_frontend_flowz:latest --build-arg accesskey="$AWS_ACCESSKEY" --build-arg secretkey="$AWS_SECRETKEY" --build-arg domainKey="$DOMAINKEY_MASTER" ./client/;
+            docker images;
+            docker tag flowz_frontend_flowz:latest $DOCKER_USERNAME_FLOWZ/flowz_frontend_flowz:latest;
+            docker push $DOCKER_USERNAME_FLOWZ/flowz_frontend_flowz:latest;
+        elif [ ${TRAVIS_BRANCH} = "develop" ]; then
+            docker login -u="$DOCKER_USERNAME" -p="$DOCKER_PASSWORD";
+            docker build -t flowz_frontend_flowz:dev --build-arg accesskey="$AWS_ACCESSKEY" --build-arg secretkey="$AWS_SECRETKEY" --build-arg domainKey="$DOMAINKEY_DEVELOP" ./client/;
+            docker images;
+            docker tag flowz_frontend_flowz:dev $DOCKER_USERNAME/flowz_frontend_flowz:dev;
+            docker push $DOCKER_USERNAME/flowz_frontend_flowz:dev;
+        elif [ ${TRAVIS_BRANCH} = "staging" ]; then
+            docker login -u="$DOCKER_USERNAME" -p="$DOCKER_PASSWORD";
+            docker build -t flowz_frontend_flowz:$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER --build-arg accesskey="$AWS_ACCESSKEY" --build-arg secretkey="$AWS_SECRETKEY" --build-arg domainKey="$DOMAINKEY_STAGING" ./client/;
+            docker tag flowz_frontend_flowz:$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER $DOCKER_USERNAME/flowz_frontend_flowz:$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER;
+            docker tag flowz_frontend_flowz:$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER $DOCKER_USERNAME/flowz_frontend_flowz:staging;
+            docker images;
+            docker push $DOCKER_USERNAME/flowz_frontend_flowz:$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER;
+            docker push $DOCKER_USERNAME/flowz_frontend_flowz:staging;
         else
             docker login -u="$DOCKER_USERNAME" -p="$DOCKER_PASSWORD";
-            docker build -t seneca_jobqueue_flowz:$TRAVIS_BRANCH ./JobQue/;
+            docker build -t flowz_frontend_flowz:qa --build-arg accesskey="$AWS_ACCESSKEY" --build-arg secretkey="$AWS_SECRETKEY" --build-arg domainKey="$DOMAINKEY_QA" ./client/;
             docker images;
-            docker tag seneca_jobqueue_flowz:$TRAVIS_BRANCH $DOCKER_USERNAME/seneca_jobqueue_flowz:$TRAVIS_BRANCH;
-            docker push $DOCKER_USERNAME/seneca_jobqueue_flowz:$TRAVIS_BRANCH;
+            docker tag flowz_frontend_flowz:qa $DOCKER_USERNAME/flowz_frontend_flowz:qa;
+            docker push $DOCKER_USERNAME/flowz_frontend_flowz:qa;
         fi'
-    - stage: Flowz-Frontend-Flowz
-      script:
-      - docker login -u="$DOCKER_USERNAME" -p="$DOCKER_PASSWORD"
-      - docker build -t flowz_frontend_flowz:$TRAVIS_BRANCH --build-arg accesskey="$AWS_ACCESSKEY" --build-arg secretkey="$AWS_SECRETKEY" ./client/
-      - docker images
-      - docker tag flowz_frontend_flowz:$TRAVIS_BRANCH $DOCKER_USERNAME/flowz_frontend_flowz:$TRAVIS_BRANCH
-      - docker push $DOCKER_USERNAME/flowz_frontend_flowz:$TRAVIS_BRANCH
     - stage: Flowz-Backend-Flowz
       script:
-      - 'if [ ${TRAVIS_BRANCH} = "demo" ]; then
-            echo "not build flowz_backend_flowz image";
+      - 'if [ ${TRAVIS_BRANCH} = "master" ]; then
+            docker login -u="$DOCKER_USERNAME_FLOWZ" -p="$DOCKER_PASSWORD_FLOWZ";
+            docker build -t flowz_backend_flowz:latest ./service/;
+            docker images;
+            docker tag flowz_backend_flowz:latest $DOCKER_USERNAME_FLOWZ/flowz_backend_flowz:latest;
+            docker push $DOCKER_USERNAME_FLOWZ/flowz_backend_flowz:latest;
+        elif [ ${TRAVIS_BRANCH} = "develop" ]; then
+            docker login -u="$DOCKER_USERNAME" -p="$DOCKER_PASSWORD";
+            docker build -t flowz_backend_flowz:dev ./service/;
+            docker images;
+            docker tag flowz_backend_flowz:dev $DOCKER_USERNAME/flowz_backend_flowz:dev;
+            docker push $DOCKER_USERNAME/flowz_backend_flowz:dev;
+        elif [ ${TRAVIS_BRANCH} = "staging" ]; then
+            docker login -u="$DOCKER_USERNAME" -p="$DOCKER_PASSWORD";
+            docker build -t flowz_backend_flowz:$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER ./service/;
+            docker tag flowz_backend_flowz:$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER $DOCKER_USERNAME/flowz_backend_flowz:$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER;
+            docker tag flowz_backend_flowz:$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER $DOCKER_USERNAME/flowz_backend_flowz:staging;
+            docker images;
+            docker push $DOCKER_USERNAME/flowz_backend_flowz:$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER;
+            docker push $DOCKER_USERNAME/flowz_backend_flowz:staging;
         else
             docker login -u="$DOCKER_USERNAME" -p="$DOCKER_PASSWORD";
-            docker build -t flowz_backend_flowz:$TRAVIS_BRANCH ./service/;
+            docker build -t flowz_backend_flowz:qa ./service/;
             docker images;
-            docker tag flowz_backend_flowz:$TRAVIS_BRANCH $DOCKER_USERNAME/flowz_backend_flowz:$TRAVIS_BRANCH;
-            docker push $DOCKER_USERNAME/flowz_backend_flowz:$TRAVIS_BRANCH;
+            docker tag flowz_backend_flowz:qa $DOCKER_USERNAME/flowz_backend_flowz:qa;
+            docker push $DOCKER_USERNAME/flowz_backend_flowz:qa;
         fi'
     - stage: Upgrade Rancher Service
       script:
-      - 'if [ ${TRAVIS_BRANCH} = "demo" ]; then
-            bash ./upgrade_demo.sh;
-         else
-            bash ./upgrade.sh;
-         fi'
-
+      - bash ./upgrade.sh
     - stage: Finish Rancher Service
       script:
-      - 'if [ ${TRAVIS_BRANCH} = "demo" ]; then
-            bash ./finish_demo.sh;
-         else
-            bash ./finish.sh;
-         fi'
+      - bash ./finish.sh
 
 notifications:
   email:

JobQue/.gitignore

@@ -10,6 +10,7 @@ selenium-debug.log
 package-lock.json
 worker/
 example/worker/
+seneca-job-queue/logs
 
 # Editor directories and files
 .idea

JobQue/docs/README.md

@@ -0,0 +1 @@
+flowcharts related to scheduler and process flow of jobqueue can be found in this folder

JobQue/scheduler/default.json

@@ -15,7 +15,7 @@
   "scheduler_table": "scheduler",
   "system_logs_table": "system_logs",
   "qJobTimeout": 3600000,
-  "qJobRetryMax": 0,
+  "qJobRetryMax": 5,
   "chokidar": {
     "persistent": true,
     "ignoreInitial": false,

JobQue/scheduler/function.js

@@ -37,7 +37,7 @@ module.exports = function (options, PINO_DB_OPTION, PINO_C_OPTION) {
         if (numberOfExternalSchema != 0 && !externalCheck) {
 
           let sourceCounts = targetJob.data.sourceCount ? Object.values(targetJob.data.sourceCount) : []
-          
+
           if (sourceCounts.length < numberOfExternalSchema || (capacity && Math.min(...sourceCounts) != capacity)) {
             await this.updateProcess(targetJob, 'created')
             await this.updateLog(targetJob, 'created', false)
@@ -568,6 +568,7 @@ module.exports = function (options, PINO_DB_OPTION, PINO_C_OPTION) {
   this.constructor.prototype.newInstance = async function (flowInstance, fId, next) {
     return new Promise (async (resolve, reject) => {
       try{
+
         for (let i=0; i<flowInstance.start_states.length; i++) {
 
           let startStateId = flowInstance.start_states[i]
@@ -695,16 +696,16 @@ module.exports = function (options, PINO_DB_OPTION, PINO_C_OPTION) {
               if (inputAvailability) {
                 //if all inputs are available, update the latest mapping done and begin the process
                 //i.e. change status of the job for that process in it's respective worker queue to `waiting`
-                pino(PINO_DB_OPTION,fs.createWriteStream('./logs')).info({'fId': jobData.fId, 'job': targetSchema.id, 'jobId': targetJobs[0].id}, 'next job')
-                pino(PINO_C_OPTION).info({'fId': jobData.fId, 'job': targetSchema.id, 'jobId': targetJobs[0].id}, 'next job')
+                //pino(PINO_DB_OPTION,fs.createWriteStream('./logs')).info({'fId': jobData.fId, 'job': targetSchema.id, 'jobId': targetJobs[0].id}, 'next job')
+                //pino(PINO_C_OPTION).info({'fId': jobData.fId, 'job': targetSchema.id, 'jobId': targetJobs[0].id}, 'next job')
                 // await this.beginProcess(targetJobs[i])
                 await this.beginProcess(targetJobs[0])
               }
               else {
                 //i.e. all inputs are still not available so just update the latest mapping done
                 //(it's status in it's respective worker queue will not be changed)
-                pino(PINO_DB_OPTION,fs.createWriteStream('./logs')).warn({'fId': jobData.fId, 'job': targetSchema.id, 'jobId': targetJobs[0].id}, 'all inputs not available')
-                pino(PINO_C_OPTION).warn({'fId': jobData.fId, 'job': targetSchema.id, 'jobId': targetJobs[0].id}, 'all inputs not available')
+                //pino(PINO_DB_OPTION,fs.createWriteStream('./logs')).warn({'fId': jobData.fId, 'job': targetSchema.id, 'jobId': targetJobs[0].id}, 'all inputs not available')
+                //pino(PINO_C_OPTION).warn({'fId': jobData.fId, 'job': targetSchema.id, 'jobId': targetJobs[0].id}, 'all inputs not available')
                 // let tmp = await this.updateProcess(targetJobs[i])
               }
             // }
@@ -796,21 +797,21 @@ module.exports = function (options, PINO_DB_OPTION, PINO_C_OPTION) {
             if (inputAvailability) {
               //if all inputs are available, update the latest mapping done and begin the process
               //i.e. change status of the job for that process in it's respective worker queue to `waiting`
-              pino(PINO_DB_OPTION,fs.createWriteStream('./logs')).info({'fId': fId, 'jobId': targetId}, 'next job')
-              pino(PINO_C_OPTION).info({'fId': fId, 'jobId': targetId}, 'next job')
+              //pino(PINO_DB_OPTION,fs.createWriteStream('./logs')).info({'fId': fId, 'jobId': targetId}, 'next job')
+              //pino(PINO_C_OPTION).info({'fId': fId, 'jobId': targetId}, 'next job')
               await this.beginProcess(targetJobs[i])
             }
             else {
               //i.e. all inputs are still not available so just update the latest mapping done
               //(it's status in it's respective worker queue will not be changed)
               var mappingRequired = capacity ? false : this.externalMappingRequired(targetJobs[i])
               if (mappingRequired) {
-                pino(PINO_DB_OPTION,fs.createWriteStream('./logs')).warn({'fId': fId, 'jobId': targetId}, 'mapping required')
-                pino(PINO_C_OPTION).warn({'fId': fId, 'jobId': targetId}, 'mapping required')
+                //pino(PINO_DB_OPTION,fs.createWriteStream('./logs')).warn({'fId': fId, 'jobId': targetId}, 'mapping required')
+                //pino(PINO_C_OPTION).warn({'fId': fId, 'jobId': targetId}, 'mapping required')
               }
               else {
-                pino(PINO_DB_OPTION,fs.createWriteStream('./logs')).warn({'fId': fId, 'job': targetId, 'jobId': targetJobs[i].id}, 'all inputs not available')
-                pino(PINO_C_OPTION).warn({'fId': fId, 'job': targetId, 'jobId': targetJobs[i].id}, 'all inputs not available')
+                //pino(PINO_DB_OPTION,fs.createWriteStream('./logs')).warn({'fId': fId, 'job': targetId, 'jobId': targetJobs[i].id}, 'all inputs not available')
+                //pino(PINO_C_OPTION).warn({'fId': fId, 'job': targetId, 'jobId': targetJobs[i].id}, 'all inputs not available')
               }
               // let tmp = await this.updateProcess(targetJobs[i])
             }

JobQue/scheduler/index.js

@@ -22,13 +22,11 @@ module.exports = function (options) {
     if (options.logs.console == false) PINO_C_OPTION.level = 'silent'
     if (options.logs.db == false) {
       PINO_DB_OPTION.level = 'silent'
-    }
-    else {
+    } else {
       const SYSTEM_LOGS_TABLE = options.logs.table ? options.logs.table : app.system_logs_table
       enableWatcher(logs_table)
     }
-  }
-  else {
+  } else {
     enableWatcher(app.system_logs_table)
   }
 
@@ -37,48 +35,47 @@ module.exports = function (options) {
     const CHOKIDAR_OPTION = app.chokidar
     var watcher = chokidar.watch('./logs', CHOKIDAR_OPTION)
 
-    watcher.on('change', path =>
-      fs.readFile('./logs','utf8', function (err, data) {
-        if (err) throw err
-        let parsedData = JSON.parse(data)
-        rdash.table(SYSTEM_LOGS_TABLE).insert(parsedData).run(function (err , result) {
-          if (err) {
-            pino(PINO_DB_OPTION,fs.createWriteStream('./logs')).error({},err)
-            pino(PINO_C_OPTION).error({},err)
-          }
-        })
-      })
-    )
+    // watcher.on('change', path =>
+    //   // fs.readFile('./logs','utf8', function (err, data) {
+    //   //   if (err) throw err
+    //   //   let parsedData = JSON.parse(data)
+    //   //   rdash.table(SYSTEM_LOGS_TABLE).insert(parsedData).run(function (err , result) {
+    //   //     if (err) {
+    //   //       pino(PINO_DB_OPTION,fs.createWriteStream('./logs')).error({},err)
+    //   //       pino(PINO_C_OPTION).error({},err)
+    //   //     }
+    //   //   })
+    //   // })
+    // )
   }
 
   const func = new scheduler_functions(options, PINO_DB_OPTION, PINO_C_OPTION)
 
-  q.process(async (job, next) => {
+  q.process(async(job, next) => {
+
     try {
 
       const fId = job.data.fId
-      const flowInstance = await func.getFlowInstance(fId) //get flow instance from db
-      // pino(PINO_C_OPTION).info(job.data)
+      const flowInstance = await func.getFlowInstance(fId).catch(err => {throw err}); //get flow instance from db
+        // pino(PINO_C_OPTION).info(job.data)
       if (job.data.processNotification) {
         //the condition will be satisfied if the job is created by some process worker as a part
         //of notifying schcduler that the process completed succesfully a that particular worker
 
-        func.notificationACK(flowInstance, fId, job.data, next, cxnOptions, qOptions)
-      }
-      else if (job.data.isExternalInput) {
+        func.notificationACK(flowInstance, fId, job.data, next, cxnOptions, qOptions).catch(err => {throw err});
+      } else if (job.data.isExternalInput) {
         //the condition will be satisfied if the job is created in
         //order to provide external input to certain process
 
-        await func.performExternalOperation(flowInstance, job.data, fId, next)
-      }
-      else {
+        await func.performExternalOperation(flowInstance, job.data, fId, next).catch(err => {throw err});
+      } else {
         //i.e. the job in scheduler was created as a result of a new flowz instance
-        await func.newInstance(flowInstance, fId, next)
+        await func.newInstance(flowInstance, fId, next).catch(err => {throw err});
       }
     } catch (err) {
-      pino(PINO_DB_OPTION,fs.createWriteStream('./logs')).error({},'... error in process\n'+err)
-      pino(PINO_C_OPTION).error({},'... error in process '+err)
-      return next(err)
+      pino(PINO_DB_OPTION, fs.createWriteStream('./logs')).error({}, '... error in process\n' + err)
+      pino(PINO_C_OPTION).error({}, '... error in process ' + err)
+      return next(new Error(err.message),null)
     }
   })
 }

JobQue/scheduler/logs

@@ -1 +1 @@
-{"level":40,"time":1514959372879,"msg":"all inputs not available","pid":19672,"hostname":"OBLAP-51","fId":"1d54ad55-3606-4a22-af98-29895fa4e0c8","job":"Task_0a65jhx","jobId":"73de6e4e-2732-47c2-a737-e3f5269721a1","v":1}
+{"level":30,"time":1532702300476,"msg":"End Process","pid":18901,"hostname":"OBLAP-28","fId":"c8fe645b-4d59-4b08-9f9f-4fbb079c9fe3","jobId":"EndEvent_0f72gx5","v":1}

JobQue/scheduler/package.json

@@ -20,6 +20,7 @@
     "rethinkdb": "^2.3.3",
     "rethinkdb-job-queue": "^3.1.4",
     "twit": "^2.2.9",
+    "underscore": "^1.8.3",
     "vm": "^0.1.0"
   },
   "devDependencies": {

JobQue/scheduler/process.js

@@ -2,15 +2,16 @@ const vm = require('vm')
 const rp = require('request-promise')
 const pino = require('pino')
 const config = require('./default.json')
+const { exec, fork } = require('child_process')
 const PINO = config.pinoConsole
 
 global.options = []
-for (var i=4; i<process.argv.length; i++) {
+for (var i = 4; i < process.argv.length; i++) {
   options.push(process.argv[i])
 }
 const PROCESS_URL = process.argv[3]
 
-async function getProcess (jobType) {
+async function getProcess(jobType) {
   return new Promise((resolve, reject) => {
     rp(PROCESS_URL + jobType)
       .then(function (jobProcessCode) {
@@ -25,12 +26,23 @@ async function getProcess (jobType) {
 let executeProcess = async function (jobType) {
   try {
     let processCode = await getProcess(jobType)
-    const script = new vm.Script(`(function(require) {`+processCode +`})`, { filename: 'processTrace.vm' })
+    const script = new vm.Script(`(function(require) {` + processCode + `})`, { filename: 'processTrace.vm' })
     script.runInThisContext()(require)
-    pino(PINO).info({Process: jobType, fId: process.argv[6], pId: process.pid}, 'process created')
+    pino(PINO).info({ Process: jobType, fId: process.argv[6], pId: process.pid }, 'process created')
   } catch (e) {
-    pino(PINO).error(e)
-    pino(PINO).error('unable to load child process')
+    let file = __filename;
+    pino(PINO).error(e.message)
+    let error = e.message.split("'");
+    error[0] = error[0].slice(0, -1);
+    if (error[0] === 'Cannot find module') {
+      exec('cd ../scheduler && npm i ' + error[1], (error, stdout, stderr) => {
+        console.log(`${stdout}`);
+        process.argv.splice(0, 2);
+        fork(`${file}`, process.argv)
+      })
+    } else {
+      pino(PINO).error('unable to load child worker', jobType)
+    }
   }
 }
 

JobQue/seneca-job-queue/package.json

@@ -18,6 +18,7 @@
     "child_process": "^1.0.2",
     "chokidar": "^1.7.0",
     "cors": "^2.8.4",
+    "emailjs": "^2.2.0",
     "express": "^4.16.2",
     "express-fileupload": "^0.3.0",
     "fs": "0.0.1-security",
@@ -26,13 +27,14 @@
     "request": "^2.83.0",
     "request-promise": "^4.2.2",
     "rethinkdb": "^2.3.3",
-    "rethinkdb-job-queue": "^3.1.4",
+    "rethinkdb-job-queue": "^3.1.7",
     "rethinkdbdash": "^2.3.31",
     "seneca": "^3.4.3",
     "seneca-web": "^2.1.0",
     "seneca-web-adapter-express": "^1.0.2",
     "socket.io": "^2.0.4",
     "socket.io-client": "^2.0.4",
+    "underscore": "^1.8.3",
     "vm": "^0.1.0"
   }
 }