support schema change

Signed-off-by: husharp <ihusharp@gmail.com>

Author: HuSharp

pkg/controller/pd_control.go

@@ -119,3 +119,11 @@ func NewFakePDClientWithAddress(pdControl *pdapi.FakePDControl, peerURL string)
 	pdControl.SetPDClientWithAddress(peerURL, pdClient)
 	return pdClient
 }
+
+// NewFakeEtcdClient creates a fake etcdClient that is set as the etcd client
+func NewFakeEtcdClient(pdControl *pdapi.FakePDControl, tc *v1alpha1.TidbCluster) *pdapi.FakeEtcdClient {
+	etcdClient := pdapi.NewFakeEtcdClient()
+	pdControl.SetEtcdClient(pdapi.Namespace(tc.GetNamespace()), tc.GetName(), etcdClient)
+
+	return etcdClient
+}

pkg/manager/member/pd_upgrader.go

@@ -15,14 +15,15 @@ package member
 
 import (
 	"fmt"
+	"net/url"
 
 	"github.com/pingcap/advanced-statefulset/client/apis/apps/v1/helper"
+	"github.com/pingcap/kvproto/pkg/pdpb"
 	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
 	"github.com/pingcap/tidb-operator/pkg/controller"
 	mngerutils "github.com/pingcap/tidb-operator/pkg/manager/utils"
 	"github.com/pingcap/tidb-operator/pkg/pdapi"
 	"github.com/pingcap/tidb-operator/pkg/third_party/k8s"
-
 	apps "k8s.io/api/apps/v1"
 	"k8s.io/klog/v2"
 )
@@ -154,6 +155,69 @@ func (u *pdUpgrader) upgradePDPod(tc *v1alpha1.TidbCluster, ordinal int32, newSe
 	}
 
 	mngerutils.SetUpgradePartition(newSet, ordinal)
+	return u.syncPDPeerUrls(tc, upgradePdName)
+}
+
+func (u *pdUpgrader) syncPDPeerUrls(tc *v1alpha1.TidbCluster, pdName string) error {
+	klog.Infof("pd upgrader: sync pd member: %s peer urls", pdName)
+	var (
+		tlsEnabled = tc.IsTLSClusterEnabled()
+		pdCli      = controller.GetPDClient(u.deps.PDControl, tc)
+		member     *pdpb.Member
+	)
+	members, err := pdCli.GetMembers()
+	if err != nil {
+		return err
+	}
+	for _, m := range members.Members {
+		if m.Name == pdName {
+			member = m
+			break
+		}
+	}
+	if member == nil {
+		return fmt.Errorf("failed to find member %s in pd cluster", pdName)
+	}
+
+	var (
+		newPeers []string
+		needSync bool
+	)
+	for _, peerUrl := range member.PeerUrls {
+		u, err := url.Parse(peerUrl)
+		if err != nil {
+			return fmt.Errorf("failed to parse peer url %s, %v", peerUrl, err)
+		}
+		// check if peer url need to be updated
+		if tlsEnabled != (u.Scheme == "https") {
+			needSync = true
+			if !tlsEnabled {
+				u.Scheme = "http"
+			} else {
+				u.Scheme = "https"
+			}
+			newPeers = append(newPeers, u.String())
+		} else {
+			newPeers = append(newPeers, peerUrl)
+		}
+	}
+
+	if needSync {
+		pdEtcdClient, err := u.deps.PDControl.GetPDEtcdClient(pdapi.Namespace(tc.Namespace), tc.Name,
+			tc.IsTLSClusterEnabled(), pdapi.ClusterRef(tc.Spec.ClusterDomain))
+
+		if err != nil {
+			return err
+		}
+		defer pdEtcdClient.Close()
+		err = pdEtcdClient.UpdateMember(member.GetMemberId(), newPeers)
+		if err != nil {
+			return err
+		}
+		klog.Infof("pd upgrader: sync pd member: %s peer urls successfully, from %v to %v",
+			pdName, member.PeerUrls, newPeers)
+	}
+
 	return nil
 }
 

pkg/manager/member/pd_upgrader_test.go

@@ -17,13 +17,13 @@ import (
 	"fmt"
 	"testing"
 
+	. "github.com/onsi/gomega"
+	"github.com/pingcap/kvproto/pkg/pdpb"
 	"github.com/pingcap/tidb-operator/pkg/apis/label"
 	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
 	"github.com/pingcap/tidb-operator/pkg/controller"
 	mngerutils "github.com/pingcap/tidb-operator/pkg/manager/utils"
 	"github.com/pingcap/tidb-operator/pkg/pdapi"
-
-	. "github.com/onsi/gomega"
 	apps "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -51,6 +51,7 @@ func TestPDUpgraderUpgrade(t *testing.T) {
 		upgrader, pdControl, _, podInformer := newPDUpgrader()
 		tc := newTidbClusterForPDUpgrader()
 		pdClient := controller.NewFakePDClient(pdControl, tc)
+		etcdClient := controller.NewFakeEtcdClient(pdControl, tc)
 
 		if test.changeFn != nil {
 			test.changeFn(tc)
@@ -95,6 +96,30 @@ func TestPDUpgraderUpgrade(t *testing.T) {
 			return healthInfo, nil
 		})
 
+		pdClient.AddReaction(pdapi.GetMembersActionType, func(action *pdapi.Action) (interface{}, error) {
+			membersInfo := &pdapi.MembersInfo{
+				Members: []*pdpb.Member{
+					{
+						Name:     PdPodName(upgradeTcName, 0),
+						MemberId: 111,
+						PeerUrls: []string{"http://upgrader-pd-0:2380"},
+					},
+					{
+						Name:     PdPodName(upgradeTcName, 1),
+						MemberId: 222,
+						PeerUrls: []string{"http://upgrader-pd-1:2380"},
+					},
+				},
+			}
+			return membersInfo, nil
+		})
+
+		pdClient.GetMembers()
+
+		etcdClient.AddReaction(pdapi.EtcdUpdateMemberActionType, func(action *pdapi.Action) (interface{}, error) {
+			return nil, nil
+		})
+
 		err := upgrader.Upgrade(tc, oldSet, newSet)
 		test.errExpectFn(g, err)
 		test.expectFn(g, tc, newSet)
@@ -314,12 +339,41 @@ func TestPDUpgraderUpgrade(t *testing.T) {
 				g.Expect(newSet.Spec.UpdateStrategy.RollingUpdate.Partition).To(Equal(pointer.Int32Ptr(1)))
 			},
 		},
+		{
+			name: "upgrade from non-tls to tls",
+			changeFn: func(tc *v1alpha1.TidbCluster) {
+				tc.Status.PD.Synced = true
+				tc.Spec = v1alpha1.TidbClusterSpec{
+					PD: &v1alpha1.PDSpec{
+						ComponentSpec: v1alpha1.ComponentSpec{
+							Image: "pingcap/pd:v3.1.0",
+						},
+					},
+					TiDB: &v1alpha1.TiDBSpec{
+						TLSClient: &v1alpha1.TiDBTLSClient{
+							Enabled: true,
+						},
+					},
+					TiKV:       &v1alpha1.TiKVSpec{},
+					TLSCluster: &v1alpha1.TLSCluster{Enabled: true},
+				}
+			},
+			changePods:        nil,
+			changeOldSet:      nil,
+			transferLeaderErr: false,
+			errExpectFn: func(g *GomegaWithT, err error) {
+				g.Expect(err).NotTo(HaveOccurred())
+			},
+			expectFn: func(g *GomegaWithT, tc *v1alpha1.TidbCluster, newSet *apps.StatefulSet) {
+				g.Expect(tc.Status.PD.Phase).To(Equal(v1alpha1.UpgradePhase))
+				g.Expect(newSet.Spec.UpdateStrategy.RollingUpdate.Partition).To(Equal(pointer.Int32Ptr(1)))
+			},
+		},
 	}
 
 	for i := range tests {
 		testFn(&tests[i])
 	}
-
 }
 
 func TestChoosePDToTransferFromMembers(t *testing.T) {
@@ -447,6 +501,16 @@ func TestChoosePDToTransferFromMembers(t *testing.T) {
 
 func newPDUpgrader() (Upgrader, *pdapi.FakePDControl, *controller.FakePodControl, podinformers.PodInformer) {
 	fakeDeps := controller.NewFakeDependencies()
+
+	informer := fakeDeps.KubeInformerFactory
+	informer.Core().V1().Secrets().Informer().GetIndexer().Add(&corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "upgrader-cluster-client-secret",
+			Namespace: corev1.NamespaceDefault,
+		},
+	})
+	fakeDeps.PDControl = pdapi.NewFakePDControl(informer.Core().V1().Secrets().Lister())
+
 	pdUpgrader := &pdUpgrader{deps: fakeDeps}
 	pdControl := fakeDeps.PDControl.(*pdapi.FakePDControl)
 	podControl := fakeDeps.PodControl.(*controller.FakePodControl)

pkg/pdapi/fake_pdapi.go

@@ -291,3 +291,75 @@ func (c *FakePDClient) GetRecoveringMark() (bool, error) {
 
 	return true, nil
 }
+
+const (
+	EtcdGetActionType          ActionType = "Get"
+	EtcdPutActionType          ActionType = "Put"
+	EtcdDeleteActionType       ActionType = "Delete"
+	EtcdUpdateMemberActionType ActionType = "UpdateMember"
+	EtcdCloseActionType        ActionType = "Close"
+)
+
+// FakeEtcdClient implements a fake version of EtcdClient.
+type FakeEtcdClient struct {
+	reactions map[ActionType]Reaction
+}
+
+func NewFakeEtcdClient() *FakeEtcdClient {
+	return &FakeEtcdClient{reactions: map[ActionType]Reaction{}}
+}
+
+func (c *FakeEtcdClient) AddReaction(actionType ActionType, reaction Reaction) {
+	c.reactions[actionType] = reaction
+}
+
+// fakeAPI is a small helper for fake API calls
+func (c *FakeEtcdClient) fakeAPI(actionType ActionType, action *Action) (interface{}, error) {
+	if reaction, ok := c.reactions[actionType]; ok {
+		result, err := reaction(action)
+		if err != nil {
+			return nil, err
+		}
+		return result, nil
+	}
+	return nil, &NotFoundReaction{actionType}
+}
+
+func (c *FakeEtcdClient) Get(_ string, _ bool) (kvs []*KeyValue, err error) {
+	action := &Action{}
+	nil, err := c.fakeAPI(EtcdGetActionType, action)
+	if err != nil {
+		return
+	}
+	return
+}
+
+func (c *FakeEtcdClient) PutKey(_, _ string) error {
+	action := &Action{}
+	_, err := c.fakeAPI(EtcdPutActionType, action)
+	return err
+}
+
+func (c *FakeEtcdClient) PutTTLKey(_, _ string, _ int64) error {
+	action := &Action{}
+	_, err := c.fakeAPI(EtcdPutActionType, action)
+	return err
+}
+
+func (c *FakeEtcdClient) DeleteKey(_ string) error {
+	action := &Action{}
+	_, err := c.fakeAPI(EtcdDeleteActionType, action)
+	return err
+}
+
+func (c *FakeEtcdClient) UpdateMember(_ uint64, _ []string) error {
+	action := &Action{}
+	_, err := c.fakeAPI(EtcdUpdateMemberActionType, action)
+	return err
+}
+
+func (c *FakeEtcdClient) Close() error {
+	action := &Action{}
+	_, err := c.fakeAPI(EtcdCloseActionType, action)
+	return err
+}

pkg/pdapi/pd_control.go

@@ -337,10 +337,24 @@ type FakePDControl struct {
 
 func NewFakePDControl(secretLister corelisterv1.SecretLister) *FakePDControl {
 	return &FakePDControl{
-		defaultPDControl{secretLister: secretLister, pdClients: map[string]PDClient{}},
+		defaultPDControl{secretLister: secretLister, pdClients: map[string]PDClient{}, pdEtcdClients: map[string]PDEtcdClient{}},
 	}
 }
 
+func (fpc *FakePDControl) GetPDClient(namespace Namespace, tcName string, tlsEnabled bool, opts ...Option) PDClient {
+	if tlsEnabled {
+		return fpc.defaultPDControl.pdClients[genClientKey("http", namespace, tcName, "")]
+	}
+	return fpc.defaultPDControl.GetPDClient(namespace, tcName, tlsEnabled, opts...)
+}
+
+func (fpc *FakePDControl) GetPDEtcdClient(namespace Namespace, tcName string, tlsEnabled bool, opts ...Option) (PDEtcdClient, error) {
+	if tlsEnabled {
+		return fpc.defaultPDControl.pdEtcdClients[genClientKey("http", namespace, tcName, "")], nil
+	}
+	return fpc.defaultPDControl.GetPDEtcdClient(namespace, tcName, tlsEnabled, opts...)
+}
+
 func (fpc *FakePDControl) SetPDClient(namespace Namespace, tcName string, pdclient PDClient) {
 	fpc.defaultPDControl.pdClients[genClientKey("http", namespace, tcName, "")] = pdclient
 }
@@ -352,3 +366,7 @@ func (fpc *FakePDControl) SetPDClientWithClusterDomain(namespace Namespace, tcNa
 func (fpc *FakePDControl) SetPDClientWithAddress(peerURL string, pdclient PDClient) {
 	fpc.defaultPDControl.pdClients[peerURL] = pdclient
 }
+
+func (fpc *FakePDControl) SetEtcdClient(namespace Namespace, tcName string, etcdClient PDEtcdClient) {
+	fpc.defaultPDControl.pdEtcdClients[genClientKey("http", namespace, tcName, "")] = etcdClient
+}

pkg/pdapi/pdetcd.go

@@ -33,10 +33,12 @@ type PDEtcdClient interface {
 	Get(key string, prefix bool) (kvs []*KeyValue, err error)
 	// PutKey will put key to the target pd etcd cluster
 	PutKey(key, value string) error
-	// PutKey will put key with ttl to the target pd etcd cluster
+	// PutTTLKey will put key with ttl to the target pd etcd cluster
 	PutTTLKey(key, value string, ttl int64) error
 	// DeleteKey will delete key from the target pd etcd cluster
 	DeleteKey(key string) error
+	// UpdateMember will update the member list of the target pd etcd cluster
+	UpdateMember(id uint64, peerAddrs []string) error
 	// Close will close the etcd connection
 	Close() error
 }
@@ -121,3 +123,15 @@ func (c *pdEtcdClient) DeleteKey(key string) error {
 	}
 	return nil
 }
+
+func (c *pdEtcdClient) UpdateMember(id uint64, peerAddrs []string) error {
+	ctx, cancel := context.WithTimeout(context.Background(), c.timeout)
+	defer cancel()
+
+	_, err := c.etcdClient.MemberUpdate(ctx, id, peerAddrs)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}