diff --git a/server/hdl_websock.go b/server/hdl_websock.go index 7ead35a..61ce16d 100644 --- a/server/hdl_websock.go +++ b/server/hdl_websock.go @@ -11,12 +11,17 @@ package main import ( "encoding/json" + "errors" + "io" + "net" "net/http" "sync/atomic" + "syscall" "time" "github.com/gorilla/websocket" "github.com/tinode/chat/server/logs" + immetrics "github.com/tinode/chat/server/metrics" "github.com/tinode/chat/server/store/types" ) @@ -56,9 +61,10 @@ func (sess *Session) readLoop() { // Read a ClientComMessage _, raw, err := sess.ws.ReadMessage() if err != nil { - if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure, - websocket.CloseNormalClosure) { - logs.Err.Println("ws: readLoop", sess.sid, err) + reason := classifyWSError(err) + immetrics.IncWSDisconnect(reason) + if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) { + logs.Err.Printf("ws: readLoop sid=%s uid=%s reason=%s err=%v", sess.sid, sess.uid.UserId(), reason, err) } return } @@ -68,17 +74,15 @@ func (sess *Session) readLoop() { } func (sess *Session) sendMessage(msg any) bool { - if len(sess.send) > sendQueueLimit { - logs.Err.Println("ws: outbound queue limit exceeded", sess.sid) + if depth := len(sess.send); depth > sendQueueLimit { + immetrics.ObserveWSWrite("text", "rejected", "queue_limit", time.Now(), 0) + logs.Err.Printf("ws: outbound queue limit exceeded sid=%s uid=%s depth=%d capacity=%d", + sess.sid, sess.uid.UserId(), depth, cap(sess.send)) return false } statsInc("OutgoingMessagesWebsockTotal", 1) - if err := wsWrite(sess.ws, websocket.TextMessage, msg); err != nil { - if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure, - websocket.CloseNormalClosure) { - logs.Err.Println("ws: writeLoop", sess.sid, err) - } + if err := sess.writeWSFrame(websocket.TextMessage, msg); err != nil { return false } return true @@ -100,6 +104,7 @@ func (sess *Session) writeLoop() { // Channel closed. return } + sess.observeOutboundDequeue(false) switch v := msg.(type) { case []*ServerComMessage: // batch of unserialized messages for _, msg := range v { @@ -128,7 +133,7 @@ func (sess *Session) writeLoop() { case msg := <-sess.stop: // Shutdown requested, don't care if the message is delivered if msg != nil { - wsWrite(sess.ws, websocket.TextMessage, msg) + sess.writeWSFrame(websocket.TextMessage, msg) } return @@ -136,17 +141,70 @@ func (sess *Session) writeLoop() { sess.delSub(topic) case <-ticker.C: - if err := wsWrite(sess.ws, websocket.PingMessage, nil); err != nil { - if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure, - websocket.CloseNormalClosure) { - logs.Err.Println("ws: writeLoop ping", sess.sid, err) - } + if err := sess.writeWSFrame(websocket.PingMessage, nil); err != nil { return } } } } +func (sess *Session) writeWSFrame(mt int, msg any) error { + frame := "text" + if mt == websocket.PingMessage { + frame = "ping" + } + payloadBytes := 0 + if bits, ok := msg.([]byte); ok { + payloadBytes = len(bits) + } + started := time.Now() + err := wsWrite(sess.ws, mt, msg) + if err == nil { + immetrics.ObserveWSWrite(frame, "ok", "none", started, payloadBytes) + return nil + } + reason := classifyWSError(err) + immetrics.ObserveWSWrite(frame, "error", reason, started, payloadBytes) + logs.Err.Printf("ws: writeLoop sid=%s uid=%s frame=%s queue=%d capacity=%d reason=%s err=%v", + sess.sid, sess.uid.UserId(), frame, len(sess.send), cap(sess.send), reason, err) + return err +} + +func classifyWSError(err error) string { + if err == nil { + return "none" + } + var closeErr *websocket.CloseError + if errors.As(err, &closeErr) { + switch closeErr.Code { + case websocket.CloseNormalClosure: + return "close_normal" + case websocket.CloseGoingAway: + return "close_going_away" + case websocket.CloseAbnormalClosure: + return "close_abnormal" + default: + return "close_other" + } + } + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return "timeout" + } + switch { + case errors.Is(err, syscall.ECONNRESET): + return "connection_reset" + case errors.Is(err, syscall.EPIPE): + return "broken_pipe" + case errors.Is(err, net.ErrClosed): + return "closed" + case errors.Is(err, io.EOF): + return "eof" + default: + return "other" + } +} + // Writes a message with the given message type (mt) and payload. func wsWrite(ws *websocket.Conn, mt int, msg any) error { var bits []byte diff --git a/server/hdl_websock_test.go b/server/hdl_websock_test.go new file mode 100644 index 0000000..5e740e0 --- /dev/null +++ b/server/hdl_websock_test.go @@ -0,0 +1,52 @@ +package main + +import ( + "fmt" + "io" + "net" + "os" + "syscall" + "testing" + + "github.com/gorilla/websocket" +) + +func TestClassifyWSError(t *testing.T) { + tests := []struct { + name string + err error + want string + }{ + {name: "none", want: "none"}, + {name: "normal close", err: &websocket.CloseError{Code: websocket.CloseNormalClosure}, want: "close_normal"}, + {name: "abnormal close", err: &websocket.CloseError{Code: websocket.CloseAbnormalClosure}, want: "close_abnormal"}, + {name: "timeout", err: &net.OpError{Op: "write", Err: os.ErrDeadlineExceeded}, want: "timeout"}, + {name: "reset", err: fmt.Errorf("write failed: %w", syscall.ECONNRESET), want: "connection_reset"}, + {name: "broken pipe", err: fmt.Errorf("write failed: %w", syscall.EPIPE), want: "broken_pipe"}, + {name: "closed", err: net.ErrClosed, want: "closed"}, + {name: "eof", err: io.EOF, want: "eof"}, + {name: "other", err: fmt.Errorf("unknown"), want: "other"}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := classifyWSError(tc.err); got != tc.want { + t.Fatalf("classifyWSError()=%q want=%q", got, tc.want) + } + }) + } +} + +func TestWebSocketOutboundQueueTiming(t *testing.T) { + sess := &Session{proto: WEBSOCK, send: make(chan any, 2)} + payload := []byte("message") + if !sess.queueOutBytes(payload) { + t.Fatal("queueOutBytes rejected payload") + } + + queued := <-sess.send + sess.observeOutboundDequeue(false) + if string(queued.([]byte)) != string(payload) { + t.Fatalf("payload=%q want=%q", queued, payload) + } +} diff --git a/server/metrics/metrics.go b/server/metrics/metrics.go index 718a71e..972466a 100644 --- a/server/metrics/metrics.go +++ b/server/metrics/metrics.go @@ -54,6 +54,45 @@ var ( }, func() float64 { return float64(expvarIntValue("LiveSessions")) }) + IMWSWriteTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "im_ws_write_total", + Help: "WebSocket frame write and pre-write rejection outcomes by pod, frame type, result, and bounded reason.", + }, []string{"pod", "frame", "result", "reason"}) + IMWSWriteDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "im_ws_write_duration_seconds", + Help: "WebSocket frame write duration by pod, frame type, and result.", + Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}, + }, []string{"pod", "frame", "result"}) + IMWSWriteBytesTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "im_ws_write_bytes_total", + Help: "WebSocket payload bytes submitted to the socket by pod and frame type.", + }, []string{"pod", "frame"}) + IMWSDisconnectTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "im_ws_disconnect_total", + Help: "WebSocket read-loop disconnects by pod and bounded reason.", + }, []string{"pod", "reason"}) + IMWSSendQueueCurrent = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "im_ws_send_queue_current", + Help: "Current number of queued WebSocket outbound channel entries across sessions on a pod.", + }, []string{"pod"}) + IMWSSendQueueCapacity = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "im_ws_send_queue_capacity", + Help: "Configured outbound channel capacity of one WebSocket session by pod.", + }, []string{"pod"}) + IMWSSendQueueEnqueueTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "im_ws_send_queue_enqueue_total", + Help: "WebSocket outbound channel enqueue attempts by pod and result.", + }, []string{"pod", "result"}) + IMWSSendQueueDepth = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "im_ws_send_queue_depth", + Help: "Per-session WebSocket outbound channel depth observed after enqueue attempts.", + Buckets: []float64{0, 1, 4, 8, 16, 32, 64, 96, 128, 144, 160}, + }, []string{"pod", "result"}) + IMWSSendQueueWaitDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "im_ws_send_queue_wait_duration_seconds", + Help: "Time a WebSocket outbound channel entry waited before dequeue or purge.", + Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120}, + }, []string{"pod", "result"}) IMClusterLivePeers = prometheus.NewGaugeFunc(prometheus.GaugeOpts{ Name: "im_cluster_live_peers", Help: "Current connected cluster peers.", @@ -144,7 +183,10 @@ var ( ) func init() { - Registry.MustRegister(WSConnectionsActive, IMWSConnectionsActive, IMClusterLivePeers, IMClusterExpectedPeers, + Registry.MustRegister(WSConnectionsActive, IMWSConnectionsActive, IMWSWriteTotal, IMWSWriteDuration, + IMWSWriteBytesTotal, IMWSDisconnectTotal, IMWSSendQueueCurrent, IMWSSendQueueCapacity, + IMWSSendQueueEnqueueTotal, IMWSSendQueueDepth, IMWSSendQueueWaitDuration, + IMClusterLivePeers, IMClusterExpectedPeers, HTTPRequestsTotal, HTTPRequestDuration, DBQueryDuration, RealtimeFanoutEventsTotal, RealtimeFanoutDeliveryDuration, ClusterP2MQueueLength, ClusterP2MQueueCapacity, ClusterP2MEnqueueTotal, ClusterRPCInflight, ClusterRPCDuration, @@ -203,6 +245,48 @@ func ObserveInternalQueueEnqueue(queue string, depth, capacity int, accepted boo InternalQueueCapacity.WithLabelValues(pod, queue).Set(float64(capacity)) } +func ObserveWSWrite(frame, result, reason string, started time.Time, payloadBytes int) { + pod := metricPodName() + frame = normalizeLabel(frame) + result = normalizeLabel(result) + reason = normalizeLabel(reason) + IMWSWriteTotal.WithLabelValues(pod, frame, result, reason).Inc() + IMWSWriteDuration.WithLabelValues(pod, frame, result).Observe(time.Since(started).Seconds()) + if payloadBytes > 0 { + IMWSWriteBytesTotal.WithLabelValues(pod, frame).Add(float64(payloadBytes)) + } +} + +func IncWSDisconnect(reason string) { + IMWSDisconnectTotal.WithLabelValues(metricPodName(), normalizeLabel(reason)).Inc() +} + +func ObserveWSSendQueueEnqueue(depth, capacity int, accepted bool) { + result := "ok" + if !accepted { + result = "full" + } + pod := metricPodName() + IMWSSendQueueEnqueueTotal.WithLabelValues(pod, result).Inc() + IMWSSendQueueDepth.WithLabelValues(pod, result).Observe(float64(depth)) + IMWSSendQueueCapacity.WithLabelValues(pod).Set(float64(capacity)) + if accepted { + IMWSSendQueueCurrent.WithLabelValues(pod).Inc() + } +} + +func ObserveWSSendQueueDequeue(queuedAt time.Time, purged bool) { + result := "dequeued" + if purged { + result = "purged" + } + pod := metricPodName() + IMWSSendQueueCurrent.WithLabelValues(pod).Dec() + if !queuedAt.IsZero() { + IMWSSendQueueWaitDuration.WithLabelValues(pod, result).Observe(time.Since(queuedAt).Seconds()) + } +} + func normalizeLabel(value string) string { if value == "" { return "unknown" diff --git a/server/monitoring_http_test.go b/server/monitoring_http_test.go index 9bc2523..93e4f02 100644 --- a/server/monitoring_http_test.go +++ b/server/monitoring_http_test.go @@ -91,6 +91,10 @@ func TestServeMetricsIncludesWSConnections(t *testing.T) { immetrics.KafkaConsumerLag.WithLabelValues("test", "topic", "group").Set(0) immetrics.KafkaDeadLettersTotal.WithLabelValues("p2p", "test").Add(0) immetrics.ObserveInternalQueueEnqueue("test", 0, 1, true) + immetrics.ObserveWSWrite("text", "ok", "none", time.Now(), 10) + immetrics.IncWSDisconnect("close_normal") + immetrics.ObserveWSSendQueueEnqueue(1, 160, true) + immetrics.ObserveWSSendQueueDequeue(time.Now(), false) wrt := httptest.NewRecorder() serveMetrics(wrt, httptest.NewRequest(http.MethodGet, "/metrics", nil)) @@ -102,6 +106,9 @@ func TestServeMetricsIncludesWSConnections(t *testing.T) { t.Fatalf("unexpected body %s", body) } for _, metric := range []string{"im_ws_connections_active", "im_cluster_p2m_queue_length", + "im_ws_write_total", "im_ws_write_duration_seconds", "im_ws_write_bytes_total", + "im_ws_disconnect_total", "im_ws_send_queue_current", "im_ws_send_queue_capacity", + "im_ws_send_queue_enqueue_total", "im_ws_send_queue_depth", "im_ws_send_queue_wait_duration_seconds", "im_cluster_p2m_enqueue_total", "im_cluster_rpc_duration_seconds", "im_cluster_route_total", "im_internal_queue_enqueue_total", "im_internal_queue_depth", "im_internal_queue_capacity", "im_kafka_message_persist_total", "im_kafka_consumer_lag", "im_kafka_dead_letters_total"} { diff --git a/server/session.go b/server/session.go index d8dc7e3..d5b9e06 100644 --- a/server/session.go +++ b/server/session.go @@ -141,6 +141,10 @@ type Session struct { // Outbound mesages, buffered. // The content must be serialized in format suitable for the session. send chan any + // WebSocket queue timestamps are kept separately to preserve the send channel's payload contract. + wsQueueMu sync.Mutex + wsQueueTimes []time.Time + wsQueueHead int // Channel for shutting down the session, buffer 1. // Content in the same format as for 'send' @@ -182,6 +186,52 @@ type Subscription struct { supd chan<- *sessionUpdate } +func (s *Session) enqueueOutbound(payload any) bool { + if s.proto != WEBSOCK { + select { + case s.send <- payload: + return true + default: + return false + } + } + + s.wsQueueMu.Lock() + defer s.wsQueueMu.Unlock() + select { + case s.send <- payload: + s.wsQueueTimes = append(s.wsQueueTimes, time.Now()) + immetrics.ObserveWSSendQueueEnqueue(len(s.send), cap(s.send), true) + return true + default: + immetrics.ObserveWSSendQueueEnqueue(len(s.send), cap(s.send), false) + return false + } +} + +func (s *Session) observeOutboundDequeue(purged bool) { + if s.proto != WEBSOCK { + return + } + s.wsQueueMu.Lock() + if s.wsQueueHead >= len(s.wsQueueTimes) { + s.wsQueueMu.Unlock() + return + } + queuedAt := s.wsQueueTimes[s.wsQueueHead] + s.wsQueueHead++ + if s.wsQueueHead == len(s.wsQueueTimes) { + s.wsQueueTimes = s.wsQueueTimes[:0] + s.wsQueueHead = 0 + } else if s.wsQueueHead >= 128 { + copy(s.wsQueueTimes, s.wsQueueTimes[s.wsQueueHead:]) + s.wsQueueTimes = s.wsQueueTimes[:len(s.wsQueueTimes)-s.wsQueueHead] + s.wsQueueHead = 0 + } + s.wsQueueMu.Unlock() + immetrics.ObserveWSSendQueueDequeue(queuedAt, purged) +} + func (s *Session) addSub(topic string, sub *Subscription) { if s.multi != nil { s.multi.addSub(topic, sub) @@ -333,9 +383,7 @@ func (s *Session) queueOutBatch(msgs []*ServerComMessage) bool { } if s.supportsMessageBatching() { - select { - case s.send <- msgs: - default: + if !s.enqueueOutbound(msgs) { // Never block here since it may also block the topic's run() goroutine. logs.Err.Println("s.queueOut: session's send queue2 full", s.sid) return false @@ -394,9 +442,7 @@ func (s *Session) queueOut(msg *ServerComMessage) bool { } } - select { - case s.send <- msg: - default: + if !s.enqueueOutbound(msg) { // Never block here since it may also block the topic's run() goroutine. logs.Err.Println("s.queueOut: session's send queue full", s.sid) return false @@ -414,9 +460,7 @@ func (s *Session) queueOutBytes(data []byte) bool { return true } - select { - case s.send <- data: - default: + if !s.enqueueOutbound(data) { logs.Err.Println("s.queueOutBytes: session's send queue full", s.sid) return false } @@ -451,6 +495,7 @@ func (s *Session) stopSession(data any) { func (s *Session) purgeChannels() { for len(s.send) > 0 { <-s.send + s.observeOutboundDequeue(true) } for len(s.stop) > 0 { <-s.stop