add memory filter lable select

Signed-off-by: lengrongfu <1275177125@qq.com>

pkg/storage/memorystorage/watchcache/watch_cache.go

@@ -12,6 +12,7 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/selection"
 	"k8s.io/apimachinery/pkg/watch"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/registry/generic/registry"
@@ -112,6 +113,19 @@ func GetKeyFunc(gvr schema.GroupVersionResource, isNamespaced bool) keyFunc {
 	return kc
 }
 
+func GetAttrsFunc(obj runtime.Object) (labels.Set, fields.Set, error) {
+	accessor, err := meta.Accessor(obj)
+	if err != nil {
+		return nil, nil, err
+	}
+	objLabels := accessor.GetLabels()
+	if objLabels == nil {
+		objLabels = make(map[string]string)
+	}
+	labelSet := labels.Set(objLabels)
+	return labelSet, nil, nil
+}
+
 func storeElementKey(obj interface{}) (string, error) {
 	elem, ok := obj.(*StoreElement)
 	if !ok {
@@ -153,7 +167,7 @@ func NewWatchCache(capacity int, gvr schema.GroupVersionResource, isNamespaced b
 	wc := &WatchCache{
 		capacity:     capacity,
 		KeyFunc:      GetKeyFunc(gvr, isNamespaced),
-		getAttrsFunc: nil,
+		getAttrsFunc: GetAttrsFunc,
 		cache:        make([]*watch.Event, capacity),
 		startIndex:   0,
 		endIndex:     0,
@@ -305,12 +319,67 @@ func (w *WatchCache) WaitUntilFreshAndList(opts *internal.ListOptions) ([]*Store
 					continue
 				}
 			}
+			if filterLabelSelector(opts.LabelSelector, se.Labels) {
+				continue
+			}
 			result = append(result, se)
 		}
 	}
 	return result, w.resourceVersion, nil
 }
 
+// filterLabelSelector returns true if the given label selector matches the given label set. else false.
+func filterLabelSelector(labelSelector labels.Selector, label labels.Set) bool {
+	if labelSelector != nil && label != nil {
+		if requirements, selectable := labelSelector.Requirements(); selectable {
+			for _, requirement := range requirements {
+				values := requirement.Values().List()
+				switch requirement.Operator() {
+				case selection.Exists:
+					if !label.Has(requirement.Key()) {
+						return true
+					}
+				case selection.DoesNotExist:
+					if label.Has(requirement.Key()) {
+						return true
+					}
+				case selection.Equals, selection.DoubleEquals:
+					labelValue := label.Get(requirement.Key())
+					if labelValue != values[0] {
+						return true
+					}
+				case selection.NotEquals:
+					labelValue := label.Get(requirement.Key())
+					if labelValue == values[0] {
+						return true
+					}
+				case selection.In:
+					labelValue := label.Get(requirement.Key())
+					valuesMap := make(map[string]struct{})
+					for _, value := range values {
+						valuesMap[value] = struct{}{}
+					}
+					if _, ok := valuesMap[labelValue]; !ok {
+						return true
+					}
+				case selection.NotIn:
+					labelValue := label.Get(requirement.Key())
+					valuesMap := make(map[string]struct{})
+					for _, value := range values {
+						valuesMap[value] = struct{}{}
+					}
+					if _, ok := valuesMap[labelValue]; ok {
+						return true
+					}
+				default:
+					continue
+				}
+			}
+		}
+	}
+	return false
+}
+
 // WaitUntilFreshAndGet returns list of pointers to <storeElement> objects.
 func (w *WatchCache) WaitUntilFreshAndGet(cluster, namespace, name string) (*StoreElement, error) {
 	w.RLock()