Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

mcs: add a test for starting tso server first #6535

Merged
merged 3 commits into from
May 31, 2023
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
add a test for starting tso server first
Signed-off-by: Ryan Leung <rleungx@gmail.com>
  • Loading branch information
rleungx committed May 30, 2023
commit 525c3f811d18ec4db6511b47c5931d7b9a2c5791
2 changes: 1 addition & 1 deletion pkg/mcs/resourcemanager/server/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -409,7 +409,7 @@ func (s *Server) startServer() (err error) {
s.serviceRegister = discovery.NewServiceRegister(s.ctx, s.etcdClient, strconv.FormatUint(s.clusterID, 10),
utils.ResourceManagerServiceName, s.cfg.AdvertiseListenAddr, serializedEntry, discovery.DefaultLeaseInSeconds)
if err := s.serviceRegister.Register(); err != nil {
log.Error("failed to regiser the service", zap.String("service-name", utils.ResourceManagerServiceName), errs.ZapError(err))
log.Error("failed to register the service", zap.String("service-name", utils.ResourceManagerServiceName), errs.ZapError(err))
return err
}
atomic.StoreInt64(&s.isRunning, 1)
Expand Down
3 changes: 3 additions & 0 deletions server/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -559,6 +559,9 @@ func (s *Server) Run() error {
if err := s.startEtcd(s.ctx); err != nil {
return err
}
failpoint.Inject("delayStartServer", func() {
time.Sleep(2 * time.Second)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it possible to inject an input channel and let it wait for an input after mcs.NewTestTSOCluster(ctx, 2, addr) is completed? This way could make the test more deterministic and stable.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it's ok for now and this test won't be unstable because of it.

})
if err := s.startServer(s.ctx); err != nil {
return err
}
Expand Down
69 changes: 69 additions & 0 deletions tests/integrations/mcs/tso/api_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -15,17 +15,22 @@
package tso

import (
"bytes"
"context"
"encoding/json"
"io"
"net/http"
"testing"
"time"

"github.com/pingcap/failpoint"
"github.com/stretchr/testify/require"
"github.com/stretchr/testify/suite"
tso "github.com/tikv/pd/pkg/mcs/tso/server"
apis "github.com/tikv/pd/pkg/mcs/tso/server/apis/v1"
mcsutils "github.com/tikv/pd/pkg/mcs/utils"
"github.com/tikv/pd/pkg/storage/endpoint"
"github.com/tikv/pd/server/config"
"github.com/tikv/pd/tests"
"github.com/tikv/pd/tests/integrations/mcs"
)
Expand Down Expand Up @@ -104,3 +109,67 @@ func mustGetKeyspaceGroupMembers(re *require.Assertions, server *tso.Server) map
re.NoError(json.Unmarshal(data, &resp))
return resp
}

func TestTSOServerStartFirst(t *testing.T) {
re := require.New(t)
re.NoError(failpoint.Enable("github.com/tikv/pd/server/delayStartServer", `return(true)`))
ctx, cancel := context.WithCancel(context.Background())
defer cancel()

pdCluster, err := tests.NewTestAPICluster(ctx, 1, func(conf *config.Config, serverName string) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shall we call it apiCluster to indicate it in API mode?

conf.Keyspace.PreAlloc = []string{"k1", "k2"}
})
defer pdCluster.Destroy()
re.NoError(err)
addr := pdCluster.GetConfig().GetClientURL()
ch := make(chan struct{})
defer close(ch)
clusterCh := make(chan *mcs.TestTSOCluster)
defer close(clusterCh)
go func() {
tsoCluster, err := mcs.NewTestTSOCluster(ctx, 2, addr)
re.NoError(err)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could NewTestTSOCluster fail at etcdutil.CreateClients() because pdCluster.RunInitialServers() hasn't started etcd at that moment?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, it will wait for API server to be ready.

primary := tsoCluster.WaitForDefaultPrimaryServing(re)
re.NotNil(primary)
clusterCh <- tsoCluster
ch <- struct{}{}
}()
err = pdCluster.RunInitialServers()
re.NoError(err)
leaderName := pdCluster.WaitLeader()
pdLeaderServer := pdCluster.GetServer(leaderName)
re.NoError(pdLeaderServer.BootstrapCluster())
re.NoError(err)
tsoCluster := <-clusterCh
defer tsoCluster.Destroy()
<-ch

time.Sleep(time.Second * 1)
input := make(map[string]interface{})
input["new-id"] = 1
input["keyspaces"] = []uint32{2}
jsonBody, err := json.Marshal(input)
re.NoError(err)
httpReq, err := http.NewRequest(http.MethodPost, addr+"/pd/api/v2/tso/keyspace-groups/0/split", bytes.NewBuffer(jsonBody))
re.NoError(err)
httpResp, err := dialClient.Do(httpReq)
re.NoError(err)
defer httpResp.Body.Close()
re.Equal(http.StatusOK, httpResp.StatusCode)

httpReq, err = http.NewRequest(http.MethodGet, addr+"/pd/api/v2/tso/keyspace-groups/0", nil)
re.NoError(err)
httpResp, err = dialClient.Do(httpReq)
re.NoError(err)
data, err := io.ReadAll(httpResp.Body)
re.NoError(err)
defer httpResp.Body.Close()
re.Equal(http.StatusOK, httpResp.StatusCode)

var group endpoint.KeyspaceGroup
re.NoError(json.Unmarshal(data, &group))
re.Len(group.Keyspaces, 2)
re.Len(group.Members, 2)

re.NoError(failpoint.Disable("github.com/tikv/pd/server/delayStartServer"))
}