infosync: integrate PD HTTP client into the label manager

DEPS.bzl

@@ -7132,13 +7132,13 @@ def go_deps():
         name = "com_github_tikv_pd_client",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/tikv/pd/client",
-        sha256 = "014bb8796797b8b5cecc22866a1aab8491e3718c540168ac91257cf7f220cc84",
-        strip_prefix = "github.com/tikv/pd/client@v0.0.0-20231117041718-dda748abe55d",
+        sha256 = "440821579da980d0405695b463da892608a59252a296cd7e52b4f97881c5fdb7",
+        strip_prefix = "github.com/tikv/pd/client@v0.0.0-20231121080541-8919bc11f770",
         urls = [
-            "http://bazel-cache.pingcap.net:8080/gomod/github.com/tikv/pd/client/com_github_tikv_pd_client-v0.0.0-20231117041718-dda748abe55d.zip",
-            "http://ats.apps.svc/gomod/github.com/tikv/pd/client/com_github_tikv_pd_client-v0.0.0-20231117041718-dda748abe55d.zip",
-            "https://cache.hawkingrei.com/gomod/github.com/tikv/pd/client/com_github_tikv_pd_client-v0.0.0-20231117041718-dda748abe55d.zip",
-            "https://storage.googleapis.com/pingcapmirror/gomod/github.com/tikv/pd/client/com_github_tikv_pd_client-v0.0.0-20231117041718-dda748abe55d.zip",
+            "http://bazel-cache.pingcap.net:8080/gomod/github.com/tikv/pd/client/com_github_tikv_pd_client-v0.0.0-20231121080541-8919bc11f770.zip",
+            "http://ats.apps.svc/gomod/github.com/tikv/pd/client/com_github_tikv_pd_client-v0.0.0-20231121080541-8919bc11f770.zip",
+            "https://cache.hawkingrei.com/gomod/github.com/tikv/pd/client/com_github_tikv_pd_client-v0.0.0-20231121080541-8919bc11f770.zip",
+            "https://storage.googleapis.com/pingcapmirror/gomod/github.com/tikv/pd/client/com_github_tikv_pd_client-v0.0.0-20231121080541-8919bc11f770.zip",
         ],
     )
     go_repository(

go.mod

@@ -103,7 +103,7 @@ require (
 	github.com/tdakkota/asciicheck v0.2.0
 	github.com/tiancaiamao/appdash v0.0.0-20181126055449-889f96f722a2
 	github.com/tikv/client-go/v2 v2.0.8-0.20231116051730-1c2351c28173
-	github.com/tikv/pd/client v0.0.0-20231117041718-dda748abe55d
+	github.com/tikv/pd/client v0.0.0-20231121080541-8919bc11f770
 	github.com/timakin/bodyclose v0.0.0-20230421092635-574207250966
 	github.com/twmb/murmur3 v1.1.6
 	github.com/uber/jaeger-client-go v2.22.1+incompatible

go.sum

@@ -993,8 +993,8 @@ github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a h1:J/YdBZ46WKpXsxsW
 github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a/go.mod h1:h4xBhSNtOeEosLJ4P7JyKXX7Cabg7AVkWCK5gV2vOrM=
 github.com/tikv/client-go/v2 v2.0.8-0.20231116051730-1c2351c28173 h1:lmJzX0kqrV7kO21wrZPbtjkidzwbDCfXeQrhDWEi5dE=
 github.com/tikv/client-go/v2 v2.0.8-0.20231116051730-1c2351c28173/go.mod h1:BOGTSZtbMHEnGC4HOpbONdnTQF+E9nb2Io7c3P9sb7g=
-github.com/tikv/pd/client v0.0.0-20231117041718-dda748abe55d h1:6isljjnUH8zzkJx2X8MUGh+5AlMv+pCEhCy5MSyuhSM=
-github.com/tikv/pd/client v0.0.0-20231117041718-dda748abe55d/go.mod h1:cd6zBqRM9aogxf26K8NnFRPVtq9BnRE59tKEpX8IaWQ=
+github.com/tikv/pd/client v0.0.0-20231121080541-8919bc11f770 h1:YSXDKT9+KngRSAShoSQVKD/CK1kR4X/9hutKkSK9gn0=
+github.com/tikv/pd/client v0.0.0-20231121080541-8919bc11f770/go.mod h1:cd6zBqRM9aogxf26K8NnFRPVtq9BnRE59tKEpX8IaWQ=
 github.com/timakin/bodyclose v0.0.0-20230421092635-574207250966 h1:quvGphlmUVU+nhpFa4gg4yJyTRJ13reZMDHrKwYw53M=
 github.com/timakin/bodyclose v0.0.0-20230421092635-574207250966/go.mod h1:27bSVNWSBOHm+qRp1T9qzaIpsWEP6TbUnei/43HK+PQ=
 github.com/tklauser/go-sysconf v0.3.9/go.mod h1:11DU/5sG7UexIrp/O6g35hrWzu0JxlwQ3LSFUzyeuhs=

pkg/ddl/attributes_sql_test.go

@@ -252,7 +252,7 @@ PARTITION BY RANGE (c) (
 func TestFlashbackTable(t *testing.T) {
 	store, dom := testkit.CreateMockStoreAndDomain(t)
 
-	_, err := infosync.GlobalInfoSyncerInit(context.Background(), dom.DDL().GetID(), dom.ServerID, dom.GetEtcdClient(), dom.GetEtcdClient(), dom.GetPDClient(), keyspace.CodecV1, true)
+	_, err := infosync.GlobalInfoSyncerInit(context.Background(), dom.DDL().GetID(), dom.ServerID, dom.GetEtcdClient(), dom.GetEtcdClient(), dom.GetPDClient(), dom.GetPDHTTPClient(), keyspace.CodecV1, true)
 	require.NoError(t, err)
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
@@ -310,7 +310,7 @@ PARTITION BY RANGE (c) (
 func TestDropTable(t *testing.T) {
 	store, dom := testkit.CreateMockStoreAndDomain(t)
 
-	_, err := infosync.GlobalInfoSyncerInit(context.Background(), dom.DDL().GetID(), dom.ServerID, dom.GetEtcdClient(), dom.GetEtcdClient(), dom.GetPDClient(), keyspace.CodecV1, true)
+	_, err := infosync.GlobalInfoSyncerInit(context.Background(), dom.DDL().GetID(), dom.ServerID, dom.GetEtcdClient(), dom.GetEtcdClient(), dom.GetPDClient(), dom.GetPDHTTPClient(), keyspace.CodecV1, true)
 	require.NoError(t, err)
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
@@ -363,7 +363,7 @@ PARTITION BY RANGE (c) (
 func TestCreateWithSameName(t *testing.T) {
 	store, dom := testkit.CreateMockStoreAndDomain(t)
 
-	_, err := infosync.GlobalInfoSyncerInit(context.Background(), dom.DDL().GetID(), dom.ServerID, dom.GetEtcdClient(), dom.GetEtcdClient(), dom.GetPDClient(), keyspace.CodecV1, true)
+	_, err := infosync.GlobalInfoSyncerInit(context.Background(), dom.DDL().GetID(), dom.ServerID, dom.GetEtcdClient(), dom.GetEtcdClient(), dom.GetPDClient(), dom.GetPDHTTPClient(), keyspace.CodecV1, true)
 	require.NoError(t, err)
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
@@ -427,7 +427,7 @@ PARTITION BY RANGE (c) (
 func TestPartition(t *testing.T) {
 	store, dom := testkit.CreateMockStoreAndDomain(t)
 
-	_, err := infosync.GlobalInfoSyncerInit(context.Background(), dom.DDL().GetID(), dom.ServerID, dom.GetEtcdClient(), dom.GetEtcdClient(), dom.GetPDClient(), keyspace.CodecV1, true)
+	_, err := infosync.GlobalInfoSyncerInit(context.Background(), dom.DDL().GetID(), dom.ServerID, dom.GetEtcdClient(), dom.GetEtcdClient(), dom.GetPDClient(), dom.GetPDHTTPClient(), keyspace.CodecV1, true)
 	require.NoError(t, err)
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")

pkg/ddl/label/BUILD.bazel

@@ -13,6 +13,7 @@ go_library(
         "//pkg/parser/ast",
         "//pkg/tablecodec",
         "//pkg/util/codec",
+        "@com_github_tikv_pd_client//http",
         "@in_gopkg_yaml_v2//:yaml_v2",
     ],
 )
@@ -32,6 +33,7 @@ go_test(
         "//pkg/parser/ast",
         "//pkg/testkit/testsetup",
         "@com_github_stretchr_testify//require",
+        "@com_github_tikv_pd_client//http",
         "@org_uber_go_goleak//:goleak",
     ],
 )

pkg/ddl/label/attributes.go

@@ -17,6 +17,8 @@ package label
 import (
 	"fmt"
 	"strings"
+
+	pd "github.com/tikv/pd/client/http"
 )
 
 const (
@@ -37,15 +39,9 @@ const (
 	AttributesDuplicated
 )
 
-// Label is used to describe attributes
-type Label struct {
-	Key   string `json:"key,omitempty"`
-	Value string `json:"value,omitempty"`
-}
-
 // NewLabel creates a new label for a given string.
-func NewLabel(attr string) (Label, error) {
-	l := Label{}
+func NewLabel(attr string) (pd.RegionLabel, error) {
+	l := pd.RegionLabel{}
 	kv := strings.Split(attr, "=")
 	if len(kv) != 2 {
 		return l, fmt.Errorf("%w: %s", ErrInvalidAttributesFormat, attr)
@@ -66,14 +62,14 @@ func NewLabel(attr string) (Label, error) {
 	return l, nil
 }
 
-// Restore converts a Attribute to a string.
-func (l *Label) Restore() string {
+// RestoreRegionLabel converts a Attribute to a string.
+func RestoreRegionLabel(l *pd.RegionLabel) string {
 	return l.Key + "=" + l.Value
 }
 
 // CompatibleWith will check if two constraints are compatible.
 // Return (compatible, duplicated).
-func (l *Label) CompatibleWith(o *Label) AttributesCompatibility {
+func CompatibleWith(l *pd.RegionLabel, o *pd.RegionLabel) AttributesCompatibility {
 	if l.Key != o.Key {
 		return AttributesCompatible
 	}
@@ -85,26 +81,23 @@ func (l *Label) CompatibleWith(o *Label) AttributesCompatibility {
 	return AttributesIncompatible
 }
 
-// Labels is a slice of Label.
-type Labels []Label
-
 // NewLabels creates a slice of Label for given attributes.
-func NewLabels(attrs []string) (Labels, error) {
-	labels := make(Labels, 0, len(attrs))
+func NewLabels(attrs []string) ([]pd.RegionLabel, error) {
+	labels := make([]pd.RegionLabel, 0, len(attrs))
 	for _, attr := range attrs {
 		label, err := NewLabel(attr)
 		if err != nil {
 			return nil, err
 		}
-		if err := labels.Add(label); err != nil {
+		if err := Add(&labels, label); err != nil {
 			return nil, err
 		}
 	}
 	return labels, nil
 }
 
-// Restore converts Attributes to a string.
-func (labels *Labels) Restore() string {
+// RestoreRegionLabels converts Attributes to a string.
+func RestoreRegionLabels(labels *[]pd.RegionLabel) string {
 	var sb strings.Builder
 	for i, label := range *labels {
 		switch label.Key {
@@ -117,25 +110,25 @@ func (labels *Labels) Restore() string {
 			sb.WriteByte(',')
 		}
 		sb.WriteByte('"')
-		sb.WriteString(label.Restore())
+		sb.WriteString(RestoreRegionLabel(&label))
 		sb.WriteByte('"')
 	}
 	return sb.String()
 }
 
 // Add will add a new attribute, with validation of all attributes.
-func (labels *Labels) Add(label Label) error {
+func Add(labels *[]pd.RegionLabel, label pd.RegionLabel) error {
 	for i := range *labels {
 		l := (*labels)[i]
-		res := label.CompatibleWith(&l)
+		res := CompatibleWith(&label, &l)
 		if res == AttributesCompatible {
 			continue
 		}
 		if res == AttributesDuplicated {
 			return nil
 		}
-		s1 := label.Restore()
-		s2 := l.Restore()
+		s1 := RestoreRegionLabel(&label)
+		s2 := RestoreRegionLabel(&l)
 		return fmt.Errorf("'%s' and '%s' are conflicted", s1, s2)
 	}
 

pkg/ddl/label/attributes_test.go

@@ -18,28 +18,29 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	pd "github.com/tikv/pd/client/http"
 )
 
 func TestNewLabel(t *testing.T) {
 	type TestCase struct {
 		name  string
 		input string
-		label Label
+		label pd.RegionLabel
 	}
 
 	tests := []TestCase{
 		{
 			name:  "normal",
 			input: "merge_option=allow",
-			label: Label{
+			label: pd.RegionLabel{
 				Key:   "merge_option",
 				Value: "allow",
 			},
 		},
 		{
 			name:  "normal with space",
 			input: " merge_option=allow ",
-			label: Label{
+			label: pd.RegionLabel{
 				Key:   "merge_option",
 				Value: "allow",
 			},
@@ -58,7 +59,7 @@ func TestNewLabel(t *testing.T) {
 func TestRestoreLabel(t *testing.T) {
 	type TestCase struct {
 		name   string
-		input  Label
+		input  pd.RegionLabel
 		output string
 	}
 
@@ -83,7 +84,7 @@ func TestRestoreLabel(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			output := test.input.Restore()
+			output := RestoreRegionLabel(&test.input)
 			require.Equal(t, test.output, output)
 		})
 	}
@@ -124,8 +125,8 @@ func TestNewLabels(t *testing.T) {
 func TestAddLabels(t *testing.T) {
 	type TestCase struct {
 		name   string
-		labels Labels
-		label  Label
+		labels []pd.RegionLabel
+		label  pd.RegionLabel
 		err    bool
 	}
 
@@ -154,7 +155,7 @@ func TestAddLabels(t *testing.T) {
 		},
 		{
 			"duplicated attributes, skip",
-			append(labels, Label{
+			append(labels, pd.RegionLabel{
 				Key:   "merge_option",
 				Value: "allow",
 			}),
@@ -171,7 +172,7 @@ func TestAddLabels(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			err = test.labels.Add(test.label)
+			err = Add(&test.labels, test.label)
 			if test.err {
 				require.Error(t, err)
 			} else {
@@ -185,7 +186,7 @@ func TestAddLabels(t *testing.T) {
 func TestRestoreLabels(t *testing.T) {
 	type TestCase struct {
 		name   string
-		input  Labels
+		input  []pd.RegionLabel
 		output string
 	}
 
@@ -203,29 +204,29 @@ func TestRestoreLabels(t *testing.T) {
 	tests := []TestCase{
 		{
 			"normal1",
-			Labels{},
+			[]pd.RegionLabel{},
 			"",
 		},
 		{
 			"normal2",
-			Labels{input1, input2},
+			[]pd.RegionLabel{input1, input2},
 			`"merge_option=allow","key=value"`,
 		},
 		{
 			"normal3",
-			Labels{input3, input4, input5},
+			[]pd.RegionLabel{input3, input4, input5},
 			"",
 		},
 		{
 			"normal4",
-			Labels{input1, input2, input3},
+			[]pd.RegionLabel{input1, input2, input3},
 			`"merge_option=allow","key=value"`,
 		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			output := test.input.Restore()
+			output := RestoreRegionLabels(&test.input)
 			require.Equal(t, test.output, output)
 		})
 	}