Logger improved

Author: jd78

Application/ExecuteLocked.js

@@ -1,13 +1,14 @@
 var ReadWriteLock  = require("rwlock");
 var q = require("q");
-var logger = require("./Logger");
+var Logger = require("./Logger")
 
 function ExecuteLocked(){
     this.lock = new ReadWriteLock();
 }
 
 ExecuteLocked.prototype.execRead = function(func){
     var deferred = q.defer();
+    var logger = Logger.instance();
     this.lock.readLock(function(release){
        func().then(function(obj){
             release();
@@ -22,6 +23,7 @@ ExecuteLocked.prototype.execRead = function(func){
 
 ExecuteLocked.prototype.execWrite = function(func){
     var deferred = q.defer();
+    var logger = Logger.instance();
     this.lock.writeLock(function(release){
        func().then(function(obj){
             release();

Application/Jobs.js

@@ -1,4 +1,4 @@
-var logger = require("./Logger");
+var Logger = require("./Logger");
 var Message = require("../Entities/Message");
 var util = require("util");
 var q = require("q");
@@ -7,6 +7,7 @@ function Jobs(){}
 
 Jobs.prototype.executeJob = function(job) {
     var self = this;
+    var logger = Logger.instance();
     return q.Promise(function(resolve, reject){
         logger.debug("executing job id: %d, partitionId: %d, type: %s, pid: %d", job.id, job.partitionId, job.type, process.pid);
         if(self[job.type] === undefined) {

Application/Logger.js

@@ -1,31 +1,42 @@
 var winston = require("winston");
 var fs = require("fs");
 var util = require("util");
+var q = require("q");
 
-var binPath = "bin";
-var servicePath = "parallel-queue-partitioner";
-var fullpath = util.format("./%s/%s", binPath, servicePath);
+var logger;
 
-if (!fs.existsSync(util.format("./%s", binPath))) {
-	fs.mkdirSync(util.format("./%s", binPath));
+function newLogger(){
+  return q.Promise(function(resolve){
+    var binPath = "bin";
+    var servicePath = "parallel-queue-partitioner";
+    var fullpath = util.format("./%s/%s", binPath, servicePath);
+  
+    if (!fs.existsSync(util.format("./%s", binPath))) {
+    	fs.mkdirSync(util.format("./%s", binPath));
+    }
+  
+    if (!fs.existsSync(fullpath)) {
+    	fs.mkdirSync(fullpath);
+    }
+    
+    logger = new (winston.Logger)({
+      transports: [
+        new (winston.transports.Console)(),
+        new (winston.transports.File)({ 
+            filename: util.format("%s/pid-%s-partitioner.log", fullpath, process.pid),
+            handleExceptions: true,
+            exitOnError: false,
+            level: 'error', //info, warning, error
+            maxsize: 625000,
+            zippedArchive: true
+        })
+      ]
+    });
+    resolve(logger);
+  });
 }
 
-if (!fs.existsSync(fullpath)) {
-	fs.mkdirSync(fullpath);
-}
-
-var logger = new (winston.Logger)({
-    transports: [
-      new (winston.transports.Console)(),
-      new (winston.transports.File)({ 
-          filename: util.format("%s/pid-%s-partitioner.log", fullpath, process.pid),
-          handleExceptions: true,
-          exitOnError: false,
-          level: 'error', //info, warning, error
-          maxsize: 625000,
-          zippedArchive: true
-      })
-    ]
-});
-
-module.exports = logger;
+module.exports = {
+  new: newLogger,
+  instance: function(){ return logger; }
+};

Application/Worker.js

@@ -1,10 +1,10 @@
 var jobService = require("../Services/JobService");
 var cluster = require("cluster");
 var jobs = require("./Jobs");
-var logger = require("./Logger");
 var ReadWriteLock  = require("rwlock");
 var queueLock = new ReadWriteLock();
 var progressLock = new ReadWriteLock();
+var Logger = require("./Logger");
 
 function Worker(worker){
     this.worker = worker;
@@ -15,7 +15,7 @@ function Worker(worker){
             return;
         }
 
-        logger.debug("master received completed notify for jobId %d", message.id);
+        Logger.instance().debug("master received completed notify for jobId %d", message.id);
         jobService.done(message.id);
     });
 }
@@ -24,13 +24,15 @@ var queue = [];
 var inProgress = false;
 
 if(cluster.isWorker) {
-    logger.transports.file.level = process.env["loggerLevel"];
-    logger.transports.console.level = process.env["loggerLevel"];
+    Logger.new().then(function(log){
+        log.transports.file.level = process.env["loggerLevel"];
+        log.transports.console.level = process.env["loggerLevel"];
+        
+        log.info("worker %d registered", process.pid);
+    });
 
-    logger.info("worker %d registered", process.pid);
-
     process.on('message', function(job) { 
-        logger.debug("job %d received", job.id);
+        Logger.instance().debug("job %d received", job.id);
         queueLock.writeLock(function(release){
             queue.push(function() { return jobs.executeJob(job) });
             release();

Partitioner.js

@@ -5,13 +5,12 @@ var jobs = require("./Application/Jobs");
 var Lock = require("./Application/ExecuteLocked");
 var lock = new Lock();
 var Worker = require("./Application/Worker");
-var logger = require("./Application/Logger");
 var validator = require("validator");
 
-
 var workers = [];
 var workerPartitionIndex = 0;
 var numberOfWorkers;
+var logger;
 
 var defaultConfiguration = {
   numberOfWorkers: 1,
@@ -20,6 +19,9 @@ var defaultConfiguration = {
 };
 
 function Partitioner(configuration) {
+    if(cluster.isWorker)
+        throw new Error("a worker is trying to instantiate a partitioner");
+    
     if(configuration !== undefined)
         validate(configuration);
 
@@ -28,20 +30,22 @@ function Partitioner(configuration) {
     this.partitionService = new PartitionService(config.cleanIdlePartitionsAfterMinutes || 15);
 
     var processEnv = {};
-    if(config.loggerLevel !== undefined){
-        logger.transports.file.level = config.loggerLevel;
-        logger.transports.console.level = config.loggerLevel;
-        processEnv["loggerLevel"] = config.loggerLevel;
-    }else {
-        processEnv["loggerLevel"] = defaultConfiguration.loggerLevel;
-    }
 
-    if(cluster.isWorker)
-        throw new Error("a worker is trying to instantiate a partitioner");
-    
-    for(var i=0; i < numberOfWorkers; i++){
-        workers.push(new Worker(cluster.fork(processEnv)));
-    }
+    var Logger = require("./Application/Logger");
+    Logger.new().then(function(log){
+        logger = log;    
+        if(config.loggerLevel !== undefined){
+            logger.transports.file.level = config.loggerLevel;
+            logger.transports.console.level = config.loggerLevel;
+            processEnv["loggerLevel"] = config.loggerLevel;
+        }else {
+            processEnv["loggerLevel"] = defaultConfiguration.loggerLevel;
+        }
+        
+        for(var i=0; i < numberOfWorkers; i++){
+            workers.push(new Worker(cluster.fork(processEnv)));
+        }
+    });
 }
 
 Partitioner.prototype.enqueueJob = function(job, callback){

README.md

@@ -112,7 +112,8 @@ var partitioner = new Partitioner({
 
 ## Logger
 
-A logger will be created in the folder ./bin/parallel-queue-partitioner/partitioner.log
+A logger will be created in the folder ./bin/parallel-queue-partitioner.
+Each worker will create its own log. 
 
 ## Exemple how to use the partitioner with redis
 

Samples/test-redis.js

@@ -37,25 +37,25 @@ var id=0;
 
 if(cluster.isMaster) {
     console.log('pushing messages');
-    for (var i = 0; i < 50; i++) {
-        queue.create('jobs', {
-            partitionId: 0,
-            type: "sequential",
-            sequence: i
-        }).save(function(err) {
-            if (err) console.log(err);
-        });
-    }
+    // for (var i = 0; i < 50; i++) {
+    //     queue.create('jobs', {
+    //         partitionId: 0,
+    //         type: "sequential",
+    //         sequence: i
+    //     }).save(function(err) {
+    //         if (err) console.log(err);
+    //     });
+    // }
 
-    for (var i = 0; i < 50; i++) {
-        queue.create('jobs', {
-            partitionId: 1,
-            type: "sequential",
-            sequence: i
-        }).save(function(err) {
-            if (err) console.log(err);
-        });
-    }
+    // for (var i = 0; i < 50; i++) {
+    //     queue.create('jobs', {
+    //         partitionId: 1,
+    //         type: "sequential",
+    //         sequence: i
+    //     }).save(function(err) {
+    //         if (err) console.log(err);
+    //     });
+    // }
 
     for (var i = 0; i < 150; i++) {
         queue.create('jobs', {
@@ -72,7 +72,7 @@ if(cluster.isMaster) {
 function start(){
     var partitioner = new Partitioner({
         numberOfWorkers: 4,
-        loggerLevel: 'info'
+        loggerLevel: 'debug'
     });
 
     setTimeout(function(){

Services/JobService.js

@@ -1,6 +1,6 @@
 var _ = require("underscore");
 var q = require("q");
-var logger = require("../Application/Logger");
+var Logger = require("../Application/Logger");
 
 
 var jobs = [];
@@ -29,7 +29,7 @@ JobService.prototype.push = function(id, callback) {
             jobs.push(new Job(id, callback));
             resolve();
         }catch(ex){
-            logger.error(ex);
+            Logger.instance().error(ex);
             reject(ex);
         }
     });

Services/PartitionService.js

@@ -1,7 +1,7 @@
 var _ = require("underscore");
 var moment = require("moment");
 var q = require("q");
-var logger = require("../Application/Logger");
+var Logger = require("../Application/Logger");
 var util = require("util");
 
 var Lock = require("../Application/ExecuteLocked");
@@ -17,6 +17,7 @@ function Partition(partitionId, worker) {
     this.updatedAt = moment().utc().format();
 
     setInterval(function() {
+        var logger = Logger.instance();
         logger.info(util.format("Partition cleanup fired for %d", partitionId));
         if(self.updatedAt <= moment().utc().subtract(cleanIdlePartitionsAfterMinutes, 'minutes').format())
             lock.execWrite(function(){ 
@@ -51,7 +52,7 @@ PartitionService.prototype.get = function(partitionId) {
                 resolve(maybePartition);
 
             } catch(ex){
-                logger.error(ex);
+                Logger.instance().error(ex);
                 reject(ex);
             }
         });