proxy:collects maxpipeline and maxproxyclient error (#1408)

* proxy:collects pipeline and proxyclient error

* proxy:flush stats when session exit

* proxy:delete unnecessary flushOpStats function

Author: tangcong

pkg/proxy/session.go

@@ -34,6 +34,7 @@ type Session struct {
 	stats struct {
 		opmap map[string]*opStats
 		total atomic2.Int64
+		fails atomic2.Int64
 		flush struct {
 			n    uint
 			nano int64
@@ -116,6 +117,8 @@ func (s *Session) Start(d *Router) {
 			go func() {
 				s.Conn.Encode(redis.NewErrorf("ERR max number of clients reached"), true)
 				s.CloseWithError(ErrTooManySessions)
+				s.incrFails()
+				s.flushOpStats(true)
 			}()
 			decrSessions()
 			return
@@ -125,6 +128,8 @@ func (s *Session) Start(d *Router) {
 			go func() {
 				s.Conn.Encode(redis.NewErrorf("ERR router is not online"), true)
 				s.CloseWithError(ErrRouterNotOnline)
+				s.incrFails()
+				s.flushOpStats(true)
 			}()
 			decrSessions()
 			return
@@ -162,6 +167,7 @@ func (s *Session) loopReader(tasks *RequestChan, d *Router) (err error) {
 		s.incrOpTotal()
 
 		if tasks.Buffered() > maxPipelineLen {
+			s.incrFails()
 			return ErrTooManyPipelinedRequests
 		}
 
@@ -629,6 +635,10 @@ func (s *Session) incrOpTotal() {
 	s.stats.total.Incr()
 }
 
+func (s *Session) incrFails() {
+	s.stats.fails.Incr()
+}
+
 func (s *Session) getOpStats(opstr string) *opStats {
 	e := s.stats.opmap[opstr]
 	if e == nil {
@@ -665,6 +675,7 @@ func (s *Session) flushOpStats(force bool) {
 	s.stats.flush.nano = nano
 
 	incrOpTotal(s.stats.total.Swap(0))
+	incrOpFails(s.stats.fails.Swap(0))
 	for _, e := range s.stats.opmap {
 		if e.calls.Int64() != 0 || e.fails.Int64() != 0 {
 			incrOpStats(e)

pkg/proxy/stats.go

@@ -149,6 +149,10 @@ func incrOpTotal(n int64) {
 	cmdstats.total.Add(n)
 }
 
+func incrOpFails(n int64) {
+	cmdstats.fails.Add(n)
+}
+
 func incrOpStats(e *opStats) {
 	s := getOpStats(e.opstr, true)
 	s.calls.Add(e.calls.Swap(0))