ticdc: upgrade ticdc first if version >= v5.1.0

go.mod

@@ -138,7 +138,7 @@ require (
 	github.com/cespare/xxhash/v2 v2.1.1 // indirect
 	github.com/chai2010/gettext-go v0.0.0-20170215093142-bf70f2a70fb1 // indirect
 	github.com/containerd/containerd v1.4.11 // indirect
-	github.com/coreos/go-semver v0.3.0 // indirect
+	github.com/coreos/go-semver v0.3.0
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dimchansky/utfbom v1.1.0 // indirect
 	github.com/docker/distribution v2.7.1+incompatible // indirect

pkg/apis/pingcap/v1alpha1/tidbcluster.go

@@ -26,6 +26,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"github.com/pingcap/tidb-operator/pkg/apis/label"
+	"github.com/pingcap/tidb-operator/pkg/apis/util"
 )
 
 const (
@@ -44,9 +45,6 @@ const (
 	// shutdown a TiCDC pod.
 	defaultTiCDCGracefulShutdownTimeout = 10 * time.Minute
 	defaultPDStartTimeout               = 30
-
-	// the latest version
-	versionLatest = "latest"
 )
 
 var (
@@ -92,7 +90,7 @@ func (tc *TidbCluster) PDVersion() string {
 		return ""
 	}
 
-	return getImageVersion(tc.PDImage())
+	return util.GetImageVersion(tc.PDImage())
 }
 
 // TiKVImage return the image used by TiKV.
@@ -128,7 +126,7 @@ func (tc *TidbCluster) TiKVVersion() string {
 		return ""
 	}
 
-	return getImageVersion(tc.TiKVImage())
+	return util.GetImageVersion(tc.TiKVImage())
 }
 
 func (tc *TidbCluster) TiKVContainerPrivilege() *bool {
@@ -189,7 +187,7 @@ func (tc *TidbCluster) TiFlashVersion() string {
 		return ""
 	}
 
-	return getImageVersion(tc.TiFlashImage())
+	return util.GetImageVersion(tc.TiFlashImage())
 }
 
 func (tc *TidbCluster) TiFlashContainerPrivilege() *bool {
@@ -309,17 +307,7 @@ func (tc *TidbCluster) TiDBVersion() string {
 		return ""
 	}
 
-	return getImageVersion(tc.TiDBImage())
-}
-
-// getImageVersion returns the verion of a image
-func getImageVersion(image string) string {
-	colonIdx := strings.LastIndexByte(image, ':')
-	if colonIdx >= 0 {
-		return image[colonIdx+1:]
-	}
-
-	return versionLatest
+	return util.GetImageVersion(tc.TiDBImage())
 }
 
 // PumpImage return the image used by Pump.

pkg/apis/util/util.go

@@ -0,0 +1,31 @@
+// Copyright 2023 PingCAP, Inc.
+//
+// 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 util
+
+import "strings"
+
+const (
+	// `latest` is a special version that always points to the latest version of the image.
+	VersionLatest = "latest"
+)
+
+// GetImageVersion returns the verion of a image
+func GetImageVersion(image string) string {
+	colonIdx := strings.LastIndexByte(image, ':')
+	if colonIdx >= 0 {
+		return image[colonIdx+1:]
+	}
+
+	return VersionLatest
+}

pkg/apis/util/util_test.go

@@ -0,0 +1,45 @@
+// Copyright 2023 PingCAP, Inc.
+//
+// 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 util
+
+import (
+	"testing"
+
+	. "github.com/onsi/gomega"
+)
+
+func TestGetImageVersion(t *testing.T) {
+	g := NewGomegaWithT(t)
+	cases := []struct {
+		image    string
+		expected string
+	}{
+		{
+			image:    "foo/bar:latest",
+			expected: "latest",
+		},
+		{
+			image:    "foo/bar:1.2.3",
+			expected: "1.2.3",
+		},
+		{
+			image:    "foo/bar",
+			expected: "latest",
+		},
+	}
+
+	for _, c := range cases {
+		g.Expect(GetImageVersion(c.image)).To(Equal(c.expected))
+	}
+}

pkg/controller/tidbcluster/tidb_cluster_control.go

@@ -14,15 +14,22 @@
 package tidbcluster
 
 import (
+	"errors"
+	"fmt"
+	"regexp"
+
+	"github.com/coreos/go-semver/semver"
 	v1 "k8s.io/api/core/v1"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	errorutils "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/klog/v2"
 
 	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
 	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1/defaulting"
 	v1alpha1validation "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1/validation"
+	"github.com/pingcap/tidb-operator/pkg/apis/util"
 	"github.com/pingcap/tidb-operator/pkg/controller"
 	"github.com/pingcap/tidb-operator/pkg/manager"
 	"github.com/pingcap/tidb-operator/pkg/manager/member"
@@ -58,7 +65,9 @@ func NewDefaultTidbClusterControl(
 	discoveryManager member.TidbDiscoveryManager,
 	tidbClusterStatusManager manager.Manager,
 	conditionUpdater TidbClusterConditionUpdater,
-	recorder record.EventRecorder) ControlInterface {
+	recorder record.EventRecorder,
+	deps *controller.Dependencies,
+) ControlInterface {
 	return &defaultTidbClusterControl{
 		tcControl:                tcControl,
 		pdMemberManager:          pdMemberManager,
@@ -77,6 +86,7 @@ func NewDefaultTidbClusterControl(
 		tidbClusterStatusManager: tidbClusterStatusManager,
 		conditionUpdater:         conditionUpdater,
 		recorder:                 recorder,
+		deps:                     deps,
 	}
 }
 
@@ -98,6 +108,7 @@ type defaultTidbClusterControl struct {
 	tidbClusterStatusManager manager.Manager
 	conditionUpdater         TidbClusterConditionUpdater
 	recorder                 record.EventRecorder
+	deps                     *controller.Dependencies
 }
 
 // UpdateStatefulSet executes the core logic loop for a tidbcluster.
@@ -174,6 +185,23 @@ func (c *defaultTidbClusterControl) updateTidbCluster(tc *v1alpha1.TidbCluster)
 		return err
 	}
 
+	// Upgrade TiCDC first if needed.
+	upgradeTiCDCFirst, err := NeedToUpdateTiCDCFirst(ns, tcName, c.deps)
+	if err != nil {
+		return err
+	}
+	if upgradeTiCDCFirst {
+		// works that should be done to make the ticdc cluster current state match the desired state:
+		//   - waiting for the pd cluster available(pd cluster is in quorum)
+		//   - waiting for the tikv cluster available(at least one peer works)
+		//   - create or update ticdc deployment
+		//   - sync ticdc cluster status from pd to TidbCluster object
+		if err := c.ticdcMemberManager.Sync(tc); err != nil {
+			metrics.ClusterUpdateErrors.WithLabelValues(ns, tcName, "ticdc").Inc()
+			return err
+		}
+	}
+
 	// works that should be done to make the pd cluster current state match the desired state:
 	//   - create or update the pd service
 	//   - create or update the pd headless service
@@ -250,16 +278,17 @@ func (c *defaultTidbClusterControl) updateTidbCluster(tc *v1alpha1.TidbCluster)
 		return err
 	}
 
-	// works that should be done to make the ticdc cluster current state match the desired state:
-	//   - waiting for the pd cluster available(pd cluster is in quorum)
-	//   - waiting for the tikv cluster available(at least one peer works)
-	//   - create or update ticdc deployment
-	//   - sync ticdc cluster status from pd to TidbCluster object
-	if err := c.ticdcMemberManager.Sync(tc); err != nil {
-		metrics.ClusterUpdateErrors.WithLabelValues(ns, tcName, "ticdc").Inc()
-		return err
+	if !upgradeTiCDCFirst {
+		// works that should be done to make the ticdc cluster current state match the desired state:
+		//   - waiting for the pd cluster available(pd cluster is in quorum)
+		//   - waiting for the tikv cluster available(at least one peer works)
+		//   - create or update ticdc deployment
+		//   - sync ticdc cluster status from pd to TidbCluster object
+		if err := c.ticdcMemberManager.Sync(tc); err != nil {
+			metrics.ClusterUpdateErrors.WithLabelValues(ns, tcName, "ticdc").Inc()
+			return err
+		}
 	}
-
 	// syncing the labels from Pod to PVC and PV, these labels include:
 	//   - label.StoreIDLabelKey
 	//   - label.MemberIDLabelKey
@@ -296,6 +325,75 @@ func (c *defaultTidbClusterControl) updateTidbCluster(tc *v1alpha1.TidbCluster)
 	return err
 }
 
+// NeedToUpdateTiCDCFirst checks if the TiCDC version is greater than or equal to v5.1.0.
+// If so, we need to update TiCDC first before updating other components.
+// See more: https://github.com/pingcap/tidb-operator/issues/4966#issuecomment-1606727165
+func NeedToUpdateTiCDCFirst(ns string, tcName string, deps *controller.Dependencies) (bool, error) {
+	if deps == nil {
+		return false, nil
+	}
+
+	// Check if need to upgrade TiCDC first.
+	upgradeTiCDCFirst := false
+
+	oldSts, err := deps.StatefulSetLister.StatefulSets(ns).Get(controller.TiCDCMemberName(tcName))
+	if err != nil && !apierrors.IsNotFound(err) {
+		return false, fmt.Errorf("updateTidbCluster: failed to get sts %s for cluster %s/%s, error: %s", controller.TiCDCMemberName(tcName), ns, tcName, err)
+	}
+	stsNotExist := apierrors.IsNotFound(err)
+	if !stsNotExist {
+		// Get TiCDC version from old sts.
+		ticdcImage := oldSts.Spec.Template.Spec.Containers[0].Image
+		oldTiCDCVersion := util.GetImageVersion(ticdcImage)
+		upgradeTiCDCFirst, err = needToUpdateTiCDCFirst(oldTiCDCVersion)
+		// For some reason, we can't get the old TiCDC version, so we can't determine if we need to upgrade TiCDC first.
+		// In this case, we just return false.
+		if err != nil {
+			klog.Warningf("cluster %s/%s can't determine if need to upgrade TiCDC first, error: %s", ns, tcName, err)
+			return false, nil
+		}
+	}
+
+	return upgradeTiCDCFirst, nil
+}
+
+func needToUpdateTiCDCFirst(ticdcVersion string) (bool, error) {
+	// NOTE: 2023-07-04 is the date we add this check,
+	// so the latest version should be greater than v5.1.0.
+	if ticdcVersion == util.VersionLatest {
+		return true, nil
+	}
+	ver := &semver.Version{}
+	sanitizedVersion, err := sanitizeVersion(ticdcVersion)
+	if err != nil {
+		return false, err
+	}
+	if err := ver.Set(sanitizedVersion); err != nil {
+		return false, errors.New("ticdc version is invalid")
+	}
+
+	if ver.LessThan(*semver.New("5.1.0-alpha")) {
+		return false, nil
+	} else {
+		return true, nil
+	}
+}
+
+// versionRe matches the version string like v7.2.0, v7.2.0-pre,
+// v6.1.3-20230517-5484207 and v5.1.1-20211227 to extract the version.
+// You can find all the version strings in https://hub.docker.com/r/pingcap/ticdc/tags.
+var versionRe = regexp.MustCompile(`v(\d+\.\d+\.\d+)(?:-\+)?`)
+
+// sanitizeVersion remove the prefix "v" and suffix git hash.
+// For example, the version string v3.0.0-rc.1-10-gd3f8e0a becomes 3.0.0-rc.1.
+func sanitizeVersion(v string) (string, error) {
+	match := versionRe.FindStringSubmatch(v)
+	if len(match) > 1 {
+		return match[1], nil
+	}
+	return "", fmt.Errorf("invalid version string: %s", v)
+}
+
 func (c *defaultTidbClusterControl) recordMetrics(tc *v1alpha1.TidbCluster) {
 	ns := tc.GetNamespace()
 	tcName := tc.GetName()