Return api/errors instead of raw etcd errors

pkg/api/errors/errors.go

@@ -372,6 +372,11 @@ func IsServerTimeout(err error) bool {
 	return reasonForError(err) == api.StatusReasonServerTimeout
 }
 
+// IsInternalServerError determines if err is an error which indicates that there was an internal server error.
+func IsInternalServerError(err error) bool {
+	return reasonForError(err) == api.StatusReasonInternalError
+}
+
 // IsUnexpectedServerError returns true if the server response was not in the expected API format,
 // and may be the result of another HTTP actor.
 func IsUnexpectedServerError(err error) bool {
@@ -409,6 +414,14 @@ func SuggestsClientDelay(err error) (int, bool) {
 	return 0, false
 }
 
+func IsAPIStatusError(err error) bool {
+	switch err.(type) {
+	case *StatusError:
+		return true
+	}
+	return false
+}
+
 func reasonForError(err error) api.StatusReason {
 	switch t := err.(type) {
 	case *StatusError:

pkg/api/errors/etcd/etcd.go

@@ -27,8 +27,10 @@ func InterpretGetError(err error, kind, name string) error {
 	switch {
 	case tools.IsEtcdNotFound(err):
 		return errors.NewNotFound(kind, name)
-	default:
+	case errors.IsAPIStatusError(err):
 		return err
+	default:
+		return errors.NewInternalError(err)
 	}
 }
 
@@ -38,8 +40,10 @@ func InterpretCreateError(err error, kind, name string) error {
 	switch {
 	case tools.IsEtcdNodeExist(err):
 		return errors.NewAlreadyExists(kind, name)
-	default:
+	case errors.IsAPIStatusError(err):
 		return err
+	default:
+		return errors.NewInternalError(err)
 	}
 }
 
@@ -49,8 +53,10 @@ func InterpretUpdateError(err error, kind, name string) error {
 	switch {
 	case tools.IsEtcdTestFailed(err), tools.IsEtcdNodeExist(err):
 		return errors.NewConflict(kind, name, err)
-	default:
+	case errors.IsAPIStatusError(err):
 		return err
+	default:
+		return errors.NewInternalError(err)
 	}
 }
 
@@ -60,7 +66,9 @@ func InterpretDeleteError(err error, kind, name string) error {
 	switch {
 	case tools.IsEtcdNotFound(err):
 		return errors.NewNotFound(kind, name)
-	default:
+	case errors.IsAPIStatusError(err):
 		return err
+	default:
+		return errors.NewInternalError(err)
 	}
 }

pkg/registry/etcd/etcd.go

@@ -108,7 +108,10 @@ func (r *Registry) CreateService(ctx api.Context, svc *api.Service) (*api.Servic
 	}
 	out := &api.Service{}
 	err = r.CreateObj(key, svc, out, 0)
-	return out, etcderr.InterpretCreateError(err, "service", svc.Name)
+	if err != nil {
+		err = etcderr.InterpretCreateError(err, "Service", svc.Name)
+	}
+	return out, err
 }
 
 // GetService obtains a Service specified by its name.
@@ -120,7 +123,7 @@ func (r *Registry) GetService(ctx api.Context, name string) (*api.Service, error
 	var svc api.Service
 	err = r.ExtractObj(key, &svc, false)
 	if err != nil {
-		return nil, etcderr.InterpretGetError(err, "service", name)
+		return nil, etcderr.InterpretGetError(err, "Service", name)
 	}
 	return &svc, nil
 }
@@ -133,7 +136,7 @@ func (r *Registry) DeleteService(ctx api.Context, name string) error {
 	}
 	err = r.Delete(key, true)
 	if err != nil {
-		return etcderr.InterpretDeleteError(err, "service", name)
+		return etcderr.InterpretDeleteError(err, "Service", name)
 	}
 
 	// TODO: can leave dangling endpoints, and potentially return incorrect
@@ -153,12 +156,15 @@ func (r *Registry) UpdateService(ctx api.Context, svc *api.Service) (*api.Servic
 	}
 	out := &api.Service{}
 	err = r.SetObj(key, svc, out, 0)
-	return out, etcderr.InterpretUpdateError(err, "service", svc.Name)
+	if err != nil {
+		err = etcderr.InterpretUpdateError(err, "Service", svc.Name)
+	}
+	return out, err
 }
 
 // WatchServices begins watching for new, changed, or deleted service configurations.
 func (r *Registry) WatchServices(ctx api.Context, label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
-	version, err := tools.ParseWatchResourceVersion(resourceVersion, "service")
+	version, err := tools.ParseWatchResourceVersion(resourceVersion, "Service")
 	if err != nil {
 		return nil, err
 	}

pkg/registry/generic/etcd/etcd.go

@@ -191,7 +191,9 @@ func (e *Etcd) CreateWithName(ctx api.Context, name string, obj runtime.Object)
 		return err
 	}
 	err = e.Helper.CreateObj(key, obj, nil, ttl)
-	err = etcderr.InterpretCreateError(err, e.EndpointName, name)
+	if err != nil {
+		err = etcderr.InterpretCreateError(err, e.EndpointName, name)
+	}
 	if err == nil && e.Decorator != nil {
 		err = e.Decorator(obj)
 	}
@@ -250,7 +252,9 @@ func (e *Etcd) UpdateWithName(ctx api.Context, name string, obj runtime.Object)
 		return err
 	}
 	err = e.Helper.SetObj(key, obj, nil, ttl)
-	err = etcderr.InterpretUpdateError(err, e.EndpointName, name)
+	if err != nil {
+		err = etcderr.InterpretUpdateError(err, e.EndpointName, name)
+	}
 	if err == nil && e.Decorator != nil {
 		err = e.Decorator(obj)
 	}

pkg/registry/pod/etcd/etcd.go

@@ -145,13 +145,13 @@ func (r *BindingREST) setPodHostAndAnnotations(ctx api.Context, podID, oldMachin
 	err = r.store.Helper.GuaranteedUpdate(podKey, &api.Pod{}, false, tools.SimpleUpdate(func(obj runtime.Object) (runtime.Object, error) {
 		pod, ok := obj.(*api.Pod)
 		if !ok {
-			return nil, fmt.Errorf("unexpected object: %#v", obj)
+			return nil, errors.NewInternalError(fmt.Errorf("received object is not of type pod: %#v", obj))
 		}
 		if pod.DeletionTimestamp != nil {
-			return nil, fmt.Errorf("pod %s is being deleted, cannot be assigned to a host", pod.Name)
+			return nil, errors.NewConflict("pod", podID, fmt.Errorf("pod %s is being deleted, cannot be assigned to a host", pod.Name))
 		}
 		if pod.Spec.NodeName != oldMachine {
-			return nil, fmt.Errorf("pod %v is already assigned to node %q", pod.Name, pod.Spec.NodeName)
+			return nil, errors.NewConflict("pod", podID, fmt.Errorf("pod %v is already assigned to node %q", pod.Name, pod.Spec.NodeName))
 		}
 		pod.Spec.NodeName = machine
 		if pod.Annotations == nil {
@@ -222,7 +222,7 @@ func (r *LogREST) New() runtime.Object {
 func (r *LogREST) Get(ctx api.Context, name string, opts runtime.Object) (runtime.Object, error) {
 	logOpts, ok := opts.(*api.PodLogOptions)
 	if !ok {
-		return nil, fmt.Errorf("Invalid options object: %#v", opts)
+		return nil, errors.NewInternalError(fmt.Errorf("received object is not of type PodLogOptions: %#v", opts))
 	}
 	location, transport, err := pod.LogLocation(r.store, r.kubeletConn, ctx, name, logOpts)
 	if err != nil {
@@ -270,7 +270,7 @@ func (r *ProxyREST) NewConnectOptions() (runtime.Object, bool, string) {
 func (r *ProxyREST) Connect(ctx api.Context, id string, opts runtime.Object) (rest.ConnectHandler, error) {
 	proxyOpts, ok := opts.(*api.PodProxyOptions)
 	if !ok {
-		return nil, fmt.Errorf("Invalid options object: %#v", opts)
+		return nil, errors.NewInternalError(fmt.Errorf("received object is not of type PodProxyOptions: %#v", opts))
 	}
 	location, _, err := pod.ResourceLocation(r.store, ctx, id)
 	if err != nil {
@@ -300,7 +300,7 @@ func (r *ExecREST) New() runtime.Object {
 func (r *ExecREST) Connect(ctx api.Context, name string, opts runtime.Object) (rest.ConnectHandler, error) {
 	execOpts, ok := opts.(*api.PodExecOptions)
 	if !ok {
-		return nil, fmt.Errorf("Invalid options object: %#v", opts)
+		return nil, errors.NewInternalError(fmt.Errorf("received object is not of type PodExecOptions: %#v", opts))
 	}
 	location, transport, err := pod.ExecLocation(r.store, r.kubeletConn, ctx, name, execOpts)
 	if err != nil {

pkg/registry/pod/etcd/etcd_test.go

@@ -153,7 +153,7 @@ func TestCreateRegistryError(t *testing.T) {
 
 	pod := validNewPod()
 	_, err := storage.Create(api.NewDefaultContext(), pod)
-	if err != fakeEtcdClient.Err {
+	if !errors.IsInternalServerError(err) {
 		t.Fatalf("unexpected error: %v", err)
 	}
 }

pkg/registry/resourcequota/etcd/etcd_test.go

@@ -116,7 +116,7 @@ func TestCreateRegistryError(t *testing.T) {
 
 	resourcequota := validNewResourceQuota()
 	_, err := storage.Create(api.NewDefaultContext(), resourcequota)
-	if err != fakeEtcdClient.Err {
+	if !errors.IsInternalServerError(err) {
 		t.Fatalf("unexpected error: %v", err)
 	}
 }

pkg/registry/service/allocator/etcd/etcd.go

@@ -17,7 +17,6 @@ limitations under the License.
 package etcd
 
 import (
-	"errors"
 	"fmt"
 	"sync"
 
@@ -31,7 +30,8 @@ import (
 )
 
 var (
-	errorUnableToAllocate = errors.New("unable to allocate")
+	// Placeholder error that should not be surfaced to the user.
+	errorUnableToAllocate = k8serr.NewInternalError(fmt.Errorf("unable to allocate"))
 )
 
 // Etcd exposes a service.Allocator that is backed by etcd.
@@ -121,6 +121,9 @@ func (e *Etcd) AllocateNext() (int, bool, error) {
 		}
 		return nil
 	})
+	if err != nil && err.Error() == errorUnableToAllocate.Error() {
+		err = nil
+	}
 	return offset, ok, err
 }
 
@@ -161,7 +164,10 @@ func (e *Etcd) tryUpdate(fn func() error) error {
 			return existing, nil
 		}),
 	)
-	return etcderr.InterpretUpdateError(err, e.kind, "")
+	if err != nil {
+		err = etcderr.InterpretUpdateError(err, e.kind, "")
+	}
+	return err
 }
 
 // Refresh reloads the RangeAllocation from etcd.