support etcd and resource group

Signed-off-by: husharp <jinhao.hu@pingcap.com>

client/client.go

@@ -50,9 +50,10 @@ type Region struct {
 
 // GlobalConfigItem standard format of KV pair in GlobalConfig client
 type GlobalConfigItem struct {
-	Name  string
-	Value string
-	Error error
+	Name     string
+	Value    string
+	ItemKind pdpb.ItemKind
+	Error    error
 }
 
 // Client is a PD (Placement Driver) client.
@@ -122,11 +123,11 @@ type Client interface {
 	GetOperator(ctx context.Context, regionID uint64) (*pdpb.GetOperatorResponse, error)
 
 	// LoadGlobalConfig gets the global config from etcd
-	LoadGlobalConfig(ctx context.Context, names []string) ([]GlobalConfigItem, error)
+	LoadGlobalConfig(ctx context.Context, configPath string) ([]GlobalConfigItem, error)
 	// StoreGlobalConfig set the config from etcd
-	StoreGlobalConfig(ctx context.Context, items []GlobalConfigItem) error
+	StoreGlobalConfig(ctx context.Context, configPath string, items []GlobalConfigItem) error
 	// WatchGlobalConfig returns an stream with all global config and updates
-	WatchGlobalConfig(ctx context.Context) (chan []GlobalConfigItem, error)
+	WatchGlobalConfig(ctx context.Context, configPath string, revision int64) (chan []GlobalConfigItem, error)
 	// UpdateOption updates the client option.
 	UpdateOption(option DynamicOption, value interface{}) error
 
@@ -1822,8 +1823,8 @@ func trimHTTPPrefix(str string) string {
 	return str
 }
 
-func (c *client) LoadGlobalConfig(ctx context.Context, names []string) ([]GlobalConfigItem, error) {
-	resp, err := c.getClient().LoadGlobalConfig(ctx, &pdpb.LoadGlobalConfigRequest{Names: names})
+func (c *client) LoadGlobalConfig(ctx context.Context, configPath string) ([]GlobalConfigItem, error) {
+	resp, err := c.getClient().LoadGlobalConfig(ctx, &pdpb.LoadGlobalConfigRequest{ConfigPath: configPath})
 	if err != nil {
 		return nil, err
 	}
@@ -1844,12 +1845,12 @@ func (c *client) LoadGlobalConfig(ctx context.Context, names []string) ([]Global
 	return res, nil
 }
 
-func (c *client) StoreGlobalConfig(ctx context.Context, items []GlobalConfigItem) error {
+func (c *client) StoreGlobalConfig(ctx context.Context, configPath string, items []GlobalConfigItem) error {
 	resArr := make([]*pdpb.GlobalConfigItem, len(items))
 	for i, it := range items {
 		resArr[i] = &pdpb.GlobalConfigItem{Name: it.Name, Value: it.Value}
 	}
-	res, err := c.getClient().StoreGlobalConfig(ctx, &pdpb.StoreGlobalConfigRequest{Changes: resArr})
+	res, err := c.getClient().StoreGlobalConfig(ctx, &pdpb.StoreGlobalConfigRequest{Changes: resArr, ConfigPath: configPath})
 	if err != nil {
 		return err
 	}
@@ -1860,9 +1861,13 @@ func (c *client) StoreGlobalConfig(ctx context.Context, items []GlobalConfigItem
 	return err
 }
 
-func (c *client) WatchGlobalConfig(ctx context.Context) (chan []GlobalConfigItem, error) {
+func (c *client) WatchGlobalConfig(ctx context.Context, configPath string, revision int64) (chan []GlobalConfigItem, error) {
+	// register watch components
 	globalConfigWatcherCh := make(chan []GlobalConfigItem, 16)
-	res, err := c.getClient().WatchGlobalConfig(ctx, &pdpb.WatchGlobalConfigRequest{})
+	res, err := c.getClient().WatchGlobalConfig(ctx, &pdpb.WatchGlobalConfigRequest{
+		ConfigPath: configPath,
+		Revision:   revision,
+	})
 	if err != nil {
 		close(globalConfigWatcherCh)
 		return nil, err
@@ -1886,7 +1891,7 @@ func (c *client) WatchGlobalConfig(ctx context.Context) (chan []GlobalConfigItem
 				}
 				arr := make([]GlobalConfigItem, len(m.Changes))
 				for j, i := range m.Changes {
-					arr[j] = GlobalConfigItem{i.GetName(), i.GetValue(), nil}
+					arr[j] = GlobalConfigItem{i.GetName(), i.GetValue(), i.GetKind(), nil}
 				}
 				globalConfigWatcherCh <- arr
 			}

client/go.mod

@@ -6,11 +6,13 @@ require (
 	github.com/opentracing/opentracing-go v1.2.0
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c
 	github.com/pingcap/failpoint v0.0.0-20210918120811-547c13e3eb00
-	github.com/pingcap/kvproto v0.0.0-20230110033234-055843a0a07d
+	github.com/pingcap/kvproto v0.0.0-20230111073505-de69cb94beae
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3
 	github.com/prometheus/client_golang v1.11.0
-	github.com/stretchr/testify v1.7.0
+	github.com/stretchr/testify v1.8.1
 	go.uber.org/goleak v1.1.11
 	go.uber.org/zap v1.20.0
-	google.golang.org/grpc v1.43.0
+	google.golang.org/grpc v1.52.0
 )
+
+replace google.golang.org/grpc v1.52.0 => google.golang.org/grpc v1.26.0

client/resourcemanager_client.go

@@ -16,6 +16,9 @@ package pd
 
 import (
 	"context"
+	"fmt"
+	"github.com/gogo/protobuf/proto"
+	"github.com/pingcap/kvproto/pkg/pdpb"
 	"time"
 
 	"github.com/pingcap/errors"
@@ -28,17 +31,19 @@ import (
 type actionType int
 
 const (
-	add    actionType = 0
-	modify actionType = 1
+	add                     actionType = 0
+	modify                  actionType = 1
+	groupSettingsPathPrefix            = "resource_group/settings"
 )
 
 // ResourceManagerClient manages resource group info and token request.
 type ResourceManagerClient interface {
-	ListResourceGroups(ctx context.Context) ([]*rmpb.ResourceGroup, error)
+	ListResourceGroups(ctx context.Context) ([]*rmpb.ResourceGroup, int64, error)
 	GetResourceGroup(ctx context.Context, resourceGroupName string) (*rmpb.ResourceGroup, error)
 	AddResourceGroup(ctx context.Context, metaGroup *rmpb.ResourceGroup) (string, error)
 	ModifyResourceGroup(ctx context.Context, metaGroup *rmpb.ResourceGroup) (string, error)
 	DeleteResourceGroup(ctx context.Context, resourceGroupName string) (string, error)
+	WatchResourceGroup(ctx context.Context, revision int64) (chan []*rmpb.ResourceGroup, error)
 	AcquireTokenBuckets(ctx context.Context, request *rmpb.TokenBucketsRequest) ([]*rmpb.TokenBucketResponse, error)
 }
 
@@ -51,17 +56,17 @@ func (c *client) resourceManagerClient() rmpb.ResourceManagerClient {
 }
 
 // ListResourceGroups loads and returns all metadata of resource groups.
-func (c *client) ListResourceGroups(ctx context.Context) ([]*rmpb.ResourceGroup, error) {
+func (c *client) ListResourceGroups(ctx context.Context) ([]*rmpb.ResourceGroup, int64, error) {
 	req := &rmpb.ListResourceGroupsRequest{}
 	resp, err := c.resourceManagerClient().ListResourceGroups(ctx, req)
 	if err != nil {
-		return nil, err
+		return nil, 0, err
 	}
 	resErr := resp.GetError()
 	if resErr != nil {
-		return nil, errors.Errorf("[resource_manager]" + resErr.Message)
+		return nil, 0, errors.Errorf("[resource_manager]" + resErr.Message)
 	}
-	return resp.GetGroups(), nil
+	return resp.GetGroups(), resp.GetRevision(), nil
 }
 
 func (c *client) GetResourceGroup(ctx context.Context, resourceGroupName string) (*rmpb.ResourceGroup, error) {
@@ -124,12 +129,58 @@ func (c *client) DeleteResourceGroup(ctx context.Context, resourceGroupName stri
 	return resp.GetBody(), nil
 }
 
+// WatchResourceGroup watches resource groups changes.
+// It returns a stream of slices of resource groups.
+// The first message in stream contains all current resource groups,
+// all subsequent messages contains new events[PUT/DELETE] for all resource groups.
+func (c *client) WatchResourceGroup(ctx context.Context, revision int64) (chan []*rmpb.ResourceGroup, error) {
+	configChan, err := c.WatchGlobalConfig(ctx, groupSettingsPathPrefix, revision)
+	resourceGroupWatcherChan := make(chan []*rmpb.ResourceGroup)
+	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				log.Error("[pd] panic in ResourceManagerClient `WatchResourceGroups`", zap.Any("error", r))
+				return
+			}
+		}()
+		for {
+			select {
+			case <-ctx.Done():
+				close(resourceGroupWatcherChan)
+				return
+			case res, ok := <-configChan:
+				if !ok {
+					close(resourceGroupWatcherChan)
+					return
+				}
+				fmt.Printf("[Client] watch %+v\n", res)
+				groups := make([]*rmpb.ResourceGroup, 0, len(res))
+				for _, item := range res {
+					switch item.ItemKind {
+					case pdpb.ItemKind_PUT:
+						group := &rmpb.ResourceGroup{}
+						if err := proto.Unmarshal([]byte(item.Value), group); err != nil {
+							return
+						}
+						groups = append(groups, group)
+					case pdpb.ItemKind_DELETE:
+						continue
+					}
+
+				}
+				resourceGroupWatcherChan <- groups
+			}
+		}
+	}()
+	return resourceGroupWatcherChan, err
+}
+
 func (c *client) AcquireTokenBuckets(ctx context.Context, request *rmpb.TokenBucketsRequest) ([]*rmpb.TokenBucketResponse, error) {
 	req := &tokenRequest{
 		done:       make(chan error, 1),
 		requestCtx: ctx,
 		clientCtx:  c.ctx,
-		Requeset:   request,
+		Request:    request,
 	}
 	c.tokenDispatcher.tokenBatchController.tokenRequestCh <- req
 	grantedTokens, err := req.Wait()
@@ -143,7 +194,7 @@ type tokenRequest struct {
 	clientCtx    context.Context
 	requestCtx   context.Context
 	done         chan error
-	Requeset     *rmpb.TokenBucketsRequest
+	Request      *rmpb.TokenBucketsRequest
 	TokenBuckets []*rmpb.TokenBucketResponse
 }
 
@@ -232,7 +283,7 @@ func (c *client) handleResourceTokenDispatcher(dispatcherCtx context.Context, tb
 }
 
 func (c *client) processTokenRequests(stream rmpb.ResourceManager_AcquireTokenBucketsClient, t *tokenRequest) error {
-	req := t.Requeset
+	req := t.Request
 	if err := stream.Send(req); err != nil {
 		err = errors.WithStack(err)
 		t.done <- err

go.mod

@@ -26,7 +26,7 @@ require (
 	github.com/pingcap/errcode v0.3.0
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c
 	github.com/pingcap/failpoint v0.0.0-20200702092429-9f69995143ce
-	github.com/pingcap/kvproto v0.0.0-20230110033234-055843a0a07d
+	github.com/pingcap/kvproto v0.0.0-20230111073505-de69cb94beae
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3
 	github.com/pingcap/sysutil v0.0.0-20211208032423-041a72e5860d
 	github.com/pingcap/tidb-dashboard v0.0.0-20221201151320-ea3ee6971f2e
@@ -44,10 +44,10 @@ require (
 	go.etcd.io/etcd v0.5.0-alpha.5.0.20220915004622-85b640cee793
 	go.uber.org/goleak v1.1.12
 	go.uber.org/zap v1.19.1
-	golang.org/x/text v0.3.7
+	golang.org/x/text v0.4.0
 	golang.org/x/time v0.0.0-20220224211638-0e9765cccd65
-	golang.org/x/tools v0.1.10
-	google.golang.org/grpc v1.26.0
+	golang.org/x/tools v0.1.12
+	google.golang.org/grpc v1.52.0
 	gotest.tools/gotestsum v1.7.0
 )
 
@@ -104,7 +104,7 @@ require (
 	github.com/google/uuid v1.1.2 // indirect
 	github.com/gorilla/websocket v1.4.2 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.0.1-0.20190118093823-f849b5445de4 // indirect
-	github.com/grpc-ecosystem/grpc-gateway v1.12.1 // indirect
+	github.com/grpc-ecosystem/grpc-gateway v1.16.0 // indirect
 	github.com/gtank/cryptopasta v0.0.0-20170601214702-1f550f6f2f69 // indirect
 	github.com/ianlancetaylor/demangle v0.0.0-20210905161508-09a460cdf81d // indirect
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect
@@ -165,16 +165,15 @@ require (
 	golang.org/x/crypto v0.0.0-20220411220226-7b82a4e95df4 // indirect
 	golang.org/x/exp v0.0.0-20220321173239-a90fa8a75705
 	golang.org/x/image v0.0.0-20200119044424-58c23975cae1 // indirect
-	golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3 // indirect
-	golang.org/x/net v0.0.0-20220722155237-a158d28d115b // indirect
-	golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be // indirect
-	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
-	golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect
-	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 // indirect
-	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
+	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
+	golang.org/x/net v0.2.0 // indirect
+	golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d // indirect
+	golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4 // indirect
+	golang.org/x/sys v0.2.0 // indirect
+	golang.org/x/term v0.2.0 // indirect
 	google.golang.org/appengine v1.4.0 // indirect
-	google.golang.org/genproto v0.0.0-20190927181202-20e1ac93f88c // indirect
-	google.golang.org/protobuf v1.28.0 // indirect
+	google.golang.org/genproto v0.0.0-20221202195650-67e5cbc046fd // indirect
+	google.golang.org/protobuf v1.28.1 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
@@ -185,6 +184,8 @@ require (
 	sigs.k8s.io/yaml v1.2.0 // indirect
 )
 
+replace google.golang.org/grpc v1.52.0 => google.golang.org/grpc v1.26.0
+
 // When you modify PD cooperatively with kvproto, this will be useful to submit the PR to PD and the PR to
 // kvproto at the same time. You can run `go mod tidy` to make it replaced with go-mod style specification.
 // After the PR to kvproto is merged, remember to comment this out and run `go mod tidy`.