Add message counter metrics

downstream.go

@@ -444,6 +444,15 @@ func (dc *downstreamConn) unmarshalText(uc *upstreamConn, text string) string {
 	return strings.ReplaceAll(text, "/"+uc.network.GetName(), "")
 }
 
+func (dc *downstreamConn) ReadMessage() (*irc.Message, error) {
+	msg, err := dc.conn.ReadMessage()
+	if err != nil {
+		return nil, err
+	}
+	dc.srv.metrics.downstreamInMessagesTotal.Inc()
+	return msg, nil
+}
+
 func (dc *downstreamConn) readMessages(ch chan<- event) error {
 	for {
 		msg, err := dc.ReadMessage()
@@ -498,6 +507,7 @@ func (dc *downstreamConn) SendMessage(msg *irc.Message) {
 		return
 	}
 
+	dc.srv.metrics.downstreamOutMessagesTotal.Inc()
 	dc.conn.SendMessage(msg)
 }
 

server.go

@@ -101,6 +101,11 @@ type Server struct {
 	metrics struct {
 		downstreams int64Gauge
 		upstreams   int64Gauge
+
+		upstreamOutMessagesTotal   prometheus.Counter
+		upstreamInMessagesTotal    prometheus.Counter
+		downstreamOutMessagesTotal prometheus.Counter
+		downstreamInMessagesTotal  prometheus.Counter
 	}
 }
 
@@ -170,6 +175,26 @@ func (s *Server) registerMetrics() {
 		Name: "soju_upstreams_active",
 		Help: "Current number of upstream connections",
 	}, s.metrics.upstreams.Float64)
+
+	s.metrics.upstreamOutMessagesTotal = factory.NewCounter(prometheus.CounterOpts{
+		Name: "soju_upstream_out_messages_total",
+		Help: "Total number of outgoing messages sent to upstream servers",
+	})
+
+	s.metrics.upstreamInMessagesTotal = factory.NewCounter(prometheus.CounterOpts{
+		Name: "soju_upstream_in_messages_total",
+		Help: "Total number of incoming messages received from upstream servers",
+	})
+
+	s.metrics.downstreamOutMessagesTotal = factory.NewCounter(prometheus.CounterOpts{
+		Name: "soju_downstream_out_messages_total",
+		Help: "Total number of outgoing messages sent to downstream clients",
+	})
+
+	s.metrics.downstreamInMessagesTotal = factory.NewCounter(prometheus.CounterOpts{
+		Name: "soju_downstream_in_messages_total",
+		Help: "Total number of incoming messages received from downstream clients",
+	})
 }
 
 func (s *Server) Shutdown() {

upstream.go

@@ -1806,6 +1806,15 @@ func (uc *upstreamConn) register() {
 	})
 }
 
+func (uc *upstreamConn) ReadMessage() (*irc.Message, error) {
+	msg, err := uc.conn.ReadMessage()
+	if err != nil {
+		return nil, err
+	}
+	uc.srv.metrics.upstreamInMessagesTotal.Inc()
+	return msg, nil
+}
+
 func (uc *upstreamConn) runUntilRegistered() error {
 	for !uc.registered {
 		msg, err := uc.ReadMessage()
@@ -1856,6 +1865,7 @@ func (uc *upstreamConn) SendMessage(msg *irc.Message) {
 		msg.Tags = nil
 	}
 
+	uc.srv.metrics.upstreamOutMessagesTotal.Inc()
 	uc.conn.SendMessage(msg)
 }