fix race

Signed-off-by: Ryan Leung <rleungx@gmail.com>

errors.toml

@@ -16,6 +16,11 @@ error = '''
 redirect failed
 '''
 
+["PD:apiutil:ErrRedirectToNotLeader"]
+error = '''
+redirect to not leader
+'''
+
 ["PD:autoscaling:ErrEmptyMetricsResponse"]
 error = '''
 metrics response from Prometheus is empty

pkg/errs/errno.go

@@ -197,6 +197,8 @@ var (
 var (
 	ErrRedirect       = errors.Normalize("redirect failed", errors.RFCCodeText("PD:apiutil:ErrRedirect"))
 	ErrOptionNotExist = errors.Normalize("the option %s does not exist", errors.RFCCodeText("PD:apiutil:ErrOptionNotExist"))
+	// ErrRedirectToNotLeader is the error message for redirect to not leader.
+	ErrRedirectToNotLeader = errors.Normalize("redirect to not leader", errors.RFCCodeText("PD:apiutil:ErrRedirectToNotLeader"))
 )
 
 // grpcutil errors

pkg/utils/apiutil/apiutil.go

@@ -61,11 +61,6 @@ const (
 	// ForwardToMicroServiceHeader is used to mark the request is forwarded to micro service.
 	ForwardToMicroServiceHeader = "Forward-To-Micro-Service"
 
-	// ErrRedirectFailed is the error message for redirect failed.
-	ErrRedirectFailed = "redirect failed"
-	// ErrRedirectToNotLeader is the error message for redirect to not leader.
-	ErrRedirectToNotLeader = "redirect to not leader"
-
 	chunkSize = 4096
 )
 
@@ -459,7 +454,7 @@ func (p *customReverseProxies) ServeHTTP(w http.ResponseWriter, r *http.Request)
 		}
 		return
 	}
-	http.Error(w, ErrRedirectFailed, http.StatusInternalServerError)
+	http.Error(w, errs.ErrRedirect.FastGenByArgs().Error(), http.StatusInternalServerError)
 }
 
 // copyHeader duplicates the HTTP headers from the source `src` to the destination `dst`.

pkg/utils/apiutil/serverapi/middleware.go

@@ -162,15 +162,21 @@ func (h *redirector) matchMicroServiceRedirectRules(r *http.Request) (bool, stri
 func (h *redirector) ServeHTTP(w http.ResponseWriter, r *http.Request, next http.HandlerFunc) {
 	redirectToMicroService, targetAddr := h.matchMicroServiceRedirectRules(r)
 	allowFollowerHandle := len(r.Header.Get(apiutil.PDAllowFollowerHandleHeader)) > 0
-	if !h.s.IsClosed() && (allowFollowerHandle || h.s.GetMember().IsLeader()) && !redirectToMicroService {
+
+	if h.s.IsClosed() {
+		http.Error(w, errs.ErrServerNotStarted.FastGenByArgs().Error(), http.StatusInternalServerError)
+		return
+	}
+
+	if (allowFollowerHandle || h.s.GetMember().IsLeader()) && !redirectToMicroService {
 		next(w, r)
 		return
 	}
 
 	// Prevent more than one redirection.
 	if name := r.Header.Get(apiutil.PDRedirectorHeader); len(name) != 0 {
 		log.Error("redirect but server is not leader", zap.String("from", name), zap.String("server", h.s.Name()), errs.ZapError(errs.ErrRedirect))
-		http.Error(w, apiutil.ErrRedirectToNotLeader, http.StatusInternalServerError)
+		http.Error(w, errs.ErrRedirectToNotLeader.FastGenByArgs().Error(), http.StatusInternalServerError)
 		return
 	}
 
@@ -189,7 +195,7 @@ func (h *redirector) ServeHTTP(w http.ResponseWriter, r *http.Request, next http
 	var clientUrls []string
 	if redirectToMicroService {
 		if len(targetAddr) == 0 {
-			http.Error(w, apiutil.ErrRedirectFailed, http.StatusInternalServerError)
+			http.Error(w, errs.ErrRedirect.FastGenByArgs().Error(), http.StatusInternalServerError)
 			return
 		}
 		clientUrls = append(clientUrls, targetAddr)

server/apiv2/middlewares/redirector.go

@@ -30,8 +30,13 @@ import (
 func Redirector() gin.HandlerFunc {
 	return func(c *gin.Context) {
 		svr := c.MustGet(ServerContextKey).(*server.Server)
+
+		if svr.IsClosed() {
+			c.AbortWithStatusJSON(http.StatusInternalServerError, errs.ErrServerNotStarted.FastGenByArgs().Error())
+			return
+		}
 		allowFollowerHandle := len(c.Request.Header.Get(apiutil.PDAllowFollowerHandleHeader)) > 0
-		if !svr.IsClosed() && (allowFollowerHandle || svr.GetMember().IsLeader()) {
+		if allowFollowerHandle || svr.GetMember().IsLeader() {
 			c.Next()
 			return
 		}
@@ -45,12 +50,11 @@ func Redirector() gin.HandlerFunc {
 
 		c.Request.Header.Set(apiutil.PDRedirectorHeader, svr.Name())
 
-		leader := svr.GetMember().GetLeader()
-		if leader == nil {
+		if svr.GetMember().GetLeader() == nil {
 			c.AbortWithStatusJSON(http.StatusServiceUnavailable, errs.ErrLeaderNil.FastGenByArgs().Error())
 			return
 		}
-		clientUrls := leader.GetClientUrls()
+		clientUrls := svr.GetMember().GetLeader().GetClientUrls()
 		urls := make([]url.URL, 0, len(clientUrls))
 		for _, item := range clientUrls {
 			u, err := url.Parse(item)