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)
This commit is contained in:
@@ -4,7 +4,7 @@ import { randomBytes } from 'crypto'
|
|||||||
import Long from 'long'
|
import Long from 'long'
|
||||||
import { proto } from '../../WAProto/index.js'
|
import { proto } from '../../WAProto/index.js'
|
||||||
import { DEFAULT_CACHE_TTLS, KEY_BUNDLE_TYPE, MIN_PREKEY_COUNT } from '../Defaults'
|
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 {
|
import type {
|
||||||
GroupParticipant,
|
GroupParticipant,
|
||||||
MessageReceiptType,
|
MessageReceiptType,
|
||||||
@@ -1387,6 +1387,18 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
|||||||
if (msg.key.id && msg.key.remoteJid) {
|
if (msg.key.id && msg.key.remoteJid) {
|
||||||
logMessageReceived(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) {
|
} catch (error) {
|
||||||
logger.error({ error, node: binaryNodeToString(node) }, 'error in handling message')
|
logger.error({ error, node: binaryNodeToString(node) }, 'error in handling message')
|
||||||
|
|||||||
@@ -56,6 +56,7 @@ import {
|
|||||||
S_WHATSAPP_NET
|
S_WHATSAPP_NET
|
||||||
} from '../WABinary'
|
} from '../WABinary'
|
||||||
import { USyncQuery, USyncUser } from '../WAUSync'
|
import { USyncQuery, USyncUser } from '../WAUSync'
|
||||||
|
import { recordMessageSent, recordMessageRetry, recordMessageFailure } from '../Utils/prometheus-metrics'
|
||||||
import { makeNewsletterSocket } from './newsletter'
|
import { makeNewsletterSocket } from './newsletter'
|
||||||
|
|
||||||
export const makeMessagesSocket = (config: SocketConfig) => {
|
export const makeMessagesSocket = (config: SocketConfig) => {
|
||||||
@@ -1038,6 +1039,17 @@ export const makeMessagesSocket = (config: SocketConfig) => {
|
|||||||
// Log with [BAILEYS] prefix
|
// Log with [BAILEYS] prefix
|
||||||
logMessageSent(msgId, destinationJid)
|
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
|
// Add message to retry cache if enabled
|
||||||
if (messageRetryManager && !participant) {
|
if (messageRetryManager && !participant) {
|
||||||
messageRetryManager.addRecentMessage(destinationJid, msgId, message)
|
messageRetryManager.addRecentMessage(destinationJid, msgId, message)
|
||||||
|
|||||||
+14
-1
@@ -54,7 +54,12 @@ import {
|
|||||||
jidEncode,
|
jidEncode,
|
||||||
S_WHATSAPP_NET
|
S_WHATSAPP_NET
|
||||||
} from '../WABinary'
|
} from '../WABinary'
|
||||||
import { recordConnectionError } from '../Utils/prometheus-metrics'
|
import {
|
||||||
|
recordConnectionError,
|
||||||
|
recordConnectionAttempt,
|
||||||
|
incrementActiveConnections,
|
||||||
|
decrementActiveConnections
|
||||||
|
} from '../Utils/prometheus-metrics'
|
||||||
import { BinaryInfo } from '../WAM/BinaryInfo.js'
|
import { BinaryInfo } from '../WAM/BinaryInfo.js'
|
||||||
import { USyncQuery, USyncUser } from '../WAUSync/'
|
import { USyncQuery, USyncUser } from '../WAUSync/'
|
||||||
import { WebSocketClient } from './Client'
|
import { WebSocketClient } from './Client'
|
||||||
@@ -764,8 +769,12 @@ export const makeSocket = (config: SocketConfig) => {
|
|||||||
errorType = `error_${statusCode}`
|
errorType = `error_${statusCode}`
|
||||||
}
|
}
|
||||||
recordConnectionError(errorType)
|
recordConnectionError(errorType)
|
||||||
|
recordConnectionAttempt('failure')
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Decrement active connections
|
||||||
|
decrementActiveConnections()
|
||||||
|
|
||||||
clearInterval(keepAliveReq)
|
clearInterval(keepAliveReq)
|
||||||
clearTimeout(qrTimer)
|
clearTimeout(qrTimer)
|
||||||
|
|
||||||
@@ -1080,6 +1089,10 @@ export const makeSocket = (config: SocketConfig) => {
|
|||||||
|
|
||||||
ev.emit('connection.update', { connection: 'open' })
|
ev.emit('connection.update', { connection: 'open' })
|
||||||
|
|
||||||
|
// Record successful connection metrics
|
||||||
|
recordConnectionAttempt('success')
|
||||||
|
incrementActiveConnections()
|
||||||
|
|
||||||
if (node.attrs.lid && authState.creds.me?.id) {
|
if (node.attrs.lid && authState.creds.me?.id) {
|
||||||
const myLID = node.attrs.lid
|
const myLID = node.attrs.lid
|
||||||
process.nextTick(async () => {
|
process.nextTick(async () => {
|
||||||
|
|||||||
@@ -16,7 +16,7 @@ import type {
|
|||||||
WAMessage,
|
WAMessage,
|
||||||
WAMessageKey
|
WAMessageKey
|
||||||
} from '../Types'
|
} from '../Types'
|
||||||
import { metrics } from './prometheus-metrics.js'
|
import { metrics, recordHistorySyncMessages } from './prometheus-metrics.js'
|
||||||
import { WAMessageStubType } from '../Types'
|
import { WAMessageStubType } from '../Types'
|
||||||
import { getContentType, normalizeMessageContent } from '../Utils/messages'
|
import { getContentType, normalizeMessageContent } from '../Utils/messages'
|
||||||
import {
|
import {
|
||||||
@@ -353,6 +353,11 @@ const processMessage = async (
|
|||||||
isLatest: histNotification.syncType !== proto.HistorySync.HistorySyncType.ON_DEMAND ? isLatest : undefined,
|
isLatest: histNotification.syncType !== proto.HistorySync.HistorySyncType.ON_DEMAND ? isLatest : undefined,
|
||||||
peerDataRequestSessionId: histNotification.peerDataRequestSessionId
|
peerDataRequestSessionId: histNotification.peerDataRequestSessionId
|
||||||
})
|
})
|
||||||
|
|
||||||
|
// Record history sync metrics
|
||||||
|
if (data.messages?.length) {
|
||||||
|
recordHistorySyncMessages(data.messages.length)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
break
|
break
|
||||||
|
|||||||
@@ -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
|
// Global Instance
|
||||||
// ============================================
|
// ============================================
|
||||||
|
|||||||
Reference in New Issue
Block a user