pd, openapi(ticdc): retry one reset by peer when kill the pd instance, fix healthy panic

cdc/api/v1/api.go

@@ -874,7 +874,15 @@ func (h *OpenAPI) ServerStatus(c *gin.Context) {
 func (h *OpenAPI) Health(c *gin.Context) {
 	ctx := c.Request.Context()
 
-	health, err := h.statusProvider().IsHealthy(ctx)
+	var err error
+	provider := h.statusProvider()
+	if provider == nil {
+		err = cerror.ErrOwnerNotFound.FastGenByArgs()
+		c.JSON(http.StatusInternalServerError, model.NewHTTPError(err))
+		return
+	}
+
+	health, err := provider.IsHealthy(ctx)
 	if err != nil {
 		c.IndentedJSON(http.StatusInternalServerError, model.NewHTTPError(err))
 		return

cdc/api/v1/api_test.go

@@ -896,11 +896,21 @@ func TestHealth(t *testing.T) {
 	// capture is owner
 	ctrl := gomock.NewController(t)
 	cp := mock_capture.NewMockCapture(ctrl)
-	sp := mock_owner.NewMockStatusProvider(ctrl)
-	ownerRouter := newRouter(cp, sp)
-	api := testCase{url: "/api/v1/health", method: "GET"}
+	ownerRouter := newRouter(cp, nil)
 
 	cp.EXPECT().IsReady().Return(true).AnyTimes()
+	cp.EXPECT().StatusProvider().Return(nil).Times(1)
+
+	api := testCase{url: "/api/v1/health", method: "GET"}
+
+	w := httptest.NewRecorder()
+	req, _ := http.NewRequestWithContext(context.Background(), api.method, api.url, nil)
+	ownerRouter.ServeHTTP(w, req)
+	require.Equal(t, 500, w.Code)
+
+	sp := mock_owner.NewMockStatusProvider(ctrl)
+	ownerRouter = newRouter(cp, sp)
+
 	cp.EXPECT().Info().DoAndReturn(func() (model.CaptureInfo, error) {
 		return model.CaptureInfo{}, nil
 	}).AnyTimes()
@@ -911,8 +921,8 @@ func TestHealth(t *testing.T) {
 	// IsHealthy returns error.
 	isHealthError := sp.EXPECT().IsHealthy(gomock.Any()).
 		Return(false, cerror.ErrOwnerNotFound)
-	w := httptest.NewRecorder()
-	req, _ := http.NewRequestWithContext(context.Background(), api.method, api.url, nil)
+	w = httptest.NewRecorder()
+	req, _ = http.NewRequestWithContext(context.Background(), api.method, api.url, nil)
 	ownerRouter.ServeHTTP(w, req)
 	require.Equal(t, 500, w.Code)
 

pkg/errorutil/ignore.go

@@ -84,5 +84,10 @@ func IsRetryableEtcdError(err error) bool {
 	if strings.Contains(etcdErr.Error(), "received prior goaway: code: NO_ERROR") {
 		return true
 	}
+
+	// this may happen if the PD instance shutdown by `kill -9`, no matter the instance is the leader or not.
+	if strings.Contains(etcdErr.Error(), "connection reset by peer") {
+		return true
+	}
 	return false
 }

pkg/errorutil/ignore_test.go

@@ -63,6 +63,8 @@ func TestIsRetryableEtcdError(t *testing.T) {
 		{errors.New("rpc error: code = Unavailable desc = closing transport due to: " +
 			"connection error: desc = \\\"error reading from server: EOF\\\", " +
 			"received prior goaway: code: NO_ERROR\""), true},
+		{errors.New("rpc error: code = Unavailable desc = error reading from server: " +
+			"xxx: read: connection reset by peer"), true},
 	}
 
 	for _, item := range cases {