test: move heavy tests to integration_test package

pkg/integration_test/cluster.go

@@ -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
+}

pkg/integration_test/config.go

@@ -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
+}