add api dialogs iter duration histogram metric

package.json

@@ -1,7 +1,7 @@
 {
     "name": "mtproto_exporter",
     "type": "module",
-    "version": "1.2.0",
+    "version": "1.3.0",
     "packageManager": "pnpm@10.6.5",
     "license": "MIT",
     "scripts": {

src/config.ts

@@ -1,5 +1,5 @@
 import type { OptionDefinition } from "command-line-args";
-import type { RawKeywordLike } from "./keywords.js";
+import type { RawKeywordLike } from "./metrics/keywords.js";
 import { readFile } from "node:fs/promises";
 import cmdline from "command-line-args";
 import yaml from "js-yaml";

src/main.ts

@@ -5,8 +5,8 @@ import { collectDefaultMetrics, Registry } from "prom-client";
 
 import { config, readKeywords } from "./config.js";
 import * as env from "./env.js";
-import { rawToPatterns } from "./keywords.js";
-import * as metrics from "./metrics.js";
+import { rawToPatterns } from "./metrics/keywords.js";
+import * as metrics from "./metrics/metrics.js";
 import MetricsServer from "./server.js";
 
 const registry = new Registry();
@@ -32,9 +32,8 @@ const user = await tg.start({
 
 console.log("Logged in as", user.username);
 
-registry.registerMetric(metrics.newStaticPeerInfoGauge(tg));
-registry.registerMetric(metrics.newUnreadCountGauge(tg));
-registry.registerMetric(metrics.newMessagesCounter(dp));
+metrics.collectDialogMetrics(tg, registry);
+metrics.collectNewMessageMetrics(dp, registry);
 
 if (config.keywords) {
     const counter = new metrics.KeywordsCounter(dp, rawToPatterns(config.keywords));

src/metrics.ts

@@ -1,95 +0,0 @@
-import type { Dispatcher } from "@mtcute/dispatcher";
-import type { TelegramClient } from "@mtcute/node";
-import { PropagationAction } from "@mtcute/dispatcher";
-import { Counter, Gauge } from "prom-client";
-
-import { config } from "./config.js";
-import { peersConfigBoolFilter, peersConfigFilter } from "./filters.js";
-import { KeywordsCounter } from "./keywords.js";
-
-function newMessagesCounter(dp: Dispatcher) {
-    const counter = new Counter({
-        name: "messenger_dialog_messages_count",
-        help: "Messages count since exporter startup",
-        labelNames: ["peerId"],
-    });
-
-    dp.onNewMessage(peersConfigFilter(config), async (msg) => {
-        counter.inc({
-            peerId: msg.chat.id,
-        });
-        return PropagationAction.Continue;
-    });
-    return counter;
-}
-
-function newStaticPeerInfoGauge(tg: TelegramClient) {
-    const gauge = new Gauge({
-        name: "messenger_dialog_info",
-        help: "Dialog information exposed as labels",
-        labelNames: ["peerId", "peerType", "displayName"],
-        collect: async () => {
-            gauge.reset();
-            for await (const d of tg.iterDialogs()) {
-                if (!peersConfigBoolFilter(config, d.peer.id)) {
-                    continue;
-                }
-
-                gauge.set({
-                    peerId: d.peer.id,
-                    peerType: d.peer.type,
-                    displayName: d.peer.displayName,
-                }, 1);
-            }
-        },
-    });
-    return gauge;
-}
-
-function newUnreadCountGauge(tg: TelegramClient) {
-    const gauge = new Gauge({
-        name: "messenger_dialog_unread_messages_count",
-        help: "Number of unread messages in dialogs",
-        labelNames: ["peerId"],
-        collect: async () => {
-            gauge.reset();
-            for await (const d of tg.iterDialogs()) {
-                if (!peersConfigBoolFilter(config, d.peer.id)) {
-                    continue;
-                }
-
-                gauge.set({
-                    peerId: d.peer.id,
-                }, d.unreadCount);
-            }
-        },
-    });
-    return gauge;
-}
-
-function newWordsCounter(dp: Dispatcher) {
-    const counter = new Counter({
-        name: "messenger_dialog_words_count",
-        help: "Number of words in messages since exporter startup",
-        labelNames: ["peerId", "word"],
-    });
-    dp.onNewMessage(peersConfigFilter(config), async (msg) => {
-        const words = msg.text.toLowerCase().split(" ");
-        for (const w of words) {
-            counter.inc({
-                peerId: msg.chat.id,
-                word: w,
-            });
-        }
-        return PropagationAction.Continue;
-    });
-    return counter;
-}
-
-export {
-    KeywordsCounter,
-    newMessagesCounter,
-    newStaticPeerInfoGauge,
-    newUnreadCountGauge,
-    newWordsCounter,
-};

src/metrics/dialogs.ts

@@ -0,0 +1,94 @@
+import type { Dialog, TelegramClient } from "@mtcute/node";
+
+import type { Registry } from "prom-client";
+import process from "node:process";
+import timers from "node:timers/promises";
+
+import { Gauge, Histogram } from "prom-client";
+import { config } from "../config.js";
+import { peersConfigBoolFilter } from "../filters.js";
+
+export function collectDialogMetrics(tg: TelegramClient, registry: Registry) {
+    const dialogs = new DialogsHolder(tg, 1000, 5000);
+
+    const info = new Gauge({
+        name: "messenger_dialog_info",
+        help: "Dialog information exposed as labels",
+        labelNames: ["peerId", "peerType", "displayName"],
+        collect: async () => {
+            info.reset();
+            for (const d of await dialogs.get()) {
+                info.set({
+                    peerId: d.peer.id,
+                    peerType: d.peer.type,
+                    displayName: d.peer.displayName,
+                }, 1);
+            }
+        },
+    });
+
+    const unread = new Gauge({
+        name: "messenger_dialog_unread_messages_count",
+        help: "Number of unread messages in dialogs",
+        labelNames: ["peerId"],
+        collect: async () => {
+            unread.reset();
+            for (const d of await dialogs.get()) {
+                unread.set({
+                    peerId: d.peer.id,
+                }, d.unreadCount);
+            }
+        },
+    });
+
+    dialogs.registerMetrics(registry);
+    registry.registerMetric(info);
+    registry.registerMetric(unread);
+}
+
+class DialogsHolder {
+    private lastUpdate = 0n;
+    private dialogs: Dialog[] = [];
+    private isUpdating = false;
+    private ttl: bigint;
+
+    private dialogsIterDurationHistogram: Histogram;
+
+    constructor(private tg: TelegramClient, ttl: number, private timeout: number, private pollInterval = 10) {
+        this.ttl = BigInt(ttl) * 1000000n;
+        this.dialogsIterDurationHistogram = new Histogram({
+            name: "telegram_api_dialogs_iter_duration",
+            help: "Duration of iteration over telegram dialogs",
+        });
+    }
+
+    public registerMetrics(registry: Registry) {
+        registry.registerMetric(this.dialogsIterDurationHistogram);
+    }
+
+    public async get() {
+        if (this.isUpdating) {
+            for (let i = 0; i < this.timeout && this.isUpdating; i += this.pollInterval) {
+                await timers.setTimeout(this.pollInterval);
+            }
+            if (this.isUpdating) {
+                throw new Error("Timed out fetching dialogs");
+            }
+        }
+        if (process.hrtime.bigint() - this.lastUpdate > this.ttl) {
+            this.isUpdating = true;
+            this.dialogs = [];
+            const end = this.dialogsIterDurationHistogram.startTimer();
+            for await (const d of this.tg.iterDialogs()) {
+                if (!peersConfigBoolFilter(config, d.peer.id)) {
+                    continue;
+                }
+                this.dialogs.push(d);
+            }
+            end();
+            this.lastUpdate = process.hrtime.bigint();
+            this.isUpdating = false;
+        }
+        return this.dialogs;
+    }
+}

src/metrics/keywords.ts

@@ -1,9 +1,11 @@
 import type { Dispatcher } from "@mtcute/dispatcher";
+
 import { PropagationAction } from "@mtcute/dispatcher";
 import { Counter } from "prom-client";
-import { config } from "./config.js";
-import { peersConfigFilter } from "./filters.js";
-import { escapeRegex } from "./utils.js";
+
+import { config } from "../config.js";
+import { peersConfigFilter } from "../filters.js";
+import { escapeRegex } from "../utils.js";
 
 export interface RawKeywordPattern {
     name: string;
@@ -86,3 +88,22 @@ export class KeywordsCounter extends Counter {
         this._keywords = keywords;
     }
 }
+
+export function newWordsCounter(dp: Dispatcher) {
+    const counter = new Counter({
+        name: "messenger_dialog_words_count",
+        help: "Number of words in messages since exporter startup",
+        labelNames: ["peerId", "word"],
+    });
+    dp.onNewMessage(peersConfigFilter(config), async (msg) => {
+        const words = msg.text.toLowerCase().split(" ");
+        for (const w of words) {
+            counter.inc({
+                peerId: msg.chat.id,
+                word: w,
+            });
+        }
+        return PropagationAction.Continue;
+    });
+    return counter;
+}

src/metrics/message.ts

@@ -0,0 +1,78 @@
+import type { Dispatcher } from "@mtcute/dispatcher";
+import type { Registry } from "prom-client";
+
+import { PropagationAction } from "@mtcute/dispatcher";
+import { Counter } from "prom-client";
+
+import { config } from "../config.js";
+import { peersConfigFilter } from "../filters.js";
+
+export function collectNewMessageMetrics(dp: Dispatcher, registry: Registry) {
+    const messages = new Counter({
+        name: "messenger_dialog_messages_count",
+        help: "Messages count since exporter startup",
+        labelNames: ["peerId"],
+    });
+
+    const media = new Counter({
+        name: "messenger_dialog_media_sent_count",
+        help: "Medias sent since exporter startup",
+        labelNames: ["peerId"],
+    });
+
+    const stickers = new Counter({
+        name: "messenger_dialog_stickers_sent_count",
+        help: "Stickers sent since exporter startup",
+        labelNames: ["peerId"],
+    });
+
+    const voice = new Counter({
+        name: "messenger_dialog_voice_messages_count",
+        help: "Voice messages sent since exporter startup",
+        labelNames: ["peerId"],
+    });
+
+    dp.onNewMessage(peersConfigFilter(config), (msg) => {
+        if (msg.media) {
+            let counter;
+            switch (msg.media.type) {
+                case "photo": case "audio": case "document": {
+                    counter = media;
+                    break;
+                }
+                case "sticker": {
+                    counter = stickers;
+                    break;
+                }
+                case "voice": {
+                    counter = voice;
+                    break;
+                }
+                case "video": {
+                    if (msg.media.isRound) {
+                        counter = voice;
+                    } else {
+                        counter = media;
+                    }
+                    break;
+                }
+            }
+            if (counter) {
+                counter.inc({
+                    peerId: msg.chat.id,
+                });
+            }
+        }
+
+        messages.inc({
+            peerId: msg.chat.id,
+        });
+
+        return PropagationAction.Continue;
+    });
+
+    registry.registerMetric(media);
+    registry.registerMetric(stickers);
+    registry.registerMetric(voice);
+    registry.registerMetric(messages);
+}

src/metrics/metrics.ts

@@ -0,0 +1,37 @@
+import type { Dispatcher } from "@mtcute/dispatcher";
+
+import { PropagationAction } from "@mtcute/dispatcher";
+import { Counter } from "prom-client";
+
+import { config } from "../config.js";
+import { peersConfigFilter } from "../filters.js";
+
+import { collectDialogMetrics } from "./dialogs.js";
+import { KeywordsCounter } from "./keywords.js";
+import { collectNewMessageMetrics } from "./message.js";
+
+function newWordsCounter(dp: Dispatcher) {
+    const counter = new Counter({
+        name: "messenger_dialog_words_count",
+        help: "Number of words in messages since exporter startup",
+        labelNames: ["peerId", "word"],
+    });
+    dp.onNewMessage(peersConfigFilter(config), async (msg) => {
+        const words = msg.text.toLowerCase().split(" ");
+        for (const w of words) {
+            counter.inc({
+                peerId: msg.chat.id,
+                word: w,
+            });
+        }
+        return PropagationAction.Continue;
+    });
+    return counter;
+}
+
+export {
+    collectDialogMetrics,
+    collectNewMessageMetrics,
+    KeywordsCounter,
+    newWordsCounter,
+};