http_server: add http handler for failpoint (#1722) (#1732)

cdc/http_status.go

@@ -23,7 +23,10 @@ import (
 	"net/http/pprof"
 	"os"
 
+	"github.com/pingcap/ticdc/pkg/util"
+
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/cdc/kv"
 	"github.com/pingcap/ticdc/pkg/config"
@@ -54,6 +57,11 @@ func (s *Server) startStatusHTTP() error {
 
 	serverMux.HandleFunc("/admin/log", handleAdminLogLevel)
 
+	if util.FailpointBuild {
+		// `http.StripPrefix` is needed because `failpoint.HttpHandler` assumes that it handles the prefix `/`.
+		serverMux.Handle("/debug/fail/", http.StripPrefix("/debug/fail", &failpoint.HttpHandler{}))
+	}
+
 	prometheus.DefaultGatherer = registry
 	serverMux.Handle("/metrics", promhttp.Handler())
 	conf := config.GetGlobalServerConfig()

cdc/http_status_test.go

@@ -14,13 +14,15 @@
 package cdc
 
 import (
+	"bytes"
 	"fmt"
 	"io/ioutil"
 	"net/http"
 	"net/url"
 	"time"
 
 	"github.com/pingcap/check"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/ticdc/pkg/config"
 	cerror "github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/util/testleak"
@@ -72,6 +74,7 @@ func (s *httpStatusSuite) TestHTTPStatus(c *check.C) {
 	testHandleRebalance(c)
 	testHandleMoveTable(c)
 	testHandleChangefeedQuery(c)
+	testHandleFailpoint(c)
 }
 
 func testPprof(c *check.C) {
@@ -132,3 +135,37 @@ func testRequestNonOwnerFailed(c *check.C, uri string) {
 	c.Assert(resp.StatusCode, check.Equals, http.StatusBadRequest)
 	c.Assert(string(data), check.Equals, concurrency.ErrElectionNotLeader.Error())
 }
+
+func testHandleFailpoint(c *check.C) {
+	fp := "github.com/pingcap/ticdc/cdc/TestHandleFailpoint"
+	uri := fmt.Sprintf("http://%s/debug/fail/%s", advertiseAddr4Test, fp)
+	body := bytes.NewReader([]byte("return(true)"))
+	req, err := http.NewRequest("PUT", uri, body)
+	c.Assert(err, check.IsNil)
+
+	resp, err := http.DefaultClient.Do(req)
+	c.Assert(err, check.IsNil)
+	defer resp.Body.Close()
+	c.Assert(resp.StatusCode, check.GreaterEqual, 200)
+	c.Assert(resp.StatusCode, check.Less, 300)
+
+	failpointHit := false
+	failpoint.Inject("TestHandleFailpoint", func() {
+		failpointHit = true
+	})
+	c.Assert(failpointHit, check.IsTrue)
+
+	req, err = http.NewRequest("DELETE", uri, body)
+	c.Assert(err, check.IsNil)
+	resp, err = http.DefaultClient.Do(req)
+	c.Assert(err, check.IsNil)
+	defer resp.Body.Close()
+	c.Assert(resp.StatusCode, check.GreaterEqual, 200)
+	c.Assert(resp.StatusCode, check.Less, 300)
+
+	failpointHit = false
+	failpoint.Inject("TestHandleFailpoint", func() {
+		failpointHit = true
+	})
+	c.Assert(failpointHit, check.IsFalse)
+}