feat(recycler): add error cecycler (#10)

pkg/dns/dns.go

@@ -8,6 +8,7 @@ import (
 	"github.com/li4n0/revsuit/internal/database"
 	"github.com/li4n0/revsuit/internal/newdns"
 	"github.com/li4n0/revsuit/internal/qqwry"
+	"github.com/li4n0/revsuit/internal/recycler"
 	"github.com/li4n0/revsuit/internal/rule"
 	"github.com/patrickmn/go-cache"
 	log "unknwon.dev/clog/v2"
@@ -52,6 +53,12 @@ func (s *Server) Run() {
 
 	//create new dns zone with root domain
 	newZone := func(name string) *newdns.Zone {
+		defer func() {
+			if err := recover(); err != nil {
+				recycler.Recycle(err)
+			}
+		}()
+
 		domain := strings.TrimSuffix(name, ".")
 		frags := strings.Split(domain, ".")
 		zoneName := ""
@@ -79,7 +86,7 @@ func (s *Server) Run() {
 
 					r, err := newRecord(_rule, flag, domain, ip, qqwry.Area(ip))
 					if err != nil {
-						log.Error("DNS record(rule_id:%s) created failed :%s", _rule.Name, err)
+						log.Warn("DNS record(rule_id:%s) created failed :%s", _rule.Name, err)
 						return nil, nil
 					}
 					log.Info("DNS record[id:%d rule:%s remote_ip:%s] has been created", r.ID, _rule.Name, ip)

pkg/ftp/ftp.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/li4n0/revsuit/internal/database"
 	"github.com/li4n0/revsuit/internal/qqwry"
+	"github.com/li4n0/revsuit/internal/recycler"
 	"github.com/li4n0/revsuit/internal/rule"
 	log "unknwon.dev/clog/v2"
 )
@@ -54,14 +55,19 @@ func (s *Server) updateRules() error {
 }
 
 func (s *Server) handleConnection(conn net.Conn) {
-	defer conn.Close()
+	defer func() {
+		_ = conn.Close()
+		if err := recover(); err != nil {
+			recycler.Recycle(err)
+		}
+	}()
 
 	if err := conn.SetDeadline(time.Now().Add(time.Second * 30)); err != nil {
-		log.Error("FTP set connection deadline error:%v", err)
+		log.Warn("FTP set connection deadline error:%v", err)
 	}
 
 	if _, err := conn.Write([]byte("220 (vsFTPd 3.0.2)\r\n")); err != nil {
-		log.Error("FTP write connection error:%v", err)
+		log.Warn("FTP write connection error:%v", err)
 	}
 
 	ip := strings.Split(conn.RemoteAddr().String(), ":")[0]
@@ -158,7 +164,7 @@ loop:
 		// create new record
 		r, err := NewRecord(_rule, flag, user, password, path, ip, area, status)
 		if err != nil {
-			log.Error("FTP record[rule_id:%d] created failed :%s", _rule.ID, err)
+			log.Warn("FTP record[rule_id:%d] created failed :%s", _rule.ID, err)
 			return
 		}
 		log.Info("FTP record[id:%d rule:%s remote_ip:%s] has been created", r.ID, _rule.Name, ip)
@@ -210,7 +216,7 @@ func (s *Server) Run() {
 		for {
 			tcpConn, err := listener.Accept()
 			if err != nil {
-				log.Error("FTP accept connection error: %v", err)
+				log.Warn("FTP accept connection error: %v", err)
 				continue
 			}
 			_ = tcpConn.Close()
@@ -220,7 +226,7 @@ func (s *Server) Run() {
 	for {
 		tcpConn, err := listener.Accept()
 		if err != nil {
-			log.Error("FTP accept connection error: %v", err)
+			log.Warn("FTP accept connection error: %v", err)
 			continue
 		}
 		go s.handleConnection(tcpConn)

pkg/mysql/mysql.go

@@ -97,7 +97,7 @@ func (s *Server) ConnectionClosed(c *vmysql.Conn) {
 
 	cr, ok := s.connRulePool.Load(c.ConnectionID)
 	if !ok {
-		log.Error("MySQL Connection rule(%d) not match flag", c.ConnectionID)
+		log.Warn("MySQL Connection rule(%d) not match flag", c.ConnectionID)
 		return
 	}
 
@@ -127,7 +127,7 @@ func (s *Server) ConnectionClosed(c *vmysql.Conn) {
 
 	r, err := newRecord(_rule, flag, user, clientName, clientOS, ip, qqwry.Area(ip), supportLoadLocalData, files)
 	if err != nil {
-		log.Error("MySQL record(rule_id:%s) created failed :%s", _rule.Name, err)
+		log.Warn("MySQL record(rule_id:%s) created failed :%s", _rule.Name, err)
 		return
 	}
 	log.Info("MySQL record[id:%d rule:%s remote_ip:%s] has been created", r.ID, _rule.Name, ip)
@@ -275,7 +275,7 @@ func (s *Server) Run() {
 	log.Info("Starting MySQL Server at %s", s.Addr)
 	s.listener, err = vmysql.NewListener("tcp", s.Addr, authServer, s, s.VersionString, 0, 0)
 	if err != nil {
-		log.Error("New MySQL Server failed: %s", err)
+		log.Warn("New MySQL Server failed: %s", err)
 		os.Exit(-1)
 	}
 

pkg/mysql/vmysql/auth_server.go

@@ -86,7 +86,7 @@ var authServers = make(map[string]AuthServer)
 // RegisterAuthServerImpl registers an implementations of AuthServer.
 func RegisterAuthServerImpl(name string, authServer AuthServer) {
 	if _, ok := authServers[name]; ok {
-		log.Error("AuthServer named %v already exists", name)
+		log.Warn("AuthServer named %v already exists", name)
 	}
 	authServers[name] = authServer
 }

pkg/mysql/vmysql/conn.go

@@ -344,7 +344,7 @@ func (c *Conn) readUploadFileEphemeralPacket() []byte {
 
 		_, err := io.ReadFull(r, data)
 		if err != nil {
-			log.Error("Error while reading data: %s", err)
+			log.Warn("Error while reading data: %s", err)
 			return nil
 		} else {
 			fileChunkData = append(fileChunkData, data)
@@ -718,7 +718,7 @@ func (c *Conn) handleNextCommand(handler Handler) error {
 		// case though, and very unlikely to happen,
 		// and the only downside is we log a bit more then.
 		if err != io.EOF {
-			log.Error("Error reading packet from %s: %v", c, err)
+			log.Warn("Error reading packet from %s: %v", c, err)
 		}
 		return err
 	}
@@ -732,7 +732,7 @@ func (c *Conn) handleNextCommand(handler Handler) error {
 		c.RecycleReadPacket()
 		c.SchemaName = db
 		if err := c.writeOKPacket(0, 0, c.StatusFlags, 0); err != nil {
-			log.Error("Error writing ComInitDB result to %s: %v", c, err)
+			log.Warn("Error writing ComInitDB result to %s: %v", c, err)
 			return err
 		}
 	case ComQuery:
@@ -749,10 +749,10 @@ func (c *Conn) handleNextCommand(handler Handler) error {
 		if c.Capabilities&CapabilityClientMultiStatements != 0 {
 			queries, err = sqlparser.SplitStatementToPieces(query)
 			if err != nil {
-				log.Error("Conn %v: Error splitting query: %v", c, err)
+				log.Warn("Conn %v: Error splitting query: %v", c, err)
 				if werr := c.writeErrorPacketFromError(err); werr != nil {
 					// If we can't even write the error, we're done.
-					log.Error("Conn %v: Error writing query error: %v", c, werr)
+					log.Warn("Conn %v: Error writing query error: %v", c, werr)
 					return werr
 				}
 			}
@@ -772,7 +772,7 @@ func (c *Conn) handleNextCommand(handler Handler) error {
 		timings.Record(queryTimingKey, queryStart)
 
 		if err := c.flush(); err != nil {
-			log.Error("Conn %v: Flush() failed: %v", c.ID(), err)
+			log.Warn("Conn %v: Flush() failed: %v", c.ID(), err)
 			return err
 		}
 
@@ -781,12 +781,12 @@ func (c *Conn) handleNextCommand(handler Handler) error {
 		// Return error if listener was shut down and OK otherwise
 		if c.listener.isShutdown() {
 			if err := c.writeErrorPacket(ERServerShutdown, SSServerShutdown, "Server shutdown in progress"); err != nil {
-				log.Error("Error writing ComPing error to %s: %v", c, err)
+				log.Warn("Error writing ComPing error to %s: %v", c, err)
 				return err
 			}
 		} else {
 			if err := c.writeOKPacket(0, 0, c.StatusFlags, 0); err != nil {
-				log.Error("Error writing ComPing result to %s: %v", c, err)
+				log.Warn("Error writing ComPing result to %s: %v", c, err)
 				return err
 			}
 		}
@@ -800,28 +800,28 @@ func (c *Conn) handleNextCommand(handler Handler) error {
 			case 1:
 				c.Capabilities &^= CapabilityClientMultiStatements
 			default:
-				log.Error("Got unhandled packet (ComSetOption default) from client %v, returning error: %v", c.ConnectionID, data)
+				log.Warn("Got unhandled packet (ComSetOption default) from client %v, returning error: %v", c.ConnectionID, data)
 				if err := c.writeErrorPacket(ERUnknownComError, SSUnknownComError, "error handling packet: %v", data); err != nil {
-					log.Error("Error writing error packet to client: %v", err)
+					log.Warn("Error writing error packet to client: %v", err)
 					return err
 				}
 			}
 			if err := c.writeEndResult(false, 0, 0, 0); err != nil {
-				log.Error("Error writeEndResult error %v ", err)
+				log.Warn("Error writeEndResult error %v ", err)
 				return err
 			}
 		} else {
-			log.Error("Got unhandled packet (ComSetOption else) from client %v, returning error: %v", c.ConnectionID, data)
+			log.Warn("Got unhandled packet (ComSetOption else) from client %v, returning error: %v", c.ConnectionID, data)
 			if err := c.writeErrorPacket(ERUnknownComError, SSUnknownComError, "error handling packet: %v", data); err != nil {
-				log.Error("Error writing error packet to client: %v", err)
+				log.Warn("Error writing error packet to client: %v", err)
 				return err
 			}
 		}
 	default:
-		log.Error("Got unhandled packet (default) from %s, returning error: %v", c, data)
+		log.Warn("Got unhandled packet (default) from %s, returning error: %v", c, data)
 		c.RecycleReadPacket()
 		if err := c.writeErrorPacket(ERUnknownComError, SSUnknownComError, "command handling not implemented yet: %v", data[0]); err != nil {
-			log.Error("Error writing error packet to %s: %s", c, err)
+			log.Warn("Error writing error packet to %s: %s", c, err)
 			return err
 		}
 	}
@@ -872,7 +872,7 @@ func (c *Conn) execQuery(query string, handler Handler, more bool) error {
 		// was a read operation.
 		if !sendFinished {
 			if err := c.writeEndResult(more, 0, 0, handler.WarningCount(c)); err != nil {
-				log.Error("Error writing result to %s: %v", c, err)
+				log.Warn("Error writing result to %s: %v", c, err)
 				return err
 			}
 		}

pkg/mysql/vmysql/server.go

@@ -23,6 +23,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/li4n0/revsuit/internal/recycler"
 	log "unknwon.dev/clog/v2"
 	"vitess.io/vitess/go/netutil"
 	"vitess.io/vitess/go/sqltypes"
@@ -222,6 +223,11 @@ func (l *Listener) Addr() net.Addr {
 
 // Accept runs an accept loop until the listener is closed.
 func (l *Listener) Accept() {
+	defer func() {
+		if err := recover(); err != nil {
+			recycler.Recycle(err)
+		}
+	}()
 	for {
 		conn, err := l.listener.Accept()
 		if err != nil {
@@ -253,7 +259,7 @@ func (l *Listener) handle(conn net.Conn, connectionID uint32, acceptTime time.Ti
 	// Catch panics, and close the connection in any case.
 	defer func() {
 		if x := recover(); x != nil {
-			log.Error("mysql_server caught panic:\n%v\n%s", x, tb.Stack(4))
+			log.Warn("mysql_server caught panic:\n%v\n%s", x, tb.Stack(4))
 		}
 		// We call flush here in case there's a premature return after
 		// startWriterBuffering is called
@@ -269,7 +275,7 @@ func (l *Listener) handle(conn net.Conn, connectionID uint32, acceptTime time.Ti
 	salt, err := c.writeHandshakeV10(l.ServerVersion, l.authServer, l.TLSConfig != nil)
 	if err != nil {
 		if err != io.EOF {
-			log.Error("Cannot send HandshakeV10 packet to %s: %v", c, err)
+			log.Warn("Cannot send HandshakeV10 packet to %s: %v", c, err)
 		}
 		return
 	}
@@ -280,13 +286,13 @@ func (l *Listener) handle(conn net.Conn, connectionID uint32, acceptTime time.Ti
 	if err != nil {
 		// Don't log EOF errors. They cause too much spam, same as main read loop.
 		if err != io.EOF {
-			log.Error("Cannot read client handshake response from %s: %v", c, err)
+			log.Warn("Cannot read client handshake response from %s: %v", c, err)
 		}
 		return
 	}
 	user, authMethod, authResponse, err := l.parseClientHandshakePacket(c, true, response)
 	if err != nil {
-		log.Error("Cannot parse client handshake response from %s: %v", c, err)
+		log.Warn("Cannot parse client handshake response from %s: %v", c, err)
 		return
 	}
 
@@ -299,14 +305,14 @@ func (l *Listener) handle(conn net.Conn, connectionID uint32, acceptTime time.Ti
 		// SSL was enabled. We need to re-read the auth packet.
 		response, err = c.readEphemeralPacket()
 		if err != nil {
-			log.Error("Cannot read post-SSL client handshake response from %s: %v", c, err)
+			log.Warn("Cannot read post-SSL client handshake response from %s: %v", c, err)
 			return
 		}
 
 		// Returns copies of the data, so we can recycle the buffer.
 		user, authMethod, authResponse, err = l.parseClientHandshakePacket(c, false, response)
 		if err != nil {
-			log.Error("Cannot parse post-SSL client handshake response from %s: %v", c, err)
+			log.Warn("Cannot parse post-SSL client handshake response from %s: %v", c, err)
 			return
 		}
 		c.RecycleReadPacket()
@@ -360,13 +366,13 @@ func (l *Listener) handle(conn net.Conn, connectionID uint32, acceptTime time.Ti
 		data := make([]byte, 21) //nolint:ineffassign,staticcheck // SA4006 This line is required because the binary protocol requires padding with 0
 		data = append(salt, byte(0x00))
 		if err := c.writeAuthSwitchRequest(MysqlNativePassword, data); err != nil {
-			log.Error("Error writing auth switch packet for %s: %v", c, err)
+			log.Warn("Error writing auth switch packet for %s: %v", c, err)
 			return
 		}
 
 		response, err := c.readEphemeralPacket()
 		if err != nil {
-			log.Error("Error reading auth switch response for %s: %v", c, err)
+			log.Warn("Error reading auth switch response for %s: %v", c, err)
 			return
 		}
 		c.RecycleReadPacket()
@@ -396,7 +402,7 @@ func (l *Listener) handle(conn net.Conn, connectionID uint32, acceptTime time.Ti
 			data = authServerDialogSwitchData()
 		}
 		if err := c.writeAuthSwitchRequest(authServerMethod, data); err != nil {
-			log.Error("Error writing auth switch packet for %s: %v", c, err)
+			log.Warn("Error writing auth switch packet for %s: %v", c, err)
 			return
 		}
 
@@ -418,7 +424,7 @@ func (l *Listener) handle(conn net.Conn, connectionID uint32, acceptTime time.Ti
 
 	// Negotiation worked, send OK packet.
 	if err := c.writeOKPacket(0, 0, c.StatusFlags, 0); err != nil {
-		log.Error("Cannot write OK packet to %s: %v", c, err)
+		log.Warn("Cannot write OK packet to %s: %v", c, err)
 		return
 	}
 

pkg/rhttp/http.go

@@ -88,7 +88,7 @@ func (s *Server) updateRules() error {
 
 func (s *Server) Run() {
 	if err := s.updateRules(); err != nil {
-		log.Error(err.Error())
+		log.Warn(err.Error())
 	}
 	log.Info("Starting HTTP Server at %s, token:%s", s.Addr, s.Token)
 	err := s.Router.Run(s.Addr)
@@ -152,7 +152,7 @@ func (s *Server) Receive(c *gin.Context) {
 		// create new record
 		r, err := NewRecord(_rule, flag, c.Request.Method, u, ip, area, string(raw))
 		if err != nil {
-			log.Error("HTTP record[rule_id:%d] created failed :%s", _rule.ID, err)
+			log.Warn("HTTP record[rule_id:%d] created failed :%s", _rule.ID, err)
 			code, err := strconv.Atoi(compileTpl(c, _rule.ResponseStatusCode, vars))
 			if err != nil || code < 100 || code > 600 {
 				code = 400

pkg/rmi/rmi.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/li4n0/revsuit/internal/database"
 	"github.com/li4n0/revsuit/internal/qqwry"
+	"github.com/li4n0/revsuit/internal/recycler"
 	log "unknwon.dev/clog/v2"
 )
 
@@ -46,10 +47,15 @@ func (s *Server) updateRules() error {
 }
 
 func (s *Server) handleConnection(conn net.Conn) {
-	defer conn.Close()
+	defer func() {
+		_ = conn.Close()
+		if err := recover(); err != nil {
+			recycler.Recycle(err)
+		}
+	}()
 
 	if err := conn.SetDeadline(time.Now().Add(time.Second * 30)); err != nil {
-		log.Error("RMI set connection deadline error:%v", err)
+		log.Warn("RMI set connection deadline error:%v", err)
 	}
 
 	ip, port, _ := net.SplitHostPort(conn.RemoteAddr().String())
@@ -57,7 +63,7 @@ func (s *Server) handleConnection(conn net.Conn) {
 	buf := make([]byte, 1024)
 	_, err := conn.Read(buf)
 	if err != nil {
-		log.Error("RMI read connection error:%v", err)
+		log.Warn("RMI read connection error:%v", err)
 	}
 
 	if !bytes.Contains(buf, []byte{0x4a, 0x52, 0x4d, 0x49}) {
@@ -77,7 +83,7 @@ func (s *Server) handleConnection(conn net.Conn) {
 
 	_, err = conn.Write(send)
 	if err != nil {
-		log.Error("RMI write connection error: %v", err)
+		log.Warn("RMI write connection error: %v", err)
 	}
 
 	data := make([]byte, 512)
@@ -85,7 +91,7 @@ func (s *Server) handleConnection(conn net.Conn) {
 	for length := 0; length < 50; {
 		n, err := conn.Read(data)
 		if err != nil {
-			log.Error("RMI read connection error: %v", err)
+			log.Warn("RMI read connection error: %v", err)
 		}
 		length += n
 	}
@@ -104,7 +110,7 @@ func (s *Server) handleConnection(conn net.Conn) {
 		// create new record
 		r, err := NewRecord(_rule, flag, path, ip, area)
 		if err != nil {
-			log.Error("RMI record[rule_id:%d] created failed :%s", _rule.ID, err)
+			log.Warn("RMI record[rule_id:%d] created failed :%s", _rule.ID, err)
 			return
 		}
 		log.Info("RMI record[id:%d rule:%s remote_ip:%s] has been created", r.ID, _rule.Name, ip)
@@ -149,7 +155,7 @@ func (s *Server) Run() {
 	for {
 		tcpConn, err := listener.Accept()
 		if err != nil {
-			log.Error("RMI accept connection error: %v", err)
+			log.Warn("RMI accept connection error: %v", err)
 			continue
 		}
 		go s.handleConnection(tcpConn)

pkg/server/controller.go

@@ -1,11 +1,18 @@
 package server
 
 import (
+	"fmt"
 	"io"
+	"net"
 	"net/http"
+	"net/http/httputil"
+	"os"
+	"strings"
+	"time"
 
 	"github.com/gin-gonic/gin"
 	"github.com/li4n0/revsuit/internal/record"
+	"github.com/li4n0/revsuit/internal/recycler"
 	log "unknwon.dev/clog/v2"
 )
 
@@ -33,3 +40,48 @@ func events(c *gin.Context) {
 	})
 	log.Info(c.Request.RemoteAddr, "disconnect")
 }
+
+func recovery(c *gin.Context) {
+	timeFormat := func(t time.Time) string {
+		var timeString = t.Format("2006/01/02 - 15:04:05")
+		return timeString
+	}
+	defer func() {
+		if err := recover(); err != nil {
+			// Check for a broken connection, as it is not really a
+			// condition that warrants a panic stack trace.
+			var brokenPipe bool
+			if ne, ok := err.(*net.OpError); ok {
+				if se, ok := ne.Err.(*os.SyscallError); ok {
+					if strings.Contains(strings.ToLower(se.Error()), "broken pipe") || strings.Contains(strings.ToLower(se.Error()), "connection reset by peer") {
+						brokenPipe = true
+					}
+				}
+			}
+			httpRequest, _ := httputil.DumpRequest(c.Request, false)
+			headers := strings.Split(string(httpRequest), "\r\n")
+			for idx, header := range headers {
+				current := strings.Split(header, ":")
+				if current[0] == "Authorization" {
+					headers[idx] = current[0] + ": *"
+				}
+			}
+			if brokenPipe {
+				recycler.Recycle(fmt.Sprintf("%s\n%s", err, string(httpRequest)))
+			} else if gin.IsDebugging() {
+				recycler.Recycle(fmt.Sprintf("[Recovery] %s panic recovered:\n%s\n%s\n", timeFormat(time.Now()), strings.Join(headers, "\r\n"), err))
+			} else {
+				recycler.Recycle(fmt.Sprintf("[Recovery] %s panic recovered:\n%s\n", timeFormat(time.Now()), err))
+			}
+
+			// If the connection is dead, we can't write a status to it.
+			if brokenPipe {
+				_ = c.Error(err.(error)) // nolint: errcheck
+				c.Abort()
+			} else {
+				c.AbortWithStatus(http.StatusInternalServerError)
+			}
+		}
+	}()
+	c.Next()
+}

pkg/server/router.go

@@ -15,10 +15,10 @@ import (
 )
 
 func (revsuit *Revsuit) registerRouter() {
-	revsuit.http.Router = gin.Default()
-	if revsuit.logLevel != log.LevelTrace {
-		revsuit.http.Router = gin.New()
-		revsuit.http.Router.Use(gin.Recovery())
+	revsuit.http.Router = gin.New()
+	revsuit.http.Router.Use(recovery)
+	if revsuit.logLevel == log.LevelTrace {
+		revsuit.http.Router.Use(gin.Logger())
 	}
 
 	revsuit.registerPlatformRouter()