etcd-tester: support per-agent client/peer/failpoint ports

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

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{

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

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