Added Websocket's broadcast queue load metrics (#13935)

Now we send metrics when the websocket's broadcast queue receive or
consume a message from the queue
Этот коммит содержится в:
Mario de Frutos Dieguez
2020-02-28 11:44:56 +01:00
коммит произвёл GitHub
родитель b07898b8a2
Коммит 75197d291a
3 изменённых файлов: 18 добавлений и 0 удалений

Просмотреть файл

@@ -343,6 +343,9 @@ func (h *Hub) Unregister(webConn *WebConn) {
func (h *Hub) Broadcast(message *model.WebSocketEvent) { func (h *Hub) Broadcast(message *model.WebSocketEvent) {
if h != nil && h.broadcast != nil && message != nil { if h != nil && h.broadcast != nil && message != nil {
if metrics := h.app.Metrics(); metrics != nil {
metrics.IncrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1)
}
select { select {
case h.broadcast <- message: case h.broadcast <- message:
case <-h.didStop: case <-h.didStop:
@@ -433,6 +436,9 @@ func (h *Hub) Start() {
} }
} }
case msg := <-h.broadcast: case msg := <-h.broadcast:
if metrics := h.app.Metrics(); metrics != nil {
metrics.DecrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1)
}
candidates := connections.All() candidates := connections.All()
if msg.GetBroadcast().UserId != "" { if msg.GetBroadcast().UserId != "" {
candidates = connections.ForUser(msg.GetBroadcast().UserId) candidates = connections.ForUser(msg.GetBroadcast().UserId)

Просмотреть файл

@@ -37,6 +37,8 @@ type MetricsInterface interface {
IncrementWebsocketEvent(eventType string) IncrementWebsocketEvent(eventType string)
IncrementWebSocketBroadcast(eventType string) IncrementWebSocketBroadcast(eventType string)
IncrementWebSocketBroadcastBufferSize(hub string, amount float64)
DecrementWebSocketBroadcastBufferSize(hub string, amount float64)
AddMemCacheHitCounter(cacheName string, amount float64) AddMemCacheHitCounter(cacheName string, amount float64)
AddMemCacheMissCounter(cacheName string, amount float64) AddMemCacheMissCounter(cacheName string, amount float64)

Просмотреть файл

@@ -21,6 +21,11 @@ func (_m *MetricsInterface) AddMemCacheMissCounter(cacheName string, amount floa
_m.Called(cacheName, amount) _m.Called(cacheName, amount)
} }
// DecrementWebSocketBroadcastBufferSize provides a mock function with given fields: hub, amount
func (_m *MetricsInterface) DecrementWebSocketBroadcastBufferSize(hub string, amount float64) {
_m.Called(hub, amount)
}
// IncrementChannelIndexCounter provides a mock function with given fields: // IncrementChannelIndexCounter provides a mock function with given fields:
func (_m *MetricsInterface) IncrementChannelIndexCounter() { func (_m *MetricsInterface) IncrementChannelIndexCounter() {
_m.Called() _m.Called()
@@ -141,6 +146,11 @@ func (_m *MetricsInterface) IncrementWebSocketBroadcast(eventType string) {
_m.Called(eventType) _m.Called(eventType)
} }
// IncrementWebSocketBroadcastBufferSize provides a mock function with given fields: hub, amount
func (_m *MetricsInterface) IncrementWebSocketBroadcastBufferSize(hub string, amount float64) {
_m.Called(hub, amount)
}
// IncrementWebhookPost provides a mock function with given fields: // IncrementWebhookPost provides a mock function with given fields:
func (_m *MetricsInterface) IncrementWebhookPost() { func (_m *MetricsInterface) IncrementWebhookPost() {
_m.Called() _m.Called()