From 55ba3d95fb5ac1f14f1dda7192d4911606bbc826 Mon Sep 17 00:00:00 2001 From: Anthony Romano Date: Thu, 8 Sep 2016 16:09:50 -0700 Subject: [PATCH] etcd-tester: support per-agent client/peer/failpoint ports --- .../functional-tester/etcd-tester/cluster.go | 44 ++++++++-------- .../etcd-tester/failure_agent.go | 4 +- tools/functional-tester/etcd-tester/main.go | 52 ++++++++++++++++++- tools/functional-tester/etcd-tester/member.go | 12 +++++ 4 files changed, 86 insertions(+), 26 deletions(-) diff --git a/tools/functional-tester/etcd-tester/cluster.go b/tools/functional-tester/etcd-tester/cluster.go index 52d11b10b..263b55dfa 100644 --- a/tools/functional-tester/etcd-tester/cluster.go +++ b/tools/functional-tester/etcd-tester/cluster.go @@ -29,15 +29,21 @@ import ( "google.golang.org/grpc" ) -const ( - peerURLPort = 2380 - failpointPort = 2381 -) +// agentConfig holds information needed to interact/configure an agent and its etcd process +type agentConfig struct { + endpoint string + clientPort int + peerPort int + failpointPort int + + datadir string +} type cluster struct { + agents []agentConfig + v2Only bool // to be deprecated - datadir string stressQPS int stressKeyLargeSize int stressKeySize int @@ -53,27 +59,27 @@ type ClusterStatus struct { AgentStatuses map[string]client.Status } -func (c *cluster) bootstrap(agentEndpoints []string) error { - size := len(agentEndpoints) +func (c *cluster) bootstrap() error { + size := len(c.agents) members := make([]*member, size) memberNameURLs := make([]string, size) - for i, u := range agentEndpoints { - agent, err := client.NewAgent(u) + for i, a := range c.agents { + agent, err := client.NewAgent(a.endpoint) if err != nil { return err } - host, _, err := net.SplitHostPort(u) + host, _, err := net.SplitHostPort(a.endpoint) if err != nil { return err } members[i] = &member{ Agent: agent, - Endpoint: u, + Endpoint: a.endpoint, Name: fmt.Sprintf("etcd-%d", i), - ClientURL: fmt.Sprintf("http://%s:2379", host), - PeerURL: fmt.Sprintf("http://%s:%d", host, peerURLPort), - FailpointURL: fmt.Sprintf("http://%s:%d", host, failpointPort), + ClientURL: fmt.Sprintf("http://%s:%d", host, a.clientPort), + PeerURL: fmt.Sprintf("http://%s:%d", host, a.peerPort), + FailpointURL: fmt.Sprintf("http://%s:%d", host, a.failpointPort), } memberNameURLs[i] = members[i].ClusterEntry() } @@ -83,7 +89,7 @@ func (c *cluster) bootstrap(agentEndpoints []string) error { for i, m := range members { flags := append( m.Flags(), - "--data-dir", c.datadir, + "--data-dir", c.agents[i].datadir, "--initial-cluster-token", token, "--initial-cluster", clusterStr) @@ -127,13 +133,7 @@ func (c *cluster) bootstrap(agentEndpoints []string) error { return nil } -func (c *cluster) Reset() error { - eps := make([]string, len(c.Members)) - for i, m := range c.Members { - eps[i] = m.Endpoint - } - return c.bootstrap(eps) -} +func (c *cluster) Reset() error { return c.bootstrap() } func (c *cluster) WaitHealth() error { var err error diff --git a/tools/functional-tester/etcd-tester/failure_agent.go b/tools/functional-tester/etcd-tester/failure_agent.go index 2e164e224..c539eafcc 100644 --- a/tools/functional-tester/etcd-tester/failure_agent.go +++ b/tools/functional-tester/etcd-tester/failure_agent.go @@ -78,8 +78,8 @@ func newFailureKillLeaderForLongTime() failure { return &failureUntilSnapshot{newFailureKillLeader()} } -func injectDropPort(m *member) error { return m.Agent.DropPort(peerURLPort) } -func recoverDropPort(m *member) error { return m.Agent.RecoverPort(peerURLPort) } +func injectDropPort(m *member) error { return m.Agent.DropPort(m.peerPort()) } +func recoverDropPort(m *member) error { return m.Agent.RecoverPort(m.peerPort()) } func newFailureIsolate() failure { return &failureOne{ diff --git a/tools/functional-tester/etcd-tester/main.go b/tools/functional-tester/etcd-tester/main.go index f3a9117c1..f36396292 100644 --- a/tools/functional-tester/etcd-tester/main.go +++ b/tools/functional-tester/etcd-tester/main.go @@ -18,6 +18,7 @@ import ( "flag" "fmt" "net/http" + "os" "strings" "github.com/coreos/pkg/capnslog" @@ -26,8 +27,18 @@ import ( var plog = capnslog.NewPackageLogger("github.com/coreos/etcd", "etcd-tester") +const ( + defaultClientPort = 2379 + defaultPeerPort = 2380 + defaultFailpointPort = 2381 +) + func main() { endpointStr := flag.String("agent-endpoints", "localhost:9027", "HTTP RPC endpoints of agents. Do not specify the schema.") + clientPorts := flag.String("client-ports", "", "etcd client port for each agent endpoint") + peerPorts := flag.String("peer-ports", "", "etcd peer port for each agent endpoint") + failpointPorts := flag.String("failpoint-ports", "", "etcd failpoint port for each agent endpoint") + datadir := flag.String("data-dir", "agent.etcd", "etcd data directory location on agent machine.") stressKeyLargeSize := flag.Uint("stress-key-large-size", 32*1024+1, "the size of each large key written into etcd.") stressKeySize := flag.Uint("stress-key-size", 100, "the size of each small key written into etcd.") @@ -39,15 +50,29 @@ func main() { isV2Only := flag.Bool("v2-only", false, "'true' to run V2 only tester.") flag.Parse() + eps := strings.Split(*endpointStr, ",") + cports := portsFromArg(*clientPorts, len(eps), defaultClientPort) + pports := portsFromArg(*peerPorts, len(eps), defaultPeerPort) + fports := portsFromArg(*failpointPorts, len(eps), defaultFailpointPort) + agents := make([]agentConfig, len(eps)) + for i := range eps { + agents[i].endpoint = eps[i] + agents[i].clientPort = cports[i] + agents[i].peerPort = pports[i] + agents[i].failpointPort = fports[i] + agents[i].datadir = *datadir + } + c := &cluster{ + agents: agents, v2Only: *isV2Only, - datadir: *datadir, stressQPS: *stressQPS, stressKeyLargeSize: int(*stressKeyLargeSize), stressKeySize: int(*stressKeySize), stressKeySuffixRange: int(*stressKeySuffixRange), } - if err := c.bootstrap(strings.Split(*endpointStr, ",")); err != nil { + + if err := c.bootstrap(); err != nil { plog.Fatal(err) } defer c.Terminate() @@ -102,3 +127,26 @@ func main() { t.runLoop() } + +// portsFromArg converts a comma separated list into a slice of ints +func portsFromArg(arg string, n, defaultPort int) []int { + ret := make([]int, n) + if len(arg) == 0 { + for i := range ret { + ret[i] = defaultPort + } + return ret + } + s := strings.Split(arg, ",") + if len(s) != n { + fmt.Printf("expected %d ports, got %d (%s)\n", n, len(s), arg) + os.Exit(1) + } + for i := range s { + if _, err := fmt.Sscanf(s[i], "%d", &ret[i]); err != nil { + fmt.Println(err) + os.Exit(1) + } + } + return ret +} diff --git a/tools/functional-tester/etcd-tester/member.go b/tools/functional-tester/etcd-tester/member.go index 5d5a12568..26a807be3 100644 --- a/tools/functional-tester/etcd-tester/member.go +++ b/tools/functional-tester/etcd-tester/member.go @@ -16,6 +16,7 @@ package main import ( "fmt" + "net" "net/url" "time" @@ -165,3 +166,14 @@ func (m *member) grpcAddr() string { } return u.Host } + +func (m *member) peerPort() (port int) { + _, portStr, err := net.SplitHostPort(m.PeerURL) + if err != nil { + panic(err) + } + if _, err = fmt.Sscanf(portStr, "%d", &port); err != nil { + panic(err) + } + return port +}