Use etcd embed package, switch apiserver to use etcdv3

cmd/localkube/cmd/start.go

@@ -90,7 +90,7 @@ func SetupServer(s *localkube.LocalkubeServer) {
 	capabilities.Initialize(c)
 
 	// setup etcd
-	etcd, err := s.NewEtcd(localkube.KubeEtcdClientURLs, localkube.KubeEtcdPeerURLs, "kubeetcd", s.GetEtcdDataDirectory())
+	etcd, err := s.NewEtcd(s.GetEtcdDataDirectory())
 	if err != nil {
 		panic(err)
 	}

pkg/localkube/apiserver.go

@@ -21,6 +21,8 @@ import (
 	"path"
 	"strconv"
 
+	"github.com/coreos/etcd/embed"
+
 	apiserveroptions "k8s.io/apiserver/pkg/server/options"
 	"k8s.io/apiserver/pkg/storage/storagebackend"
 
@@ -55,8 +57,8 @@ func StartAPIServer(lk LocalkubeServer) func() error {
 	}
 	// use localkube etcd
 
-	config.Etcd.StorageConfig.ServerList = KubeEtcdClientURLs
-	config.Etcd.StorageConfig.Type = storagebackend.StorageTypeETCD2
+	config.Etcd.StorageConfig.ServerList = []string{embed.DefaultListenClientURLs}
+	config.Etcd.StorageConfig.Type = storagebackend.StorageTypeETCD3
 
 	// set Service IP range
 	config.ServiceClusterIPRange = lk.ServiceClusterIPRange

pkg/localkube/etcd.go

@@ -17,138 +17,59 @@ limitations under the License.
 package localkube
 
 import (
-	"crypto/tls"
-	"fmt"
-	"net"
-	"net/http"
 	"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/etcd/embed"
+	"github.com/golang/glog"
 )
 
 const (
+	// EtcdName is the name of the extra-config component for etcd
 	EtcdName = "etcd"
 )
 
-var (
-	// EtcdClientURLs have listeners created and handle etcd API traffic
-	KubeEtcdClientURLs = []string{"http://0.0.0.0:2379"}
-
-	// EtcdPeerURLs don't have listeners created for them, they are used to pass Etcd validation
-	KubeEtcdPeerURLs = []string{"http://0.0.0.0:2380"}
-)
-
-// Etcd is a Server which manages an Etcd cluster
+// EtcdServer is a Server which manages an Etcd cluster
 type EtcdServer struct {
-	*etcdserver.EtcdServer
-	config          *etcdserver.ServerConfig
-	clientListeners []net.Listener
+	Etcd   *embed.Etcd
+	Config *embed.Config
 }
 
 // NewEtcd creates a new default etcd Server using 'dataDir' for persistence. Panics if could not be configured.
-func (lk LocalkubeServer) NewEtcd(clientURLStrs, peerURLStrs []string, name, dataDirectory string) (*EtcdServer, error) {
-	clientURLs, err := types.NewURLs(clientURLStrs)
-	if err != nil {
-		return nil, err
-	}
-
-	peerURLs, err := types.NewURLs(peerURLStrs)
-	if err != nil {
-		return nil, err
-	}
-
-	urlsMap := map[string]types.URLs{
-		name: peerURLs,
-	}
-
-	config := &etcdserver.ServerConfig{
-		Name:               name,
-		ClientURLs:         clientURLs,
-		PeerURLs:           peerURLs,
-		DataDir:            dataDirectory,
-		InitialPeerURLsMap: urlsMap,
-
-		NewCluster: true,
-
-		SnapCount:     etcdserver.DefaultSnapCount,
-		MaxSnapFiles:  5,
-		MaxWALFiles:   5,
-		TickMs:        100,
-		ElectionTicks: 10,
-	}
-
-	lk.SetExtraConfigForComponent(EtcdName, &config)
+func (lk LocalkubeServer) NewEtcd(dataDir string) (*EtcdServer, error) {
+	cfg := embed.NewConfig()
+	cfg.Dir = dataDir
 
+	lk.SetExtraConfigForComponent(EtcdName, &cfg)
 	return &EtcdServer{
-		config: config,
+		Config: cfg,
 	}, nil
 }
 
-// Starts starts the etcd server and listening for client connections
+// Start starts the etcd server and listening for client connections
 func (e *EtcdServer) Start() {
 	var err error
-	e.EtcdServer, err = etcdserver.NewServer(e.config)
+	e.Etcd, err = embed.StartEtcd(e.Config)
 	if err != nil {
-		msg := fmt.Sprintf("Etcd config error: %v", err)
-		panic(msg)
+		glog.Fatalf("Error starting up etcd: %s", err)
 	}
 
-	// create client listeners
-	e.clientListeners = createListenersOrPanic(e.config.ClientURLs)
-
-	// start etcd
-	e.EtcdServer.Start()
-
-	// setup client listeners
-	ch := v2http.NewClientHandler(e.EtcdServer, e.requestTimeout())
-	for _, l := range e.clientListeners {
-		go func(l net.Listener) {
-			srv := &http.Server{
-				Handler:     ch,
-				ReadTimeout: 5 * time.Minute,
-			}
-			panic(srv.Serve(l))
-		}(l)
+	select {
+	case <-e.Etcd.Server.ReadyNotify():
+		glog.Infoln("Etcd server is ready")
+	case <-time.After(60 * time.Second):
+		e.Etcd.Server.Stop() // trigger a shutdown
+		glog.Fatalf("Etcd took too long to start")
 	}
 }
 
 // Stop closes all connections and stops the Etcd server
 func (e *EtcdServer) Stop() {
-	if e.EtcdServer != nil {
-		e.EtcdServer.Stop()
-	}
-
-	for _, l := range e.clientListeners {
-		l.Close()
+	if e.Etcd != nil {
+		e.Etcd.Server.Stop()
 	}
 }
 
 // Name returns the servers unique name
-func (EtcdServer) Name() string {
-	return EtcdName
-}
-
-func (e *EtcdServer) requestTimeout() time.Duration {
-	// from github.com/coreos/etcd/etcdserver/config.go
-	return 5*time.Second + 2*time.Duration(e.config.ElectionTicks)*time.Duration(e.config.TickMs)*time.Millisecond
-}
-
-func createListenersOrPanic(urls types.URLs) (listeners []net.Listener) {
-	for _, url := range urls {
-		l, err := net.Listen("tcp", url.Host)
-		if err != nil {
-			panic(err)
-		}
-
-		l, err = transport.NewKeepAliveListener(l, url.Scheme, &tls.Config{})
-		if err != nil {
-			panic(err)
-		}
-
-		listeners = append(listeners, l)
-	}
-	return listeners
+func (e EtcdServer) Name() string {
+	return e.Name()
 }