diff --git a/src/Socket/messages-recv.ts b/src/Socket/messages-recv.ts index 707a884d..4be626e9 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, recordMessageRetry, recordMessageFailure } from '../Utils/prometheus-metrics.js' import type { GroupParticipant, MessageReceiptType, @@ -409,11 +409,13 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { if (messageRetryManager.hasExceededMaxRetries(msgId)) { logger.debug({ msgId }, 'reached retry limit with new retry manager, clearing') messageRetryManager.markRetryFailed(msgId) + recordMessageFailure('retry', 'max_retries_reached') return } // Increment retry count using new system const retryCount = messageRetryManager.incrementRetryCount(msgId) + recordMessageRetry('retry') // Use the new retry count for the rest of the logic const key = `${msgId}:${msgKey?.participant}` @@ -425,11 +427,13 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { if (retryCount >= maxMsgRetryCount) { logger.debug({ retryCount, msgId }, 'reached retry limit, clearing') await msgRetryCache.del(key) + recordMessageFailure('retry', 'max_retries_reached') return } retryCount += 1 await msgRetryCache.set(key, retryCount) + recordMessageRetry('retry') } const key = `${msgId}:${msgKey?.participant}` @@ -1387,6 +1391,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..ae2be48d 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) => { @@ -595,6 +596,7 @@ export const makeMessagesSocket = (config: SocketConfig) => { const nodes = (await Promise.all(encryptionPromises)).filter(node => node !== null) as BinaryNode[] if (recipientJids.length > 0 && nodes.length === 0) { + recordMessageFailure('send', 'encryption_failed') throw new Boom('All encryptions failed', { statusCode: 500 }) } @@ -1038,6 +1040,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 08f45ac4..2d3b9f2c 100644 --- a/src/Socket/socket.ts +++ b/src/Socket/socket.ts @@ -54,6 +54,12 @@ import { jidEncode, S_WHATSAPP_NET } from '../WABinary' +import { + recordConnectionError, + recordConnectionAttempt, + incrementActiveConnections, + decrementActiveConnections +} from '../Utils/prometheus-metrics' import { BinaryInfo } from '../WAM/BinaryInfo.js' import { USyncQuery, USyncUser } from '../WAUSync/' import { WebSocketClient } from './Client' @@ -730,6 +736,45 @@ export const makeSocket = (config: SocketConfig) => { closed = true logger.info({ trace: error?.stack }, error ? 'connection errored' : 'connection closed') + // Record connection error metric + if (error) { + const statusCode = (error as Boom)?.output?.statusCode || 0 + let errorType = 'unknown' + switch (statusCode) { + case DisconnectReason.connectionClosed: + errorType = 'connection_closed' + break + case DisconnectReason.connectionLost: + errorType = 'connection_lost' + break + case DisconnectReason.connectionReplaced: + errorType = 'connection_replaced' + break + case DisconnectReason.timedOut: + errorType = 'timed_out' + break + case DisconnectReason.loggedOut: + errorType = 'logged_out' + break + case DisconnectReason.badSession: + errorType = 'bad_session' + break + case DisconnectReason.restartRequired: + errorType = 'restart_required' + break + case DisconnectReason.multideviceMismatch: + errorType = 'multidevice_mismatch' + break + default: + errorType = `error_${statusCode}` + } + recordConnectionError(errorType) + recordConnectionAttempt('failure') + } + + // Decrement active connections + decrementActiveConnections() + clearInterval(keepAliveReq) clearTimeout(qrTimer) @@ -1044,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/circuit-breaker.ts b/src/Utils/circuit-breaker.ts index fc20e135..143f0729 100644 --- a/src/Utils/circuit-breaker.ts +++ b/src/Utils/circuit-breaker.ts @@ -380,6 +380,11 @@ export class CircuitBreaker extends EventEmitter { this.options.onOpen() this.emit('open') + // Record circuit breaker trip metric + if (this.options.collectMetrics) { + metrics.circuitBreakerTrips?.inc({ name: this.options.name }) + } + // Schedule transition to HALF_OPEN this.resetTimer = setTimeout(() => { this.transitionTo('half-open') diff --git a/src/Utils/event-buffer.ts b/src/Utils/event-buffer.ts index 15a0f47c..64aec7b2 100644 --- a/src/Utils/event-buffer.ts +++ b/src/Utils/event-buffer.ts @@ -61,7 +61,7 @@ export function loadBufferConfig(): BufferConfig { maxBufferSize: parseInt(process.env.BAILEYS_BUFFER_MAX_SIZE || '5000', 10), flushDebounceMs: parseInt(process.env.BAILEYS_BUFFER_FLUSH_DEBOUNCE_MS || '100', 10), enableAdaptiveTimeout: process.env.BAILEYS_BUFFER_ADAPTIVE_TIMEOUT !== 'false', - enableMetrics: process.env.BAILEYS_BUFFER_METRICS === 'true', + enableMetrics: process.env.BAILEYS_BUFFER_METRICS === 'true' || process.env.BAILEYS_PROMETHEUS_ENABLED === 'true', lruCleanupRatio: parseFloat(process.env.BAILEYS_BUFFER_LRU_CLEANUP_RATIO || '0.2'), bufferWarnThreshold: parseFloat(process.env.BAILEYS_BUFFER_WARN_THRESHOLD || '0.8') } @@ -329,6 +329,31 @@ class AdaptiveTimeoutCalculator { } return 'balanced' } + + /** + * Get current event rate in events per second + */ + getEventRate(): number { + if (this.eventTimestamps.length < 2) { + return 0 + } + const oldest = this.eventTimestamps[0]! + const newest = this.eventTimestamps[this.eventTimestamps.length - 1]! + const timeSpan = newest - oldest + if (timeSpan === 0) return 0 + return (this.eventTimestamps.length / timeSpan) * 1000 + } + + /** + * Check if system is healthy based on current mode + * Conservative mode with low event rate is healthy + * Aggressive mode might indicate high load + */ + isHealthy(): boolean { + const mode = this.getMode() + // System is considered healthy if not in aggressive mode (high load) + return mode !== 'aggressive' + } } // ============================================================================ @@ -410,9 +435,9 @@ export const makeEventBuffer = ( } } - const recordFlushMetrics = (eventCount: number, forced: boolean) => { + const recordFlushMetrics = (eventCount: number, forced: boolean, cacheSize: number) => { if (metricsModule) { - metricsModule.recordBufferFlush(eventCount, forced) + metricsModule.recordBufferFlush(eventCount, forced, cacheSize) } } @@ -449,6 +474,10 @@ export const makeEventBuffer = ( currentSize: currentEventCount, maxSize: config.maxBufferSize }) + // Record overflow metric + if (metricsModule) { + metricsModule.recordBufferOverflow() + } flush(true) return true } @@ -482,6 +511,10 @@ export const makeEventBuffer = ( removed: removed.length, remaining: historyCache.size }) + // Record metrics for cache cleanup + if (metricsModule) { + metricsModule.recordCacheCleanup(removed.length) + } } } @@ -553,7 +586,12 @@ export const makeEventBuffer = ( stats.historyCacheSize = historyCache.size // Record metrics - recordFlushMetrics(eventCount, force) + recordFlushMetrics(eventCount, force, historyCache.size) + + // Update adaptive metrics + if (config.enableAdaptiveTimeout && metricsModule) { + metricsModule.updateAdaptiveMetrics(adaptiveTimeout.getEventRate(), adaptiveTimeout.isHealthy()) + } // Log with [BAILEYS] prefix - use getMode() to avoid duplicating mode calculation logic const flushDuration = Date.now() - flushStartTime @@ -599,11 +637,16 @@ export const makeEventBuffer = ( } logger.debug('Destroying event buffer') + const hadPendingFlush = isBuffering destroyed = true // Flush any remaining events - if (isBuffering) { + if (hadPendingFlush) { flush(true) + // Record final flush metric + if (metricsModule) { + metricsModule.recordBufferFinalFlush() + } } // Clear all timers @@ -616,6 +659,11 @@ export const makeEventBuffer = ( // Remove all event listeners ev.removeAllListeners() + // Record buffer destroyed metric + if (metricsModule) { + metricsModule.recordBufferDestroyed('normal', hadPendingFlush) + } + logger.debug('Event buffer destroyed successfully') } 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 378ce449..c24a43fe 100644 --- a/src/Utils/prometheus-metrics.ts +++ b/src/Utils/prometheus-metrics.ts @@ -95,6 +95,31 @@ export function getConfiguredPrefix(): string { return configuredPrefix } +/** + * Check if a metric with the given name already exists in the custom registry + */ +function metricExists(name: string): boolean { + const fullName = getFullMetricName(name) + try { + const existing = customRegistry.getSingleMetric(fullName) + return existing !== undefined + } catch { + return false + } +} + +/** + * Get an existing metric from the custom registry + */ +function getExistingMetric(name: string): T | undefined { + const fullName = getFullMetricName(name) + try { + return customRegistry.getSingleMetric(fullName) as T | undefined + } catch { + return undefined + } +} + // ============================================ // Configuration // ============================================ @@ -1051,9 +1076,11 @@ export class SystemMetricsCollector { new Gauge('system_load_average', 'System load average', ['period']) ) - // Event loop lag + // Event loop lag - use different name to avoid conflict with collectDefaultMetrics + // prom-client's collectDefaultMetrics creates nodejs_eventloop_lag_seconds + // We use system_eventloop_lag_seconds to avoid duplicate registration this.eventLoopLag = registry.register( - new Histogram('nodejs_eventloop_lag_seconds', 'Event loop lag in seconds', [], DEFAULT_LATENCY_BUCKETS) + new Histogram('system_eventloop_lag_seconds', 'Event loop lag in seconds', [], DEFAULT_LATENCY_BUCKETS) ) // Initialize CPU baseline @@ -1364,6 +1391,12 @@ export const metrics = { connectionLatency: baileysMetrics.register( new Histogram('connection_latency_ms', 'Connection establishment latency in ms', [], [100, 250, 500, 1000, 2500, 5000, 10000]) ), + activeConnections: baileysMetrics.register( + new Gauge('active_connections', 'Number of active WhatsApp connections') + ), + connectionErrors: baileysMetrics.register( + new Counter('connection_errors_total', 'Total connection errors', ['error_type']) + ), // ========== Message Metrics ========== messagesSent: baileysMetrics.register( @@ -1445,6 +1478,18 @@ export const metrics = { eventsProcessed: baileysMetrics.register( new Counter('events_processed_total', 'Total events processed from buffer', ['event_type']) ), + bufferDestroyed: baileysMetrics.register( + new Counter('buffer_destroyed_total', 'Total buffers destroyed', ['reason', 'had_pending_flush']) + ), + bufferFinalFlush: baileysMetrics.register( + new Counter('buffer_final_flush_total', 'Total final flushes during destruction') + ), + bufferCacheCleanup: baileysMetrics.register( + new Counter('buffer_cache_cleanup_total', 'Total cache cleanup operations') + ), + bufferCacheSize: baileysMetrics.register( + new Gauge('buffer_cache_size', 'Current buffer cache size') + ), // ========== Adaptive Flush Metrics ========== adaptiveFlushInterval: baileysMetrics.register( @@ -1462,6 +1507,12 @@ export const metrics = { adaptiveFlushEfficiency: baileysMetrics.register( new Gauge('adaptive_flush_efficiency_percent', 'Flush efficiency percentage') ), + adaptiveHealthStatus: baileysMetrics.register( + new Gauge('adaptive_health_status', 'Adaptive system health status (0=unhealthy, 1=healthy)') + ), + adaptiveEventRate: baileysMetrics.register( + new Gauge('adaptive_event_rate', 'Current event rate per second') + ), // ========== Error Metrics ========== errors: baileysMetrics.register( @@ -1747,31 +1798,19 @@ export function trackOperation( // Event Buffer Metrics Integration // ============================================ -// Event buffer metrics (lazy initialized) +// Event buffer metrics (lazy initialized) - for detailed buffer tracking +// Note: bufferFlushes is NOT here - use metrics.bufferFlushes from main metrics object let eventBufferMetrics: { - eventsBuffered: Counter | null - bufferFlushes: Counter | null bufferFlushEvents: Histogram | null bufferCurrentSize: Gauge | null bufferPeakSize: Gauge | null bufferHistoryCacheSize: Gauge | null - bufferOverflows: Counter | null bufferLruCleanups: Counter | null } | null = null function getEventBufferMetrics() { if (!eventBufferMetrics) { eventBufferMetrics = { - eventsBuffered: new Counter( - 'events_buffered_total', - 'Total number of events buffered', - ['event_type'] - ), - bufferFlushes: new Counter( - 'buffer_flushes_total', - 'Total number of buffer flushes', - ['forced'] - ), bufferFlushEvents: new Histogram( 'buffer_flush_events', 'Number of events per buffer flush', @@ -1790,24 +1829,11 @@ function getEventBufferMetrics() { 'buffer_history_cache_size', 'Current size of history cache' ), - bufferOverflows: new Counter( - 'buffer_overflows_total', - 'Total number of buffer overflows detected' - ), bufferLruCleanups: new Counter( 'buffer_lru_cleanups_total', 'Total number of LRU cache cleanups performed' ) } - // Register all metrics - if (eventBufferMetrics.eventsBuffered) baileysMetrics.register(eventBufferMetrics.eventsBuffered) - if (eventBufferMetrics.bufferFlushes) baileysMetrics.register(eventBufferMetrics.bufferFlushes) - if (eventBufferMetrics.bufferFlushEvents) baileysMetrics.register(eventBufferMetrics.bufferFlushEvents) - if (eventBufferMetrics.bufferCurrentSize) baileysMetrics.register(eventBufferMetrics.bufferCurrentSize) - if (eventBufferMetrics.bufferPeakSize) baileysMetrics.register(eventBufferMetrics.bufferPeakSize) - if (eventBufferMetrics.bufferHistoryCacheSize) baileysMetrics.register(eventBufferMetrics.bufferHistoryCacheSize) - if (eventBufferMetrics.bufferOverflows) baileysMetrics.register(eventBufferMetrics.bufferOverflows) - if (eventBufferMetrics.bufferLruCleanups) baileysMetrics.register(eventBufferMetrics.bufferLruCleanups) } return eventBufferMetrics } @@ -1818,7 +1844,7 @@ function getEventBufferMetrics() { */ export function recordEventBuffered(eventType: string, count: number = 1): void { try { - const metrics = getEventBufferMetrics() + // Use the main metrics object which has eventsBuffered with label ['event_type'] metrics.eventsBuffered?.inc({ event_type: eventType }, count) } catch { // Metrics not initialized, ignore silently @@ -1829,12 +1855,23 @@ export function recordEventBuffered(eventType: string, count: number = 1): void * Record a buffer flush operation * Used by event-buffer.ts for metrics integration */ -export function recordBufferFlush(eventCount: number, forced: boolean): void { +export function recordBufferFlush(eventCount: number, forced: boolean, historyCacheSize?: number): void { try { - const metrics = getEventBufferMetrics() - metrics.bufferFlushes?.inc({ forced: forced ? 'true' : 'false' }) - metrics.bufferFlushEvents?.observe({}, eventCount) - metrics.bufferCurrentSize?.set({}, 0) // Reset after flush + // Use the main metrics object which has the correct labels ['type', 'reason'] + metrics.bufferFlushes?.inc({ type: 'event', reason: forced ? 'forced' : 'normal' }) + + // Update buffer cache size if provided + if (typeof historyCacheSize === 'number') { + metrics.bufferCacheSize?.set({}, historyCacheSize) + } + + // Also update the lazy-loaded event buffer metrics for detailed tracking + const ebMetrics = getEventBufferMetrics() + ebMetrics.bufferFlushEvents?.observe({}, eventCount) + ebMetrics.bufferCurrentSize?.set({}, 0) // Reset after flush + if (typeof historyCacheSize === 'number') { + ebMetrics.bufferHistoryCacheSize?.set({}, historyCacheSize) + } } catch { // Metrics not initialized, ignore silently } @@ -1852,10 +1889,197 @@ export function updateBufferStatistics(stats: { lruCleanups: number }): void { try { - const metrics = getEventBufferMetrics() - metrics.bufferCurrentSize?.set({}, stats.currentSize) - metrics.bufferPeakSize?.set({}, stats.peakSize) - metrics.bufferHistoryCacheSize?.set({}, stats.historyCacheSize) + const ebMetrics = getEventBufferMetrics() + ebMetrics.bufferCurrentSize?.set({}, stats.currentSize) + ebMetrics.bufferPeakSize?.set({}, stats.peakSize) + ebMetrics.bufferHistoryCacheSize?.set({}, stats.historyCacheSize) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Record a cache cleanup operation + * Used by event-buffer.ts when LRU cleanup is performed + */ +export function recordCacheCleanup(removedCount: number): void { + try { + metrics.bufferCacheCleanup?.inc({}) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Record a buffer overflow event + * Used by event-buffer.ts when buffer exceeds max size + */ +export function recordBufferOverflow(): void { + try { + metrics.bufferOverflows?.inc({ type: 'event' }) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Record a connection error + * Used by socket.ts when connection fails + */ +export function recordConnectionError(errorType: string): void { + try { + metrics.connectionErrors?.inc({ error_type: errorType }) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Record buffer destruction + * Used by event-buffer.ts when buffer is destroyed + */ +export function recordBufferDestroyed(reason: string, hadPendingFlush: boolean): void { + try { + metrics.bufferDestroyed?.inc({ reason, had_pending_flush: hadPendingFlush ? 'true' : 'false' }) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Record final flush during buffer destruction + * Used by event-buffer.ts when buffer flushes remaining events on destroy + */ +export function recordBufferFinalFlush(): void { + try { + metrics.bufferFinalFlush?.inc({}) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Update adaptive system metrics + * Used by event-buffer.ts to report adaptive timeout health and event rate + */ +export function updateAdaptiveMetrics(eventRate: number, isHealthy: boolean): void { + try { + metrics.adaptiveEventRate?.set({}, eventRate) + metrics.adaptiveHealthStatus?.set({}, isHealthy ? 1 : 0) + } catch { + // Metrics not initialized, ignore silently + } +} + +// ============================================ +// 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 } @@ -1880,12 +2104,59 @@ export function getMetricsManager(config?: Partial): PrometheusMe return globalMetricsManager } +/** + * Initialize metrics with labels to zero values + * This ensures they appear in Prometheus output even before first increment + */ +function initializeMetricsWithLabels(): void { + try { + // Buffer destroyed metric + metrics.bufferDestroyed?.inc({ reason: 'normal', had_pending_flush: 'true' }, 0) + metrics.bufferDestroyed?.inc({ reason: 'normal', had_pending_flush: 'false' }, 0) + metrics.bufferDestroyed?.inc({ reason: 'error', had_pending_flush: 'true' }, 0) + metrics.bufferDestroyed?.inc({ reason: 'error', had_pending_flush: 'false' }, 0) + + // Buffer flushes metric + metrics.bufferFlushes?.inc({ type: 'event', reason: 'normal' }, 0) + metrics.bufferFlushes?.inc({ type: 'event', reason: 'forced' }, 0) + + // Connection metrics + metrics.connectionAttempts?.inc({ status: 'success' }, 0) + metrics.connectionAttempts?.inc({ status: 'failure' }, 0) + metrics.connectionErrors?.inc({ error_type: 'timeout' }, 0) + metrics.connectionErrors?.inc({ error_type: 'closed' }, 0) + metrics.connectionErrors?.inc({ error_type: 'unknown' }, 0) + + // Message metrics + metrics.messagesSent?.inc({ type: 'text' }, 0) + metrics.messagesSent?.inc({ type: 'media' }, 0) + metrics.messagesSent?.inc({ type: 'other' }, 0) + metrics.messagesReceived?.inc({ type: 'text' }, 0) + metrics.messagesReceived?.inc({ type: 'media' }, 0) + metrics.messagesReceived?.inc({ type: 'notification' }, 0) + metrics.messagesReceived?.inc({ type: 'other' }, 0) + metrics.messageRetries?.inc({ type: 'text' }, 0) + metrics.messageRetries?.inc({ type: 'other' }, 0) + metrics.messageFailures?.inc({ type: 'text', reason: 'max_retries' }, 0) + metrics.messageFailures?.inc({ type: 'other', reason: 'max_retries' }, 0) + + // Circuit breaker metric + metrics.circuitBreakerTrips?.inc({ name: 'main' }, 0) + + console.log('[Prometheus] Initialized metrics with labels') + } catch (error) { + console.error('[Prometheus] Error initializing metrics with labels:', error) + } +} + /** * Initialize global metrics (call once at application startup) */ export async function initializeMetrics(config?: Partial): Promise { const manager = getMetricsManager(config) await manager.initialize() + // Initialize metrics with labels to zero values + initializeMetricsWithLabels() return manager }