Compare commits

...

11 Commits

3 changed files with 90 additions and 9 deletions

46
src/helper/userHelper.js Normal file
View File

@ -0,0 +1,46 @@
// This file specifies functions to help a user with e.g. configuration-errors
function detectStreamData(stream, timeout = 5000){
return new Promise((resolve, reject) => {
let timeoutHandler;
if(timeout){
timeoutHandler = setTimeout(() => {
reject('timeout');
remListeners();
},
timeout);
}
function remListeners(){
stream.removeListener('error', errorHandler);
stream.removeListener('data', dataHandler);
if(timeoutHandler) clearTimeout(timeoutHandler);
}
function errorHandler(err) {
remListeners();
}
function dataHandler(data) {
resolve(data);
remListeners();
}
stream.on('error', errorHandler);
stream.on('data', dataHandler);
});
}
function detectStreamsData(streams, timeout = 5000){
let promises = [];
streams.forEach((stream) => {
promises.push(detectStreamData(stream, timeout));
})
return promises;
}
// Specify exports
module.exports = {
detectStreamData,
detectStreamsData,
};

View File

@ -1,5 +1,6 @@
"use strict"; "use strict";
const logger = require("./helper/logger.js")("main"); const logFactory = require("./helper/logger.js");
const logger = logFactory("main");
const { requireEnvVars } = require("./helper/env.js"); const { requireEnvVars } = require("./helper/env.js");
const { exit } = require("process"); const { exit } = require("process");
@ -13,6 +14,9 @@ const { PacketStreamFactory } = require("./streamHandler/PacketStreamFactory.js"
const { PacketInfluxPointFactory } = require("./streamHandler/PacketInfluxPointFactory.js"); const { PacketInfluxPointFactory } = require("./streamHandler/PacketInfluxPointFactory.js");
const { InfluxPointWriter } = require("./streamHandler/InfluxPointWriter.js"); const { InfluxPointWriter } = require("./streamHandler/InfluxPointWriter.js");
const userHelper = require("./helper/userHelper.js");
/// Setup ENVs /// Setup ENVs
const env = process.env; const env = process.env;
// Defaults // Defaults
@ -60,23 +64,53 @@ if(errorMsg){
let proc = exec(cmd); let proc = exec(cmd);
logger.debug("Creating & Attaching streams.."); logger.debug("Creating & Attaching streams..");
let regexBlockStream = new RegexBlockStream(/^\d{2}:\d{2}:\d{2}.\d{6}.*(\n( {4,8}|\t\t?).*)+\n/gm);
let packetStreamFactory = new PacketStreamFactory();
let packetInfluxPointFactory = new PacketInfluxPointFactory();
let influxPointWriter = new InfluxPointWriter(influxDb, env.INFLUX_ORG, env.INFLUX_BUCKET);
proc.stdout proc.stdout
.setEncoding("utf8") .setEncoding("utf8")
.pipe(new RegexBlockStream(/^\d{2}:\d{2}:\d{2}.\d{6}.*(\n( {4,8}|\t\t?).*)+\n/gm)) .pipe(regexBlockStream)
.pipe(new PacketStreamFactory()) .pipe(packetStreamFactory)
.pipe(new PacketInfluxPointFactory()) .pipe(packetInfluxPointFactory)
.pipe(new InfluxPointWriter(influxDb, env.INFLUX_ORG, env.INFLUX_BUCKET)); .pipe(influxPointWriter);
logger.debug("Attaching error-logger.."); logger.debug("Attaching error-logger..");
const loggerTcpdump = logFactory("tcpdump");
proc.stderr.setEncoding("utf8").on("data", (data) => { proc.stderr.setEncoding("utf8").on("data", (data) => {
logger.error(data); if(!data.match(/^(tcpdump: )?listening on /i)) // Catch start-error
loggerTcpdump.error(data);
});
regexBlockStream.on('error', (err) => {
if(err) loggerTcpdump.error(err);
})
proc.on("error", (err) => {
loggerTcpdump.error(err);
});
const loggerPacketStream = logFactory("PacketStreamFactory");
userHelper.detectStreamData(proc.stdout, 10000) // Expect tcpdump-logs to have data after max. 10s
.then(() => {
loggerTcpdump.debug("Got first data");
userHelper.detectStreamData(packetStreamFactory, 10000) // Expect then to have packets after further 10s
.then(() => {
loggerPacketStream.debug("Got first packet");
})
.catch((err) => {
if(err == 'timeout') loggerPacketStream.warn("No packets");
});
})
.catch((err) => {
if(err == 'timeout') loggerTcpdump.warn("No data after 10s! Wrong configuration?");
}); });
logger.debug("Attaching exit-handler.."); logger.debug("Attaching exit-handler..");
proc.on("exit", (code) => { proc.on("exit", (code) => {
logger.info(`tcpdump exited code: ${code}`); loggerTcpdump.debug(`tcpdump exited code: ${code}`);
if (code) { if (code) {
logger.fatal(`tcpdump exited with non-zero code: ${code}`); loggerTcpdump.fatal(`tcpdump exited with non-zero code: ${code}`);
exit(1); exit(1);
} }
logger.info("Shutdown"); logger.info("Shutdown");

View File

@ -9,6 +9,7 @@ const TAG_LIST = [
"bssid", "bssid",
"frequency", "frequency",
"flags", "flags",
"packetType",
]; ];
/** Measurement-name and corresponding field-key */ /** Measurement-name and corresponding field-key */