Add debug-dump-ws command

Add debug command to dump the WebSocket messages being sent and received
by the master server to aid in finding chatty events.

clusterctl.js

@@ -310,6 +310,14 @@ commands.push(new Command({
 	},
 }));
 
+commands.push(new Command({
+	definition: ["debug-dump-ws", "Dump WebSocket messages sent and received by master", (yargs) => { }],
+	handler: async function(args, control) {
+		await link.messages.debugDumpWs.send(control);
+		return new Promise(() => {});
+	},
+}));
+
 // Convert to mapping from name to command instance
 commands = new Map([...commands.map(command => [command.name, command])]);
 
@@ -352,6 +360,10 @@ class Control extends link.Link {
 		console.log(formatOutputColored(output));
 	}
 
+	async debugWsMessageEventHandler(message) {
+		console.log("WS", message.data.direction, message.data.content);
+	}
+
 	async shutdown() {
 		this.connector.setTimeout(30);
 

lib/link/messages.js

@@ -195,6 +195,12 @@ messages.ping = new Request({
 	links: wsLinks,
 });
 
+messages.debugDumpWs = new Request({
+	type: "debug_dump_ws",
+	links: ["control-master"],
+});
+
+
 // Management requests
 messages.listSlaves = new Request({
 	type: 'list_slaves',
@@ -484,6 +490,15 @@ class Event {
 	}
 }
 
+messages.debugWsMessage = new Event({
+	type: "debug_ws_message",
+	links: ["master-control"],
+	eventProperties: {
+		"direction": { type: "string" },
+		"content": { type: "string" },
+	},
+});
+
 messages.instanceStopped = new Event({
 	type: 'instance_stopped',
 	links: ['instance-slave'],

master.js

@@ -30,6 +30,7 @@ const version = require("./package").version;
 let masterConfig;
 let masterConfigPath;
 let stopAcceptingNewSessions = false;
+let debugEvents = new events.EventEmitter();
 
 // homebrew modules
 const generateSSLcert = require("lib/generateSSLcert");
@@ -429,6 +430,16 @@ class ControlConnection extends BaseConnection {
 		});
 
 		this.instanceOutputSubscriptions = new Set();
+
+		this.ws_dumper = null;
+		this.connector.on("connected", () => {
+			this.connector._socket.clusterio_ignore_dump = !!this.ws_dumper;
+		});
+		this.connector.on("close", () => {
+			if (this.ws_dumper) {
+				debugEvents.off("message", this.ws_dumper);
+			}
+		});
 	}
 
 	async listSlavesRequestHandler(message) {
@@ -551,6 +562,14 @@ class ControlConnection extends BaseConnection {
 	async setInstanceOutputSubscriptionsRequestHandler(message) {
 		this.instanceOutputSubscriptions = new Set(message.data.instance_ids);
 	}
+
+	async debugDumpWsRequestHandler(message) {
+		this.ws_dumper = data => {
+			link.messages.debugWsMessage.send(this, data);
+		};
+		this.connector._socket.clusterio_ignore_dump = true;
+		debugEvents.on("message", this.ws_dumper)
+	}
 }
 
 var slaveConnections = new Map();
@@ -828,10 +847,18 @@ wss.on("connection", function (socket, req) {
 
 	// Track statistics
 	wsConnectionsCounter.inc();
-	socket.on("message", () => { wsMessageCounter.labels("in").inc(); });
+	socket.on("message", (message) => {
+		wsMessageCounter.labels("in").inc();
+		if (!socket.clusterio_ignore_dump) {
+			debugEvents.emit("message", { direction: "in", content: message });
+		}
+	});
 	let originalSend = socket.send;
 	socket.send = (...args) => {
 		wsMessageCounter.labels("out").inc();
+		if (typeof args[0] === "string" && !socket.clusterio_ignore_dump) {
+			debugEvents.emit("message", { direction: "out", content: args[0] });
+		}
 		return originalSend.call(socket, ...args);
 	};
 

test/integration/link.js

@@ -12,6 +12,7 @@ class TestControl extends link.Link {
 		link.attachAllMessages(this);
 	}
 
+	async debugWsMessageEventHandler() { }
 	async prepareDisconnectRequestHandler() { }
 	async instanceOutputEventHandler() { }
 }