Modifications to to remove FakeEtcdClient. Enables starting & stopping

an etcd server per unit tests that need them.

Godeps/Godeps.json

@@ -180,11 +180,6 @@
 			"Comment": "v2.2.1-1-g4dc835c",
 			"Rev": "4dc835c718bbdbb9a1c36ef5cdf1921a423cbf70"
 		},
-		{
-			"ImportPath": "github.com/coreos/etcd/pkg/testutil",
-			"Comment": "v2.2.0-17-g45c86af",
-			"Rev": "45c86af0eb195f6f833cab6fb176a60fc8c47185"
-		},
 		{
 			"ImportPath": "github.com/coreos/etcd/pkg/timeutil",
 			"Comment": "v2.2.1-1-g4dc835c",

Godeps/_workspace/src/github.com/coreos/etcd/pkg/testutil/pauseable_handler.go

@@ -1,57 +0,0 @@
-// Copyright 2015 CoreOS, Inc.
-//
-// 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 testutil
-
-import (
-	"net/http"
-	"sync"
-)
-
-type PauseableHandler struct {
-	Next   http.Handler
-	mu     sync.Mutex
-	paused bool
-}
-
-func (ph *PauseableHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	ph.mu.Lock()
-	paused := ph.paused
-	ph.mu.Unlock()
-	if !paused {
-		ph.Next.ServeHTTP(w, r)
-	} else {
-		hj, ok := w.(http.Hijacker)
-		if !ok {
-			panic("webserver doesn't support hijacking")
-		}
-		conn, _, err := hj.Hijack()
-		if err != nil {
-			panic(err.Error())
-		}
-		conn.Close()
-	}
-}
-
-func (ph *PauseableHandler) Pause() {
-	ph.mu.Lock()
-	defer ph.mu.Unlock()
-	ph.paused = true
-}
-
-func (ph *PauseableHandler) Resume() {
-	ph.mu.Lock()
-	defer ph.mu.Unlock()
-	ph.paused = false
-}

Godeps/_workspace/src/github.com/coreos/etcd/pkg/testutil/recorder.go

@@ -1,40 +0,0 @@
-// Copyright 2015 CoreOS, Inc.
-//
-// 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 testutil
-
-import "sync"
-
-type Action struct {
-	Name   string
-	Params []interface{}
-}
-
-type Recorder struct {
-	sync.Mutex
-	actions []Action
-}
-
-func (r *Recorder) Record(a Action) {
-	r.Lock()
-	r.actions = append(r.actions, a)
-	r.Unlock()
-}
-func (r *Recorder) Action() []Action {
-	r.Lock()
-	cpy := make([]Action, len(r.actions))
-	copy(cpy, r.actions)
-	r.Unlock()
-	return cpy
-}

Godeps/_workspace/src/github.com/coreos/etcd/pkg/testutil/testutil.go

@@ -1,46 +0,0 @@
-// Copyright 2015 CoreOS, Inc.
-//
-// 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 testutil
-
-import (
-	"net/url"
-	"testing"
-	"time"
-)
-
-// TODO: improve this when we are able to know the schedule or status of target go-routine.
-func WaitSchedule() {
-	time.Sleep(10 * time.Millisecond)
-}
-
-func MustNewURLs(t *testing.T, urls []string) []url.URL {
-	if urls == nil {
-		return nil
-	}
-	var us []url.URL
-	for _, url := range urls {
-		u := MustNewURL(t, url)
-		us = append(us, *u)
-	}
-	return us
-}
-
-func MustNewURL(t *testing.T, s string) *url.URL {
-	u, err := url.Parse(s)
-	if err != nil {
-		t.Fatalf("parse %v error: %v", s, err)
-	}
-	return u
-}

pkg/storage/etcd/etcd_helper_test.go

@@ -405,27 +405,28 @@ func TestGetNotFoundErr(t *testing.T) {
 
 func TestCreate(t *testing.T) {
 	obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
-	fakeClient := tools.NewFakeEtcdClient(t)
-	helper := newEtcdHelper(fakeClient, testapi.Default.Codec(), etcdtest.PathPrefix())
+	server := NewEtcdTestClientServer(t)
+	defer server.Terminate(t)
+	helper := newEtcdHelper(server.client, testapi.Default.Codec(), etcdtest.PathPrefix())
 	returnedObj := &api.Pod{}
 	err := helper.Create(context.TODO(), "/some/key", obj, returnedObj, 5)
 	if err != nil {
 		t.Errorf("Unexpected error %#v", err)
 	}
-	data, err := testapi.Default.Codec().Encode(obj)
+	_, err = testapi.Default.Codec().Encode(obj)
 	if err != nil {
 		t.Errorf("Unexpected error %#v", err)
 	}
-	key := etcdtest.AddPrefix("/some/key")
-	node := fakeClient.Data[key].R.Node
-	if e, a := string(data), node.Value; e != a {
-		t.Errorf("Wanted %v, got %v", e, a)
+	err = helper.Get(context.TODO(), "/some/key", returnedObj, false)
+	if err != nil {
+		t.Errorf("Unexpected error %#v", err)
 	}
-	if e, a := uint64(5), fakeClient.LastSetTTL; e != a {
-		t.Errorf("Wanted %v, got %v", e, a)
+	_, err = testapi.Default.Codec().Encode(returnedObj)
+	if err != nil {
+		t.Errorf("Unexpected error %#v", err)
 	}
-	if obj.ResourceVersion != returnedObj.ResourceVersion || obj.Name != returnedObj.Name {
-		t.Errorf("If set was successful but returned object did not have correct resource version")
+	if obj.Name != returnedObj.Name {
+		t.Errorf("Wanted %v, got %v", obj.Name, returnedObj.Name)
 	}
 }
 

pkg/storage/etcd/etcd_test_util.go

@@ -0,0 +1,157 @@
+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+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 etcd
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+	"time"
+
+	"k8s.io/kubernetes/pkg/tools"
+
+	"github.com/coreos/etcd/etcdserver"
+	"github.com/coreos/etcd/etcdserver/etcdhttp"
+	"github.com/coreos/etcd/pkg/transport"
+	"github.com/coreos/etcd/pkg/types"
+	"github.com/coreos/etcd/rafthttp"
+	goetcd "github.com/coreos/go-etcd/etcd"
+)
+
+// EtcdTestServer encapsulates the datastructures needed to start local instance for testing
+type EtcdTestServer struct {
+	etcdserver.ServerConfig
+	PeerListeners, ClientListeners []net.Listener
+	client                         tools.EtcdClient
+
+	raftHandler http.Handler
+	s           *etcdserver.EtcdServer
+	hss         []*httptest.Server
+}
+
+// newLocalListener opens a port localhost using any port
+func newLocalListener(t *testing.T) net.Listener {
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	return l
+}
+
+// configureTestCluster will set the params to start an etcd server
+func configureTestCluster(t *testing.T, name string) *EtcdTestServer {
+	var err error
+	m := &EtcdTestServer{}
+
+	pln := newLocalListener(t)
+	m.PeerListeners = []net.Listener{pln}
+	m.PeerURLs, err = types.NewURLs([]string{"http://" + pln.Addr().String()})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	cln := newLocalListener(t)
+	m.ClientListeners = []net.Listener{cln}
+	m.ClientURLs, err = types.NewURLs([]string{"http://" + cln.Addr().String()})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	m.Name = name
+	m.DataDir, err = ioutil.TempDir(os.TempDir(), "etcd")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	clusterStr := fmt.Sprintf("%s=http://%s", name, pln.Addr().String())
+	m.InitialPeerURLsMap, err = types.NewURLsMap(clusterStr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	m.Transport, err = transport.NewTimeoutTransport(transport.TLSInfo{}, time.Second, rafthttp.ConnReadTimeout, rafthttp.ConnWriteTimeout)
+	if err != nil {
+		t.Fatal(err)
+	}
+	m.NewCluster = true
+	m.ForceNewCluster = false
+	m.ElectionTicks = 10
+	m.TickMs = uint(10)
+
+	return m
+}
+
+// launch will attempt to start the etcd server
+func (m *EtcdTestServer) launch(t *testing.T) error {
+	var err error
+	if m.s, err = etcdserver.NewServer(&m.ServerConfig); err != nil {
+		return fmt.Errorf("failed to initialize the etcd server: %v", err)
+	}
+	m.s.SyncTicker = time.Tick(500 * time.Millisecond)
+	m.s.Start()
+	m.raftHandler = etcdhttp.NewPeerHandler(m.s.Cluster(), m.s.RaftHandler())
+	for _, ln := range m.PeerListeners {
+		hs := &httptest.Server{
+			Listener: ln,
+			Config:   &http.Server{Handler: m.raftHandler},
+		}
+		hs.Start()
+		m.hss = append(m.hss, hs)
+	}
+	for _, ln := range m.ClientListeners {
+		hs := &httptest.Server{
+			Listener: ln,
+			Config:   &http.Server{Handler: etcdhttp.NewClientHandler(m.s, m.ServerConfig.ReqTimeout())},
+		}
+		hs.Start()
+		m.hss = append(m.hss, hs)
+	}
+	return nil
+}
+
+// Terminate will shutdown the running etcd server
+func (m *EtcdTestServer) Terminate(t *testing.T) {
+	m.client.(*goetcd.Client).Close()
+	m.s.Stop()
+	for _, hs := range m.hss {
+		hs.CloseClientConnections()
+		hs.Close()
+	}
+	if err := os.RemoveAll(m.ServerConfig.DataDir); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// NewEtcdTestClientServer creates a new client and server for testing
+func NewEtcdTestClientServer(t *testing.T) *EtcdTestServer {
+	server := configureTestCluster(t, "foo")
+	err := server.launch(t)
+	if err != nil {
+		t.Fatal("Failed to start etcd server error=%v", err)
+		return nil
+	}
+	server.client = goetcd.NewClient(server.ClientURLs.StringSlice())
+	if server.client == nil {
+		t.Errorf("Failed to connect to local etcd server")
+		defer server.Terminate(t)
+		return nil
+	}
+	return server
+}

pkg/storage/etcd/etcd_util.go

@@ -23,10 +23,9 @@ import (
 	"net/http"
 	"os/exec"
 
-	"k8s.io/kubernetes/pkg/tools"
-
 	goetcd "github.com/coreos/go-etcd/etcd"
 	"github.com/golang/glog"
+	"k8s.io/kubernetes/pkg/tools"
 )
 
 // IsEtcdNotFound returns true if and only if err is an etcd not found error.