Enable support for indexes in WatchTracker

Kubernetes-commit: 21ec77dc19047215094a538b717e24035ca000ab

pkg/util/flowcontrol/watch_tracker.go

@@ -20,8 +20,13 @@ import (
 	"net/http"
 	"sync"
 
+	metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
+	"k8s.io/apimachinery/pkg/apis/meta/internalversion/scheme"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apiserver/pkg/endpoints/request"
+
+	"k8s.io/klog/v2"
 )
 
 // readOnlyVerbs contains verbs for read-only requests.
@@ -63,20 +68,66 @@ type WatchTracker interface {
 	GetInterestedWatchCount(requestInfo *request.RequestInfo) int
 }
 
+// builtinIndexes represents of set of indexes registered in
+// watchcache that are indexing watches and increase speed of
+// their processing.
+// We define the indexes as a map from a resource to the path
+// to the field in the object on which the index is built.
+type builtinIndexes map[string]string
+
+func getBuiltinIndexes() builtinIndexes {
+	// The only existing indexes as of now are:
+	// - spec.nodeName for pods
+	// - metadata.Name for nodes, secrets and configmaps
+	// However, we can ignore the latter, because the requestInfo.Name
+	// is set for them (i.e. we already catch them correctly).
+	return map[string]string{
+		"pods": "spec.nodeName",
+	}
+}
+
 // watchTracker tracks the number of watches in the system for
 // the purpose of estimating the cost of incoming mutating requests.
 type watchTracker struct {
-	lock sync.Mutex
+	// indexes represents a set of registered indexes.
+	// It can't change after creation.
+	indexes builtinIndexes
 
+	lock       sync.Mutex
 	watchCount map[watchIdentifier]int
 }
 
 func NewWatchTracker() WatchTracker {
 	return &watchTracker{
+		indexes:    getBuiltinIndexes(),
 		watchCount: make(map[watchIdentifier]int),
 	}
 }
 
+const (
+	unsetValue = "<unset>"
+)
+
+func getIndexValue(r *http.Request, field string) string {
+	opts := metainternalversion.ListOptions{}
+	if err := scheme.ParameterCodec.DecodeParameters(r.URL.Query(), metav1.SchemeGroupVersion, &opts); err != nil {
+		klog.Warningf("Couldn't parse list options for %v: %v", r.URL.Query(), err)
+		return unsetValue
+	}
+	if opts.FieldSelector == nil {
+		return unsetValue
+	}
+	if value, ok := opts.FieldSelector.RequiresExactMatch(field); ok {
+		return value
+	}
+	return unsetValue
+}
+
+type indexValue struct {
+	resource string
+	value    string
+}
+
 // RegisterWatch implements WatchTracker interface.
 func (w *watchTracker) RegisterWatch(r *http.Request) ForgetWatchFunc {
 	requestInfo, ok := request.RequestInfoFrom(r.Context())
@@ -84,6 +135,14 @@ func (w *watchTracker) RegisterWatch(r *http.Request) ForgetWatchFunc {
 		return nil
 	}
 
+	var index *indexValue
+	if indexField, ok := w.indexes[requestInfo.Resource]; ok {
+		index = &indexValue{
+			resource: requestInfo.Resource,
+			value:    getIndexValue(r, indexField),
+		}
+	}
+
 	identifier := &watchIdentifier{
 		apiGroup:  requestInfo.APIGroup,
 		resource:  requestInfo.Resource,
@@ -93,16 +152,40 @@ func (w *watchTracker) RegisterWatch(r *http.Request) ForgetWatchFunc {
 
 	w.lock.Lock()
 	defer w.lock.Unlock()
-	w.watchCount[*identifier]++
-	return w.forgetWatch(identifier)
+	w.updateIndexLocked(identifier, index, 1)
+	return w.forgetWatch(identifier, index)
 }
 
-func (w *watchTracker) forgetWatch(identifier *watchIdentifier) ForgetWatchFunc {
+func (w *watchTracker) updateIndexLocked(identifier *watchIdentifier, index *indexValue, incr int) {
+	if index == nil {
+		w.watchCount[*identifier] += incr
+	} else {
+		// For resources with defined index, for a given watch event we are
+		// only processing the watchers that:
+		// (a) do not specify field selector for an index field
+		// (b) do specify field selector with the value equal to the value
+		//     coming from the processed object
+		//
+		// TODO(wojtek-t): For the sake of making progress and initially
+		// simplifying the implementation, we approximate (b) for all values
+		// as the value for an empty string. The assumption we're making here
+		// is that the difference between the actual number of watchers that
+		// will be processed, i.e. (a)+(b) above and the one from our
+		// approximation i.e. (a)+[(b) for field value of ""] will be small.
+		// This seem to be true in almost all production clusters, which makes
+		// it a reasonable first step simplification to unblock progres on it.
+		if index.value == unsetValue || index.value == "" {
+			w.watchCount[*identifier]++
+		}
+	}
+}
+
+func (w *watchTracker) forgetWatch(identifier *watchIdentifier, index *indexValue) ForgetWatchFunc {
 	return func() {
 		w.lock.Lock()
 		defer w.lock.Unlock()
 
-		w.watchCount[*identifier]--
+		w.updateIndexLocked(identifier, index, -1)
 		if w.watchCount[*identifier] == 0 {
 			delete(w.watchCount, *identifier)
 		}

pkg/util/flowcontrol/watch_tracker_test.go

@@ -257,3 +257,73 @@ func TestGetInterestedWatchCount(t *testing.T) {
 	}
 
 }
+
+func TestGetInterestedWatchCountWithIndex(t *testing.T) {
+	watchTracker := NewWatchTracker()
+
+	registeredWatches := []*http.Request{
+		httpRequest("GET", "api/v1/pods", "watch=true"),
+		httpRequest("GET", "api/v1/namespaces/foo/pods", "watch=true"),
+		httpRequest("GET", "api/v1/namespaces/foo/pods", "watch=true&fieldSelector=metadata.name=mypod"),
+		httpRequest("GET", "api/v1/namespaces/foo/pods", "watch=true&fieldSelector=spec.nodeName"),
+		// The watches below will be ignored due to index.
+		httpRequest("GET", "api/v1/namespaces/foo/pods", "watch=true&fieldSelector=spec.nodeName=node1"),
+		httpRequest("GET", "api/v1/namespaces/foo/pods", "watch=true&fieldSelector=spec.nodeName=node2"),
+	}
+	requestInfoFactory := &request.RequestInfoFactory{
+		APIPrefixes:          sets.NewString("api", "apis"),
+		GrouplessAPIPrefixes: sets.NewString("api"),
+	}
+	for _, req := range registeredWatches {
+		requestInfo, err := requestInfoFactory.NewRequestInfo(req)
+		if err != nil {
+			t.Fatalf("unexpected error from requestInfo creation: %#v", err)
+		}
+		r := req.WithContext(request.WithRequestInfo(context.Background(), requestInfo))
+		if forget := watchTracker.RegisterWatch(r); forget == nil {
+			t.Errorf("watch wasn't registered: %#v", requestInfo)
+		}
+	}
+
+	testCases := []struct {
+		name     string
+		request  *http.Request
+		expected int
+	}{
+		{
+			name:     "pod creation in foo namespace",
+			request:  httpRequest("POST", "/api/v1/namespaces/foo/pods", ""),
+			expected: 3,
+		},
+		{
+			name:     "mypod update in foo namespace",
+			request:  httpRequest("PUT", "/api/v1/namespaces/foo/pods/mypod", ""),
+			expected: 4,
+		},
+		{
+			name:     "mypod patch in foo namespace",
+			request:  httpRequest("PATCH", "/api/v1/namespaces/foo/pods/mypod", ""),
+			expected: 4,
+		},
+		{
+			name:     "mypod deletion in foo namespace",
+			request:  httpRequest("DELETE", "/api/v1/namespaces/foo/pods/mypod", ""),
+			expected: 4,
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			requestInfo, err := requestInfoFactory.NewRequestInfo(testCase.request)
+			if err != nil {
+				t.Fatalf("unexpected error from requestInfo creation: %#v", err)
+			}
+
+			count := watchTracker.GetInterestedWatchCount(requestInfo)
+			if count != testCase.expected {
+				t.Errorf("unexpected interested watch count: %d, expected %d", count, testCase.expected)
+			}
+		})
+	}
+
+}