Initial commit

Author: Brian Crowell

.gitignore

@@ -0,0 +1,3 @@
+/node_modules
+/logs
+config.json

LICENSE

@@ -0,0 +1,14 @@
+Copyright 2016–7 Anonymous
+Copyright 2018 Brian Crowell
+
+Permission to use, copy, modify, and/or distribute this software for any
+purpose with or without fee is hereby granted, provided that the above
+copyright notice and this permission notice appear in all copies.
+
+THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
+REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
+INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR
+OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+PERFORMANCE OF THIS SOFTWARE.

README.md

@@ -0,0 +1,82 @@
+# node-log-forwarder
+
+Node.js-based log parsing, forwarding, and notifications.
+
+This is a server that provides the functionality of products like
+Logstash and Fluentd—it accepts log data from remote sources,
+parses them, processes them, and forwards them to notification or
+database systems such as Slack, IRC, or Elasticsearch.
+
+But unlike Logstash or Fluentd...
+
+**Processing log entries is done in user-provided scripts written in JavaScript.** Make the processing logic as complex as you need to, and take advantage of any Node.js-compatible library.
+
+A syslog receiver that sends fully-attributed structured syslog to Elasticsearch and messages to IRC channels might look like this:
+
+```javascript
+const d3 = require('d3');
+const dateFormat = d3.timeFormat('%Y.%m.%d');
+
+function preprocess(ctx, line) {
+  return {
+    reportingIp: ctx.meta.remoteAddress,
+    receivingPort: ctx.meta.localPort,
+    receivedTime: ctx.meta.receiveTime,
+    eventTime: ctx.meta.receiveTime,
+    message: (line instanceof Buffer) ? line.toString('latin1') : line,
+    tag: ['raw'],
+  };
+}
+
+function process(ctx, msg) {
+  ctx.sendElasticsearch('raw-syslog-' + dateFormat(msg.eventTime), 'raw-syslog');
+  ctx.sendIrc('#syslog');
+
+  if(msg.message.startsWith('ERROR')) {
+    msg.tag.push('error');
+    ctx.sendIrc('#syslog_errors');
+  }
+
+  return { log: msg };
+}
+```
+
+**Almost any configuration or script change takes effect immediately** without restarting, closing sockets, losing connections, or losing messages. Alter the script above and save it and instantly see the effects of your changes.
+
+It also features very fast startup, and it spreads messages across multiple worker processes for increased throughput on multiprocessor systems.
+
+## Provided inputs
+
+Built-in support for receiving:
+
+* Netflow V9
+* UDP (such as syslog)
+* Line-based TCP (such as syslog, Bunyan, or custom log formats)
+
+## Provided outputs
+
+Built-in support for sending formatted results to:
+
+* Local files, with filenames supplied by the user script. Use this to structure your log files in directories by source IP, date, both, or whatever other naming scheme you like.
+* Elasticsearch, with support for bulk uploads and throughput statistics by worker
+* Slack, with full custom formatting
+* IRC
+* SMTP
+
+# Setup
+
+TODO, but read [config.json.example](./config.json.example)
+
+# Writing a user script
+
+TODO, but see the [sample user scripts](./sample-user-scripts)
+
+# Writing an input module
+
+TODO, but see the [built-in input modules](./inputs)
+
+# Contact, acknowledgements
+
+Written by Brian Crowell, with special thanks to the organization that supported this project, who has asked to remain anonymous.
+
+Please do get in touch if you use this project, I would love to hear about it!

config.json.example

@@ -0,0 +1,105 @@
+// This is a sample configuration file for the forwarder. Copy
+// this to config.json to create your configuration file.
+//
+// Settings marked "(dynamic)" can be changed while the program
+// is running and will take effect when the file is saved.
+// Settings marked "(static)" should not be changed while the
+// program is running.
+
+{
+  // The name for this instance, used in logging statistics (dynamic)
+  "name": "test-logger",
+
+  // Map of input names to modules (dynamic)
+  //
+  // Inputs can be added or removed at run-time.
+  "inputs": {
+
+    // Name of the input (static). Changes to the config are matched back
+    // to the input module by name, so don't change this while the
+    // forwarder is running.
+    "bunyan": {
+
+      // Path to the module that provides the input (static)
+      //
+      // Different modules will have different options. This input uses
+      // tcp-lines, which accepts TCP connections, splits input by newlines,
+      // and provides each message to the given user script.
+      "module": "./inputs/tcp-lines.js",
+
+      // Path to the user script (dynamic)
+      //
+      // This script decides how each message is processed. Changes to this
+      // path or to the script are reflected as soon as they are seen.
+      // For tcp-lines, see sample scripts msvistalog.js, bunyan.js, or cylance.js.
+      "script": "sample-user-scripts/bunyan.js",
+
+      // Host to bind the socket to (optional, dynamic); if unspecified, binds
+      // to all interfaces
+      // "host": "localhost"
+
+      // Port to bind the socket to (dynamic)
+      "port": 5022
+    },
+    "syslog": {
+      // Same as above, but for a UDP input.
+      "module": "./inputs/udp.js",
+      "script": "sample-user-scripts/syslog.js",
+
+      // Optional "host"
+      // "host": "localhost"
+
+      "port": 5144
+    }
+  },
+
+  // Number of workers spawned (dynamic).
+  //
+  // Each worker can use up to one CPU core, so set this to the
+  // number of cores available on your system. It can be adjusted
+  // at run time.
+  "workerCount": 2,
+
+  // Log file writing
+  "file": {
+    // The base path to which log files will be written; paths outside this are not allowed
+    "basePath": "/home/bcrowell/software/node-log-forwarder/logs"
+  },
+
+  // SMTP support (optional, dynamic)
+  "mail": {
+    // Transport options, either a URL or an object (dynamic)
+    // See https://www.npmjs.com/package/nodemailer#set-up-smtp
+    "transportOptions": "smtp://my.mail.server"
+  },
+
+  // IRC support (optional, dynamic)
+  "irc": {
+    "server": "myirc.company.net",
+    "nick": "logbot",
+
+    // Options for the IRC library; see https://node-irc.readthedocs.org/en/latest/API.html#client
+    "options": {
+      // These are the defaults we use
+      // "port": 6667,
+      // "encoding": "utf8",
+      // "autoRejoin": true
+      //"debug": true
+    }
+  },
+
+  // Elasticsearch support (optional, dynamic)
+  "elasticsearch": {
+    "clientSettings": {
+      "hosts": [
+        "http://10.0.0.1:9200",
+        "http://10.0.0.2:9200"
+      ],
+      "apiVersion": "5.0",
+      "sniffOnStart": true,
+      "sniffInterval": 60000,
+      "sniffOnConnectionFault": true
+    },
+    "bulkSizeMegabytes": 10
+  }
+}

index.js

@@ -0,0 +1,45 @@
+'use strict';
+
+const config = require('./lib/config.js');
+const cluster = require('cluster');
+
+const log = config.log;
+
+if(cluster.isMaster) {
+  log.info({messageId: 'master/started'}, 'Starting up master process.');
+
+  cluster.on('fork', worker => {
+    checkRunningWorkers();
+  });
+
+  cluster.on('disconnect', worker => {
+    log.warn({messageId: 'master/worker-disconnected'}, `Worker #${worker.id} (PID ${worker.process.pid}) has disconnected.`);
+    checkRunningWorkers();
+  });
+
+  cluster.on('exit', worker => {
+    log.warn({messageId: 'master/worker-exited'}, `Worker #${worker.id} (PID ${worker.process.pid}) has exited.`);
+    checkRunningWorkers();
+  });
+
+  function checkRunningWorkers() {
+    const targetWorkerCount = config.config.workerCount || 0;
+    const workerIds = Object.keys(cluster.workers);
+
+    if(workerIds.length > targetWorkerCount) {
+      log.warn({messageId: 'master/too-many-workers'}, `Killing worker to meet worker count.`);
+      cluster.workers[workerIds[0]].kill();
+    }
+    else if(workerIds.length < targetWorkerCount) {
+      cluster.fork();
+    }
+  }
+
+  config.on('change', newConfig => {
+    checkRunningWorkers();
+  });
+}
+else {
+  log.info({messageId: 'worker/started'}, `Starting up worker process ${cluster.worker.id}.`);
+}
+