Use shared watch code in kubelet etcd config

pkg/kubelet/config/etcd.go

@@ -18,6 +18,7 @@ limitations under the License.
 package config
 
 import (
+	"errors"
 	"fmt"
 	"path"
 	"time"
@@ -25,96 +26,72 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	_ "github.com/GoogleCloudPlatform/kubernetes/pkg/api/v1beta1"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
-	"github.com/coreos/go-etcd/etcd"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
 	"github.com/golang/glog"
-	"gopkg.in/v1/yaml"
 )
 
 func EtcdKeyForHost(hostname string) string {
 	return path.Join("/", "registry", "hosts", hostname, "kubelet")
 }
 
-// TODO(lavalamp): Use a watcher interface instead of the etcd client directly
 type SourceEtcd struct {
 	key     string
-	client  tools.EtcdClient
+	helper  tools.EtcdHelper
 	updates chan<- interface{}
-
-	interval time.Duration
-	timeout  time.Duration
 }
 
-// NewSourceEtcd creates a config source that watches and pulls from a key in etcd.
+// NewSourceEtcd creates a config source that watches and pulls from a key in etcd
 func NewSourceEtcd(key string, client tools.EtcdClient, updates chan<- interface{}) *SourceEtcd {
-	config := &SourceEtcd{
+	helper := tools.EtcdHelper{
+		client,
+		runtime.Codec,
+		runtime.ResourceVersioner,
+	}
+	source := &SourceEtcd{
 		key:     key,
-		client:  client,
+		helper:  helper,
 		updates: updates,
-
-		timeout: 1 * time.Minute,
 	}
 	glog.Infof("Watching etcd for %s", key)
-	go util.Forever(config.run, time.Second)
+	go util.Forever(source.run, time.Second)
-	return config
+	return source
 }
 
-// run loops forever looking for changes to a key in etcd.
 func (s *SourceEtcd) run() {
-	index := uint64(0)
+	watching, err := s.helper.Watch(s.key, 0)
-	for {
-		nextIndex, err := s.fetchNextState(index)
 	if err != nil {
-			if !tools.IsEtcdNotFound(err) {
+		glog.Errorf("Failed to initialize etcd watch: %v", err)
-				glog.Errorf("Unable to extract from the response (%s): %%v", s.key, err)
-			}
 		return
 	}
-		index = nextIndex
+	for {
+		select {
+		case event, ok := <-watching.ResultChan():
+			if !ok {
+				return
 			}
-}
 
-// fetchNextState fetches the key (or waits for a change to a key) and then returns
+			pods, err := eventToPods(event)
-// the nextIndex to read.  It will watch no longer than s.waitDuration and then return.
-func (s *SourceEtcd) fetchNextState(fromIndex uint64) (nextIndex uint64, err error) {
-	var response *etcd.Response
-
-	if fromIndex == 0 {
-		response, err = s.client.Get(s.key, true, false)
-	} else {
-		response, err = s.client.Watch(s.key, fromIndex, false, nil, stopChannel(s.timeout))
-		if tools.IsEtcdWatchStoppedByUser(err) {
-			return fromIndex, nil
-		}
-	}
 			if err != nil {
-		return fromIndex, err
+				glog.Errorf("Failed to parse result from etcd watch: %v", err)
+				continue
 			}
 
-	pods, err := responseToPods(response)
+			glog.Infof("Received state from etcd watch: %+v", pods)
-	if err != nil {
-		glog.Infof("Response was in error: %#v", response)
-		return 0, fmt.Errorf("error parsing response: %#v", err)
-	}
-
-	glog.Infof("Got state from etcd: %+v", pods)
 			s.updates <- kubelet.PodUpdate{pods, kubelet.SET}
-
+		}
-	return response.Node.ModifiedIndex + 1, nil
+	}
 }
 
-// responseToPods takes an etcd Response object, and turns it into a structured list of containers.
+// eventToPods takes a watch.Event object, and turns it into a structured list of pods.
 // It returns a list of containers, or an error if one occurs.
-func responseToPods(response *etcd.Response) ([]kubelet.Pod, error) {
+func eventToPods(ev watch.Event) ([]kubelet.Pod, error) {
 	pods := []kubelet.Pod{}
-	if response.Node == nil || len(response.Node.Value) == 0 {
+	manifests, ok := ev.Object.(*api.ContainerManifestList)
-		return pods, fmt.Errorf("no nodes field: %v", response)
+	if !ok {
-	}
+		return pods, errors.New("unable to parse response as ContainerManifestList")
-
-	manifests := api.ContainerManifestList{}
-	if err := yaml.Unmarshal([]byte(response.Node.Value), &manifests); err != nil {
-		return pods, fmt.Errorf("could not unmarshal manifests: %v", err)
 	}
 
 	for i, manifest := range manifests.Items {
@@ -124,22 +101,10 @@ func responseToPods(response *etcd.Response) ([]kubelet.Pod, error) {
 		}
 		pods = append(pods, kubelet.Pod{Name: name, Manifest: manifest})
 	}
+
 	return pods, nil
 }
 
-// stopChannel creates a channel that is closed after a duration for use with etcd client API.
+func makeContainerKey(machine string) string {
-// If until is 0, the channel will never close.
+	return "/registry/hosts/" + machine + "/kubelet"
-func stopChannel(until time.Duration) chan bool {
-	stop := make(chan bool)
-	if until == 0 {
-		return stop
-	}
-	go func() {
-		select {
-		case <-time.After(until):
-		}
-		stop <- true
-		close(stop)
-	}()
-	return stop
 }

pkg/kubelet/config/etcd_test.go

@@ -19,149 +19,67 @@ package config
 import (
 	"reflect"
 	"testing"
-	"time"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
-	"github.com/coreos/go-etcd/etcd"
 )
 
-// TODO(lavalamp): Use the etcd watcher from the tools package, and make sure all test cases here are tested there.
+func TestEventToPods(t *testing.T) {
-
+	tests := []struct {
-func TestGetEtcdData(t *testing.T) {
+		input watch.Event
-	fakeClient := tools.NewFakeEtcdClient(t)
+		pods  []kubelet.Pod
-	ch := make(chan interface{})
+		fail  bool
-	fakeClient.Data["/registry/hosts/machine/kubelet"] = tools.EtcdResponseWithError{
+	}{
-		R: &etcd.Response{
+		{
-			Node: &etcd.Node{
+			input: watch.Event{Object: nil},
-				Value: runtime.EncodeOrDie(&api.ContainerManifestList{
+			pods:  []kubelet.Pod{},
-					Items: []api.ContainerManifest{{ID: "foo"}},
+			fail:  true,
-				}),
+		},
-				ModifiedIndex: 1,
+		{
+			input: watch.Event{Object: &api.ContainerManifestList{}},
+			pods:  []kubelet.Pod{},
+			fail:  false,
+		},
+		{
+			input: watch.Event{
+				Object: &api.ContainerManifestList{
+					Items: []api.ContainerManifest{
+						api.ContainerManifest{ID: "foo"},
+						api.ContainerManifest{ID: "bar"},
 					},
 				},
-		E: nil,
+			},
+			pods: []kubelet.Pod{
+				kubelet.Pod{Name: "foo", Manifest: api.ContainerManifest{ID: "foo"}},
+				kubelet.Pod{Name: "bar", Manifest: api.ContainerManifest{ID: "bar"}},
+			},
+			fail: false,
+		},
+		{
+			input: watch.Event{
+				Object: &api.ContainerManifestList{
+					Items: []api.ContainerManifest{
+						api.ContainerManifest{ID: ""},
+						api.ContainerManifest{ID: ""},
+					},
+				},
+			},
+			pods: []kubelet.Pod{
+				kubelet.Pod{Name: "1", Manifest: api.ContainerManifest{ID: ""}},
+				kubelet.Pod{Name: "2", Manifest: api.ContainerManifest{ID: ""}},
+			},
+			fail: false,
+		},
 	}
-	NewSourceEtcd("/registry/hosts/machine/kubelet", fakeClient, ch)
 
-	//TODO: update FakeEtcdClient.Watch to handle receiver=nil with a given index
+	for i, tt := range tests {
-	//returns an infinite stream of updates
+		pods, err := eventToPods(tt.input)
-	for i := 0; i < 2; i++ {
+		if !reflect.DeepEqual(tt.pods, pods) {
-		update := (<-ch).(kubelet.PodUpdate)
+			t.Errorf("case %d: expected output %#v, got %#v", i, tt.pods, pods)
-		expected := CreatePodUpdate(kubelet.SET, kubelet.Pod{Name: "foo", Manifest: api.ContainerManifest{ID: "foo"}})
+		}
-		if !reflect.DeepEqual(expected, update) {
+		if tt.fail != (err != nil) {
-			t.Errorf("Expected %#v, Got %#v", expected, update)
+			t.Errorf("case %d: got fail=%t but err=%v", i, tt.fail, err)
 		}
 	}
 }
-
-func TestGetEtcdNoData(t *testing.T) {
-	fakeClient := tools.NewFakeEtcdClient(t)
-	ch := make(chan interface{}, 1)
-	fakeClient.Data["/registry/hosts/machine/kubelet"] = tools.EtcdResponseWithError{
-		R: &etcd.Response{},
-		E: nil,
-	}
-	c := SourceEtcd{"/registry/hosts/machine/kubelet", fakeClient, ch, time.Millisecond, time.Minute}
-	_, err := c.fetchNextState(0)
-	if err == nil {
-		t.Errorf("Expected error")
-	}
-	expectEmptyChannel(t, ch)
-}
-
-func TestGetEtcd(t *testing.T) {
-	fakeClient := tools.NewFakeEtcdClient(t)
-	ch := make(chan interface{}, 1)
-	manifest := api.ContainerManifest{ID: "foo", Version: "v1beta1", Containers: []api.Container{{Name: "1", Image: "foo"}}}
-	fakeClient.Data["/registry/hosts/machine/kubelet"] = tools.EtcdResponseWithError{
-		R: &etcd.Response{
-			Node: &etcd.Node{
-				Value: runtime.EncodeOrDie(&api.ContainerManifestList{
-					Items: []api.ContainerManifest{manifest},
-				}),
-				ModifiedIndex: 1,
-			},
-		},
-		E: nil,
-	}
-	c := SourceEtcd{"/registry/hosts/machine/kubelet", fakeClient, ch, time.Millisecond, time.Minute}
-	lastIndex, err := c.fetchNextState(0)
-	if err != nil {
-		t.Errorf("Unexpected error: %v", err)
-	}
-	if lastIndex != 2 {
-		t.Errorf("Expected %#v, Got %#v", 2, lastIndex)
-	}
-	update := (<-ch).(kubelet.PodUpdate)
-	expected := CreatePodUpdate(kubelet.SET, kubelet.Pod{Name: "foo", Manifest: manifest})
-	if !reflect.DeepEqual(expected, update) {
-		t.Errorf("Expected %#v, Got %#v", expected, update)
-	}
-	for i := range update.Pods {
-		if errs := kubelet.ValidatePod(&update.Pods[i]); len(errs) != 0 {
-			t.Errorf("Expected no validation errors on %#v, Got %#v", update.Pods[i], errs)
-		}
-	}
-}
-
-func TestWatchEtcd(t *testing.T) {
-	fakeClient := tools.NewFakeEtcdClient(t)
-	ch := make(chan interface{}, 1)
-	fakeClient.Data["/registry/hosts/machine/kubelet"] = tools.EtcdResponseWithError{
-		R: &etcd.Response{
-			Node: &etcd.Node{
-				Value:         runtime.EncodeOrDie(&api.ContainerManifestList{}),
-				ModifiedIndex: 2,
-			},
-		},
-		E: nil,
-	}
-	c := SourceEtcd{"/registry/hosts/machine/kubelet", fakeClient, ch, time.Millisecond, time.Minute}
-	lastIndex, err := c.fetchNextState(1)
-	if err != nil {
-		t.Errorf("Unexpected error: %v", err)
-	}
-	if lastIndex != 3 {
-		t.Errorf("Expected %d, Got %d", 3, lastIndex)
-	}
-	update := (<-ch).(kubelet.PodUpdate)
-	expected := CreatePodUpdate(kubelet.SET)
-	if !reflect.DeepEqual(expected, update) {
-		t.Errorf("Expected %#v, Got %#v", expected, update)
-	}
-}
-
-func TestGetEtcdNotFound(t *testing.T) {
-	fakeClient := tools.NewFakeEtcdClient(t)
-	ch := make(chan interface{}, 1)
-	fakeClient.Data["/registry/hosts/machine/kubelet"] = tools.EtcdResponseWithError{
-		R: &etcd.Response{},
-		E: tools.EtcdErrorNotFound,
-	}
-	c := SourceEtcd{"/registry/hosts/machine/kubelet", fakeClient, ch, time.Millisecond, time.Minute}
-	_, err := c.fetchNextState(0)
-	if err == nil {
-		t.Errorf("Expected error")
-	}
-	expectEmptyChannel(t, ch)
-}
-
-func TestGetEtcdError(t *testing.T) {
-	fakeClient := tools.NewFakeEtcdClient(t)
-	ch := make(chan interface{}, 1)
-	fakeClient.Data["/registry/hosts/machine/kubelet"] = tools.EtcdResponseWithError{
-		R: &etcd.Response{},
-		E: &etcd.EtcdError{
-			ErrorCode: 200, // non not found error
-		},
-	}
-	c := SourceEtcd{"/registry/hosts/machine/kubelet", fakeClient, ch, time.Millisecond, time.Minute}
-	_, err := c.fetchNextState(0)
-	if err == nil {
-		t.Errorf("Expected error")
-	}
-	expectEmptyChannel(t, ch)
-}