Merge branch 'feature/add-region-max-size' of github.com:bufferflies/pd into feature/add-region-max-size

Author: bufferflies

client/Makefile

@@ -16,6 +16,8 @@ GO_TOOLS_BIN_PATH := $(shell pwd)/../.tools/bin
 PATH := $(GO_TOOLS_BIN_PATH):$(PATH)
 SHELL := env PATH='$(PATH)' GOBIN='$(GO_TOOLS_BIN_PATH)' $(shell which bash)
 
+default: static tidy test
+
 test:
 	CGO_ENABLE=1 go test -race -cover
 

pd.code-workspace

@@ -0,0 +1,11 @@
+{
+	"folders": [
+		{
+			"path": "."
+		},
+		{
+			"path": "client"
+		}
+	],
+	"settings": {}
+}

pkg/mock/mockcluster/mockcluster.go

@@ -80,7 +80,7 @@ func (mc *Cluster) GetOpts() *config.PersistOptions {
 	return mc.PersistOptions
 }
 
-// GetConfig returns the cluster immutable configuration.
+// GetImmutableCfg returns the cluster immutable configuration.
 func (mc *Cluster) GetImmutableCfg() *config.ImmutableConfig {
 	return mc.ImmutableConfig
 }

server/api/operator.go

@@ -17,6 +17,7 @@ package api
 import (
 	"net/http"
 	"strconv"
+	"time"
 
 	"github.com/gorilla/mux"
 	"github.com/tikv/pd/pkg/apiutil"
@@ -344,6 +345,32 @@ func (h *operatorHandler) Delete(w http.ResponseWriter, r *http.Request) {
 	h.r.JSON(w, http.StatusOK, "The pending operator is canceled.")
 }
 
+// @Tags operator
+// @Summary lists the finished operators since the given timestamp in second.
+// @Param from query integer false "From Unix timestamp"
+// @Produce json
+// @Success 200 {object} []operator.OpRecord
+// @Failure 400 {string} string "The request is invalid."
+// @Failure 500 {string} string "PD server failed to proceed the request."
+// @Router /operators/records [get]
+func (h *operatorHandler) Records(w http.ResponseWriter, r *http.Request) {
+	var from time.Time
+	if fromStr := r.URL.Query()["from"]; len(fromStr) > 0 {
+		fromInt, err := strconv.ParseInt(fromStr[0], 10, 64)
+		if err != nil {
+			h.r.JSON(w, http.StatusBadRequest, err.Error())
+			return
+		}
+		from = time.Unix(fromInt, 0)
+	}
+	records, err := h.GetRecords(from)
+	if err != nil {
+		h.r.JSON(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	h.r.JSON(w, http.StatusOK, records)
+}
+
 func parseStoreIDsAndPeerRole(ids interface{}, roles interface{}) (map[uint64]placement.PeerRoleType, bool) {
 	items, ok := ids.([]interface{})
 	if !ok {

server/api/operator_test.go

@@ -19,7 +19,9 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"strconv"
 	"strings"
+	"time"
 
 	. "github.com/pingcap/check"
 	"github.com/pingcap/failpoint"
@@ -79,6 +81,9 @@ func (s *testOperatorSuite) TestAddRemovePeer(c *C) {
 	regionURL := fmt.Sprintf("%s/operators/%d", s.urlPrefix, region.GetId())
 	operator := mustReadURL(c, regionURL)
 	c.Assert(strings.Contains(operator, "operator not found"), IsTrue)
+	recordURL := fmt.Sprintf("%s/operators/records?from=%s", s.urlPrefix, strconv.FormatInt(time.Now().Unix(), 10))
+	records := mustReadURL(c, recordURL)
+	c.Assert(strings.Contains(records, "operator not found"), IsTrue)
 
 	mustPutStore(c, s.svr, 3, metapb.StoreState_Up, nil)
 	err := postJSON(testDialClient, fmt.Sprintf("%s/operators", s.urlPrefix), []byte(`{"name":"add-peer", "region_id": 1, "store_id": 3}`))
@@ -89,6 +94,8 @@ func (s *testOperatorSuite) TestAddRemovePeer(c *C) {
 
 	_, err = doDelete(testDialClient, regionURL)
 	c.Assert(err, IsNil)
+	records = mustReadURL(c, recordURL)
+	c.Assert(strings.Contains(records, "admin-add-peer {add peer: store [3]}"), IsTrue)
 
 	err = postJSON(testDialClient, fmt.Sprintf("%s/operators", s.urlPrefix), []byte(`{"name":"remove-peer", "region_id": 1, "store_id": 2}`))
 	c.Assert(err, IsNil)
@@ -98,6 +105,8 @@ func (s *testOperatorSuite) TestAddRemovePeer(c *C) {
 
 	_, err = doDelete(testDialClient, regionURL)
 	c.Assert(err, IsNil)
+	records = mustReadURL(c, recordURL)
+	c.Assert(strings.Contains(records, "admin-remove-peer {rm peer: store [2]}"), IsTrue)
 
 	mustPutStore(c, s.svr, 4, metapb.StoreState_Up, nil)
 	err = postJSON(testDialClient, fmt.Sprintf("%s/operators", s.urlPrefix), []byte(`{"name":"add-learner", "region_id": 1, "store_id": 4}`))
@@ -114,6 +123,11 @@ func (s *testOperatorSuite) TestAddRemovePeer(c *C) {
 	c.Assert(err, NotNil)
 	err = postJSON(testDialClient, fmt.Sprintf("%s/operators", s.urlPrefix), []byte(`{"name":"transfer-region", "region_id": 1, "to_store_ids": [1, 2, 3]}`))
 	c.Assert(err, NotNil)
+
+	// Fail to get operator if from is latest.
+	time.Sleep(time.Second)
+	records = mustReadURL(c, fmt.Sprintf("%s/operators/records?from=%s", s.urlPrefix, strconv.FormatInt(time.Now().Unix(), 10)))
+	c.Assert(strings.Contains(records, "operator not found"), IsTrue)
 }
 
 func (s *testOperatorSuite) TestMergeRegionOperator(c *C) {

server/api/router.go

@@ -150,6 +150,7 @@ func createRouter(prefix string, svr *server.Server) *mux.Router {
 	operatorHandler := newOperatorHandler(handler, rd)
 	registerFunc(apiRouter, "GetOperators", "/operators", operatorHandler.List, setMethods("GET"))
 	registerFunc(apiRouter, "SetOperators", "/operators", operatorHandler.Post, setMethods("POST"))
+	registerFunc(apiRouter, "GetOperatorRecords", "/operators/records", operatorHandler.Records, setMethods("GET"))
 	registerFunc(apiRouter, "GetRegionOperator", "/operators/{region_id}", operatorHandler.Get, setMethods("GET"))
 	registerFunc(apiRouter, "DeleteRegionOperator", "/operators/{region_id}", operatorHandler.Delete, setMethods("DELETE"))
 

server/cluster/cluster.go

@@ -330,7 +330,6 @@ func (c *RaftCluster) runBackgroundJobs(interval time.Duration) {
 		case <-ticker.C:
 			c.checkStores()
 			c.collectMetrics()
-			c.coordinator.opController.PruneHistory()
 		}
 	}
 }
@@ -492,7 +491,7 @@ func (c *RaftCluster) GetOpts() *config.PersistOptions {
 	return c.opt
 }
 
-// GetConfig gets the cluster configuration.
+// GetImmutableCfg gets the cluster Immutable configuration.
 func (c *RaftCluster) GetImmutableCfg() *config.ImmutableConfig {
 	return c.immutableCfg
 }

server/config/config.go

@@ -600,7 +600,7 @@ func (c *Config) Adjust(meta *toml.MetaData, reloading bool) error {
 	c.Security.Encryption.Adjust()
 
 	adjustUint64(&c.MaxRegionSize, defaultMaxRegionSize)
-	adjustUint64(&c.MaxRegionSize, defaultMaxSplitSize)
+	adjustUint64(&c.MaxSplitSize, defaultMaxSplitSize)
 
 	return nil
 }

server/config/config_test.go

@@ -524,5 +524,4 @@ func (s *testConfigSuite) TestImmutableConfig(c *C) {
 	iconfig = NewImmutableConfig(config)
 	c.Assert(iconfig.GetMaxRegionSize(), Equals, uint64(200))
 	c.Assert(iconfig.GetMaxRegionSize(), Equals, uint64(300))
-
 }

server/grpc_service.go

@@ -1410,8 +1410,8 @@ func (s *GrpcServer) GetOperator(ctx context.Context, request *pdpb.GetOperatorR
 	return &pdpb.GetOperatorResponse{
 		Header:   s.header(),
 		RegionId: requestID,
-		Desc:     []byte(r.Op.Desc()),
-		Kind:     []byte(r.Op.Kind().String()),
+		Desc:     []byte(r.Desc()),
+		Kind:     []byte(r.Kind().String()),
 		Status:   r.Status,
 	}, nil
 }

server/handler.go

@@ -457,6 +457,19 @@ func (h *Handler) GetHistory(start time.Time) ([]operator.OpHistory, error) {
 	return c.GetHistory(start), nil
 }
 
+// GetRecords returns finished operators since start.
+func (h *Handler) GetRecords(from time.Time) ([]*operator.OpRecord, error) {
+	c, err := h.GetOperatorController()
+	if err != nil {
+		return nil, err
+	}
+	records := c.GetRecords(from)
+	if len(records) == 0 {
+		return nil, ErrOperatorNotFound
+	}
+	return records, nil
+}
+
 // SetAllStoresLimit is used to set limit of all stores.
 func (h *Handler) SetAllStoresLimit(ratePerMin float64, limitType storelimit.Type) error {
 	c, err := h.GetRaftCluster()

server/schedule/checker/merge_checker_test.go

@@ -114,11 +114,14 @@ func (s *testMergeCheckerSuite) TestBasic(c *C) {
 	c.Assert(ops[1].RegionID(), Equals, s.regions[1].GetID())
 
 	// change the max region size
-	s.mc.config = config.NewImmutableConfig(config.NewConfig(), config.WithMaxRegionSize(200))
+	cfg := config.NewConfig()
+	cfg.MaxRegionSize = 200
+	s.mc.config = config.NewImmutableConfig(cfg)
 	s.cluster.PutRegion(s.regions[1].Clone(core.SetApproximateSize(200)))
 	ops = s.mc.Check(s.regions[2])
 	c.Assert(ops, NotNil)
-	s.mc.config = config.NewImmutableConfig(config.NewConfig(), config.WithMaxRegionSize(96))
+	cfg.MaxRegionSize = 96
+	s.mc.config = config.NewImmutableConfig(cfg)
 	// Test the peer store check.
 	store := s.cluster.GetStore(1)
 	c.Assert(store, NotNil)

server/schedule/operator/operator.go

@@ -357,6 +357,37 @@ func (o *Operator) History() []OpHistory {
 	return histories
 }
 
+// OpRecord is used to log and visualize completed operators.
+type OpRecord struct {
+	*Operator
+	FinishTime time.Time
+	duration   time.Duration
+}
+
+func (o *OpRecord) String() string {
+	return fmt.Sprintf("%s (finishAt:%v, duration:%v)", o.Operator.String(), o.FinishTime, o.duration)
+}
+
+// MarshalJSON returns the status of operator as a JSON string
+func (o *OpRecord) MarshalJSON() ([]byte, error) {
+	return []byte(`"` + o.String() + `"`), nil
+}
+
+// Record transfers the operator to OpRecord.
+func (o *Operator) Record(finishTime time.Time) *OpRecord {
+	step := atomic.LoadInt32(&o.currentStep)
+	record := &OpRecord{
+		Operator:   o,
+		FinishTime: finishTime,
+	}
+	start := o.GetStartTime()
+	if o.Status() != SUCCESS && 0 < step && int(step-1) < len(o.stepsTime) {
+		start = time.Unix(0, o.stepsTime[int(step-1)])
+	}
+	record.duration = finishTime.Sub(start)
+	return record
+}
+
 // GetAdditionalInfo returns additional info with string
 func (o *Operator) GetAdditionalInfo() string {
 	if len(o.AdditionalInfos) != 0 {

server/schedule/operator/operator_test.go

@@ -418,3 +418,12 @@ func (s *testOperatorSuite) TestSchedulerKind(c *C) {
 		c.Assert(v.op.SchedulerKind(), Equals, v.expect)
 	}
 }
+
+func (s *testOperatorSuite) TestRecord(c *C) {
+	operator := s.newTestOperator(1, OpLeader, AddLearner{ToStore: 1, PeerID: 1}, RemovePeer{FromStore: 1, PeerID: 1})
+	now := time.Now()
+	time.Sleep(time.Second)
+	ob := operator.Record(now)
+	c.Assert(ob.FinishTime, Equals, now)
+	c.Assert(ob.duration.Seconds(), Greater, time.Second.Seconds())
+}