Skip to content

Commit 9927d60

Browse files
shiyuhang0shiyuhang0
committed
add mysql and fix all filed
1 parent 9b2069d commit 9927d60

49 files changed

Lines changed: 3850 additions & 3071 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎internal/cli/serverless/changefeed/create.go‎

Lines changed: 65 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,7 @@ package changefeed
1717
import (
1818
"encoding/json"
1919
"fmt"
20-
"slices"
2120
"strconv"
22-
"time"
2321

2422
"github.com/aws/aws-sdk-go-v2/aws"
2523
"github.com/fatih/color"
@@ -43,6 +41,7 @@ func (c CreateOpts) NonInteractiveFlags() []string {
4341
flag.ClusterID,
4442
flag.ChangefeedType,
4543
flag.ChangefeedKafka,
44+
flag.ChangefeedMySQL,
4645
flag.ChangefeedFilter,
4746
flag.ChangefeedStartTSO,
4847
flag.ChangefeedName,
@@ -53,7 +52,6 @@ func (c CreateOpts) RequiredFlags() []string {
5352
return []string{
5453
flag.ClusterID,
5554
flag.ChangefeedType,
56-
flag.ChangefeedKafka,
5755
flag.ChangefeedFilter,
5856
}
5957
}
@@ -105,10 +103,12 @@ func CreateCmd(h *internal.Helper) *cobra.Command {
105103
}
106104
ctx := cmd.Context()
107105

108-
var clusterID, name, changefeedType, kafkaStr, filterStr string
106+
var clusterID, name, kafkaStr, mysqlStr, filterStr string
109107
var startTSO uint64
110-
var kafkaInfo cdc.KafkaInfo
111108
var filter cdc.CDCFilter
109+
var kafkaInfo cdc.Kafka
110+
var mysqlInfo cdc.MySQL
111+
var changefeedType cdc.ChangefeedTypeEnum
112112

113113
if opts.interactive {
114114
if !h.IOStreams.CanPrompt {
@@ -123,17 +123,36 @@ func CreateCmd(h *internal.Helper) *cobra.Command {
123123
return err
124124
}
125125
clusterID = cluster.ID
126-
127-
inputs := []string{flag.ChangefeedName, flag.ChangefeedType, flag.ChangefeedKafka, flag.ChangefeedFilter, flag.ChangefeedStartTSO}
128-
textInput, err := ui.InitialInputModel(inputs, inputDescription)
126+
changefeedType, err = GetSelectedChangefeedType()
129127
if err != nil {
130128
return err
131129
}
132-
name = textInput.Inputs[0].Value()
133-
changefeedType = textInput.Inputs[1].Value()
134-
kafkaStr = textInput.Inputs[2].Value()
135-
filterStr = textInput.Inputs[3].Value()
136-
startTSOStr := textInput.Inputs[4].Value()
130+
131+
var startTSOStr string
132+
switch changefeedType {
133+
case cdc.CHANGEFEEDTYPEENUM_KAFKA:
134+
inputs := []string{flag.ChangefeedName, flag.ChangefeedKafka, flag.ChangefeedFilter, flag.ChangefeedStartTSO}
135+
textInput, err := ui.InitialInputModel(inputs, createKafkaInputDescription)
136+
if err != nil {
137+
return err
138+
}
139+
name = textInput.Inputs[0].Value()
140+
kafkaStr = textInput.Inputs[1].Value()
141+
filterStr = textInput.Inputs[2].Value()
142+
startTSOStr = textInput.Inputs[3].Value()
143+
case cdc.CHANGEFEEDTYPEENUM_MYSQL:
144+
inputs := []string{flag.ChangefeedName, flag.ChangefeedMySQL, flag.ChangefeedFilter, flag.ChangefeedStartTSO}
145+
textInput, err := ui.InitialInputModel(inputs, createKafkaInputDescription)
146+
if err != nil {
147+
return err
148+
}
149+
name = textInput.Inputs[0].Value()
150+
mysqlStr = textInput.Inputs[1].Value()
151+
filterStr = textInput.Inputs[2].Value()
152+
startTSOStr = textInput.Inputs[3].Value()
153+
default:
154+
return errors.Errorf("currently only %s and %s type is supported", cdc.CHANGEFEEDTYPEENUM_KAFKA, cdc.CHANGEFEEDTYPEENUM_MYSQL)
155+
}
137156
if startTSOStr == "" {
138157
startTSO = 0
139158
} else {
@@ -152,72 +171,70 @@ func CreateCmd(h *internal.Helper) *cobra.Command {
152171
if err != nil {
153172
return errors.Trace(err)
154173
}
155-
changefeedType, err = cmd.Flags().GetString(flag.ChangefeedType)
174+
changefeedTypeStr, err := cmd.Flags().GetString(flag.ChangefeedType)
156175
if err != nil {
157176
return errors.Trace(err)
158177
}
159-
if changefeedType == "" {
160-
return errors.New("type is required")
161-
}
178+
changefeedType = cdc.ChangefeedTypeEnum(changefeedTypeStr)
162179
kafkaStr, err = cmd.Flags().GetString(flag.ChangefeedKafka)
163180
if err != nil {
164181
return errors.Trace(err)
165182
}
166-
if kafkaStr == "" {
167-
return errors.New("kafka info is required")
168-
}
169-
if err := json.Unmarshal([]byte(kafkaStr), &kafkaInfo); err != nil {
170-
return errors.New("invalid kafka info, please use JSON format")
183+
mysqlStr, err = cmd.Flags().GetString(flag.ChangefeedMySQL)
184+
if err != nil {
185+
return errors.Trace(err)
171186
}
172187
filterStr, err = cmd.Flags().GetString(flag.ChangefeedFilter)
173188
if err != nil {
174189
return errors.Trace(err)
175190
}
176-
if filterStr == "" {
177-
return errors.New("filter is required")
178-
}
179-
if err := json.Unmarshal([]byte(filterStr), &filter); err != nil {
180-
return errors.New("invalid filter, please use JSON format")
181-
}
182191
startTSO, err = cmd.Flags().GetUint64(flag.ChangefeedStartTSO)
183192
if err != nil {
184193
return errors.Trace(err)
185194
}
186195
}
187196

188197
// check all the parameters
189-
if changefeedType == "" {
190-
return errors.New("type is required")
191-
}
192-
if kafkaStr == "" {
193-
return errors.New("kafka info is required")
194-
}
195-
if err := json.Unmarshal([]byte(kafkaStr), &kafkaInfo); err != nil {
196-
return errors.New("invalid kafka info, please use JSON format")
198+
switch changefeedType {
199+
case cdc.CHANGEFEEDTYPEENUM_KAFKA:
200+
if kafkaStr == "" {
201+
return errors.New("kafka info is required")
202+
}
203+
if err := json.Unmarshal([]byte(kafkaStr), &kafkaInfo); err != nil {
204+
return errors.New("invalid kafka info, please use JSON format")
205+
}
206+
case cdc.CHANGEFEEDTYPEENUM_MYSQL:
207+
if mysqlStr == "" {
208+
return errors.New("mysql info is required")
209+
}
210+
if err := json.Unmarshal([]byte(mysqlStr), &mysqlInfo); err != nil {
211+
return errors.New("invalid mysql info, please use JSON format")
212+
}
213+
default:
214+
return errors.New("currently only KAFKA and MYSQL type is supported")
197215
}
216+
198217
if filterStr == "" {
199218
return errors.New("filter is required")
200219
}
201220
if err := json.Unmarshal([]byte(filterStr), &filter); err != nil {
202221
return errors.New("invalid filter, please use JSON format")
203222
}
204223

205-
if !slices.Contains(cdc.AllowedConnectorTypeEnumEnumValues, cdc.ConnectorTypeEnum(changefeedType)) {
206-
return errors.New("currently only KAFKA type is supported")
207-
}
208-
209224
// create the changefeed
210-
body := &cdc.ConnectorServiceCreateConnectorBody{
211-
Name: &name,
225+
body := &cdc.ChangefeedServiceCreateChangefeedBody{
226+
DisplayName: &name,
212227
Sink: cdc.SinkInfo{
213-
Type: cdc.ConnectorTypeEnum(changefeedType),
228+
Type: cdc.ChangefeedTypeEnum(changefeedType),
214229
},
215230
Filter: filter,
216231
}
217232

218233
switch body.Sink.Type {
219-
case cdc.CONNECTORTYPEENUM_KAFKA:
234+
case cdc.CHANGEFEEDTYPEENUM_KAFKA:
220235
body.Sink.Kafka = &kafkaInfo
236+
case cdc.CHANGEFEEDTYPEENUM_MYSQL:
237+
body.Sink.Mysql = &mysqlInfo
221238
}
222239
if startTSO == 0 {
223240
mode := cdc.STARTMODEENUM_FROM_NOW
@@ -232,18 +249,11 @@ func CreateCmd(h *internal.Helper) *cobra.Command {
232249
}
233250
}
234251

235-
now := time.Now().UTC().Format(time.RFC3339)
236-
mode := cdc.STARTMODEENUM_FROM_UTC
237-
body.StartPosition = &cdc.StartPosition{
238-
Mode: &mode,
239-
Utc: &now,
240-
}
241-
242-
resp, err := d.CreateConnector(ctx, clusterID, body)
252+
resp, err := d.CreateChangefeed(ctx, clusterID, body)
243253
if err != nil {
244254
return errors.Trace(err)
245255
}
246-
_, err = fmt.Fprintln(h.IOStreams.Out, color.GreenString("changefeed %s created", resp.ConnectorId))
256+
_, err = fmt.Fprintln(h.IOStreams.Out, color.GreenString("changefeed %s created", resp.ChangefeedId))
247257
if err != nil {
248258
return err
249259
}
@@ -253,19 +263,11 @@ func CreateCmd(h *internal.Helper) *cobra.Command {
253263

254264
createCmd.Flags().StringP(flag.ClusterID, flag.ClusterIDShort, "", "The ID of the cluster.")
255265
createCmd.Flags().String(flag.ChangefeedName, "", "The name of the changefeed.")
256-
createCmd.Flags().String(flag.ChangefeedType, "", fmt.Sprintf("The type of the changefeed, one of %q", cdc.AllowedConnectorTypeEnumEnumValues))
266+
createCmd.Flags().String(flag.ChangefeedType, "", fmt.Sprintf("The type of the changefeed, one of %q", cdc.AllowedChangefeedTypeEnumEnumValues))
257267
createCmd.Flags().String(flag.ChangefeedKafka, "", "Kafka information in JSON format, use \"ticloud serverless changefeed template\" to see templates.")
268+
createCmd.Flags().String(flag.ChangefeedMySQL, "", "MySQL information in JSON format, use \"ticloud serverless changefeed template\" to see templates.")
258269
createCmd.Flags().String(flag.ChangefeedFilter, "", "Filter in JSON format, use \"ticloud serverless changefeed template\" to see templates.")
259270
createCmd.Flags().Uint64(flag.ChangefeedStartTSO, 0, "Start TSO for the changefeed, default to current TSO. See https://docs.pingcap.com/tidb/stable/tso/ for more information about TSO.")
260271

261272
return createCmd
262273
}
263-
264-
// inputDescription 用于交互式输入提示
265-
var inputDescription = map[string]string{
266-
flag.ChangefeedName: "The name of the changefeed, skip to use the default name",
267-
flag.ChangefeedType: fmt.Sprintf("The type of the changefeed, one of %q", cdc.AllowedConnectorTypeEnumEnumValues),
268-
flag.ChangefeedKafka: "Kafka information in JSON format, use \"ticloud serverless changefeed template\" to see templates.",
269-
flag.ChangefeedFilter: "Filter in JSON format, use \"ticloud serverless changefeed template\" to see templates.",
270-
flag.ChangefeedStartTSO: "Start TSO (uint64) for the changefeed, skip to use the current TSO. See https://docs.pingcap.com/tidb/stable/tso/ for more information about TSO.",
271-
}

‎internal/cli/serverless/changefeed/delete.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -150,7 +150,7 @@ func DeleteCmd(h *internal.Helper) *cobra.Command {
150150
}
151151
}
152152

153-
_, err = d.DeleteConnector(ctx, clusterID, changefeedID)
153+
_, err = d.DeleteChangefeed(ctx, clusterID, changefeedID)
154154
if err != nil {
155155
return errors.Trace(err)
156156
}

‎internal/cli/serverless/changefeed/describe.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -118,7 +118,7 @@ func DescribeCmd(h *internal.Helper) *cobra.Command {
118118
return errors.New("changefeed-id is required")
119119
}
120120

121-
changefeed, err := d.GetConnector(ctx, clusterID, changefeedID)
121+
changefeed, err := d.GetChangefeed(ctx, clusterID, changefeedID)
122122
if err != nil {
123123
return errors.Trace(err)
124124
}

‎internal/cli/serverless/changefeed/edit.go‎

Lines changed: 53 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,7 @@ func EditCmd(h *internal.Helper) *cobra.Command {
7777

7878
var editCmd = &cobra.Command{
7979
Use: "edit",
80-
Short: "Edit a changefeed (kafka and filter must be fully specified)",
80+
Short: "Edit a changefeed",
8181
Args: cobra.NoArgs,
8282
Example: fmt.Sprintf(` Update a changefeed in interactive mode:
8383
$ %[1]s serverless changefeed edit
@@ -95,8 +95,9 @@ func EditCmd(h *internal.Helper) *cobra.Command {
9595
}
9696
ctx := cmd.Context()
9797

98-
var clusterID, changefeedID, kafkaStr, filterStr string
99-
var kafkaInfo cdc.KafkaInfo
98+
var clusterID, changefeedID, kafkaStr, mysqlStr, filterStr string
99+
var kafkaInfo cdc.Kafka
100+
var mysqlInfo cdc.MySQL
100101
var filter cdc.CDCFilter
101102
var name *string
102103

@@ -120,16 +121,33 @@ func EditCmd(h *internal.Helper) *cobra.Command {
120121
}
121122
changefeedID = cf.ID
122123

123-
inputs := []string{flag.ChangefeedName, flag.ChangefeedKafka, flag.ChangefeedFilter}
124-
textInput, err := ui.InitialInputModel(inputs, updateInputDescriptionInteractive)
125-
if err != nil {
126-
return err
127-
}
128-
nameStr := textInput.Inputs[0].Value()
129-
kafkaStr = textInput.Inputs[1].Value()
130-
filterStr = textInput.Inputs[2].Value()
131-
if nameStr != "" {
132-
name = &nameStr
124+
switch cf.Type {
125+
case string(cdc.CHANGEFEEDTYPEENUM_KAFKA):
126+
inputs := []string{flag.ChangefeedName, flag.ChangefeedKafka, flag.ChangefeedFilter}
127+
textInput, err := ui.InitialInputModel(inputs, updateKafkaInputDescriptionInteractive)
128+
if err != nil {
129+
return err
130+
}
131+
nameStr := textInput.Inputs[0].Value()
132+
kafkaStr = textInput.Inputs[1].Value()
133+
filterStr = textInput.Inputs[2].Value()
134+
if nameStr != "" {
135+
name = &nameStr
136+
}
137+
case string(cdc.CHANGEFEEDTYPEENUM_MYSQL):
138+
inputs := []string{flag.ChangefeedName, flag.ChangefeedMySQL, flag.ChangefeedFilter}
139+
textInput, err := ui.InitialInputModel(inputs, updateMySQLInputDescriptionInteractive)
140+
if err != nil {
141+
return err
142+
}
143+
nameStr := textInput.Inputs[0].Value()
144+
kafkaStr = textInput.Inputs[1].Value()
145+
filterStr = textInput.Inputs[2].Value()
146+
if nameStr != "" {
147+
name = &nameStr
148+
}
149+
default:
150+
return errors.Errorf("unsupported changefeed type: %s", cf.Type)
133151
}
134152
} else {
135153
var err error
@@ -152,6 +170,10 @@ func EditCmd(h *internal.Helper) *cobra.Command {
152170
if err != nil {
153171
return errors.Trace(err)
154172
}
173+
mysqlStr, err = cmd.Flags().GetString(flag.ChangefeedMySQL)
174+
if err != nil {
175+
return errors.Trace(err)
176+
}
155177
filterStr, err = cmd.Flags().GetString(flag.ChangefeedFilter)
156178
if err != nil {
157179
return errors.Trace(err)
@@ -165,17 +187,17 @@ func EditCmd(h *internal.Helper) *cobra.Command {
165187
return errors.New("invalid filter, please use JSON format")
166188
}
167189

168-
body := &cdc.ConnectorServiceEditConnectorBody{
169-
Name: name,
170-
Filter: filter,
190+
body := &cdc.ChangefeedServiceEditChangefeedBody{
191+
DisplayName: name,
192+
Filter: filter,
171193
}
172194

173-
changefeed, err := d.GetConnector(ctx, clusterID, changefeedID)
195+
changefeed, err := d.GetChangefeed(ctx, clusterID, changefeedID)
174196
if err != nil {
175197
return errors.Trace(err)
176198
}
177199
switch changefeed.Sink.Type {
178-
case cdc.CONNECTORTYPEENUM_KAFKA:
200+
case cdc.CHANGEFEEDTYPEENUM_KAFKA:
179201
if kafkaStr == "" {
180202
return errors.New("kafka info (--kafka) is required and must be fully specified")
181203
}
@@ -186,11 +208,22 @@ func EditCmd(h *internal.Helper) *cobra.Command {
186208
Type: changefeed.Sink.Type,
187209
Kafka: &kafkaInfo,
188210
}
211+
case cdc.CHANGEFEEDTYPEENUM_MYSQL:
212+
if mysqlStr == "" {
213+
return errors.New("mysql info (--mysql) is required and must be fully specified")
214+
}
215+
if err := json.Unmarshal([]byte(kafkaStr), &mysqlInfo); err != nil {
216+
return errors.New("invalid kafka info, please use JSON format")
217+
}
218+
body.Sink = cdc.SinkInfo{
219+
Type: changefeed.Sink.Type,
220+
Mysql: &mysqlInfo,
221+
}
189222
default:
190223
return errors.Errorf("unsupported changefeed sink type: %s", changefeed.Sink.Type)
191224
}
192225

193-
_, err = d.EditConnector(ctx, clusterID, changefeedID, body)
226+
_, err = d.EditChangefeed(ctx, clusterID, changefeedID, body)
194227
if err != nil {
195228
return errors.Trace(err)
196229
}
@@ -206,14 +239,8 @@ func EditCmd(h *internal.Helper) *cobra.Command {
206239
editCmd.Flags().String(flag.ChangefeedID, "", "The ID of the changefeed to be updated.")
207240
editCmd.Flags().String(flag.ChangefeedName, "", "The new name of the changefeed.")
208241
editCmd.Flags().String(flag.ChangefeedKafka, "", "Complete kafka information in JSON format, use \"ticloud serverless changefeed template\" to see templates.")
242+
editCmd.Flags().String(flag.ChangefeedMySQL, "", "Complete mysql information in JSON format, use \"ticloud serverless changefeed template\" to see templates.")
209243
editCmd.Flags().String(flag.ChangefeedFilter, "", "Complete filter in JSON format, use \"ticloud serverless changefeed template\" to see templates.")
210244

211245
return editCmd
212246
}
213-
214-
// updateInputDescriptionInteractive 用于交互式输入提示（kafka 和 filter 必填）
215-
var updateInputDescriptionInteractive = map[string]string{
216-
flag.ChangefeedName: "The new name of the changefeed, skip to keep the current name.",
217-
flag.ChangefeedKafka: "Complete Kafka information in JSON format, use \"ticloud serverless changefeed template\" to see templates.",
218-
flag.ChangefeedFilter: "Complete Filter in JSON format, use \"ticloud serverless changefeed template\" to see templates.",
219-
}

0 commit comments

Comments
 (0)