Add error type to watch.

pkg/api/errors/errors.go

@@ -21,6 +21,7 @@ import (
 	"net/http"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 )
 
 // statusError is an error intended for consumption by a REST API server.
@@ -38,6 +39,16 @@ func (e *statusError) Status() api.Status {
 	return e.status
 }
 
+// FromObject generates an statusError from an api.Status, if that is the type of obj; otherwise,
+// returns an error created by fmt.Errorf.
+func FromObject(obj runtime.Object) error {
+	switch t := obj.(type) {
+	case *api.Status:
+		return &statusError{*t}
+	}
+	return fmt.Errorf("unexpected object: %v", obj)
+}
+
 // NewNotFound returns a new error which indicates that the resource of the kind and the name was not found.
 func NewNotFound(kind, name string) error {
 	return &statusError{api.Status{

pkg/api/errors/errors_test.go

@@ -23,6 +23,7 @@ import (
 	"testing"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 )
 
 func TestErrorNew(t *testing.T) {
@@ -131,3 +132,23 @@ func Test_reasonForError(t *testing.T) {
 		t.Errorf("unexpected reason type: %#v", a)
 	}
 }
+
+type TestType struct{}
+
+func (*TestType) IsAnAPIObject() {}
+
+func TestFromObject(t *testing.T) {
+	table := []struct {
+		obj     runtime.Object
+		message string
+	}{
+		{&api.Status{Message: "foobar"}, "foobar"},
+		{&TestType{}, "unexpected object: &{}"},
+	}
+
+	for _, item := range table {
+		if e, a := item.message, FromObject(item.obj).Error(); e != a {
+			t.Errorf("Expected %v, got %v", e, a)
+		}
+	}
+}

pkg/client/cache/reflector.go

@@ -22,6 +22,7 @@ import (
 	"reflect"
 	"time"
 
+	apierrs "github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
@@ -101,7 +102,7 @@ func (r *Reflector) listAndWatch() {
 			return
 		}
 		if err := r.watchHandler(w, &resourceVersion); err != nil {
-			glog.Errorf("failed to watch %v: %v", r.expectedType, err)
+			glog.Errorf("watch of %v ended with error: %v", r.expectedType, err)
 			return
 		}
 	}
@@ -131,6 +132,9 @@ func (r *Reflector) watchHandler(w watch.Interface, resourceVersion *uint64) err
 		if !ok {
 			break
 		}
+		if event.Type == watch.Error {
+			return apierrs.FromObject(event.Object)
+		}
 		if e, a := r.expectedType, reflect.TypeOf(event.Object); e != a {
 			glog.Errorf("expected type %v, but watch event object had type %v", e, a)
 			continue
@@ -162,6 +166,6 @@ func (r *Reflector) watchHandler(w watch.Interface, resourceVersion *uint64) err
 		glog.Errorf("unexpected watch close - watch lasted less than a second and no items received")
 		return errors.New("very short watch")
 	}
-	glog.Infof("unexpected watch close - %v total items received", eventCount)
+	glog.Infof("watch close - %v total items received", eventCount)
 	return nil
 }

pkg/kubelet/config/etcd.go

@@ -60,18 +60,18 @@ func NewSourceEtcd(key string, client tools.EtcdClient, updates chan<- interface
 }
 
 func (s *SourceEtcd) run() {
-	watching, err := s.helper.Watch(s.key, 0)
-	if err != nil {
-		glog.Errorf("Failed to initialize etcd watch: %v", err)
-		return
-	}
+	watching := s.helper.Watch(s.key, 0)
 	for {
 		select {
 		case event, ok := <-watching.ResultChan():
 			if !ok {
 				return
 			}
-
+			if event.Type == watch.Error {
+				glog.Errorf("Watch error: %v", event.Object)
+				watching.Stop()
+				return
+			}
 			pods, err := eventToPods(event)
 			if err != nil {
 				glog.Errorf("Failed to parse result from etcd watch: %v", err)

pkg/registry/etcd/etcd.go

@@ -343,7 +343,7 @@ func (r *Registry) WatchServices(label, field labels.Selector, resourceVersion u
 		return nil, fmt.Errorf("label selectors are not supported on services")
 	}
 	if value, found := field.RequiresExactMatch("ID"); found {
-		return r.Watch(makeServiceKey(value), resourceVersion)
+		return r.Watch(makeServiceKey(value), resourceVersion), nil
 	}
 	if field.Empty() {
 		return r.WatchList("/registry/services/specs", resourceVersion, tools.Everything)
@@ -375,7 +375,7 @@ func (r *Registry) WatchEndpoints(label, field labels.Selector, resourceVersion
 		return nil, fmt.Errorf("label selectors are not supported on endpoints")
 	}
 	if value, found := field.RequiresExactMatch("ID"); found {
-		return r.Watch(makeServiceEndpointsKey(value), resourceVersion)
+		return r.Watch(makeServiceEndpointsKey(value), resourceVersion), nil
 	}
 	if field.Empty() {
 		return r.WatchList("/registry/services/endpoints", resourceVersion, tools.Everything)

pkg/tools/etcd_tools_watch.go

@@ -20,6 +20,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
@@ -48,7 +49,8 @@ func (h *EtcdHelper) WatchList(key string, resourceVersion uint64, filter Filter
 
 // Watch begins watching the specified key. Events are decoded into
 // API objects and sent down the returned watch.Interface.
-func (h *EtcdHelper) Watch(key string, resourceVersion uint64) (watch.Interface, error) {
+// Errors will be sent down the channel.
+func (h *EtcdHelper) Watch(key string, resourceVersion uint64) watch.Interface {
 	return h.WatchAndTransform(key, resourceVersion, nil)
 }
 
@@ -67,10 +69,11 @@ func (h *EtcdHelper) Watch(key string, resourceVersion uint64) (watch.Interface,
 //     return value, nil
 //   })
 //
-func (h *EtcdHelper) WatchAndTransform(key string, resourceVersion uint64, transform TransformFunc) (watch.Interface, error) {
+// Errors will be sent down the channel.
+func (h *EtcdHelper) WatchAndTransform(key string, resourceVersion uint64, transform TransformFunc) watch.Interface {
 	w := newEtcdWatcher(false, Everything, h.Codec, h.ResourceVersioner, transform)
 	go w.etcdWatch(h.Client, key, resourceVersion)
-	return w, <-w.immediateError
+	return w
 }
 
 // TransformFunc attempts to convert an object to another object for use with a watcher.
@@ -86,14 +89,10 @@ type etcdWatcher struct {
 	filter FilterFunc
 
 	etcdIncoming  chan *etcd.Response
+	etcdError     chan error
 	etcdStop      chan bool
 	etcdCallEnded chan struct{}
 
-	// etcdWatch will send an error down this channel if the Watch fails.
-	// Otherwise, a nil will be sent down this channel watchWaitDuration
-	// after the watch starts.
-	immediateError chan error
-
 	outgoing chan watch.Event
 	userStop chan struct{}
 	stopped  bool
@@ -110,64 +109,53 @@ const watchWaitDuration = 100 * time.Millisecond
 // and a versioner, the versioner must be able to handle the objects that transform creates.
 func newEtcdWatcher(list bool, filter FilterFunc, encoding runtime.Codec, versioner runtime.ResourceVersioner, transform TransformFunc) *etcdWatcher {
 	w := &etcdWatcher{
-		encoding:       encoding,
-		versioner:      versioner,
-		transform:      transform,
-		list:           list,
-		filter:         filter,
-		etcdIncoming:   make(chan *etcd.Response),
-		etcdStop:       make(chan bool),
-		etcdCallEnded:  make(chan struct{}),
-		immediateError: make(chan error),
-		outgoing:       make(chan watch.Event),
-		userStop:       make(chan struct{}),
+		encoding:     encoding,
+		versioner:    versioner,
+		transform:    transform,
+		list:         list,
+		filter:       filter,
+		etcdIncoming: make(chan *etcd.Response),
+		etcdError:    make(chan error, 1),
+		etcdStop:     make(chan bool),
+		outgoing:     make(chan watch.Event),
+		userStop:     make(chan struct{}),
 	}
 	w.emit = func(e watch.Event) { w.outgoing <- e }
 	go w.translate()
 	return w
 }
 
 // etcdWatch calls etcd's Watch function, and handles any errors. Meant to be called
-// as a goroutine. Will either send an error over w.immediateError if Watch fails, or in 100ms will
+// as a goroutine.
 func (w *etcdWatcher) etcdWatch(client EtcdGetSet, key string, resourceVersion uint64) {
 	defer util.HandleCrash()
-	defer close(w.etcdCallEnded)
-	go func() {
-		// This is racy; assume that Watch will fail within 100ms if it is going to fail.
-		// It's still more useful than blocking until the first result shows up.
-		// Trying to detect the 401: watch window expired error.
-		<-time.After(watchWaitDuration)
-		w.immediateError <- nil
-	}()
+	defer close(w.etcdError)
 	if resourceVersion == 0 {
-		latest, ok := etcdGetInitialWatchState(client, key, w.list, w.etcdIncoming)
-		if !ok {
+		latest, err := etcdGetInitialWatchState(client, key, w.list, w.etcdIncoming)
+		if err != nil {
+			w.etcdError <- err
 			return
 		}
 		resourceVersion = latest + 1
 	}
 	_, err := client.Watch(key, resourceVersion, w.list, w.etcdIncoming, w.etcdStop)
-	if err != etcd.ErrWatchStoppedByUser {
-		glog.Errorf("etcd.Watch stopped unexpectedly: %v (%#v)", err, key)
-		w.immediateError <- err
+	if err != nil && err != etcd.ErrWatchStoppedByUser {
+		w.etcdError <- err
 	}
 }
 
 // etcdGetInitialWatchState turns an etcd Get request into a watch equivalent
-func etcdGetInitialWatchState(client EtcdGetSet, key string, recursive bool, incoming chan<- *etcd.Response) (resourceVersion uint64, success bool) {
-	success = true
-
+func etcdGetInitialWatchState(client EtcdGetSet, key string, recursive bool, incoming chan<- *etcd.Response) (resourceVersion uint64, err error) {
 	resp, err := client.Get(key, false, recursive)
 	if err != nil {
 		if !IsEtcdNotFound(err) {
 			glog.Errorf("watch was unable to retrieve the current index for the provided key: %v (%#v)", err, key)
-			success = false
-			return
+			return resourceVersion, err
 		}
 		if index, ok := etcdErrorIndex(err); ok {
 			resourceVersion = index
 		}
-		return
+		return resourceVersion, nil
 	}
 	resourceVersion = resp.EtcdIndex
 	convertRecursiveResponse(resp.Node, resp, incoming)
@@ -189,24 +177,34 @@ func convertRecursiveResponse(node *etcd.Node, response *etcd.Response, incoming
 	incoming <- &copied
 }
 
-// translate pulls stuff from etcd, convert, and push out the outgoing channel. Meant to be
+// translate pulls stuff from etcd, converts, and pushes out the outgoing channel. Meant to be
 // called as a goroutine.
 func (w *etcdWatcher) translate() {
 	defer close(w.outgoing)
 	defer util.HandleCrash()
 
 	for {
 		select {
-		case <-w.etcdCallEnded:
+		case err := <-w.etcdError:
+			if err != nil {
+				w.emit(watch.Event{
+					watch.Error,
+					&api.Status{
+						Status:  api.StatusFailure,
+						Message: err.Error(),
+					},
+				})
+			}
 			return
 		case <-w.userStop:
 			w.etcdStop <- true
 			return
 		case res, ok := <-w.etcdIncoming:
-			if !ok {
-				return
+			if ok {
+				w.sendResult(res)
 			}
-			w.sendResult(res)
+			// If !ok, don't return here-- must wait for etcdError channel
+			// to give an error or be closed.
 		}
 	}
 }