machine status info and input

Author: David Gómez Matarrodona

examples/config/data/replaceProblem.txt

@@ -0,0 +1,43 @@
+{
+	"config" : {
+		"datadir" : "/tmp/nsyslog",
+		"input" : {"buffer" : 100},
+		"collector" : {
+			"key" : "ctx",
+			"ival" : 1000
+		}
+	},
+
+	"inputs" : {
+		"stats" : {
+			"type" : "machine",
+			"config" : {
+			}
+		}
+	},
+
+	"processors" : {
+	},
+
+	"transporters" : {
+		"console" : {
+			"type" : "console",
+			"attach" : ["flow1"],
+			"config" : {
+				"format" : "${JSON:originalMessage.memory}",
+				"json" : {
+					"format" : true,
+					"spaces" : 2,
+					"color" : true
+				}
+			}
+		},
+		"null" : {
+			"type" : "null"
+		}
+	},
+
+	"flows" : [
+		{"id":"flow1", "from":"stats", "fork":true, "processors":[], "transporters":"console"}
+	]
+}

examples/config/test051_machine.json

@@ -0,0 +1,44 @@
+{
+	"config" : {
+		"datadir" : "/tmp/nsyslog",
+		"input" : {"buffer" : 100}
+	},
+
+	"inputs" : {
+		"lines" : {
+			"type" : "file",
+			"config" : {
+				"path" : "/opt/nsyslog/examples/config/data/replaceProblem.txt",
+				"readmode" : "offset",
+				"offset" : "start"
+			}
+		}
+	},
+
+	"processors" : {
+		"replace" : {
+			"type" : "properties",
+			"config" : {
+				"set" : {
+					"parsed" : "${originalMessage.replace(/(\\\\\\\\)/g,'\\\\').replace(/(\\\\)r(\\\\)n/g,' ')}"
+				}
+			}
+		}
+	},
+
+	"transporters" : {
+		"console" : {
+			"type" : "console",
+			"config" : {
+				"format" : "${parsed}"
+			}
+		},
+		"null" : {
+			"type" : "null"
+		}
+	},
+
+	"flows" : [
+		{"id":"flow1", "from":"lines", "fork":false, "processors":["replace"], "transporters":"console"}
+	]
+}

examples/config/test052_regex.json

@@ -2,6 +2,7 @@ const Watermark = require('../watermark');
 
 const
 	colorize = require('json-colorizer'),
+	jsexpr = require('jsexpr'),
 	{inspect} = require('util'),
 	Stats = require('../stats'),
 	cluster = require('../cluster'),
@@ -63,10 +64,16 @@ else {
 		  action(fnstats);
 
 		vorpal.
-		  command('global').
+		  command('global [expression]').
 			description('Shows global memory').
 		  action(function(args,callback){
-				vorpal.log(global._);
+				if(!args.expression) {
+					vorpal.log(global._);
+				}
+				else {
+					let xpr = jsexpr.expr("${"+args.expression+"}");
+					vorpal.log(xpr(global._));
+				}
 				callback();
 			});
 

lib/cli/basic.js

@@ -24,7 +24,8 @@ const INPUTS = {
 	sqlserver : srequire("../input/sqlserver"),
 	zmq : srequire("../input/zmq"),
 	ws : srequire("../input/ws"),
-	elastic : srequire("../input/elastic")
+	elastic : srequire("../input/elastic"),
+	machine : srequire("../input/machine")
 };
 
 Object.keys(INPUTS).forEach(p=>{

lib/config/inputs.js

@@ -0,0 +1,57 @@
+const
+	Input = require('.'),
+	MachineCollector = require('../machine');
+
+class MachineInput extends Input {
+	constructor(id,type) {
+		super(id,type);
+		this.collector = MachineCollector.default;
+		this.cb = null;
+	}
+
+	get mode() {
+		return Input.MODE.push;
+	}
+
+	async configure(config,callback) {
+		callback();
+	}
+
+	start(callback) {
+		this.cb = (data)=>callback(null,{
+			id : this.id,
+			type : this.type,
+			originalMessage : data
+		});
+		this.collector.on(MachineCollector.Event.status,this.cb);
+	}
+
+	stop(callback) {
+		if(this.cb) {
+			this.collector.removeListener(MachineCollector.Event.status,this.cb);
+			this.cb = null;
+		}
+		callback();
+	}
+
+	pause(callback) {
+		if(this.cb) {
+			this.collector.removeListener(MachineCollector.Event.status,this.cb);
+		}
+		callback();
+	}
+
+	resume(callback) {
+		if(this.cb) {
+			this.collector.removeListener(MachineCollector.Event.status,this.cb);
+			this.collector.on(MachineCollector.Event.status,this.cb);
+		}
+		callback();
+	}
+
+	key(entry) {
+		return `${entry.input}:${entry.type}`;
+	}
+}
+
+module.exports = MachineInput;

lib/input/machine.js

@@ -0,0 +1,90 @@
+const 
+    extend = require('extend'),
+    os = require('os'),
+    oss = require('os-utils'),
+    nodeDiskInfo = require('node-disk-info'),
+    EventEmitter = require('events'),
+    Shm = require('./shm');
+
+const DEF_CONF = {
+    key : 'ctx',
+    ival : 30000
+}
+
+const EVENTS = {
+    status : 'status'
+};
+
+var INSTANCE = null;
+
+class MachineCollector extends EventEmitter {
+    constructor() {
+        super();
+
+        this.key = DEF_CONF.key;
+        this.ival = DEF_CONF.ival;
+        this.to = null;
+    }
+
+    static get Event() {
+        return EVENTS;
+    }
+
+    static get Defaults() {
+        return DEF_CONF
+    }
+    
+    static get default() {
+        if(!INSTANCE) {
+            INSTANCE = new MachineCollector();
+        }
+        return INSTANCE;
+    }
+
+    configure(options) {
+        options = extend(true,{},DEF_CONF,options);
+        
+        this.ival = options.ival;
+        this.key = options.key;
+    }
+
+    start() {
+        this.loop();
+    }
+
+    stop() {
+        clearTimeout(this.to);
+        this.to = null;
+    }
+
+    async loop() {
+        let res = await this.collectMachine();
+        Shm.hset('global',this.key,res);
+        this.emit('status',res);
+        this.to = setTimeout(()=>this.loop(),this.ival);
+    }
+
+    async collectMachine() {
+        return {
+            platform : process.platform,
+            arch : process.arch,
+            sysUpTime : oss.sysUptime(),
+            processUpTime : oss.processUptime(),
+            cpu : os.cpus(),
+            env : process.env,
+            ifaces : Object.entries(os.networkInterfaces()).map(i=>({name:i[0],data:i[1]})),
+            disk : await nodeDiskInfo.getDiskInfo(),
+            cpuLoad : {
+                min1:oss.loadavg(1),
+                min5:oss.loadavg(5),
+                min15:oss.loadavg(15)
+            },
+            memory : {
+                total : os.totalmem(),
+                free : os.freemem()
+            }
+        }
+    }    
+}
+
+module.exports = MachineCollector;

lib/machine.js

@@ -1,11 +1,13 @@
 const
 	logger = require("../logger"),
+	extend = require('extend'),
 	Logger = require("../logger"),
 	mingo = require('../mingo'),
 	ClusterAware = require("./clusteraware"),
 	GlobalDB = require('./globaldb'),
 	FileQueue = require("fileq"),
 	Reemiter = require("../reemiter"),
+	MachineCollector = require('../machine'),
 	Config = require("../config"),
 	InputWrapper = require("../input/wrapper"),
 	Transporters = Config.Transporters,
@@ -77,6 +79,8 @@ class NSyslog extends ClusterAware {
 		this.flowmap = {};
 		this.pubsub = null;
 
+		this.collector = MachineCollector.default;
+
 		Reemiter.configure(this);
 	}
 
@@ -315,6 +319,15 @@ class NSyslog extends ClusterAware {
 		});
 	}
 
+	async startCollector() {
+		let conf = extend(true,{config:{collector:MachineCollector.Defaults}},this.config);
+		if(conf.config.collector.enabled!==false) {
+			logger.info('Starting machine collector...');
+			this.collector.configure(conf.config.collector);
+			this.collector.start();
+		}
+	}
+
 	/**
 	 * Pushes data manually to the nsyslog instance, writing them to the input stream,
 	 * in order to be sent to the flows
@@ -352,6 +365,7 @@ class NSyslog extends ClusterAware {
 		logger.info(`\tProcessors`,Object.keys(modules.processors));
 		logger.info(`\tTransporters`,Object.keys(modules.transporters));
 
+		await this.startCollector();
 		await this.startPubSub();
 		this.startChildren();
 		this.startProcessorStream();
@@ -387,6 +401,8 @@ class NSyslog extends ClusterAware {
 
 		logger.info('Closing child processes');
 		this.children.forEach(f=>f.fstream.closefork());
+		logger.info('Stopping machine collector...');
+		this.collector.stop();
 		logger.info('Stopping inputs...');
 		await Promise.all(Object.keys(inputs).map(key=>inputs[key]).filter(m=>m.own).map(m=>prfn(m,'stop')));
 		logger.info('Stopping processors...');

lib/nsyslog/index.js

@@ -60,9 +60,11 @@
     "minimatch": "^3.0.4",
     "mkdirp": "^0.5.1",
     "mongodb": "^3.3.3",
+    "node-disk-info": "^1.3.0",
     "node-forge": "^0.10.0",
     "nools": "^0.4.4",
     "nsyslog-parser": "^0.9.6",
+    "os-utils": "0.0.14",
     "pemtools": "^0.4.7",
     "promise-stream-queue": "^0.4.5",
     "request": "^2.87.0",