Use go mod instead of GO_PATH and add more cgo interfeces

Signed-off-by: Nbigsheeper <yihao.dai@zilliz.com>

core/src/dog_segment/Collection.h

@@ -20,6 +20,10 @@ public:
       return schema_;
     }
 
+    std::string& get_collection_name() {
+      return collection_name_;
+    }
+
 private:
     // TODO: add Index ptr
     // IndexPtr index_ = nullptr;

core/src/dog_segment/Partition.h

@@ -13,6 +13,10 @@ public:
       return schema_;
     }
 
+    std::string& get_partition_name() {
+      return partition_name_;
+    }
+
 private:
     std::string partition_name_;
     SchemaPtr schema_;

core/src/dog_segment/SegmentBase.h

@@ -89,7 +89,7 @@ class SegmentBase {
     void set_time_end(Timestamp time_end) {
         this->time_end_ = time_end;
     }
-    uint64_t get_segment_id(uint64_t segment_id) {
+    uint64_t get_segment_id() {
         return segment_id_;
     }
     uint64_t set_segment_id(uint64_t segment_id) {

core/src/dog_segment/collection_c.cpp

@@ -8,6 +8,8 @@ NewCollection(const char* collection_name, const char* schema_conf) {
 
   auto collection = std::make_unique<milvus::dog_segment::Collection>(name, conf);
 
+  // TODO: delete print
+  std::cout << "create collection " << collection_name << std::endl;
   return (void*)collection.release();
 }
 
@@ -15,5 +17,7 @@ void
 DeleteCollection(CCollection collection) {
   auto col = (milvus::dog_segment::Collection*)collection;
 
+  // TODO: delete print
+  std::cout << "delete collection " << col->get_collection_name() << std::endl;
   delete col;
 }

core/src/dog_segment/partition_c.cpp

@@ -12,11 +12,15 @@ NewPartition(CCollection collection, const char* partition_name) {
 
   auto partition = std::make_unique<milvus::dog_segment::Partition>(name, schema);
 
+  // TODO: delete print
+  std::cout << "create partition " << name << std::endl;
   return (void*)partition.release();
 }
 
 void DeletePartition(CPartition partition) {
   auto p = (milvus::dog_segment::Partition*)partition;
 
+  // TODO: delete print
+  std::cout << "delete partition " << p->get_partition_name() <<std::endl;
   delete p;
 }

core/src/dog_segment/segment_c.cpp

@@ -10,12 +10,16 @@ NewSegment(CPartition partition, unsigned long segment_id) {
 
   segment->set_segment_id(segment_id);
 
+  // TODO: delete print
+  std::cout << "create segment " << segment_id << std::endl;
   return (void*)segment.release();
 }
 
 void DeleteSegment(CSegmentBase segment) {
   auto s = (milvus::dog_segment::SegmentBase*)segment;
 
+  // TODO: delete print
+  std::cout << "delete segment " << s->get_segment_id() << std::endl;
   delete s;
 }
 

core/unittest/test_c_api.cpp

@@ -2,14 +2,47 @@
 #include <string>
 #include <random>
 #include <gtest/gtest.h>
-#include <dog_segment/SegmentBase.h>
 
-#include "dog_segment/segment_c.h"
 #include "dog_segment/collection_c.h"
+#include "dog_segment/segment_c.h"
+
+TEST(CApiTest, CollectionTest) {
+  auto collection_name = "collection0";
+  auto schema_tmp_conf = "null_schema";
+  auto collection = NewCollection(collection_name, schema_tmp_conf);
+  DeleteCollection(collection);
+}
+
+TEST(CApiTest, PartitonTest) {
+  auto collection_name = "collection0";
+  auto schema_tmp_conf = "null_schema";
+  auto collection = NewCollection(collection_name, schema_tmp_conf);
+  auto partition_name = "partition0";
+  auto partition = NewPartition(collection, partition_name);
+  DeleteCollection(collection);
+  DeletePartition(partition);
+}
 
-TEST(SegmentTest, InsertTest) {
-  auto fake_schema = std::make_shared<milvus::dog_segment::Schema>();
-  auto s = milvus::dog_segment::CreateSegment(fake_schema).release();
+TEST(CApiTest, SegmentTest) {
+  auto collection_name = "collection0";
+  auto schema_tmp_conf = "null_schema";
+  auto collection = NewCollection(collection_name, schema_tmp_conf);
+  auto partition_name = "partition0";
+  auto partition = NewPartition(collection, partition_name);
+  auto segment = NewSegment(partition, 0);
+  DeleteCollection(collection);
+  DeletePartition(partition);
+  DeleteSegment(segment);
+}
+
+
+TEST(CApiTest, InsertTest) {
+  auto collection_name = "collection0";
+  auto schema_tmp_conf = "null_schema";
+  auto collection = NewCollection(collection_name, schema_tmp_conf);
+  auto partition_name = "partition0";
+  auto partition = NewPartition(collection, partition_name);
+  auto segment = NewSegment(partition, 0);
 
   std::vector<char> raw_data;
   std::vector<uint64_t> timestamps;
@@ -31,7 +64,11 @@ TEST(SegmentTest, InsertTest) {
 
   auto line_sizeof = (sizeof(int) + sizeof(float) * 16);
 
-  auto res = Insert(s, N, uids.data(), timestamps.data(), raw_data.data(), (int)line_sizeof, N);
+  auto res = Insert(segment, N, uids.data(), timestamps.data(), raw_data.data(), (int)line_sizeof, N);
+
+  assert(res == 0);
 
-  std::cout << res << std::endl;
-}
\ No newline at end of file
+  DeleteCollection(collection);
+  DeletePartition(partition);
+  DeleteSegment(segment);
+}

errors/errors.go

+
+package errors
+
+import (
+	"fmt"
+	"io"
+)
+
+// New returns an error with the supplied message.
+// New also records the stack trace at the point it was called.
+func New(message string) error {
+	return &fundamental{
+		msg:   message,
+		stack: callers(),
+	}
+}
+
+// Errorf formats according to a format specifier and returns the string
+// as a value that satisfies error.
+// Errorf also records the stack trace at the point it was called.
+func Errorf(format string, args ...interface{}) error {
+	return &fundamental{
+		msg:   fmt.Sprintf(format, args...),
+		stack: callers(),
+	}
+}
+
+// StackTraceAware is an optimization to avoid repetitive traversals of an error chain.
+// HasStack checks for this marker first.
+// Annotate/Wrap and Annotatef/Wrapf will produce this marker.
+type StackTraceAware interface {
+	HasStack() bool
+}
+
+// HasStack tells whether a StackTracer exists in the error chain
+func HasStack(err error) bool {
+	if errWithStack, ok := err.(StackTraceAware); ok {
+		return errWithStack.HasStack()
+	}
+	return GetStackTracer(err) != nil
+}
+
+// fundamental is an error that has a message and a stack, but no caller.
+type fundamental struct {
+	msg string
+	*stack
+}
+
+func (f *fundamental) Error() string { return f.msg }
+
+func (f *fundamental) Format(s fmt.State, verb rune) {
+	switch verb {
+	case 'v':
+		if s.Flag('+') {
+			io.WriteString(s, f.msg)
+			f.stack.Format(s, verb)
+			return
+		}
+		fallthrough
+	case 's':
+		io.WriteString(s, f.msg)
+	case 'q':
+		fmt.Fprintf(s, "%q", f.msg)
+	}
+}
+
+// WithStack annotates err with a stack trace at the point WithStack was called.
+// If err is nil, WithStack returns nil.
+//
+// For most use cases this is deprecated and AddStack should be used (which will ensure just one stack trace).
+// However, one may want to use this in some situations, for example to create a 2nd trace across a goroutine.
+func WithStack(err error) error {
+	if err == nil {
+		return nil
+	}
+
+	return &withStack{
+		err,
+		callers(),
+	}
+}
+
+// AddStack is similar to WithStack.
+// However, it will first check with HasStack to see if a stack trace already exists in the causer chain before creating another one.
+func AddStack(err error) error {
+	if HasStack(err) {
+		return err
+	}
+	return WithStack(err)
+}
+
+type withStack struct {
+	error
+	*stack
+}
+
+func (w *withStack) Cause() error { return w.error }
+
+func (w *withStack) Format(s fmt.State, verb rune) {
+	switch verb {
+	case 'v':
+		if s.Flag('+') {
+			fmt.Fprintf(s, "%+v", w.Cause())
+			w.stack.Format(s, verb)
+			return
+		}
+		fallthrough
+	case 's':
+		io.WriteString(s, w.Error())
+	case 'q':
+		fmt.Fprintf(s, "%q", w.Error())
+	}
+}
+
+// Wrap returns an error annotating err with a stack trace
+// at the point Wrap is called, and the supplied message.
+// If err is nil, Wrap returns nil.
+//
+// For most use cases this is deprecated in favor of Annotate.
+// Annotate avoids creating duplicate stack traces.
+func Wrap(err error, message string) error {
+	if err == nil {
+		return nil
+	}
+	hasStack := HasStack(err)
+	err = &withMessage{
+		cause:         err,
+		msg:           message,
+		causeHasStack: hasStack,
+	}
+	return &withStack{
+		err,
+		callers(),
+	}
+}
+
+// Wrapf returns an error annotating err with a stack trace
+// at the point Wrapf is call, and the format specifier.
+// If err is nil, Wrapf returns nil.
+//
+// For most use cases this is deprecated in favor of Annotatef.
+// Annotatef avoids creating duplicate stack traces.
+func Wrapf(err error, format string, args ...interface{}) error {
+	if err == nil {
+		return nil
+	}
+	hasStack := HasStack(err)
+	err = &withMessage{
+		cause:         err,
+		msg:           fmt.Sprintf(format, args...),
+		causeHasStack: hasStack,
+	}
+	return &withStack{
+		err,
+		callers(),
+	}
+}
+
+// WithMessage annotates err with a new message.
+// If err is nil, WithMessage returns nil.
+func WithMessage(err error, message string) error {
+	if err == nil {
+		return nil
+	}
+	return &withMessage{
+		cause:         err,
+		msg:           message,
+		causeHasStack: HasStack(err),
+	}
+}
+
+type withMessage struct {
+	cause         error
+	msg           string
+	causeHasStack bool
+}
+
+func (w *withMessage) Error() string  { return w.msg + ": " + w.cause.Error() }
+func (w *withMessage) Cause() error   { return w.cause }
+func (w *withMessage) HasStack() bool { return w.causeHasStack }
+
+func (w *withMessage) Format(s fmt.State, verb rune) {
+	switch verb {
+	case 'v':
+		if s.Flag('+') {
+			fmt.Fprintf(s, "%+v\n", w.Cause())
+			io.WriteString(s, w.msg)
+			return
+		}
+		fallthrough
+	case 's', 'q':
+		io.WriteString(s, w.Error())
+	}
+}
+
+// Cause returns the underlying cause of the error, if possible.
+// An error value has a cause if it implements the following
+// interface:
+//
+//     type causer interface {
+//            Cause() error
+//     }
+//
+// If the error does not implement Cause, the original error will
+// be returned. If the error is nil, nil will be returned without further
+// investigation.
+func Cause(err error) error {
+	cause := Unwrap(err)
+	if cause == nil {
+		return err
+	}
+	return Cause(cause)
+}
+
+// Unwrap uses causer to return the next error in the chain or nil.
+// This goes one-level deeper, whereas Cause goes as far as possible
+func Unwrap(err error) error {
+	type causer interface {
+		Cause() error
+	}
+	if unErr, ok := err.(causer); ok {
+		return unErr.Cause()
+	}
+	return nil
+}
+
+// Find an error in the chain that matches a test function.
+// returns nil if no error is found.
+func Find(origErr error, test func(error) bool) error {
+	var foundErr error
+	WalkDeep(origErr, func(err error) bool {
+		if test(err) {
+			foundErr = err
+			return true
+		}
+		return false
+	})
+	return foundErr
+}

errors/group.go

+package errors
+
+// ErrorGroup is an interface for multiple errors that are not a chain.
+// This happens for example when executing multiple operations in parallel.
+type ErrorGroup interface {
+	Errors() []error
+}
+
+// Errors uses the ErrorGroup interface to return a slice of errors.
+// If the ErrorGroup interface is not implemented it returns an array containing just the given error.
+func Errors(err error) []error {
+	if eg, ok := err.(ErrorGroup); ok {
+		return eg.Errors()
+	}
+	return []error{err}
+}
+
+// WalkDeep does a depth-first traversal of all errors.
+// Any ErrorGroup is traversed (after going deep).
+// The visitor function can return true to end the traversal early
+// In that case, WalkDeep will return true, otherwise false.
+func WalkDeep(err error, visitor func(err error) bool) bool {
+	// Go deep
+	unErr := err
+	for unErr != nil {
+		if done := visitor(unErr); done {
+			return true
+		}
+		unErr = Unwrap(unErr)
+	}
+
+	// Go wide
+	if group, ok := err.(ErrorGroup); ok {
+		for _, err := range group.Errors() {
+			if early := WalkDeep(err, visitor); early {
+				return true
+			}
+		}
+	}
+
+	return false
+}

errors/stack.go

+package errors
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"path"
+	"runtime"
+	"strconv"
+	"strings"
+)
+
+// StackTracer retrieves the StackTrace
+// Generally you would want to use the GetStackTracer function to do that.
+type StackTracer interface {
+	StackTrace() StackTrace
+}
+
+// GetStackTracer will return the first StackTracer in the causer chain.
+// This function is used by AddStack to avoid creating redundant stack traces.
+//
+// You can also use the StackTracer interface on the returned error to get the stack trace.
+func GetStackTracer(origErr error) StackTracer {
+	var stacked StackTracer
+	WalkDeep(origErr, func(err error) bool {
+		if stackTracer, ok := err.(StackTracer); ok {
+			stacked = stackTracer
+			return true
+		}
+		return false
+	})
+	return stacked
+}
+
+// Frame represents a program counter inside a stack frame.
+type Frame uintptr
+
+// pc returns the program counter for this frame;
+// multiple frames may have the same PC value.
+func (f Frame) pc() uintptr { return uintptr(f) - 1 }
+
+// file returns the full path to the file that contains the
+// function for this Frame's pc.
+func (f Frame) file() string {
+	fn := runtime.FuncForPC(f.pc())
+	if fn == nil {
+		return "unknown"
+	}
+	file, _ := fn.FileLine(f.pc())
+	return file
+}
+
+// line returns the line number of source code of the
+// function for this Frame's pc.
+func (f Frame) line() int {
+	fn := runtime.FuncForPC(f.pc())
+	if fn == nil {
+		return 0
+	}
+	_, line := fn.FileLine(f.pc())
+	return line
+}
+
+// Format formats the frame according to the fmt.Formatter interface.
+//
+//    %s    source file
+//    %d    source line
+//    %n    function name
+//    %v    equivalent to %s:%d
+//
+// Format accepts flags that alter the printing of some verbs, as follows:
+//
+//    %+s   function name and path of source file relative to the compile time
+//          GOPATH separated by \n\t (<funcname>\n\t<path>)
+//    %+v   equivalent to %+s:%d
+func (f Frame) Format(s fmt.State, verb rune) {
+	f.format(s, s, verb)
+}
+
+// format allows stack trace printing calls to be made with a bytes.Buffer.
+func (f Frame) format(w io.Writer, s fmt.State, verb rune) {
+	switch verb {
+	case 's':
+		switch {
+		case s.Flag('+'):
+			pc := f.pc()
+			fn := runtime.FuncForPC(pc)
+			if fn == nil {
+				io.WriteString(w, "unknown")
+			} else {
+				file, _ := fn.FileLine(pc)
+				io.WriteString(w, fn.Name())
+				io.WriteString(w, "\n\t")
+				io.WriteString(w, file)
+			}
+		default:
+			io.WriteString(w, path.Base(f.file()))
+		}
+	case 'd':
+		io.WriteString(w, strconv.Itoa(f.line()))
+	case 'n':
+		name := runtime.FuncForPC(f.pc()).Name()
+		io.WriteString(w, funcname(name))
+	case 'v':
+		f.format(w, s, 's')
+		io.WriteString(w, ":")
+		f.format(w, s, 'd')
+	}
+}
+
+// StackTrace is stack of Frames from innermost (newest) to outermost (oldest).
+type StackTrace []Frame
+
+// Format formats the stack of Frames according to the fmt.Formatter interface.
+//
+//    %s	lists source files for each Frame in the stack
+//    %v	lists the source file and line number for each Frame in the stack
+//
+// Format accepts flags that alter the printing of some verbs, as follows:
+//
+//    %+v   Prints filename, function, and line number for each Frame in the stack.
+func (st StackTrace) Format(s fmt.State, verb rune) {
+	var b bytes.Buffer
+	switch verb {
+	case 'v':
+		switch {
+		case s.Flag('+'):
+			b.Grow(len(st) * stackMinLen)
+			for _, fr := range st {
+				b.WriteByte('\n')
+				fr.format(&b, s, verb)
+			}
+		case s.Flag('#'):
+			fmt.Fprintf(&b, "%#v", []Frame(st))
+		default:
+			st.formatSlice(&b, s, verb)
+		}
+	case 's':
+		st.formatSlice(&b, s, verb)
+	}
+	io.Copy(s, &b)
+}
+
+// formatSlice will format this StackTrace into the given buffer as a slice of
+// Frame, only valid when called with '%s' or '%v'.
+func (st StackTrace) formatSlice(b *bytes.Buffer, s fmt.State, verb rune) {
+	b.WriteByte('[')
+	if len(st) == 0 {
+		b.WriteByte(']')
+		return
+	}
+
+	b.Grow(len(st) * (stackMinLen / 4))
+	st[0].format(b, s, verb)
+	for _, fr := range st[1:] {
+		b.WriteByte(' ')
+		fr.format(b, s, verb)
+	}
+	b.WriteByte(']')
+}
+
+// stackMinLen is a best-guess at the minimum length of a stack trace. It
+// doesn't need to be exact, just give a good enough head start for the buffer
+// to avoid the expensive early growth.
+const stackMinLen = 96
+
+// stack represents a stack of program counters.
+type stack []uintptr
+
+func (s *stack) Format(st fmt.State, verb rune) {
+	switch verb {
+	case 'v':
+		switch {
+		case st.Flag('+'):
+			var b bytes.Buffer
+			b.Grow(len(*s) * stackMinLen)
+			for _, pc := range *s {
+				f := Frame(pc)
+				b.WriteByte('\n')
+				f.format(&b, st, 'v')
+			}
+			io.Copy(st, &b)
+		}
+	}
+}
+
+func (s *stack) StackTrace() StackTrace {
+	f := make([]Frame, len(*s))
+	for i := 0; i < len(f); i++ {
+		f[i] = Frame((*s)[i])
+	}
+	return f
+}
+
+func callers() *stack {
+	return callersSkip(4)
+}
+
+func callersSkip(skip int) *stack {
+	const depth = 32
+	var pcs [depth]uintptr
+	n := runtime.Callers(skip, pcs[:])
+	var st stack = pcs[0:n]
+	return &st
+}
+
+// funcname removes the path prefix component of a function's name reported by func.Name().
+func funcname(name string) string {
+	i := strings.LastIndex(name, "/")
+	name = name[i+1:]
+	i = strings.Index(name, ".")
+	return name[i+1:]
+}
+
+// NewStack is for library implementers that want to generate a stack trace.
+// Normally you should insted use AddStack to get an error with a stack trace.
+//
+// The result of this function can be turned into a stack trace by calling .StackTrace()
+//
+// This function takes an argument for the number of stack frames to skip.
+// This avoids putting stack generation function calls like this one in the stack trace.
+// A value of 0 will give you the line that called NewStack(0)
+// A library author wrapping this in their own function will want to use a value of at least 1.
+func NewStack(skip int) StackTracer {
+	return callersSkip(skip + 3)
+}

go.mod

 module github.com/czs007/suvlim
 
-go 1.14
+go 1.15
 
 require (
 	code.cloudfoundry.org/bytefmt v0.0.0-20200131002437-cf55d5288a48 // indirect
-	github.com/onsi/ginkgo v1.14.0 // indirect
 	github.com/BurntSushi/toml v0.3.1
+	github.com/apache/pulsar/pulsar-client-go v0.0.0-20200901051823-800681aaa9af
 	github.com/coreos/etcd v3.3.25+incompatible // indirect
-	github.com/envoyproxy/protoc-gen-validate v0.1.0 // indirect
-	github.com/gogo/protobuf v1.3.1 // indirect
+	github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f
+	github.com/docker/go-units v0.4.0
+	github.com/gogo/protobuf v1.3.1
 	github.com/golang/protobuf v1.4.2
+	github.com/google/btree v1.0.0
 	github.com/minio/minio-go/v7 v7.0.5
-	github.com/onsi/ginkgo v1.14.0 // indirect
+	github.com/opentracing/opentracing-go v1.2.0
+	github.com/pingcap/check v0.0.0-20200212061837-5e12011dc712
+	github.com/pingcap/errors v0.11.4
+	github.com/pingcap/log v0.0.0-20200828042413-fce0951f1463
 	github.com/pivotal-golang/bytefmt v0.0.0-20200131002437-cf55d5288a48
+	github.com/sirupsen/logrus v1.6.0
 	github.com/stretchr/testify v1.6.1
-	github.com/tikv/client-go v0.0.0-20200723074018-095b94dc2430
+	github.com/tikv/client-go v0.0.0-20200824032810-95774393107b
+	github.com/tikv/pd v2.1.19+incompatible
 	go.etcd.io/etcd v3.3.25+incompatible
-	google.golang.org/grpc v1.23.1
+	go.uber.org/zap v1.15.0
+	golang.org/x/net v0.0.0-20200822124328-c89045814202
+	google.golang.org/grpc v1.31.1
+	google.golang.org/grpc/examples v0.0.0-20200828165940-d8ef479ab79a // indirect
+	gopkg.in/natefinch/lumberjack.v2 v2.0.0
+	sigs.k8s.io/yaml v1.2.0 // indirect
 )

master/client/base_client.go

+// Copyright 2019 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package pd
+
+import (
+	"context"
+	"reflect"
+	"sort"
+	"sync"
+	"time"
+
+	"github.com/czs007/suvlim/errors"
+	"github.com/czs007/suvlim/pkg/pdpb"
+	"github.com/czs007/suvlim/util/grpcutil"
+	"github.com/pingcap/log"
+	"go.uber.org/zap"
+	"google.golang.org/grpc"
+)
+
+// baseClient is a basic client for all other complex client.
+type baseClient struct {
+	urls      []string
+	clusterID uint64
+	connMu    struct {
+		sync.RWMutex
+		clientConns map[string]*grpc.ClientConn
+		leader      string
+	}
+
+	checkLeaderCh chan struct{}
+
+	wg     sync.WaitGroup
+	ctx    context.Context
+	cancel context.CancelFunc
+
+	security SecurityOption
+
+	gRPCDialOptions []grpc.DialOption
+	timeout         time.Duration
+}
+
+// SecurityOption records options about tls
+type SecurityOption struct {
+	CAPath   string
+	CertPath string
+	KeyPath  string
+}
+
+// ClientOption configures client.
+type ClientOption func(c *baseClient)
+
+// WithGRPCDialOptions configures the client with gRPC dial options.
+func WithGRPCDialOptions(opts ...grpc.DialOption) ClientOption {
+	return func(c *baseClient) {
+		c.gRPCDialOptions = append(c.gRPCDialOptions, opts...)
+	}
+}
+
+// WithCustomTimeoutOption configures the client with timeout option.
+func WithCustomTimeoutOption(timeout time.Duration) ClientOption {
+	return func(c *baseClient) {
+		c.timeout = timeout
+	}
+}
+
+// newBaseClient returns a new baseClient.
+func newBaseClient(ctx context.Context, urls []string, security SecurityOption, opts ...ClientOption) (*baseClient, error) {
+	ctx1, cancel := context.WithCancel(ctx)
+	c := &baseClient{
+		urls:          urls,
+		checkLeaderCh: make(chan struct{}, 1),
+		ctx:           ctx1,
+		cancel:        cancel,
+		security:      security,
+		timeout:       defaultPDTimeout,
+	}
+	c.connMu.clientConns = make(map[string]*grpc.ClientConn)
+	for _, opt := range opts {
+		opt(c)
+	}
+
+	log.Info("[pd] init cluster id", zap.Uint64("cluster-id", c.clusterID))
+
+	c.wg.Add(1)
+	go c.leaderLoop()
+
+	return c, nil
+}
+
+func (c *baseClient) initRetry(f func() error) error {
+	var err error
+	for i := 0; i < maxInitClusterRetries; i++ {
+		if err = f(); err == nil {
+			return nil
+		}
+		select {
+		case <-c.ctx.Done():
+			return err
+		case <-time.After(time.Second):
+		}
+	}
+	return errors.WithStack(err)
+}
+
+func (c *baseClient) leaderLoop() {
+	defer c.wg.Done()
+
+	ctx, cancel := context.WithCancel(c.ctx)
+	defer cancel()
+
+	for {
+		select {
+		case <-c.checkLeaderCh:
+		case <-time.After(time.Minute):
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// ScheduleCheckLeader is used to check leader.
+func (c *baseClient) ScheduleCheckLeader() {
+	select {
+	case c.checkLeaderCh <- struct{}{}:
+	default:
+	}
+}
+
+// GetClusterID returns the ClusterID.
+func (c *baseClient) GetClusterID(context.Context) uint64 {
+	return c.clusterID
+}
+
+// GetLeaderAddr returns the leader address.
+// For testing use.
+func (c *baseClient) GetLeaderAddr() string {
+	c.connMu.RLock()
+	defer c.connMu.RUnlock()
+	return c.connMu.leader
+}
+
+// GetURLs returns the URLs.
+// For testing use. It should only be called when the client is closed.
+func (c *baseClient) GetURLs() []string {
+	return c.urls
+}
+
+
+func (c *baseClient) updateURLs(members []*pdpb.Member) {
+	urls := make([]string, 0, len(members))
+	for _, m := range members {
+		urls = append(urls, m.GetClientUrls()...)
+	}
+
+	sort.Strings(urls)
+	// the url list is same.
+	if reflect.DeepEqual(c.urls, urls) {
+		return
+	}
+
+	log.Info("[pd] update member urls", zap.Strings("old-urls", c.urls), zap.Strings("new-urls", urls))
+	c.urls = urls
+}
+
+func (c *baseClient) switchLeader(addrs []string) error {
+	// FIXME: How to safely compare leader urls? For now, only allows one client url.
+	addr := addrs[0]
+
+	c.connMu.RLock()
+	oldLeader := c.connMu.leader
+	c.connMu.RUnlock()
+
+	if addr == oldLeader {
+		return nil
+	}
+
+	log.Info("[pd] switch leader", zap.String("new-leader", addr), zap.String("old-leader", oldLeader))
+	if _, err := c.getOrCreateGRPCConn(addr); err != nil {
+		return err
+	}
+
+	c.connMu.Lock()
+	defer c.connMu.Unlock()
+	c.connMu.leader = addr
+	return nil
+}
+
+func (c *baseClient) getOrCreateGRPCConn(addr string) (*grpc.ClientConn, error) {
+	c.connMu.RLock()
+	conn, ok := c.connMu.clientConns[addr]
+	c.connMu.RUnlock()
+	if ok {
+		return conn, nil
+	}
+	tlsCfg, err := grpcutil.SecurityConfig{
+		CAPath:   c.security.CAPath,
+		CertPath: c.security.CertPath,
+		KeyPath:  c.security.KeyPath,
+	}.ToTLSConfig()
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	dctx, cancel := context.WithTimeout(c.ctx, dialTimeout)
+	defer cancel()
+	cc, err := grpcutil.GetClientConn(dctx, addr, tlsCfg, c.gRPCDialOptions...)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	c.connMu.Lock()
+	defer c.connMu.Unlock()
+	if old, ok := c.connMu.clientConns[addr]; ok {
+		cc.Close()
+		log.Debug("use old connection", zap.String("target", cc.Target()), zap.String("state", cc.GetState().String()))
+		return old, nil
+	}
+
+	c.connMu.clientConns[addr] = cc
+	return cc, nil
+}

master/client/client.go

+// Copyright 2016 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package pd
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/opentracing/opentracing-go"
+	"github.com/czs007/suvlim/errors"
+	"github.com/czs007/suvlim/pkg/pdpb"
+	"github.com/pingcap/log"
+	"go.uber.org/zap"
+)
+
+
+// Client is a PD (Placement Driver) client.
+// It should not be used after calling Close().
+type Client interface {
+	// GetClusterID gets the cluster ID from PD.
+	GetClusterID(ctx context.Context) uint64
+	// GetMemberInfo gets the members Info from PD
+	//GetMemberInfo(ctx context.Context) ([]*pdpb.Member, error)
+	// GetLeaderAddr returns current leader's address. It returns "" before
+	// syncing leader from server.
+	GetLeaderAddr() string
+	// GetTS gets a timestamp from PD.
+	GetTS(ctx context.Context) (int64, int64, error)
+	// GetTSAsync gets a timestamp from PD, without block the caller.
+	GetTSAsync(ctx context.Context) TSFuture
+	Close()
+}
+
+type tsoRequest struct {
+	start    time.Time
+	ctx      context.Context
+	done     chan error
+	physical int64
+	logical  int64
+}
+
+const (
+	defaultPDTimeout      = 3 * time.Second
+	dialTimeout           = 3 * time.Second
+	updateLeaderTimeout   = time.Second // Use a shorter timeout to recover faster from network isolation.
+	maxMergeTSORequests   = 10000       // should be higher if client is sending requests in burst
+	maxInitClusterRetries = 100
+)
+
+var (
+	// errFailInitClusterID is returned when failed to load clusterID from all supplied PD addresses.
+	errFailInitClusterID = errors.New("[pd] failed to get cluster id")
+	// errClosing is returned when request is canceled when client is closing.
+	errClosing = errors.New("[pd] closing")
+	// errTSOLength is returned when the number of response timestamps is inconsistent with request.
+	errTSOLength = errors.New("[pd] tso length in rpc response is incorrect")
+)
+
+type client struct {
+	*baseClient
+	tsoRequests chan *tsoRequest
+
+	lastPhysical int64
+	lastLogical  int64
+
+	tsDeadlineCh chan deadline
+}
+
+// NewClient creates a PD client.
+func NewClient(pdAddrs []string, security SecurityOption, opts ...ClientOption) (Client, error) {
+	return NewClientWithContext(context.Background(), pdAddrs, security, opts...)
+}
+
+// NewClientWithContext creates a PD client with context.
+func NewClientWithContext(ctx context.Context, pdAddrs []string, security SecurityOption, opts ...ClientOption) (Client, error) {
+	log.Info("[pd] create pd client with endpoints", zap.Strings("pd-address", pdAddrs))
+	base, err := newBaseClient(ctx, addrsToUrls(pdAddrs), security, opts...)
+	if err != nil {
+		return nil, err
+	}
+	c := &client{
+		baseClient:   base,
+		tsoRequests:  make(chan *tsoRequest, maxMergeTSORequests),
+		tsDeadlineCh: make(chan deadline, 1),
+	}
+
+	c.wg.Add(2)
+	go c.tsLoop()
+	go c.tsCancelLoop()
+
+	return c, nil
+}
+
+type deadline struct {
+	timer  <-chan time.Time
+	done   chan struct{}
+	cancel context.CancelFunc
+}
+
+func (c *client) tsCancelLoop() {
+	defer c.wg.Done()
+
+	ctx, cancel := context.WithCancel(c.ctx)
+	defer cancel()
+
+	for {
+		select {
+		case d := <-c.tsDeadlineCh:
+			select {
+			case <-d.timer:
+				log.Error("tso request is canceled due to timeout")
+				d.cancel()
+			case <-d.done:
+			case <-ctx.Done():
+				return
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (c *client) checkStreamTimeout(loopCtx context.Context, cancel context.CancelFunc, createdCh chan struct{}) {
+	select {
+	case <-time.After(c.timeout):
+		cancel()
+	case <-createdCh:
+		return
+	case <-loopCtx.Done():
+		return
+	}
+}
+
+
+func (c *client) tsLoop() {
+	defer c.wg.Done()
+
+	loopCtx, loopCancel := context.WithCancel(c.ctx)
+	defer loopCancel()
+
+	defaultSize := maxMergeTSORequests + 1
+	requests := make([]*tsoRequest, defaultSize)
+	createdCh := make(chan struct{})
+
+	var opts []opentracing.StartSpanOption
+	var stream pdpb.PD_TsoClient
+	var cancel context.CancelFunc
+
+	for {
+		var err error
+
+		if stream == nil {
+			var ctx context.Context
+			ctx, cancel = context.WithCancel(loopCtx)
+			go c.checkStreamTimeout(loopCtx, cancel, createdCh)
+			stream, err = c.leaderClient().Tso(ctx)
+			if stream != nil {
+				createdCh <- struct{}{}
+			}
+			if err != nil {
+				select {
+				case <-loopCtx.Done():
+					cancel()
+					return
+				default:
+				}
+				log.Error("[pd] create tso stream error")
+				c.ScheduleCheckLeader()
+				cancel()
+				c.revokeTSORequest(errors.WithStack(err))
+				select {
+				case <-time.After(time.Second):
+				case <-loopCtx.Done():
+					return
+				}
+				continue
+			}
+		}
+
+		select {
+		case first := <-c.tsoRequests:
+			pendingPlus1 := len(c.tsoRequests) + 1
+			requests[0] = first
+			for i := 1; i < pendingPlus1; i++ {
+				requests[i] = <-c.tsoRequests
+			}
+			done := make(chan struct{})
+			dl := deadline{
+				timer:  time.After(c.timeout),
+				done:   done,
+				cancel: cancel,
+			}
+			select {
+			case c.tsDeadlineCh <- dl:
+			case <-loopCtx.Done():
+				cancel()
+				return
+			}
+			opts = extractSpanReference(requests[:pendingPlus1], opts[:0])
+			err = c.processTSORequests(stream, requests[:pendingPlus1], opts)
+			close(done)
+		case <-loopCtx.Done():
+			cancel()
+			return
+		}
+
+		if err != nil {
+			select {
+			case <-loopCtx.Done():
+				cancel()
+				return
+			default:
+			}
+			log.Error("[pd] getTS error")
+			c.ScheduleCheckLeader()
+			cancel()
+			stream, cancel = nil, nil
+		}
+	}
+}
+
+func extractSpanReference(requests []*tsoRequest, opts []opentracing.StartSpanOption) []opentracing.StartSpanOption {
+	for _, req := range requests {
+		if span := opentracing.SpanFromContext(req.ctx); span != nil {
+			opts = append(opts, opentracing.ChildOf(span.Context()))
+		}
+	}
+	return opts
+}
+
+func (c *client) processTSORequests(stream pdpb.PD_TsoClient, requests []*tsoRequest, opts []opentracing.StartSpanOption) error {
+	if len(opts) > 0 {
+		span := opentracing.StartSpan("pdclient.processTSORequests", opts...)
+		defer span.Finish()
+	}
+	count := len(requests)
+	//start := time.Now()
+	req := &pdpb.TsoRequest{
+		Header: c.requestHeader(),
+		Count:  uint32(count),
+	}
+
+	if err := stream.Send(req); err != nil {
+		err = errors.WithStack(err)
+		c.finishTSORequest(requests, 0, 0, err)
+		return err
+	}
+	resp, err := stream.Recv()
+	if err != nil {
+		err = errors.WithStack(err)
+		c.finishTSORequest(requests, 0, 0, err)
+		return err
+	}
+
+	if resp.GetCount() != uint32(len(requests)) {
+		err = errors.WithStack(errTSOLength)
+		c.finishTSORequest(requests, 0, 0, err)
+		return err
+	}
+
+	physical, logical := resp.GetTimestamp().GetPhysical(), resp.GetTimestamp().GetLogical()
+	// Server returns the highest ts.
+	logical -= int64(resp.GetCount() - 1)
+	if tsLessEqual(physical, logical, c.lastPhysical, c.lastLogical) {
+		panic(errors.Errorf("timestamp fallback, newly acquired ts (%d,%d) is less or equal to last one (%d, %d)",
+			physical, logical, c.lastLogical, c.lastLogical))
+	}
+	c.lastPhysical = physical
+	c.lastLogical = logical + int64(len(requests)) - 1
+	c.finishTSORequest(requests, physical, logical, nil)
+	return nil
+}
+
+func tsLessEqual(physical, logical, thatPhysical, thatLogical int64) bool {
+	if physical == thatPhysical {
+		return logical <= thatLogical
+	}
+	return physical < thatPhysical
+}
+
+func (c *client) finishTSORequest(requests []*tsoRequest, physical, firstLogical int64, err error) {
+	for i := 0; i < len(requests); i++ {
+		if span := opentracing.SpanFromContext(requests[i].ctx); span != nil {
+			span.Finish()
+		}
+		requests[i].physical, requests[i].logical = physical, firstLogical+int64(i)
+		requests[i].done <- err
+	}
+}
+
+func (c *client) revokeTSORequest(err error) {
+	n := len(c.tsoRequests)
+	for i := 0; i < n; i++ {
+		req := <-c.tsoRequests
+		req.done <- err
+	}
+}
+
+func (c *client) Close() {
+	c.cancel()
+	c.wg.Wait()
+
+	c.revokeTSORequest(errors.WithStack(errClosing))
+
+	c.connMu.Lock()
+	defer c.connMu.Unlock()
+	for _, cc := range c.connMu.clientConns {
+		if err := cc.Close(); err != nil {
+			log.Error("[pd] failed to close gRPC clientConn")
+		}
+	}
+}
+
+// leaderClient gets the client of current PD leader.
+func (c *client) leaderClient() pdpb.PDClient {
+	c.connMu.RLock()
+	defer c.connMu.RUnlock()
+
+	return pdpb.NewPDClient(c.connMu.clientConns[c.connMu.leader])
+}
+
+var tsoReqPool = sync.Pool{
+	New: func() interface{} {
+		return &tsoRequest{
+			done:     make(chan error, 1),
+			physical: 0,
+			logical:  0,
+		}
+	},
+}
+
+func (c *client) GetTSAsync(ctx context.Context) TSFuture {
+	if span := opentracing.SpanFromContext(ctx); span != nil {
+		span = opentracing.StartSpan("GetTSAsync", opentracing.ChildOf(span.Context()))
+		ctx = opentracing.ContextWithSpan(ctx, span)
+	}
+	req := tsoReqPool.Get().(*tsoRequest)
+	req.ctx = ctx
+	req.start = time.Now()
+	c.tsoRequests <- req
+
+	return req
+}
+
+// TSFuture is a future which promises to return a TSO.
+type TSFuture interface {
+	// Wait gets the physical and logical time, it would block caller if data is not available yet.
+	Wait() (int64, int64, error)
+}
+
+func (req *tsoRequest) Wait() (physical int64, logical int64, err error) {
+	// If tso command duration is observed very high, the reason could be it
+	// takes too long for Wait() be called.
+	select {
+	case err = <-req.done:
+		err = errors.WithStack(err)
+		defer tsoReqPool.Put(req)
+		if err != nil {
+			return 0, 0, err
+		}
+		physical, logical = req.physical, req.logical
+		return
+	case <-req.ctx.Done():
+		return 0, 0, errors.WithStack(req.ctx.Err())
+	}
+}
+
+func (c *client) GetTS(ctx context.Context) (physical int64, logical int64, err error) {
+	resp := c.GetTSAsync(ctx)
+	return resp.Wait()
+}
+
+
+
+func (c *client) requestHeader() *pdpb.RequestHeader {
+	return &pdpb.RequestHeader{
+		ClusterId: c.clusterID,
+	}
+}
+
+func addrsToUrls(addrs []string) []string {
+	// Add default schema "http://" to addrs.
+	urls := make([]string, 0, len(addrs))
+	for _, addr := range addrs {
+		if strings.Contains(addr, "://") {
+			urls = append(urls, addr)
+		} else {
+			urls = append(urls, "http://"+addr)
+		}
+	}
+	return urls
+}

master/config/config.go

+// Copyright 2016 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+import (
+	"bytes"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"github.com/czs007/suvlim/util/grpcutil"
+	//"google.golang.org/grpc"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/BurntSushi/toml"
+	"github.com/czs007/suvlim/errors"
+	"github.com/czs007/suvlim/util/typeutil"
+	"github.com/pingcap/log"
+	"go.etcd.io/etcd/embed"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+)
+
+// Config is the pd server configuration.
+type Config struct {
+	flagSet *flag.FlagSet
+	Version bool `json:"-"`
+	ConfigCheck bool `json:"-"`
+
+	ClientUrls          string `toml:"client-urls" json:"client-urls"`
+	PeerUrls            string `toml:"peer-urls" json:"peer-urls"`
+	AdvertiseClientUrls string `toml:"advertise-client-urls" json:"advertise-client-urls"`
+	AdvertisePeerUrls   string `toml:"advertise-peer-urls" json:"advertise-peer-urls"`
+
+
+	Name    string `toml:"name" json:"name"`
+	DataDir string     `toml:"data-dir" json:"data-dir"`
+	EnableGRPCGateway bool   `json:"enable-grpc-gateway"`
+
+	InitialCluster      string `toml:"initial-cluster" json:"initial-cluster"`
+	InitialClusterState string `toml:"initial-cluster-state" json:"initial-cluster-state"`
+	InitialClusterToken string `toml:"initial-cluster-token" json:"initial-cluster-token"`
+
+
+	LeaderLease int64 `toml:"lease" json:"lease"`
+	Log     log.Config `toml:"log" json:"log"`
+	LogFileDeprecated  string `toml:"log-file" json:"log-file,omitempty"`
+	LogLevelDeprecated string `toml:"log-level" json:"log-level,omitempty"`
+
+	PDServerCfg PDServerConfig `toml:"pd-server" json:"pd-server"`
+
+	TickInterval typeutil.Duration `toml:"tick-interval"`
+	ElectionInterval typeutil.Duration `toml:"election-interval"`
+	DisableStrictReconfigCheck bool
+
+	// TsoSaveInterval is the interval to save timestamp.
+	TsoSaveInterval typeutil.Duration `toml:"tso-save-interval" json:"tso-save-interval"`
+
+	PreVote bool `toml:"enable-prevote"`
+	Security grpcutil.SecurityConfig `toml:"security" json:"security"`
+
+	configFile string
+
+	// For all warnings during parsing.
+	WarningMsgs []string
+
+	HeartbeatStreamBindInterval typeutil.Duration
+
+	logger   *zap.Logger
+	logProps *log.ZapProperties
+
+	//ServiceRegister func(*grpc.Server) `json:"-"`
+}
+
+// NewConfig creates a new config.
+func NewConfig() *Config {
+	cfg := &Config{}
+	cfg.flagSet = flag.NewFlagSet("pd", flag.ContinueOnError)
+	fs := cfg.flagSet
+
+	fs.BoolVar(&cfg.Version, "V", false, "print version information and exit")
+	fs.BoolVar(&cfg.Version, "version", false, "print version information and exit")
+	fs.StringVar(&cfg.configFile, "config", "", "config file")
+	fs.BoolVar(&cfg.ConfigCheck, "config-check", false, "check config file validity and exit")
+
+	fs.StringVar(&cfg.Name, "name", "", "human-readable name for this pd member")
+
+	fs.StringVar(&cfg.DataDir, "data-dir", "", "path to the data directory (default 'default.${name}')")
+	fs.StringVar(&cfg.ClientUrls, "client-urls", defaultClientUrls, "url for client traffic")
+	fs.StringVar(&cfg.AdvertiseClientUrls, "advertise-client-urls", "", "advertise url for client traffic (default '${client-urls}')")
+	fs.StringVar(&cfg.PeerUrls, "peer-urls", defaultPeerUrls, "url for peer traffic")
+	fs.StringVar(&cfg.AdvertisePeerUrls, "advertise-peer-urls", "", "advertise url for peer traffic (default '${peer-urls}')")
+	fs.StringVar(&cfg.InitialCluster, "initial-cluster", "", "initial cluster configuration for bootstrapping, e,g. pd=http://127.0.0.1:2380")
+
+	fs.StringVar(&cfg.Log.Level, "L", "", "log level: debug, info, warn, error, fatal (default 'info')")
+	fs.StringVar(&cfg.Log.File.Filename, "log-file", "", "log file path")
+
+	return cfg
+}
+
+
+const (
+	defaultLeaderLease             = int64(3)
+	defaultName                = "pd"
+	defaultClientUrls          = "http://127.0.0.1:2379"
+	defaultPeerUrls            = "http://127.0.0.1:2380"
+	defaultInitialClusterState = embed.ClusterStateFlagNew
+	defaultInitialClusterToken = "pd-cluster"
+
+	// etcd use 100ms for heartbeat and 1s for election timeout.
+	// We can enlarge both a little to reduce the network aggression.
+	// now embed etcd use TickMs for heartbeat, we will update
+	// after embed etcd decouples tick and heartbeat.
+	defaultTickInterval = 500 * time.Millisecond
+	// embed etcd has a check that `5 * tick > election`
+	defaultElectionInterval = 3000 * time.Millisecond
+
+	defaultHeartbeatStreamRebindInterval = time.Minute
+
+	defaultMaxResetTSGap    = 24 * time.Hour
+	defaultEnableGRPCGateway   = true
+	defaultDisableErrorVerbose = true
+)
+
+func init() {
+
+}
+
+func adjustString(v *string, defValue string) {
+	if len(*v) == 0 {
+		*v = defValue
+	}
+}
+
+func adjustUint64(v *uint64, defValue uint64) {
+	if *v == 0 {
+		*v = defValue
+	}
+}
+
+func adjustInt64(v *int64, defValue int64) {
+	if *v == 0 {
+		*v = defValue
+	}
+}
+
+func adjustFloat64(v *float64, defValue float64) {
+	if *v == 0 {
+		*v = defValue
+	}
+}
+
+func adjustDuration(v *typeutil.Duration, defValue time.Duration) {
+	if v.Duration == 0 {
+		v.Duration = defValue
+	}
+}
+
+func adjustPath(p *string) {
+	absPath, err := filepath.Abs(*p)
+	if err == nil {
+		*p = absPath
+	}
+}
+
+// Parse parses flag definitions from the argument list.
+func (c *Config) Parse(arguments []string) error {
+	// Parse first to get config file.
+	err := c.flagSet.Parse(arguments)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	// Load config file if specified.
+	var meta *toml.MetaData
+	if c.configFile != "" {
+		meta, err = c.configFromFile(c.configFile)
+		if err != nil {
+			return err
+		}
+
+		// Backward compatibility for toml config
+		if c.LogFileDeprecated != "" {
+			msg := fmt.Sprintf("log-file in %s is deprecated, use [log.file] instead", c.configFile)
+			c.WarningMsgs = append(c.WarningMsgs, msg)
+			if c.Log.File.Filename == "" {
+				c.Log.File.Filename = c.LogFileDeprecated
+			}
+		}
+		if c.LogLevelDeprecated != "" {
+			msg := fmt.Sprintf("log-level in %s is deprecated, use [log] instead", c.configFile)
+			c.WarningMsgs = append(c.WarningMsgs, msg)
+			if c.Log.Level == "" {
+				c.Log.Level = c.LogLevelDeprecated
+			}
+		}
+		if meta.IsDefined("schedule", "disable-raft-learner") {
+			msg := fmt.Sprintf("disable-raft-learner in %s is deprecated", c.configFile)
+			c.WarningMsgs = append(c.WarningMsgs, msg)
+		}
+		if meta.IsDefined("dashboard", "disable-telemetry") {
+			msg := fmt.Sprintf("disable-telemetry in %s is deprecated, use enable-telemetry instead", c.configFile)
+			c.WarningMsgs = append(c.WarningMsgs, msg)
+		}
+	}
+
+	// Parse again to replace with command line options.
+	err = c.flagSet.Parse(arguments)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	if len(c.flagSet.Args()) != 0 {
+		return errors.Errorf("'%s' is an invalid flag", c.flagSet.Arg(0))
+	}
+
+	err = c.Adjust(meta)
+	return err
+}
+
+// Validate is used to validate if some configurations are right.
+func (c *Config) Validate() error {
+	dataDir, err := filepath.Abs(c.DataDir)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	logFile, err := filepath.Abs(c.Log.File.Filename)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	rel, err := filepath.Rel(dataDir, filepath.Dir(logFile))
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	if !strings.HasPrefix(rel, "..") {
+		return errors.New("log directory shouldn't be the subdirectory of data directory")
+	}
+
+	return nil
+}
+
+// Utility to test if a configuration is defined.
+type configMetaData struct {
+	meta *toml.MetaData
+	path []string
+}
+
+func newConfigMetadata(meta *toml.MetaData) *configMetaData {
+	return &configMetaData{meta: meta}
+}
+
+func (m *configMetaData) IsDefined(key string) bool {
+	if m.meta == nil {
+		return false
+	}
+	keys := append([]string(nil), m.path...)
+	keys = append(keys, key)
+	return m.meta.IsDefined(keys...)
+}
+
+func (m *configMetaData) Child(path ...string) *configMetaData {
+	newPath := append([]string(nil), m.path...)
+	newPath = append(newPath, path...)
+	return &configMetaData{
+		meta: m.meta,
+		path: newPath,
+	}
+}
+
+func (m *configMetaData) CheckUndecoded() error {
+	if m.meta == nil {
+		return nil
+	}
+	undecoded := m.meta.Undecoded()
+	if len(undecoded) == 0 {
+		return nil
+	}
+	errInfo := "Config contains undefined item: "
+	for _, key := range undecoded {
+		errInfo += key.String() + ", "
+	}
+	return errors.New(errInfo[:len(errInfo)-2])
+}
+
+// Adjust is used to adjust the PD configurations.
+func (c *Config) Adjust(meta *toml.MetaData) error {
+	configMetaData := newConfigMetadata(meta)
+	if err := configMetaData.CheckUndecoded(); err != nil {
+		c.WarningMsgs = append(c.WarningMsgs, err.Error())
+	}
+
+	if c.Name == "" {
+		hostname, err := os.Hostname()
+		if err != nil {
+			return err
+		}
+		adjustString(&c.Name, fmt.Sprintf("%s-%s", defaultName, hostname))
+	}
+	adjustString(&c.DataDir, fmt.Sprintf("default.%s", c.Name))
+	adjustPath(&c.DataDir)
+
+	if err := c.Validate(); err != nil {
+		return err
+	}
+
+	adjustString(&c.ClientUrls, defaultClientUrls)
+	adjustString(&c.AdvertiseClientUrls, c.ClientUrls)
+	adjustString(&c.PeerUrls, defaultPeerUrls)
+	adjustString(&c.AdvertisePeerUrls, c.PeerUrls)
+
+	if len(c.InitialCluster) == 0 {
+		// The advertise peer urls may be http://127.0.0.1:2380,http://127.0.0.1:2381
+		// so the initial cluster is pd=http://127.0.0.1:2380,pd=http://127.0.0.1:2381
+		items := strings.Split(c.AdvertisePeerUrls, ",")
+
+		sep := ""
+		for _, item := range items {
+			c.InitialCluster += fmt.Sprintf("%s%s=%s", sep, c.Name, item)
+			sep = ","
+		}
+	}
+
+	adjustString(&c.InitialClusterState, defaultInitialClusterState)
+	adjustString(&c.InitialClusterToken, defaultInitialClusterToken)
+
+	adjustInt64(&c.LeaderLease, defaultLeaderLease)
+
+	adjustDuration(&c.TsoSaveInterval, time.Duration(defaultLeaderLease)*time.Second)
+
+	adjustDuration(&c.TickInterval, defaultTickInterval)
+	adjustDuration(&c.ElectionInterval, defaultElectionInterval)
+
+
+	if err := c.PDServerCfg.adjust(configMetaData.Child("pd-server")); err != nil {
+		return err
+	}
+
+	c.adjustLog(configMetaData.Child("log"))
+	adjustDuration(&c.HeartbeatStreamBindInterval, defaultHeartbeatStreamRebindInterval)
+
+	if !configMetaData.IsDefined("enable-prevote") {
+		c.PreVote = true
+	}
+	if !configMetaData.IsDefined("enable-grpc-gateway") {
+		c.EnableGRPCGateway = defaultEnableGRPCGateway
+	}
+
+	return nil
+}
+
+func (c *Config) adjustLog(meta *configMetaData) {
+	if !meta.IsDefined("disable-error-verbose") {
+		c.Log.DisableErrorVerbose = defaultDisableErrorVerbose
+	}
+}
+
+// Clone returns a cloned configuration.
+func (c *Config) Clone() *Config {
+	cfg := &Config{}
+	*cfg = *c
+	return cfg
+}
+
+func (c *Config) String() string {
+	data, err := json.MarshalIndent(c, "", "  ")
+	if err != nil {
+		return "<nil>"
+	}
+	return string(data)
+}
+
+// configFromFile loads config from file.
+func (c *Config) configFromFile(path string) (*toml.MetaData, error) {
+	meta, err := toml.DecodeFile(path, c)
+	return &meta, errors.WithStack(err)
+}
+
+
+// PDServerConfig is the configuration for pd server.
+type PDServerConfig struct {
+	// MaxResetTSGap is the max gap to reset the tso.
+	MaxResetTSGap typeutil.Duration `toml:"max-gap-reset-ts" json:"max-gap-reset-ts"`
+}
+
+func (c *PDServerConfig) adjust(meta *configMetaData) error {
+	adjustDuration(&c.MaxResetTSGap, defaultMaxResetTSGap)
+	return nil
+}
+
+// Clone returns a cloned PD server config.
+func (c *PDServerConfig) Clone() *PDServerConfig {
+	return &PDServerConfig{
+		MaxResetTSGap:    c.MaxResetTSGap,
+	}
+}
+
+
+// ParseUrls parse a string into multiple urls.
+// Export for api.
+func ParseUrls(s string) ([]url.URL, error) {
+	items := strings.Split(s, ",")
+	urls := make([]url.URL, 0, len(items))
+	for _, item := range items {
+		u, err := url.Parse(item)
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+
+		urls = append(urls, *u)
+	}
+
+	return urls, nil
+}
+
+// SetupLogger setup the logger.
+func (c *Config) SetupLogger() error {
+	lg, p, err := log.InitLogger(&c.Log, zap.AddStacktrace(zapcore.FatalLevel))
+	if err != nil {
+		return err
+	}
+	c.logger = lg
+	c.logProps = p
+	return nil
+}
+
+// GetZapLogger gets the created zap logger.
+func (c *Config) GetZapLogger() *zap.Logger {
+	return c.logger
+}
+
+// GetZapLogProperties gets properties of the zap logger.
+func (c *Config) GetZapLogProperties() *log.ZapProperties {
+	return c.logProps
+}
+
+// GetConfigFile gets the config file.
+func (c *Config) GetConfigFile() string {
+	return c.configFile
+}
+
+// RewriteFile rewrites the config file after updating the config.
+func (c *Config) RewriteFile(new *Config) error {
+	filePath := c.GetConfigFile()
+	if filePath == "" {
+		return nil
+	}
+	var buf bytes.Buffer
+	if err := toml.NewEncoder(&buf).Encode(*new); err != nil {
+		return err
+	}
+	dir := filepath.Dir(filePath)
+	tmpfile := filepath.Join(dir, "tmp_pd.toml")
+
+	f, err := os.Create(tmpfile)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	if _, err := f.Write(buf.Bytes()); err != nil {
+		return err
+	}
+	if err := f.Sync(); err != nil {
+		return err
+	}
+	return os.Rename(tmpfile, filePath)
+}
+
+func (c *Config) GenEmbedEtcdConfig() (*embed.Config, error) {
+	cfg := embed.NewConfig()
+	cfg.Name = c.Name
+	cfg.Dir = c.DataDir
+	cfg.WalDir = ""
+	cfg.InitialCluster = c.InitialCluster
+	cfg.ClusterState = c.InitialClusterState
+	cfg.InitialClusterToken = c.InitialClusterToken
+	cfg.EnablePprof = true
+	cfg.PreVote = c.PreVote
+	cfg.StrictReconfigCheck = !c.DisableStrictReconfigCheck
+	cfg.TickMs = uint(c.TickInterval.Duration / time.Millisecond)
+	cfg.ElectionMs = uint(c.ElectionInterval.Duration / time.Millisecond)
+
+	allowedCN, serr := c.Security.GetOneAllowedCN()
+	if serr != nil {
+		return nil, serr
+	}
+	cfg.ClientTLSInfo.ClientCertAuth = len(c.Security.CAPath) != 0
+	cfg.ClientTLSInfo.TrustedCAFile = c.Security.CAPath
+	cfg.ClientTLSInfo.CertFile = c.Security.CertPath
+	cfg.ClientTLSInfo.KeyFile = c.Security.KeyPath
+	// Client no need to set the CN. (cfg.ClientTLSInfo.AllowedCN = allowedCN)
+	cfg.PeerTLSInfo.ClientCertAuth = len(c.Security.CAPath) != 0
+	cfg.PeerTLSInfo.TrustedCAFile = c.Security.CAPath
+	cfg.PeerTLSInfo.CertFile = c.Security.CertPath
+	cfg.PeerTLSInfo.KeyFile = c.Security.KeyPath
+	cfg.PeerTLSInfo.AllowedCN = allowedCN
+	cfg.ZapLoggerBuilder = embed.NewZapCoreLoggerBuilder(c.logger, c.logger.Core(), c.logProps.Syncer)
+	cfg.EnableGRPCGateway = c.EnableGRPCGateway
+	cfg.EnableV2 = true
+	cfg.Logger = "zap"
+	var err error
+
+	cfg.LPUrls, err = ParseUrls(c.PeerUrls)
+	if err != nil {
+		return nil, err
+	}
+
+	cfg.APUrls, err = ParseUrls(c.AdvertisePeerUrls)
+	if err != nil {
+		return nil, err
+	}
+
+	cfg.LCUrls, err = ParseUrls(c.ClientUrls)
+	if err != nil {
+		return nil, err
+	}
+
+	cfg.ACUrls, err = ParseUrls(c.AdvertiseClientUrls)
+	if err != nil {
+		return nil, err
+	}
+
+	return cfg, nil
+}

master/config/persist_options.go

+// Copyright 2017 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+import (
+	"sync/atomic"
+	"time"
+	//"github.com/czs007/suvlim/master/kv"
+	"github.com/czs007/suvlim/master/meta"
+
+)
+
+// PersistOptions wraps all configurations that need to persist to storage and
+// allows to access them safely.
+type PersistOptions struct {
+	pdServerConfig  atomic.Value
+}
+
+// NewPersistOptions creates a new PersistOptions instance.
+func NewPersistOptions(cfg *Config) *PersistOptions {
+	o := &PersistOptions{}
+	o.pdServerConfig.Store(&cfg.PDServerCfg)
+	return o
+}
+
+// GetPDServerConfig returns pd server configurations.
+func (o *PersistOptions) GetPDServerConfig() *PDServerConfig {
+	return o.pdServerConfig.Load().(*PDServerConfig)
+}
+
+// SetPDServerConfig sets the PD configuration.
+func (o *PersistOptions) SetPDServerConfig(cfg *PDServerConfig) {
+	o.pdServerConfig.Store(cfg)
+}
+
+
+// GetMaxResetTSGap gets the max gap to reset the tso.
+func (o *PersistOptions) GetMaxResetTSGap() time.Duration {
+	return o.GetPDServerConfig().MaxResetTSGap.Duration
+}
+
+
+// Persist saves the configuration to the storage.
+func (o *PersistOptions) Persist(storage *meta.Storage) error {
+	cfg := &Config{
+		PDServerCfg:     *o.GetPDServerConfig(),
+	}
+	err := storage.SaveConfig(cfg)
+	return err
+}
+
+// Reload reloads the configuration from the storage.
+func (o *PersistOptions) Reload(storage *meta.Storage) error {
+	cfg := &Config{}
+	// pass nil to initialize cfg to default values (all items undefined)
+	cfg.Adjust(nil)
+
+	isExist, err := storage.LoadConfig(cfg)
+	if err != nil {
+		return err
+	}
+	if isExist {
+		o.pdServerConfig.Store(&cfg.PDServerCfg)
+	}
+	return nil
+}

master/config/util.go

+// Copyright 2019 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+import (
+	"net/url"
+	"regexp"
+
+	"github.com/czs007/suvlim/errors"
+	"github.com/czs007/suvlim/pkg/metapb"
+)
+
+const (
+	// Label key consists of alphanumeric characters, '-', '_', '.' or '/', and must start and end with an
+	// alphanumeric character. If can also contain an extra '$' at the beginning.
+	keyFormat = "^[$]?[A-Za-z0-9]([-A-Za-z0-9_./]*[A-Za-z0-9])?$"
+	// Value key can be any combination of alphanumeric characters, '-', '_', '.' or '/'. It can also be empty to
+	// mark the label as deleted.
+	valueFormat = "^[-A-Za-z0-9_./]*$"
+)
+
+func validateFormat(s, format string) error {
+	isValid, _ := regexp.MatchString(format, s)
+	if !isValid {
+		return errors.Errorf("%s does not match format %q", s, format)
+	}
+	return nil
+}
+
+// ValidateLabels checks the legality of the labels.
+func ValidateLabels(labels []*metapb.StoreLabel) error {
+	for _, label := range labels {
+		if err := validateFormat(label.Key, keyFormat); err != nil {
+			return err
+		}
+		if err := validateFormat(label.Value, valueFormat); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// ValidateURLWithScheme checks the format of the URL.
+func ValidateURLWithScheme(rawURL string) error {
+	u, err := url.ParseRequestURI(rawURL)
+	if err != nil {
+		return err
+	}
+	if u.Scheme == "" || u.Host == "" {
+		return errors.Errorf("%s has no scheme", rawURL)
+	}
+	return nil
+}

master/config/util_test.go

+// Copyright 2018 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+import (
+	"github.com/czs007/suvlim/pkg/metapb"
+	. "github.com/pingcap/check"
+)
+
+var _ = Suite(&testUtilSuite{})
+
+type testUtilSuite struct{}
+
+func (s *testUtilSuite) TestValidateLabels(c *C) {
+	tests := []struct {
+		label  string
+		hasErr bool
+	}{
+		{"z1", false},
+		{"z-1", false},
+		{"h1;", true},
+		{"z_1", false},
+		{"z_1&", true},
+		{"cn", false},
+		{"Zo^ne", true},
+		{"z_", true},
+		{"hos&t-15", true},
+		{"_test1", true},
+		{"-test1", true},
+		{"192.168.199.1", false},
+		{"www.pingcap.com", false},
+		{"h_127.0.0.1", false},
+		{"a", false},
+		{"a/b", false},
+		{"ab/", true},
+		{"/ab", true},
+		{"$abc", false},
+		{"$", true},
+		{"a$b", true},
+		{"$$", true},
+	}
+	for _, t := range tests {
+		c.Assert(ValidateLabels([]*metapb.StoreLabel{{Key: t.label}}) != nil, Equals, t.hasErr)
+	}
+}
+
+func (s *testUtilSuite) TestValidateURLWithScheme(c *C) {
+	tests := []struct {
+		addr   string
+		hasErr bool
+	}{
+		{"", true},
+		{"foo", true},
+		{"/foo", true},
+		{"http", true},
+		{"http://", true},
+		{"http://foo", false},
+		{"https://foo", false},
+		{"http://127.0.0.1", false},
+		{"http://127.0.0.1/", false},
+		{"https://foo.com/bar", false},
+		{"https://foo.com/bar/", false},
+	}
+	for _, t := range tests {
+		c.Assert(ValidateURLWithScheme(t.addr) != nil, Equals, t.hasErr)
+	}
+}

master/election/leadership.go

+// Copyright 2020 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package election
+
+import (
+	"context"
+	"sync/atomic"
+	"github.com/czs007/suvlim/errors"
+	"github.com/czs007/suvlim/pkg/pdpb"
+	"github.com/czs007/suvlim/util/etcdutil"
+	"github.com/czs007/suvlim/master/kv"
+	"go.etcd.io/etcd/clientv3"
+)
+
+// GetLeader gets the corresponding leader from etcd by given leaderPath (as the key).
+func GetLeader(c *clientv3.Client, leaderPath string) (*pdpb.Member, int64, error) {
+	leader := &pdpb.Member{}
+	ok, rev, err := etcdutil.GetProtoMsgWithModRev(c, leaderPath, leader)
+	if err != nil {
+		return nil, 0, err
+	}
+	if !ok {
+		return nil, 0, nil
+	}
+
+	return leader, rev, nil
+}
+
+// Leadership is used to manage the leadership campaigning.
+type Leadership struct {
+	// purpose is used to show what this election for
+	purpose string
+	// The lease which is used to get this leadership
+	lease  atomic.Value // stored as *lease
+	client *clientv3.Client
+	// leaderKey and leaderValue are key-value pair in etcd
+	leaderKey   string
+	leaderValue string
+}
+
+// NewLeadership creates a new Leadership.
+func NewLeadership(client *clientv3.Client, leaderKey, purpose string) *Leadership {
+	leadership := &Leadership{
+		purpose:   purpose,
+		client:    client,
+		leaderKey: leaderKey,
+	}
+	return leadership
+}
+
+// getLease gets the lease of leadership, only if leadership is valid,
+// i.e the owner is a true leader, the lease is not nil.
+func (ls *Leadership) getLease() *lease {
+	l := ls.lease.Load()
+	if l == nil {
+		return nil
+	}
+	return l.(*lease)
+}
+
+func (ls *Leadership) setLease(lease *lease) {
+	ls.lease.Store(lease)
+}
+
+// GetClient is used to get the etcd client.
+func (ls *Leadership) GetClient() *clientv3.Client {
+	return ls.client
+}
+
+// Keep will keep the leadership available by update the lease's expired time continuously
+func (ls *Leadership) Keep(ctx context.Context) {
+	ls.getLease().KeepAlive(ctx)
+}
+
+// Check returns whether the leadership is still available
+func (ls *Leadership) Check() bool {
+	return ls != nil && ls.getLease() != nil && !ls.getLease().IsExpired()
+}
+
+// LeaderTxn returns txn() with a leader comparison to guarantee that
+// the transaction can be executed only if the server is leader.
+func (ls *Leadership) LeaderTxn(cs ...clientv3.Cmp) clientv3.Txn {
+	txn := kv.NewSlowLogTxn(ls.client)
+	return txn.If(append(cs, ls.leaderCmp())...)
+}
+
+func (ls *Leadership) leaderCmp() clientv3.Cmp {
+	return clientv3.Compare(clientv3.Value(ls.leaderKey), "=", ls.leaderValue)
+}
+
+// DeleteLeader deletes the corresponding leader from etcd by given leaderPath (as the key).
+func (ls *Leadership) DeleteLeader() error {
+	// delete leader itself and let others start a new election again.
+	resp, err := ls.LeaderTxn().Then(clientv3.OpDelete(ls.leaderKey)).Commit()
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	if !resp.Succeeded {
+		return errors.New("resign leader failed, we are not leader already")
+	}
+
+	return nil
+}
+
+// Reset does some defer job such as closing lease, resetting lease etc.
+func (ls *Leadership) Reset() {
+	ls.getLease().Close()
+}

master/election/lease.go

+// Copyright 2019 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package election
+
+import (
+	"context"
+	"sync/atomic"
+	"time"
+
+	"github.com/czs007/suvlim/errors"
+	"github.com/pingcap/log"
+	"github.com/tikv/pd/pkg/etcdutil"
+	"go.etcd.io/etcd/clientv3"
+	"go.uber.org/zap"
+)
+
+const (
+	revokeLeaseTimeout = time.Second
+	requestTimeout     = etcdutil.DefaultRequestTimeout
+	slowRequestTime    = etcdutil.DefaultSlowRequestTime
+)
+
+// lease is used as the low-level mechanism for campaigning and renewing elected leadership.
+// The way to gain and maintain leadership is to update and keep the lease alive continuously.
+type lease struct {
+	// purpose is used to show what this election for
+	Purpose string
+	// etcd client and lease
+	client *clientv3.Client
+	lease  clientv3.Lease
+	ID     clientv3.LeaseID
+	// leaseTimeout and expireTime are used to control the lease's lifetime
+	leaseTimeout time.Duration
+	expireTime   atomic.Value
+}
+
+// Grant uses `lease.Grant` to initialize the lease and expireTime.
+func (l *lease) Grant(leaseTimeout int64) error {
+	start := time.Now()
+	ctx, cancel := context.WithTimeout(l.client.Ctx(), requestTimeout)
+	leaseResp, err := l.lease.Grant(ctx, leaseTimeout)
+	cancel()
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	if cost := time.Since(start); cost > slowRequestTime {
+		log.Warn("lease grants too slow", zap.Duration("cost", cost), zap.String("purpose", l.Purpose))
+	}
+	log.Info("lease granted", zap.Int64("lease-id", int64(leaseResp.ID)), zap.Int64("lease-timeout", leaseTimeout), zap.String("purpose", l.Purpose))
+	l.ID = leaseResp.ID
+	l.leaseTimeout = time.Duration(leaseTimeout) * time.Second
+	l.expireTime.Store(start.Add(time.Duration(leaseResp.TTL) * time.Second))
+	return nil
+}
+
+// Close releases the lease.
+func (l *lease) Close() error {
+	// Reset expire time.
+	l.expireTime.Store(time.Time{})
+	// Try to revoke lease to make subsequent elections faster.
+	ctx, cancel := context.WithTimeout(l.client.Ctx(), revokeLeaseTimeout)
+	defer cancel()
+	l.lease.Revoke(ctx, l.ID)
+	return l.lease.Close()
+}
+
+// IsExpired checks if the lease is expired. If it returns true,
+// current leader should step down and try to re-elect again.
+func (l *lease) IsExpired() bool {
+	if l.expireTime.Load() == nil {
+		return false
+	}
+	return time.Now().After(l.expireTime.Load().(time.Time))
+}
+
+// KeepAlive auto renews the lease and update expireTime.
+func (l *lease) KeepAlive(ctx context.Context) {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	timeCh := l.keepAliveWorker(ctx, l.leaseTimeout/3)
+
+	var maxExpire time.Time
+	for {
+		select {
+		case t := <-timeCh:
+			if t.After(maxExpire) {
+				maxExpire = t
+				l.expireTime.Store(t)
+			}
+		case <-time.After(l.leaseTimeout):
+			log.Info("lease timeout", zap.Time("expire", l.expireTime.Load().(time.Time)), zap.String("purpose", l.Purpose))
+			return
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// Periodically call `lease.KeepAliveOnce` and post back latest received expire time into the channel.
+func (l *lease) keepAliveWorker(ctx context.Context, interval time.Duration) <-chan time.Time {
+	ch := make(chan time.Time)
+
+	go func() {
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+
+		log.Info("start lease keep alive worker", zap.Duration("interval", interval), zap.String("purpose", l.Purpose))
+		defer log.Info("stop lease keep alive worker", zap.String("purpose", l.Purpose))
+
+		for {
+			go func() {
+				start := time.Now()
+				ctx1, cancel := context.WithTimeout(ctx, l.leaseTimeout)
+				defer cancel()
+				res, err := l.lease.KeepAliveOnce(ctx1, l.ID)
+				if err != nil {
+					log.Warn("lease keep alive failed", zap.Error(err), zap.String("purpose", l.Purpose))
+					return
+				}
+				if res.TTL > 0 {
+					expire := start.Add(time.Duration(res.TTL) * time.Second)
+					select {
+					case ch <- expire:
+					case <-ctx1.Done():
+					}
+				}
+			}()
+
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
+		}
+	}()
+
+	return ch
+}

master/grpc_service.go

+// Copyright 2017 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"context"
+	"io"
+	//"strconv"
+	"sync/atomic"
+	"time"
+
+	"github.com/czs007/suvlim/pkg/pdpb"
+	"github.com/czs007/suvlim/errors"
+	"github.com/pingcap/log"
+	//"github.com/czs007/suvlim/util/tsoutil"
+	//"github.com/tikv/pd/server/cluster"
+	//"github.com/tikv/pd/server/core"
+	//"github.com/tikv/pd/server/versioninfo"
+	"go.uber.org/zap"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+const slowThreshold = 5 * time.Millisecond
+
+// gRPC errors
+var (
+	// ErrNotLeader is returned when current server is not the leader and not possible to process request.
+	// TODO: work as proxy.
+	ErrNotLeader  = status.Errorf(codes.Unavailable, "not leader")
+	ErrNotStarted = status.Errorf(codes.Unavailable, "server not started")
+)
+
+// Tso implements gRPC PDServer.
+func (s *Server) Tso(stream pdpb.PD_TsoServer) error {
+	for {
+		request, err := stream.Recv()
+		if err == io.EOF {
+			return nil
+		}
+		if err != nil {
+			return errors.WithStack(err)
+		}
+		start := time.Now()
+		// TSO uses leader lease to determine validity. No need to check leader here.
+		if s.IsClosed() {
+			return status.Errorf(codes.Unknown, "server not started")
+		}
+		if request.GetHeader().GetClusterId() != s.clusterID {
+			return status.Errorf(codes.FailedPrecondition, "mismatch cluster id, need %d but got %d", s.clusterID, request.GetHeader().GetClusterId())
+		}
+		count := request.GetCount()
+		ts, err := s.tsoAllocator.GenerateTSO(count)
+		if err != nil {
+			return status.Errorf(codes.Unknown, err.Error())
+		}
+
+		elapsed := time.Since(start)
+		if elapsed > slowThreshold {
+			log.Warn("get timestamp too slow", zap.Duration("cost", elapsed))
+		}
+		response := &pdpb.TsoResponse{
+			Header:    s.header(),
+			Timestamp: &ts,
+			Count:     count,
+		}
+		if err := stream.Send(response); err != nil {
+			return errors.WithStack(err)
+		}
+	}
+}
+
+
+// AllocID implements gRPC PDServer.
+func (s *Server) AllocID(ctx context.Context, request *pdpb.AllocIDRequest) (*pdpb.AllocIDResponse, error) {
+	if err := s.validateRequest(request.GetHeader()); err != nil {
+		return nil, err
+	}
+
+	// We can use an allocator for all types ID allocation.
+	id, err := s.idAllocator.Alloc()
+	if err != nil {
+		return nil, status.Errorf(codes.Unknown, err.Error())
+	}
+
+	return &pdpb.AllocIDResponse{
+		Header: s.header(),
+		Id:     id,
+	}, nil
+}
+
+
+const heartbeatSendTimeout = 5 * time.Second
+
+var errSendHeartbeatTimeout = errors.New("send region heartbeat timeout")
+
+// heartbeatServer wraps PD_RegionHeartbeatServer to ensure when any error
+// occurs on Send() or Recv(), both endpoints will be closed.
+type heartbeatServer struct {
+	stream pdpb.PD_HeartbeatServer
+	closed int32
+}
+
+func (s *heartbeatServer) Send(m *pdpb.HeartbeatResponse) error {
+	if atomic.LoadInt32(&s.closed) == 1 {
+		return io.EOF
+	}
+	done := make(chan error, 1)
+	go func() { done <- s.stream.Send(m) }()
+	select {
+	case err := <-done:
+		if err != nil {
+			atomic.StoreInt32(&s.closed, 1)
+		}
+		return errors.WithStack(err)
+	case <-time.After(heartbeatSendTimeout):
+		atomic.StoreInt32(&s.closed, 1)
+		return errors.WithStack(errSendHeartbeatTimeout)
+	}
+}
+
+func (s *heartbeatServer) Recv() (*pdpb.HeartbeatRequest, error) {
+	if atomic.LoadInt32(&s.closed) == 1 {
+		return nil, io.EOF
+	}
+	req, err := s.stream.Recv()
+	if err != nil {
+		atomic.StoreInt32(&s.closed, 1)
+		return nil, errors.WithStack(err)
+	}
+	return req, nil
+}
+
+// RegionHeartbeat implements gRPC PDServer.
+func (s *Server) Heartbeat(stream pdpb.PD_HeartbeatServer) error {
+	server := &heartbeatServer{stream: stream}
+
+	for {
+		request, err := server.Recv()
+		if err == io.EOF {
+			return nil
+		}
+		if err != nil {
+			return errors.WithStack(err)
+		}
+
+		if err = s.validateRequest(request.GetHeader()); err != nil {
+			return err
+		}
+		//msg:= "OK"
+		//s.hbStreams.SendMsg(msg)
+	}
+}
+
+// validateRequest checks if Server is leader and clusterID is matched.
+// TODO: Call it in gRPC interceptor.
+func (s *Server) validateRequest(header *pdpb.RequestHeader) error {
+	if s.IsClosed() {
+		return errors.WithStack(ErrNotLeader)
+	}
+	if header.GetClusterId() != s.clusterID {
+		return status.Errorf(codes.FailedPrecondition, "mismatch cluster id, need %d but got %d", s.clusterID, header.GetClusterId())
+	}
+	return nil
+}
+
+func (s *Server) header() *pdpb.ResponseHeader {
+	return &pdpb.ResponseHeader{ClusterId: s.clusterID}
+}
+
+func (s *Server) errorHeader(err *pdpb.Error) *pdpb.ResponseHeader {
+	return &pdpb.ResponseHeader{
+		ClusterId: s.clusterID,
+		Error:     err,
+	}
+}
+
+func (s *Server) notBootstrappedHeader() *pdpb.ResponseHeader {
+	return s.errorHeader(&pdpb.Error{
+		Type:    pdpb.ErrorType_NOT_BOOTSTRAPPED,
+		Message: "cluster is not bootstrapped",
+	})
+}
+
+

master/heartbeat_streams.go

+// Copyright 2017 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/czs007/suvlim/util/logutil"
+	"github.com/czs007/suvlim/pkg/pdpb"
+)
+
+// HeartbeatStream is an interface.
+type HeartbeatStream interface {
+	Send(*pdpb.HeartbeatResponse) error
+}
+
+// HeartbeatStreams is an interface of async region heartbeat.
+type HeartbeatStreams interface {
+	SendMsg(msg *pdpb.HeartbeatResponse)
+	BindStream(peerID uint64, stream HeartbeatStream)
+}
+
+const (
+	heartbeatStreamKeepAliveInterval = time.Minute
+	heartbeatChanCapacity            = 1024
+)
+
+type streamUpdate struct {
+	peerID uint64
+	stream HeartbeatStream
+}
+
+type heartbeatStreams struct {
+	wg             sync.WaitGroup
+	hbStreamCtx    context.Context
+	hbStreamCancel context.CancelFunc
+	clusterID      uint64
+	streams        map[uint64]HeartbeatStream
+	msgCh          chan *pdpb.HeartbeatResponse
+	streamCh       chan streamUpdate
+}
+
+func newHeartbeatStreams(ctx context.Context, clusterID uint64) *heartbeatStreams {
+	hbStreamCtx, hbStreamCancel := context.WithCancel(ctx)
+	hs := &heartbeatStreams{
+		hbStreamCtx:    hbStreamCtx,
+		hbStreamCancel: hbStreamCancel,
+		clusterID:      clusterID,
+		streams:        make(map[uint64]HeartbeatStream),
+		msgCh:          make(chan *pdpb.HeartbeatResponse, heartbeatChanCapacity),
+		streamCh:       make(chan streamUpdate, 1),
+	}
+	hs.wg.Add(1)
+	go hs.run()
+	return hs
+}
+
+func (s *heartbeatStreams) run() {
+	defer logutil.LogPanic()
+
+	defer s.wg.Done()
+
+	keepAliveTicker := time.NewTicker(heartbeatStreamKeepAliveInterval)
+	defer keepAliveTicker.Stop()
+
+	//keepAlive := &pdpb.HeartbeatResponse{Header: &pdpb.ResponseHeader{ClusterId: s.clusterID}}
+
+	for {
+		select {
+		case update := <-s.streamCh:
+			s.streams[update.peerID] = update.stream
+		case msg := <-s.msgCh:
+			println("msgCh", msg)
+		case <-keepAliveTicker.C:
+			println("keepAlive")
+		case <-s.hbStreamCtx.Done():
+			return
+		}
+	}
+}
+
+func (s *heartbeatStreams) Close() {
+	s.hbStreamCancel()
+	s.wg.Wait()
+}
+
+func (s *heartbeatStreams) BindStream(peerID uint64, stream HeartbeatStream) {
+	update := streamUpdate{
+		peerID: peerID,
+		stream: stream,
+	}
+	select {
+	case s.streamCh <- update:
+	case <-s.hbStreamCtx.Done():
+	}
+}
+
+func (s *heartbeatStreams) SendMsg(msg *pdpb.HeartbeatResponse) {
+	msg.Header = &pdpb.ResponseHeader{ClusterId: s.clusterID}
+	select {
+	case s.msgCh <- msg:
+	case <-s.hbStreamCtx.Done():
+	}
+}
+
+func (s *heartbeatStreams) sendErr(errType pdpb.ErrorType, errMsg string) {
+
+	msg := &pdpb.HeartbeatResponse{
+		Header: &pdpb.ResponseHeader{
+			ClusterId: s.clusterID,
+			Error: &pdpb.Error{
+				Type:    errType,
+				Message: errMsg,
+			},
+		},
+	}
+
+	select {
+	case s.msgCh <- msg:
+	case <-s.hbStreamCtx.Done():
+	}
+}

master/id/id.go

+// Copyright 2016 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package id
+
+import (
+	"path"
+	"sync"
+
+	"github.com/czs007/suvlim/errors"
+	"github.com/czs007/suvlim/master/kv"
+	"github.com/czs007/suvlim/util/etcdutil"
+	"github.com/czs007/suvlim/util/typeutil"
+	"github.com/pingcap/log"
+	"go.etcd.io/etcd/clientv3"
+	"go.uber.org/zap"
+)
+
+// Allocator is the allocator to generate unique ID.
+type Allocator interface {
+	Alloc() (uint64, error)
+}
+
+const allocStep = uint64(1000)
+
+// AllocatorImpl is used to allocate ID.
+type AllocatorImpl struct {
+	mu   sync.Mutex
+	base uint64
+	end  uint64
+
+	client   *clientv3.Client
+	rootPath string
+	member   string
+}
+
+// NewAllocatorImpl creates a new IDAllocator.
+func NewAllocatorImpl(client *clientv3.Client, rootPath string, member string) *AllocatorImpl {
+	return &AllocatorImpl{client: client, rootPath: rootPath, member: member}
+}
+
+// Alloc returns a new id.
+func (alloc *AllocatorImpl) Alloc() (uint64, error) {
+	alloc.mu.Lock()
+	defer alloc.mu.Unlock()
+
+	if alloc.base == alloc.end {
+		end, err := alloc.generate()
+		if err != nil {
+			return 0, err
+		}
+
+		alloc.end = end
+		alloc.base = alloc.end - allocStep
+	}
+
+	alloc.base++
+
+	return alloc.base, nil
+}
+
+func (alloc *AllocatorImpl) generate() (uint64, error) {
+	key := alloc.getAllocIDPath()
+	value, err := etcdutil.GetValue(alloc.client, key)
+	if err != nil {
+		return 0, err
+	}
+
+	var (
+		cmp clientv3.Cmp
+		end uint64
+	)
+
+	if value == nil {
+		// create the key
+		cmp = clientv3.Compare(clientv3.CreateRevision(key), "=", 0)
+	} else {
+		// update the key
+		end, err = typeutil.BytesToUint64(value)
+		if err != nil {
+			return 0, err
+		}
+
+		cmp = clientv3.Compare(clientv3.Value(key), "=", string(value))
+	}
+
+	end += allocStep
+	value = typeutil.Uint64ToBytes(end)
+	txn := kv.NewSlowLogTxn(alloc.client)
+	leaderPath := path.Join(alloc.rootPath, "leader")
+	t := txn.If(append([]clientv3.Cmp{cmp}, clientv3.Compare(clientv3.Value(leaderPath), "=", alloc.member))...)
+	resp, err := t.Then(clientv3.OpPut(key, string(value))).Commit()
+	if err != nil {
+		return 0, err
+	}
+	if !resp.Succeeded {
+		return 0, errors.New("generate id failed, we may not leader")
+	}
+
+	log.Info("idAllocator allocates a new id", zap.Uint64("alloc-id", end))
+	return end, nil
+}
+
+func (alloc *AllocatorImpl) getAllocIDPath() string {
+	return path.Join(alloc.rootPath, "alloc_id")
+}

master/kv/etcd_kv.go

+// Copyright 2016 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kv
+
+import (
+	"context"
+	"path"
+
+	//"strings"
+	"time"
+
+	"github.com/czs007/suvlim/util/etcdutil"
+	"github.com/pingcap/errors"
+	"github.com/pingcap/log"
+	"go.etcd.io/etcd/clientv3"
+	"go.uber.org/zap"
+)
+
+const (
+	requestTimeout  = 10 * time.Second
+	slowRequestTime = 1 * time.Second
+)
+
+var (
+	errTxnFailed = errors.New("failed to commit transaction")
+)
+
+type etcdKVBase struct {
+	client   *clientv3.Client
+	rootPath string
+}
+
+// NewEtcdKVBase creates a new etcd kv.
+func NewEtcdKVBase(client *clientv3.Client, rootPath string) *etcdKVBase {
+	return &etcdKVBase{
+		client:   client,
+		rootPath: rootPath,
+	}
+}
+
+func (kv *etcdKVBase) Load(key string) (string, error) {
+	key = path.Join(kv.rootPath, key)
+
+	resp, err := etcdutil.EtcdKVGet(kv.client, key)
+	if err != nil {
+		return "", err
+	}
+	if n := len(resp.Kvs); n == 0 {
+		return "", nil
+	} else if n > 1 {
+		return "", errors.Errorf("load more than one kvs: key %v kvs %v", key, n)
+	}
+	return string(resp.Kvs[0].Value), nil
+}
+
+func (kv *etcdKVBase) Save(key, value string) error {
+	key = path.Join(kv.rootPath, key)
+
+	txn := NewSlowLogTxn(kv.client)
+	resp, err := txn.Then(clientv3.OpPut(key, value)).Commit()
+	if err != nil {
+		log.Error("save to etcd meet error", zap.String("key", key), zap.String("value", value))
+		return errors.WithStack(err)
+	}
+	if !resp.Succeeded {
+		return errors.WithStack(errTxnFailed)
+	}
+	return nil
+}
+
+func (kv *etcdKVBase) Remove(key string) error {
+	key = path.Join(kv.rootPath, key)
+
+	txn := NewSlowLogTxn(kv.client)
+	resp, err := txn.Then(clientv3.OpDelete(key)).Commit()
+	if err != nil {
+		log.Error("remove from etcd meet error", zap.String("key", key))
+		return errors.WithStack(err)
+	}
+	if !resp.Succeeded {
+		return errors.WithStack(errTxnFailed)
+	}
+	return nil
+}
+
+// SlowLogTxn wraps etcd transaction and log slow one.
+type SlowLogTxn struct {
+	clientv3.Txn
+	cancel context.CancelFunc
+}
+
+// NewSlowLogTxn create a SlowLogTxn.
+func NewSlowLogTxn(client *clientv3.Client) clientv3.Txn {
+	ctx, cancel := context.WithTimeout(client.Ctx(), requestTimeout)
+	return &SlowLogTxn{
+		Txn:    client.Txn(ctx),
+		cancel: cancel,
+	}
+}
+
+// If takes a list of comparison. If all comparisons passed in succeed,
+// the operations passed into Then() will be executed. Or the operations
+// passed into Else() will be executed.
+func (t *SlowLogTxn) If(cs ...clientv3.Cmp) clientv3.Txn {
+	return &SlowLogTxn{
+		Txn:    t.Txn.If(cs...),
+		cancel: t.cancel,
+	}
+}
+
+// Then takes a list of operations. The Ops list will be executed, if the
+// comparisons passed in If() succeed.
+func (t *SlowLogTxn) Then(ops ...clientv3.Op) clientv3.Txn {
+	return &SlowLogTxn{
+		Txn:    t.Txn.Then(ops...),
+		cancel: t.cancel,
+	}
+}
+
+// Commit implements Txn Commit interface.
+func (t *SlowLogTxn) Commit() (*clientv3.TxnResponse, error) {
+	start := time.Now()
+	resp, err := t.Txn.Commit()
+	t.cancel()
+
+	cost := time.Since(start)
+	if cost > slowRequestTime {
+		log.Warn("txn runs too slow",
+			zap.Error(err),
+			zap.Reflect("response", resp),
+			zap.Duration("cost", cost))
+	}
+	//label := "success"
+	//if err != nil {
+	//	label = "failed"
+	//}
+	//txnCounter.WithLabelValues(label).Inc()
+	//txnDuration.WithLabelValues(label).Observe(cost.Seconds())
+	return resp, errors.WithStack(err)
+}

master/kv/kv.go

+// Copyright 2017 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kv
+
+// Base is an abstract interface for load/save pd cluster data.
+type Base interface {
+	Load(key string) (string, error)	
+	Save(key, value string) error
+	Remove(key string) error
+}

master/kv/mem_kv.go

+// Copyright 2017 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kv
+
+import (
+	"sync"
+
+	"github.com/google/btree"
+)
+
+type memoryKV struct {
+	sync.RWMutex
+	tree *btree.BTree
+}
+
+// NewMemoryKV returns an in-memory kvBase for testing.
+func NewMemoryKV() Base {
+	return &memoryKV{
+		tree: btree.New(2),
+	}
+}
+
+type memoryKVItem struct {
+	key, value string
+}
+
+func (s memoryKVItem) Less(than btree.Item) bool {
+	return s.key < than.(memoryKVItem).key
+}
+
+func (kv *memoryKV) Load(key string) (string, error) {
+	kv.RLock()
+	defer kv.RUnlock()
+	item := kv.tree.Get(memoryKVItem{key, ""})
+	if item == nil {
+		return "", nil
+	}
+	return item.(memoryKVItem).value, nil
+}
+
+func (kv *memoryKV) LoadRange(key, endKey string, limit int) ([]string, []string, error) {
+	kv.RLock()
+	defer kv.RUnlock()
+	keys := make([]string, 0, limit)
+	values := make([]string, 0, limit)
+	kv.tree.AscendRange(memoryKVItem{key, ""}, memoryKVItem{endKey, ""}, func(item btree.Item) bool {
+		keys = append(keys, item.(memoryKVItem).key)
+		values = append(values, item.(memoryKVItem).value)
+		if limit > 0 {
+			return len(keys) < limit
+		}
+		return true
+	})
+	return keys, values, nil
+}
+
+func (kv *memoryKV) Save(key, value string) error {
+	kv.Lock()
+	defer kv.Unlock()
+	kv.tree.ReplaceOrInsert(memoryKVItem{key, value})
+	return nil
+}
+
+func (kv *memoryKV) Remove(key string) error {
+	kv.Lock()
+	defer kv.Unlock()
+
+	kv.tree.Delete(memoryKVItem{key, ""})
+	return nil
+}

master/member/member.go

+// Copyright 2016 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package member
+
+import (
+	"strings"
+	"time"
+
+	"github.com/czs007/suvlim/master/election"
+
+	"github.com/czs007/suvlim/pkg/pdpb"
+
+	"github.com/czs007/suvlim/master/config"
+
+	"go.etcd.io/etcd/clientv3"
+	"go.etcd.io/etcd/embed"
+)
+
+const (
+	// The timeout to wait transfer etcd leader to complete.
+	moveLeaderTimeout = 5 * time.Second
+)
+
+// Member is used for the election related logic.
+type Member struct {
+	// etcd and cluster information.
+	leadership *election.Leadership
+	etcd       *embed.Etcd
+	client     *clientv3.Client
+	id         uint64       // etcd server id.
+	member     *pdpb.Member // current PD's info.
+	// memberValue is the serialized string of `member`. It will be save in
+	// etcd leader key when the PD node is successfully elected as the PD leader
+	// of the cluster. Every write will use it to check PD leadership.
+	memberValue string
+}
+
+// NewMember create a new Member.
+func NewMember(etcd *embed.Etcd, client *clientv3.Client, id uint64) *Member {
+	return &Member{
+		etcd:   etcd,
+		client: client,
+		id:     id,
+	}
+}
+
+// ID returns the unique etcd ID for this server in etcd cluster.
+func (m *Member) ID() uint64 {
+	return m.id
+}
+
+// MemberValue returns the member value.
+func (m *Member) MemberValue() string {
+	return m.memberValue
+}
+
+// Member returns the member.
+func (m *Member) Member() *pdpb.Member {
+	return m.member
+}
+
+// Etcd returns etcd related information.
+func (m *Member) Etcd() *embed.Etcd {
+	return m.etcd
+}
+
+// GetEtcdLeader returns the etcd leader ID.
+func (m *Member) GetEtcdLeader() uint64 {
+	return m.etcd.Server.Lead()
+}
+
+// GetLeadership returns the leadership of the PD member.
+func (m *Member) GetLeadership() *election.Leadership {
+	return m.leadership
+}
+
+// MemberInfo initializes the member info.
+func (m *Member) MemberInfo(cfg *config.Config, name string) {
+	leader := &pdpb.Member{
+		Name:       name,
+		MemberId:   m.ID(),
+		ClientUrls: strings.Split(cfg.AdvertiseClientUrls, ","),
+		PeerUrls:   strings.Split(cfg.AdvertisePeerUrls, ","),
+	}
+	m.member = leader
+}
+
+// Close gracefully shuts down all servers/listeners.
+func (m *Member) Close() {
+	m.Etcd().Close()
+}

master/meta/storage.go

+// Copyright 2017 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package meta
+
+import (
+	"encoding/json"
+	"path"
+	"sync"
+	"github.com/gogo/protobuf/proto"
+	"github.com/czs007/suvlim/errors"
+	"github.com/czs007/suvlim/pkg/metapb"
+	"github.com/czs007/suvlim/master/kv"
+)
+
+const (
+	clusterPath              = "raft"
+	configPath               = "config"
+)
+
+
+// Storage wraps all kv operations, keep it stateless.
+type Storage struct {
+	kv.Base
+	mu               sync.Mutex
+}
+
+// NewStorage creates Storage instance with Base.
+func NewStorage(base kv.Base) *Storage {
+	return &Storage{
+		Base: base,
+	}
+}
+
+// LoadMeta loads cluster meta from storage.
+func (s *Storage) LoadMeta(meta *metapb.Cluster) (bool, error) {
+	return loadProto(s.Base, clusterPath, meta)
+}
+
+// SaveMeta save cluster meta to storage.
+func (s *Storage) SaveMeta(meta *metapb.Cluster) error {
+	return saveProto(s.Base, clusterPath, meta)
+}
+
+func (s *Storage) SaveConfig(cfg interface{}) error {
+	value, err := json.Marshal(cfg)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	return s.Save(configPath, string(value))
+}
+
+// LoadConfig loads config from configPath then unmarshal it to cfg.
+func (s *Storage) LoadConfig(cfg interface{}) (bool, error) {
+	value, err := s.Load(configPath)
+	if err != nil {
+		return false, err
+	}
+	if value == "" {
+		return false, nil
+	}
+	err = json.Unmarshal([]byte(value), cfg)
+	if err != nil {
+		return false, errors.WithStack(err)
+	}
+	return true, nil
+}
+
+// SaveJSON saves json format data to storage.
+func (s *Storage) SaveJSON(prefix, key string, data interface{}) error {
+	value, err := json.Marshal(data)
+	if err != nil {
+		return err
+	}
+	return s.Save(path.Join(prefix, key), string(value))
+}
+
+
+// Close closes the s.
+func (s *Storage) Close() error {
+	return nil
+}
+
+func loadProto(s kv.Base, key string, msg proto.Message) (bool, error) {
+	value, err := s.Load(key)
+	if err != nil {
+		return false, err
+	}
+	if value == "" {
+		return false, nil
+	}
+	err = proto.Unmarshal([]byte(value), msg)
+	return true, errors.WithStack(err)
+}
+
+func saveProto(s kv.Base, key string, msg proto.Message) error {
+	value, err := proto.Marshal(msg)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	return s.Save(key, string(value))
+}

master/server.go

+// Copyright 2016 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"context"
+
+	"github.com/czs007/suvlim/master/kv"
+
+	//"fmt"
+	"math/rand"
+	"path"
+
+	"github.com/czs007/suvlim/master/member"
+
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/czs007/suvlim/errors"
+
+	//"github.com/czs007/suvlim/pkg/"
+	"github.com/czs007/suvlim/pkg/pdpb"
+	"github.com/czs007/suvlim/util/etcdutil"
+	"github.com/pingcap/log"
+
+	"github.com/czs007/suvlim/master/config"
+	"github.com/czs007/suvlim/master/id"
+	"github.com/czs007/suvlim/master/meta"
+	"github.com/czs007/suvlim/util/typeutil"
+	"github.com/czs007/suvlim/util/logutil"
+
+	//"github.com/czs007/suvlim/master/kv"
+	"github.com/czs007/suvlim/master/tso"
+	"go.etcd.io/etcd/clientv3"
+	"go.etcd.io/etcd/embed"
+	//"go.etcd.io/etcd/pkg/types"
+	"go.uber.org/zap"
+	"google.golang.org/grpc"
+)
+
+const (
+	etcdTimeout     = time.Second * 3
+	pdRootPath      = "/pd"
+	pdAPIPrefix     = "/pd/"
+	pdClusterIDPath = "/pd/cluster_id"
+)
+
+var (
+	// EnableZap enable the zap logger in embed etcd.
+	EnableZap = true
+	// EtcdStartTimeout the timeout of the startup etcd.
+	EtcdStartTimeout = time.Minute * 5
+)
+
+// Server is the pd server.
+type Server struct {
+	// Server state.
+	isServing int64
+
+	// Server start timestamp
+	startTimestamp int64
+
+	// Configs and initial fields.
+	cfg            *config.Config
+	etcdCfg        *embed.Config
+	persistOptions *config.PersistOptions
+
+	// for PD leader election.
+	member *member.Member
+
+	// Server services.
+	// for id allocator, we can use one allocator for
+	// store, region and peer, because we just need
+	// a unique ID.
+
+	// handler        *Handler
+
+	ctx              context.Context
+	serverLoopCtx    context.Context
+	serverLoopCancel func()
+	serverLoopWg     sync.WaitGroup
+
+	// etcd client
+	client    *clientv3.Client
+	clusterID uint64 // pd cluster id.
+	rootPath  string
+
+	// Server services.
+	// for id allocator, we can use one allocator for
+	// store, region and peer, because we just need
+	// a unique ID.
+	idAllocator *id.AllocatorImpl
+	storage     *meta.Storage
+
+	// for tso.
+	tsoAllocator tso.Allocator
+	//
+
+	hbStreams *heartbeatStreams
+
+	// Zap logger
+	lg       *zap.Logger
+	logProps *log.ZapProperties
+
+	// Add callback functions at different stages
+	startCallbacks []func()
+	closeCallbacks []func()
+}
+
+// CreateServer creates the UNINITIALIZED pd server with given configuration.
+func CreateServer(ctx context.Context, cfg *config.Config) (*Server, error) {
+	log.Info("PD Config", zap.Reflect("config", cfg))
+	rand.Seed(time.Now().UnixNano())
+
+	s := &Server{
+		cfg:            cfg,
+		persistOptions: config.NewPersistOptions(cfg),
+		member:         &member.Member{},
+		ctx:            ctx,
+		startTimestamp: time.Now().Unix(),
+	}
+
+	// s.handler = newHandler(s)
+
+	// Adjust etcd config.
+	etcdCfg, err := s.cfg.GenEmbedEtcdConfig()
+	if err != nil {
+		return nil, err
+	}
+
+	etcdCfg.ServiceRegister = func(gs *grpc.Server) {
+		pdpb.RegisterPDServer(gs, s)
+		//diagnosticspb.RegisterDiagnosticsServer(gs, s)
+	}
+	s.etcdCfg = etcdCfg
+
+	s.lg = cfg.GetZapLogger()
+	s.logProps = cfg.GetZapLogProperties()
+
+	s.lg = cfg.GetZapLogger()
+	s.logProps = cfg.GetZapLogProperties()
+	return s, nil
+}
+
+func (s *Server) startEtcd(ctx context.Context) error {
+	newCtx, cancel := context.WithTimeout(ctx, EtcdStartTimeout)
+	defer cancel()
+
+	etcd, err := embed.StartEtcd(s.etcdCfg)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	// Check cluster ID
+	//urlMap, err := types.NewURLsMap(s.cfg.InitialCluster)
+	//if err != nil {
+	//	return errors.WithStack(err)
+	//}
+	tlsConfig, err := s.cfg.Security.ToTLSConfig()
+	if err != nil {
+		return err
+	}
+
+	//if err = etcdutil.CheckClusterID(etcd.Server.Cluster().ID(), urlMap, tlsConfig); err != nil {
+	//	return err
+	//}
+
+	select {
+	// Wait etcd until it is ready to use
+	case <-etcd.Server.ReadyNotify():
+	case <-newCtx.Done():
+		return errors.Errorf("canceled when waiting embed etcd to be ready")
+	}
+
+	endpoints := []string{s.etcdCfg.ACUrls[0].String()}
+	log.Info("create etcd v3 client", zap.Strings("endpoints", endpoints), zap.Reflect("cert", s.cfg.Security))
+
+	client, err := clientv3.New(clientv3.Config{
+		Endpoints:   endpoints,
+		DialTimeout: etcdTimeout,
+		TLS:         tlsConfig,
+	})
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	etcdServerID := uint64(etcd.Server.ID())
+
+	// update advertise peer urls.
+	etcdMembers, err := etcdutil.ListEtcdMembers(client)
+	if err != nil {
+		return err
+	}
+	for _, m := range etcdMembers.Members {
+		if etcdServerID == m.ID {
+			etcdPeerURLs := strings.Join(m.PeerURLs, ",")
+			if s.cfg.AdvertisePeerUrls != etcdPeerURLs {
+				log.Info("update advertise peer urls", zap.String("from", s.cfg.AdvertisePeerUrls), zap.String("to", etcdPeerURLs))
+				s.cfg.AdvertisePeerUrls = etcdPeerURLs
+			}
+		}
+	}
+	s.client = client
+
+	return nil
+}
+
+// AddStartCallback adds a callback in the startServer phase.
+func (s *Server) AddStartCallback(callbacks ...func()) {
+	s.startCallbacks = append(s.startCallbacks, callbacks...)
+}
+
+func (s *Server) startServer(ctx context.Context) error {
+	var err error
+	if err = s.initClusterID(); err != nil {
+		return err
+	}
+	log.Info("init cluster id", zap.Uint64("cluster-id", s.clusterID))
+	// It may lose accuracy if use float64 to store uint64. So we store the
+	// cluster id in label.
+	//metadataGauge.WithLabelValues(fmt.Sprintf("cluster%d", s.clusterID)).Set(0)
+
+	s.rootPath = path.Join(pdRootPath, strconv.FormatUint(s.clusterID, 10))
+
+	s.idAllocator = id.NewAllocatorImpl(s.client, s.rootPath, s.member.MemberValue())
+
+	s.tsoAllocator = tso.NewGlobalTSOAllocator(
+		s.member.GetLeadership(),
+		s.rootPath,
+		s.cfg.TsoSaveInterval.Duration,
+		func() time.Duration { return s.persistOptions.GetMaxResetTSGap() },
+	)
+	kvBase := kv.NewEtcdKVBase(s.client, s.rootPath)
+	// path := filepath.Join(s.cfg.DataDir, "region-meta")
+
+	// regionStorage, err := core.NewRegionStorage(ctx, path)
+	// if err != nil {
+	// return err
+	// }
+	s.storage = meta.NewStorage(kvBase)
+	// s.basicCluster = core.NewBasicCluster()
+	// s.cluster = cluster.NewRaftCluster(ctx, s.GetClusterRootPath(), s.clusterID, syncer.NewRegionSyncer(s), s.client, s.httpClient)
+	s.hbStreams = newHeartbeatStreams(ctx, s.clusterID)
+
+	// Run callbacks
+	for _, cb := range s.startCallbacks {
+		cb()
+	}
+
+	// Server has started.
+	atomic.StoreInt64(&s.isServing, 1)
+	return nil
+}
+
+func (s *Server) initClusterID() error {
+	// Get any cluster key to parse the cluster ID.
+	resp, err := etcdutil.EtcdKVGet(s.client, pdClusterIDPath)
+	if err != nil {
+		return err
+	}
+
+	// If no key exist, generate a random cluster ID.
+	if len(resp.Kvs) == 0 {
+		s.clusterID, err = initOrGetClusterID(s.client, pdClusterIDPath)
+		return err
+	}
+	s.clusterID, err = typeutil.BytesToUint64(resp.Kvs[0].Value)
+	return err
+}
+
+// AddCloseCallback adds a callback in the Close phase.
+func (s *Server) AddCloseCallback(callbacks ...func()) {
+	s.closeCallbacks = append(s.closeCallbacks, callbacks...)
+}
+
+// Close closes the server.
+func (s *Server) Close() {
+	if !atomic.CompareAndSwapInt64(&s.isServing, 1, 0) {
+		// server is already closed
+		return
+	}
+
+	log.Info("closing server")
+
+	s.stopServerLoop()
+
+	if s.client != nil {
+		s.client.Close()
+	}
+
+	if s.member.Etcd() != nil {
+		s.member.Close()
+	}
+
+	if s.hbStreams != nil {
+		s.hbStreams.Close()
+	}
+	if err := s.storage.Close(); err != nil {
+		log.Error("close storage meet error", zap.Error(err))
+	}
+
+	// Run callbacks
+	for _, cb := range s.closeCallbacks {
+		cb()
+	}
+
+	log.Info("close server")
+}
+
+// IsClosed checks whether server is closed or not.
+func (s *Server) IsClosed() bool {
+	return atomic.LoadInt64(&s.isServing) == 0
+}
+
+// Run runs the pd server.
+func (s *Server) Run() error {
+	//go StartMonitor(s.ctx, time.Now, func() {
+	//	log.Error("system time jumps backward")
+	//	timeJumpBackCounter.Inc()
+	//})
+	if err := s.startEtcd(s.ctx); err != nil {
+		return err
+	}
+
+	if err := s.startServer(s.ctx); err != nil {
+		return err
+	}
+
+	s.startServerLoop(s.ctx)
+
+	return nil
+}
+
+// Context returns the context of server.
+func (s *Server) Context() context.Context {
+	return s.ctx
+}
+
+// LoopContext returns the loop context of server.
+func (s *Server) LoopContext() context.Context {
+	return s.serverLoopCtx
+}
+
+func (s *Server) startServerLoop(ctx context.Context) {
+	s.serverLoopCtx, s.serverLoopCancel = context.WithCancel(ctx)
+	s.serverLoopWg.Add(1)
+	go s.etcdLeaderLoop()
+}
+
+func (s *Server) stopServerLoop() {
+	s.serverLoopCancel()
+	s.serverLoopWg.Wait()
+}
+
+// GetAddr returns the server urls for clients.
+func (s *Server) GetAddr() string {
+	return s.cfg.AdvertiseClientUrls
+}
+
+// GetClientScheme returns the client URL scheme
+func (s *Server) GetClientScheme() string {
+	if len(s.cfg.Security.CertPath) == 0 && len(s.cfg.Security.KeyPath) == 0 {
+		return "http"
+	}
+	return "https"
+}
+
+// GetClient returns builtin etcd client.
+func (s *Server) GetClient() *clientv3.Client {
+	return s.client
+}
+
+// GetPersistOptions returns the schedule option.
+func (s *Server) GetPersistOptions() *config.PersistOptions {
+	return s.persistOptions
+}
+
+// GetStorage returns the backend storage of server.
+func (s *Server) GetStorage() *meta.Storage {
+	return s.storage
+}
+
+// GetHBStreams returns the heartbeat streams.
+func (s *Server) GetHBStreams() HeartbeatStreams {
+	return s.hbStreams
+}
+
+// GetAllocator returns the ID allocator of server.
+func (s *Server) GetAllocator() *id.AllocatorImpl {
+	return s.idAllocator
+}
+
+// Name returns the unique etcd Name for this server in etcd cluster.
+func (s *Server) Name() string {
+	return s.cfg.Name
+}
+
+// ClusterID returns the cluster ID of this server.
+func (s *Server) ClusterID() uint64 {
+	return s.clusterID
+}
+
+// StartTimestamp returns the start timestamp of this server
+func (s *Server) StartTimestamp() int64 {
+	return s.startTimestamp
+}
+
+// GetConfig gets the config information.
+func (s *Server) GetConfig() *config.Config {
+	cfg := s.cfg.Clone()
+	cfg.PDServerCfg = *s.persistOptions.GetPDServerConfig()
+	storage := s.GetStorage()
+	if storage == nil {
+		return cfg
+	}
+	return cfg
+}
+
+// GetServerOption gets the option of the server.
+func (s *Server) GetServerOption() *config.PersistOptions {
+	return s.persistOptions
+}
+
+// SetLogLevel sets log level.
+func (s *Server) SetLogLevel(level string) error {
+	if !isLevelLegal(level) {
+		return errors.Errorf("log level %s is illegal", level)
+	}
+	s.cfg.Log.Level = level
+	log.SetLevel(logutil.StringToZapLogLevel(level))
+	log.Warn("log level changed", zap.String("level", log.GetLevel().String()))
+	return nil
+}
+
+func isLevelLegal(level string) bool {
+	switch strings.ToLower(level) {
+	case "fatal", "error", "warn", "warning", "debug", "info":
+		return true
+	default:
+		return false
+	}
+}
+
+func (s *Server) etcdLeaderLoop() {
+	defer logutil.LogPanic()
+	defer s.serverLoopWg.Done()
+
+	ctx, cancel := context.WithCancel(s.serverLoopCtx)
+	defer cancel()
+	for {
+		select {
+		// case <-time.After(s.cfg.LeaderPriorityCheckInterval.Duration):
+		// 	s.member.CheckPriority(ctx)
+		case <-ctx.Done():
+			log.Info("server is closed, exit etcd leader loop")
+			return
+		}
+	}
+}

master/tso/global_allocator.go

+// Copyright 2020 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package tso
+
+import (
+	"sync/atomic"
+	"time"
+
+	"github.com/czs007/suvlim/errors"
+	"github.com/czs007/suvlim/master/election"
+	"github.com/czs007/suvlim/pkg/pdpb"
+	"github.com/czs007/suvlim/util/typeutil"
+	"github.com/pingcap/log"
+	"go.uber.org/zap"
+)
+
+// Allocator is a Timestamp Oracle allocator.
+type Allocator interface {
+	// Initialize is used to initialize a TSO allocator.
+	// It will synchronize TSO with etcd and initialize the
+	// memory for later allocation work.
+	Initialize() error
+	// UpdateTSO is used to update the TSO in memory and the time window in etcd.
+	UpdateTSO() error
+	// SetTSO sets the physical part with given tso. It's mainly used for BR restore
+	// and can not forcibly set the TSO smaller than now.
+	SetTSO(tso uint64) error
+	// GenerateTSO is used to generate a given number of TSOs.
+	// Make sure you have initialized the TSO allocator before calling.
+	GenerateTSO(count uint32) (pdpb.Timestamp, error)
+	// Reset is used to reset the TSO allocator.
+	Reset()
+}
+
+// GlobalTSOAllocator is the global single point TSO allocator.
+type GlobalTSOAllocator struct {
+	// leadership is used to check the current PD server's leadership
+	// to determine whether a tso request could be processed.
+	leadership      *election.Leadership
+	timestampOracle *timestampOracle
+}
+
+// NewGlobalTSOAllocator creates a new global TSO allocator.
+func NewGlobalTSOAllocator(leadership *election.Leadership, rootPath string, saveInterval time.Duration, maxResetTSGap func() time.Duration) Allocator {
+	return &GlobalTSOAllocator{
+		leadership: leadership,
+		timestampOracle: &timestampOracle{
+			client:        leadership.GetClient(),
+			rootPath:      rootPath,
+			saveInterval:  saveInterval,
+			maxResetTSGap: maxResetTSGap,
+		},
+	}
+}
+
+// Initialize will initialize the created global TSO allocator.
+func (gta *GlobalTSOAllocator) Initialize() error {
+	return gta.timestampOracle.SyncTimestamp(gta.leadership)
+}
+
+// UpdateTSO is used to update the TSO in memory and the time window in etcd.
+func (gta *GlobalTSOAllocator) UpdateTSO() error {
+	return gta.timestampOracle.UpdateTimestamp(gta.leadership)
+}
+
+// SetTSO sets the physical part with given tso.
+func (gta *GlobalTSOAllocator) SetTSO(tso uint64) error {
+	return gta.timestampOracle.ResetUserTimestamp(gta.leadership, tso)
+}
+
+// GenerateTSO is used to generate a given number of TSOs.
+// Make sure you have initialized the TSO allocator before calling.
+func (gta *GlobalTSOAllocator) GenerateTSO(count uint32) (pdpb.Timestamp, error) {
+	var resp pdpb.Timestamp
+
+	if count == 0 {
+		return resp, errors.New("tso count should be positive")
+	}
+
+	maxRetryCount := 10
+	// failpoint.Inject("skipRetryGetTS", func() {
+	// 	maxRetryCount = 1
+	// })
+
+	for i := 0; i < maxRetryCount; i++ {
+		current := (*atomicObject)(atomic.LoadPointer(&gta.timestampOracle.TSO))
+		if current == nil || current.physical == typeutil.ZeroTime {
+			// If it's leader, maybe SyncTimestamp hasn't completed yet
+			if gta.leadership.Check() {
+				log.Info("sync hasn't completed yet, wait for a while")
+				time.Sleep(200 * time.Millisecond)
+				continue
+			}
+			log.Error("invalid timestamp", zap.Any("timestamp", current))
+			return pdpb.Timestamp{}, errors.New("can not get timestamp, may be not leader")
+		}
+
+		resp.Physical = current.physical.UnixNano() / int64(time.Millisecond)
+		resp.Logical = atomic.AddInt64(&current.logical, int64(count))
+		if resp.Logical >= maxLogical {
+			log.Error("logical part outside of max logical interval, please check ntp time",
+				zap.Reflect("response", resp),
+				zap.Int("retry-count", i))
+			time.Sleep(UpdateTimestampStep)
+			continue
+		}
+		// In case lease expired after the first check.
+		if !gta.leadership.Check() {
+			return pdpb.Timestamp{}, errors.New("alloc timestamp failed, lease expired")
+		}
+		return resp, nil
+	}
+	return resp, errors.New("can not get timestamp")
+}
+
+// Reset is used to reset the TSO allocator.
+func (gta *GlobalTSOAllocator) Reset() {
+	gta.timestampOracle.ResetTimestamp()
+}

master/tso/tso.go

+// Copyright 2016 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package tso
+
+import (
+	"path"
+	"sync/atomic"
+	"time"
+	"unsafe"
+
+	"github.com/czs007/suvlim/errors"
+	"github.com/czs007/suvlim/master/election"
+	"github.com/czs007/suvlim/util/etcdutil"
+	"github.com/czs007/suvlim/util/tsoutil"
+	"github.com/czs007/suvlim/util/typeutil"
+	"github.com/pingcap/log"
+	"go.etcd.io/etcd/clientv3"
+	"go.uber.org/zap"
+)
+
+const (
+	// UpdateTimestampStep is used to update timestamp.
+	UpdateTimestampStep = 50 * time.Millisecond
+	// updateTimestampGuard is the min timestamp interval.
+	updateTimestampGuard = time.Millisecond
+	// maxLogical is the max upper limit for logical time.
+	// When a TSO's logical time reaches this limit,
+	// the physical time will be forced to increase.
+	maxLogical = int64(1 << 18)
+)
+
+// atomicObject is used to store the current TSO in memory.
+type atomicObject struct {
+	physical time.Time
+	logical  int64
+}
+
+// timestampOracle is used to maintain the logic of tso.
+type timestampOracle struct {
+	client   *clientv3.Client
+	rootPath string
+	// TODO: remove saveInterval
+	saveInterval  time.Duration
+	maxResetTSGap func() time.Duration
+	// For tso, set after the PD becomes a leader.
+	TSO           unsafe.Pointer
+	lastSavedTime atomic.Value
+}
+
+func (t *timestampOracle) getTimestampPath() string {
+	return path.Join(t.rootPath, "timestamp")
+}
+
+func (t *timestampOracle) loadTimestamp() (time.Time, error) {
+	data, err := etcdutil.GetValue(t.client, t.getTimestampPath())
+	if err != nil {
+		return typeutil.ZeroTime, err
+	}
+	if len(data) == 0 {
+		return typeutil.ZeroTime, nil
+	}
+	return typeutil.ParseTimestamp(data)
+}
+
+// save timestamp, if lastTs is 0, we think the timestamp doesn't exist, so create it,
+// otherwise, update it.
+func (t *timestampOracle) saveTimestamp(leadership *election.Leadership, ts time.Time) error {
+	key := t.getTimestampPath()
+	data := typeutil.Uint64ToBytes(uint64(ts.UnixNano()))
+	resp, err := leadership.LeaderTxn().
+		Then(clientv3.OpPut(key, string(data))).
+		Commit()
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	if !resp.Succeeded {
+		return errors.New("save timestamp failed, maybe we lost leader")
+	}
+
+	t.lastSavedTime.Store(ts)
+
+	return nil
+}
+
+// SyncTimestamp is used to synchronize the timestamp.
+func (t *timestampOracle) SyncTimestamp(leadership *election.Leadership) error {
+
+	last, err := t.loadTimestamp()
+	if err != nil {
+		return err
+	}
+
+	next := time.Now()
+
+	// If the current system time minus the saved etcd timestamp is less than `updateTimestampGuard`,
+	// the timestamp allocation will start from the saved etcd timestamp temporarily.
+	if typeutil.SubTimeByWallClock(next, last) < updateTimestampGuard {
+		next = last.Add(updateTimestampGuard)
+	}
+
+	save := next.Add(t.saveInterval)
+	if err = t.saveTimestamp(leadership, save); err != nil {
+		return err
+	}
+
+	log.Info("sync and save timestamp", zap.Time("last", last), zap.Time("save", save), zap.Time("next", next))
+
+	current := &atomicObject{
+		physical: next,
+	}
+	atomic.StorePointer(&t.TSO, unsafe.Pointer(current))
+
+	return nil
+}
+
+// ResetUserTimestamp update the physical part with specified tso.
+func (t *timestampOracle) ResetUserTimestamp(leadership *election.Leadership, tso uint64) error {
+	if !leadership.Check() {
+		return errors.New("Setup timestamp failed, lease expired")
+	}
+	physical, _ := tsoutil.ParseTS(tso)
+	next := physical.Add(time.Millisecond)
+	prev := (*atomicObject)(atomic.LoadPointer(&t.TSO))
+
+	// do not update
+	if typeutil.SubTimeByWallClock(next, prev.physical) <= 3*updateTimestampGuard {
+		return errors.New("the specified ts too small than now")
+	}
+
+	if typeutil.SubTimeByWallClock(next, prev.physical) >= t.maxResetTSGap() {
+		return errors.New("the specified ts too large than now")
+	}
+
+	save := next.Add(t.saveInterval)
+	if err := t.saveTimestamp(leadership, save); err != nil {
+		return err
+	}
+	update := &atomicObject{
+		physical: next,
+	}
+	atomic.CompareAndSwapPointer(&t.TSO, unsafe.Pointer(prev), unsafe.Pointer(update))
+	return nil
+}
+
+// UpdateTimestamp is used to update the timestamp.
+// This function will do two things:
+// 1. When the logical time is going to be used up, increase the current physical time.
+// 2. When the time window is not big enough, which means the saved etcd time minus the next physical time
+//    will be less than or equal to `updateTimestampGuard`, then the time window needs to be updated and
+//    we also need to save the next physical time plus `TsoSaveInterval` into etcd.
+//
+// Here is some constraints that this function must satisfy:
+// 1. The saved time is monotonically increasing.
+// 2. The physical time is monotonically increasing.
+// 3. The physical time is always less than the saved timestamp.
+func (t *timestampOracle) UpdateTimestamp(leadership *election.Leadership) error {
+	prev := (*atomicObject)(atomic.LoadPointer(&t.TSO))
+	now := time.Now()
+
+	jetLag := typeutil.SubTimeByWallClock(now, prev.physical)
+	if jetLag > 3*UpdateTimestampStep {
+		log.Warn("clock offset", zap.Duration("jet-lag", jetLag), zap.Time("prev-physical", prev.physical), zap.Time("now", now))
+	}
+
+	var next time.Time
+	prevLogical := atomic.LoadInt64(&prev.logical)
+	// If the system time is greater, it will be synchronized with the system time.
+	if jetLag > updateTimestampGuard {
+		next = now
+	} else if prevLogical > maxLogical/2 {
+		// The reason choosing maxLogical/2 here is that it's big enough for common cases.
+		// Because there is enough timestamp can be allocated before next update.
+		log.Warn("the logical time may be not enough", zap.Int64("prev-logical", prevLogical))
+		next = prev.physical.Add(time.Millisecond)
+	} else {
+		// It will still use the previous physical time to alloc the timestamp.
+		return nil
+	}
+
+	// It is not safe to increase the physical time to `next`.
+	// The time window needs to be updated and saved to etcd.
+	if typeutil.SubTimeByWallClock(t.lastSavedTime.Load().(time.Time), next) <= updateTimestampGuard {
+		save := next.Add(t.saveInterval)
+		if err := t.saveTimestamp(leadership, save); err != nil {
+			return err
+		}
+	}
+
+	current := &atomicObject{
+		physical: next,
+		logical:  0,
+	}
+
+	atomic.StorePointer(&t.TSO, unsafe.Pointer(current))
+
+	return nil
+}
+
+// ResetTimestamp is used to reset the timestamp.
+func (t *timestampOracle) ResetTimestamp() {
+	zero := &atomicObject{
+		physical: typeutil.ZeroTime,
+	}
+	atomic.StorePointer(&t.TSO, unsafe.Pointer(zero))
+}

master/util.go

+// Copyright 2016 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"context"
+	"math/rand"
+	"time"
+
+	"github.com/czs007/suvlim/errors"
+	"github.com/czs007/suvlim/util/etcdutil"
+	"github.com/czs007/suvlim/util/typeutil"
+	"go.etcd.io/etcd/clientv3"
+)
+
+const (
+	requestTimeout = etcdutil.DefaultRequestTimeout
+)
+
+func initOrGetClusterID(c *clientv3.Client, key string) (uint64, error) {
+	ctx, cancel := context.WithTimeout(c.Ctx(), requestTimeout)
+	defer cancel()
+
+	// Generate a random cluster ID.
+	ts := uint64(time.Now().Unix())
+	clusterID := (ts << 32) + uint64(rand.Uint32())
+	value := typeutil.Uint64ToBytes(clusterID)
+
+	// Multiple PDs may try to init the cluster ID at the same time.
+	// Only one PD can commit this transaction, then other PDs can get
+	// the committed cluster ID.
+	resp, err := c.Txn(ctx).
+		If(clientv3.Compare(clientv3.CreateRevision(key), "=", 0)).
+		Then(clientv3.OpPut(key, string(value))).
+		Else(clientv3.OpGet(key)).
+		Commit()
+	if err != nil {
+		return 0, errors.WithStack(err)
+	}
+
+	// Txn commits ok, return the generated cluster ID.
+	if resp.Succeeded {
+		return clusterID, nil
+	}
+
+	// Otherwise, parse the committed cluster ID.
+	if len(resp.Responses) == 0 {
+		return 0, errors.Errorf("txn returns empty response: %v", resp)
+	}
+
+	response := resp.Responses[0].GetResponseRange()
+	if response == nil || len(response.Kvs) != 1 {
+		return 0, errors.Errorf("txn returns invalid range response: %v", resp)
+	}
+
+	return typeutil.BytesToUint64(response.Kvs[0].Value)
+}

pkg/metapb/metapb.pb.go

+// Code generated by protoc-gen-go. DO NOT EDIT.
+// source: metapb.proto
+
+/*
+Package metapb is a generated protocol buffer package.
+
+It is generated from these files:
+	metapb.proto
+
+It has these top-level messages:
+	Cluster
+	Peer
+*/
+package metapb
+
+import proto "github.com/golang/protobuf/proto"
+import fmt "fmt"
+import math "math"
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ = proto.Marshal
+var _ = fmt.Errorf
+var _ = math.Inf
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the proto package it is being compiled against.
+// A compilation error at this line likely means your copy of the
+// proto package needs to be updated.
+const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package
+
+type PeerRole int32
+
+const (
+	PeerRole_Master  PeerRole = 0
+	PeerRole_Reader  PeerRole = 1
+	PeerRole_Write   PeerRole = 2
+	PeerRole_Proxyer PeerRole = 3
+)
+
+var PeerRole_name = map[int32]string{
+	0: "Master",
+	1: "Reader",
+	2: "Write",
+	3: "Proxyer",
+}
+var PeerRole_value = map[string]int32{
+	"Master":  0,
+	"Reader":  1,
+	"Write":   2,
+	"Proxyer": 3,
+}
+
+func (x PeerRole) String() string {
+	return proto.EnumName(PeerRole_name, int32(x))
+}
+func (PeerRole) EnumDescriptor() ([]byte, []int) { return fileDescriptor0, []int{0} }
+
+type Cluster struct {
+	Id uint64 `protobuf:"varint,1,opt,name=id" json:"id,omitempty"`
+	// max peer count for a region.
+	// pd will do the auto-balance if region peer count mismatches.
+	MaxPeerCount uint32 `protobuf:"varint,1024,opt,name=max_peer_count,json=maxPeerCount" json:"max_peer_count,omitempty"`
+}
+
+func (m *Cluster) Reset()                    { *m = Cluster{} }
+func (m *Cluster) String() string            { return proto.CompactTextString(m) }
+func (*Cluster) ProtoMessage()               {}
+func (*Cluster) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{0} }
+
+func (m *Cluster) GetId() uint64 {
+	if m != nil {
+		return m.Id
+	}
+	return 0
+}
+
+func (m *Cluster) GetMaxPeerCount() uint32 {
+	if m != nil {
+		return m.MaxPeerCount
+	}
+	return 0
+}
+
+type Peer struct {
+	Id     uint64   `protobuf:"varint,1,opt,name=id" json:"id,omitempty"`
+	PeerId uint64   `protobuf:"varint,2,opt,name=peer_id,json=peerId" json:"peer_id,omitempty"`
+	Role   PeerRole `protobuf:"varint,3,opt,name=role,enum=metapb.PeerRole" json:"role,omitempty"`
+}
+
+func (m *Peer) Reset()                    { *m = Peer{} }
+func (m *Peer) String() string            { return proto.CompactTextString(m) }
+func (*Peer) ProtoMessage()               {}
+func (*Peer) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{1} }
+
+func (m *Peer) GetId() uint64 {
+	if m != nil {
+		return m.Id
+	}
+	return 0
+}
+
+func (m *Peer) GetPeerId() uint64 {
+	if m != nil {
+		return m.PeerId
+	}
+	return 0
+}
+
+func (m *Peer) GetRole() PeerRole {
+	if m != nil {
+		return m.Role
+	}
+	return PeerRole_Master
+}
+
+func init() {
+	proto.RegisterType((*Cluster)(nil), "metapb.Cluster")
+	proto.RegisterType((*Peer)(nil), "metapb.Peer")
+	proto.RegisterEnum("metapb.PeerRole", PeerRole_name, PeerRole_value)
+}
+
+func init() { proto.RegisterFile("metapb.proto", fileDescriptor0) }
+
+var fileDescriptor0 = []byte{
+	// 210 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x64, 0x8f, 0xb1, 0x4b, 0xc5, 0x30,
+	0x10, 0xc6, 0x4d, 0x5e, 0x4d, 0x9f, 0xe7, 0xb3, 0x84, 0x2c, 0x66, 0x2c, 0x0f, 0x85, 0xe2, 0xd0,
+	0x41, 0x37, 0x27, 0xa1, 0x93, 0x83, 0x50, 0x02, 0xe2, 0x58, 0x52, 0x73, 0x43, 0xa0, 0x35, 0xe5,
+	0x4c, 0xa1, 0x6e, 0xfe, 0xe9, 0x92, 0xa8, 0xd3, 0xdb, 0xee, 0xf7, 0x3b, 0xbe, 0xe3, 0x3b, 0x38,
+	0xcc, 0x18, 0xed, 0x32, 0xb6, 0x0b, 0x85, 0x18, 0x94, 0xf8, 0xa5, 0xe3, 0x13, 0x94, 0xdd, 0xb4,
+	0x7e, 0x46, 0x24, 0x55, 0x01, 0xf7, 0x4e, 0xb3, 0x9a, 0x35, 0x85, 0xe1, 0xde, 0xa9, 0x5b, 0xa8,
+	0x66, 0xbb, 0x0d, 0x0b, 0x22, 0x0d, 0xef, 0x61, 0xfd, 0x88, 0xfa, 0x7b, 0x5f, 0xb3, 0xe6, 0xca,
+	0x1c, 0x66, 0xbb, 0xf5, 0x88, 0xd4, 0x25, 0x79, 0x7c, 0x85, 0x22, 0xc1, 0x49, 0xfc, 0x1a, 0xca,
+	0x1c, 0xf5, 0x4e, 0xf3, 0x2c, 0x45, 0xc2, 0x67, 0xa7, 0x6e, 0xa0, 0xa0, 0x30, 0xa1, 0xde, 0xd5,
+	0xac, 0xa9, 0xee, 0x65, 0xfb, 0xd7, 0x2b, 0x1d, 0x31, 0x61, 0x42, 0x93, 0xb7, 0x77, 0x8f, 0xb0,
+	0xff, 0x37, 0x0a, 0x40, 0xbc, 0xd8, 0xd4, 0x51, 0x9e, 0xa5, 0xd9, 0xa0, 0x75, 0x48, 0x92, 0xa9,
+	0x0b, 0x38, 0x7f, 0x23, 0x1f, 0x51, 0x72, 0x75, 0x09, 0x65, 0x4f, 0x61, 0xfb, 0x42, 0x92, 0xbb,
+	0x51, 0xe4, 0x1f, 0x1f, 0x7e, 0x02, 0x00, 0x00, 0xff, 0xff, 0x03, 0x47, 0x25, 0x66, 0xf3, 0x00,
+	0x00, 0x00,
+}

proto/metapb.proto

+syntax = "proto3";
+package metapb;
+
+
+message Cluster {
+    uint64 id = 1;
+    // max peer count for a region.
+    // pd will do the auto-balance if region peer count mismatches.
+    uint32 max_peer_count = 1024;
+    // more attributes......
+}
+
+enum PeerRole {
+
+    Master = 0;
+
+    Reader = 1;
+
+    Write = 2;
+
+    Proxyer = 3;
+
+}
+
+message Peer {
+    uint64 id = 1;
+    uint64 peer_id = 2;
+    PeerRole role = 3;
+}

proto/pdpb.proto

+syntax = "proto3";
+package pdpb;
+
+import "metapb.proto";
+
+service PD {
+    rpc Tso(stream TsoRequest) returns (stream TsoResponse) {}
+    rpc AllocID(AllocIDRequest) returns (AllocIDResponse) {}
+    rpc Heartbeat(stream HeartbeatRequest) returns (stream HeartbeatResponse) {}
+}
+
+message Member {
+    // name is the name of the PD member.
+    string name = 1;
+    // member_id is the unique id of the PD member.
+    uint64 member_id = 2;
+    repeated string peer_urls = 3;
+    repeated string client_urls = 4;
+}
+
+message RequestHeader {
+    // cluster_id is the ID of the cluster which be sent to.
+    uint64 cluster_id = 1;
+}
+
+message ResponseHeader {
+    // cluster_id is the ID of the cluster which sent the response.
+    uint64 cluster_id = 1;
+    Error error = 2;
+}
+
+enum ErrorType {
+    OK = 0;
+    UNKNOWN = 1;
+    NOT_BOOTSTRAPPED = 2;
+    STORE_TOMBSTONE = 3;
+    ALREADY_BOOTSTRAPPED = 4;
+    INCOMPATIBLE_VERSION = 5;
+    REGION_NOT_FOUND = 6;
+}
+
+message Error {
+    ErrorType type = 1;
+    string message = 2;
+}
+
+message TsoRequest {
+    RequestHeader header = 1;
+
+    uint32 count = 2;
+}
+
+message Timestamp {
+    int64 physical = 1;
+    int64 logical = 2;
+}
+
+message TsoResponse {
+    ResponseHeader header = 1;
+
+    uint32 count = 2;
+    Timestamp timestamp = 3;
+}
+
+message AllocIDRequest {
+    RequestHeader header = 1;
+}
+
+message AllocIDResponse {
+    ResponseHeader header = 1;
+    uint64 id = 2;
+}
+
+message HeartbeatRequest {
+    RequestHeader header = 1;
+    metapb.Peer peer = 2;
+
+}
+
+message HeartbeatResponse {
+    ResponseHeader header = 1;
+}

pulsar/schema/message.go

@@ -101,14 +101,14 @@ const (
 
 type PulsarMessage struct {
 	CollectionName string
-	Fields []*FieldValue
-	EntityId int64
-	PartitionTag string
-	VectorParam *VectorParam
-	Segments []*SegmentRecord
-	Timestamp int64
-	ClientId int64
-	MsgType OpType
+	Fields         []*FieldValue
+	EntityId       int64
+	PartitionTag   string
+	VectorParam    *VectorParam
+	Segments       []*SegmentRecord
+	Timestamp      int64
+	ClientId       int64
+	MsgType        OpType
 }
 
 type Message interface {
@@ -122,6 +122,7 @@ type InsertMsg struct {
 	Fields         []*FieldValue
 	EntityId       int64
 	PartitionTag   string
+	SegmentId      uint64
 	Timestamp      uint64
 	ClientId       int64
 	MsgType        OpType
@@ -189,5 +190,5 @@ func (kms *Key2SegMsg) GetType() OpType {
 }
 
 type SyncEofMsg struct {
-	MsgType  OpType
-}
\ No newline at end of file
+	MsgType OpType
+}

reader/collection.go

 package reader
 
+/*
+
+#cgo CFLAGS: -I../core/include
+
+#cgo LDFLAGS: -L../core/lib -lmilvus_dog_segment -Wl,-rpath=../core/lib
+
+#include "partition_c.h"
+
+*/
 import "C"
 import (
 	"errors"
@@ -11,29 +20,20 @@ type Collection struct {
 	Partitions []*Partition
 }
 
-func (c *Collection) NewPartition(partitionName string) (*Partition, error) {
+func (c *Collection) NewPartition(partitionName string) *Partition {
 	cName := C.CString(partitionName)
-	partitionPtr, status := C.NewPartition(c.CollectionPtr, cName)
-
-	if status != 0 {
-		return nil, errors.New("create partition failed")
-	}
+	partitionPtr := C.NewPartition(c.CollectionPtr, cName)
 
 	var newPartition = &Partition{PartitionPtr: partitionPtr, PartitionName: partitionName}
 	c.Partitions = append(c.Partitions, newPartition)
-	return newPartition, nil
+	return newPartition
 }
 
-func (c *Collection) DeletePartition(partitionName string) error {
-	cName := C.CString(partitionName)
-	status := C.DeletePartition(c.CollectionPtr, cName)
-
-	if status != 0 {
-		return errors.New("create partition failed")
-	}
+func (c *Collection) DeletePartition(partition *Partition) {
+	cPtr := partition.PartitionPtr
+	C.DeletePartition(cPtr)
 
 	// TODO: remove from c.Partitions
-	return nil
 }
 
 func (c *Collection) GetSegments() ([]*Segment, error) {

reader/index.go

 package reader
 
-import "suvlim/pulsar/schema"
+import "github.com/czs007/suvlim/pulsar/schema"
 
 type IndexConfig struct {}