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>

hack/local-up-cluster.sh

@@ -86,6 +86,10 @@ DEFAULT_STORAGE_CLASS=${KUBE_DEFAULT_STORAGE_CLASS:-true}
 KUBE_CACHE_MUTATION_DETECTOR="${KUBE_CACHE_MUTATION_DETECTOR:-true}"
 export KUBE_CACHE_MUTATION_DETECTOR
 
+# panic the server on watch decode errors since they are considered coder mistakes
+KUBE_PANIC_WATCH_DECODE_ERROR="${KUBE_PANIC_WATCH_DECODE_ERROR:-true}"
+export KUBE_PANIC_WATCH_DECODE_ERROR
+
 ADMISSION_CONTROL_CONFIG_FILE=${ADMISSION_CONTROL_CONFIG_FILE:-""}
 
 # START_MODE can be 'all', 'kubeletonly', or 'nokubelet'

hack/make-rules/test.sh

@@ -27,6 +27,10 @@ kube::golang::setup_env
 KUBE_CACHE_MUTATION_DETECTOR="${KUBE_CACHE_MUTATION_DETECTOR:-true}"
 export KUBE_CACHE_MUTATION_DETECTOR
 
+# panic the server on watch decode errors since they are considered coder mistakes
+KUBE_PANIC_WATCH_DECODE_ERROR="${KUBE_PANIC_WATCH_DECODE_ERROR:-true}"
+export KUBE_PANIC_WATCH_DECODE_ERROR
+
 # Handle case where OS has sha#sum commands, instead of shasum.
 if which shasum >/dev/null 2>&1; then
   SHA1SUM="shasum -a1"

staging/src/k8s.io/apiserver/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

staging/src/k8s.io/apiserver/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) {