Merge pull request #57504 from yue9944882/fix-fake-client-dummy-watch

Automatic merge from submit-queue. If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

feat(fakeclient): push event on watched channel on add/update/delete

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

This PR enables watch function for kubernetes [fakeclient](https://github.com/kubernetes/kubernetes/blob/1bcf0b0a227d57057bde1f6fd50f26224483b324/staging/src/k8s.io/client-go/kubernetes/fake/clientset_generated.go#L88).

This fake client add watchReactorFunction by wrapping [watch.NewFake](https://github.com/kubernetes/kubernetes/blob/1bcf0b0a227d57057bde1f6fd50f26224483b324/staging/src/k8s.io/client-go/kubernetes/fake/clientset_generated.go#L98) which is a `chan Event` but actually nothing pushes objects into this channel. So all watch function called by fake client will never return or never receive any object.

This PR intercepts ReactionFunc of `Create / Update / DeleteActionImpl` and will push the requested object to channel.

Which issue(s) this PR fixes (optional, in fixes #<issue number>(, fixes #<issue_number>, ...) format, will close the issue(s) when PR gets merged):

Fixes #54075

**Special notes for your reviewer**:

**Release note**:

```dev-release-note
enable watch function for fake client
```

Kubernetes-commit: 268555a30a0f028762854f5b0d3ebb587e2ee4ee

vendor/k8s.io/apimachinery/pkg/api/errors/errors.go

@@ -352,6 +352,14 @@ func NewGenericServerResponse(code int, verb string, qualifiedResource schema.Gr
 		reason = metav1.StatusReasonForbidden
 		// the server message has details about who is trying to perform what action.  Keep its message.
 		message = serverMessage
+	case http.StatusNotAcceptable:
+		reason = metav1.StatusReasonNotAcceptable
+		// the server message has details about what types are acceptable
+		message = serverMessage
+	case http.StatusUnsupportedMediaType:
+		reason = metav1.StatusReasonUnsupportedMediaType
+		// the server message has details about what types are acceptable
+		message = serverMessage
 	case http.StatusMethodNotAllowed:
 		reason = metav1.StatusReasonMethodNotAllowed
 		message = "the server does not allow this method on the requested resource"
@@ -434,6 +442,16 @@ func IsResourceExpired(err error) bool {
 	return ReasonForError(err) == metav1.StatusReasonExpired
 }
 
+// IsNotAcceptable determines if err is an error which indicates that the request failed due to an invalid Accept header
+func IsNotAcceptable(err error) bool {
+	return ReasonForError(err) == metav1.StatusReasonNotAcceptable
+}
+
+// IsUnsupportedMediaType determines if err is an error which indicates that the request failed due to an invalid Content-Type header
+func IsUnsupportedMediaType(err error) bool {
+	return ReasonForError(err) == metav1.StatusReasonUnsupportedMediaType
+}
+
 // IsMethodNotSupported determines if the err is an error which indicates the provided action could not
 // be performed because it is not supported by the server.
 func IsMethodNotSupported(err error) bool {

vendor/k8s.io/apimachinery/pkg/apis/meta/v1/types.go

@@ -651,6 +651,18 @@ const (
 	// can only be created. API calls that return MethodNotAllowed can never succeed.
 	StatusReasonMethodNotAllowed StatusReason = "MethodNotAllowed"
 
+	// StatusReasonNotAcceptable means that the accept types indicated by the client were not acceptable
+	// to the server - for instance, attempting to receive protobuf for a resource that supports only json and yaml.
+	// API calls that return NotAcceptable can never succeed.
+	// Status code 406
+	StatusReasonNotAcceptable StatusReason = "NotAcceptable"
+
+	// StatusReasonUnsupportedMediaType means that the content type sent by the client is not acceptable
+	// to the server - for instance, attempting to send protobuf for a resource that supports only json and yaml.
+	// API calls that return UnsupportedMediaType can never succeed.
+	// Status code 415
+	StatusReasonUnsupportedMediaType StatusReason = "UnsupportedMediaType"
+
 	// StatusReasonInternalError indicates that an internal error occurred, it is unexpected
 	// and the outcome of the call is unknown.
 	// Details (optional):

vendor/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured/helpers.go

@@ -43,14 +43,15 @@ func NestedFieldCopy(obj map[string]interface{}, fields ...string) (interface{},
 
 func nestedFieldNoCopy(obj map[string]interface{}, fields ...string) (interface{}, bool, error) {
 	var val interface{} = obj
-	for _, field := range fields {
+
+	for i, field := range fields {
 		if m, ok := val.(map[string]interface{}); ok {
 			val, ok = m[field]
 			if !ok {
 				return nil, false, nil
 			}
 		} else {
-			return nil, false, fmt.Errorf("%v is of the type %T, expected map[string]interface{}", val, val)
+			return nil, false, fmt.Errorf("%v accessor error: %v is of the type %T, expected map[string]interface{}", jsonPath(fields[:i+1]), val, val)
 		}
 	}
 	return val, true, nil
@@ -65,7 +66,7 @@ func NestedString(obj map[string]interface{}, fields ...string) (string, bool, e
 	}
 	s, ok := val.(string)
 	if !ok {
-		return "", false, fmt.Errorf("%v is of the type %T, expected string", val, val)
+		return "", false, fmt.Errorf("%v accessor error: %v is of the type %T, expected string", jsonPath(fields), val, val)
 	}
 	return s, true, nil
 }
@@ -79,7 +80,7 @@ func NestedBool(obj map[string]interface{}, fields ...string) (bool, bool, error
 	}
 	b, ok := val.(bool)
 	if !ok {
-		return false, false, fmt.Errorf("%v is of the type %T, expected bool", val, val)
+		return false, false, fmt.Errorf("%v accessor error: %v is of the type %T, expected bool", jsonPath(fields), val, val)
 	}
 	return b, true, nil
 }
@@ -93,7 +94,7 @@ func NestedFloat64(obj map[string]interface{}, fields ...string) (float64, bool,
 	}
 	f, ok := val.(float64)
 	if !ok {
-		return 0, false, fmt.Errorf("%v is of the type %T, expected float64", val, val)
+		return 0, false, fmt.Errorf("%v accessor error: %v is of the type %T, expected float64", jsonPath(fields), val, val)
 	}
 	return f, true, nil
 }
@@ -107,7 +108,7 @@ func NestedInt64(obj map[string]interface{}, fields ...string) (int64, bool, err
 	}
 	i, ok := val.(int64)
 	if !ok {
-		return 0, false, fmt.Errorf("%v is of the type %T, expected int64", val, val)
+		return 0, false, fmt.Errorf("%v accessor error: %v is of the type %T, expected int64", jsonPath(fields), val, val)
 	}
 	return i, true, nil
 }
@@ -121,14 +122,14 @@ func NestedStringSlice(obj map[string]interface{}, fields ...string) ([]string,
 	}
 	m, ok := val.([]interface{})
 	if !ok {
-		return nil, false, fmt.Errorf("%v is of the type %T, expected []interface{}", val, val)
+		return nil, false, fmt.Errorf("%v accessor error: %v is of the type %T, expected []interface{}", jsonPath(fields), val, val)
 	}
 	strSlice := make([]string, 0, len(m))
 	for _, v := range m {
 		if str, ok := v.(string); ok {
 			strSlice = append(strSlice, str)
 		} else {
-			return nil, false, fmt.Errorf("contains non-string key in the slice: %v is of the type %T, expected string", v, v)
+			return nil, false, fmt.Errorf("%v accessor error: contains non-string key in the slice: %v is of the type %T, expected string", jsonPath(fields), v, v)
 		}
 	}
 	return strSlice, true, nil
@@ -143,7 +144,7 @@ func NestedSlice(obj map[string]interface{}, fields ...string) ([]interface{}, b
 	}
 	_, ok := val.([]interface{})
 	if !ok {
-		return nil, false, fmt.Errorf("%v is of the type %T, expected []interface{}", val, val)
+		return nil, false, fmt.Errorf("%v accessor error: %v is of the type %T, expected []interface{}", jsonPath(fields), val, val)
 	}
 	return runtime.DeepCopyJSONValue(val).([]interface{}), true, nil
 }
@@ -160,7 +161,7 @@ func NestedStringMap(obj map[string]interface{}, fields ...string) (map[string]s
 		if str, ok := v.(string); ok {
 			strMap[k] = str
 		} else {
-			return nil, false, fmt.Errorf("contains non-string key in the map: %v is of the type %T, expected string", v, v)
+			return nil, false, fmt.Errorf("%v accessor error: contains non-string key in the map: %v is of the type %T, expected string", jsonPath(fields), v, v)
 		}
 	}
 	return strMap, true, nil
@@ -185,25 +186,26 @@ func nestedMapNoCopy(obj map[string]interface{}, fields ...string) (map[string]i
 	}
 	m, ok := val.(map[string]interface{})
 	if !ok {
-		return nil, false, fmt.Errorf("%v is of the type %T, expected map[string]interface{}", val, val)
+		return nil, false, fmt.Errorf("%v accessor error: %v is of the type %T, expected map[string]interface{}", jsonPath(fields), val, val)
 	}
 	return m, true, nil
 }
 
 // SetNestedField sets the value of a nested field to a deep copy of the value provided.
-// Returns false if value cannot be set because one of the nesting levels is not a map[string]interface{}.
-func SetNestedField(obj map[string]interface{}, value interface{}, fields ...string) bool {
+// Returns an error if value cannot be set because one of the nesting levels is not a map[string]interface{}.
+func SetNestedField(obj map[string]interface{}, value interface{}, fields ...string) error {
 	return setNestedFieldNoCopy(obj, runtime.DeepCopyJSONValue(value), fields...)
 }
 
-func setNestedFieldNoCopy(obj map[string]interface{}, value interface{}, fields ...string) bool {
+func setNestedFieldNoCopy(obj map[string]interface{}, value interface{}, fields ...string) error {
 	m := obj
-	for _, field := range fields[:len(fields)-1] {
+
+	for i, field := range fields[:len(fields)-1] {
 		if val, ok := m[field]; ok {
 			if valMap, ok := val.(map[string]interface{}); ok {
 				m = valMap
 			} else {
-				return false
+				return fmt.Errorf("value cannot be set because %v is not a map[string]interface{}", jsonPath(fields[:i+1]))
 			}
 		} else {
 			newVal := make(map[string]interface{})
@@ -212,12 +214,12 @@ func setNestedFieldNoCopy(obj map[string]interface{}, value interface{}, fields
 		}
 	}
 	m[fields[len(fields)-1]] = value
-	return true
+	return nil
 }
 
 // SetNestedStringSlice sets the string slice value of a nested field.
-// Returns false if value cannot be set because one of the nesting levels is not a map[string]interface{}.
-func SetNestedStringSlice(obj map[string]interface{}, value []string, fields ...string) bool {
+// Returns an error if value cannot be set because one of the nesting levels is not a map[string]interface{}.
+func SetNestedStringSlice(obj map[string]interface{}, value []string, fields ...string) error {
 	m := make([]interface{}, 0, len(value)) // convert []string into []interface{}
 	for _, v := range value {
 		m = append(m, v)
@@ -226,14 +228,14 @@ func SetNestedStringSlice(obj map[string]interface{}, value []string, fields ...
 }
 
 // SetNestedSlice sets the slice value of a nested field.
-// Returns false if value cannot be set because one of the nesting levels is not a map[string]interface{}.
-func SetNestedSlice(obj map[string]interface{}, value []interface{}, fields ...string) bool {
+// Returns an error if value cannot be set because one of the nesting levels is not a map[string]interface{}.
+func SetNestedSlice(obj map[string]interface{}, value []interface{}, fields ...string) error {
 	return SetNestedField(obj, value, fields...)
 }
 
 // SetNestedStringMap sets the map[string]string value of a nested field.
-// Returns false if value cannot be set because one of the nesting levels is not a map[string]interface{}.
-func SetNestedStringMap(obj map[string]interface{}, value map[string]string, fields ...string) bool {
+// Returns an error if value cannot be set because one of the nesting levels is not a map[string]interface{}.
+func SetNestedStringMap(obj map[string]interface{}, value map[string]string, fields ...string) error {
 	m := make(map[string]interface{}, len(value)) // convert map[string]string into map[string]interface{}
 	for k, v := range value {
 		m[k] = v
@@ -242,8 +244,8 @@ func SetNestedStringMap(obj map[string]interface{}, value map[string]string, fie
 }
 
 // SetNestedMap sets the map[string]interface{} value of a nested field.
-// Returns false if value cannot be set because one of the nesting levels is not a map[string]interface{}.
-func SetNestedMap(obj map[string]interface{}, value map[string]interface{}, fields ...string) bool {
+// Returns an error if value cannot be set because one of the nesting levels is not a map[string]interface{}.
+func SetNestedMap(obj map[string]interface{}, value map[string]interface{}, fields ...string) error {
 	return SetNestedField(obj, value, fields...)
 }
 
@@ -268,6 +270,10 @@ func getNestedString(obj map[string]interface{}, fields ...string) string {
 	return val
 }
 
+func jsonPath(fields []string) string {
+	return "." + strings.Join(fields, ".")
+}
+
 func extractOwnerReference(v map[string]interface{}) metav1.OwnerReference {
 	// though this field is a *bool, but when decoded from JSON, it's
 	// unmarshalled as bool.

vendor/k8s.io/client-go/testing/BUILD

@@ -3,6 +3,7 @@ package(default_visibility = ["//visibility:public"])
 load(
     "@io_bazel_rules_go//go:def.bzl",
     "go_library",
+    "go_test",
 )
 
 go_library(
@@ -28,6 +29,24 @@ go_library(
     ],
 )
 
+go_test(
+    name = "go_default_test",
+    srcs = [
+        "fixture_test.go",
+    ],
+    embed = [":go_default_library"],
+    importpath = "k8s.io/client-go/testing",
+    deps = [
+        "//vendor/github.com/stretchr/testify/assert:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/api/meta:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/watch:go_default_library",
+    ],
+)
+
 filegroup(
     name = "package-srcs",
     srcs = glob(["**"]),

vendor/k8s.io/client-go/testing/fixture.go

@@ -29,6 +29,11 @@ import (
 	restclient "k8s.io/client-go/rest"
 )
 
+// FakeWatchBufferSize is the max num of watch event can be buffered in the
+// watch channel. Note that when watch event overflows or exceed this buffer
+// size, manipulations via fake client may be blocked.
+const FakeWatchBufferSize = 128
+
 // ObjectTracker keeps track of objects. It is intended to be used to
 // fake calls to a server by returning objects based on their kind,
 // namespace and name.
@@ -54,6 +59,10 @@ type ObjectTracker interface {
 	// didn't exist in the tracker prior to deletion, Delete returns
 	// no error.
 	Delete(gvr schema.GroupVersionResource, ns, name string) error
+
+	// Watch watches objects from the tracker. Watch returns a channel
+	// which will push added / modified / deleted object.
+	Watch(gvr schema.GroupVersionResource, ns string) (watch.Interface, error)
 }
 
 // ObjectScheme abstracts the implementation of common operations on objects.
@@ -132,6 +141,13 @@ type tracker struct {
 	decoder runtime.Decoder
 	lock    sync.RWMutex
 	objects map[schema.GroupVersionResource][]runtime.Object
+	// The value type of watchers is a map of which the key is either a namespace or
+	// all/non namespace aka "" and its value is list of fake watchers. Each of
+	// fake watcher holds a buffered channel of size "FakeWatchBufferSize" which
+	// is default to 128. Manipulations on resources will broadcast the notification
+	// events into the watchers' channel and note that too many unhandled event may
+	// potentially block the tracker.
+	watchers map[schema.GroupVersionResource]map[string][]*watch.FakeWatcher
 }
 
 var _ ObjectTracker = &tracker{}
@@ -140,9 +156,10 @@ var _ ObjectTracker = &tracker{}
 // of objects for the fake clientset. Mostly useful for unit tests.
 func NewObjectTracker(scheme ObjectScheme, decoder runtime.Decoder) ObjectTracker {
 	return &tracker{
-		scheme:  scheme,
-		decoder: decoder,
-		objects: make(map[schema.GroupVersionResource][]runtime.Object),
+		scheme:   scheme,
+		decoder:  decoder,
+		objects:  make(map[schema.GroupVersionResource][]runtime.Object),
+		watchers: make(map[schema.GroupVersionResource]map[string][]*watch.FakeWatcher),
 	}
 }
 
@@ -185,6 +202,19 @@ func (t *tracker) List(gvr schema.GroupVersionResource, gvk schema.GroupVersionK
 	return list.DeepCopyObject(), nil
 }
 
+func (t *tracker) Watch(gvr schema.GroupVersionResource, ns string) (watch.Interface, error) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+
+	fakewatcher := watch.NewFakeWithChanSize(FakeWatchBufferSize, true)
+
+	if _, exists := t.watchers[gvr]; !exists {
+		t.watchers[gvr] = make(map[string][]*watch.FakeWatcher)
+	}
+	t.watchers[gvr][ns] = append(t.watchers[gvr][ns], fakewatcher)
+	return fakewatcher, nil
+}
+
 func (t *tracker) Get(gvr schema.GroupVersionResource, ns, name string) (runtime.Object, error) {
 	errNotFound := errors.NewNotFound(gvr.GroupResource(), name)
 
@@ -263,6 +293,19 @@ func (t *tracker) Update(gvr schema.GroupVersionResource, obj runtime.Object, ns
 	return t.add(gvr, obj, ns, true)
 }
 
+func (t *tracker) getWatches(gvr schema.GroupVersionResource, ns string) []*watch.FakeWatcher {
+	watches := []*watch.FakeWatcher{}
+	if t.watchers[gvr] != nil {
+		if w := t.watchers[gvr][ns]; w != nil {
+			watches = append(watches, w...)
+		}
+		if w := t.watchers[gvr][""]; w != nil {
+			watches = append(watches, w...)
+		}
+	}
+	return watches
+}
+
 func (t *tracker) add(gvr schema.GroupVersionResource, obj runtime.Object, ns string, replaceExisting bool) error {
 	t.lock.Lock()
 	defer t.lock.Unlock()
@@ -296,6 +339,9 @@ func (t *tracker) add(gvr schema.GroupVersionResource, obj runtime.Object, ns st
 		}
 		if oldMeta.GetNamespace() == newMeta.GetNamespace() && oldMeta.GetName() == newMeta.GetName() {
 			if replaceExisting {
+				for _, w := range t.getWatches(gvr, ns) {
+					w.Modify(obj)
+				}
 				t.objects[gvr][i] = obj
 				return nil
 			}
@@ -310,6 +356,10 @@ func (t *tracker) add(gvr schema.GroupVersionResource, obj runtime.Object, ns st
 
 	t.objects[gvr] = append(t.objects[gvr], obj)
 
+	for _, w := range t.getWatches(gvr, ns) {
+		w.Add(obj)
+	}
+
 	return nil
 }
 
@@ -342,7 +392,11 @@ func (t *tracker) Delete(gvr schema.GroupVersionResource, ns, name string) error
 			return err
 		}
 		if objMeta.GetNamespace() == ns && objMeta.GetName() == name {
+			obj := t.objects[gvr][i]
 			t.objects[gvr] = append(t.objects[gvr][:i], t.objects[gvr][i+1:]...)
+			for _, w := range t.getWatches(gvr, ns) {
+				w.Delete(obj)
+			}
 			found = true
 			break
 		}