Panic server on watch errors in test environment

This change makes it so that errors during watch decoding panic the
server if it is in a test environment.  This allows us to catch coder
errors related to storing incompatible types at the same location in
etcd.

Signed-off-by: Monis Khan <mkhan@redhat.com>

Kubernetes-commit: a13f026fd012859f04467e6007e2cafe4a788927

pkg/storage/etcd3/watcher.go

@@ -17,8 +17,11 @@ limitations under the License.
 package etcd3
 
 import (
+	"errors"
 	"fmt"
 	"net/http"
+	"os"
+	"strconv"
 	"strings"
 	"sync"
 
@@ -40,6 +43,24 @@ const (
 	outgoingBufSize = 100
 )
 
+// fatalOnDecodeError is used during testing to panic the server if watcher encounters a decoding error
+var fatalOnDecodeError = false
+
+// errTestingDecode is the only error that testingDeferOnDecodeError catches during a panic
+var errTestingDecode = errors.New("sentinel error only used during testing to indicate watch decoding error")
+
+// testingDeferOnDecodeError is used during testing to recover from a panic caused by errTestingDecode, all other values continue to panic
+func testingDeferOnDecodeError() {
+	if r := recover(); r != nil && r != errTestingDecode {
+		panic(r)
+	}
+}
+
+func init() {
+	// check to see if we are running in a test environment
+	fatalOnDecodeError, _ = strconv.ParseBool(os.Getenv("KUBE_PANIC_WATCH_DECODE_ERROR"))
+}
+
 type watcher struct {
 	client      *clientv3.Client
 	codec       runtime.Codec
@@ -373,9 +394,18 @@ func (wc *watchChan) prepareObjs(e *event) (curObj runtime.Object, oldObj runtim
 	return curObj, oldObj, nil
 }
 
-func decodeObj(codec runtime.Codec, versioner storage.Versioner, data []byte, rev int64) (runtime.Object, error) {
+func decodeObj(codec runtime.Codec, versioner storage.Versioner, data []byte, rev int64) (_ runtime.Object, err error) {
 	obj, err := runtime.Decode(codec, []byte(data))
 	if err != nil {
+		if fatalOnDecodeError {
+			// catch watch decode error iff we caused it on
+			// purpose during a unit test
+			defer testingDeferOnDecodeError()
+			// we are running in a test environment and thus an
+			// error here is due to a coder mistake if the defer
+			// does not catch it
+			panic(err)
+		}
 		return nil, err
 	}
 	// ensure resource version is set on the object we load from etcd

pkg/storage/etcd3/watcher_test.go

@@ -17,7 +17,6 @@ limitations under the License.
 package etcd3
 
 import (
-	"errors"
 	"fmt"
 	"reflect"
 	"strconv"
@@ -327,7 +326,7 @@ type testCodec struct {
 }
 
 func (c *testCodec) Decode(data []byte, defaults *schema.GroupVersionKind, into runtime.Object) (runtime.Object, *schema.GroupVersionKind, error) {
-	return nil, nil, errors.New("Expected decoding failure")
+	return nil, nil, errTestingDecode
 }
 
 func testCheckEventType(t *testing.T, expectEventType watch.EventType, w watch.Interface) {