playground: Add tidb-cse mode (#2386)

components/playground/instance/instance.go

@@ -37,6 +37,14 @@ type Config struct {
 	Version    string `yaml:"version"`
 }
 
+// CSEOptions contains configs to run TiDB cluster in CSE mode.
+type CSEOptions struct {
+	S3Endpoint string `yaml:"s3_endpoint"`
+	Bucket     string `yaml:"bucket"`
+	AccessKey  string `yaml:"access_key"`
+	SecretKey  string `yaml:"secret_key"`
+}
+
 type instance struct {
 	ID         int
 	Dir        string

components/playground/instance/pd.go

@@ -46,10 +46,11 @@ type PDInstance struct {
 	joinEndpoints []*PDInstance
 	pds           []*PDInstance
 	Process
+	isCSEMode bool
 }
 
 // NewPDInstance return a PDInstance
-func NewPDInstance(role PDRole, binPath, dir, host, configPath string, id int, pds []*PDInstance, port int) *PDInstance {
+func NewPDInstance(role PDRole, binPath, dir, host, configPath string, id int, pds []*PDInstance, port int, isCSEMode bool) *PDInstance {
 	if port <= 0 {
 		port = 2379
 	}
@@ -63,8 +64,9 @@ func NewPDInstance(role PDRole, binPath, dir, host, configPath string, id int, p
 			StatusPort: utils.MustGetFreePort(host, port),
 			ConfigPath: configPath,
 		},
-		Role: role,
-		pds:  pds,
+		Role:      role,
+		pds:       pds,
+		isCSEMode: isCSEMode,
 	}
 }
 
@@ -112,8 +114,8 @@ func (inst *PDInstance) Start(ctx context.Context, version utils.Version) error
 			fmt.Sprintf("--client-urls=http://%s", utils.JoinHostPort(inst.Host, inst.StatusPort)),
 			fmt.Sprintf("--advertise-client-urls=http://%s", utils.JoinHostPort(AdvertiseHost(inst.Host), inst.StatusPort)),
 			fmt.Sprintf("--log-file=%s", inst.LogFile()),
+			fmt.Sprintf("--config=%s", configPath),
 		}...)
-
 		switch {
 		case len(inst.initEndpoints) > 0:
 			endpoints := make([]string, 0)
@@ -140,9 +142,7 @@ func (inst *PDInstance) Start(ctx context.Context, version utils.Version) error
 			fmt.Sprintf("--advertise-listen-addr=http://%s", utils.JoinHostPort(AdvertiseHost(inst.Host), inst.StatusPort)),
 			fmt.Sprintf("--backend-endpoints=%s", strings.Join(endpoints, ",")),
 			fmt.Sprintf("--log-file=%s", inst.LogFile()),
-		}
-		if inst.ConfigPath != "" {
-			args = append(args, fmt.Sprintf("--config=%s", inst.ConfigPath))
+			fmt.Sprintf("--config=%s", configPath),
 		}
 	case PDRoleScheduling:
 		endpoints := pdEndpoints(inst.pds, true)
@@ -153,9 +153,7 @@ func (inst *PDInstance) Start(ctx context.Context, version utils.Version) error
 			fmt.Sprintf("--advertise-listen-addr=http://%s", utils.JoinHostPort(AdvertiseHost(inst.Host), inst.StatusPort)),
 			fmt.Sprintf("--backend-endpoints=%s", strings.Join(endpoints, ",")),
 			fmt.Sprintf("--log-file=%s", inst.LogFile()),
-		}
-		if inst.ConfigPath != "" {
-			args = append(args, fmt.Sprintf("--config=%s", inst.ConfigPath))
+			fmt.Sprintf("--config=%s", configPath),
 		}
 	}
 

components/playground/instance/pd_config.go

@@ -16,5 +16,15 @@ package instance
 func (inst *PDInstance) getConfig() map[string]any {
 	config := make(map[string]any)
 	config["schedule.patrol-region-interval"] = "100ms"
+
+	if inst.isCSEMode {
+		config["keyspace.pre-alloc"] = []string{"mykeyspace"}
+		config["replication.enable-placement-rules"] = true
+		config["replication.max-replica"] = 1
+		config["schedule.merge-schedule-limit"] = 0
+		config["schedule.low-space-ration"] = 1.0
+		config["schedule.replica-schedule-limit"] = 500
+	}
+
 	return config
 }

components/playground/instance/tidb.go

@@ -31,11 +31,11 @@ type TiDBInstance struct {
 	Process
 	tiproxyCertDir string
 	enableBinlog   bool
-	isDisaggMode   bool
+	isCSEMode      bool
 }
 
 // NewTiDBInstance return a TiDBInstance
-func NewTiDBInstance(binPath string, dir, host, configPath string, id, port int, pds []*PDInstance, tiproxyCertDir string, enableBinlog bool, isDisaggMode bool) *TiDBInstance {
+func NewTiDBInstance(binPath string, dir, host, configPath string, id, port int, pds []*PDInstance, tiproxyCertDir string, enableBinlog bool, isCSEMode bool) *TiDBInstance {
 	if port <= 0 {
 		port = 4000
 	}
@@ -52,7 +52,7 @@ func NewTiDBInstance(binPath string, dir, host, configPath string, id, port int,
 		tiproxyCertDir: tiproxyCertDir,
 		pds:            pds,
 		enableBinlog:   enableBinlog,
-		isDisaggMode:   isDisaggMode,
+		isCSEMode:      isCSEMode,
 	}
 }
 

components/playground/instance/tidb_config.go

@@ -22,9 +22,36 @@ func (inst *TiDBInstance) getConfig() map[string]any {
 	config := make(map[string]any)
 	config["security.auto-tls"] = true
 
-	if inst.isDisaggMode {
+	if inst.isCSEMode {
+		config["keyspace-name"] = "mykeyspace"
+		config["enable-safe-point-v2"] = true
+		config["force-enable-vector-type"] = true
 		config["use-autoscaler"] = false
 		config["disaggregated-tiflash"] = true
+		config["ratelimit.full-speed"] = 1048576000
+		config["ratelimit.full-speed-capacity"] = 1048576000
+		config["ratelimit.low-speed-watermark"] = 1048576000000
+		config["ratelimit.block-write-watermark"] = 1048576000000
+		config["security.enable-sem"] = false
+		config["tiflash-replicas.constraints"] = []any{
+			map[string]any{
+				"key": "engine",
+				"op":  "in",
+				"values": []string{
+					"tiflash",
+				},
+			},
+			map[string]any{
+				"key": "engine_role",
+				"op":  "in",
+				"values": []string{
+					"write",
+				},
+			},
+		}
+		config["tiflash-replicas.group-id"] = "enable_s3_wn_region"
+		config["tiflash-replicas.extra-s3-rule"] = false
+		config["tiflash-replicas.min-count"] = 1
 	}
 
 	tiproxyCrtPath := filepath.Join(inst.tiproxyCertDir, "tiproxy.crt")

components/playground/instance/tiflash.go

@@ -39,19 +39,11 @@ const (
 	TiFlashRoleDisaggCompute TiFlashRole = "compute"
 )
 
-// DisaggOptions contains configs to run TiFlash in disaggregated mode.
-type DisaggOptions struct {
-	S3Endpoint string `yaml:"s3_endpoint"`
-	Bucket     string `yaml:"bucket"`
-	AccessKey  string `yaml:"access_key"`
-	SecretKey  string `yaml:"secret_key"`
-}
-
 // TiFlashInstance represent a running TiFlash
 type TiFlashInstance struct {
 	instance
 	Role            TiFlashRole
-	DisaggOpts      DisaggOptions
+	cseOpts         CSEOptions
 	TCPPort         int
 	ServicePort     int
 	ProxyPort       int
@@ -62,7 +54,7 @@ type TiFlashInstance struct {
 }
 
 // NewTiFlashInstance return a TiFlashInstance
-func NewTiFlashInstance(role TiFlashRole, disaggOptions DisaggOptions, binPath, dir, host, configPath string, id int, pds []*PDInstance, dbs []*TiDBInstance, version string) *TiFlashInstance {
+func NewTiFlashInstance(role TiFlashRole, cseOptions CSEOptions, binPath, dir, host, configPath string, id int, pds []*PDInstance, dbs []*TiDBInstance, version string) *TiFlashInstance {
 	if role != TiFlashRoleNormal && role != TiFlashRoleDisaggWrite && role != TiFlashRoleDisaggCompute {
 		panic(fmt.Sprintf("Unknown TiFlash role %s", role))
 	}
@@ -82,7 +74,7 @@ func NewTiFlashInstance(role TiFlashRole, disaggOptions DisaggOptions, binPath,
 			ConfigPath: configPath,
 		},
 		Role:            role,
-		DisaggOpts:      disaggOptions,
+		cseOpts:         cseOptions,
 		TCPPort:         utils.MustGetFreePort(host, 9100), // 9000 for default object store port
 		ServicePort:     utils.MustGetFreePort(host, 3930),
 		ProxyPort:       utils.MustGetFreePort(host, 20170),

components/playground/instance/tiflash_config.go

@@ -21,6 +21,18 @@ func (inst *TiFlashInstance) getProxyConfig() map[string]any {
 	config["raftdb.max-open-files"] = 256
 	config["storage.reserve-space"] = 0
 	config["storage.reserve-raft-space"] = 0
+
+	if inst.Role == TiFlashRoleDisaggWrite {
+		config["storage.api-version"] = 2
+		config["storage.enable-ttl"] = true
+		config["dfs.prefix"] = "tikv"
+		config["dfs.s3-endpoint"] = inst.cseOpts.S3Endpoint
+		config["dfs.s3-key-id"] = inst.cseOpts.AccessKey
+		config["dfs.s3-secret-key"] = inst.cseOpts.SecretKey
+		config["dfs.s3-bucket"] = inst.cseOpts.Bucket
+		config["dfs.s3-region"] = "local"
+	}
+
 	return config
 }
 
@@ -31,23 +43,26 @@ func (inst *TiFlashInstance) getConfig() map[string]any {
 	config["logger.level"] = "debug"
 
 	if inst.Role == TiFlashRoleDisaggWrite {
-		config["storage.s3.endpoint"] = inst.DisaggOpts.S3Endpoint
-		config["storage.s3.bucket"] = inst.DisaggOpts.Bucket
-		config["storage.s3.root"] = "/"
-		config["storage.s3.access_key_id"] = inst.DisaggOpts.AccessKey
-		config["storage.s3.secret_access_key"] = inst.DisaggOpts.SecretKey
+		config["enable_safe_point_v2"] = true
+		config["storage.api_version"] = 2
+		config["storage.s3.endpoint"] = inst.cseOpts.S3Endpoint
+		config["storage.s3.bucket"] = inst.cseOpts.Bucket
+		config["storage.s3.root"] = "/tiflash-cse/"
+		config["storage.s3.access_key_id"] = inst.cseOpts.AccessKey
+		config["storage.s3.secret_access_key"] = inst.cseOpts.SecretKey
+		config["storage.main.dir"] = []string{filepath.Join(inst.Dir, "main_data")}
 		config["flash.disaggregated_mode"] = "tiflash_write"
-		config["flash.use_autoscaler"] = false
 	} else if inst.Role == TiFlashRoleDisaggCompute {
-		config["storage.s3.endpoint"] = inst.DisaggOpts.S3Endpoint
-		config["storage.s3.bucket"] = inst.DisaggOpts.Bucket
-		config["storage.s3.root"] = "/"
-		config["storage.s3.access_key_id"] = inst.DisaggOpts.AccessKey
-		config["storage.s3.secret_access_key"] = inst.DisaggOpts.SecretKey
+		config["enable_safe_point_v2"] = true
+		config["storage.s3.endpoint"] = inst.cseOpts.S3Endpoint
+		config["storage.s3.bucket"] = inst.cseOpts.Bucket
+		config["storage.s3.root"] = "/tiflash-cse/"
+		config["storage.s3.access_key_id"] = inst.cseOpts.AccessKey
+		config["storage.s3.secret_access_key"] = inst.cseOpts.SecretKey
 		config["storage.remote.cache.dir"] = filepath.Join(inst.Dir, "remote_cache")
-		config["storage.remote.cache.capacity"] = 1000000000 // 1GB
+		config["storage.remote.cache.capacity"] = 20000000000 // 20GB
+		config["storage.main.dir"] = []string{filepath.Join(inst.Dir, "main_data")}
 		config["flash.disaggregated_mode"] = "tiflash_compute"
-		config["flash.use_autoscaler"] = false
 	}
 
 	return config

components/playground/instance/tikv.go

@@ -28,10 +28,12 @@ type TiKVInstance struct {
 	instance
 	pds []*PDInstance
 	Process
+	isCSEMode bool
+	cseOpts   CSEOptions
 }
 
 // NewTiKVInstance return a TiKVInstance
-func NewTiKVInstance(binPath string, dir, host, configPath string, id int, port int, pds []*PDInstance) *TiKVInstance {
+func NewTiKVInstance(binPath string, dir, host, configPath string, id int, port int, pds []*PDInstance, isCSEMode bool, cseOptions CSEOptions) *TiKVInstance {
 	if port <= 0 {
 		port = 20160
 	}
@@ -45,7 +47,9 @@ func NewTiKVInstance(binPath string, dir, host, configPath string, id int, port
 			StatusPort: utils.MustGetFreePort(host, 20180),
 			ConfigPath: configPath,
 		},
-		pds: pds,
+		pds:       pds,
+		isCSEMode: isCSEMode,
+		cseOpts:   cseOptions,
 	}
 }
 

components/playground/instance/tikv_config.go

@@ -19,5 +19,17 @@ func (inst *TiKVInstance) getConfig() map[string]any {
 	config["raftdb.max-open-files"] = 256
 	config["storage.reserve-space"] = 0
 	config["storage.reserve-raft-space"] = 0
+
+	if inst.isCSEMode {
+		config["storage.api-version"] = 2
+		config["storage.enable-ttl"] = true
+		config["dfs.prefix"] = "tikv"
+		config["dfs.s3-endpoint"] = inst.cseOpts.S3Endpoint
+		config["dfs.s3-key-id"] = inst.cseOpts.AccessKey
+		config["dfs.s3-secret-key"] = inst.cseOpts.SecretKey
+		config["dfs.s3-bucket"] = inst.cseOpts.Bucket
+		config["dfs.s3-region"] = "local"
+	}
+
 	return config
 }