From c348f7190432111f686c994be38a60408c39a651 Mon Sep 17 00:00:00 2001
From: Wang Kang <wangkang@pingcap.com>
Date: Fri, 16 Sep 2022 00:53:13 +0800
Subject: [PATCH] cluster: support TiCDC cluster id

---
 embed/templates/scripts/run_cdc.sh.tpl | 3 +++
 pkg/cluster/spec/cdc.go                | 7 ++++++-
 pkg/cluster/template/scripts/cdc.go    | 7 +++++++
 pkg/tidbver/tidbver.go                 | 5 +++++
 4 files changed, 21 insertions(+), 1 deletion(-)

diff --git a/embed/templates/scripts/run_cdc.sh.tpl b/embed/templates/scripts/run_cdc.sh.tpl
index 028d12642d..388f2deb7a 100644
--- a/embed/templates/scripts/run_cdc.sh.tpl
+++ b/embed/templates/scripts/run_cdc.sh.tpl
@@ -42,6 +42,9 @@ exec bin/cdc server \
 {{- if .TZ}}
     --tz "{{.TZ}}" \
 {{- end}}
+{{- if .ClusterID}}
+    --cluster-id {{.ClusterID}} \
+{{- end}}
 {{- if .ConfigFileEnabled}}
     --config conf/cdc.toml \
 {{- end}}
diff --git a/pkg/cluster/spec/cdc.go b/pkg/cluster/spec/cdc.go
index b03f2ef956..9a866647ea 100644
--- a/pkg/cluster/spec/cdc.go
+++ b/pkg/cluster/spec/cdc.go
@@ -43,6 +43,7 @@ type CDCSpec struct {
 	Offline         bool                   `yaml:"offline,omitempty"`
 	GCTTL           int64                  `yaml:"gc-ttl,omitempty" validate:"gc-ttl:editable"`
 	TZ              string                 `yaml:"tz,omitempty" validate:"tz:editable"`
+	TiCDCClusterID  string                 `yaml:"ticdc_cluster_id"`
 	NumaNode        string                 `yaml:"numa_node,omitempty" validate:"numa_node:editable"`
 	Config          map[string]interface{} `yaml:"config,omitempty" validate:"config:ignore"`
 	ResourceControl meta.ResourceControl   `yaml:"resource_control,omitempty" validate:"resource_control:editable"`
@@ -171,6 +172,10 @@ func (i *CDCInstance) InitConfig(
 		}
 	}
 
+	if !tidbver.TiCDCSupportClusterID(clusterVersion) && spec.TiCDCClusterID != "" {
+		return errors.New("ticdc_cluster_id is only supported with TiCDC version v6.2.0 or later")
+	}
+
 	cfg := scripts.NewCDCScript(
 		i.GetHost(),
 		paths.Deploy,
@@ -178,7 +183,7 @@ func (i *CDCInstance) InitConfig(
 		enableTLS,
 		spec.GCTTL,
 		spec.TZ,
-	).WithPort(spec.Port).WithNumaNode(spec.NumaNode).AppendEndpoints(topo.Endpoints(deployUser)...)
+	).WithPort(spec.Port).WithNumaNode(spec.NumaNode).AppendEndpoints(topo.Endpoints(deployUser)...).WithCDCClusterID(spec.TiCDCClusterID)
 
 	// doesn't work
 	if _, err := i.setTLSConfig(ctx, false, nil, paths); err != nil {
diff --git a/pkg/cluster/template/scripts/cdc.go b/pkg/cluster/template/scripts/cdc.go
index 9775323544..48b7082d30 100644
--- a/pkg/cluster/template/scripts/cdc.go
+++ b/pkg/cluster/template/scripts/cdc.go
@@ -33,6 +33,7 @@ type CDCScript struct {
 	NumaNode          string
 	GCTTL             int64
 	TZ                string
+	ClusterID         string
 	TLSEnabled        bool
 	Endpoints         []*PDScript
 	ConfigFileEnabled bool
@@ -82,6 +83,12 @@ func (c *CDCScript) WithDataDirEnabled() *CDCScript {
 	return c
 }
 
+// WithCDCClusterID set CDC cluster-id
+func (c *CDCScript) WithCDCClusterID(clusterID string) *CDCScript {
+	c.ClusterID = clusterID
+	return c
+}
+
 // Config generate the config file data.
 func (c *CDCScript) Config() ([]byte, error) {
 	fp := path.Join("templates", "scripts", "run_cdc.sh.tpl")
diff --git a/pkg/tidbver/tidbver.go b/pkg/tidbver/tidbver.go
index a56576aaa3..6428cc96a1 100644
--- a/pkg/tidbver/tidbver.go
+++ b/pkg/tidbver/tidbver.go
@@ -76,6 +76,11 @@ func TiCDCSupportDataDir(version string) bool {
 	return semver.Major(version) == "v4" && semver.Compare(version, "v4.0.14") >= 0
 }
 
+// TiCDCSupportClusterID return if the given version of TiCDC support --cluster-id param to identify TiCDC cluster
+func TiCDCSupportClusterID(version string) bool {
+	return semver.Compare(version, "v6.2.0") >= 0 || strings.Contains(version, "nightly")
+}
+
 // TiCDCSupportRollingUpgrade return if the given version of TiCDC support rolling upgrade
 // TiCDC support graceful rolling upgrade since v6.3.0
 func TiCDCSupportRollingUpgrade(version string) bool {