mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-07-30 15:05:27 +00:00
Merge pull request #1138 from bcwaldon/common-watch-code
Use shared watch code in kubelet etcd config
This commit is contained in:
Daniel Smith
commit
779de7361a
@ -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"
|
||||
}
|
||||
|
||||
@ -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)
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user