Skip to content

Commit

Permalink
Add GetTransactionInfo in VTadmin API and page (#17142)
Browse files Browse the repository at this point in the history
Signed-off-by: Manan Gupta <[email protected]>
  • Loading branch information
GuptaManan100 authored Nov 8, 2024
1 parent d9ab9f7 commit 469bdcc
Show file tree
Hide file tree
Showing 21 changed files with 2,881 additions and 1,469 deletions.
18 changes: 18 additions & 0 deletions go/test/endtoend/cluster/cluster_process.go
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,7 @@ type LocalProcessCluster struct {
VtctlclientProcess VtctlClientProcess
VtctldClientProcess VtctldClientProcess
VtctlProcess VtctlProcess
VtadminProcess VtAdminProcess

// background executable processes
TopoProcess TopoProcess
Expand Down Expand Up @@ -1061,6 +1062,10 @@ func (cluster *LocalProcessCluster) Teardown() {
}
}

if err := cluster.VtadminProcess.TearDown(); err != nil {
log.Errorf("Error in vtadmin teardown: %v", err)
}

var mysqlctlProcessList []*exec.Cmd
var mysqlctlTabletUIDs []int
for _, keyspace := range cluster.Keyspaces {
Expand Down Expand Up @@ -1301,6 +1306,19 @@ func (cluster *LocalProcessCluster) NewVTOrcProcess(config VTOrcConfiguration) *
}
}

// NewVTAdminProcess creates a new VTAdminProcess object
func (cluster *LocalProcessCluster) NewVTAdminProcess() {
cluster.VtadminProcess = VtAdminProcess{
Binary: "vtadmin",
Port: cluster.GetAndReservePort(),
LogDir: cluster.TmpDirectory,
VtGateGrpcPort: cluster.VtgateProcess.GrpcPort,
VtGateWebPort: cluster.VtgateProcess.Port,
VtctldWebPort: cluster.VtctldProcess.Port,
VtctldGrpcPort: cluster.VtctldProcess.GrpcPort,
}
}

// VtprocessInstanceFromVttablet creates a new vttablet object
func (cluster *LocalProcessCluster) VtprocessInstanceFromVttablet(tablet *Vttablet, shardName string, ksName string) *VttabletProcess {
return VttabletProcessInstance(
Expand Down
235 changes: 235 additions & 0 deletions go/test/endtoend/cluster/vtadmin_process.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,235 @@
/*
Copyright 2024 The Vitess Authors.
Licensed 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 cluster

import (
"fmt"
"io"
"net/http"
"os"
"os/exec"
"path"
"strings"
"syscall"
"testing"
"time"

"github.com/stretchr/testify/require"

"vitess.io/vitess/go/vt/log"
)

// VtAdminProcess is a test struct for running
// vtorc as a separate process for testing
type VtAdminProcess struct {
Binary string
Port int
LogDir string
ExtraArgs []string
VtGateGrpcPort int
VtGateWebPort int
VtctldGrpcPort int
VtctldWebPort int
proc *exec.Cmd
exit chan error
}

// Setup starts orc process with required arguements
func (vp *VtAdminProcess) Setup() (err error) {
// create the configuration file
timeNow := time.Now().UnixNano()
err = os.MkdirAll(vp.LogDir, 0755)
if err != nil {
log.Errorf("cannot create log directory for vtadmin: %v", err)
return err
}
rbacFile, err := vp.CreateAndWriteFile("rbac", `rules:
- resource: "*"
actions: ["*"]
subjects: ["*"]
clusters: ["*"]
`, "yaml")
if err != nil {
return err
}

discoveryFile, err := vp.CreateAndWriteFile("discovery", fmt.Sprintf(`
{
"vtctlds": [
{
"host": {
"fqdn": "localhost:%d",
"hostname": "localhost:%d"
}
}
],
"vtgates": [
{
"host": {
"fqdn": "localhost:%d",
"hostname": "localhost:%d"
}
}
]
}
`, vp.VtctldWebPort, vp.VtctldGrpcPort, vp.VtGateWebPort, vp.VtGateGrpcPort), "json")
if err != nil {
return err
}

vp.proc = exec.Command(
vp.Binary,
"--addr", vp.Address(),
"--http-tablet-url-tmpl", `"http://{{ .Tablet.Hostname }}:15{{ .Tablet.Alias.Uid }}"`,
"--tracer", `"opentracing-jaeger"`,
"--grpc-tracing",
"--http-tracing",
"--logtostderr",
"--alsologtostderr",
"--rbac",
"--rbac-config", rbacFile,
"--cluster", fmt.Sprintf(`id=local,name=local,discovery=staticfile,discovery-staticfile-path=%s,tablet-fqdn-tmpl=http://{{ .Tablet.Hostname }}:15{{ .Tablet.Alias.Uid }},schema-cache-default-expiration=1m`, discoveryFile),
)

if *isCoverage {
vp.proc.Args = append(vp.proc.Args, "--test.coverprofile="+getCoveragePath("vp.out"))
}

vp.proc.Args = append(vp.proc.Args, vp.ExtraArgs...)
vp.proc.Args = append(vp.proc.Args, "--alsologtostderr")

logFile := fmt.Sprintf("vtadmin-stderr-%d.txt", timeNow)
errFile, err := os.Create(path.Join(vp.LogDir, logFile))
if err != nil {
log.Errorf("cannot create error log file for vtadmin: %v", err)
return err
}
vp.proc.Stderr = errFile

vp.proc.Env = append(vp.proc.Env, os.Environ()...)
vp.proc.Env = append(vp.proc.Env, DefaultVttestEnv)

log.Infof("Running vtadmin with command: %v", strings.Join(vp.proc.Args, " "))

err = vp.proc.Start()
if err != nil {
return
}

vp.exit = make(chan error)
go func() {
if vp.proc != nil {
exitErr := vp.proc.Wait()
if exitErr != nil {
log.Errorf("vtadmin process exited with error: %v", exitErr)
data, _ := os.ReadFile(logFile)
log.Errorf("vtadmin stderr - %s", string(data))
}
vp.exit <- exitErr
close(vp.exit)
}
}()

return nil
}

// CreateAndWriteFile creates a file and writes the content to it.
func (vp *VtAdminProcess) CreateAndWriteFile(prefix string, content string, extension string) (string, error) {
timeNow := time.Now().UnixNano()
file, err := os.Create(path.Join(vp.LogDir, fmt.Sprintf("%s-%d.%s", prefix, timeNow, extension)))
if err != nil {
log.Errorf("cannot create file for vtadmin: %v", err)
return "", err
}

_, err = file.WriteString(content)
if err != nil {
return "", err
}
fileName := file.Name()
err = file.Close()
if err != nil {
return "", err
}
return fileName, nil
}

// Address returns the address of the running vtadmin service.
func (vp *VtAdminProcess) Address() string {
return fmt.Sprintf("localhost:%d", vp.Port)
}

// TearDown shuts down the running vtorc service
func (vp *VtAdminProcess) TearDown() error {
if vp.proc == nil || vp.exit == nil {
return nil
}
// Attempt graceful shutdown with SIGTERM first
_ = vp.proc.Process.Signal(syscall.SIGTERM)

select {
case <-vp.exit:
vp.proc = nil
return nil

case <-time.After(30 * time.Second):
_ = vp.proc.Process.Kill()
err := <-vp.exit
vp.proc = nil
return err
}
}

// MakeAPICall makes an API call on the given endpoint of VTOrc
func (vp *VtAdminProcess) MakeAPICall(endpoint string) (status int, response string, err error) {
url := fmt.Sprintf("http://%s/%s", vp.Address(), endpoint)
resp, err := http.Get(url)
if err != nil {
if resp != nil {
status = resp.StatusCode
}
return status, "", err
}
defer func() {
if resp != nil && resp.Body != nil {
resp.Body.Close()
}
}()

respByte, _ := io.ReadAll(resp.Body)
return resp.StatusCode, string(respByte), err
}

// MakeAPICallRetry is used to make an API call and retries until success
func (vp *VtAdminProcess) MakeAPICallRetry(t *testing.T, url string) string {
t.Helper()
timeout := time.After(10 * time.Second)
for {
select {
case <-timeout:
require.FailNow(t, "timed out waiting for api to work")
return ""
default:
status, resp, err := vp.MakeAPICall(url)
if err == nil && status == 200 {
return resp
}
time.Sleep(100 * time.Millisecond)
}
}
}
5 changes: 5 additions & 0 deletions go/test/endtoend/transaction/twopc/main_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,11 @@ func TestMain(m *testing.M) {
vtParams = clusterInstance.GetVTParams(keyspaceName)
vtgateGrpcAddress = fmt.Sprintf("%s:%d", clusterInstance.Hostname, clusterInstance.VtgateGrpcPort)

clusterInstance.NewVTAdminProcess()
if err := clusterInstance.VtadminProcess.Setup(); err != nil {
return 1
}

// create mysql instance and connection parameters
conn, closer, err := utils.NewMySQL(clusterInstance, keyspaceName, SchemaSQL)
if err != nil {
Expand Down
7 changes: 7 additions & 0 deletions go/test/endtoend/transaction/twopc/twopc_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import (
"fmt"
"reflect"
"sort"
"strconv"
"strings"
"sync"
"testing"
Expand Down Expand Up @@ -1428,6 +1429,12 @@ func TestReadTransactionStatus(t *testing.T) {
require.Contains(t, out, "insert into twopc_t1(id, col) values (9, 4)")
require.Contains(t, out, unresTransaction.Dtid)

// Read the data from vtadmin API, and verify that too has the same information.
apiRes := clusterInstance.VtadminProcess.MakeAPICallRetry(t, fmt.Sprintf("/api/transaction/local/%v/info", unresTransaction.Dtid))
require.Contains(t, apiRes, "insert into twopc_t1(id, col) values (9, 4)")
require.Contains(t, apiRes, unresTransaction.Dtid)
require.Contains(t, apiRes, strconv.FormatInt(res.TimeCreated, 10))

// Wait for the commit to have returned.
wg.Wait()
}
Expand Down
Loading

0 comments on commit 469bdcc

Please sign in to comment.