Better logging

Separated log streams for access log (defaults to stdout) and general
log (defaults to stderr). Added support for writing logs to a file.
Improved formatting and added support for JSON formatted logs.

lib/cluster.coffee

@@ -1,58 +0,0 @@
-###
-# Copyright 2013, 2014  Zaid Abdulla
-#
-# GenieACS is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Affero General Public License as
-# published by the Free Software Foundation, either version 3 of the
-# License, or (at your option) any later version.
-#
-# GenieACS is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU Affero General Public License for more details.
-#
-# You should have received a copy of the GNU Affero General Public License
-# along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
-###
-
-config = require './config'
-cluster = require 'cluster'
-util = require 'util'
-
-
-cluster.on('listening', (worker, address) ->
-  util.log("Worker #{worker.process.pid} listening to #{address.address}:#{address.port}")
-)
-
-restartWorker = (worker, code, signal) ->
-  util.log("Worker #{worker.process.pid} died (#{worker.process.exitCode}) #{signal}")
-  cluster.fork()
-
-cluster.on('exit', restartWorker)
-
-
-start = (service) ->
-  console.error('WARNING: This is a development branch of GenieACS. DO NOT use in production.')
-  workerProcesses = config.get("#{service.toUpperCase()}_WORKER_PROCESSES")
-
-  cluster.setupMaster({
-      exec : 'lib/server',
-      args : [service]
-  })
-
-  for [0 ... workerProcesses]
-    cluster.fork()
-
-
-process.on('SIGINT', () ->
-  cluster.removeListener('exit', restartWorker)
-)
-
-process.on('SIGTERM', () ->
-  cluster.removeListener('exit', restartWorker)
-  for id of cluster.workers
-    cluster.workers[id].kill()
-)
-
-
-exports.start = start

lib/cluster.js

@@ -0,0 +1,138 @@
+/**
+ * Copyright 2013-2017  Zaid Abdulla
+ *
+ * GenieACS is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * GenieACS is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
+ */
+"use strict";
+
+const cluster = require("cluster");
+const logger = require("./logger");
+
+const VERSION = require("../package.json").version;
+
+
+function getDependencyVersions() {
+  let dependencies = [`node@${process.versions.node}`];
+  try {
+    let npmls = JSON.parse(require("child_process").execSync("npm ls --prod --json"));
+
+    for (let k in npmls.dependencies)
+      dependencies.push(`${k}@${npmls.dependencies[k].version}`);
+  } catch (err) {}
+
+  return dependencies.join(",");
+}
+
+
+function getConfig() {
+  let conf = [];
+  const keys = Object.keys(process.env);
+  for (let k of keys)
+    if (k.startsWith("GENIEACS_"))
+      conf.push(`${k.slice(9)}=${process.env[k]}`);
+
+  return conf.join(",");
+}
+
+
+function restartWorker(worker, code, signal) {
+  let msg = {
+    message: "Worker died",
+    pid: worker.process.pid
+  };
+
+  if (code != null)
+    msg.exitCode = code;
+
+  if (signal != null)
+    msg.signal = signal;
+
+  logger.error(msg);
+  cluster.fork();
+}
+
+
+function start(service) {
+  logger.init(service, VERSION);
+  const config = require("./config");
+
+  logger.info({
+    message: `GenieACS (genieacs-${service}) starting`,
+    pid: process.pid,
+    version: VERSION,
+    dependencies: getDependencyVersions(),
+    config: getConfig()
+  });
+
+  cluster.on("listening", function(worker, address) {
+    logger.info({
+      message: "Worker listening",
+      pid: worker.process.pid,
+      address: address.address,
+      port: address.port
+    });
+  });
+
+  cluster.on("exit", restartWorker);
+
+  process.on("SIGINT", function() {
+    logger.info({
+      message: "Received signal SIGINT, exiting",
+      pid: process.pid
+    });
+
+    cluster.removeListener("exit", restartWorker);
+  });
+
+  process.on("SIGTERM", function() {
+    logger.info({
+      message: "Received signal SIGTERM, exiting",
+      pid: process.pid
+    });
+
+    cluster.removeListener("exit", restartWorker);
+    for (let pid in cluster.workers)
+      cluster.workers[pid].kill();
+  });
+
+  logger.warn({
+    message: "This is a development branch of GenieACS. DO NOT use in production."
+  });
+
+  cluster.setupMaster({
+    exec: "lib/server",
+    args: [service]
+  });
+
+  let workerCount = config.get(`${service.toUpperCase()}_WORKER_PROCESSES`);
+
+  for (let i = 0; i < workerCount; ++ i)
+    cluster.fork();
+}
+
+
+process.on("uncaughtException", function(err) {
+  logger.error({
+    message: "Uncaught exception in master process, exiting",
+    exception: err
+  });
+  cluster.removeListener("exit", restartWorker);
+  for (let pid in cluster.workers)
+    cluster.workers[pid].kill();
+
+  logger.close();
+});
+
+
+exports.start = start;

lib/config.coffee

@@ -29,22 +29,27 @@ options = {
   CWMP_PORT : {type : 'int', default : 7547},
   CWMP_INTERFACE : {type : 'string', default : '0.0.0.0'},
   CWMP_SSL : {type : 'bool', default : false},
+  CWMP_LOG_FILE : {type: 'path', default : ''},
+  CWMP_ACCESS_LOG_FILE : {type : 'path', default : ''},
 
   NBI_WORKER_PROCESSES : {type : 'int', default : 2},
   NBI_PORT : {type : 'int', default : 7557},
   NBI_INTERFACE : {type : 'string', default : '0.0.0.0'},
   NBI_SSL : {type : 'bool', default : false},
+  NBI_LOG_FILE : {type: 'path', default : ''},
+  NBI_ACCESS_LOG_FILE : {type : 'path', default : ''},
 
   FS_WORKER_PROCESSES : {type : 'int', default : 2},
   FS_PORT : {type : 'int', default : 7567},
   FS_INTERFACE : {type : 'string', default : '0.0.0.0'},
   FS_SSL : {type : 'bool', default : false},
   FS_IP : {type : 'string', default : '192.168.0.1'},
+  FS_LOG_FILE : {type: 'path', default : ''},
+  FS_ACCESS_LOG_FILE : {type : 'path', default : ''},
 
   DOWNLOAD_TIMEOUT: {type : 'int', default : 3600},
   EXT_TIMEOUT: {type: 'int', default: 3000},
   PRESETS_CACHE_DURATION : {type : 'int', default : 86400},
-  LOG_INFORMS : {type : 'bool', default : true},
   DEBUG : {type : 'bool', default : false},
   RETRY_DELAY : {type : 'int', default : 300},
   SESSION_TIMEOUT : {type : 'int', default : 30},
@@ -53,6 +58,8 @@ options = {
   TASK_PARAMETERS_BATCH_SIZE : {type : 'int', default : 32},
   MAX_DEPTH : {type: 'int', default : 16},
   COOKIES_PATH : {type : 'string'},
+  LOG_FORMAT : {type : 'string', default : 'simple'},
+  ACCESS_LOG_FORMAT : {type : 'string', default : ''},
 
   # XML configuration
   XML_RECOVER : {type : 'bool', default : false},
@@ -81,7 +88,7 @@ setConfig = (name, value, commandLineArgument) ->
       when 'string'
         String(val)
       when 'path'
-        path.resolve(val)
+        if val then path.resolve(val) else ''
       else
         null
 
@@ -109,16 +116,12 @@ setConfig = (name, value, commandLineArgument) ->
 
 
 # Command line arguments
-exports.argv = []
 argv = process.argv[2..]
 while argv.length
   arg = argv.shift()
   if arg[0] == '-'
     v = argv.shift()
-    exports.argv[arg] = v
     setConfig(arg[2..], v, true)
-  else
-    exports.argv.push(arg)
 
 
 # Environment variable

lib/cwmp.coffee

@@ -50,6 +50,7 @@ query = require './query'
 device = require './device'
 cache = require './cache'
 db = require './db'
+logger = require './logger'
 
 MAX_CYCLES = 4
 
@@ -92,7 +93,8 @@ writeResponse = (currentRequest, res) ->
     currentRequest.httpResponse.end(res.data)
 
 
-recordFault = (sessionData, fault, provisions, channels) ->
+recordFault = (currentRequest, fault, provisions, channels) ->
+  sessionData = currentRequest.sessionData
   faults = sessionData.faults
 
   for channel of channels
@@ -111,7 +113,14 @@ recordFault = (sessionData, fault, provisions, channels) ->
 
     sessionData.faultsTouched ?= {}
     sessionData.faultsTouched[channel] = true
-    util.log("#{sessionData.deviceId}: Fault occurred in channel #{channel} (retries: #{sessionData.retries[channel]})")
+
+    logger.accessError({
+      currentRequest: currentRequest
+      message: 'Channel has faulted'
+      fault: fault
+      channel: channel
+      retries: sessionData.retries[channel]
+    })
 
   for provision, i in provisions
     for channel of channels
@@ -128,9 +137,6 @@ recordFault = (sessionData, fault, provisions, channels) ->
 
 
 inform = (currentRequest, cwmpRequest) ->
-  if config.get('LOG_INFORMS', currentRequest.sessionData.deviceId)
-    util.log("#{currentRequest.sessionData.deviceId}: Inform (#{cwmpRequest.methodRequest.event}); retry count #{cwmpRequest.methodRequest.retryCount}")
-
   session.inform(currentRequest.sessionData, cwmpRequest.methodRequest, (err, rpcResponse) ->
     return throwError(err, currentRequest.httpResponse) if err
     res = soap.response({
@@ -149,14 +155,21 @@ inform = (currentRequest, cwmpRequest) ->
 
 
 transferComplete = (currentRequest, cwmpRequest) ->
-  session.transferComplete(currentRequest.sessionData, cwmpRequest.methodRequest, (err, rpcResponse, fault, operation) ->
+  session.transferComplete(currentRequest.sessionData, cwmpRequest.methodRequest, (err, rpcResponse, operation, fault) ->
     return throwError(err, currentRequest.httpResponse) if err
 
+    if not operation?
+      logger.accessWarn({
+        currentRequest: currentRequest
+        message: 'Unrecognized command key'
+        cpeRequest: cwmpRequest.methodRequest
+      })
+
     if fault
       for k, v of operation.retries
         currentRequest.sessionData.retries[k] = v
 
-      recordFault(currentRequest.sessionData, fault, operation.provisions, operation.channels)
+      recordFault(currentRequest, fault, operation.provisions, operation.channels)
 
     res = soap.response({
       id : cwmpRequest.id,
@@ -280,7 +293,7 @@ applyPresets = (currentRequest) ->
       return throwError(err, currentRequest.httpResponse) if err
 
       if fault
-        recordFault(sessionData, fault, sessionData.provisions, sessionData.channels)
+        recordFault(currentRequest, fault, sessionData.provisions, sessionData.channels)
         session.clearProvisions(sessionData)
         return applyPresets(currentRequest)
 
@@ -321,7 +334,7 @@ applyPresets = (currentRequest) ->
           message: 'The provision seems to be repeating indefinitely'
           timestamp: sessionData.timestamp
         }
-        recordFault(sessionData, fault, sessionData.provisions, sessionData.channels)
+        recordFault(currentRequest, fault, sessionData.provisions, sessionData.channels)
         session.clearProvisions(sessionData)
         return sendRpcRequest(currentRequest)
 
@@ -329,7 +342,7 @@ applyPresets = (currentRequest) ->
         return throwError(err, currentRequest.httpResponse) if err
 
         if fault
-          recordFault(sessionData, fault, sessionData.provisions, sessionData.channels)
+          recordFault(currentRequest, fault, sessionData.provisions, sessionData.channels)
           session.clearProvisions(sessionData)
           return applyPresets(currentRequest)
 
@@ -354,7 +367,7 @@ nextRpc = (currentRequest) ->
     return throwError(err, currentRequest.httpResponse) if err
 
     if fault
-      recordFault(currentRequest.sessionData, fault, currentRequest.sessionData.provisions, currentRequest.sessionData.channels)
+      recordFault(currentRequest, fault, currentRequest.sessionData.provisions, currentRequest.sessionData.channels)
       session.clearProvisions(currentRequest.sessionData)
       return nextRpc(currentRequest)
 
@@ -383,7 +396,11 @@ nextRpc = (currentRequest) ->
 
       # Delete if expired
       if task.expiry <= currentRequest.sessionData.timestamp
-        util.log("#{currentRequest.sessionData.deviceId}: Task is expired #{task.name}(#{task._id})")
+        logger.accessInfo({
+          currentRequest: currentRequest
+          message: 'Task expired'
+          task: task
+        })
         currentRequest.sessionData.doneTasks ?= []
         currentRequest.sessionData.doneTasks.push(String(task._id))
         if channel of currentRequest.sessionData.faults
@@ -443,7 +460,10 @@ sendRpcRequest = (currentRequest, id, rpcRequest) ->
 
       delete currentSessions.get(currentRequest.httpRequest.connection)[currentRequest.sessionData.sessionId]
       if isNew
-        util.log("#{currentRequest.sessionData.deviceId}: New device registered")
+        logger.accessInfo({
+          currentRequest: currentRequest
+          message: 'New device registered'
+        })
 
       db.clearTasks(currentRequest.sessionData.deviceId, currentRequest.sessionData.doneTasks, (err) ->
         return throwError(err, currentRequest.httpResponse) if err
@@ -500,7 +520,12 @@ sendRpcRequest = (currentRequest, id, rpcRequest) ->
         return sendRpcRequest(currentRequest, id, rpcRequest)
       )
 
-  util.log("#{currentRequest.sessionData.deviceId}: #{rpcRequest.type} (#{id})")
+  logger.accessInfo({
+    currentRequest: currentRequest
+    message: 'ACS request'
+    acsRequest: rpcRequest
+    acsRequestId: id
+  })
 
   res = soap.response({
     id : id,
@@ -647,30 +672,53 @@ listener = (httpRequest, httpResponse) ->
 
       if cwmpRequest.methodRequest?
         if cwmpRequest.methodRequest.type is 'Inform'
+          logger.accessInfo({
+            currentRequest: currentRequest
+            message: 'Inform'
+            cpeRequest: cwmpRequest.methodRequest
+            cpeRequestId: cwmpRequest.id
+          });
           inform(currentRequest, cwmpRequest)
         else if cwmpRequest.methodRequest.type is 'TransferComplete'
+          logger.accessInfo({
+            currentRequest: currentRequest
+            message: 'CPE request'
+            cpeRequest: cwmpRequest.methodRequest
+            cpeRequestId: cwmpRequest.id
+          });
           transferComplete(currentRequest, cwmpRequest)
         else if cwmpRequest.methodRequest.type is 'GetRPCMethods'
-          util.log("#{currentRequest.sessionData.deviceId}: GetRPCMethods")
+          logger.accessInfo({
+            currentRequest: currentRequest
+            message: 'CPE request'
+            cpeRequest: cwmpRequest.methodRequest
+            cpeRequestId: cwmpRequest.id
+          });
           res = soap.response({
             id : cwmpRequest.id,
             methodResponse : {type : 'GetRPCMethodsResponse', methodList : ['Inform', 'GetRPCMethods', 'TransferComplete', 'RequestDownload']},
             cwmpVersion : currentRequest.sessionData.cwmpVersion
           })
           writeResponse(currentRequest, res)
-        else if cwmpRequest.methodRequest.type is 'TransferComplete'
+        else
-          return throwError(new Error('ACS method not supported'), currentRequest.httpResponse) if err
+          return throwError(new Error('ACS method not supported'), currentRequest.httpResponse)
       else if cwmpRequest.methodResponse?
         session.rpcResponse(currentRequest.sessionData, cwmpRequest.id, cwmpRequest.methodResponse, (err) ->
           return throwError(err, currentRequest.httpResponse) if err
           nextRpc(currentRequest)
         )
       else if cwmpRequest.fault?
+        logger.accessWarn({
+          currentRequest: currentRequest
+          message: 'CPE fault'
+          cpeFault: cwmpRequest.fault.detail
+          cpeRequestId: cwmpRequest.id
+        });
         session.rpcFault(currentRequest.sessionData, cwmpRequest.id, cwmpRequest.fault, (err, fault) ->
           return throwError(err, currentRequest.httpResponse) if err
 
           if fault
-            recordFault(currentRequest.sessionData, fault, currentRequest.sessionData.provisions, currentRequest.sessionData.channels)
+            recordFault(currentRequest, fault, currentRequest.sessionData.provisions, currentRequest.sessionData.channels)
             session.clearProvisions(currentRequest.sessionData)
           nextRpc(currentRequest)
         )
@@ -682,7 +730,7 @@ listener = (httpRequest, httpResponse) ->
             for k, v of operations[i].retries
               sessionData.retries[k] = v
 
-            recordFault(currentRequest.sessionData, fault, operations[i].provisions, operations[i].channels)
+            recordFault(currentRequest, fault, operations[i].provisions, operations[i].channels)
 
           nextRpc(currentRequest)
         )

lib/logger.js

@@ -0,0 +1,315 @@
+/**
+ * Copyright 2013-2017  Zaid Abdulla
+ *
+ * GenieACS is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * GenieACS is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
+ */
+"use strict";
+
+const fs = require("fs");
+const os = require("os");
+
+const config = require("./config");
+
+const REOPEN_EVERY = 60000;
+
+const LOG_FORMAT = config.get("LOG_FORMAT");
+const ACCESS_LOG_FORMAT = config.get("ACCESS_LOG_FORMAT") || LOG_FORMAT;
+
+const defaultMeta = {};
+
+let LOG_SYSTEMD = false;
+let ACCESS_LOG_SYSTEMD = false;
+
+let LOG_FILE, ACCESS_LOG_FILE;
+
+let logStream = fs.createWriteStream(null, {fd: process.stderr.fd});
+let logStat = fs.fstatSync(logStream.fd);
+let accessLogStream = fs.createWriteStream(null, {fd: process.stdout.fd});
+let accessLogStat = fs.fstatSync(accessLogStream.fd);
+
+
+// Reopen if original files have been moved (e.g. logrotate)
+function reopen() {
+  let counter = 1;
+
+  if (LOG_FILE) {
+    ++ counter;
+    fs.stat(LOG_FILE, function(err, stat) {
+      if (err && !err.message.startsWith("ENOENT:"))
+        throw err;
+
+      if (!(stat && stat.dev === logStat.dev && stat.ino === logStat.ino)) {
+        logStream.end();
+        logStream = fs.createWriteStream(null,
+          {fd: fs.openSync(LOG_FILE, "a")});
+        logStat = fs.fstatSync(logStream.fd);
+      }
+
+      if (-- counter === 0)
+        setTimeout(reopen, REOPEN_EVERY - (Date.now() % REOPEN_EVERY)).unref();
+    });
+  }
+
+  if (ACCESS_LOG_FILE) {
+    ++ counter;
+    fs.stat(ACCESS_LOG_FILE, function(err, stat) {
+      if (err && !err.message.startsWith("ENOENT:"))
+        throw err;
+
+      if (!(stat && stat.dev === accessLogStat.dev && stat.ino === accessLogStat.ino)) {
+        accessLogStream.end();
+        accessLogStream = fs.createWriteStream(null,
+          {fd: fs.openSync(ACCESS_LOG_FILE, "a")});
+        accessLogStat = fs.fstatSync(accessLogStream.fd);
+      }
+
+      if (-- counter === 0)
+        setTimeout(reopen, REOPEN_EVERY - (Date.now() % REOPEN_EVERY)).unref();
+    });
+  }
+
+  if (-- counter === 0)
+    setTimeout(reopen, REOPEN_EVERY - (Date.now() % REOPEN_EVERY)).unref();
+}
+
+
+function init(service, version) {
+  defaultMeta.hostname = os.hostname();
+  defaultMeta.pid = process.pid;
+  defaultMeta.name = `genieacs-${service}`;
+  defaultMeta.version = version;
+
+  LOG_FILE = config.get(`${service.toUpperCase()}_LOG_FILE`);
+  ACCESS_LOG_FILE = config.get(`${service.toUpperCase()}_ACCESS_LOG_FILE`);
+
+  if (LOG_FILE) {
+    logStream = fs.createWriteStream(null,
+      {fd: fs.openSync(LOG_FILE, "a")});
+    logStat = fs.fstatSync(logStream.fd);
+  }
+
+  if (ACCESS_LOG_FILE) {
+    accessLogStream = fs.createWriteStream(null,
+      {fd: fs.openSync(ACCESS_LOG_FILE, "a")});
+    accessLogStat = fs.fstatSync(accessLogStream.fd);
+  }
+
+  // Determine if logs are going to journald
+  const JOURNAL_STREAM = process.env["JOURNAL_STREAM"];
+
+  if (JOURNAL_STREAM) {
+    let [dev, inode] = JOURNAL_STREAM.split(":");
+    dev = parseInt(dev);
+    inode = parseInt(inode);
+
+    LOG_SYSTEMD = (logStat.dev === dev && logStat.ino === inode);
+    ACCESS_LOG_SYSTEMD = (accessLogStat.dev === dev && accessLogStat.ino === inode);
+  }
+
+  if (LOG_FILE || ACCESS_LOG_FILE) {
+    // Can't use setInterval as we need all workers to cehck at the same time
+    setTimeout(reopen, REOPEN_EVERY - (Date.now() % REOPEN_EVERY)).unref();
+  }
+}
+
+
+function close() {
+  accessLogStream.end();
+  logStream.end();
+}
+
+
+function flatten(details) {
+  if (details.currentRequest) {
+    details.deviceId = details.currentRequest.sessionData.deviceId;
+    details.remoteAddress = details.currentRequest.httpRequest.connection.remoteAddress;
+    delete details.currentRequest;
+  }
+
+  if (details.exception) {
+    details.exceptionName = details.exception.name;
+    details.exceptionMessage = details.exception.message;
+    details.exceptionStack = details.exception.stack;
+    delete details.exception;
+  }
+
+  if (details.task) {
+    details.taskId = details.task._id;
+    delete details.task;
+  }
+
+  if (details.acsRequest) {
+    details.acsRequestName = details.acsRequest.type;
+    if (details.acsRequest.commandKey)
+      details.acsRequestCommandKey = details.acsRequest.commandKey;
+    delete details.acsRequest;
+  }
+
+  if (details.cpeRequest) {
+    if (details.cpeRequest.type === "Inform") {
+      details.informEvent = details.cpeRequest.event.join(",");
+      details.informRetryCount = details.cpeRequest.retryCount;
+    }
+    else {
+      details.cpeRequestName = details.cpeRequest.type;
+      if (details.cpeRequest.commandKey)
+        details.cpeRequestCommandKey = details.cpeRequest.commandKey;
+    }
+    delete details.cpeRequest;
+  }
+
+  if (details.cpeFault) {
+    details.cpeFaultCode = details.cpeFault.faultCode;
+    details.cpeFaultString = details.cpeFault.faultString;
+    delete details.cpeFault;
+  }
+
+  if (details.fault) {
+    details.faultCode = details.fault.code;
+    details.faultMessage = details.fault.message;
+    delete details.fault;
+  }
+  return details;
+}
+
+
+function formatJson(details, systemd) {
+  if (systemd) {
+    let severity = "";
+    if (details.severity === "info")
+      severity = "<6>";
+    else if (details.severity === "warn")
+      severity = "<4>";
+    else if (details.severity === "error")
+      severity = "<3>";
+
+    return `${severity}${JSON.stringify(flatten(details))}${os.EOL}`;
+  }
+
+  return `${JSON.stringify(flatten(details))}${os.EOL}`;
+}
+
+
+function formatSimple(details, systemd) {
+  flatten(details);
+
+  let remote = "";
+  if (details.remoteAddress) {
+    if (details.deviceId)
+      remote = `${details.remoteAddress} ${details.deviceId}: `;
+    else
+      remote = `${details.remoteAddress}: `;
+  }
+
+  const keys = Object.keys(details);
+  const skip = {
+    deviceId: 1,
+    remoteAddress: 1,
+    severity: 1,
+    timestamp: 1,
+    message: 1
+  };
+
+  let meta = "";
+
+  let kv = [];
+  for (let k of keys)
+    if (!skip[k])
+      kv.push(`${k}=${JSON.stringify(details[k])}`);
+
+  if (kv.length)
+    meta = `; ${kv.join(" ")}`;
+
+  if (systemd) {
+    let severity = "";
+    if (details.severity === "info")
+      severity = "<6>";
+    else if (details.severity === "warn")
+      severity = "<4>";
+    else if (details.severity === "error")
+      severity = "<3>";
+
+    return `${severity}${remote}${details.message}${meta}${os.EOL}`;
+  }
+
+  return `${details.timestamp} [${details.severity.toUpperCase()}] ${remote}${details.message}${meta}${os.EOL}`;
+}
+
+
+function log(details) {
+  details.timestamp = new Date().toISOString();
+  if (LOG_FORMAT === "json") {
+    details = Object.assign({}, defaultMeta, details);
+    logStream.write(formatJson(details, LOG_SYSTEMD));
+  } else {
+    logStream.write(formatSimple(details, LOG_SYSTEMD));
+  }
+}
+
+
+function info(details) {
+  details.severity = "info";
+  log(details);
+}
+
+
+function warn(details) {
+  details.severity = "warn";
+  log(details);
+}
+
+
+function error(details) {
+  details.severity = "error";
+  log(details);
+}
+
+
+function accessLog(details) {
+  details.timestamp = new Date().toISOString();
+  if (ACCESS_LOG_FORMAT === "json") {
+    Object.assign(details, defaultMeta);
+    accessLogStream.write(formatJson(details, ACCESS_LOG_SYSTEMD));
+  } else {
+    accessLogStream.write(formatSimple(details, ACCESS_LOG_SYSTEMD));
+  }
+}
+
+
+function accessInfo(details) {
+  details.severity = "info";
+  accessLog(details);
+}
+
+
+function accessWarn(details) {
+  details.severity = "warn";
+  accessLog(details);
+}
+
+
+function accessError(details) {
+  details.severity = "error";
+  accessLog(details);
+}
+
+
+exports.init = init;
+exports.close = close;
+exports.info = info;
+exports.warn = warn;
+exports.error = error;
+exports.accessInfo = accessInfo;
+exports.accessWarn = accessWarn;
+exports.accessError = accessError;

lib/server.coffee

@@ -15,17 +15,18 @@
 # along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
 ###
 
+service = process.argv[2]
+if not service?
+  throw new Error('Missing argument cwmp, fs, or nbi')
+
 util = require 'util'
 cluster = require 'cluster'
+logger = require './logger'
+logger.init(service, require('../package.json').version)
 config = require './config'
 db = require './db'
 extensions = require './extensions'
 
-
-service = config.argv[0]
-if not service?
-  throw new Error('Missing argument cwmp, fs, or nbi')
-
 networkInterface = config.get("#{service.toUpperCase()}_INTERFACE")
 port = config.get("#{service.toUpperCase()}_PORT")
 useHttps = config.get("#{service.toUpperCase()}_SSL")
@@ -45,21 +46,22 @@ exit = () ->
   if not server
     db.disconnect()
     extensions.killAll()
+    logger.close()
     return
 
   server.close(() ->
     db.disconnect()
     extensions.killAll()
+    logger.close()
   )
 
 
 process.on('uncaughtException', (err) ->
-  try
+  logger.error({
-    util.error("#{new Date().toISOString()} - #{err.stack}\n")
+    message: 'Uncaught exception'
+    exception: err
+  })
   exit()
-  catch err
-    util.error("#{new Date().toISOString()} - #{err.stack}\n")
-  throw err
 )
 
 

lib/session.coffee

@@ -164,7 +164,6 @@ transferComplete = (sessionData, rpcReq, callback) ->
   operation = sessionData.operations[commandKey]
 
   if not operation?
-    # TODO Show a warning
     return callback(null, {type : 'TransferCompleteResponse'})
 
   instance = operation.args.instance
@@ -182,7 +181,7 @@ transferComplete = (sessionData, rpcReq, callback) ->
         timestamp: operation.timestamp
       }
 
-      return callback(err, {type : 'TransferCompleteResponse'}, fault, operation)
+      return callback(err, {type : 'TransferCompleteResponse'}, operation, fault)
     )
 
   loadPath(sessionData, ['Downloads', instance, '*'])
@@ -218,7 +217,7 @@ transferComplete = (sessionData, rpcReq, callback) ->
       {value: [timestamp, [+rpcReq.completeTime, 'xsd:dateTime']]}, toClear)
 
     clear(sessionData, toClear, (err) ->
-      return callback(err, {type : 'TransferCompleteResponse'})
+      return callback(err, {type : 'TransferCompleteResponse'}, operation)
     )
   )