diff --git a/pkg/kubelet/config/etcd.go b/pkg/kubelet/config/etcd.go index 33d2184ff58..a48d8413c6b 100644 --- a/pkg/kubelet/config/etcd.go +++ b/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) - return config + go util.Forever(source.run, time.Second) + 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) + if err != nil { + glog.Errorf("Failed to initialize etcd watch: %v", err) + return + } for { - nextIndex, err := s.fetchNextState(index) - if err != nil { - if !tools.IsEtcdNotFound(err) { - glog.Errorf("Unable to extract from the response (%s): %%v", s.key, err) + select { + case event, ok := <-watching.ResultChan(): + if !ok { + return } - return + + pods, err := eventToPods(event) + if err != nil { + glog.Errorf("Failed to parse result from etcd watch: %v", err) + continue + } + + glog.Infof("Received state from etcd watch: %+v", pods) + s.updates <- kubelet.PodUpdate{pods, kubelet.SET} } - index = nextIndex } } -// fetchNextState fetches the key (or waits for a change to a key) and then returns -// 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 - } - - pods, err := responseToPods(response) - 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 { - return pods, fmt.Errorf("no nodes field: %v", response) - } - - manifests := api.ContainerManifestList{} - if err := yaml.Unmarshal([]byte(response.Node.Value), &manifests); err != nil { - return pods, fmt.Errorf("could not unmarshal manifests: %v", err) + manifests, ok := ev.Object.(*api.ContainerManifestList) + if !ok { + return pods, errors.New("unable to parse response as ContainerManifestList") } 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. -// If until is 0, the channel will never close. -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 +func makeContainerKey(machine string) string { + return "/registry/hosts/" + machine + "/kubelet" } diff --git a/pkg/kubelet/config/etcd_test.go b/pkg/kubelet/config/etcd_test.go index 629a626617e..ac1338f527f 100644 --- a/pkg/kubelet/config/etcd_test.go +++ b/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/tools" - "github.com/coreos/go-etcd/etcd" + "github.com/GoogleCloudPlatform/kubernetes/pkg/watch" ) -// TODO(lavalamp): Use the etcd watcher from the tools package, and make sure all test cases here are tested there. - -func TestGetEtcdData(t *testing.T) { - fakeClient := tools.NewFakeEtcdClient(t) - ch := make(chan interface{}) - fakeClient.Data["/registry/hosts/machine/kubelet"] = tools.EtcdResponseWithError{ - R: &etcd.Response{ - Node: &etcd.Node{ - Value: runtime.EncodeOrDie(&api.ContainerManifestList{ - Items: []api.ContainerManifest{{ID: "foo"}}, - }), - ModifiedIndex: 1, - }, +func TestEventToPods(t *testing.T) { + tests := []struct { + input watch.Event + pods []kubelet.Pod + fail bool + }{ + { + input: watch.Event{Object: nil}, + pods: []kubelet.Pod{}, + fail: true, + }, + { + 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"}, + }, + }, + }, + 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, }, - E: nil, } - NewSourceEtcd("/registry/hosts/machine/kubelet", fakeClient, ch) - //TODO: update FakeEtcdClient.Watch to handle receiver=nil with a given index - //returns an infinite stream of updates - for i := 0; i < 2; i++ { - update := (<-ch).(kubelet.PodUpdate) - expected := CreatePodUpdate(kubelet.SET, kubelet.Pod{Name: "foo", Manifest: api.ContainerManifest{ID: "foo"}}) - if !reflect.DeepEqual(expected, update) { - t.Errorf("Expected %#v, Got %#v", expected, update) + for i, tt := range tests { + pods, err := eventToPods(tt.input) + if !reflect.DeepEqual(tt.pods, pods) { + t.Errorf("case %d: expected output %#v, got %#v", i, tt.pods, pods) + } + if tt.fail != (err != nil) { + 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) -}