@@ -16,7 +16,7 @@ if [ -d etcd ]; then

 fi

 etcd_version=$(go run ${OS_ROOT}/hack/version.go ${OS_ROOT}/Godeps/Godeps.json \

-  github.com/coreos/etcd/server)

+  github.com/coreos/etcd/etcdserver)

 mkdir -p etcd && cd etcd

@@ -232,7 +232,8 @@ function start_etcd {

   etcd -name test -data-dir ${ETCD_DIR} -bind-addr ${host}:${port} ${initial_cluster} >/dev/null 2>/dev/null &

   export ETCD_PID=$!

-  wait_for_url "http://${host}:${port}/version" "etcd: "

+  wait_for_url "http://${host}:${port}/version" "etcd: " 0.25 80

+  curl -X PUT  "http://${host}:${port}/v2/keys/_test"

 }

 # stop_openshift_server utility function to terminate an

@@ -4,13 +4,10 @@ import (

 	"fmt"

 	"time"

-	//etcdconfig "github.com/coreos/etcd/config"

-	_ "github.com/coreos/etcd/etcdmain"

 	etcdclient "github.com/coreos/go-etcd/etcd"

 	"github.com/golang/glog"

 	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"

-	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"

 	"github.com/openshift/origin/pkg/api/latest"

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

@@ -18,40 +15,60 @@ import (

 // RunEtcd starts an etcd server and runs it forever

 func RunEtcd(etcdServerConfig *configapi.EtcdConfig) {

+	cfg := &config{

+		name: defaultName,

+		dir:  etcdServerConfig.StorageDir,

-	/*config := etcdconfig.New()

-

-	config.Addr = etcdServerConfig.Address

-	config.BindAddr = etcdServerConfig.ServingInfo.BindAddress

+		TickMs:       100,

+		ElectionMs:   1000,

+		maxSnapFiles: 5,

+		maxWalFiles:  5,

+		initialClusterToken: "etcd-cluster",

+	}

+	var err error

 	if configapi.UseTLS(etcdServerConfig.ServingInfo) {

-		config.CAFile = etcdServerConfig.ServingInfo.ClientCA

-		config.CertFile = etcdServerConfig.ServingInfo.ServerCert.CertFile

-		config.KeyFile = etcdServerConfig.ServingInfo.ServerCert.KeyFile

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

 	}

-

-	config.Peer.Addr = etcdServerConfig.PeerAddress

-	config.Peer.BindAddr = etcdServerConfig.PeerServingInfo.BindAddress

 	if configapi.UseTLS(etcdServerConfig.PeerServingInfo) {

-		config.Peer.CAFile = etcdServerConfig.PeerServingInfo.ClientCA

-		config.Peer.CertFile = etcdServerConfig.PeerServingInfo.ServerCert.CertFile

-		config.Peer.KeyFile = etcdServerConfig.PeerServingInfo.ServerCert.KeyFile

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

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

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

+	}

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

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

+	}

+

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

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

+	}

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

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

+	}

+

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

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

 	}

-	config.DataDir = etcdServerConfig.StorageDir

-	config.Name = "openshift.local"

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

-	server := etcd.New(config)

-	go util.Forever(func() {

-		glog.Infof("Started etcd at %s", config.Addr)

-		server.Run()

-		glog.Fatalf("etcd died, exiting.")

-	}, 500*time.Millisecond)

-	<-server.ReadyNotify()*/

+	stopped, err := startEtcd(cfg)

+	if err != nil {

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

+	}

+	go func() {

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

+		<-stopped

+	}()

 }

-// getAndTestEtcdClient creates an etcd client based on the provided config and waits

+// GetAndTestEtcdClient creates an etcd client based on the provided config and waits

 // until etcd server is reachable. It errors out and exits if the server cannot

 // be reached for a certain amount of time.

 func GetAndTestEtcdClient(etcdClientInfo configapi.EtcdConnectionInfo) (*etcdclient.Client, error) {

@@ -93,7 +110,7 @@ func GetAndTestEtcdClient(etcdClientInfo configapi.EtcdConnectionInfo) (*etcdcli

 	return etcdClient, nil

 }

-// newOpenShiftEtcdHelper returns an EtcdHelper for the provided arguments or an error if the version

+// NewOpenShiftEtcdHelper returns an EtcdHelper for the provided arguments or an error if the version

 // is incorrect.

 func NewOpenShiftEtcdHelper(etcdClientInfo configapi.EtcdConnectionInfo) (helper tools.EtcdHelper, err error) {

 	// Connect and setup etcd interfaces

new file mode 100644

@@ -0,0 +1,209 @@

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

+package etcd

+

+import (

+	"fmt"

+	"io/ioutil"

+	"log"

+	"net"

+	"net/http"

+	"net/url"

+	"strings"

+	"time"

+

+	"github.com/golang/glog"

+

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

+	"github.com/coreos/etcd/etcdserver/etcdhttp"

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

+	"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) {

+	cls, err := setupCluster(cfg)

+	if err != nil {

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

+	}

+

+	pt, err := transport.NewTimeoutTransport(cfg.peerTLSInfo, rafthttp.ConnReadTimeout, rafthttp.ConnWriteTimeout)

+	if err != nil {

+		return nil, 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

+		l, err = transport.NewTimeoutListener(u.Host, u.Scheme, cfg.peerTLSInfo, 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 {

+		var l net.Listener

+		l, err = transport.NewKeepAliveListener(u.Host, u.Scheme, cfg.clientTLSInfo)

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

+		MaxWALFiles:     cfg.maxWalFiles,

+		Cluster:         cls,

+		NewCluster:      true,

+		ForceNewCluster: false,

+		Transport:       pt,

+		TickMs:          cfg.TickMs,

+		ElectionTicks:   cfg.electionTicks(),

+	}

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

+	ph := etcdhttp.NewPeerHandler(s.Cluster, etcdserver.RaftTimer(s), s.RaftHandler())

+	// 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) (*etcdserver.Cluster, error) {

+	var cls *etcdserver.Cluster

+	var err error

+	switch {

+	default:

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

+		cls, err = etcdserver.NewClusterFromString(cfg.initialClusterToken, cfg.initialCluster)

+	}

+	return cls, 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 {

+	return netutil.ResolveTCPAddrs(cfg.lpurls, cfg.apurls, cfg.lcurls, cfg.acurls)

+}

+

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