@@ -1,64 +1,98 @@

 package etcdserver

 import (

-	"fmt"

+	"net/url"

+	"strings"

+	"time"

+	"github.com/coreos/etcd/embed"

+	"github.com/coreos/etcd/pkg/osutil"

+	"github.com/coreos/etcd/pkg/types"

 	"github.com/golang/glog"

 	configapi "github.com/openshift/origin/pkg/cmd/server/api"

 )

+const defaultName = "openshift.local"

+

 // RunEtcd starts an etcd server and runs it forever

 func RunEtcd(etcdServerConfig *configapi.EtcdConfig) {

-	cfg := &config{

-		name: defaultName,

-		dir:  etcdServerConfig.StorageDir,

-

-		TickMs:       100,

-		ElectionMs:   1000,

-		maxSnapFiles: 5,

-		maxWalFiles:  5,

+	cfg := embed.NewConfig()

+	cfg.Debug = true

+	cfg.Name = defaultName

+	cfg.Dir = etcdServerConfig.StorageDir

-		initialClusterToken: "etcd-cluster",

+	clientTLS := configapi.UseTLS(etcdServerConfig.ServingInfo)

+	if clientTLS {

+		cfg.ClientTLSInfo.CAFile = etcdServerConfig.ServingInfo.ClientCA

+		cfg.ClientTLSInfo.CertFile = etcdServerConfig.ServingInfo.ServerCert.CertFile

+		cfg.ClientTLSInfo.KeyFile = etcdServerConfig.ServingInfo.ServerCert.KeyFile

+		cfg.ClientTLSInfo.ClientCertAuth = len(cfg.ClientTLSInfo.CAFile) > 0

 	}

-	var err error

-	if configapi.UseTLS(etcdServerConfig.ServingInfo) {

-		cfg.clientTLSInfo.CAFile = etcdServerConfig.ServingInfo.ClientCA

-		cfg.clientTLSInfo.CertFile = etcdServerConfig.ServingInfo.ServerCert.CertFile

-		cfg.clientTLSInfo.KeyFile = etcdServerConfig.ServingInfo.ServerCert.KeyFile

-	}

-	if cfg.lcurls, err = urlsFromStrings(etcdServerConfig.ServingInfo.BindAddress, cfg.clientTLSInfo); err != nil {

-		glog.Fatalf("Unable to build etcd client URLs: %v", err)

+	u, err := types.NewURLs(addressToURLs(etcdServerConfig.ServingInfo.BindAddress, clientTLS))

+	if err != nil {

+		glog.Fatalf("Unable to build etcd peer URLs: %v", err)

 	}

+	cfg.LCUrls = []url.URL(u)

-	if configapi.UseTLS(etcdServerConfig.PeerServingInfo) {

-		cfg.peerTLSInfo.CAFile = etcdServerConfig.PeerServingInfo.ClientCA

-		cfg.peerTLSInfo.CertFile = etcdServerConfig.PeerServingInfo.ServerCert.CertFile

-		cfg.peerTLSInfo.KeyFile = etcdServerConfig.PeerServingInfo.ServerCert.KeyFile

+	peerTLS := configapi.UseTLS(etcdServerConfig.PeerServingInfo)

+	if peerTLS {

+		cfg.PeerTLSInfo.CAFile = etcdServerConfig.PeerServingInfo.ClientCA

+		cfg.PeerTLSInfo.CertFile = etcdServerConfig.PeerServingInfo.ServerCert.CertFile

+		cfg.PeerTLSInfo.KeyFile = etcdServerConfig.PeerServingInfo.ServerCert.KeyFile

+		cfg.PeerTLSInfo.ClientCertAuth = len(cfg.PeerTLSInfo.CAFile) > 0

 	}

-	if cfg.lpurls, err = urlsFromStrings(etcdServerConfig.PeerServingInfo.BindAddress, cfg.peerTLSInfo); err != nil {

+	u, err = types.NewURLs(addressToURLs(etcdServerConfig.PeerServingInfo.BindAddress, peerTLS))

+	if err != nil {

 		glog.Fatalf("Unable to build etcd peer URLs: %v", err)

 	}

+	cfg.LPUrls = []url.URL(u)

-	if cfg.acurls, err = urlsFromStrings(etcdServerConfig.Address, cfg.clientTLSInfo); err != nil {

+	u, err = types.NewURLs(addressToURLs(etcdServerConfig.Address, clientTLS))

+	if err != nil {

 		glog.Fatalf("Unable to build etcd announce client URLs: %v", err)

 	}

-	if cfg.apurls, err = urlsFromStrings(etcdServerConfig.PeerAddress, cfg.peerTLSInfo); err != nil {

+	cfg.ACUrls = []url.URL(u)

+

+	u, err = types.NewURLs(addressToURLs(etcdServerConfig.PeerAddress, peerTLS))

+	if err != nil {

 		glog.Fatalf("Unable to build etcd announce peer URLs: %v", err)

 	}

+	cfg.APUrls = []url.URL(u)

-	if err := cfg.resolveUrls(); err != nil {

-		glog.Fatalf("Unable to resolve etcd URLs: %v", err)

-	}

+	cfg.InitialCluster = cfg.InitialClusterFromName(cfg.Name)

-	cfg.initialCluster = fmt.Sprintf("%s=%s", cfg.name, cfg.apurls[0].String())

+	osutil.HandleInterrupts()

-	stopped, err := startEtcd(cfg)

+	e, err := embed.StartEtcd(cfg)

 	if err != nil {

 		glog.Fatalf("Unable to start etcd: %v", err)

 	}

+

 	go func() {

-		glog.Infof("Started etcd at %s", etcdServerConfig.Address)

-		<-stopped

+		defer e.Close()

+

+		select {

+		case <-e.Server.ReadyNotify():

+			glog.Infof("Started etcd at %s", etcdServerConfig.Address)

+		case <-time.After(60 * time.Second):

+			glog.Warning("etcd took too long to start, stopped")

+			e.Server.Stop() // trigger a shutdown

+		}

+		glog.Fatalf("etcd has returned an error: %v", <-e.Err())

 	}()

 }

+

+// addressToURLs turns a host:port comma delimited list into an array valid

+// URL strings with the appropriate prefix for the TLS mode.

+func addressToURLs(addr string, isTLS bool) []string {

+	addrs := strings.Split(addr, ",")

+	for i := range addrs {

+		if isTLS {

+			addrs[i] = "https://" + addrs[i]

+		} else {

+			addrs[i] = "http://" + addrs[i]

+		}

+	}

+	return addrs

+}

deleted file mode 100644

@@ -1,255 +0,0 @@

-// This is a somewhat faithful reproduction of etcdmain/etcd.go

-package etcdserver

-

-import (

-	"fmt"

-	"io/ioutil"

-	"log"

-	"net"

-	"net/http"

-	"net/url"

-	"strings"

-	"time"

-

-	"github.com/golang/glog"

-

-	"github.com/coreos/etcd/etcdserver"

-	etcdhttp "github.com/coreos/etcd/etcdserver/api/v2http"

-	"github.com/coreos/etcd/pkg/osutil"

-	"github.com/coreos/etcd/pkg/transport"

-	"github.com/coreos/etcd/pkg/types"

-	"github.com/coreos/etcd/rafthttp"

-)

-

-type config struct {

-	// member

-	dir            string

-	lpurls, lcurls []url.URL

-	maxSnapFiles   uint

-	maxWalFiles    uint

-	name           string

-	snapCount      uint64

-	// TODO: decouple tickMs and heartbeat tick (current heartbeat tick = 1).

-	// make ticks a cluster wide configuration.

-	TickMs     uint

-	ElectionMs uint

-

-	// clustering

-	apurls, acurls      []url.URL

-	initialCluster      string

-	initialClusterToken string

-

-	// security

-	clientTLSInfo, peerTLSInfo transport.TLSInfo

-}

-

-const (

-	// the owner can make/remove files inside the directory

-	defaultName = "openshift.local"

-)

-

-// startEtcd launches the etcd server and HTTP handlers for client/server communication.

-func startEtcd(cfg *config) (<-chan struct{}, error) {

-	initialPeers, token, err := setupCluster(cfg)

-	if err != nil {

-		return nil, fmt.Errorf("error setting up initial cluster: %v", err)

-	}

-

-	if !cfg.peerTLSInfo.Empty() {

-		glog.V(2).Infof("etcd: peerTLS: %s", cfg.peerTLSInfo)

-	}

-	plns := make([]net.Listener, 0)

-	for _, u := range cfg.lpurls {

-		var l net.Listener

-		peerTLSConfig, err := cfg.peerTLSInfo.ServerConfig()

-		if err != nil {

-			return nil, err

-		}

-		l, err = transport.NewTimeoutListener(u.Host, u.Scheme, peerTLSConfig, rafthttp.ConnReadTimeout, rafthttp.ConnWriteTimeout)

-		if err != nil {

-			return nil, err

-		}

-

-		urlStr := u.String()

-		glog.V(2).Info("etcd: listening for peers on ", urlStr)

-		defer func() {

-			if err != nil {

-				l.Close()

-				glog.V(2).Info("etcd: stopping listening for peers on ", urlStr)

-			}

-		}()

-		plns = append(plns, l)

-	}

-

-	if !cfg.clientTLSInfo.Empty() {

-		glog.V(2).Infof("etcd: clientTLS: %s", cfg.clientTLSInfo)

-	}

-	clns := make([]net.Listener, 0)

-	for _, u := range cfg.lcurls {

-		l, err := net.Listen("tcp", u.Host)

-		if err != nil {

-			return nil, err

-		}

-		clientTLSConfig, err := cfg.clientTLSInfo.ServerConfig()

-		if err != nil {

-			return nil, err

-		}

-		l, err = transport.NewKeepAliveListener(l, u.Scheme, clientTLSConfig)

-		if err != nil {

-			return nil, err

-		}

-

-		urlStr := u.String()

-		glog.V(2).Info("etcd: listening for client requests on ", urlStr)

-		defer func() {

-			if err != nil {

-				l.Close()

-				glog.V(2).Info("etcd: stopping listening for client requests on ", urlStr)

-			}

-		}()

-		clns = append(clns, l)

-	}

-

-	srvcfg := &etcdserver.ServerConfig{

-		Name:                cfg.name,

-		ClientURLs:          cfg.acurls,

-		PeerURLs:            cfg.apurls,

-		DataDir:             cfg.dir,

-		SnapCount:           cfg.snapCount,

-		MaxSnapFiles:        cfg.maxSnapFiles,

-		InitialPeerURLsMap:  initialPeers,

-		InitialClusterToken: token,

-		MaxWALFiles:         cfg.maxWalFiles,

-		NewCluster:          true,

-		ForceNewCluster:     false,

-		TickMs:              cfg.TickMs,

-		ElectionTicks:       cfg.electionTicks(),

-

-		PeerTLSInfo: cfg.peerTLSInfo,

-	}

-	var s *etcdserver.EtcdServer

-	s, err = etcdserver.NewServer(srvcfg)

-	if err != nil {

-		return nil, err

-	}

-	osutil.HandleInterrupts()

-	s.Start()

-	osutil.RegisterInterruptHandler(s.Stop)

-

-	ch := etcdhttp.NewClientHandler(s, srvcfg.ReqTimeout())

-	ph := etcdhttp.NewPeerHandler(s)

-	// Start the peer server in a goroutine

-	for _, l := range plns {

-		go func(l net.Listener) {

-			glog.Fatal(serveHTTP(l, ph, 5*time.Minute))

-		}(l)

-	}

-	// Start a client server goroutine for each listen address

-	for _, l := range clns {

-		go func(l net.Listener) {

-			// read timeout does not work with http close notify

-			// TODO: https://github.com/golang/go/issues/9524

-			glog.Fatal(serveHTTP(l, ch, 0))

-		}(l)

-	}

-	return s.StopNotify(), nil

-}

-

-// setupCluster sets up an initial cluster definition for bootstrap or discovery.

-func setupCluster(cfg *config) (types.URLsMap, string, error) {

-	// We're statically configured, and cluster has appropriately been set.

-	m, err := types.NewURLsMap(cfg.initialCluster)

-	return m, cfg.initialClusterToken, err

-}

-

-func genClusterString(name string, urls types.URLs) string {

-	addrs := make([]string, 0)

-	for _, u := range urls {

-		addrs = append(addrs, fmt.Sprintf("%v=%v", name, u.String()))

-	}

-	return strings.Join(addrs, ",")

-}

-

-func initialClusterFromName(name string) string {

-	n := name

-	if name == "" {

-		n = defaultName

-	}

-	return fmt.Sprintf("%s=http://localhost:7001", n)

-}

-

-func urlsFromStrings(input string, tlsInfo transport.TLSInfo) ([]url.URL, error) {

-	urls := []url.URL{}

-	for _, addr := range strings.Split(input, ",") {

-		addrURL := url.URL{Scheme: "http", Host: addr}

-		if !tlsInfo.Empty() {

-			addrURL.Scheme = "https"

-		}

-		urls = append(urls, addrURL)

-	}

-	return urls, nil

-}

-

-// serveHTTP accepts incoming HTTP connections on the listener l,

-// creating a new service goroutine for each. The service goroutines

-// read requests and then call handler to reply to them.

-func serveHTTP(l net.Listener, handler http.Handler, readTimeout time.Duration) error {

-	logger := log.New(ioutil.Discard, "etcdhttp", 0)

-	// TODO: add debug flag; enable logging when debug flag is set

-	srv := &http.Server{

-		Handler:     handler,

-		ReadTimeout: readTimeout,

-		ErrorLog:    logger, // do not log user error

-	}

-	return srv.Serve(l)

-}

-

-func (cfg *config) resolveUrls() error {

-	out, err := resolveTCPAddrs([][]url.URL{cfg.lpurls, cfg.apurls, cfg.lcurls, cfg.acurls})

-	if err != nil {

-		return err

-	}

-	cfg.lpurls, cfg.apurls, cfg.lcurls, cfg.acurls = out[0], out[1], out[2], out[3]

-	return nil

-}

-

-func (cfg config) electionTicks() int { return int(cfg.ElectionMs / cfg.TickMs) }

-

-// resolveTCPAddrs is a convenience wrapper for net.ResolveTCPAddr.

-// resolveTCPAddrs return a new set of url.URLs, in which all DNS hostnames

-// are resolved.

-func resolveTCPAddrs(urls [][]url.URL) ([][]url.URL, error) {

-	newurls := make([][]url.URL, 0)

-	for _, us := range urls {

-		nus := make([]url.URL, len(us))

-		for i, u := range us {

-			nu, err := url.Parse(u.String())

-			if err != nil {

-				return nil, err

-			}

-			nus[i] = *nu

-		}

-		for i, u := range nus {

-			host, _, err := net.SplitHostPort(u.Host)

-			if err != nil {

-				glog.Errorf("could not parse url %s during tcp resolving", u.Host)

-				return nil, err

-			}

-			if host == "localhost" {

-				continue

-			}

-			if net.ParseIP(host) != nil {

-				continue

-			}

-			tcpAddr, err := net.ResolveTCPAddr("tcp", u.Host)

-			if err != nil {

-				glog.Errorf("could not resolve host %s", u.Host)

-				return nil, err

-			}

-			glog.V(4).Infof("resolving %s to %s", u.Host, tcpAddr.String())

-			nus[i].Host = tcpAddr.String()

-		}

-		newurls = append(newurls, nus)

-	}

-	return newurls, nil

-}

@@ -30,12 +30,12 @@ type EtcdOptions struct {

 const etcdLong = `Start an etcd server for testing.

 This command starts an etcd server based on the config for testing.  It is not

-Intended for production use.  Running

+intended for production use.  Running

   %[1]s start %[2]s

-will start the server listening for incoming requests. The server

-will run in the foreground until you terminate the process.`

+will start the server listening for incoming requests. The server will run in

+the foreground until you terminate the process.`

 // NewCommandStartEtcdServer starts only the etcd server

 func NewCommandStartEtcdServer(name, basename string, out io.Writer) (*cobra.Command, *EtcdOptions) {