Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
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
2 changes: 1 addition & 1 deletion .travis.yml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ go:
- 1.10.x

script:
- make dev
- make ci

matrix:
include:
Expand Down
6 changes: 6 additions & 0 deletions Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ PD_PKG := github.com/pingcap/pd

TEST_PKGS := $(shell find . -iname "*_test.go" -exec dirname {} \; | \
uniq | sed -e "s/^\./github.com\/pingcap\/pd/")
BASIC_TEST_PKGS := $(filter-out github.com/pingcap/pd/pkg/integration_test,$(TEST_PKGS))

GOFILTER := grep -vE 'vendor|testutil'
GOCHECKER := $(GOFILTER) | awk '{ print } END { if (NR > 0) { exit 1 } }'
Expand All @@ -22,6 +23,8 @@ all: dev

dev: build simulator check test

ci: build simulator check basic_test

build:
ifeq ("$(WITH_RACE)", "1")
CGO_ENABLED=1 go build -race -ldflags '$(LDFLAGS)' -o bin/pd-server cmd/pd-server/main.go
Expand All @@ -36,6 +39,9 @@ test:
# testing..
CGO_ENABLED=1 go test -race -cover $(TEST_PKGS)

basic_test:
go test $(BASIC_TEST_PKGS)

check:
go get github.com/golang/lint/golint

Expand Down
2 changes: 1 addition & 1 deletion circle.yml
Original file line number Diff line number Diff line change
Expand Up @@ -9,4 +9,4 @@ jobs:
- checkout
- run:
name: "Build & Test"
command: make dev
command: make ci
243 changes: 243 additions & 0 deletions pkg/integration_test/cluster.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,243 @@
// Copyright 2018 PingCAP, Inc.
//
// 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,
// See the License for the specific language governing permissions and
// limitations under the License.

package integration

import (
"context"
"os"
"sync"
"time"

"github.com/juju/errors"
"github.com/pingcap/kvproto/pkg/pdpb"
"github.com/pingcap/pd/server"
"github.com/pingcap/pd/server/api"
)

// testServer states.
const (
Initial int32 = iota
Running
Stop
Destroy
)

type testServer struct {
sync.RWMutex
server *server.Server
state int32
}

var initHTTPClientOnce sync.Once

func newTestServer(cfg *server.Config) (*testServer, error) {
err := server.PrepareJoinCluster(cfg)
if err != nil {
return nil, errors.Trace(err)
}
svr, err := server.CreateServer(cfg, api.NewHandler)
if err != nil {
return nil, errors.Trace(err)
}
initHTTPClientOnce.Do(func() {
err = server.InitHTTPClient(svr)
})
if err != nil {
return nil, errors.Trace(err)
}
return &testServer{
server: svr,
state: Initial,
}, nil
}

func (s *testServer) Run() error {
s.Lock()
defer s.Unlock()
if s.state != Initial && s.state != Stop {
return errors.Errorf("server(state%d) cannot run", s.state)
}
if err := s.server.Run(); err != nil {
return errors.Trace(err)
}
s.state = Running
return nil
}

func (s *testServer) Stop() error {
s.Lock()
defer s.Unlock()
if s.state != Running {
return errors.Errorf("server(state%d) cannot stop", s.state)
}
s.server.Close()
s.state = Stop
return nil
}

func (s *testServer) Destory() error {
s.Lock()
defer s.Unlock()
if s.state == Running {
s.server.Close()
}
os.RemoveAll(s.server.GetConfig().DataDir)
s.state = Destroy
return nil
}

func (s *testServer) State() int32 {
s.RLock()
defer s.RUnlock()
return s.state
}

func (s *testServer) GetConfig() *server.Config {
s.RLock()
defer s.RUnlock()
return s.server.GetConfig()
}

func (s *testServer) GetClusterID() uint64 {
s.RLock()
defer s.RUnlock()
return s.server.ClusterID()
}

func (s *testServer) GetServerID() uint64 {
s.RLock()
defer s.RUnlock()
return s.server.ID()
}

func (s *testServer) IsLeader() bool {
s.RLock()
defer s.RUnlock()
return s.server.IsLeader()
}

func (s *testServer) GetEtcdLeader() (string, error) {
s.RLock()
defer s.RUnlock()
req := &pdpb.GetMembersRequest{Header: &pdpb.RequestHeader{ClusterId: s.server.ClusterID()}}
members, err := s.server.GetMembers(context.TODO(), req)
if err != nil {
return "", errors.Trace(err)
}
return members.GetEtcdLeader().GetName(), nil
}

type testCluster struct {
config *clusterConfig
servers map[string]*testServer
}

func newTestCluster(initialServerCount int) (*testCluster, error) {
config := newClusterConfig(initialServerCount)
servers := make(map[string]*testServer)
for _, conf := range config.InitialServers {
serverConf, err := conf.Generate()
if err != nil {
return nil, errors.Trace(err)
}
s, err := newTestServer(serverConf)
if err != nil {
return nil, errors.Trace(err)
}
servers[conf.Name] = s
}
return &testCluster{
config: config,
servers: servers,
}, nil
}

func (c *testCluster) RunAll() error {
var wg sync.WaitGroup
errCh := make(chan error, len(c.servers))
for _, s := range c.servers {
wg.Add(1)
go func(s *testServer) {
defer wg.Done()
if err := s.Run(); err != nil {
errCh <- err
}
}(s)
}
wg.Wait()
close(errCh)
var errs []error
for err := range errCh {
errs = append(errs, err)
}
if len(errs) != 0 {
return errors.Errorf("run server error(s): %v", errs)
}
return nil
}

func (c *testCluster) StopAll() error {
for _, s := range c.servers {
if err := s.Stop(); err != nil {
return errors.Trace(err)
}
}
return nil
}

func (c *testCluster) GetServer(name string) *testServer {
return c.servers[name]
}

func (c *testCluster) GetLeader() string {
for name, s := range c.servers {
if s.IsLeader() {
return name
}
}
return ""
}

func (c *testCluster) WaitLeader() string {
for i := 0; i < 100; i++ {
if leader := c.GetLeader(); leader != "" {
return leader
}
time.Sleep(100 * time.Millisecond)
}
return ""
}

func (c *testCluster) Join() (*testServer, error) {
conf, err := c.config.Join().Generate()
if err != nil {
return nil, errors.Trace(err)
}
s, err := newTestServer(conf)
if err != nil {
return nil, errors.Trace(err)
}
c.servers[conf.Name] = s
return s, nil
}

func (c *testCluster) Destory() error {
for _, s := range c.servers {
err := s.Destory()
if err != nil {
return errors.Trace(err)
}
}
return nil
}
106 changes: 106 additions & 0 deletions pkg/integration_test/config.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,106 @@
// Copyright 2018 PingCAP, Inc.
//
// 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,
// See the License for the specific language governing permissions and
// limitations under the License.

package integration

import (
"fmt"
"io/ioutil"
"strings"

"github.com/juju/errors"
"github.com/pingcap/pd/pkg/tempurl"
"github.com/pingcap/pd/server"
)

type serverConfig struct {
Name string
DataDir string
ClientURLs string
AdvertiseClientURLs string
PeerURLs string
AdvertisePeerURLs string
ClusterConfig *clusterConfig
Join bool
}

func newServerConfig(name string, cc *clusterConfig, join bool) *serverConfig {
tempDir, _ := ioutil.TempDir("/tmp", "pd-integration-test")
return &serverConfig{
Name: name,
DataDir: tempDir,
ClientURLs: tempurl.Alloc(),
PeerURLs: tempurl.Alloc(),
ClusterConfig: cc,
Join: join,
}
}

func (c *serverConfig) Generate() (*server.Config, error) {
arguments := []string{
"--name=" + c.Name,
"--data-dir=" + c.DataDir,
"--client-urls=" + c.ClientURLs,
"--advertise-client-urls=" + c.AdvertiseClientURLs,
"--peer-urls=" + c.PeerURLs,
"--advertise-peer-urls=" + c.AdvertisePeerURLs,
}
if c.Join {
arguments = append(arguments, "--join="+c.ClusterConfig.GetJoinAddr())
} else {
arguments = append(arguments, "--initial-cluster="+c.ClusterConfig.GetServerAddrs())
}

cfg := server.NewConfig()
err := cfg.Parse(arguments)
if err != nil {
return nil, errors.Trace(err)
}
return cfg, nil
}

type clusterConfig struct {
InitialServers []*serverConfig
JoinServers []*serverConfig
}

func newClusterConfig(n int) *clusterConfig {
var cc clusterConfig
for i := 0; i < n; i++ {
c := newServerConfig(cc.nextServerName(), &cc, false)
cc.InitialServers = append(cc.InitialServers, c)
}
return &cc
}

func (c *clusterConfig) Join() *serverConfig {
sc := newServerConfig(c.nextServerName(), c, true)
c.JoinServers = append(c.JoinServers, sc)
return sc
}

func (c *clusterConfig) nextServerName() string {
return fmt.Sprintf("pd%d", len(c.InitialServers)+len(c.JoinServers)+1)
}

func (c *clusterConfig) GetServerAddrs() string {
var addrs []string
for _, s := range c.InitialServers {
addrs = append(addrs, fmt.Sprintf("%s=%s", s.Name, s.PeerURLs))
}
return strings.Join(addrs, ",")
}

func (c *clusterConfig) GetJoinAddr() string {
return c.InitialServers[0].PeerURLs
}
Loading