Statically link etcd into node e2e.

test/e2e_node/e2e_service.go

@@ -98,7 +98,7 @@ func (e *E2EServices) Stop() error {
 		if manifestPath != "" {
 			err := os.RemoveAll(manifestPath)
 			if err != nil {
-				glog.Errorf("Failed to delete static pod manifest directory %s.\n%v", manifestPath, err)
+				glog.Errorf("Failed to delete static pod manifest directory %s: %v", manifestPath, err)
 			}
 		}
 	}()
@@ -119,7 +119,6 @@ func RunE2EServices() {
 
 // Ports of different e2e services.
 const (
-	etcdPort            = "4001"
 	apiserverPort       = "8080"
 	kubeletPort         = "10250"
 	kubeletReadOnlyPort = "10255"
@@ -127,7 +126,6 @@ const (
 
 // Health check urls of different e2e services.
 var (
-	etcdHealthCheckURL      = getEndpoint(etcdPort) + "/v2/keys/" // Trailing slash is required,
 	apiserverHealthCheckURL = getEndpoint(apiserverPort) + "/healthz"
 	kubeletHealthCheckURL   = getEndpoint(kubeletReadOnlyPort) + "/healthz"
 )
@@ -139,7 +137,7 @@ func getEndpoint(port string) string {
 
 func getHealthCheckURLs() []string {
 	return []string{
-		etcdHealthCheckURL,
+		getEtcdHealthCheckURL(),
 		apiserverHealthCheckURL,
 		kubeletHealthCheckURL,
 	}
@@ -150,6 +148,9 @@ type e2eService struct {
 	services []*server
 	rmDirs   []string
 	logFiles map[string]logFileData
+
+	// All statically linked e2e services
+	etcdServer *EtcdServer
 }
 
 type logFileData struct {
@@ -201,13 +202,12 @@ func (es *e2eService) start() error {
 		return err
 	}
 
-	s, err := es.startEtcd()
+	err := es.startEtcd()
 	if err != nil {
 		return err
 	}
-	es.services = append(es.services, s)
 
-	s, err = es.startApiServer()
+	s, err := es.startApiServer()
 	if err != nil {
 		return err
 	}
@@ -284,6 +284,10 @@ func (es *e2eService) stop() {
 			glog.Errorf("Failed to stop %v: %v", s.name, err)
 		}
 	}
+	// TODO(random-liu): Use a loop to stop all services after introducing service interface.
+	if err := es.etcdServer.Stop(); err != nil {
+		glog.Errorf("Failed to stop %q: %v", es.etcdServer.Name(), err)
+	}
 	for _, d := range es.rmDirs {
 		err := os.RemoveAll(d)
 		if err != nil {
@@ -292,45 +296,20 @@ func (es *e2eService) stop() {
 	}
 }
 
-func (es *e2eService) startEtcd() (*server, error) {
+func (es *e2eService) startEtcd() error {
 	dataDir, err := ioutil.TempDir("", "node-e2e")
 	if err != nil {
-		return nil, err
+		return err
 	}
 	// Mark the dataDir as directories to remove.
 	es.rmDirs = append(es.rmDirs, dataDir)
-	var etcdPath string
+	es.etcdServer = NewEtcd(dataDir)
-	// CoreOS ships a binary named 'etcd' which is really old, so prefer 'etcd2' if it exists
+	return es.etcdServer.Start()
-	etcdPath, err = exec.LookPath("etcd2")
-	if err != nil {
-		etcdPath, err = exec.LookPath("etcd")
-	}
-	if err != nil {
-		glog.Infof("etcd not found in PATH. Defaulting to %s...", defaultEtcdPath)
-		_, err = os.Stat(defaultEtcdPath)
-		if err != nil {
-			return nil, fmt.Errorf("etcd binary not found")
-		}
-		etcdPath = defaultEtcdPath
-	}
-	cmd := exec.Command(etcdPath,
-		"--listen-client-urls=http://0.0.0.0:2379,http://0.0.0.0:4001",
-		"--advertise-client-urls=http://0.0.0.0:2379,http://0.0.0.0:4001")
-	// Execute etcd in the data directory instead of using --data-dir because the flag sometimes requires additional
-	// configuration (e.g. --name in version 0.4.9)
-	cmd.Dir = dataDir
-	server := newServer(
-		"etcd",
-		cmd,
-		nil,
-		[]string{etcdHealthCheckURL},
-		"etcd.log")
-	return server, server.start()
 }
 
 func (es *e2eService) startApiServer() (*server, error) {
 	cmd := exec.Command("sudo", getApiServerBin(),
-		"--etcd-servers", getEndpoint(etcdPort),
+		"--etcd-servers", getEtcdClientURL(),
 		"--insecure-bind-address", "0.0.0.0",
 		"--service-cluster-ip-range", "10.0.0.1/24",
 		"--kubelet-port", kubeletPort,

test/e2e_node/etcd.go

@@ -0,0 +1,165 @@
+/*
+Copyright 2016 The Kubernetes Authors.
+
+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,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package e2e_node
+
+import (
+	"crypto/tls"
+	"net"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/coreos/etcd/etcdserver"
+	"github.com/coreos/etcd/etcdserver/api/v2http"
+	"github.com/coreos/etcd/pkg/transport"
+	"github.com/coreos/etcd/pkg/types"
+	"github.com/coreos/pkg/capnslog"
+	"github.com/golang/glog"
+)
+
+// TODO(random-liu): Add service interface to manage services with the same behaviour.
+
+func init() {
+	// github.com/coreos/etcd/etcdserver/api package is too spammy, set the log level to NOTICE.
+	capnslog.MustRepoLogger("github.com/coreos/etcd/etcdserver/api").SetRepoLogLevel(capnslog.NOTICE)
+}
+
+// All following configurations are got from etcd source code.
+// TODO(random-liu): Use embed.NewConfig after etcd3 is supported.
+const (
+	etcdName           = "etcd"
+	clientURLStr       = "http://localhost:4001" // clientURL has listener created and handles etcd API traffic
+	peerURLStr         = "http://localhost:7001" // peerURL does't have listener created, it is used to pass Etcd validation
+	snapCount          = etcdserver.DefaultSnapCount
+	maxSnapFiles       = 5
+	maxWALFiles        = 5
+	tickMs             = 100
+	electionTicks      = 10
+	etcdHealthCheckURL = clientURLStr + "/v2/keys/" // Trailing slash is required,
+)
+
+// EtcdServer is a server which manages etcd.
+type EtcdServer struct {
+	*etcdserver.EtcdServer
+	config       *etcdserver.ServerConfig
+	clientListen net.Listener
+}
+
+// NewEtcd creates a new default etcd server using 'dataDir' for persistence.
+func NewEtcd(dataDir string) *EtcdServer {
+	clientURLs, err := types.NewURLs([]string{clientURLStr})
+	if err != nil {
+		glog.Fatalf("Failed to parse client url %q: %v", clientURLStr, err)
+	}
+	peerURLs, err := types.NewURLs([]string{peerURLStr})
+	if err != nil {
+		glog.Fatalf("Failed to parse peer url %q: %v", peerURLStr, err)
+	}
+
+	config := &etcdserver.ServerConfig{
+		Name:               etcdName,
+		ClientURLs:         clientURLs,
+		PeerURLs:           peerURLs,
+		DataDir:            dataDir,
+		InitialPeerURLsMap: map[string]types.URLs{etcdName: peerURLs},
+		NewCluster:         true,
+		SnapCount:          snapCount,
+		MaxSnapFiles:       maxSnapFiles,
+		MaxWALFiles:        maxWALFiles,
+		TickMs:             tickMs,
+		ElectionTicks:      electionTicks,
+	}
+
+	return &EtcdServer{
+		config: config,
+	}
+}
+
+// Start starts the etcd server and listening for client connections
+func (e *EtcdServer) Start() error {
+	var err error
+	e.EtcdServer, err = etcdserver.NewServer(e.config)
+	if err != nil {
+		return err
+	}
+	// create client listener, there should be only one url
+	e.clientListen, err = createListener(e.config.ClientURLs[0])
+	if err != nil {
+		return err
+	}
+
+	// start etcd
+	e.EtcdServer.Start()
+
+	// setup client listener
+	ch := v2http.NewClientHandler(e.EtcdServer, e.config.ReqTimeout())
+	errCh := make(chan error)
+	go func(l net.Listener) {
+		defer close(errCh)
+		srv := &http.Server{
+			Handler:     ch,
+			ReadTimeout: 5 * time.Minute,
+		}
+		// Serve always returns a non-nil error.
+		errCh <- srv.Serve(l)
+	}(e.clientListen)
+
+	err = readinessCheck([]string{etcdHealthCheckURL}, errCh)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// Stop closes all connections and stops the Etcd server
+func (e *EtcdServer) Stop() error {
+	if e.EtcdServer != nil {
+		e.EtcdServer.Stop()
+	}
+	if e.clientListen != nil {
+		err := e.clientListen.Close()
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Name returns the server's unique name
+func (e *EtcdServer) Name() string {
+	return etcdName
+}
+
+func createListener(url url.URL) (net.Listener, error) {
+	l, err := net.Listen("tcp", url.Host)
+	if err != nil {
+		return nil, err
+	}
+	l, err = transport.NewKeepAliveListener(l, url.Scheme, &tls.Config{})
+	if err != nil {
+		return nil, err
+	}
+	return l, nil
+}
+
+func getEtcdClientURL() string {
+	return clientURLStr
+}
+
+func getEtcdHealthCheckURL() string {
+	return etcdHealthCheckURL
+}