Merge pull request #6389 from heyitsanthony/func-tester-noroot

functional-tester: run locally

tools/functional-tester/Procfile

@@ -0,0 +1,4 @@
+agent-1: mkdir -p agent-1 && cd agent-1 && ../bin/etcd-agent -etcd-path ../bin/etcd -port localhost:9027 -use-root=false
+agent-2: mkdir -p agent-2 && cd agent-2 && ../bin/etcd-agent -etcd-path ../bin/etcd -port localhost:9028 -use-root=false
+agent-3: mkdir -p agent-3 && cd agent-3 && ../bin/etcd-agent -etcd-path ../bin/etcd -port localhost:9029 -use-root=false
+stresser: sleep 1s && bin/etcd-tester -agent-endpoints "localhost:9027,localhost:9028,localhost:9029"  -client-ports 12379,22379,32379 -peer-ports 12380,22380,32380

tools/functional-tester/README.md

@@ -35,3 +35,17 @@ Notes:
 - Docker image is based on Alpine Linux OS running in privileged mode to allow iptables manipulation.
 - To specify testing parameters (etcd-tester arguments) modify tools/functional-tester/docker/docker-compose.yml or start etcd-tester manually
 - (OSX) make sure that etcd binary is built for linux/amd64 (eg. `rm bin/etcd;GOOS=linux GOARCH=amd64 ./tools/functional-tester/test`) otherwise you get `exec format error`
+
+
+## with Goreman
+
+To run the functional tests on a single machine using Goreman, build with the provided build script and run with the provided Procfile:
+
+```sh
+./tools/functional-tester/build
+goreman -f tools/functional-tester/Procfile
+```
+
+Notes:
+- The etcd-agent will not run with root privileges; iptables manipulation is disabled.
+- To specify testing parameters (etcd-tester arguments) modify tools/functional-tester/Procfile or start etcd-tester manually

tools/functional-tester/etcd-agent/agent.go

@@ -39,36 +39,44 @@ type Agent struct {
 
 	cmd     *exec.Cmd
 	logfile *os.File
-	logDir  string
+
+	cfg AgentConfig
+}
+
+type AgentConfig struct {
+	EtcdPath      string
+	LogDir        string
+	FailpointAddr string
+	UseRoot       bool
 }
 
-func newAgent(etcd, logDir string) (*Agent, error) {
+func newAgent(cfg AgentConfig) (*Agent, error) {
 	// check if the file exists
-	_, err := os.Stat(etcd)
+	_, err := os.Stat(cfg.EtcdPath)
 	if err != nil {
 		return nil, err
 	}
 
-	c := exec.Command(etcd)
+	c := exec.Command(cfg.EtcdPath)
 
-	err = fileutil.TouchDirAll(logDir)
+	err = fileutil.TouchDirAll(cfg.LogDir)
 	if err != nil {
 		return nil, err
 	}
 
 	var f *os.File
-	f, err = os.Create(filepath.Join(logDir, "etcd.log"))
+	f, err = os.Create(filepath.Join(cfg.LogDir, "etcd.log"))
 	if err != nil {
 		return nil, err
 	}
 
-	return &Agent{state: stateUninitialized, cmd: c, logfile: f, logDir: logDir}, nil
+	return &Agent{state: stateUninitialized, cmd: c, logfile: f, cfg: cfg}, nil
 }
 
 // start starts a new etcd process with the given args.
 func (a *Agent) start(args ...string) error {
 	a.cmd = exec.Command(a.cmd.Path, args...)
-	a.cmd.Env = []string{"GOFAIL_HTTP=:2381"}
+	a.cmd.Env = []string{"GOFAIL_HTTP=" + a.cfg.FailpointAddr}
 	a.cmd.Stdout = a.logfile
 	a.cmd.Stderr = a.logfile
 	err := a.cmd.Start()
@@ -131,15 +139,15 @@ func (a *Agent) cleanup() error {
 	a.state = stateUninitialized
 
 	a.logfile.Close()
-	if err := archiveLogAndDataDir(a.logDir, a.dataDir()); err != nil {
+	if err := archiveLogAndDataDir(a.cfg.LogDir, a.dataDir()); err != nil {
 		return err
 	}
 
-	if err := fileutil.TouchDirAll(a.logDir); err != nil {
+	if err := fileutil.TouchDirAll(a.cfg.LogDir); err != nil {
 		return err
 	}
 
-	f, err := os.Create(filepath.Join(a.logDir, "etcd.log"))
+	f, err := os.Create(filepath.Join(a.cfg.LogDir, "etcd.log"))
 	if err != nil {
 		return err
 	}
@@ -170,14 +178,23 @@ func (a *Agent) terminate() error {
 }
 
 func (a *Agent) dropPort(port int) error {
+	if !a.cfg.UseRoot {
+		return nil
+	}
 	return netutil.DropPort(port)
 }
 
 func (a *Agent) recoverPort(port int) error {
+	if !a.cfg.UseRoot {
+		return nil
+	}
 	return netutil.RecoverPort(port)
 }
 
 func (a *Agent) setLatency(ms, rv int) error {
+	if !a.cfg.UseRoot {
+		return nil
+	}
 	if ms == 0 {
 		return netutil.RemoveLatency()
 	}

tools/functional-tester/etcd-agent/agent_test.go

@@ -79,7 +79,7 @@ func TestAgentTerminate(t *testing.T) {
 
 // newTestAgent creates a test agent and with a temp data directory.
 func newTestAgent(t *testing.T) (*Agent, string) {
-	a, err := newAgent(etcdPath, "etcd.log")
+	a, err := newAgent(AgentConfig{EtcdPath: etcdPath, LogDir: "etcd.log"})
 	if err != nil {
 		t.Fatal(err)
 	}

tools/functional-tester/etcd-agent/main.go

@@ -16,6 +16,7 @@ package main
 
 import (
 	"flag"
+	"fmt"
 	"os"
 	"path/filepath"
 
@@ -28,9 +29,26 @@ func main() {
 	etcdPath := flag.String("etcd-path", filepath.Join(os.Getenv("GOPATH"), "bin/etcd"), "the path to etcd binary")
 	etcdLogDir := flag.String("etcd-log-dir", "etcd-log", "directory to store etcd logs")
 	port := flag.String("port", ":9027", "port to serve agent server")
+	useRoot := flag.Bool("use-root", true, "use root permissions")
+	failpointAddr := flag.String("failpoint-addr", ":2381", "interface for gofail's HTTP server")
 	flag.Parse()
 
-	a, err := newAgent(*etcdPath, *etcdLogDir)
+	cfg := AgentConfig{
+		EtcdPath:      *etcdPath,
+		LogDir:        *etcdLogDir,
+		FailpointAddr: *failpointAddr,
+		UseRoot:       *useRoot,
+	}
+
+	if *useRoot && os.Getuid() != 0 {
+		fmt.Println("got --use-root=true but not root user")
+		os.Exit(1)
+	}
+	if *useRoot == false {
+		fmt.Println("root permissions disabled, agent will not modify network")
+	}
+
+	a, err := newAgent(cfg)
 	if err != nil {
 		plog.Fatal(err)
 	}

tools/functional-tester/etcd-agent/rpc_test.go

@@ -25,7 +25,7 @@ import (
 )
 
 func init() {
-	defaultAgent, err := newAgent(etcdPath, "etcd.log")
+	defaultAgent, err := newAgent(AgentConfig{EtcdPath: etcdPath, LogDir: "etcd.log"})
 	if err != nil {
 		log.Panic(err)
 	}

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