binlog: support binlog.pump.config configurations for pump and drainer

charts/tidb-cluster/templates/_helpers.tpl

@@ -79,3 +79,34 @@ config-file: |-
 {{ include "tidb-configmap.data" . | sha256sum | trunc 8 }}
 {{- end -}}
 
+{{/*
+Encapsulate pump configmap data for consistent digest calculation
+*/}}
+{{- define "pump-configmap.data" -}}
+pump-config: |-
+    {{- if .Values.binlog.pump.config }}
+{{ .Values.binlog.pump.config | indent 2 }}
+    {{- else -}}
+{{ tuple "config/_pump-config.tpl" . | include "helm-toolkit.utils.template" | indent 2 }}
+    {{- end -}}
+{{- end -}}
+
+{{- define "pump-configmap.data-digest" -}}
+{{ include "pump-configmap.data" . | sha256sum | trunc 8 }}
+{{- end -}}
+
+{{/*
+Encapsulate drainer configmap data for consistent digest calculation
+*/}}
+{{- define "drainer-configmap.data" -}}
+drainer-config: |-
+    {{- if .Values.binlog.drainer.config }}
+{{ .Values.binlog.drainer.config | indent 2 }}
+    {{- else -}}
+{{ tuple "config/_drainer-config.tpl" . | include "helm-toolkit.utils.template" | indent 2 }}
+    {{- end -}}
+{{- end -}}
+
+{{- define "drainer-configmap.data-digest" -}}
+{{ include "drainer-configmap.data" . | sha256sum | trunc 8 }}
+{{- end -}}

charts/tidb-cluster/templates/drainer-configmap-rollout.yaml

@@ -0,0 +1,16 @@
+{{- if .Values.enableConfigMapRollout }}
+{{- if .Values.binlog.drainer.create }}
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: {{ template "cluster.name" . }}-drainer-{{ template "drainer-configmap.data-digest" . }}
+  labels:
+    app.kubernetes.io/name: {{ template "chart.name" . }}
+    app.kubernetes.io/managed-by: {{ .Release.Service }}
+    app.kubernetes.io/instance: {{ .Release.Name }}
+    app.kubernetes.io/component: drainer
+    helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+"  "_" }}
+data:
+{{ include "drainer-configmap.data" . | indent 2 }}
+{{- end -}}
+{{- end -}}

charts/tidb-cluster/templates/drainer-configmap.yaml

@@ -10,6 +10,5 @@ metadata:
     app.kubernetes.io/component: drainer
     helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+"  "_" }}
 data:
-  drainer-config: |-
-{{ tuple "config/_drainer-config.tpl" . | include "helm-toolkit.utils.template" | indent 4 }}
+{{ include "drainer-configmap.data" . | indent 2 }}
 {{- end -}}

charts/tidb-cluster/templates/drainer-statefulset.yaml

@@ -58,7 +58,11 @@ spec:
       volumes:
       - name: config
         configMap:
+        {{- if .Values.enableConfigMapRollout }}
+          name: {{ template "cluster.name" . }}-drainer-{{ template "drainer-configmap.data-digest" . }}
+        {{- else }}
           name: {{ template "cluster.name" . }}-drainer
+        {{- end }}
           items:
           - key: drainer-config
             path: drainer.toml

charts/tidb-cluster/templates/pump-configmap-rollout.yaml

@@ -0,0 +1,16 @@
+{{- if .Values.enableConfigMapRollout }}
+{{- if .Values.binlog.pump.create }}
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: {{ template "cluster.name" . }}-pump-{{ template "pump-configmap.data-digest" . }}
+  labels:
+    app.kubernetes.io/name: {{ template "chart.name" . }}
+    app.kubernetes.io/managed-by: {{ .Release.Service }}
+    app.kubernetes.io/instance: {{ .Release.Name }}
+    app.kubernetes.io/component: pump
+    helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+"  "_" }}
+data:
+{{ include "pump-configmap.data" . | indent 2 }}
+{{- end }}
+{{- end }}

charts/tidb-cluster/templates/pump-configmap.yaml

@@ -10,6 +10,5 @@ metadata:
     app.kubernetes.io/component: pump
     helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+"  "_" }}
 data:
-  pump-config: |-
-{{ tuple "config/_pump-config.tpl" . | include "helm-toolkit.utils.template" | indent 4 }}
-{{- end -}}
+{{ include "pump-configmap.data" . | indent 2 }}
+{{ end }}

charts/tidb-cluster/templates/pump-statefulset.yaml

@@ -58,7 +58,11 @@ spec:
       volumes:
       - name: config
         configMap:
+        {{- if .Values.enableConfigMapRollout }}
+          name: {{ template "cluster.name" . }}-pump-{{ template "pump-configmap.data-digest" . }}
+        {{- else }}
           name: {{ template "cluster.name" . }}-pump
+        {{- end }}
           items:
           - key: pump-config
             path: pump.toml

charts/tidb-cluster/templates/scripts/_start_drainer.sh.tpl

@@ -26,6 +26,7 @@ done
 
 /drainer \
 -L={{ .Values.binlog.drainer.logLevel | default "info" }} \
+-pd-urls=http://{{ template "cluster.name" . }}-pd:2379 \
 -addr=`echo ${HOSTNAME}`.{{ template "cluster.name" . }}-drainer:8249 \
 -config=/etc/drainer/drainer.toml \
 -disable-detect={{ .Values.binlog.drainer.disableDetect | default false }} \

charts/tidb-cluster/templates/scripts/_start_pump.sh.tpl

@@ -1,5 +1,6 @@
 set -euo pipefail
 /pump \
+-pd-urls=http://{{ template "cluster.name" . }}-pd:2379 \
 -L={{ .Values.binlog.pump.logLevel | default "info" }} \
 -advertise-addr=`echo ${HOSTNAME}`.{{ template "cluster.name" . }}-pump:8250 \
 -config=/etc/pump/pump.toml \

charts/tidb-cluster/values.yaml

@@ -423,6 +423,17 @@ binlog:
     # number of seconds between heartbeat ticks (in 2 seconds)
     heartbeatInterval: 2
 
+    # Please refer to https://github.com/pingcap/tidb-binlog/blob/master/cmd/pump/pump.toml for the default
+    # pump configurations (change to the tags of your pump version),
+    # just follow the format in the file and configure in the 'config' section
+    # as below if you want to customize any configuration.
+    # config: |
+    #   gc = 7
+    #   heartbeat-interval = 2
+    #   [storage]
+    #   sync-log = true
+    #   stop-write-at-available-space = "10Gi"
+
   drainer:
     create: false
     image: pingcap/tidb-binlog:v3.0.1
@@ -474,6 +485,24 @@ binlog:
       # kafkaAddrs: "127.0.0.1:9092"
       # kafkaVersion: "0.8.2.0"
 
+    # Please refer to https://github.com/pingcap/tidb-binlog/blob/master/cmd/drainer/drainer.toml for the default
+    # drainer configurations (change to the tags of your drainer version),
+    # just follow the format in the file and configure in the 'config' section
+    # as below if you want to customize any configuration.
+    # config: |
+    #   worker-count = 16
+    #   detect-interval = 10
+    #   disable-dispatch = false
+    #   ignore-schemas = "INFORMATION_SCHEMA,PERFORMANCE_SCHEMA,mysql"
+    #   safe-mode = false
+    #   txn-batch = 20
+    #   db-type = "mysql"
+    #   [syncer.to]
+    #   # host = "127.0.0.1"
+    #   # user = "root"
+    #   # password = ""
+    #   # port = 3306
+
 scheduledBackup:
   create: false
   # https://github.com/pingcap/tidb-cloud-backup

tests/actions.go

@@ -263,6 +263,9 @@ type TidbClusterConfig struct {
 	BlockWriteConfig blockwriter.Config
 	GrafanaClient    *metrics.Client
 	TopologyKey      string
+
+	pumpConfig    []string
+	drainerConfig []string
 }
 
 func (tc *TidbClusterConfig) String() string {
@@ -2289,13 +2292,7 @@ func (oa *operatorActions) DeployIncrementalBackup(from *TidbClusterConfig, to *
 	glog.Infof("begin to deploy incremental backup cluster[%s] namespace[%s]", from.ClusterName, from.Namespace)
 
 	sets := map[string]string{
-		"binlog.pump.create":            "true",
-		"binlog.drainer.destDBType":     "mysql",
-		"binlog.drainer.mysql.host":     fmt.Sprintf("%s-tidb.%s", to.ClusterName, to.Namespace),
-		"binlog.drainer.mysql.user":     "root",
-		"binlog.drainer.mysql.password": to.Password,
-		"binlog.drainer.mysql.port":     "4000",
-		"binlog.drainer.ignoreSchemas":  "",
+		"binlog.pump.create": "true",
 	}
 	if withDrainer {
 		sets["binlog.drainer.create"] = "true"
@@ -2304,14 +2301,29 @@ func (oa *operatorActions) DeployIncrementalBackup(from *TidbClusterConfig, to *
 		sets["binlog.drainer.initialCommitTs"] = ts
 	}
 
+	from.drainerConfig = []string{
+		"worker-count = 16",
+		"detect-interval = 10",
+		"disable-dispatch = false",
+		`ignore-schemas = ""`,
+		`safe-mode = false`,
+		`txn-batch = 20`,
+		`db-type = "mysql"`,
+		`[syncer.to]`,
+		fmt.Sprintf(`host = "%s-tidb.%s"`, to.ClusterName, to.Namespace),
+		fmt.Sprintf(`user = "%s"`, "root"),
+		fmt.Sprintf(`password = "%s"`, to.Password),
+		fmt.Sprintf(`port = %d`, 4000),
+	}
+
 	cmd, err := oa.getHelmUpgradeClusterCmd(from, sets)
 	if err != nil {
 		return err
 	}
 	glog.Infof(cmd)
 	res, err := exec.Command("/bin/sh", "-c", cmd).CombinedOutput()
 	if err != nil {
-		return fmt.Errorf("failed to launch scheduler backup job: %v, %s", err, string(res))
+		return fmt.Errorf("failed to launch incremental backup job: %v, %s", err, string(res))
 	}
 	return nil
 }
@@ -2439,7 +2451,7 @@ func (oa *operatorActions) CheckIncrementalBackup(info *TidbClusterConfig, withD
 
 	err := wait.Poll(oa.pollInterval, DefaultPollTimeout, fn)
 	if err != nil {
-		return fmt.Errorf("failed to launch scheduler backup job: %v", err)
+		return fmt.Errorf("failed to check incremental backup job: %v", err)
 	}
 	return nil
 

tests/cluster_info.go

@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"os"
 	"strconv"
+
+	"github.com/golang/glog"
 )
 
 func (tc *TidbClusterConfig) set(name string, value string) (string, bool) {
@@ -113,7 +115,7 @@ func (tc *TidbClusterConfig) BuildSubValues(path string) (string, error) {
 		"[log]",
 		`level = "info"`,
 	}
-	subValues := GetSubValuesOrDie(tc.ClusterName, tc.Namespace, tc.TopologyKey, pdConfig, tikvConfig, tidbConfig)
+	subValues := GetSubValuesOrDie(tc.ClusterName, tc.Namespace, tc.TopologyKey, pdConfig, tikvConfig, tidbConfig, tc.pumpConfig, tc.drainerConfig)
 	subVaulesPath := fmt.Sprintf("%s/%s.yaml", path, tc.ClusterName)
 	_, err := os.Stat(subVaulesPath)
 	if err != nil {
@@ -134,5 +136,6 @@ func (tc *TidbClusterConfig) BuildSubValues(path string) (string, error) {
 	if err != nil {
 		return "", err
 	}
+	glog.Infof("subValues:\n %s", subValues)
 	return subVaulesPath, nil
 }

tests/util.go

@@ -104,7 +104,9 @@ var affinityTemp string = `{{.Kind}}:
           topologyKey: {{.TopologyKey}}
           namespaces:
           - {{.Namespace}}
-binlog:
+`
+
+var binlogTemp string = `binlog:
   pump:
     tolerations:
     - key: node-role
@@ -119,6 +121,10 @@ binlog:
             topologyKey: {{.TopologyKey}}
             namespaces:
             - {{.Namespace}}
+{{if .PumpConfig}}
+  	config: |
+{{range .PumpConfig}}      {{.}}
+{{end}}{{end}}
   drainer:
     tolerations:
     - key: node-role
@@ -133,6 +139,10 @@ binlog:
             topologyKey: {{.TopologyKey}}
             namespaces:
             - {{.Namespace}}
+{{if .DrainerConfig}}
+    config: |
+{{range .DrainerConfig}}      {{.}}
+{{end}}{{end}}
 `
 
 type AffinityInfo struct {
@@ -144,7 +154,14 @@ type AffinityInfo struct {
 	Config      []string
 }
 
-func GetSubValuesOrDie(clusterName, namespace, topologyKey string, pdConfig []string, tikvConfig []string, tidbConfig []string) string {
+type BinLogInfo struct {
+	PumpConfig    []string
+	DrainerConfig []string
+	Namespace     string
+	TopologyKey   string
+}
+
+func GetSubValuesOrDie(clusterName, namespace, topologyKey string, pdConfig []string, tikvConfig []string, tidbConfig []string, pumpConfig []string, drainerConfig []string) string {
 	temp, err := template.New("dt-affinity").Parse(affinityTemp)
 	if err != nil {
 		slack.NotifyAndPanic(err)
@@ -165,7 +182,23 @@ func GetSubValuesOrDie(clusterName, namespace, topologyKey string, pdConfig []st
 	if err != nil {
 		slack.NotifyAndPanic(err)
 	}
-	return fmt.Sprintf("%s%s%s", pdbuff.String(), tikvbuff.String(), tidbbuff.String())
+	subValues := fmt.Sprintf("%s%s%s", pdbuff.String(), tikvbuff.String(), tidbbuff.String())
+
+	if pumpConfig == nil && drainerConfig == nil {
+		return subValues
+	}
+
+	btemp, err := template.New("binlog").Parse(binlogTemp)
+	if err != nil {
+		slack.NotifyAndPanic(err)
+	}
+	binlogbuff := new(bytes.Buffer)
+	err = btemp.Execute(binlogbuff, &BinLogInfo{PumpConfig: pumpConfig, DrainerConfig: drainerConfig, Namespace: namespace, TopologyKey: topologyKey})
+	if err != nil {
+		slack.NotifyAndPanic(err)
+	}
+	subValues = fmt.Sprintf("%s%s", subValues, binlogbuff.String())
+	return subValues
 }
 
 const (