diff --git a/action/common/args.go b/action/common/args.go index 4579c5f..43425f8 100644 --- a/action/common/args.go +++ b/action/common/args.go @@ -27,22 +27,29 @@ import ( ) func ReadArgs(in io.Reader) error { - os.Args = []string{""} + return ReadArgsFromScanner(bufio.NewScanner(in)) +} - scanner := bufio.NewScanner(in) +func ReadArgsFromScanner(scanner *bufio.Scanner) error { + os.Args = []string{""} var lines []string - for scanner.Scan() { - line := strings.Trim(scanner.Text(), "\r\n ") - if line[len(line)-1] == '\\' { - line = line[:len(line)-1] - lines = append(lines, line) + current := strings.Trim(scanner.Text(), "\r\n ") + if strings.HasSuffix(current, "\\") { + current = strings.TrimSuffix(current, "\\") + lines = append(lines, current) } else { - lines = append(lines, line) + lines = append(lines, current) break } } + if err := scanner.Err(); err != nil { + return err + } + if len(lines) == 0 { + return io.EOF + } argsStr := strings.Join(lines, " ") rawArgs := strings.Split(argsStr, "'") diff --git a/action/common/args_test.go b/action/common/args_test.go index 882fdf5..4d31d2f 100644 --- a/action/common/args_test.go +++ b/action/common/args_test.go @@ -18,8 +18,11 @@ package common import ( + "bufio" "bytes" + "io" "os" + "strings" "testing" "github.com/stretchr/testify/assert" @@ -134,13 +137,28 @@ func TestReadArgs(t *testing.T) { continue } assert.Nil(t, ReadArgs(&stdin)) - assert.Equal(t, len(os.Args), len(testCase.args)) - for i := 0; i < len(os.Args); i++ { - assert.Equal(t, os.Args[i], testCase.args[i]) - } + assert.Equal(t, testCase.args, os.Args[1:]) } } +func TestReadArgsPreservesFollowingCommands(t *testing.T) { + scanner := bufio.NewScanner(strings.NewReader("transaction --help\nquit\n")) + + assert.NoError(t, ReadArgsFromScanner(scanner)) + assert.Equal(t, []string{"transaction", "--help"}, os.Args[1:]) + + assert.NoError(t, ReadArgsFromScanner(scanner)) + assert.Equal(t, []string{"quit"}, os.Args[1:]) +} + +func TestReadArgsHandlesEmptyLineAndEOF(t *testing.T) { + stdin := strings.NewReader("\n") + + assert.NoError(t, ReadArgs(stdin)) + assert.Equal(t, []string{}, os.Args[1:]) + assert.ErrorIs(t, ReadArgs(stdin), io.EOF) +} + func TestParseDictArg(t *testing.T) { for _, testCase := range dictArgTestCases { if !testCase.valid { diff --git a/action/common/flags.go b/action/common/flags.go new file mode 100644 index 0000000..8b2bf83 --- /dev/null +++ b/action/common/flags.go @@ -0,0 +1,43 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package common + +import ( + "github.com/spf13/cobra" + "github.com/spf13/pflag" +) + +func ResetLocalFlags(cmd *cobra.Command) { + cmd.Flags().VisitAll(func(flag *pflag.Flag) { + _ = cmd.Flags().Set(flag.Name, flag.DefValue) + flag.Changed = false + }) +} + +func ResetLocalFlagsOnParseErrorAndHelp(cmd *cobra.Command) { + cmd.SetFlagErrorFunc(func(cmd *cobra.Command, err error) error { + ResetLocalFlags(cmd) + return err + }) + + helpFunc := cmd.HelpFunc() + cmd.SetHelpFunc(func(cmd *cobra.Command, args []string) { + defer ResetLocalFlags(cmd) + helpFunc(cmd, args) + }) +} diff --git a/action/diagnose/diagnose.go b/action/diagnose/diagnose.go new file mode 100644 index 0000000..7a1df73 --- /dev/null +++ b/action/diagnose/diagnose.go @@ -0,0 +1,37 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package diagnose + +import ( + "github.com/seata/seata-ctl/action/common" + "github.com/spf13/cobra" +) + +func init() { + DiagnoseCmd.AddCommand(RunCmd) + DiagnoseCmd.SetUsageTemplate(common.GetUsageTmpl("diagnose")) + DiagnoseCmd.SetHelpTemplate(common.GetHelpTmpl()) +} + +var DiagnoseCmd = &cobra.Command{ + Use: "diagnose", + Short: "Run read-only diagnostics", + Run: func(cmd *cobra.Command, _ []string) { + _ = cmd.Help() + }, +} diff --git a/action/diagnose/run.go b/action/diagnose/run.go new file mode 100644 index 0000000..098ecda --- /dev/null +++ b/action/diagnose/run.go @@ -0,0 +1,88 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package diagnose + +import ( + "fmt" + "strings" + "time" + + "github.com/seata/seata-ctl/action/common" + "github.com/seata/seata-ctl/seata" + "github.com/spf13/cobra" +) + +var RunCmd = newRunCommand() + +func newRunCommand() *cobra.Command { + var ( + output string + checkDB bool + dbAddress string + tcpTimeout time.Duration + pageNum int + pageSize int + lockPageNum int + lockPageSize int + ) + + cmd := &cobra.Command{ + Use: "run", + Short: "Run diagnostics against the current Seata server", + RunE: func(cmd *cobra.Command, _ []string) error { + defer common.ResetLocalFlags(cmd) + + output = strings.ToLower(output) + if _, err := seata.NormalizeOutput(output); err != nil { + return err + } + + report, err := seata.RunDiagnostics(seata.DiagnoseOptions{ + CheckDB: checkDB, + DBAddress: dbAddress, + TCPTimeout: tcpTimeout, + PageNum: pageNum, + PageSize: pageSize, + LockPageNum: lockPageNum, + LockPageSize: lockPageSize, + }) + if err != nil { + return err + } + + result, err := seata.FormatDiagnoseReport(report, output) + if err != nil { + return err + } + fmt.Println(result) + return nil + }, + } + common.ResetLocalFlagsOnParseErrorAndHelp(cmd) + cmd.SetUsageTemplate(common.GetUsageTmpl("diagnose run")) + cmd.SetHelpTemplate(common.GetHelpTmpl()) + cmd.Flags().BoolVar(&checkDB, "check-db", false, "Check database connectivity") + cmd.Flags().StringVar(&dbAddress, "db-address", "", "Database address in host:port form") + cmd.Flags().DurationVar(&tcpTimeout, "tcp-timeout", 3*time.Second, "TCP timeout") + cmd.Flags().IntVar(&pageNum, "page-num", 1, "Transaction page number") + cmd.Flags().IntVar(&pageSize, "page-size", 1, "Transaction page size") + cmd.Flags().IntVar(&lockPageNum, "lock-page-num", 1, "Lock page number") + cmd.Flags().IntVar(&lockPageSize, "lock-page-size", 1, "Lock page size") + cmd.Flags().StringVar(&output, "output", seata.OutputTable, "Output format: table, json, yaml") + return cmd +} diff --git a/action/diagnose/run_test.go b/action/diagnose/run_test.go new file mode 100644 index 0000000..af34beb --- /dev/null +++ b/action/diagnose/run_test.go @@ -0,0 +1,122 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package diagnose + +import ( + "bytes" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "net/url" + "os" + "strconv" + "strings" + "testing" + + "github.com/seata/seata-ctl/seata" + "github.com/spf13/cobra" +) + +func TestRunCommandBuildsReport(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case seata.LoginURL: + fmt.Fprint(w, `{"code":"200","message":"success","data":"test-token","success":true}`) + case seata.HealthCheckURL: + if got := r.Header.Get("authorization"); got != "test-token" { + t.Fatalf("authorization = %q, want test-token", got) + } + fmt.Fprint(w, `{"code":"200","message":"success","success":true,"data":[{"type":"nacos","address":"127.0.0.1:7091","status":"ok"}]}`) + case seata.GlobalSessionQueryURL: + if got := r.Header.Get("authorization"); got != "test-token" { + t.Fatalf("authorization = %q, want test-token", got) + } + fmt.Fprint(w, `{"code":"200","message":"success","success":true,"pageSize":1,"pageNum":1,"total":1,"pages":1,"data":[{"xid":"xid-1","transactionId":"1001","status":1,"applicationId":"order-service","transactionServiceGroup":"default_tx_group","transactionName":"createOrder","timeout":60000,"beginTime":1710000000000}]}`) + case seata.GlobalLockQueryURL: + if got := r.Header.Get("authorization"); got != "test-token" { + t.Fatalf("authorization = %q, want test-token", got) + } + fmt.Fprint(w, `{"code":"200","message":"success","success":true,"pageSize":1,"pageNum":1,"total":1,"pages":1,"data":[{"xid":"xid-1","transactionId":"1001","branchId":"2001","resourceId":"jdbc:mysql://127.0.0.1:3306/seata","tableName":"account","pk":"1","rowKey":"1","vgroup":"default_tx_group"}]}`) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + loginToServer(t, server) + + cmd := newRunCommand() + output, err := executeCommand(t, cmd, "--output", "json") + if err != nil { + t.Fatalf("execute command: %v", err) + } + if !strings.Contains(output, `"stage": "status"`) || !strings.Contains(output, `"stage": "transaction query"`) || !strings.Contains(output, `"stage": "global lock query"`) { + t.Fatalf("output = %s, want JSON diagnose report", output) + } +} + +func loginToServer(t *testing.T, server *httptest.Server) { + t.Helper() + parsedURL, err := url.Parse(server.URL) + if err != nil { + t.Fatalf("parse server url: %v", err) + } + host, portStr, err := net.SplitHostPort(parsedURL.Host) + if err != nil { + t.Fatalf("split host port: %v", err) + } + port, err := strconv.Atoi(portStr) + if err != nil { + t.Fatalf("parse port: %v", err) + } + auth := seata.GetAuth() + auth.ServerIP = host + auth.ServerPort = port + auth.Username = "seata" + auth.Password = "seata" + if err = auth.Login(); err != nil { + t.Fatalf("login to test server: %v", err) + } +} + +func executeCommand(t *testing.T, cmd *cobra.Command, args ...string) (string, error) { + t.Helper() + cmd.SetArgs(args) + cmd.SilenceUsage = true + cmd.SilenceErrors = true + + var output bytes.Buffer + cmd.SetOut(&output) + cmd.SetErr(&output) + + oldStdout := os.Stdout + reader, writer, err := os.Pipe() + if err != nil { + t.Fatalf("create stdout pipe: %v", err) + } + os.Stdout = writer + execErr := cmd.Execute() + _ = writer.Close() + os.Stdout = oldStdout + if _, err = io.Copy(&output, reader); err != nil { + t.Fatalf("read stdout: %v", err) + } + _ = reader.Close() + return output.String(), execErr +} diff --git a/action/lock/check.go b/action/lock/check.go new file mode 100644 index 0000000..c09d197 --- /dev/null +++ b/action/lock/check.go @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package lock + +import ( + "fmt" + "strings" + + "github.com/seata/seata-ctl/action/common" + "github.com/seata/seata-ctl/seata" + "github.com/spf13/cobra" +) + +var CheckCmd = newCheckCommand() + +func newCheckCommand() *cobra.Command { + var ( + xid string + branchID string + output string + ) + + cmd := &cobra.Command{ + Use: "check", + Short: "Check whether a global lock exists", + RunE: func(cmd *cobra.Command, _ []string) error { + defer common.ResetLocalFlags(cmd) + + output = strings.ToLower(output) + if _, err := seata.NormalizeOutput(output); err != nil { + return err + } + + response, err := seata.CheckGlobalLock(xid, branchID) + if err != nil { + return err + } + result, err := seata.FormatGlobalLockCheck(response, output) + if err != nil { + return err + } + fmt.Println(result) + return nil + }, + } + common.ResetLocalFlagsOnParseErrorAndHelp(cmd) + cmd.SetUsageTemplate(common.GetUsageTmpl("lock check")) + cmd.SetHelpTemplate(common.GetHelpTmpl()) + cmd.Flags().StringVar(&xid, "xid", "", "Transaction XID") + cmd.Flags().StringVar(&branchID, "branch-id", "", "Branch ID") + cmd.Flags().StringVar(&output, "output", seata.OutputTable, "Output format: table, json, yaml") + return cmd +} diff --git a/action/lock/list.go b/action/lock/list.go new file mode 100644 index 0000000..a5e4519 --- /dev/null +++ b/action/lock/list.go @@ -0,0 +1,101 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package lock + +import ( + "fmt" + "strings" + + "github.com/seata/seata-ctl/action/common" + "github.com/seata/seata-ctl/seata" + "github.com/spf13/cobra" +) + +var ListCmd = newListCommand() + +func newListCommand() *cobra.Command { + var ( + xid string + tableName string + transactionID string + branchID string + pk string + resourceID string + pageNum int + pageSize int + timeStart int64 + timeEnd int64 + output string + ) + + cmd := &cobra.Command{ + Use: "list", + Short: "List global locks", + RunE: func(cmd *cobra.Command, _ []string) error { + defer common.ResetLocalFlags(cmd) + + output = strings.ToLower(output) + if _, err := seata.NormalizeOutput(output); err != nil { + return err + } + + query := seata.GlobalLockQuery{ + XID: xid, + TableName: tableName, + TransactionID: transactionID, + BranchID: branchID, + PK: pk, + ResourceID: resourceID, + PageNum: pageNum, + PageSize: pageSize, + } + if cmd.Flags().Changed("time-start") { + query.TimeStart = &timeStart + } + if cmd.Flags().Changed("time-end") { + query.TimeEnd = &timeEnd + } + + response, err := seata.QueryGlobalLocks(query) + if err != nil { + return err + } + result, err := seata.FormatGlobalLockPage(response, output) + if err != nil { + return err + } + fmt.Println(result) + return nil + }, + } + common.ResetLocalFlagsOnParseErrorAndHelp(cmd) + cmd.SetUsageTemplate(common.GetUsageTmpl("lock list")) + cmd.SetHelpTemplate(common.GetHelpTmpl()) + cmd.Flags().StringVar(&xid, "xid", "", "Filter by transaction XID") + cmd.Flags().StringVar(&tableName, "table-name", "", "Filter by table name") + cmd.Flags().StringVar(&transactionID, "transaction-id", "", "Filter by transaction ID") + cmd.Flags().StringVar(&branchID, "branch-id", "", "Filter by branch ID") + cmd.Flags().StringVar(&pk, "pk", "", "Filter by primary key") + cmd.Flags().StringVar(&resourceID, "resource-id", "", "Filter by resource ID") + cmd.Flags().IntVar(&pageNum, "page-num", 1, "Page number") + cmd.Flags().IntVar(&pageSize, "page-size", 20, "Page size") + cmd.Flags().Int64Var(&timeStart, "time-start", 0, "Filter by lock time start, in milliseconds") + cmd.Flags().Int64Var(&timeEnd, "time-end", 0, "Filter by lock time end, in milliseconds") + cmd.Flags().StringVar(&output, "output", seata.OutputTable, "Output format: table, json, yaml") + return cmd +} diff --git a/action/lock/lock.go b/action/lock/lock.go new file mode 100644 index 0000000..e1a5217 --- /dev/null +++ b/action/lock/lock.go @@ -0,0 +1,38 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package lock + +import ( + "github.com/seata/seata-ctl/action/common" + "github.com/spf13/cobra" +) + +func init() { + LockCmd.AddCommand(ListCmd) + LockCmd.AddCommand(CheckCmd) + LockCmd.SetUsageTemplate(common.GetUsageTmpl("lock")) + LockCmd.SetHelpTemplate(common.GetHelpTmpl()) +} + +var LockCmd = &cobra.Command{ + Use: "lock", + Short: "Query global locks", + Run: func(cmd *cobra.Command, _ []string) { + _ = cmd.Help() + }, +} diff --git a/action/lock/lock_test.go b/action/lock/lock_test.go new file mode 100644 index 0000000..c5ea3d9 --- /dev/null +++ b/action/lock/lock_test.go @@ -0,0 +1,173 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package lock + +import ( + "bytes" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "net/url" + "os" + "strconv" + "strings" + "testing" + + "github.com/seata/seata-ctl/seata" + "github.com/spf13/cobra" +) + +func TestListCommandBuildsQuery(t *testing.T) { + server := newConsoleServer(t, func(r *http.Request) { + wantQuery := map[string]string{ + "xid": "xid-1", + "tableName": "account", + "transactionId": "1001", + "branchId": "2001", + "pk": "1", + "resourceId": "jdbc:mysql://127.0.0.1:3306/seata", + "pageNum": "2", + "pageSize": "5", + "timeStart": "1710000000000", + "timeEnd": "1710003600000", + } + for key, want := range wantQuery { + if got := r.URL.Query().Get(key); got != want { + t.Fatalf("query[%s] = %q, want %q", key, got, want) + } + } + }) + defer server.Close() + loginToServer(t, server) + + cmd := newListCommand() + output, err := executeCommand(t, cmd, + "--xid", "xid-1", + "--table-name", "account", + "--transaction-id", "1001", + "--branch-id", "2001", + "--pk", "1", + "--resource-id", "jdbc:mysql://127.0.0.1:3306/seata", + "--page-num", "2", + "--page-size", "5", + "--time-start", "1710000000000", + "--time-end", "1710003600000", + "--output", "json", + ) + if err != nil { + t.Fatalf("execute command: %v", err) + } + if !strings.Contains(output, `"pageNum": 2`) || !strings.Contains(output, `"data"`) { + t.Fatalf("output = %s, want JSON page output", output) + } +} + +func TestCheckCommandBuildsQuery(t *testing.T) { + server := newConsoleServer(t, nil) + defer server.Close() + loginToServer(t, server) + + cmd := newCheckCommand() + output, err := executeCommand(t, cmd, "--xid", "xid-1", "--branch-id", "2001") + if err != nil { + t.Fatalf("execute command: %v", err) + } + if !strings.Contains(output, "locked") && !strings.Contains(output, "true") { + t.Fatalf("output = %s, want lock status", output) + } +} + +func newConsoleServer(t *testing.T, onQuery func(*http.Request)) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case seata.LoginURL: + fmt.Fprint(w, `{"code":"200","message":"success","data":"test-token","success":true}`) + case seata.GlobalLockQueryURL: + if onQuery != nil { + onQuery(r) + } + pageNum := r.URL.Query().Get("pageNum") + if pageNum == "" { + pageNum = "1" + } + pageSize := r.URL.Query().Get("pageSize") + if pageSize == "" { + pageSize = "20" + } + fmt.Fprintf(w, `{"code":"200","message":"success","success":true,"pageSize":%s,"pageNum":%s,"total":1,"pages":1,"data":[{"xid":"xid-1","transactionId":"1001","branchId":"2001","resourceId":"jdbc:mysql://127.0.0.1:3306/seata","tableName":"account","pk":"1","rowKey":"1","vgroup":"default_tx_group"}]}`, pageSize, pageNum) + case seata.GlobalLockCheckURL: + fmt.Fprint(w, `{"code":"200","message":"success","success":true,"data":true}`) + default: + http.NotFound(w, r) + } + })) +} + +func loginToServer(t *testing.T, server *httptest.Server) { + t.Helper() + parsedURL, err := url.Parse(server.URL) + if err != nil { + t.Fatalf("parse server url: %v", err) + } + host, portStr, err := net.SplitHostPort(parsedURL.Host) + if err != nil { + t.Fatalf("split host port: %v", err) + } + port, err := strconv.Atoi(portStr) + if err != nil { + t.Fatalf("parse port: %v", err) + } + auth := seata.GetAuth() + auth.ServerIP = host + auth.ServerPort = port + auth.Username = "seata" + auth.Password = "seata" + if err = auth.Login(); err != nil { + t.Fatalf("login to test server: %v", err) + } +} + +func executeCommand(t *testing.T, cmd *cobra.Command, args ...string) (string, error) { + t.Helper() + cmd.SetArgs(args) + cmd.SilenceUsage = true + cmd.SilenceErrors = true + + var output bytes.Buffer + cmd.SetOut(&output) + cmd.SetErr(&output) + + oldStdout := os.Stdout + reader, writer, err := os.Pipe() + if err != nil { + t.Fatalf("create stdout pipe: %v", err) + } + os.Stdout = writer + execErr := cmd.Execute() + _ = writer.Close() + os.Stdout = oldStdout + if _, err = io.Copy(&output, reader); err != nil { + t.Fatalf("read stdout: %v", err) + } + _ = reader.Close() + return output.String(), execErr +} + diff --git a/action/root.go b/action/root.go index 53a7f35..a3ae3e8 100644 --- a/action/root.go +++ b/action/root.go @@ -20,14 +20,18 @@ package action import ( "github.com/seata/seata-ctl/action/common" "github.com/seata/seata-ctl/action/config" + "github.com/seata/seata-ctl/action/diagnose" "github.com/seata/seata-ctl/action/get" "github.com/seata/seata-ctl/action/k8s" + "github.com/seata/seata-ctl/action/lock" "github.com/seata/seata-ctl/action/log" "github.com/seata/seata-ctl/action/login" "github.com/seata/seata-ctl/action/prometheus" "github.com/seata/seata-ctl/action/reload" se "github.com/seata/seata-ctl/action/set" + "github.com/seata/seata-ctl/action/transaction" del "github.com/seata/seata-ctl/action/try" + "github.com/seata/seata-ctl/action/tui" "github.com/spf13/cobra" ) @@ -48,6 +52,10 @@ func init() { k8s.ScaleCmd, prometheus.MetricsCmd, log.LogCmd, + diagnose.DiagnoseCmd, + tui.TuiCmd, + lock.LockCmd, + transaction.TransactionCmd, ) rootCmd.SetHelpTemplate(common.GetHelpTmplWithOnlyAvailableCmd()) rootCmd.CompletionOptions = cobra.CompletionOptions{ diff --git a/action/transaction/list.go b/action/transaction/list.go new file mode 100644 index 0000000..66f0578 --- /dev/null +++ b/action/transaction/list.go @@ -0,0 +1,105 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package transaction + +import ( + "fmt" + "strings" + + "github.com/seata/seata-ctl/action/common" + "github.com/seata/seata-ctl/seata" + "github.com/spf13/cobra" +) + +var ListCmd = newListCommand() + +func newListCommand() *cobra.Command { + var ( + xid string + applicationID string + status int + transactionName string + vgroup string + withBranch bool + pageNum int + pageSize int + timeStart int64 + timeEnd int64 + output string + ) + + cmd := &cobra.Command{ + Use: "list", + Short: "List global transactions", + RunE: func(cmd *cobra.Command, _ []string) error { + defer common.ResetLocalFlags(cmd) + + output = strings.ToLower(output) + if output != seata.OutputTable && output != seata.OutputJSON && output != seata.OutputYAML { + return fmt.Errorf("unsupported output format %q", output) + } + + query := seata.GlobalSessionQuery{ + XID: xid, + ApplicationID: applicationID, + TransactionName: transactionName, + Vgroup: vgroup, + PageNum: pageNum, + PageSize: pageSize, + } + if cmd.Flags().Changed("status") { + query.Status = &status + } + if cmd.Flags().Changed("with-branch") { + query.WithBranch = &withBranch + } + if cmd.Flags().Changed("time-start") { + query.TimeStart = &timeStart + } + if cmd.Flags().Changed("time-end") { + query.TimeEnd = &timeEnd + } + + response, err := seata.QueryGlobalSessions(query) + if err != nil { + return err + } + result, err := seata.FormatGlobalSessionPage(response, output) + if err != nil { + return err + } + fmt.Println(result) + return nil + }, + } + common.ResetLocalFlagsOnParseErrorAndHelp(cmd) + cmd.SetUsageTemplate(common.GetUsageTmpl("transaction list")) + cmd.SetHelpTemplate(common.GetHelpTmpl()) + cmd.Flags().StringVar(&xid, "xid", "", "Filter by transaction XID") + cmd.Flags().StringVar(&applicationID, "application-id", "", "Filter by application ID") + cmd.Flags().IntVar(&status, "status", 0, "Filter by global transaction status code") + cmd.Flags().StringVar(&transactionName, "transaction-name", "", "Filter by transaction name") + cmd.Flags().StringVar(&vgroup, "vgroup", "", "Filter by transaction service group") + cmd.Flags().BoolVar(&withBranch, "with-branch", false, "Include branch sessions") + cmd.Flags().IntVar(&pageNum, "page-num", 1, "Page number") + cmd.Flags().IntVar(&pageSize, "page-size", 20, "Page size") + cmd.Flags().Int64Var(&timeStart, "time-start", 0, "Filter by begin time start, in milliseconds") + cmd.Flags().Int64Var(&timeEnd, "time-end", 0, "Filter by begin time end, in milliseconds") + cmd.Flags().StringVar(&output, "output", seata.OutputTable, "Output format: table, json, yaml") + return cmd +} diff --git a/action/transaction/list_test.go b/action/transaction/list_test.go new file mode 100644 index 0000000..cb7ecd7 --- /dev/null +++ b/action/transaction/list_test.go @@ -0,0 +1,220 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package transaction + +import ( + "bytes" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "net/url" + "os" + "strconv" + "strings" + "sync" + "testing" + + "github.com/seata/seata-ctl/seata" + "github.com/spf13/cobra" +) + +func TestListCommandBuildsQuery(t *testing.T) { + var ( + mu sync.Mutex + gotHeader string + gotQuery url.Values + ) + server := newConsoleServer(t, func(r *http.Request) { + mu.Lock() + defer mu.Unlock() + gotHeader = r.Header.Get("authorization") + gotQuery = r.URL.Query() + }) + defer server.Close() + loginToServer(t, server) + + cmd := newListCommand() + output, err := executeCommand(t, cmd, + "--xid", "xid-1", + "--application-id", "order-service", + "--status", "1", + "--transaction-name", "createOrder", + "--vgroup", "default_tx_group", + "--with-branch", + "--page-num", "2", + "--page-size", "5", + "--time-start", "1710000000000", + "--time-end", "1710003600000", + "--output", "json", + ) + if err != nil { + t.Fatalf("execute command: %v", err) + } + if !strings.Contains(output, `"pageNum": 2`) { + t.Fatalf("output = %s, want JSON pageNum", output) + } + + mu.Lock() + defer mu.Unlock() + if gotHeader != "test-token" { + t.Fatalf("authorization = %q, want test-token", gotHeader) + } + wantQuery := map[string]string{ + "xid": "xid-1", + "applicationId": "order-service", + "status": "1", + "transactionName": "createOrder", + "vgroup": "default_tx_group", + "withBranch": "true", + "pageNum": "2", + "pageSize": "5", + "timeStart": "1710000000000", + "timeEnd": "1710003600000", + } + for key, want := range wantQuery { + if got := gotQuery.Get(key); got != want { + t.Fatalf("query[%s] = %q, want %q", key, got, want) + } + } +} + +func TestListCommandResetsFlagsAfterParseError(t *testing.T) { + server := newConsoleServer(t, nil) + defer server.Close() + loginToServer(t, server) + + cmd := newListCommand() + _, err := executeCommand(t, cmd, "--output", "json", "--unknown") + if err == nil || !strings.Contains(err.Error(), "unknown flag") { + t.Fatalf("error = %v, want unknown flag", err) + } + + output, err := executeCommand(t, cmd) + if err != nil { + t.Fatalf("execute command after parse error: %v", err) + } + if strings.Contains(output, `"code":`) { + t.Fatalf("output = %s, want default table output after parse error", output) + } + if !strings.Contains(output, "transaction_id") { + t.Fatalf("output = %s, want table header", output) + } +} + +func TestListCommandResetsFlagsAfterHelp(t *testing.T) { + server := newConsoleServer(t, nil) + defer server.Close() + loginToServer(t, server) + + cmd := newListCommand() + helpOutput, err := executeCommand(t, cmd, "--output", "json", "--help") + if err != nil { + t.Fatalf("execute help command: %v", err) + } + if !strings.Contains(helpOutput, "List global transactions") { + t.Fatalf("help output = %s, want transaction list help", helpOutput) + } + + output, err := executeCommand(t, cmd) + if err != nil { + t.Fatalf("execute command after help: %v", err) + } + if strings.Contains(output, `"code":`) { + t.Fatalf("output = %s, want default table output after help", output) + } + if !strings.Contains(output, "transaction_id") { + t.Fatalf("output = %s, want table header", output) + } +} + +func newConsoleServer(t *testing.T, onQuery func(*http.Request)) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case seata.LoginURL: + fmt.Fprint(w, `{"code":"200","message":"success","data":"test-token","success":true}`) + case seata.GlobalSessionQueryURL: + if onQuery != nil { + onQuery(r) + } + pageNum := r.URL.Query().Get("pageNum") + if pageNum == "" { + pageNum = "1" + } + pageSize := r.URL.Query().Get("pageSize") + if pageSize == "" { + pageSize = "20" + } + fmt.Fprintf(w, `{"code":"200","message":"success","success":true,"pageSize":%s,"pageNum":%s,"total":1,"pages":1,"data":[{"xid":"xid-1","transactionId":"1001","status":1,"applicationId":"order-service","transactionServiceGroup":"default_tx_group","transactionName":"createOrder","timeout":60000,"beginTime":1710000000000}]}`, pageSize, pageNum) + default: + http.NotFound(w, r) + } + })) +} + +func loginToServer(t *testing.T, server *httptest.Server) { + t.Helper() + parsedURL, err := url.Parse(server.URL) + if err != nil { + t.Fatalf("parse server url: %v", err) + } + host, portStr, err := net.SplitHostPort(parsedURL.Host) + if err != nil { + t.Fatalf("split host port: %v", err) + } + port, err := strconv.Atoi(portStr) + if err != nil { + t.Fatalf("parse port: %v", err) + } + auth := seata.GetAuth() + auth.ServerIP = host + auth.ServerPort = port + auth.Username = "seata" + auth.Password = "seata" + if err = auth.Login(); err != nil { + t.Fatalf("login to test server: %v", err) + } +} + +func executeCommand(t *testing.T, cmd *cobra.Command, args ...string) (string, error) { + t.Helper() + cmd.SetArgs(args) + cmd.SilenceUsage = true + cmd.SilenceErrors = true + + var output bytes.Buffer + cmd.SetOut(&output) + cmd.SetErr(&output) + + oldStdout := os.Stdout + reader, writer, err := os.Pipe() + if err != nil { + t.Fatalf("create stdout pipe: %v", err) + } + os.Stdout = writer + execErr := cmd.Execute() + _ = writer.Close() + os.Stdout = oldStdout + if _, err = io.Copy(&output, reader); err != nil { + t.Fatalf("read stdout: %v", err) + } + _ = reader.Close() + return output.String(), execErr +} diff --git a/action/transaction/show.go b/action/transaction/show.go new file mode 100644 index 0000000..88821e8 --- /dev/null +++ b/action/transaction/show.go @@ -0,0 +1,69 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package transaction + +import ( + "fmt" + "strings" + + "github.com/seata/seata-ctl/action/common" + "github.com/seata/seata-ctl/seata" + "github.com/spf13/cobra" +) + +var ShowCmd = newShowCommand() + +func newShowCommand() *cobra.Command { + var ( + xid string + output string + ) + + cmd := &cobra.Command{ + Use: "show", + Short: "Show one global transaction", + RunE: func(cmd *cobra.Command, _ []string) error { + defer common.ResetLocalFlags(cmd) + + output = strings.ToLower(output) + if _, err := seata.NormalizeOutput(output); err != nil { + return err + } + if xid == "" { + return fmt.Errorf("xid is required") + } + + session, err := seata.QueryGlobalSessionByXID(xid) + if err != nil { + return err + } + result, err := seata.FormatGlobalSessionDetail(session, output) + if err != nil { + return err + } + fmt.Println(result) + return nil + }, + } + common.ResetLocalFlagsOnParseErrorAndHelp(cmd) + cmd.SetUsageTemplate(common.GetUsageTmpl("transaction show")) + cmd.SetHelpTemplate(common.GetHelpTmpl()) + cmd.Flags().StringVar(&xid, "xid", "", "Transaction XID") + cmd.Flags().StringVar(&output, "output", seata.OutputTable, "Output format: table, json, yaml") + return cmd +} diff --git a/action/transaction/show_test.go b/action/transaction/show_test.go new file mode 100644 index 0000000..22b271b --- /dev/null +++ b/action/transaction/show_test.go @@ -0,0 +1,50 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package transaction + +import ( + "strings" + "testing" +) + +func TestShowCommandBuildsQuery(t *testing.T) { + server := newConsoleServer(t, nil) + defer server.Close() + loginToServer(t, server) + + cmd := newShowCommand() + output, err := executeCommand(t, cmd, "--xid", "xid-1", "--output", "json") + if err != nil { + t.Fatalf("execute command: %v", err) + } + if !strings.Contains(output, `"xid": "xid-1"`) || !strings.Contains(output, `"transactionId": "1001"`) { + t.Fatalf("output = %s, want JSON transaction detail", output) + } +} + +func TestShowCommandRequiresXID(t *testing.T) { + server := newConsoleServer(t, nil) + defer server.Close() + loginToServer(t, server) + + cmd := newShowCommand() + _, err := executeCommand(t, cmd, "--output", "json") + if err == nil || !strings.Contains(err.Error(), "xid is required") { + t.Fatalf("error = %v, want xid is required", err) + } +} diff --git a/action/transaction/transaction.go b/action/transaction/transaction.go new file mode 100644 index 0000000..8912719 --- /dev/null +++ b/action/transaction/transaction.go @@ -0,0 +1,38 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package transaction + +import ( + "github.com/seata/seata-ctl/action/common" + "github.com/spf13/cobra" +) + +func init() { + TransactionCmd.AddCommand(ListCmd) + TransactionCmd.AddCommand(ShowCmd) + TransactionCmd.SetUsageTemplate(common.GetUsageTmpl("transaction")) + TransactionCmd.SetHelpTemplate(common.GetHelpTmpl()) +} + +var TransactionCmd = &cobra.Command{ + Use: "transaction", + Short: "Query transaction resources", + Run: func(cmd *cobra.Command, _ []string) { + _ = cmd.Help() + }, +} diff --git a/action/tui/tui.go b/action/tui/tui.go new file mode 100644 index 0000000..a9333d0 --- /dev/null +++ b/action/tui/tui.go @@ -0,0 +1,277 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package tui + +import ( + "fmt" + "strings" + "time" + + tea "github.com/charmbracelet/bubbletea" + "github.com/seata/seata-ctl/action/common" + "github.com/seata/seata-ctl/seata" + "github.com/spf13/cobra" +) + +func init() { + TuiCmd.SetUsageTemplate(common.GetUsageTmpl("tui")) + TuiCmd.SetHelpTemplate(common.GetHelpTmpl()) + TuiCmd.Flags().Duration("refresh", 5*time.Second, "Auto refresh interval") + TuiCmd.Flags().Int("page-size", 20, "Transaction and lock page size") + TuiCmd.Flags().Bool("check-db", false, "Check database connectivity") + TuiCmd.Flags().String("db-address", "", "Database address in host:port form") +} + +var TuiCmd = &cobra.Command{ + Use: "tui", + Short: "Open the diagnostic TUI", + RunE: func(cmd *cobra.Command, _ []string) error { + refreshEvery, _ := cmd.Flags().GetDuration("refresh") + pageSize, _ := cmd.Flags().GetInt("page-size") + checkDB, _ := cmd.Flags().GetBool("check-db") + dbAddress, _ := cmd.Flags().GetString("db-address") + return tea.NewProgram(newModel(checkDB, dbAddress, pageSize, refreshEvery), tea.WithAltScreen()).Start() + }, +} + +type page int + +const ( + pageDiagnose page = iota + pageTransaction + pageLock +) + +type snapshot = seata.DiagnoseSnapshot + +type snapshotMsg struct{ snapshot snapshot } +type refreshMsg struct{} + +type model struct { + activePage page + autoRefresh bool + refreshEvery time.Duration + pageSize int + checkDB bool + dbAddress string + snapshot snapshot + refreshing bool + lastUpdated time.Time +} + +func newModel(checkDB bool, dbAddress string, pageSize int, refreshEvery time.Duration) model { + if pageSize <= 0 { + pageSize = 20 + } + if refreshEvery <= 0 { + refreshEvery = 5 * time.Second + } + return model{ + activePage: pageDiagnose, + autoRefresh: true, + refreshEvery: refreshEvery, + pageSize: pageSize, + checkDB: checkDB, + dbAddress: dbAddress, + refreshing: true, + } +} + +func (m model) Init() tea.Cmd { + return tea.Batch(m.refreshCmd(), m.tickCmd()) +} + +func (m model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { + switch msg := msg.(type) { + case tea.KeyMsg: + switch msg.String() { + case "q", "ctrl+c": + return m, tea.Quit + case "1": + m.activePage = pageDiagnose + case "2": + m.activePage = pageTransaction + case "3": + m.activePage = pageLock + case "tab": + m.activePage = (m.activePage + 1) % 3 + case "shift+tab": + m.activePage = (m.activePage + 2) % 3 + case "a": + m.autoRefresh = !m.autoRefresh + return m, m.tickCmd() + case "r": + if m.refreshing { + return m, nil + } + m.refreshing = true + return m, m.refreshCmd() + } + case tea.WindowSizeMsg: + return m, nil + case refreshMsg: + if !m.autoRefresh || m.refreshing { + return m, nil + } + m.refreshing = true + return m, m.refreshCmd() + case snapshotMsg: + m.refreshing = false + m.applySnapshot(msg.snapshot) + if msg.snapshot.Report != nil && msg.snapshot.Report.Success != nil && *msg.snapshot.Report.Success { + m.lastUpdated = time.Now() + } + return m, m.tickCmd() + } + return m, nil +} + +func (m model) View() string { + var body string + switch m.activePage { + case pageDiagnose: + body = m.viewDiagnose() + case pageTransaction: + body = m.viewTransaction() + case pageLock: + body = m.viewLock() + } + + state := "manual" + if m.autoRefresh { + state = "auto" + } + updated := "loading" + if !m.lastUpdated.IsZero() { + updated = m.lastUpdated.Format("15:04:05") + } + return fmt.Sprintf("seata-ctl tui | [1] diagnose [2] transaction [3] lock | refresh=%s | %s | %s\n\n%s\n", m.refreshEvery, state, updated, body) +} + +func (m model) refreshCmd() tea.Cmd { + return func() tea.Msg { + return snapshotMsg{snapshot: collectSnapshot(m.checkDB, m.dbAddress, m.pageSize)} + } +} + +func (m model) tickCmd() tea.Cmd { + if !m.autoRefresh || m.refreshEvery <= 0 { + return nil + } + return tea.Tick(m.refreshEvery, func(time.Time) tea.Msg { return refreshMsg{} }) +} + +func (m *model) applySnapshot(next snapshot) { + if next.Report != nil { + m.snapshot.Report = next.Report + } + if next.Transactions != nil { + m.snapshot.Transactions = next.Transactions + } + if next.TransactionErr != nil { + m.snapshot.TransactionErr = next.TransactionErr + } else { + m.snapshot.TransactionErr = nil + } + if next.Locks != nil { + m.snapshot.Locks = next.Locks + } + if next.LockErr != nil { + m.snapshot.LockErr = next.LockErr + } else { + m.snapshot.LockErr = nil + } +} + +func (m model) viewDiagnose() string { + if m.snapshot.Report == nil { + return "loading diagnostics..." + } + return strings.TrimSpace(renderDiagnosePage(m.snapshot.Report)) +} + +func (m model) viewTransaction() string { + issue := diagnoseStageIssue(m.snapshot.Report, "transaction query") + if m.snapshot.Transactions == nil { + if m.snapshot.TransactionErr != nil { + return "transaction query failed: " + m.snapshot.TransactionErr.Error() + } + if issue != "" { + return "transaction query unavailable: " + issue + } + return "loading transactions..." + } + output := strings.TrimSpace(seata.FormatGlobalSessionTable(m.snapshot.Transactions.Data)) + if issue != "" { + return output + "\n\nWARN: transaction query " + issue + } + return output +} + +func (m model) viewLock() string { + issue := diagnoseStageIssue(m.snapshot.Report, "global lock query") + if m.snapshot.Locks == nil { + if m.snapshot.LockErr != nil { + return "lock query failed: " + m.snapshot.LockErr.Error() + } + if issue != "" { + return "lock query unavailable: " + issue + } + return "loading locks..." + } + output := strings.TrimSpace(seata.FormatGlobalLockTable(m.snapshot.Locks.Data)) + if issue != "" { + return output + "\n\nWARN: lock query " + issue + } + return output +} + +func collectSnapshot(checkDB bool, dbAddress string, pageSize int) snapshot { + return *seata.CollectDiagnoseSnapshot(seata.DiagnoseOptions{ + CheckDB: checkDB, + DBAddress: dbAddress, + TCPTimeout: 3 * time.Second, + PageNum: 1, + PageSize: pageSize, + LockPageNum: 1, + LockPageSize: pageSize, + }) +} + +func renderDiagnosePage(report *seata.DiagnoseReport) string { + output, err := seata.FormatDiagnoseReport(report, seata.OutputTable) + if err != nil { + return err.Error() + } + if report.Success != nil && !*report.Success { + return output + "\n\nWARN: diagnostics report has failures" + } + return output +} + +func diagnoseStageIssue(report *seata.DiagnoseReport, stage string) string { + if report == nil { + return "" + } + for _, result := range report.Data { + if result.Stage == stage && result.Status != seata.DiagnoseStatusPass { + return result.Status + ": " + result.Message + } + } + return "" +} diff --git a/action/tui/tui_test.go b/action/tui/tui_test.go new file mode 100644 index 0000000..17a587e --- /dev/null +++ b/action/tui/tui_test.go @@ -0,0 +1,128 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package tui + +import ( + "errors" + "strings" + "testing" + "time" + + tea "github.com/charmbracelet/bubbletea" + "github.com/seata/seata-ctl/seata" +) + +func TestModelNavigationAndRefreshControls(t *testing.T) { + m := newModel(false, "", 20, 5) + if !m.refreshing || m.activePage != pageDiagnose { + t.Fatalf("initial model = %+v", m) + } + + updated, cmd := m.Update(key('2')) + m = updated.(model) + if cmd != nil || m.activePage != pageTransaction { + t.Fatalf("transaction navigation = page %d, cmd %v", m.activePage, cmd) + } + + m.refreshing = false + updated, cmd = m.Update(key('r')) + m = updated.(model) + if !m.refreshing || cmd == nil { + t.Fatalf("manual refresh = %+v, cmd %v", m, cmd) + } + updated, cmd = m.Update(key('r')) + m = updated.(model) + if !m.refreshing || cmd != nil { + t.Fatalf("duplicate refresh = %+v, cmd %v", m, cmd) + } + + updated, cmd = m.Update(snapshotMsg{}) + m = updated.(model) + if m.refreshing || cmd == nil { + t.Fatalf("completed refresh = %+v, cmd %v", m, cmd) + } + + updated, cmd = m.Update(key('a')) + m = updated.(model) + if m.autoRefresh || cmd != nil { + t.Fatalf("disabled auto refresh = %+v, cmd %v", m, cmd) + } + updated, cmd = m.Update(refreshMsg{}) + m = updated.(model) + if cmd != nil || m.refreshing { + t.Fatalf("refresh while disabled = %+v, cmd %v", m, cmd) + } +} + +func TestModelKeepsPreviousDataAndShowsWarning(t *testing.T) { + m := newModel(false, "", 20, 5) + transactionID := "1001" + transactions := &seata.GlobalSessionPageResult{ + Data: []seata.GlobalSession{{TransactionID: &transactionID}}, + } + m.applySnapshot(snapshot{ + Report: reportWithStage("transaction query", seata.DiagnoseStatusPass, "ok"), + Transactions: transactions, + }) + + m.applySnapshot(snapshot{ + Report: reportWithStage("transaction query", seata.DiagnoseStatusFail, "server unavailable"), + TransactionErr: errors.New("server unavailable"), + }) + output := m.viewTransaction() + if !strings.Contains(output, "1001") || !strings.Contains(output, "WARN") { + t.Fatalf("transaction view = %q, want previous data and warning", output) + } +} + +func TestModelShowsSkippedPageAsUnavailable(t *testing.T) { + m := newModel(false, "", 20, 5) + m.applySnapshot(snapshot{ + Report: reportWithStage("global lock query", seata.DiagnoseStatusSkip, "status failed"), + }) + if output := m.viewLock(); !strings.Contains(output, "unavailable") || !strings.Contains(output, "status failed") { + t.Fatalf("lock view = %q, want skipped warning", output) + } +} + +func TestModelKeepsLastUpdatedWhenRefreshFails(t *testing.T) { + m := newModel(false, "", 20, 5) + old := time.Unix(1710000000, 0) + m.lastUpdated = old + + updated, cmd := m.Update(snapshotMsg{snapshot: snapshot{Report: reportWithStage("transaction query", seata.DiagnoseStatusFail, "server unavailable")}}) + m = updated.(model) + if cmd == nil { + t.Fatal("expected tick command after refresh") + } + if !m.lastUpdated.Equal(old) { + t.Fatalf("lastUpdated = %v, want %v", m.lastUpdated, old) + } +} + +func key(runeValue rune) tea.KeyMsg { + return tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune{runeValue}} +} + +func reportWithStage(stage, status, message string) *seata.DiagnoseReport { + success := status == seata.DiagnoseStatusPass + return &seata.DiagnoseReport{ + Success: &success, + Data: []seata.CheckResult{{Stage: stage, Status: status, Message: message}}, + } +} diff --git a/changes/en-us/0.0.1.md b/changes/en-us/0.0.1.md index 4fb5530..5ba1fd1 100644 --- a/changes/en-us/0.0.1.md +++ b/changes/en-us/0.0.1.md @@ -15,3 +15,5 @@ See the License for the specific language governing permissions and limitations under the License. --> ### 0.0.1 + +- Add read-only diagnostics: `diagnose run`, `transaction list/show`, `lock list/check`, and a terminal TUI, all backed by table, JSON, and YAML output. diff --git a/changes/zh-cn/0.0.1.md b/changes/zh-cn/0.0.1.md index 4fb5530..aab9bba 100644 --- a/changes/zh-cn/0.0.1.md +++ b/changes/zh-cn/0.0.1.md @@ -15,3 +15,5 @@ See the License for the specific language governing permissions and limitations under the License. --> ### 0.0.1 + +- 新增只读诊断能力:`diagnose run`、`transaction list/show`、`lock list/check`,以及终端界面,并支持 table、JSON、YAML 输出。 diff --git a/cmd/root.go b/cmd/root.go index 2eb470b..86fb221 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -18,7 +18,10 @@ package cmd import ( + "bufio" + "errors" "fmt" + "io" "os" "github.com/seata/seata-ctl/action/login" @@ -64,20 +67,22 @@ func Execute() { tool.InitLogger() - var address = "" - for _, arg := range os.Args { if arg == "-h" || arg == "--help" || arg == "version" { os.Exit(0) } } var err error + scanner := bufio.NewScanner(os.Stdin) for { if login.Address != "" { - printPrompt(address) + printPrompt(login.Address) } - err = common.ReadArgs(os.Stdin) + err = common.ReadArgsFromScanner(scanner) if err != nil { + if errors.Is(err, io.EOF) { + return + } fmt.Println(err) continue } diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md new file mode 100644 index 0000000..9a7376c --- /dev/null +++ b/docs/troubleshooting.md @@ -0,0 +1,48 @@ + + +# Troubleshooting + +`seata-ctl` provides read-only diagnostics for Seata Server. + +## Diagnose + +Run a quick health sweep: + +```bash +seata-ctl +login --ip 127.0.0.1 --port 7091 --username seata --password seata +diagnose run +``` + +The command checks: + +- server configuration +- TCP connectivity +- login token +- status endpoint +- global transaction query +- global lock query +- optional database connectivity + +Use `--output table|json|yaml` to switch formats. + +## Common failures + +- `server address is not configured`: log in again with the right IP and port. +- `please login`: log in through the REPL first. +- `tcp connectivity failed`: check the Seata Server port and network path. diff --git a/docs/tui.md b/docs/tui.md new file mode 100644 index 0000000..5ecab7e --- /dev/null +++ b/docs/tui.md @@ -0,0 +1,45 @@ + + +# TUI + +Open the diagnostic terminal interface: + +```bash +seata-ctl +login --ip 127.0.0.1 --port 7091 --username seata --password seata +tui +``` + +Pages: + +- `1` or `Tab`: diagnostics +- `2`: global transactions +- `3`: global locks + +Keys: + +- `r`: refresh now +- `a`: toggle auto refresh +- `q` or `Ctrl+C`: quit + +Flags: + +- `--refresh 5s` +- `--page-size 20` +- `--check-db` +- `--db-address host:port` diff --git a/go.mod b/go.mod index 54dcc50..0179ddf 100644 --- a/go.mod +++ b/go.mod @@ -3,12 +3,14 @@ module github.com/seata/seata-ctl go 1.23.1 require ( + github.com/charmbracelet/bubbletea v1.2.4 github.com/elastic/go-elasticsearch/v8 v8.15.0 github.com/guptarohit/asciigraph v0.7.3 github.com/jedib0t/go-pretty/v6 v6.4.7 github.com/olivere/elastic/v7 v7.0.32 github.com/sirupsen/logrus v1.9.3 github.com/spf13/cobra v1.8.1 + github.com/spf13/pflag v1.0.5 github.com/spf13/viper v1.16.0 github.com/stretchr/testify v1.9.0 gopkg.in/yaml.v3 v3.0.1 @@ -19,9 +21,14 @@ require ( ) require ( + github.com/aymanbagabas/go-osc52/v2 v2.0.1 // indirect + github.com/charmbracelet/lipgloss v1.0.0 // indirect + github.com/charmbracelet/x/ansi v0.4.5 // indirect + github.com/charmbracelet/x/term v0.2.1 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/elastic/elastic-transport-go/v8 v8.6.0 // indirect github.com/emicklei/go-restful/v3 v3.11.0 // indirect + github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f // indirect github.com/fsnotify/fsnotify v1.7.0 // indirect github.com/fxamacker/cbor/v2 v2.7.0 // indirect github.com/go-logr/logr v1.4.2 // indirect @@ -40,21 +47,26 @@ require ( github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/josharian/intern v1.0.0 // indirect github.com/json-iterator/go v1.1.12 // indirect + github.com/lucasb-eyer/go-colorful v1.2.0 // indirect github.com/magiconair/properties v1.8.7 // indirect github.com/mailru/easyjson v0.7.7 // indirect - github.com/mattn/go-runewidth v0.0.13 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/mattn/go-localereader v0.0.1 // indirect + github.com/mattn/go-runewidth v0.0.15 // indirect github.com/mitchellh/mapstructure v1.5.0 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/muesli/ansi v0.0.0-20230316100256-276c6243b2f6 // indirect + github.com/muesli/cancelreader v0.2.2 // indirect + github.com/muesli/termenv v0.15.2 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/pelletier/go-toml/v2 v2.0.8 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect - github.com/rivo/uniseg v0.2.0 // indirect + github.com/rivo/uniseg v0.4.7 // indirect github.com/spf13/afero v1.9.5 // indirect github.com/spf13/cast v1.5.1 // indirect github.com/spf13/jwalterweatherman v1.1.0 // indirect - github.com/spf13/pflag v1.0.5 // indirect github.com/subosito/gotenv v1.4.2 // indirect github.com/x448/float16 v0.8.4 // indirect go.opentelemetry.io/otel v1.28.0 // indirect @@ -63,7 +75,8 @@ require ( go.opentelemetry.io/otel/trace v1.28.0 // indirect golang.org/x/net v0.26.0 // indirect golang.org/x/oauth2 v0.21.0 // indirect - golang.org/x/sys v0.21.0 // indirect + golang.org/x/sync v0.9.0 // indirect + golang.org/x/sys v0.27.0 // indirect golang.org/x/term v0.21.0 // indirect golang.org/x/text v0.16.0 // indirect golang.org/x/time v0.3.0 // indirect diff --git a/go.sum b/go.sum index e8c04dd..14ab947 100644 --- a/go.sum +++ b/go.sum @@ -38,7 +38,17 @@ cloud.google.com/go/storage v1.14.0/go.mod h1:GrKmX003DSIwi9o29oFT7YDnHYwZoctc3f dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo= +github.com/aymanbagabas/go-osc52/v2 v2.0.1 h1:HwpRHbFMcZLEVr42D4p7XBqjyuxQH5SMiErDT4WkJ2k= +github.com/aymanbagabas/go-osc52/v2 v2.0.1/go.mod h1:uYgXzlJ7ZpABp8OJ+exZzJJhRNQ2ASbcXHWsFqH8hp8= github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU= +github.com/charmbracelet/bubbletea v1.2.4 h1:KN8aCViA0eps9SCOThb2/XPIlea3ANJLUkv3KnQRNCE= +github.com/charmbracelet/bubbletea v1.2.4/go.mod h1:Qr6fVQw+wX7JkWWkVyXYk/ZUQ92a6XNekLXa3rR18MM= +github.com/charmbracelet/lipgloss v1.0.0 h1:O7VkGDvqEdGi93X+DeqsQ7PKHDgtQfF8j8/O2qFMQNg= +github.com/charmbracelet/lipgloss v1.0.0/go.mod h1:U5fy9Z+C38obMs+T+tJqst9VGzlOYGj4ri9reL3qUlo= +github.com/charmbracelet/x/ansi v0.4.5 h1:LqK4vwBNaXw2AyGIICa5/29Sbdq58GbGdFngSexTdRM= +github.com/charmbracelet/x/ansi v0.4.5/go.mod h1:dk73KoMTT5AX5BsX0KrqhsTqAnhZZoCBjs7dGWp4Ktw= +github.com/charmbracelet/x/term v0.2.1 h1:AQeHeLZ1OqSXhrAWpYUtZyX1T3zVxfpZuEQMIQaGIAQ= +github.com/charmbracelet/x/term v0.2.1/go.mod h1:oQ4enTYFV7QN4m0i9mzHrViD7TQKvNEEkHUMCmsxdUg= github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI= github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI= github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU= @@ -64,6 +74,8 @@ github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1m github.com/envoyproxy/go-control-plane v0.9.7/go.mod h1:cwu0lG7PUMfa9snN8LXBig5ynNVH9qI8YYLbd1fK2po= github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk= github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c= +github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f h1:Y/CXytFA4m6baUTXGLOoWe4PQhGxaX0KpnayAqC48p4= +github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f/go.mod h1:vw97MGsxSvLiUE2X8qFplwetxpGLQrlU1Q9AUEIzCaM= github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw= github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g= github.com/frankban/quicktest v1.14.4 h1:g2rn0vABPOOXmZUj+vbmUp0lPoXEMuhTpIluN0XL9UY= @@ -190,12 +202,19 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/lucasb-eyer/go-colorful v1.2.0 h1:1nnpGOrhyZZuNyfu1QjKiUICQ74+3FNCN69Aj6K7nkY= +github.com/lucasb-eyer/go-colorful v1.2.0/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0= github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY= github.com/magiconair/properties v1.8.7/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0= github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc= -github.com/mattn/go-runewidth v0.0.13 h1:lTGmDsbAYt5DmK6OnoV7EuIF1wEIFAcxld6ypU4OSgU= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/mattn/go-localereader v0.0.1 h1:ygSAOl7ZXTx4RdPYinUpg6W99U8jWvWi9Ye2JC/oIi4= +github.com/mattn/go-localereader v0.0.1/go.mod h1:8fBrzywKY7BI3czFoHkuzRoWE9C+EiG4R1k4Cjx5p88= github.com/mattn/go-runewidth v0.0.13/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w= +github.com/mattn/go-runewidth v0.0.15 h1:UNAjwbU9l54TA3KzvqLGxwWjHmMgBUVhBiTjelZgg3U= +github.com/mattn/go-runewidth v0.0.15/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w= github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY= github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= @@ -203,6 +222,12 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/muesli/ansi v0.0.0-20230316100256-276c6243b2f6 h1:ZK8zHtRHOkbHy6Mmr5D264iyp3TiX5OmNcI5cIARiQI= +github.com/muesli/ansi v0.0.0-20230316100256-276c6243b2f6/go.mod h1:CJlz5H+gyd6CUWT45Oy4q24RdLyn7Md9Vj2/ldJBSIo= +github.com/muesli/cancelreader v0.2.2 h1:3I4Kt4BQjOR54NavqnDogx/MIoWBFa0StPA8ELUXHmA= +github.com/muesli/cancelreader v0.2.2/go.mod h1:3XuTXfFS2VjM+HTLZY9Ak0l6eUKfijIfMUZ4EgX0QYo= +github.com/muesli/termenv v0.15.2 h1:GohcuySI0QmI3wN8Ok9PtKGkgkFIk7y6Vpb5PvrY+Wo= +github.com/muesli/termenv v0.15.2/go.mod h1:Epx+iuz8sNs7mNKhxzH4fWXGNpZwUaJKRS1noLXviQ8= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/olivere/elastic/v7 v7.0.32 h1:R7CXvbu8Eq+WlsLgxmKVKPox0oOwAE/2T9Si5BnvK6E= @@ -221,8 +246,9 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= -github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY= github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc= +github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ= +github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88= github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4= github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8= github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4= @@ -373,6 +399,8 @@ golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.9.0 h1:fEo0HyrW1GIgZdpbhCRO0PkJajUS5H9IFUztCgEo2jQ= +golang.org/x/sync v0.9.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -408,10 +436,12 @@ golang.org/x/sys v0.0.0-20210225134936-a50acf3fe073/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210423185535-09eb48e85fd7/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.21.0 h1:rF+pYz3DAGSQAxAu1CbC7catZg4ebC4UIeIhKxBZvws= -golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.27.0 h1:wBqf8DvsY9Y/2P8gAfPDEYNuS30J4lPHJxXSb/nJZ+s= +golang.org/x/sys v0.27.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.21.0 h1:WVXCp+/EBEHOj53Rvu+7KiT/iElMrO8ACK16SMZ3jaA= golang.org/x/term v0.21.0/go.mod h1:ooXLefLobQVslOqselCNF4SxFAaoS6KujMbsGzSDmX0= diff --git a/seata/auth.go b/seata/auth.go index ee89fef..0f0e68b 100644 --- a/seata/auth.go +++ b/seata/auth.go @@ -25,6 +25,7 @@ import ( "io" "net/http" "strconv" + "strings" ) var auth Auth @@ -60,19 +61,46 @@ func GetAuth() *Auth { } func (auth *Auth) Login() error { + auth.token = "" url := HTTPProtocol + auth.GetAddress() + LoginURL - jsonStr := []byte(fmt.Sprintf(`{"username":"%s","password":"%s"}`, auth.Username, auth.Password)) - resp, err := http.Post(url, "application/json", bytes.NewBuffer(jsonStr)) + jsonStr, err := json.Marshal(map[string]string{ + "username": auth.Username, + "password": auth.Password, + }) + if err != nil { + return err + } + request, err := http.NewRequest(http.MethodPost, url, bytes.NewBuffer(jsonStr)) + if err != nil { + return err + } + request.Header.Set("Content-Type", "application/json") + resp, err := defaultHTTPClient.Do(request) if err != nil { return err } defer resp.Body.Close() - body, _ := io.ReadAll(resp.Body) + body, err := io.ReadAll(resp.Body) + if err != nil { + return err + } + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return fmt.Errorf("login failed: http status %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + } var jsonResp Response err = json.Unmarshal(body, &jsonResp) if err != nil { return err } + if jsonResp.Code != CodeOK { + if jsonResp.Message == "" { + return errors.New("login failed") + } + return errors.New(jsonResp.Message) + } + if jsonResp.Data == "" { + return errors.New("login failed: empty token") + } auth.token = jsonResp.Data return nil } diff --git a/seata/auth_test.go b/seata/auth_test.go new file mode 100644 index 0000000..08f5dd4 --- /dev/null +++ b/seata/auth_test.go @@ -0,0 +1,111 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package seata + +import ( + "encoding/json" + "fmt" + "net" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "strings" + "testing" +) + +func TestLoginUsesJSONAndStoresToken(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var credentials map[string]string + if err := json.NewDecoder(r.Body).Decode(&credentials); err != nil { + t.Fatalf("decode login request: %v", err) + } + if credentials["username"] != `user"name` || credentials["password"] != `p\ss` { + t.Fatalf("credentials = %#v", credentials) + } + fmt.Fprint(w, `{"code":"200","message":"success","data":"new-token","success":true}`) + })) + defer server.Close() + + auth := GetAuth() + old := *auth + defer func() { *auth = old }() + setAuthAddress(t, auth, server.URL) + auth.Username = `user"name` + auth.Password = `p\ss` + if err := auth.Login(); err != nil { + t.Fatalf("Login returned error: %v", err) + } + if token, err := auth.GetToken(); err != nil || token != "new-token" { + t.Fatalf("token = %q, err = %v", token, err) + } +} + +func TestLoginRejectsInvalidResponsesAndClearsToken(t *testing.T) { + tests := []struct { + name string + statusCode int + body string + want string + }{ + {name: "http error", statusCode: http.StatusBadGateway, body: "upstream failed", want: "http status 502"}, + {name: "business error", statusCode: http.StatusOK, body: `{"code":"401","message":"unauthorized"}`, want: "unauthorized"}, + {name: "empty token", statusCode: http.StatusOK, body: `{"code":"200","message":"success","data":""}`, want: "empty token"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(test.statusCode) + fmt.Fprint(w, test.body) + })) + defer server.Close() + + auth := GetAuth() + old := *auth + defer func() { *auth = old }() + setAuthAddress(t, auth, server.URL) + auth.token = "old-token" + + err := auth.Login() + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("Login error = %v, want %q", err, test.want) + } + if _, err = auth.GetToken(); err == nil { + t.Fatal("GetToken succeeded after failed login") + } + }) + } +} + +func setAuthAddress(t *testing.T, auth *Auth, serverURL string) { + t.Helper() + parsed, err := url.Parse(serverURL) + if err != nil { + t.Fatalf("parse server URL: %v", err) + } + host, portString, err := net.SplitHostPort(parsed.Host) + if err != nil { + t.Fatalf("split server address: %v", err) + } + port, err := strconv.Atoi(portString) + if err != nil { + t.Fatalf("parse server port: %v", err) + } + auth.ServerIP = host + auth.ServerPort = port +} diff --git a/seata/config.go b/seata/config.go index 4e9554d..db79d0d 100644 --- a/seata/config.go +++ b/seata/config.go @@ -43,7 +43,7 @@ func GetConfigurations(params []string) (string, error) { return "", err } - resp, err := (&http.Client{}).Do(request) + resp, err := defaultHTTPClient.Do(request) if err != nil { return "", err } @@ -81,7 +81,7 @@ func SetConfiguration(data map[string]string, configType ConfigType) (string, er return "", err } - resp, err := (&http.Client{}).Do(request) + resp, err := defaultHTTPClient.Do(request) if err != nil { return "", err } @@ -115,7 +115,7 @@ func ReloadConfiguration() { request, _ := http.NewRequest("POST", url, nil) request.Header.Set("authorization", token) request.Header.Set("Content-Type", "application/json") - resp, err := (&http.Client{}).Do(request) + resp, err := defaultHTTPClient.Do(request) if err != nil { return } diff --git a/seata/console_client.go b/seata/console_client.go new file mode 100644 index 0000000..3380ef2 --- /dev/null +++ b/seata/console_client.go @@ -0,0 +1,89 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package seata + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strings" +) + +type ConsoleClient struct { + HTTPClient *http.Client +} + +func NewConsoleClient() *ConsoleClient { + return &ConsoleClient{HTTPClient: defaultHTTPClient} +} + +func (client *ConsoleClient) Get(path string, values url.Values, target interface{}) error { + token, err := GetAuth().GetToken() + if err != nil { + return errors.New("please login") + } + + request, err := http.NewRequest(http.MethodGet, consoleURL(path, values), nil) + if err != nil { + return err + } + request.Header.Set("authorization", token) + + httpClient := client.HTTPClient + if httpClient == nil { + httpClient = defaultHTTPClient + } + response, err := httpClient.Do(request) + if err != nil { + return fmt.Errorf("get %s: %w", path, err) + } + defer response.Body.Close() + + body, err := io.ReadAll(response.Body) + if err != nil { + return err + } + if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices { + return fmt.Errorf("get %s: http status %d: %s", path, response.StatusCode, strings.TrimSpace(string(body))) + } + if err = json.Unmarshal(body, target); err != nil { + return fmt.Errorf("decode %s response: %w", path, err) + } + return nil +} + +func consoleURL(path string, values url.Values) string { + rawURL := HTTPProtocol + GetAuth().GetAddress() + path + if len(values) == 0 { + return rawURL + } + return rawURL + "?" + values.Encode() +} + +func checkConsoleCode(code string, message string, fallback string) error { + if code == CodeOK { + return nil + } + if message == "" { + message = fallback + } + return errors.New(message) +} diff --git a/seata/const.go b/seata/const.go index bfb3512..ef4f9dd 100644 --- a/seata/const.go +++ b/seata/const.go @@ -33,6 +33,11 @@ const ( TryBeginTxnURL = TryTxnURL + "/begin" TryCommitTxnURL = TryTxnURL + "/commit" TryRollBackTxnURL = TryTxnURL + "/rollback" + GlobalSessionURL = AdminURL + "/globalSession" + GlobalSessionQueryURL = GlobalSessionURL + "/query" + GlobalLockURL = AdminURL + "/globalLock" + GlobalLockQueryURL = GlobalLockURL + "/query" + GlobalLockCheckURL = GlobalLockURL + "/check" ) const ( diff --git a/seata/diagnose.go b/seata/diagnose.go new file mode 100644 index 0000000..94e7f35 --- /dev/null +++ b/seata/diagnose.go @@ -0,0 +1,255 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package seata + +import ( + "fmt" + "net" + "strings" + "sync" + "time" + + "github.com/jedib0t/go-pretty/v6/table" + "github.com/jedib0t/go-pretty/v6/text" +) + +const ( + DiagnoseStatusPass = "PASS" + DiagnoseStatusWarn = "WARN" + DiagnoseStatusFail = "FAIL" + DiagnoseStatusSkip = "SKIP" +) + +type DiagnoseOptions struct { + CheckDB bool + DBAddress string + TCPTimeout time.Duration + PageNum int + PageSize int + LockPageNum int + LockPageSize int +} + +type DiagnoseReport struct { + Code string `json:"code" yaml:"code"` + Message string `json:"message" yaml:"message"` + Success *bool `json:"success" yaml:"success"` + Data []CheckResult `json:"data" yaml:"data"` +} + +type DiagnoseSnapshot struct { + Report *DiagnoseReport + Transactions *GlobalSessionPageResult + TransactionErr error + Locks *GlobalLockPageResult + LockErr error +} + +type CheckResult struct { + Stage string `json:"stage" yaml:"stage"` + Status string `json:"status" yaml:"status"` + Message string `json:"message" yaml:"message"` + Detail string `json:"detail,omitempty" yaml:"detail,omitempty"` + Evidence string `json:"evidence,omitempty" yaml:"evidence,omitempty"` +} + +func RunDiagnostics(opts DiagnoseOptions) (*DiagnoseReport, error) { + snapshot := CollectDiagnoseSnapshot(opts) + return snapshot.Report, nil +} + +func CollectDiagnoseSnapshot(opts DiagnoseOptions) *DiagnoseSnapshot { + opts = normalizeDiagnoseOptions(opts) + snapshot := &DiagnoseSnapshot{} + results := make([]CheckResult, 0, 7) + passed := true + add := func(result CheckResult) { + if result.Status == DiagnoseStatusFail { + passed = false + } + results = append(results, result) + } + finish := func() { + snapshot.Report = &DiagnoseReport{ + Code: CodeOK, + Message: "diagnostics finished", + Success: boolPtr(passed), + Data: results, + } + } + + address := strings.TrimSpace(GetAuth().GetAddress()) + if address == "" || address == ":0" || address == ":" { + add(CheckResult{Stage: "config", Status: DiagnoseStatusFail, Message: "server address is not configured"}) + addSkipped(add, "config failed", "tcp connectivity", "login token", "status", "transaction query", "global lock query") + add(diagnoseDBResult(opts, DiagnoseStatusSkip, "skipped because config failed")) + finish() + return snapshot + } + add(CheckResult{Stage: "config", Status: DiagnoseStatusPass, Message: "server address is configured", Evidence: address}) + + if err := checkTCP(address, opts.TCPTimeout); err != nil { + add(CheckResult{Stage: "tcp connectivity", Status: DiagnoseStatusFail, Message: err.Error(), Evidence: address}) + addSkipped(add, "tcp connectivity failed", "login token", "status", "transaction query", "global lock query") + add(diagnoseDBResult(opts, DiagnoseStatusSkip, "skipped because tcp connectivity failed")) + finish() + return snapshot + } + add(CheckResult{Stage: "tcp connectivity", Status: DiagnoseStatusPass, Message: "tcp connection succeeded", Evidence: address}) + + token, err := GetAuth().GetToken() + if err != nil { + add(CheckResult{Stage: "login token", Status: DiagnoseStatusFail, Message: err.Error()}) + addSkipped(add, "login token failed", "status", "transaction query", "global lock query") + add(diagnoseDBResult(opts, DiagnoseStatusSkip, "skipped because login token failed")) + finish() + return snapshot + } + add(CheckResult{Stage: "login token", Status: DiagnoseStatusPass, Message: "login token is available", Evidence: maskToken(token)}) + + if response, err := QueryStatus(); err != nil { + add(CheckResult{Stage: "status", Status: DiagnoseStatusFail, Message: err.Error(), Evidence: HealthCheckURL}) + addSkipped(add, "status failed", "transaction query", "global lock query") + add(diagnoseDBResult(opts, DiagnoseStatusSkip, "skipped because status failed")) + finish() + return snapshot + } else { + add(CheckResult{Stage: "status", Status: DiagnoseStatusPass, Message: fmt.Sprintf("found %d node(s)", len(response.Data)), Evidence: HealthCheckURL}) + } + + var ( + transactionResponse *GlobalSessionPageResult + transactionErr error + lockResponse *GlobalLockPageResult + lockErr error + ) + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + transactionResponse, transactionErr = QueryGlobalSessions(GlobalSessionQuery{PageNum: opts.PageNum, PageSize: opts.PageSize}) + }() + go func() { + defer wg.Done() + lockResponse, lockErr = QueryGlobalLocks(GlobalLockQuery{PageNum: opts.LockPageNum, PageSize: opts.LockPageSize}) + }() + wg.Wait() + + if transactionErr != nil { + snapshot.TransactionErr = transactionErr + add(CheckResult{Stage: "transaction query", Status: DiagnoseStatusFail, Message: transactionErr.Error(), Evidence: GlobalSessionQueryURL}) + } else { + snapshot.Transactions = transactionResponse + add(CheckResult{Stage: "transaction query", Status: DiagnoseStatusPass, Message: fmt.Sprintf("found %d global session(s)", len(transactionResponse.Data)), Evidence: GlobalSessionQueryURL}) + } + + if lockErr != nil { + snapshot.LockErr = lockErr + add(CheckResult{Stage: "global lock query", Status: DiagnoseStatusFail, Message: lockErr.Error(), Evidence: GlobalLockQueryURL}) + } else { + snapshot.Locks = lockResponse + add(CheckResult{Stage: "global lock query", Status: DiagnoseStatusPass, Message: fmt.Sprintf("found %d global lock(s)", len(lockResponse.Data)), Evidence: GlobalLockQueryURL}) + } + + add(diagnoseDBResult(opts, DiagnoseStatusSkip, "database/schema check is optional in this build")) + finish() + return snapshot +} + +func normalizeDiagnoseOptions(opts DiagnoseOptions) DiagnoseOptions { + if opts.TCPTimeout <= 0 { + opts.TCPTimeout = 3 * time.Second + } + if opts.PageNum <= 0 { + opts.PageNum = 1 + } + if opts.PageSize <= 0 { + opts.PageSize = 1 + } + if opts.LockPageNum <= 0 { + opts.LockPageNum = 1 + } + if opts.LockPageSize <= 0 { + opts.LockPageSize = 1 + } + return opts +} + +func addSkipped(add func(CheckResult), reason string, stages ...string) { + for _, stage := range stages { + add(CheckResult{Stage: stage, Status: DiagnoseStatusSkip, Message: "skipped because " + reason}) + } +} + +func diagnoseDBResult(opts DiagnoseOptions, fallbackStatus string, fallbackMessage string) CheckResult { + if !opts.CheckDB { + return CheckResult{Stage: "db/schema", Status: DiagnoseStatusSkip, Message: "database/schema check not requested"} + } + if strings.TrimSpace(opts.DBAddress) == "" { + return CheckResult{Stage: "db/schema", Status: fallbackStatus, Message: fallbackMessage} + } + if err := checkTCP(opts.DBAddress, opts.TCPTimeout); err != nil { + return CheckResult{Stage: "db/schema", Status: DiagnoseStatusFail, Message: err.Error(), Evidence: opts.DBAddress} + } + return CheckResult{Stage: "db/schema", Status: DiagnoseStatusPass, Message: "database endpoint is reachable", Evidence: opts.DBAddress} +} + +func checkTCP(address string, timeout time.Duration) error { + conn, err := net.DialTimeout("tcp", address, timeout) + if err != nil { + return err + } + return conn.Close() +} + +func FormatDiagnoseReport(response *DiagnoseReport, output string) (string, error) { + output, err := NormalizeOutput(output) + if err != nil { + return "", err + } + switch output { + case OutputTable: + return FormatDiagnoseTable(response.Data), nil + default: + return FormatStructuredOutput(response, output) + } +} + +func FormatDiagnoseTable(results []CheckResult) string { + t := table.NewWriter() + t.Style().Format.Header = text.FormatDefault + t.AppendHeader(table.Row{"stage", "status", "message", "detail", "evidence"}) + for _, result := range results { + t.AppendRow(table.Row{result.Stage, result.Status, result.Message, result.Detail, result.Evidence}) + } + return t.Render() +} + +func maskToken(token string) string { + if token == "" { + return "" + } + if len(token) <= 8 { + return token + } + return token[:4] + "..." + token[len(token)-4:] +} + +func boolPtr(value bool) *bool { + return &value +} diff --git a/seata/diagnose_test.go b/seata/diagnose_test.go new file mode 100644 index 0000000..ae27e32 --- /dev/null +++ b/seata/diagnose_test.go @@ -0,0 +1,203 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package seata + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "testing" + "time" + + yaml "gopkg.in/yaml.v3" +) + +func TestRunDiagnostics(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("authorization"); got != "test-token" { + t.Fatalf("authorization = %q, want test-token", got) + } + switch r.URL.Path { + case HealthCheckURL: + fmt.Fprint(w, `{"code":"200","message":"success","success":true,"data":[{"type":"nacos","address":"127.0.0.1:7091","status":"ok"}]}`) + case GlobalSessionQueryURL: + if got := r.URL.Query().Get("pageNum"); got != "1" { + t.Fatalf("pageNum = %q, want 1", got) + } + if got := r.URL.Query().Get("pageSize"); got != "1" { + t.Fatalf("pageSize = %q, want 1", got) + } + fmt.Fprint(w, `{"code":"200","message":"success","success":true,"pageSize":1,"pageNum":1,"total":1,"pages":1,"data":[{"xid":"xid-1","transactionId":"1001","status":1,"applicationId":"order-service","transactionServiceGroup":"default_tx_group","transactionName":"createOrder","timeout":60000,"beginTime":1710000000000}]}`) + case GlobalLockQueryURL: + if got := r.URL.Query().Get("pageNum"); got != "1" { + t.Fatalf("lock pageNum = %q, want 1", got) + } + if got := r.URL.Query().Get("pageSize"); got != "1" { + t.Fatalf("lock pageSize = %q, want 1", got) + } + fmt.Fprint(w, `{"code":"200","message":"success","success":true,"pageSize":1,"pageNum":1,"total":1,"pages":1,"data":[{"xid":"xid-1","transactionId":"1001","branchId":"2001","resourceId":"jdbc:mysql://127.0.0.1:3306/seata","tableName":"account","pk":"1","rowKey":"1","vgroup":"default_tx_group"}]}`) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + defer setTestAuth(t, server.URL, "test-token")() + + report, err := RunDiagnostics(DiagnoseOptions{ + TCPTimeout: time.Second, + PageNum: 1, + PageSize: 1, + LockPageNum: 1, + LockPageSize: 1, + }) + if err != nil { + t.Fatalf("RunDiagnostics returned error: %v", err) + } + if report.Success == nil || !*report.Success { + t.Fatalf("unexpected success flag: %+v", report.Success) + } + wantStages := []string{"config", "tcp connectivity", "login token", "status", "transaction query", "global lock query", "db/schema"} + if len(report.Data) != len(wantStages) { + t.Fatalf("len(report.Data) = %d, want %d", len(report.Data), len(wantStages)) + } + for i, want := range wantStages { + if report.Data[i].Stage != want { + t.Fatalf("stage[%d] = %q, want %q", i, report.Data[i].Stage, want) + } + } + for _, stage := range report.Data[:6] { + if stage.Status != DiagnoseStatusPass { + t.Fatalf("stage %+v not pass", stage) + } + } + if report.Data[6].Status != DiagnoseStatusSkip { + t.Fatalf("db stage = %+v, want skip", report.Data[6]) + } + + tableOutput, err := FormatDiagnoseReport(report, OutputTable) + if err != nil { + t.Fatalf("table output error: %v", err) + } + for _, want := range []string{"config", "tcp connectivity", "transaction query", "global lock query", "db/schema"} { + if !strings.Contains(tableOutput, want) { + t.Fatalf("table output %q does not contain %q", tableOutput, want) + } + } + + jsonOutput, err := FormatDiagnoseReport(report, OutputJSON) + if err != nil { + t.Fatalf("json output error: %v", err) + } + var jsonResult DiagnoseReport + if err = json.Unmarshal([]byte(jsonOutput), &jsonResult); err != nil { + t.Fatalf("unmarshal json output: %v", err) + } + if !reflect.DeepEqual(jsonResult.Data, report.Data) { + t.Fatalf("unexpected json result: %+v", jsonResult) + } + + yamlOutput, err := FormatDiagnoseReport(report, OutputYAML) + if err != nil { + t.Fatalf("yaml output error: %v", err) + } + var yamlResult DiagnoseReport + if err = yaml.Unmarshal([]byte(yamlOutput), &yamlResult); err != nil { + t.Fatalf("unmarshal yaml output: %v", err) + } + if !reflect.DeepEqual(yamlResult.Data, report.Data) { + t.Fatalf("unexpected yaml result: %+v", yamlResult) + } +} + +func TestCollectDiagnoseSnapshotStopsAfterStatusFailure(t *testing.T) { + var transactionQueries, lockQueries int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case HealthCheckURL: + fmt.Fprint(w, `{"code":"500","message":"status unavailable","success":false}`) + case GlobalSessionQueryURL: + transactionQueries++ + case GlobalLockQueryURL: + lockQueries++ + default: + http.NotFound(w, r) + } + })) + defer server.Close() + defer setTestAuth(t, server.URL, "test-token")() + + snapshot := CollectDiagnoseSnapshot(DiagnoseOptions{TCPTimeout: time.Second}) + if snapshot.Report == nil || snapshot.Report.Success == nil || *snapshot.Report.Success { + t.Fatalf("unexpected report: %+v", snapshot.Report) + } + if transactionQueries != 0 || lockQueries != 0 { + t.Fatalf("unexpected downstream queries: transactions=%d locks=%d", transactionQueries, lockQueries) + } + wantStages := []string{"config", "tcp connectivity", "login token", "status", "transaction query", "global lock query", "db/schema"} + if len(snapshot.Report.Data) != len(wantStages) { + t.Fatalf("len(report.Data) = %d, want %d", len(snapshot.Report.Data), len(wantStages)) + } + for i, want := range wantStages { + if snapshot.Report.Data[i].Stage != want { + t.Fatalf("stage[%d] = %q, want %q", i, snapshot.Report.Data[i].Stage, want) + } + } + if snapshot.Report.Data[4].Status != DiagnoseStatusSkip || snapshot.Report.Data[5].Status != DiagnoseStatusSkip { + t.Fatalf("downstream stages = %+v, want skipped", snapshot.Report.Data[4:6]) + } +} + +func TestCollectDiagnoseSnapshotKeepsIndependentQueriesAvailable(t *testing.T) { + var transactionQueries, lockQueries int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case HealthCheckURL: + fmt.Fprint(w, `{"code":"200","message":"success","success":true,"data":[]}`) + case GlobalSessionQueryURL: + transactionQueries++ + http.Error(w, "transaction query unavailable", http.StatusBadGateway) + case GlobalLockQueryURL: + lockQueries++ + fmt.Fprint(w, `{"code":"200","message":"success","success":true,"data":[]}`) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + defer setTestAuth(t, server.URL, "test-token")() + + snapshot := CollectDiagnoseSnapshot(DiagnoseOptions{TCPTimeout: time.Second}) + if transactionQueries != 1 || lockQueries != 1 { + t.Fatalf("queries = transactions:%d locks:%d, want one of each", transactionQueries, lockQueries) + } + if snapshot.Transactions != nil || snapshot.TransactionErr == nil { + t.Fatalf("transaction snapshot = %+v, error = %v, want query error", snapshot.Transactions, snapshot.TransactionErr) + } + if snapshot.Locks == nil || snapshot.LockErr != nil { + t.Fatalf("lock snapshot = %+v, error = %v, want successful query", snapshot.Locks, snapshot.LockErr) + } + if snapshot.Report == nil || len(snapshot.Report.Data) != 7 { + t.Fatalf("report = %+v, want all stages", snapshot.Report) + } + if snapshot.Report.Data[4].Status != DiagnoseStatusFail || snapshot.Report.Data[5].Status != DiagnoseStatusPass { + t.Fatalf("query stages = %+v, want fail/pass", snapshot.Report.Data[4:6]) + } +} diff --git a/seata/http_client.go b/seata/http_client.go new file mode 100644 index 0000000..8f26aea --- /dev/null +++ b/seata/http_client.go @@ -0,0 +1,26 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package seata + +import ( + "net/http" + "time" +) + +var defaultHTTPClient = &http.Client{Timeout: 10 * time.Second} + diff --git a/seata/lock.go b/seata/lock.go new file mode 100644 index 0000000..6a3efa2 --- /dev/null +++ b/seata/lock.go @@ -0,0 +1,203 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package seata + +import ( + "errors" + "net/url" + "strconv" + + "github.com/jedib0t/go-pretty/v6/table" + "github.com/jedib0t/go-pretty/v6/text" +) + +type GlobalLockQuery struct { + XID string + TableName string + TransactionID string + BranchID string + PK string + ResourceID string + PageNum int + PageSize int + TimeStart *int64 + TimeEnd *int64 +} + +type GlobalLockPageResult struct { + Code string `json:"code" yaml:"code"` + Message string `json:"message" yaml:"message"` + Success *bool `json:"success" yaml:"success"` + PageSize *int `json:"pageSize" yaml:"pageSize"` + PageNum *int `json:"pageNum" yaml:"pageNum"` + CurrPage *int `json:"currPage" yaml:"currPage"` + Total *int `json:"total" yaml:"total"` + Pages *int `json:"pages" yaml:"pages"` + Data []GlobalLock `json:"data" yaml:"data"` +} + +type GlobalLock struct { + XID *string `json:"xid" yaml:"xid"` + TransactionID *string `json:"transactionId" yaml:"transactionId"` + BranchID *string `json:"branchId" yaml:"branchId"` + ResourceID *string `json:"resourceId" yaml:"resourceId"` + TableName *string `json:"tableName" yaml:"tableName"` + PK *string `json:"pk" yaml:"pk"` + RowKey *string `json:"rowKey" yaml:"rowKey"` + Vgroup *string `json:"vgroup" yaml:"vgroup"` + GmtCreate *int64 `json:"gmtCreate" yaml:"gmtCreate"` + GmtModified *int64 `json:"gmtModified" yaml:"gmtModified"` +} + +type GlobalLockCheckResult struct { + Code string `json:"code" yaml:"code"` + Message string `json:"message" yaml:"message"` + Success *bool `json:"success" yaml:"success"` + Data *bool `json:"data" yaml:"data"` +} + +func QueryGlobalLocks(query GlobalLockQuery) (*GlobalLockPageResult, error) { + return queryGlobalLocks(NewConsoleClient(), query) +} + +func queryGlobalLocks(client *ConsoleClient, query GlobalLockQuery) (*GlobalLockPageResult, error) { + if query.PageNum <= 0 { + return nil, errors.New("page-num must be greater than 0") + } + if query.PageSize <= 0 { + return nil, errors.New("page-size must be greater than 0") + } + + var response GlobalLockPageResult + if err := client.Get(GlobalLockQueryURL, query.values(), &response); err != nil { + return nil, err + } + if err := checkConsoleCode(response.Code, response.Message, "query global locks failed"); err != nil { + return nil, err + } + return &response, nil +} + +func CheckGlobalLock(xid string, branchID string) (*GlobalLockCheckResult, error) { + return checkGlobalLock(NewConsoleClient(), xid, branchID) +} + +func checkGlobalLock(client *ConsoleClient, xid string, branchID string) (*GlobalLockCheckResult, error) { + if xid == "" { + return nil, errors.New("xid is required") + } + if branchID == "" { + return nil, errors.New("branch-id is required") + } + + values := url.Values{} + values.Set("xid", xid) + values.Set("branchId", branchID) + var response GlobalLockCheckResult + if err := client.Get(GlobalLockCheckURL, values, &response); err != nil { + return nil, err + } + if err := checkConsoleCode(response.Code, response.Message, "check global lock failed"); err != nil { + return nil, err + } + return &response, nil +} + +func (query GlobalLockQuery) values() url.Values { + values := url.Values{} + values.Set("pageNum", strconv.Itoa(query.PageNum)) + values.Set("pageSize", strconv.Itoa(query.PageSize)) + if query.XID != "" { + values.Set("xid", query.XID) + } + if query.TableName != "" { + values.Set("tableName", query.TableName) + } + if query.TransactionID != "" { + values.Set("transactionId", query.TransactionID) + } + if query.BranchID != "" { + values.Set("branchId", query.BranchID) + } + if query.PK != "" { + values.Set("pk", query.PK) + } + if query.ResourceID != "" { + values.Set("resourceId", query.ResourceID) + } + if query.TimeStart != nil { + values.Set("timeStart", strconv.FormatInt(*query.TimeStart, 10)) + } + if query.TimeEnd != nil { + values.Set("timeEnd", strconv.FormatInt(*query.TimeEnd, 10)) + } + return values +} + +func FormatGlobalLockPage(response *GlobalLockPageResult, output string) (string, error) { + output, err := NormalizeOutput(output) + if err != nil { + return "", err + } + switch output { + case OutputTable: + return FormatGlobalLockTable(response.Data), nil + default: + return FormatStructuredOutput(response, output) + } +} + +func FormatGlobalLockCheck(response *GlobalLockCheckResult, output string) (string, error) { + output, err := NormalizeOutput(output) + if err != nil { + return "", err + } + switch output { + case OutputTable: + locked := "" + if response.Data != nil { + locked = strconv.FormatBool(*response.Data) + } + t := table.NewWriter() + t.Style().Format.Header = text.FormatDefault + t.AppendHeader(table.Row{"locked", "message"}) + t.AppendRow(table.Row{locked, response.Message}) + return t.Render(), nil + default: + return FormatStructuredOutput(response, output) + } +} + +func FormatGlobalLockTable(locks []GlobalLock) string { + t := table.NewWriter() + t.Style().Format.Header = text.FormatDefault + t.AppendHeader(table.Row{"xid", "transaction_id", "branch_id", "resource_id", "table_name", "pk", "row_key", "vgroup"}) + for _, lock := range locks { + t.AppendRow(table.Row{ + stringValue(lock.XID), + stringValue(lock.TransactionID), + stringValue(lock.BranchID), + stringValue(lock.ResourceID), + stringValue(lock.TableName), + stringValue(lock.PK), + stringValue(lock.RowKey), + stringValue(lock.Vgroup), + }) + } + return t.Render() +} diff --git a/seata/lock_test.go b/seata/lock_test.go new file mode 100644 index 0000000..a1f3fb7 --- /dev/null +++ b/seata/lock_test.go @@ -0,0 +1,223 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package seata + +import ( + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "testing" + + yaml "gopkg.in/yaml.v3" +) + +func TestQueryGlobalLocksBuildsRequest(t *testing.T) { + timeStart := int64(1710000000000) + timeEnd := int64(1710003600000) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + t.Fatalf("method = %s, want %s", r.Method, http.MethodGet) + } + if r.URL.Path != GlobalLockQueryURL { + t.Fatalf("path = %s, want %s", r.URL.Path, GlobalLockQueryURL) + } + if got := r.Header.Get("authorization"); got != "test-token" { + t.Fatalf("authorization = %q, want test-token", got) + } + wantQuery := map[string]string{ + "xid": "xid-1", + "tableName": "account", + "transactionId": "1001", + "branchId": "2001", + "pk": "1", + "resourceId": "jdbc:mysql://127.0.0.1:3306/seata", + "pageNum": "2", + "pageSize": "10", + "timeStart": "1710000000000", + "timeEnd": "1710003600000", + } + for key, want := range wantQuery { + if got := r.URL.Query().Get(key); got != want { + t.Fatalf("query[%s] = %q, want %q", key, got, want) + } + } + fmt.Fprint(w, `{"code":"200","message":"success","success":true,"pageSize":10,"pageNum":2,"total":1,"pages":1,"data":[{"xid":"xid-1","transactionId":"1001","branchId":"2001","resourceId":"jdbc:mysql://127.0.0.1:3306/seata","tableName":"account","pk":"1","rowKey":"1","vgroup":"default_tx_group"}]}`) + })) + defer server.Close() + defer setTestAuth(t, server.URL, "test-token")() + + response, err := queryGlobalLocks(&ConsoleClient{HTTPClient: server.Client()}, GlobalLockQuery{ + XID: "xid-1", + TableName: "account", + TransactionID: "1001", + BranchID: "2001", + PK: "1", + ResourceID: "jdbc:mysql://127.0.0.1:3306/seata", + PageNum: 2, + PageSize: 10, + TimeStart: &timeStart, + TimeEnd: &timeEnd, + }) + if err != nil { + t.Fatalf("queryGlobalLocks returned error: %v", err) + } + if len(response.Data) != 1 || stringValue(response.Data[0].TableName) != "account" { + t.Fatalf("unexpected response: %+v", response) + } +} + +func TestCheckGlobalLockBuildsRequest(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != GlobalLockCheckURL { + t.Fatalf("path = %s, want %s", r.URL.Path, GlobalLockCheckURL) + } + if got := r.URL.Query().Get("xid"); got != "xid-1" { + t.Fatalf("xid = %q, want xid-1", got) + } + if got := r.URL.Query().Get("branchId"); got != "2001" { + t.Fatalf("branchId = %q, want 2001", got) + } + fmt.Fprint(w, `{"code":"200","message":"success","success":true,"data":true}`) + })) + defer server.Close() + defer setTestAuth(t, server.URL, "test-token")() + + response, err := checkGlobalLock(&ConsoleClient{HTTPClient: server.Client()}, "xid-1", "2001") + if err != nil { + t.Fatalf("checkGlobalLock returned error: %v", err) + } + if response.Data == nil || !*response.Data { + t.Fatalf("unexpected response: %+v", response) + } +} + +func TestQueryGlobalLocksErrors(t *testing.T) { + tests := []struct { + name string + body string + want string + }{ + {name: "server code", body: `{"code":"500","message":"boom"}`, want: "boom"}, + {name: "invalid json", body: `{`, want: "decode /api/v1/console/globalLock/query response"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + fmt.Fprint(w, tt.body) + })) + defer server.Close() + defer setTestAuth(t, server.URL, "test-token")() + + _, err := queryGlobalLocks(&ConsoleClient{HTTPClient: server.Client()}, GlobalLockQuery{PageNum: 1, PageSize: 20}) + if err == nil || !strings.Contains(err.Error(), tt.want) { + t.Fatalf("error = %v, want containing %q", err, tt.want) + } + }) + } +} + +func TestFormatGlobalLockPageAndCheck(t *testing.T) { + success := true + response := &GlobalLockPageResult{ + Code: CodeOK, + Message: "success", + Success: &success, + PageNum: intPtr(1), + PageSize: intPtr(20), + Total: intPtr(1), + Pages: intPtr(1), + Data: []GlobalLock{{ + XID: stringPtr("xid-1"), + TransactionID: stringPtr("1001"), + BranchID: stringPtr("2001"), + ResourceID: stringPtr("jdbc:mysql://127.0.0.1:3306/seata"), + TableName: stringPtr("account"), + PK: stringPtr("1"), + RowKey: stringPtr("1"), + Vgroup: stringPtr("default_tx_group"), + }}, + } + + tableOutput, err := FormatGlobalLockPage(response, OutputTable) + if err != nil { + t.Fatalf("table output error: %v", err) + } + for _, want := range []string{"xid", "transaction_id", "branch_id", "resource_id", "table_name", "pk", "row_key", "vgroup"} { + if !strings.Contains(tableOutput, want) { + t.Fatalf("table output %q does not contain %q", tableOutput, want) + } + } + + jsonOutput, err := FormatGlobalLockPage(response, OutputJSON) + if err != nil { + t.Fatalf("json output error: %v", err) + } + var jsonResult GlobalLockPageResult + if err = json.Unmarshal([]byte(jsonOutput), &jsonResult); err != nil { + t.Fatalf("unmarshal json output: %v", err) + } + if !reflect.DeepEqual(&jsonResult, response) { + t.Fatalf("unexpected json result: %+v", jsonResult) + } + + yamlOutput, err := FormatGlobalLockPage(response, OutputYAML) + if err != nil { + t.Fatalf("yaml output error: %v", err) + } + var yamlResult GlobalLockPageResult + if err = yaml.Unmarshal([]byte(yamlOutput), &yamlResult); err != nil { + t.Fatalf("unmarshal yaml output: %v", err) + } + if len(yamlResult.Data) != 1 || stringValue(yamlResult.Data[0].TableName) != "account" { + t.Fatalf("unexpected yaml result: %+v", yamlResult) + } + + checkOutput, err := FormatGlobalLockCheck(&GlobalLockCheckResult{Code: CodeOK, Message: "locked", Data: &success}, OutputTable) + if err != nil { + t.Fatalf("check table output error: %v", err) + } + for _, want := range []string{"locked", "message"} { + if !strings.Contains(checkOutput, want) { + t.Fatalf("check output %q does not contain %q", checkOutput, want) + } + } + + _, err = FormatGlobalLockPage(response, "xml") + if err == nil || !strings.Contains(err.Error(), "unsupported output format") { + t.Fatalf("error = %v, want unsupported output format", err) + } +} + +func TestCheckGlobalLockValidation(t *testing.T) { + client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + return nil, errors.New("should not be called") + })} + _, err := checkGlobalLock(&ConsoleClient{HTTPClient: client}, "", "2001") + if err == nil || !strings.Contains(err.Error(), "xid is required") { + t.Fatalf("error = %v, want xid is required", err) + } + _, err = checkGlobalLock(&ConsoleClient{HTTPClient: client}, "xid-1", "") + if err == nil || !strings.Contains(err.Error(), "branch-id is required") { + t.Fatalf("error = %v, want branch-id is required", err) + } +} diff --git a/seata/render.go b/seata/render.go new file mode 100644 index 0000000..9515b73 --- /dev/null +++ b/seata/render.go @@ -0,0 +1,76 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package seata + +import ( + "encoding/json" + "fmt" + "strings" + + yaml "gopkg.in/yaml.v3" +) + +const ( + OutputTable = "table" + OutputJSON = "json" + OutputYAML = "yaml" +) + +func NormalizeOutput(output string) (string, error) { + output = strings.ToLower(output) + switch output { + case OutputTable, OutputJSON, OutputYAML: + return output, nil + default: + return "", fmt.Errorf("unsupported output format %q", output) + } +} + +func FormatStructuredOutput(value interface{}, output string) (string, error) { + switch output { + case OutputJSON: + data, err := json.MarshalIndent(value, "", " ") + return string(data), err + case OutputYAML: + data, err := yaml.Marshal(value) + return string(data), err + default: + return "", fmt.Errorf("unsupported structured output format %q", output) + } +} + +func stringValue(value *string) string { + if value == nil { + return "" + } + return *value +} + +func intValue(value *int) interface{} { + if value == nil { + return "" + } + return *value +} + +func int64Value(value *int64) interface{} { + if value == nil { + return "" + } + return *value +} diff --git a/seata/status.go b/seata/status.go index c0de41c..05a22d7 100644 --- a/seata/status.go +++ b/seata/status.go @@ -18,62 +18,72 @@ package seata import ( - "encoding/json" "fmt" - "io" - "net/http" "os" "github.com/jedib0t/go-pretty/v6/table" + "github.com/jedib0t/go-pretty/v6/text" ) type NodeStatusResponse struct { BaseResponse - Data []NodeStatus `json:"data"` + Data []NodeStatus `json:"data" yaml:"data"` } type NodeStatus struct { - Address string `json:"address"` - Status string `json:"status"` - Type string `json:"type"` + Address string `json:"address" yaml:"address"` + Status string `json:"status" yaml:"status"` + Type string `json:"type" yaml:"type"` } -func GetStatus() { - url := HTTPProtocol + GetAuth().GetAddress() + HealthCheckURL - token, err := GetAuth().GetToken() - if err != nil { - fmt.Println("Please login again!") - os.Exit(0) +func QueryStatus() (*NodeStatusResponse, error) { + return queryStatus(NewConsoleClient()) +} + +func queryStatus(client *ConsoleClient) (*NodeStatusResponse, error) { + var response NodeStatusResponse + if err := client.Get(HealthCheckURL, nil, &response); err != nil { + return nil, err + } + if err := checkConsoleCode(response.Code, response.Message, "query status failed"); err != nil { + return nil, err } - request, _ := http.NewRequest("GET", url, nil) - request.Header.Set("authorization", token) - resp, err := (&http.Client{}).Do(request) + return &response, nil +} + +func GetStatus() { + response, err := QueryStatus() if err != nil { + fmt.Println(err) return } - defer resp.Body.Close() - body, err := io.ReadAll(resp.Body) + result, err := FormatNodeStatusResponse(response, OutputTable) if err != nil { fmt.Println(err) + return } + fmt.Fprintln(os.Stdout, result) +} - var response NodeStatusResponse - err = json.Unmarshal(body, &response) +func FormatNodeStatusResponse(response *NodeStatusResponse, output string) (string, error) { + output, err := NormalizeOutput(output) if err != nil { - fmt.Println(err) + return "", err } - - if response.Code != "200" { - fmt.Println(response.Message) + switch output { + case OutputTable: + return FormatNodeStatusTable(response.Data), nil + default: + return FormatStructuredOutput(response, output) } +} +func FormatNodeStatusTable(statuses []NodeStatus) string { t := table.NewWriter() - header := table.Row{"type", "address", "status"} - t.AppendHeader(header) - for _, data := range response.Data { - row := table.Row{data.Type, data.Address, data.Status} - t.AppendRow(row) + t.Style().Format.Header = text.FormatDefault + t.AppendHeader(table.Row{"type", "address", "status"}) + for _, status := range statuses { + t.AppendRow(table.Row{status.Type, status.Address, status.Status}) } - fmt.Println(t.Render()) - t.Style() + return t.Render() } diff --git a/seata/status_test.go b/seata/status_test.go new file mode 100644 index 0000000..3ab3822 --- /dev/null +++ b/seata/status_test.go @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package seata + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + + yaml "gopkg.in/yaml.v3" +) + +func TestQueryStatusBuildsRequest(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + t.Fatalf("method = %s, want %s", r.Method, http.MethodGet) + } + if r.URL.Path != HealthCheckURL { + t.Fatalf("path = %s, want %s", r.URL.Path, HealthCheckURL) + } + if got := r.Header.Get("authorization"); got != "test-token" { + t.Fatalf("authorization = %q, want test-token", got) + } + fmt.Fprint(w, `{"code":"200","message":"success","success":true,"data":[{"type":"nacos","address":"127.0.0.1:7091","status":"ok"}]}`) + })) + defer server.Close() + defer setTestAuth(t, server.URL, "test-token")() + + response, err := queryStatus(&ConsoleClient{HTTPClient: server.Client()}) + if err != nil { + t.Fatalf("queryStatus returned error: %v", err) + } + if len(response.Data) != 1 || response.Data[0].Type != "nacos" || response.Data[0].Status != "ok" { + t.Fatalf("unexpected response: %+v", response) + } +} + +func TestFormatNodeStatusResponse(t *testing.T) { + response := &NodeStatusResponse{ + BaseResponse: BaseResponse{Code: CodeOK, Message: "success", Success: true}, + Data: []NodeStatus{{Type: "nacos", Address: "127.0.0.1:7091", Status: "ok"}}, + } + + tableOutput, err := FormatNodeStatusResponse(response, OutputTable) + if err != nil { + t.Fatalf("table output error: %v", err) + } + for _, want := range []string{"type", "address", "status", "nacos", "127.0.0.1:7091", "ok"} { + if !strings.Contains(tableOutput, want) { + t.Fatalf("table output %q does not contain %q", tableOutput, want) + } + } + + jsonOutput, err := FormatNodeStatusResponse(response, OutputJSON) + if err != nil { + t.Fatalf("json output error: %v", err) + } + var jsonResult NodeStatusResponse + if err = json.Unmarshal([]byte(jsonOutput), &jsonResult); err != nil { + t.Fatalf("unmarshal json output: %v", err) + } + if len(jsonResult.Data) != 1 || jsonResult.Data[0].Address != "127.0.0.1:7091" { + t.Fatalf("unexpected json result: %+v", jsonResult) + } + + yamlOutput, err := FormatNodeStatusResponse(response, OutputYAML) + if err != nil { + t.Fatalf("yaml output error: %v", err) + } + var yamlResult NodeStatusResponse + if err = yaml.Unmarshal([]byte(yamlOutput), &yamlResult); err != nil { + t.Fatalf("unmarshal yaml output: %v", err) + } + if len(yamlResult.Data) != 1 || yamlResult.Data[0].Type != "nacos" { + t.Fatalf("unexpected yaml result: %+v", yamlResult) + } + + _, err = FormatNodeStatusResponse(response, "xml") + if err == nil || !strings.Contains(err.Error(), "unsupported output format") { + t.Fatalf("error = %v, want unsupported output format", err) + } +} diff --git a/seata/transaction.go b/seata/transaction.go new file mode 100644 index 0000000..f9fc47f --- /dev/null +++ b/seata/transaction.go @@ -0,0 +1,237 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package seata + +import ( + "errors" + "fmt" + "net/url" + "strconv" + + "github.com/jedib0t/go-pretty/v6/table" + "github.com/jedib0t/go-pretty/v6/text" +) + +type GlobalSessionQuery struct { + XID string + ApplicationID string + Status *int + TransactionName string + Vgroup string + WithBranch *bool + PageNum int + PageSize int + TimeStart *int64 + TimeEnd *int64 +} + +type GlobalSessionPageResult struct { + Code string `json:"code" yaml:"code"` + Message string `json:"message" yaml:"message"` + Success *bool `json:"success" yaml:"success"` + PageSize *int `json:"pageSize" yaml:"pageSize"` + PageNum *int `json:"pageNum" yaml:"pageNum"` + CurrPage *int `json:"currPage" yaml:"currPage"` + Total *int `json:"total" yaml:"total"` + Pages *int `json:"pages" yaml:"pages"` + Data []GlobalSession `json:"data" yaml:"data"` +} + +type GlobalSession struct { + XID *string `json:"xid" yaml:"xid"` + TransactionID *string `json:"transactionId" yaml:"transactionId"` + Status *int `json:"status" yaml:"status"` + ApplicationID *string `json:"applicationId" yaml:"applicationId"` + TransactionServiceGroup *string `json:"transactionServiceGroup" yaml:"transactionServiceGroup"` + TransactionName *string `json:"transactionName" yaml:"transactionName"` + Timeout *int64 `json:"timeout" yaml:"timeout"` + BeginTime *int64 `json:"beginTime" yaml:"beginTime"` + ApplicationData *string `json:"applicationData" yaml:"applicationData"` + GmtCreate *int64 `json:"gmtCreate" yaml:"gmtCreate"` + GmtModified *int64 `json:"gmtModified" yaml:"gmtModified"` + BranchSessionVOs *[]BranchSession `json:"branchSessionVOs" yaml:"branchSessionVOs"` +} + +type BranchSession struct { + XID *string `json:"xid" yaml:"xid"` + TransactionID *string `json:"transactionId" yaml:"transactionId"` + BranchID *string `json:"branchId" yaml:"branchId"` + ResourceGroupID *string `json:"resourceGroupId" yaml:"resourceGroupId"` + ResourceID *string `json:"resourceId" yaml:"resourceId"` + BranchType *string `json:"branchType" yaml:"branchType"` + Status *int `json:"status" yaml:"status"` + ClientID *string `json:"clientId" yaml:"clientId"` + ApplicationData *string `json:"applicationData" yaml:"applicationData"` + GmtCreate *int64 `json:"gmtCreate" yaml:"gmtCreate"` + GmtModified *int64 `json:"gmtModified" yaml:"gmtModified"` +} + +func QueryGlobalSessions(query GlobalSessionQuery) (*GlobalSessionPageResult, error) { + return queryGlobalSessions(NewConsoleClient(), query) +} + +func queryGlobalSessions(client *ConsoleClient, query GlobalSessionQuery) (*GlobalSessionPageResult, error) { + if query.PageNum <= 0 { + return nil, errors.New("page-num must be greater than 0") + } + if query.PageSize <= 0 { + return nil, errors.New("page-size must be greater than 0") + } + + var response GlobalSessionPageResult + if err := client.Get(GlobalSessionQueryURL, query.values(), &response); err != nil { + return nil, err + } + if err := checkConsoleCode(response.Code, response.Message, "query global sessions failed"); err != nil { + return nil, err + } + + return &response, nil +} + +func QueryGlobalSessionByXID(xid string) (*GlobalSession, error) { + withBranch := true + response, err := QueryGlobalSessions(GlobalSessionQuery{ + XID: xid, + WithBranch: &withBranch, + PageNum: 1, + PageSize: 1, + }) + if err != nil { + return nil, err + } + if len(response.Data) == 0 { + return nil, fmt.Errorf("global session %q not found", xid) + } + return &response.Data[0], nil +} + +func (query GlobalSessionQuery) values() url.Values { + values := url.Values{} + values.Set("pageNum", strconv.Itoa(query.PageNum)) + values.Set("pageSize", strconv.Itoa(query.PageSize)) + if query.XID != "" { + values.Set("xid", query.XID) + } + if query.ApplicationID != "" { + values.Set("applicationId", query.ApplicationID) + } + if query.Status != nil { + values.Set("status", strconv.Itoa(*query.Status)) + } + if query.TransactionName != "" { + values.Set("transactionName", query.TransactionName) + } + if query.Vgroup != "" { + values.Set("vgroup", query.Vgroup) + } + if query.WithBranch != nil { + values.Set("withBranch", strconv.FormatBool(*query.WithBranch)) + } + if query.TimeStart != nil { + values.Set("timeStart", strconv.FormatInt(*query.TimeStart, 10)) + } + if query.TimeEnd != nil { + values.Set("timeEnd", strconv.FormatInt(*query.TimeEnd, 10)) + } + return values +} + +func FormatGlobalSessionPage(response *GlobalSessionPageResult, output string) (string, error) { + output, err := NormalizeOutput(output) + if err != nil { + return "", err + } + switch output { + case OutputTable: + return FormatGlobalSessionTable(response.Data), nil + default: + return FormatStructuredOutput(response, output) + } +} + +func FormatGlobalSessionDetail(session *GlobalSession, output string) (string, error) { + output, err := NormalizeOutput(output) + if err != nil { + return "", err + } + switch output { + case OutputTable: + return FormatGlobalSessionDetailTable(session), nil + default: + return FormatStructuredOutput(session, output) + } +} + +func FormatGlobalSessionTable(sessions []GlobalSession) string { + t := table.NewWriter() + t.Style().Format.Header = text.FormatDefault + t.AppendHeader(table.Row{"xid", "transaction_id", "status", "application_id", "vgroup", "transaction_name", "begin_time", "timeout"}) + for _, session := range sessions { + t.AppendRow(table.Row{ + stringValue(session.XID), + stringValue(session.TransactionID), + intValue(session.Status), + stringValue(session.ApplicationID), + stringValue(session.TransactionServiceGroup), + stringValue(session.TransactionName), + int64Value(session.BeginTime), + int64Value(session.Timeout), + }) + } + return t.Render() +} + +func FormatGlobalSessionDetailTable(session *GlobalSession) string { + t := table.NewWriter() + t.Style().Format.Header = text.FormatDefault + t.AppendHeader(table.Row{"field", "value"}) + t.AppendRows([]table.Row{ + {"xid", stringValue(session.XID)}, + {"transaction_id", stringValue(session.TransactionID)}, + {"status", intValue(session.Status)}, + {"application_id", stringValue(session.ApplicationID)}, + {"vgroup", stringValue(session.TransactionServiceGroup)}, + {"transaction_name", stringValue(session.TransactionName)}, + {"begin_time", int64Value(session.BeginTime)}, + {"timeout", int64Value(session.Timeout)}, + {"application_data", stringValue(session.ApplicationData)}, + {"gmt_create", int64Value(session.GmtCreate)}, + {"gmt_modified", int64Value(session.GmtModified)}, + }) + + if session.BranchSessionVOs == nil || len(*session.BranchSessionVOs) == 0 { + return t.Render() + } + + branches := table.NewWriter() + branches.Style().Format.Header = text.FormatDefault + branches.AppendHeader(table.Row{"branch_id", "status", "resource_id", "branch_type", "client_id", "gmt_create", "gmt_modified"}) + for _, branch := range *session.BranchSessionVOs { + branches.AppendRow(table.Row{ + stringValue(branch.BranchID), + intValue(branch.Status), + stringValue(branch.ResourceID), + stringValue(branch.BranchType), + stringValue(branch.ClientID), + int64Value(branch.GmtCreate), + int64Value(branch.GmtModified), + }) + } + return t.Render() + "\n" + branches.Render() +} diff --git a/seata/transaction_test.go b/seata/transaction_test.go new file mode 100644 index 0000000..b254eaa --- /dev/null +++ b/seata/transaction_test.go @@ -0,0 +1,306 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package seata + +import ( + "encoding/json" + "errors" + "fmt" + "net" + "net/http" + "net/http/httptest" + "net/url" + "reflect" + "strconv" + "strings" + "testing" + + yaml "gopkg.in/yaml.v3" +) + +func TestQueryGlobalSessionsBuildsRequest(t *testing.T) { + status := 1 + withBranch := true + timeStart := int64(1710000000000) + timeEnd := int64(1710003600000) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + t.Fatalf("method = %s, want %s", r.Method, http.MethodGet) + } + if r.URL.Path != GlobalSessionQueryURL { + t.Fatalf("path = %s, want %s", r.URL.Path, GlobalSessionQueryURL) + } + if got := r.Header.Get("authorization"); got != "test-token" { + t.Fatalf("authorization = %q, want test-token", got) + } + wantQuery := map[string]string{ + "xid": "xid-1", + "applicationId": "account-service", + "status": "1", + "transactionName": "create-order", + "vgroup": "default_tx_group", + "withBranch": "true", + "pageNum": "2", + "pageSize": "10", + "timeStart": "1710000000000", + "timeEnd": "1710003600000", + } + for key, want := range wantQuery { + if got := r.URL.Query().Get(key); got != want { + t.Fatalf("query[%s] = %q, want %q", key, got, want) + } + } + fmt.Fprint(w, `{ + "code":"200", + "message":"success", + "success":true, + "pageSize":10, + "pageNum":2, + "total":1, + "pages":1, + "data":[{ + "xid":"xid-1", + "transactionId":"1001", + "status":1, + "applicationId":"account-service", + "transactionServiceGroup":"default_tx_group", + "transactionName":"create-order", + "timeout":30000, + "beginTime":1710000000000 + }] + }`) + })) + defer server.Close() + defer setTestAuth(t, server.URL, "test-token")() + + response, err := queryGlobalSessions(&ConsoleClient{HTTPClient: server.Client()}, GlobalSessionQuery{ + XID: "xid-1", + ApplicationID: "account-service", + Status: &status, + TransactionName: "create-order", + Vgroup: "default_tx_group", + WithBranch: &withBranch, + PageNum: 2, + PageSize: 10, + TimeStart: &timeStart, + TimeEnd: &timeEnd, + }) + if err != nil { + t.Fatalf("queryGlobalSessions returned error: %v", err) + } + if len(response.Data) != 1 { + t.Fatalf("len(response.Data) = %d, want 1", len(response.Data)) + } + session := response.Data[0] + if stringValue(session.XID) != "xid-1" || stringValue(session.TransactionID) != "1001" || stringValue(session.TransactionServiceGroup) != "default_tx_group" { + t.Fatalf("unexpected session: %+v", session) + } +} + +func TestQueryGlobalSessionsOmitsUnsetFilters(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + for _, key := range []string{"xid", "applicationId", "status", "transactionName", "vgroup", "withBranch", "timeStart", "timeEnd"} { + if _, ok := r.URL.Query()[key]; ok { + t.Fatalf("query[%s] should be omitted", key) + } + } + if got := r.URL.Query().Get("pageNum"); got != "1" { + t.Fatalf("pageNum = %q, want 1", got) + } + if got := r.URL.Query().Get("pageSize"); got != "20" { + t.Fatalf("pageSize = %q, want 20", got) + } + fmt.Fprint(w, `{"code":"200","message":"success","data":[]}`) + })) + defer server.Close() + defer setTestAuth(t, server.URL, "test-token")() + + _, err := queryGlobalSessions(&ConsoleClient{HTTPClient: server.Client()}, GlobalSessionQuery{PageNum: 1, PageSize: 20}) + if err != nil { + t.Fatalf("queryGlobalSessions returned error: %v", err) + } +} + +func TestQueryGlobalSessionsErrors(t *testing.T) { + tests := []struct { + name string + body string + want string + }{ + {name: "server code", body: `{"code":"500","message":"boom"}`, want: "boom"}, + {name: "invalid json", body: `{`, want: "decode /api/v1/console/globalSession/query response"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + fmt.Fprint(w, tt.body) + })) + defer server.Close() + defer setTestAuth(t, server.URL, "test-token")() + + _, err := queryGlobalSessions(&ConsoleClient{HTTPClient: server.Client()}, GlobalSessionQuery{PageNum: 1, PageSize: 20}) + if err == nil || !strings.Contains(err.Error(), tt.want) { + t.Fatalf("error = %v, want containing %q", err, tt.want) + } + }) + } +} + +func TestQueryGlobalSessionsHTTPAndNetworkErrors(t *testing.T) { + t.Run("http status", func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "no route", http.StatusNotFound) + })) + defer server.Close() + defer setTestAuth(t, server.URL, "test-token")() + + _, err := queryGlobalSessions(&ConsoleClient{HTTPClient: server.Client()}, GlobalSessionQuery{PageNum: 1, PageSize: 20}) + if err == nil || !strings.Contains(err.Error(), "http status 404") { + t.Fatalf("error = %v, want http status 404", err) + } + }) + + t.Run("network", func(t *testing.T) { + oldAuth := auth + auth = Auth{ServerIP: "127.0.0.1", ServerPort: 7091, token: "test-token"} + defer func() { auth = oldAuth }() + + client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + return nil, errors.New("dial failed") + })} + _, err := queryGlobalSessions(&ConsoleClient{HTTPClient: client}, GlobalSessionQuery{PageNum: 1, PageSize: 20}) + if err == nil || !strings.Contains(err.Error(), "dial failed") { + t.Fatalf("error = %v, want dial failed", err) + } + }) +} + +func TestFormatGlobalSessionPage(t *testing.T) { + success := true + response := &GlobalSessionPageResult{ + Code: CodeOK, + Message: "success", + Success: &success, + PageNum: intPtr(1), + PageSize: intPtr(20), + Total: intPtr(1), + Pages: intPtr(1), + Data: []GlobalSession{{ + XID: stringPtr("xid-1"), + TransactionID: stringPtr("1001"), + Status: intPtr(1), + ApplicationID: stringPtr("account-service"), + TransactionServiceGroup: stringPtr("default_tx_group"), + TransactionName: stringPtr("create-order"), + BeginTime: int64Ptr(1710000000000), + Timeout: int64Ptr(30000), + }}, + } + + tableOutput, err := FormatGlobalSessionPage(response, OutputTable) + if err != nil { + t.Fatalf("table output error: %v", err) + } + for _, want := range []string{ + "| xid | transaction_id | status | application_id | vgroup | transaction_name | begin_time | timeout |", + "| xid-1 | 1001 | 1 | account-service | default_tx_group | create-order | 1710000000000 | 30000 |", + } { + if !strings.Contains(tableOutput, want) { + t.Fatalf("table output %q does not contain %q", tableOutput, want) + } + } + + jsonOutput, err := FormatGlobalSessionPage(response, OutputJSON) + if err != nil { + t.Fatalf("json output error: %v", err) + } + var jsonResult GlobalSessionPageResult + if err = json.Unmarshal([]byte(jsonOutput), &jsonResult); err != nil { + t.Fatalf("unmarshal json output: %v", err) + } + if !reflect.DeepEqual(&jsonResult, response) { + t.Fatalf("unexpected json result: %+v", jsonResult) + } + for _, want := range []string{`"applicationData": null`, `"branchSessionVOs": null`} { + if !strings.Contains(jsonOutput, want) { + t.Fatalf("json output %q does not contain %q", jsonOutput, want) + } + } + + yamlOutput, err := FormatGlobalSessionPage(response, OutputYAML) + if err != nil { + t.Fatalf("yaml output error: %v", err) + } + var yamlResult GlobalSessionPageResult + if err = yaml.Unmarshal([]byte(yamlOutput), &yamlResult); err != nil { + t.Fatalf("unmarshal yaml output: %v", err) + } + if yamlResult.PageNum == nil || *yamlResult.PageNum != 1 || len(yamlResult.Data) != 1 || stringValue(yamlResult.Data[0].XID) != "xid-1" || yamlResult.Data[0].ApplicationData != nil { + t.Fatalf("unexpected yaml result: %+v", yamlResult) + } + for _, want := range []string{"applicationData: null", "branchSessionVOs: null"} { + if !strings.Contains(yamlOutput, want) { + t.Fatalf("yaml output %q does not contain %q", yamlOutput, want) + } + } + + _, err = FormatGlobalSessionPage(response, "xml") + if err == nil || !strings.Contains(err.Error(), "unsupported output format") { + t.Fatalf("error = %v, want unsupported output format", err) + } +} + +func setTestAuth(t *testing.T, serverURL string, token string) func() { + t.Helper() + parsedURL, err := url.Parse(serverURL) + if err != nil { + t.Fatalf("parse server url: %v", err) + } + host, portStr, err := net.SplitHostPort(parsedURL.Host) + if err != nil { + t.Fatalf("split host port: %v", err) + } + port, err := strconv.Atoi(portStr) + if err != nil { + t.Fatalf("parse port: %v", err) + } + oldAuth := auth + auth = Auth{ServerIP: host, ServerPort: port, token: token} + return func() { + auth = oldAuth + } +} + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (fn roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) { + return fn(request) +} + +func stringPtr(value string) *string { + return &value +} + +func intPtr(value int) *int { + return &value +} + +func int64Ptr(value int64) *int64 { + return &value +} diff --git a/seata/txn.go b/seata/txn.go index dcd94aa..723bdad 100644 --- a/seata/txn.go +++ b/seata/txn.go @@ -41,7 +41,7 @@ func BeginTxn(timeout int) { } request, _ := http.NewRequest("POST", url, nil) request.Header.Set("authorization", token) - resp, err := (&http.Client{}).Do(request) + resp, err := defaultHTTPClient.Do(request) if err != nil { return } @@ -74,7 +74,7 @@ func CommitTxn(xid string) { } request, _ := http.NewRequest("POST", url, nil) request.Header.Set("authorization", token) - resp, err := (&http.Client{}).Do(request) + resp, err := defaultHTTPClient.Do(request) if err != nil { return } @@ -107,7 +107,7 @@ func RollbackTxn(xid string) { } request, _ := http.NewRequest("POST", url, nil) request.Header.Set("authorization", token) - resp, err := (&http.Client{}).Do(request) + resp, err := defaultHTTPClient.Do(request) if err != nil { return }