Metrics

package.json

@@ -47,6 +47,7 @@
     "node-html-parser": "^1.1.11",
     "p-queue": "^5.0.0",
     "pg-promise": "^8.5.1",
+    "prom-client": "^11.3.0",
     "tslint": "^5.11.0",
     "typescript": "^3.1.3",
     "winston": "^3.0.0",

src/bot.ts

@@ -37,6 +37,7 @@ import * as Discord from "discord.js";
 import * as mime from "mime";
 import { IMatrixEvent, IMatrixMediaInfo } from "./matrixtypes";
 import { DiscordCommandHandler } from "./discordcommandhandler";
+import { MetricPeg } from "./metrics";
 
 const log = new Log("DiscordBot");
 
@@ -288,12 +289,14 @@ export class DiscordBot {
         });
         client.on("message", async (msg: Discord.Message) => {
             try {
+                MetricPeg.get.registerRequest(msg.id);
                 await this.waitUnlock(msg.channel);
                 this.discordMessageQueue[msg.channel.id] = (async () => {
                     await (this.discordMessageQueue[msg.channel.id] || Promise.resolve());
                     try {
                         await this.OnMessage(msg);
                     } catch (err) {
+                        MetricPeg.get.requestOutcome(msg.id, true, "fail");
                         log.error("Caught while handing 'message'", err);
                     }
                 })();
@@ -389,6 +392,7 @@ export class DiscordBot {
             }
             const channel = guild.channels.get(room);
             if (channel && channel.type === "text") {
+                this.ClientFactory.bindMetricsToChannel(channel as Discord.TextChannel);
                 const lookupResult = new ChannelLookupResult();
                 lookupResult.channel = channel as Discord.TextChannel;
                 lookupResult.botUser = this.bot.user.id === client.user.id;
@@ -448,6 +452,7 @@ export class DiscordBot {
                 // NOTE: Don't send replies to discord if we are a puppet.
                 msg = await chan.send(embed.description, opts);
             } else if (hook) {
+                MetricPeg.get.remoteCall("hook.send");
                 msg = await hook.send(embed.description, {
                     avatarURL: embed!.author!.icon_url,
                     embeds: embedSet.replyEmbed ? [embedSet.replyEmbed] : undefined,
@@ -539,6 +544,7 @@ export class DiscordBot {
         if (guild) {
             const channel = client.channels.get(entry.remote!.get("discord_channel") as string);
             if (channel) {
+                this.ClientFactory.bindMetricsToChannel(channel as Discord.TextChannel);
                 return channel;
             }
             throw Error("Channel given in room entry not found");
@@ -738,11 +744,13 @@ export class DiscordBot {
         if (indexOfMsg !== -1) {
             log.verbose("Got repeated message, ignoring.");
             delete this.sentMessages[indexOfMsg];
+            MetricPeg.get.requestOutcome(msg.id, true, "dropped");
             return; // Skip *our* messages
         }
         const chan = msg.channel as Discord.TextChannel;
         if (msg.author.id === this.bot.user.id) {
             // We don't support double bridging.
+            MetricPeg.get.requestOutcome(msg.id, true, "dropped");
             return;
         }
         // Test for webhooks
@@ -751,13 +759,15 @@ export class DiscordBot {
                             .filterArray((h) => h.name === "_matrix").pop();
             if (webhook && msg.webhookID === webhook.id) {
               // Filter out our own webhook messages.
+                MetricPeg.get.requestOutcome(msg.id, true, "dropped");
                 return;
             }
         }
 
         // check if it is a command to process by the bot itself
         if (msg.content.startsWith("!matrix")) {
             await this.discordCommandHandler.Process(msg);
+            MetricPeg.get.requestOutcome(msg.id, true, "success");
             return;
         }
 
@@ -766,14 +776,13 @@ export class DiscordBot {
         let rooms;
         try {
             rooms = await this.channelSync.GetRoomIdsFromChannel(msg.channel);
+            if (rooms === null) { throw Error(); }
         } catch (err) {
             log.verbose("No bridged rooms to send message to. Oh well.");
+            MetricPeg.get.requestOutcome(msg.id, true, "dropped");
             return null;
         }
         try {
-            if (rooms === null) {
-              return null;
-            }
             const intent = this.GetIntentFromDiscordMember(msg.author, msg.webhookID);
             // Check Attachements
             await Util.AsyncForEach(msg.attachments.array(), async (attachment) => {
@@ -854,7 +863,9 @@ export class DiscordBot {
                     await afterSend(res);
                 }
             });
+            MetricPeg.get.requestOutcome(msg.id, true, "success");
         } catch (err) {
+            MetricPeg.get.requestOutcome(msg.id, true, "fail");
             log.verbose("Failed to send message into room.", err);
         }
     }

src/clientfactory.ts

@@ -16,14 +16,12 @@ limitations under the License.
 
 import { DiscordBridgeConfigAuth } from "./config";
 import { DiscordStore } from "./store";
-import { Client as DiscordClient } from "discord.js";
+import { Client as DiscordClient, TextChannel } from "discord.js";
 import { Log } from "./log";
-import { Util } from "./util";
+import { MetricPeg } from "./metrics";
 
 const log = new Log("ClientFactory");
 
-const READY_TIMEOUT = 30000;
-
 export class DiscordClientFactory {
     private config: DiscordBridgeConfigAuth;
     private store: DiscordStore;
@@ -110,4 +108,12 @@ export class DiscordClientFactory {
             return this.botClient;
         }
     }
+
+    public bindMetricsToChannel(channel: TextChannel) {
+        // tslint:disable-next-line:only-arrow-functions
+        channel.send = function() {
+            MetricPeg.get.remoteCall("channel.send");
+            return channel.send.apply(channel, arguments);
+        };
+    }
 }

src/config.ts

@@ -56,6 +56,7 @@ class DiscordBridgeConfigBridge {
     public disableEveryoneMention: boolean = false;
     public disableHereMention: boolean = false;
     public disableJoinLeaveNotifications: boolean = false;
+    public enableMetrics: boolean = false;
 }
 
 export class DiscordBridgeConfigDatabase {

src/db/roomstore.ts

@@ -19,6 +19,7 @@ import { Util } from "../util";
 
 import * as uuid from "uuid/v4";
 import { Postgres } from "./postgres";
+import { MetricPeg } from "../metrics";
 
 const log = new Log("DbRoomStore");
 
@@ -155,8 +156,10 @@ export class DbRoomStore {
     public async getEntriesByMatrixId(matrixId: string): Promise<IRoomStoreEntry[]> {
         const cached = this.entriesMatrixIdCache.get(matrixId);
         if (cached && cached.ts + ENTRY_CACHE_LIMETIME > Date.now()) {
+            MetricPeg.get.storeCall("RoomStore.getEntriesByMatrixId", true);
             return cached.e;
         }
+        MetricPeg.get.storeCall("RoomStore.getEntriesByMatrixId", false);
         const entries = await this.db.All(
             "SELECT * FROM room_entries WHERE matrix_id = $id", {id: matrixId},
         );
@@ -190,6 +193,7 @@ export class DbRoomStore {
     }
 
     public async getEntriesByMatrixIds(matrixIds: string[]): Promise<IRoomStoreEntry[]> {
+        MetricPeg.get.storeCall("RoomStore.getEntriesByMatrixIds", false);
         const mxIdMap = { };
         matrixIds.forEach((mxId, i) => mxIdMap[i] = mxId);
         const sql = `SELECT * FROM room_entries WHERE matrix_id IN (${matrixIds.map((_, id) => `\$${id}`).join(", ")})`;
@@ -222,6 +226,7 @@ export class DbRoomStore {
     }
 
     public async linkRooms(matrixRoom: MatrixStoreRoom, remoteRoom: RemoteStoreRoom) {
+        MetricPeg.get.storeCall("RoomStore.linkRooms", false);
         await this.upsertRoom(remoteRoom);
 
         const values = {
@@ -244,6 +249,7 @@ export class DbRoomStore {
     }
 
     public async getEntriesByRemoteRoomData(data: IRemoteRoomDataLazy): Promise<IRoomStoreEntry[]> {
+        MetricPeg.get.storeCall("RoomStore.getEntriesByRemoteRoomData", false);
         Object.keys(data).filter((k) => typeof(data[k]) === "boolean").forEach((k) => {
             data[k] = Number(data[k]);
         });
@@ -270,11 +276,13 @@ export class DbRoomStore {
     }
 
     public async removeEntriesByRemoteRoomId(remoteId: string) {
+        MetricPeg.get.storeCall("RoomStore.removeEntriesByRemoteRoomId", false);
         await this.db.Run(`DELETE FROM room_entries WHERE remote_id = $remoteId`, {remoteId});
         await this.db.Run(`DELETE FROM remote_room_data WHERE room_id = $remoteId`, {remoteId});
     }
 
     public async removeEntriesByMatrixRoomId(matrixId: string) {
+        MetricPeg.get.storeCall("RoomStore.removeEntriesByMatrixRoomId", false);
         const entries = (await this.db.All(`SELECT * FROM room_entries WHERE matrix_id = $matrixId`, {matrixId})) || [];
         await Util.AsyncForEach(entries, async (entry) => {
             if (entry.remote_id) {
@@ -286,6 +294,7 @@ export class DbRoomStore {
     }
 
     private async upsertRoom(room: RemoteStoreRoom) {
+        MetricPeg.get.storeCall("RoomStore.upsertRoom", false);
         if (!room.data) {
             throw new Error("Tried to upsert a room with undefined data");
         }

src/db/userstore.ts

@@ -15,8 +15,8 @@ limitations under the License.
 */
 
 import { IDatabaseConnector } from "./connector";
-import * as uuid from "uuid/v4";
 import { Log } from "../log";
+import { MetricPeg } from "../metrics";
 
 /**
  * A UserStore compatible with
@@ -54,8 +54,11 @@ export class DbUserStore {
     public async getRemoteUser(remoteId: string): Promise<RemoteUser|null> {
         const cached = this.remoteUserCache.get(remoteId);
         if (cached && cached.ts + ENTRY_CACHE_LIMETIME > Date.now()) {
+            MetricPeg.get.storeCall("UserStore.getRemoteUser", true);
             return cached.e;
         }
+        MetricPeg.get.storeCall("UserStore.getRemoteUser", false);
+
         const row = await this.db.Get(
             "SELECT * FROM user_entries WHERE remote_id = $id", {id: remoteId},
         );
@@ -86,6 +89,7 @@ export class DbUserStore {
     }
 
     public async setRemoteUser(user: RemoteUser) {
+        MetricPeg.get.storeCall("UserStore.setRemoteUser", false);
         this.remoteUserCache.delete(user.id);
         const existingData = await this.db.Get(
             "SELECT * FROM remote_user_data WHERE remote_id = $remoteId",
@@ -156,6 +160,7 @@ AND guild_id = $guild_id`,
     }
 
     public async linkUsers(matrixId: string, remoteId: string) {
+        MetricPeg.get.storeCall("UserStore.linkUsers", false);
         // This is used  ONCE in the bridge to link two IDs, so do not UPSURT data.
         try {
             await this.db.Run(`INSERT INTO user_entries VALUES ($matrixId, $remoteId)`, {

src/discordas.ts

@@ -22,6 +22,7 @@ import { DiscordBot } from "./bot";
 import { DiscordStore } from "./store";
 import { Log } from "./log";
 import "source-map-support/register";
+import { MetricPeg, PrometheusBridgeMetrics } from "./metrics";
 
 const log = new Log("DiscordAS");
 
@@ -93,13 +94,16 @@ async function run(port: number, fileConfig: DiscordBridgeConfig) {
                 } catch (err) { log.error("Exception thrown while handling \"onAliasQuery\" event", err); }
             },
             onEvent: async (request) => {
+                const data = request.getData();
                 try {
+                    MetricPeg.get.registerRequest(data.event_id);
                     // Build our own context.
                     if (!store.roomStore) {
                         log.warn("Discord store not ready yet, dropping message");
+                        MetricPeg.get.requestOutcome(data.event_id, false, "dropped");
                         return;
                     }
-                    const roomId = request.getData().room_id;
+                    const roomId = data.room_id;
 
                     const context: BridgeContext = {
                         rooms: {},
@@ -111,7 +115,9 @@ async function run(port: number, fileConfig: DiscordBridgeConfig) {
                     }
 
                     await request.outcomeFrom(callbacks.onEvent(request, context));
+                    MetricPeg.get.requestOutcome(data.event_id, false, "success");
                 } catch (err) {
+                    MetricPeg.get.requestOutcome(data.event_id, false, "fail");
                     log.error("Exception thrown while handling \"onEvent\" event", err);
                     await request.outcomeFrom(Promise.reject("Failed to handle"));
                 }
@@ -159,6 +165,11 @@ async function run(port: number, fileConfig: DiscordBridgeConfig) {
     await bridge.run(port, config);
     log.info(`Started listening on port ${port}`);
 
+    if (config.bridge.enableMetrics) {
+        log.info("Enabled metrics");
+        MetricPeg.set(new PrometheusBridgeMetrics().init(bridge));
+    }
+
     try {
         await store.init(undefined, bridge.getRoomStore(), bridge.getUserStore());
     } catch (ex) {