From de5988ba8fc88e9022b512cb8e8668874dc9be0f Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 22 Jan 2026 14:20:17 +0000 Subject: [PATCH] feat: implement WhatsApp connection and message metrics Added tracking for all WhatsApp-related metrics: Connection metrics: - active_connections: Gauge tracking active connections (inc on open, dec on close) - connection_attempts_total: Counter for connection attempts (success/failure) Message metrics: - messages_sent_total: Counter with type label (text, image, video, audio, etc) - messages_received_total: Counter with type label - history_sync_messages_total: Counter for history sync messages Added helper functions in prometheus-metrics.ts: - incrementActiveConnections(), decrementActiveConnections(), setActiveConnections() - recordConnectionAttempt(status) - recordMessageSent(type), recordMessageReceived(type) - recordMessageRetry(type), recordMessageFailure(type, reason) - setMessagesQueued(count, priority) - recordHistorySyncMessages(count) --- src/Socket/messages-recv.ts | 14 +++- src/Socket/messages-send.ts | 12 ++++ src/Socket/socket.ts | 15 ++++- src/Utils/process-message.ts | 7 +- src/Utils/prometheus-metrics.ts | 114 ++++++++++++++++++++++++++++++++ 5 files changed, 159 insertions(+), 3 deletions(-) diff --git a/src/Socket/messages-recv.ts b/src/Socket/messages-recv.ts index 707a884d..36deaf09 100644 --- a/src/Socket/messages-recv.ts +++ b/src/Socket/messages-recv.ts @@ -4,7 +4,7 @@ import { randomBytes } from 'crypto' import Long from 'long' import { proto } from '../../WAProto/index.js' import { DEFAULT_CACHE_TTLS, KEY_BUNDLE_TYPE, MIN_PREKEY_COUNT } from '../Defaults' -import { metrics } from '../Utils/prometheus-metrics.js' +import { metrics, recordMessageReceived, recordHistorySyncMessages } from '../Utils/prometheus-metrics.js' import type { GroupParticipant, MessageReceiptType, @@ -1387,6 +1387,18 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { if (msg.key.id && msg.key.remoteJid) { logMessageReceived(msg.key.id, msg.key.remoteJid) } + + // Record message received metric + const msgContent = msg.message + const msgType = msgContent?.conversation ? 'text' + : msgContent?.imageMessage ? 'image' + : msgContent?.videoMessage ? 'video' + : msgContent?.audioMessage ? 'audio' + : msgContent?.documentMessage ? 'document' + : msgContent?.stickerMessage ? 'sticker' + : msgContent?.reactionMessage ? 'reaction' + : 'other' + recordMessageReceived(msgType) }) } catch (error) { logger.error({ error, node: binaryNodeToString(node) }, 'error in handling message') diff --git a/src/Socket/messages-send.ts b/src/Socket/messages-send.ts index d95102c5..8e24aedc 100644 --- a/src/Socket/messages-send.ts +++ b/src/Socket/messages-send.ts @@ -56,6 +56,7 @@ import { S_WHATSAPP_NET } from '../WABinary' import { USyncQuery, USyncUser } from '../WAUSync' +import { recordMessageSent, recordMessageRetry, recordMessageFailure } from '../Utils/prometheus-metrics' import { makeNewsletterSocket } from './newsletter' export const makeMessagesSocket = (config: SocketConfig) => { @@ -1038,6 +1039,17 @@ export const makeMessagesSocket = (config: SocketConfig) => { // Log with [BAILEYS] prefix logMessageSent(msgId, destinationJid) + // Record message sent metric + const msgType = message.conversation ? 'text' + : message.imageMessage ? 'image' + : message.videoMessage ? 'video' + : message.audioMessage ? 'audio' + : message.documentMessage ? 'document' + : message.stickerMessage ? 'sticker' + : message.reactionMessage ? 'reaction' + : 'other' + recordMessageSent(msgType) + // Add message to retry cache if enabled if (messageRetryManager && !participant) { messageRetryManager.addRecentMessage(destinationJid, msgId, message) diff --git a/src/Socket/socket.ts b/src/Socket/socket.ts index 4542d288..2d3b9f2c 100644 --- a/src/Socket/socket.ts +++ b/src/Socket/socket.ts @@ -54,7 +54,12 @@ import { jidEncode, S_WHATSAPP_NET } from '../WABinary' -import { recordConnectionError } from '../Utils/prometheus-metrics' +import { + recordConnectionError, + recordConnectionAttempt, + incrementActiveConnections, + decrementActiveConnections +} from '../Utils/prometheus-metrics' import { BinaryInfo } from '../WAM/BinaryInfo.js' import { USyncQuery, USyncUser } from '../WAUSync/' import { WebSocketClient } from './Client' @@ -764,8 +769,12 @@ export const makeSocket = (config: SocketConfig) => { errorType = `error_${statusCode}` } recordConnectionError(errorType) + recordConnectionAttempt('failure') } + // Decrement active connections + decrementActiveConnections() + clearInterval(keepAliveReq) clearTimeout(qrTimer) @@ -1080,6 +1089,10 @@ export const makeSocket = (config: SocketConfig) => { ev.emit('connection.update', { connection: 'open' }) + // Record successful connection metrics + recordConnectionAttempt('success') + incrementActiveConnections() + if (node.attrs.lid && authState.creds.me?.id) { const myLID = node.attrs.lid process.nextTick(async () => { diff --git a/src/Utils/process-message.ts b/src/Utils/process-message.ts index 3487232b..b1743cca 100644 --- a/src/Utils/process-message.ts +++ b/src/Utils/process-message.ts @@ -16,7 +16,7 @@ import type { WAMessage, WAMessageKey } from '../Types' -import { metrics } from './prometheus-metrics.js' +import { metrics, recordHistorySyncMessages } from './prometheus-metrics.js' import { WAMessageStubType } from '../Types' import { getContentType, normalizeMessageContent } from '../Utils/messages' import { @@ -353,6 +353,11 @@ const processMessage = async ( isLatest: histNotification.syncType !== proto.HistorySync.HistorySyncType.ON_DEMAND ? isLatest : undefined, peerDataRequestSessionId: histNotification.peerDataRequestSessionId }) + + // Record history sync metrics + if (data.messages?.length) { + recordHistorySyncMessages(data.messages.length) + } } break diff --git a/src/Utils/prometheus-metrics.ts b/src/Utils/prometheus-metrics.ts index 85b1bec1..4543d383 100644 --- a/src/Utils/prometheus-metrics.ts +++ b/src/Utils/prometheus-metrics.ts @@ -1971,6 +1971,120 @@ export function updateAdaptiveMetrics(eventRate: number, isHealthy: boolean): vo } } +// ============================================ +// WhatsApp Connection & Message Metrics +// ============================================ + +/** + * Increment active connections gauge + */ +export function incrementActiveConnections(): void { + try { + metrics.activeConnections?.inc({}) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Decrement active connections gauge + */ +export function decrementActiveConnections(): void { + try { + metrics.activeConnections?.dec({}) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Set active connections to specific value + */ +export function setActiveConnections(count: number): void { + try { + metrics.activeConnections?.set({}, count) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Record a connection attempt + */ +export function recordConnectionAttempt(status: 'success' | 'failure'): void { + try { + metrics.connectionAttempts?.inc({ status }) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Record a message sent + */ +export function recordMessageSent(type: string = 'text'): void { + try { + metrics.messagesSent?.inc({ type }) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Record a message received + */ +export function recordMessageReceived(type: string = 'text'): void { + try { + metrics.messagesReceived?.inc({ type }) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Record a message retry attempt + */ +export function recordMessageRetry(type: string = 'text'): void { + try { + metrics.messageRetries?.inc({ type }) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Record a message failure + */ +export function recordMessageFailure(type: string = 'text', reason: string = 'unknown'): void { + try { + metrics.messageFailures?.inc({ type, reason }) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Update messages queued gauge + */ +export function setMessagesQueued(count: number, priority: string = 'normal'): void { + try { + metrics.messagesQueued?.set({ priority }, count) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Record history sync messages + */ +export function recordHistorySyncMessages(count: number = 1): void { + try { + metrics.historySyncMessages?.inc({}, count) + } catch { + // Metrics not initialized, ignore silently + } +} + // ============================================ // Global Instance // ============================================