initial commit

Author: vbhayden

.gitignore

@@ -0,0 +1,2 @@
+/node_modules/
+config.js

Dockerfile

@@ -0,0 +1,14 @@
+FROM node:8
+
+# Create our NodeJS source directory
+RUN mkdir -p /usr/src/app
+WORKDIR /usr/src/app
+
+# Copy the project in
+COPY ./ /usr/src/app/
+
+# Install the packages
+RUN npm install --save
+
+# Start the service
+CMD ["npm", "start"]

Dockerfile.test

@@ -0,0 +1,14 @@
+FROM node:8
+
+# Create our NodeJS source directory
+RUN mkdir -p /usr/src/app
+WORKDIR /usr/src/app
+
+# Copy the project in
+COPY ./ /usr/src/app/
+
+# Install the packages
+RUN npm install --save
+
+# Start the service
+CMD ["npm", "run", "test"]

app.js

@@ -0,0 +1,145 @@
+// Entry Point for the LRS ReRoute Service.
+//
+// This is a NodeJS Express application 
+//
+const http = require("http")
+const WebSocket = require('ws');
+const express = require("express");
+const kafkaConsumer = require("simple-kafka-consumer");
+
+const config = require("./config");
+
+const APP_PORT = (process.env.APP_PORT || 3000);
+const WS_PASSWORD = (process.env.WS_PASSWORD || config.websocket.password);
+
+const app = express();
+const server = http.createServer(app);
+
+
+/**
+ * The point of this page is to montior the status of our Kafka cluster.
+ * 
+ * To that end, our monitoring page will open a socket with this service
+ * to relay messages to our page.
+ */
+const openSockets = []
+const wss = new WebSocket.Server({
+    path: config.root + "/kafka",
+    server,
+});
+
+wss.on('connection', function (ws) {
+
+    // Assign to our sockets
+    openSockets.push(ws);
+    ws.authenticated = false;
+    ws.noMessages = true;
+
+    ws.on("message", function (data) {
+        if (ws.noMessages && data == WS_PASSWORD) {
+            ws.authenticated = true;
+            console.log("AUTHENTICATED SOCKET")
+        } else if (data != "keep-alive") {
+            ws.close();
+            console.log("CLOSING: ", data)
+        }
+        ws.noMessages = false;
+    });
+
+    // If the socket is closed, stop sending messages to it
+    ws.on("close", function close() {
+        let index = openSockets.indexOf(ws);
+        openSockets.splice(index, 1);
+    });
+
+    // Send an message when they connect here
+    ws.send('Connected to Kafka Web Socket Monitor!');
+});
+
+setInterval(() => {
+    for(let socket of openSockets) {
+        if (socket.authenticated)
+            socket.send("keep-alive");
+    }
+}, 5000)
+
+// Broadcast a message to each open socket
+//
+function broadcast(message) {
+    for (let k = 0; k < openSockets.length; k++) {
+        if (openSockets[k].authenticated) {
+            openSockets[k].send(message);
+        }
+    }
+}
+
+var recentCount = 0;
+var throttleCount = 100;
+var throttleTimerMS = 500;
+var throttleWarned = false;
+
+// Configure this with our environment and config values
+kafkaConsumer.configure({
+    brokers: (process.env.KAFKA_BROKER || config.kafka.brokers.join(",")),
+    saslUser: (process.env.KAFKA_SASL_USER || config.kafka.sasl ? config.kafka.sasl.username : undefined),
+    saslPass: (process.env.KAFKA_SASL_PASS || config.kafka.sasl ? config.kafka.sasl.password : undefined),
+    topics: config.kafka.topics.map(topic => topic.name),
+    consumerGroup: config.kafka.consumerGroup
+})
+
+// Set up our consumer to broadcast its traffic to our web sockets
+kafkaConsumer.initConsumer((topic, offset, message) => {
+
+    console.log(topic, offset, message.length > 100 ? message.substr(0, 100) + " ..." : message)
+    recentCount++;
+
+    if (recentCount >= throttleCount) {
+        if (throttleWarned == false) {
+            broadcast(`[${message.topic}, # ${offset}, (throttled ${recentCount})]\n"High message rates will be throttled for performance with this page."`)
+        }
+
+        throttleWarned = true;
+        return;
+    }
+
+    broadcast(`[${topic}, # ${offset}]\n${message}\n`)
+})
+
+// Limit how many we can receive on a duration
+setInterval(function () {
+    recentCount = 0;
+    throttleWarned = false;
+}, throttleTimerMS);
+
+/**
+ * Lastly, configure that express instance to serve this page.
+ */
+app.set("view engine", "ejs");
+app.use(express.static("public"));
+app.use(express.static("scripts"));
+app.use(express.static("views"));
+app.use(config.root, express.static("public"));
+app.use(config.root, express.static("scripts"));
+app.use(config.root, express.static("views"));
+
+app.use("*", function (req, res, next) {
+
+    if (req.baseUrl.startsWith(config.root) == false)
+        res.redirect(config.root.substr(1) + req.url);
+    else
+        next();
+});
+
+// Main page.
+app.get(config.root, function (req, res, next) {
+    res.render("index.ejs", {
+        password: WS_PASSWORD,
+        root: config.root,
+        topics: config.kafka.topics
+    });
+});
+
+// Then start the server.
+server.listen(APP_PORT, function () {
+    console.log("\nKafka Web Socket Example listening on port %s", APP_PORT);
+});

config.example.js

@@ -0,0 +1,50 @@
+module.exports = {
+    
+    /**
+     * Root path of the service, so localhost:3000/this-value
+     */
+    root: "/monitor",
+
+    /**
+     * Root path of the service, so localhost:3000/this-value
+     */
+    websocket: {
+        password: "ws-password"
+    },
+
+    /** Kafka cluster settings */
+    kafka: {
+        /** Broker array for the cluster */
+        brokers: [
+            "localhost:9092"
+        ],
+
+        /** Topics we want to monitor */
+        topics: [
+            {
+                "name": "learner-xapi",
+                "pretty": "Learner xAPI",
+                "bootstrap": "primary"
+            },
+            {
+                "name": "transactional-xapi",
+                "pretty": "Transactional xAPI",
+                "bootstrap": "success"
+            },
+            {
+                "name": "authoritative-xapi",
+                "pretty": "Authoritative xAPI",
+                "bootstrap": "danger"
+            }
+        ],
+        /** SASL auth for the Kafka cluster.  Delete if not used. */
+        sasl: {
+            mechanism: 'plain', 
+            username: 'kafka-user', 
+            password: 'kafka-password' 
+        },
+
+        /** Consumer group to use for our monitor. */
+        consumerGroup: "xapi-kafka-monitor",
+    },
+}