[DNM]add user kv api

Gopkg.toml

@@ -35,3 +35,7 @@
 [[constraint]]
   name = "github.com/chzyer/readline"
   branch = "master"
+
+[[constraint]]
+  name = "github.com/wentaoxu/kvproto"
+  branch = "do_gc_distributedly"

pd-client/client.go

@@ -53,6 +53,12 @@ type Client interface {
 	// The store may expire later. Caller is responsible for caching and taking care
 	// of store change.
 	GetStore(ctx context.Context, storeID uint64) (*metapb.Store, error)
+	// GetUserKV get the user kv from pd.
+	GetUserKV(ctx context.Context, key string) (string, error)
+	// PutUserKV put the user kv to pd.
+	PutUserKV(ctx context.Context, key string, value string) error
+	// DeleteUserKV delete the user kv from pd.
+	DeleteUserKV(ctx context.Context, key string) error
 	// Close closes the client.
 	Close()
 }
@@ -622,6 +628,79 @@ func (c *client) GetStore(ctx context.Context, storeID uint64) (*metapb.Store, e
 	return store, nil
 }
 
+func (c *client) GetUserKV(ctx context.Context, key string) (string, error) {
+	if span := opentracing.SpanFromContext(ctx); span != nil {
+		span = opentracing.StartSpan("pdclient.GetUserKV", opentracing.ChildOf(span.Context()))
+		defer span.Finish()
+	}
+	start := time.Now()
+	defer func() { cmdDuration.WithLabelValues("get_user_key").Observe(time.Since(start).Seconds()) }()
+
+	ctx, cancel := context.WithTimeout(ctx, pdTimeout)
+	resp, err := c.leaderClient().GetUserKV(ctx, &pdpb.GetUserKVRequest{
+		Header: c.requestHeader(),
+		Key:    key,
+	})
+	requestDuration.WithLabelValues("get_user_key").Observe(time.Since(start).Seconds())
+	cancel()
+
+	if err != nil {
+		cmdFailedDuration.WithLabelValues("get_user_key").Observe(time.Since(start).Seconds())
+		c.scheduleCheckLeader()
+		return "", errors.Trace(err)
+	}
+	return resp.GetValue(), nil
+}
+
+func (c *client) PutUserKV(ctx context.Context, key string, value string) error {
+	if span := opentracing.SpanFromContext(ctx); span != nil {
+		span = opentracing.StartSpan("pdclient.PutUserKV", opentracing.ChildOf(span.Context()))
+		defer span.Finish()
+	}
+	start := time.Now()
+	defer func() { cmdDuration.WithLabelValues("put_user_key").Observe(time.Since(start).Seconds()) }()
+
+	ctx, cancel := context.WithTimeout(ctx, pdTimeout)
+	_, err := c.leaderClient().PutUserKV(ctx, &pdpb.PutUserKVRequest{
+		Header: c.requestHeader(),
+		Key:    key,
+		Value:  value,
+	})
+	requestDuration.WithLabelValues("put_user_key").Observe(time.Since(start).Seconds())
+	cancel()
+
+	if err != nil {
+		cmdFailedDuration.WithLabelValues("put_user_key").Observe(time.Since(start).Seconds())
+		c.scheduleCheckLeader()
+		return errors.Trace(err)
+	}
+	return nil
+}
+
+func (c *client) DeleteUserKV(ctx context.Context, key string) error {
+	if span := opentracing.SpanFromContext(ctx); span != nil {
+		span = opentracing.StartSpan("pdclient.DeleteUserKV", opentracing.ChildOf(span.Context()))
+		defer span.Finish()
+	}
+	start := time.Now()
+	defer func() { cmdDuration.WithLabelValues("delete_user_kv").Observe(time.Since(start).Seconds()) }()
+
+	ctx, cancel := context.WithTimeout(ctx, pdTimeout)
+	_, err := c.leaderClient().DeleteUserKV(ctx, &pdpb.DeleteUserKVRequest{
+		Header: c.requestHeader(),
+		Key:    key,
+	})
+	requestDuration.WithLabelValues("delete_user_kv").Observe(time.Since(start).Seconds())
+	cancel()
+
+	if err != nil {
+		cmdFailedDuration.WithLabelValues("delete_user_kv").Observe(time.Since(start).Seconds())
+		c.scheduleCheckLeader()
+		return errors.Trace(err)
+	}
+	return nil
+}
+
 func (c *client) requestHeader() *pdpb.RequestHeader {
 	return &pdpb.RequestHeader{
 		ClusterId: c.clusterID,

server/core/kv.go

@@ -29,6 +29,7 @@ const (
 	clusterPath  = "raft"
 	configPath   = "config"
 	schedulePath = "schedule"
+	userPath     = "user"
 )
 
 // KV wraps all kv operations, keep it stateless.
@@ -239,3 +240,18 @@ func (kv *KV) saveProto(key string, msg proto.Message) error {
 	}
 	return kv.Save(key, string(value))
 }
+
+// GetUserKV get key from pd under dir 'user'.
+func (kv *KV) GetUserKV(key string) (string, error) {
+	return kv.Load(path.Join(userPath, key))
+}
+
+// PutUserKV put key to pd under dir 'user'.
+func (kv *KV) PutUserKV(key string, value string) error {
+	return kv.Save(path.Join(userPath, key), value)
+}
+
+// DeleteUserKV delete key from pd under dir 'user'.
+func (kv *KV) DeleteUserKV(key string) error {
+	return kv.Delete(path.Join(userPath, key))
+}

server/grpc_service.go

@@ -555,3 +555,56 @@ func (s *Server) notBootstrappedHeader() *pdpb.ResponseHeader {
 		Message: "cluster is not bootstrapped",
 	})
 }
+
+// GetUserKV get key from pd under dir 'user'.
+func (s *Server) GetUserKV(ctx context.Context, request *pdpb.GetUserKVRequest) (*pdpb.GetUserKVResponse, error) {
+	if err := s.validateRequest(request.GetHeader()); err != nil {
+		return nil, errors.Trace(err)
+	}
+
+	value, err := s.kv.GetUserKV(request.GetKey())
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
+	return &pdpb.GetUserKVResponse{
+		Header: s.header(),
+		Key:    request.GetKey(),
+		Value:  value,
+	}, nil
+}
+
+// PutUserKV put key to pd under dir 'user'.
+func (s *Server) PutUserKV(ctx context.Context, request *pdpb.PutUserKVRequest) (*pdpb.PutUserKVResponse, error) {
+	if err := s.validateRequest(request.GetHeader()); err != nil {
+		return nil, errors.Trace(err)
+	}
+
+	err := s.kv.PutUserKV(request.GetKey(), request.GetValue())
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
+	return &pdpb.PutUserKVResponse{
+		Header: s.header(),
+		Key:    request.GetKey(),
+		Value:  request.GetValue(),
+	}, nil
+}
+
+// DeleteUserKV delete key from pd under dir 'user'.
+func (s *Server) DeleteUserKV(ctx context.Context, request *pdpb.DeleteUserKVRequest) (*pdpb.DeleteUserKVResponse, error) {
+	if err := s.validateRequest(request.GetHeader()); err != nil {
+		return nil, errors.Trace(err)
+	}
+
+	err := s.kv.DeleteUserKV(request.GetKey())
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
+	return &pdpb.DeleteUserKVResponse{
+		Header: s.header(),
+		Key:    request.GetKey(),
+	}, nil
+}