Merge pull request #46045 from enj/enj/t/watch_decode_fatal

Automatic merge from submit-queue (batch tested with PRs 46124, 46434, 46089, 45589, 46045)

Panic server on watch errors during etcd test

**What this PR does / why we need it**:

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

For example, https://github.com/kubernetes/kubernetes/pull/39164#discussion_r116619172 would have failed like so (instead of silently recording an error to the master log and passing the test):

```
I0518 14:39:32.413038    4654 storage_rbac.go:257] created rolebinding.rbac.authorization.k8s.io/system:controller:bootstrap-signer in kube-public
panic: v1beta1.NetworkPolicy is not suitable for converting to {{"networking.k8s.io" "__internal"} [{"networking.k8s.io" ""} {"networking.k8s.io" ""}]}

goroutine 799 [running]:
k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/storage/etcd3.decodeObj(0x71146c0, 0xc420eb9d80, 0x7129340, 0x733cdf8, 0xc423708360, 0x118, 0x120, 0x86, 0x0, 0x0, ...)
	/home/mkhan/go/src/k8s.io/kubernetes/_output/local/go/src/k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/storage/etcd3/watcher.go:382 +0x115
k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/storage/etcd3.(*watchChan).prepareObjs(0xc420958f60, 0xc422ee9ef0, 0xc42003c600, 0x8a3b9b, 0x70f36c0, 0xc4209585a0, 0x4583f0, 0xc4215d0f58)
	/home/mkhan/go/src/k8s.io/kubernetes/_output/local/go/src/k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/storage/etcd3/watcher.go:353 +0x2cd
k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/storage/etcd3.(*watchChan).transform(0xc420958f60, 0xc422ee9ef0, 0x0)
	/home/mkhan/go/src/k8s.io/kubernetes/_output/local/go/src/k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/storage/etcd3/watcher.go:248 +0x4d
k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/storage/etcd3.(*watchChan).processEvent(0xc420958f60, 0xc420132010)
	/home/mkhan/go/src/k8s.io/kubernetes/_output/local/go/src/k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/storage/etcd3/watcher.go:213 +0x122
created by k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/storage/etcd3.(*watchChan).run
	/home/mkhan/go/src/k8s.io/kubernetes/_output/local/go/src/k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/storage/etcd3/watcher.go:116 +0xe9
exit status 2
FAIL	k8s.io/kubernetes/test/integration/etcd	13.065s
Makefile:164: recipe for target 'test' failed
```

@liggitt PTAL

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

**Release note**:

```
NONE
```

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'
@@ -658,11 +662,11 @@ function start_kubelet {
 	${KUBELET_FLAGS} >"${KUBELET_LOG}" 2>&1 &
       KUBELET_PID=$!
       # Quick check that kubelet is running.
-      if ps -p $KUBELET_PID > /dev/null ; then 
+      if ps -p $KUBELET_PID > /dev/null ; then
 	echo "kubelet ( $KUBELET_PID ) is running."
       else
 	cat ${KUBELET_LOG} ; exit 1
-      fi	
+      fi
     else
       # Docker won't run a container with a cidfile (container id file)
       # unless that file does not already exist; clean up an existing
@@ -733,7 +737,7 @@ function start_kubedns {
         echo "Creating kube-system namespace"
         sed -e "s/{{ pillar\['dns_domain'\] }}/${DNS_DOMAIN}/g" "${KUBE_ROOT}/cluster/addons/dns/kubedns-controller.yaml.in" >| kubedns-deployment.yaml
         sed -e "s/{{ pillar\['dns_server'\] }}/${DNS_SERVER_IP}/g" "${KUBE_ROOT}/cluster/addons/dns/kubedns-svc.yaml.in" >| kubedns-svc.yaml
-        
+
         # TODO update to dns role once we have one.
         ${KUBECTL} --kubeconfig="${CERT_DIR}/admin.kubeconfig" create clusterrolebinding system:kube-dns --clusterrole=cluster-admin --serviceaccount=kube-system:default
         # use kubectl to create kubedns deployment and service
@@ -748,7 +752,7 @@ function start_kubedns {
 
 function start_kubedashboard {
     if [[ "${ENABLE_CLUSTER_DASHBOARD}" = true ]]; then
-        echo "Creating kubernetes-dashboard"       
+        echo "Creating kubernetes-dashboard"
         # use kubectl to create the dashboard
         ${KUBECTL} --kubeconfig="${CERT_DIR}/admin.kubeconfig" create -f ${KUBE_ROOT}/cluster/addons/dashboard/dashboard-controller.yaml
         ${KUBECTL} --kubeconfig="${CERT_DIR}/admin.kubeconfig" create -f ${KUBE_ROOT}/cluster/addons/dashboard/dashboard-service.yaml

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) {