sqlreplay: do not block replaying when logging or reporting hangs (#1000)

Author: djshow832

pkg/server/api/traffic.go

@@ -155,11 +155,14 @@ func (h *Server) TrafficReplay(c *gin.Context) {
 	}
 	cfg.OutputPath = c.PostForm("outputpath")
 	cfg.Addr = c.PostForm("addr")
+	h.lg.Info("request: traffic replay", zap.Any("cfg", cfg))
 
 	if err := h.mgr.ReplayJobMgr.StartReplay(cfg); err != nil {
 		c.String(http.StatusInternalServerError, err.Error())
+		h.lg.Info("response: traffic replay", zap.Error(err))
 		return
 	}
+	h.lg.Info("response: traffic replay")
 	c.String(http.StatusOK, "replay started")
 }
 
@@ -181,7 +184,9 @@ func (h *Server) TrafficCancel(c *gin.Context) {
 		}
 	}
 	cfg.Graceful = strings.EqualFold(c.PostForm("graceful"), "true")
+	h.lg.Info("request: traffic cancel", zap.Any("cfg", cfg))
 	result := h.mgr.ReplayJobMgr.Stop(cfg)
+	h.lg.Info("response: traffic cancel", zap.String("result", result))
 	c.String(http.StatusOK, result)
 }
 

pkg/sqlreplay/conn/conn.go

@@ -191,17 +191,15 @@ func (c *conn) Run(ctx context.Context) {
 			if curDB != command.Value.CurDB && command.Value.CurDB != "" {
 				// Maybe there's a USE statement already, never mind.
 				if resp := c.backendConn.ExecuteCmd(ctx, pnet.MakeInitDBRequest(command.Value.CurDB)); resp.Err != nil {
-					c.replayStats.ExceptionCmds.Add(1)
-					c.exceptionCh <- NewFailException(resp.Err, command.Value)
+					c.onExecuteFailed(command.Value, resp.Err)
 					c.lg.Info("failed to use database", zap.String("db", command.Value.CurDB), zap.Error(resp.Err))
 					continue
 				}
 				curDB = command.Value.CurDB
 			}
 			// update psID, SQL, and params for the command.
 			if err := c.updateExecuteStmt(command.Value); err != nil {
-				c.replayStats.ExceptionCmds.Add(1)
-				c.exceptionCh <- NewFailException(err, command.Value)
+				c.onExecuteFailed(command.Value, err)
 				continue
 			}
 			startTime := time.Now()
@@ -215,16 +213,20 @@ func (c *conn) Run(ctx context.Context) {
 					curDB = ""
 					continue
 				}
-				c.replayStats.ExceptionCmds.Add(1)
-				c.exceptionCh <- NewFailException(resp.Err, command.Value)
+				c.onExecuteFailed(command.Value, resp.Err)
 			} else {
 				c.updatePreparedStmts(command.Value.CapturedPsID, command.Value.Payload, resp)
 			}
-			c.execInfoCh <- ExecInfo{
+			// Logging should not block replaying.
+			select {
+			case c.execInfoCh <- ExecInfo{
 				Command:   command.Value,
 				StartTime: startTime,
 				CostTime:  latency,
+			}:
+			default:
 			}
+
 			c.replayStats.ReplayedCmds.Add(1)
 		}
 	}
@@ -340,11 +342,24 @@ func (c *conn) updateExecuteStmt(command *cmd.Command) error {
 
 func (c *conn) onDisconnected(err error) {
 	c.replayStats.ExceptionCmds.Add(1)
-	c.exceptionCh <- NewOtherException(err, c.upstreamConnID)
+	// Reporting should not block replaying.
+	select {
+	case c.exceptionCh <- NewOtherException(err, c.upstreamConnID):
+	default:
+	}
 	clear(c.psIDMapping)
 	clear(c.preparedStmts)
 }
 
+func (c *conn) onExecuteFailed(command *cmd.Command, err error) {
+	c.replayStats.ExceptionCmds.Add(1)
+	// Reporting should not block replaying.
+	select {
+	case c.exceptionCh <- NewFailException(err, command):
+	default:
+	}
+}
+
 // ExecuteCmd executes a command asynchronously by adding it to the list.
 // Adding commands should never block because it may cause cycle wait, so we don't use channels.
 // Conn A: wait for the lock held by conn B, and then its list becomes full and blocks the replay

pkg/sqlreplay/conn/conn_test.go

@@ -822,3 +822,23 @@ func TestExecInfo(t *testing.T) {
 		}
 	}
 }
+
+func TestNoBlockReplay(t *testing.T) {
+	exceptionCh, closeCh, execInfoCh := make(chan Exception, 1), make(chan uint64, 1), make(chan ExecInfo, 1)
+	stats := &ReplayStats{}
+	conn := NewConn(zap.NewNop(), ConnOpts{
+		Username:       "u1",
+		IdMgr:          id.NewIDManager(),
+		ConnID:         1,
+		UpstreamConnID: 555,
+		BcConfig:       &backend.BCConfig{},
+		ExceptionCh:    exceptionCh,
+		CloseCh:        closeCh,
+		ExecInfoCh:     execInfoCh,
+		ReplayStats:    stats,
+	})
+	for range 100 {
+		conn.onDisconnected(errors.New("conn err"))
+		conn.onExecuteFailed(&cmd.Command{}, errors.New("exec err"))
+	}
+}

pkg/sqlreplay/replay/replay.go

@@ -903,6 +903,7 @@ func (r *replay) recordExecInfoLoop() {
 	}
 }
 
+// stop is called when replaying commands fails or finishes, so no connection should be running.
 func (r *replay) stop(err error) {
 	r.Lock()
 	defer r.Unlock()