pump/* add a api to get binlog by ts (#449) (#455)

* pump/* add a api to get binlog by ts

pump/server.go

@@ -7,6 +7,7 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -400,6 +401,7 @@ func (s *Server) Start() error {
 	router.HandleFunc("/status", s.Status).Methods("GET")
 	router.HandleFunc("/state/{nodeID}/{action}", s.ApplyAction).Methods("PUT")
 	router.HandleFunc("/drainers", s.AllDrainers).Methods("GET")
+	router.HandleFunc("/debug/binlog/{ts}", s.BinlogByTS).Methods("GET")
 	http.Handle("/", router)
 	prometheus.DefaultGatherer = registry
 	http.Handle("/metrics", prometheus.Handler())
@@ -593,6 +595,31 @@ func (s *Server) Status(w http.ResponseWriter, r *http.Request) {
 	s.PumpStatus().Status(w, r)
 }
 
+// BinlogByTS exposes api get get binlog by ts
+func (s *Server) BinlogByTS(w http.ResponseWriter, r *http.Request) {
+	tsStr := mux.Vars(r)["ts"]
+	ts, err := strconv.ParseInt(tsStr, 10, 64)
+	if err != nil {
+		w.Write([]byte(fmt.Sprintf("invalid parameter ts: %s", tsStr)))
+		return
+	}
+
+	binlog, err := s.storage.GetBinlog(ts)
+	if err != nil {
+		w.Write([]byte(err.Error()))
+		return
+	}
+
+	w.Write([]byte(binlog.String()))
+	if len(binlog.PrewriteValue) > 0 {
+		prewriteValue := new(pb.PrewriteValue)
+		prewriteValue.Unmarshal(binlog.PrewriteValue)
+
+		w.Write([]byte("\n\n PrewriteValue: \n"))
+		w.Write([]byte(prewriteValue.String()))
+	}
+}
+
 // PumpStatus returns all pumps' status.
 func (s *Server) PumpStatus() *HTTPStatus {
 	status, err := s.node.NodesStatus(s.ctx)

pump/storage/storage.go

@@ -50,6 +50,9 @@ type Storage interface {
 
 	MaxCommitTS() int64
 
+	// GetBinlog return the binlog of ts
+	GetBinlog(ts int64) (binlog *pb.Binlog, err error)
+
 	// PullCommitBinlog return the chan to consume the binlog
 	PullCommitBinlog(ctx context.Context, last int64) <-chan []byte
 
@@ -427,6 +430,10 @@ func (a *Append) resolve(startTS int64) bool {
 	return false
 }
 
+func (a *Append) GetBinlog(ts int64) (*pb.Binlog, error) {
+	return a.readBinlogByTS(ts)
+}
+
 func (a *Append) readBinlogByTS(ts int64) (*pb.Binlog, error) {
 	var vp valuePointer