intercept kubelet get node request in order to reduce the traffic

pkg/yurthub/cachemanager/cache_manager.go

@@ -52,6 +52,7 @@
 
 var (
 	ErrInMemoryCacheMiss = errors.New("in-memory cache miss")
+	ErrNotNodeOrLease    = errors.New("resource is not node or lease")
 
 	nonCacheableResources = map[string]struct{}{
 		"certificatesigningrequests": {},
@@ -233,20 +234,19 @@
 
 	comp, _ := util.ClientComponentFrom(ctx)
 	// query in-memory cache first
-	var isInMemoryCache = isInMemoryCache(ctx)
 	var isInMemoryCacheMiss bool
-	if isInMemoryCache {
-		if obj, err := cm.queryInMemeryCache(info); err != nil {
-			if err == ErrInMemoryCacheMiss {
-				isInMemoryCacheMiss = true
-				klog.V(4).Infof("in-memory cache miss when handling request %s, fall back to storage query", util.ReqString(req))
-			} else {
-				klog.Errorf("cannot query in-memory cache for reqInfo %s, %v,", util.ReqInfoString(info), err)
-			}
+	if obj, err := cm.queryInMemeryCache(ctx, info); err != nil {
+		if err == ErrInMemoryCacheMiss {
+			isInMemoryCacheMiss = true
+			klog.V(4).Infof("in-memory cache miss when handling request %s, fall back to storage query", util.ReqString(req))
+		} else if err == ErrNotNodeOrLease {
+			klog.V(4).Infof("resource(%s) is not node or lease, it will be found in the disk not cache", info.Resource)
 		} else {
-			klog.V(4).Infof("in-memory cache hit when handling request %s", util.ReqString(req))
-			return obj, nil
+			klog.Errorf("cannot query in-memory cache for reqInfo %s, %v,", util.ReqInfoString(info), err)
 		}
+	} else {
+		klog.V(4).Infof("in-memory cache hit when handling request %s", util.ReqString(req))
+		return obj, nil
 	}
 
 	// fall back to normal query
@@ -275,12 +275,7 @@
 	// While cloud-edge network is broken, the inMemoryCache can only be full filled with data from edge cache,
 	// such as local disk and yurt-coordinator.
 	if isInMemoryCacheMiss {
-		if inMemoryCacheKey, err := inMemoryCacheKeyFunc(info); err != nil {
-			klog.Errorf("cannot in-memory cache key for req %s, %v", util.ReqString(req), err)
-		} else {
-			cm.inMemoryCacheFor(inMemoryCacheKey, obj)
-			klog.V(4).Infof("use obj from backend storage to update in-memory cache of key %s", inMemoryCacheKey)
-		}
+		return obj, cm.updateInMemoryCache(ctx, info, obj)
 	}
 	return obj, nil
 }
@@ -433,9 +428,15 @@
 				} else {
 					updateObjCnt++
 				}
+				errMsg := cm.updateInMemoryCache(ctx, info, obj)
+				if errMsg != nil {
+					klog.Errorf("failed to update cache, %v", errMsg)
+				}
 			case watch.Deleted:
 				err = cm.storage.Delete(key)
 				delObjCnt++
+				// for now, If it's a delete request, no need to modify the inmemory cache,
+				// because currently, there shouldn't be any delete requests for nodes or leases.
 			default:
 				// impossible go to here
 			}
@@ -606,6 +607,10 @@
 		return err
 	}
 
+	return cm.updateInMemoryCache(ctx, info, obj)
+}
+
+func (cm *cacheManager) updateInMemoryCache(ctx context.Context, info *apirequest.RequestInfo, obj runtime.Object) error {
 	// update the in-memory cache with cloud response
 	if !isInMemoryCache(ctx) {
 		return nil
@@ -803,7 +808,11 @@
 	return cm.restMapperManager.DeleteKindFor(gvr)
 }
 
-func (cm *cacheManager) queryInMemeryCache(reqInfo *apirequest.RequestInfo) (runtime.Object, error) {
+func (cm *cacheManager) queryInMemeryCache(ctx context.Context, reqInfo *apirequest.RequestInfo) (runtime.Object, error) {
+	if !isInMemoryCache(ctx) {
+		return nil, ErrNotNodeOrLease
+	}
+
 	key, err := inMemoryCacheKeyFunc(reqInfo)
 	if err != nil {
 		return nil, err

pkg/yurthub/proxy/proxy.go

@@ -191,6 +191,12 @@ func (p *yurtReverseProxy) ServeHTTP(rw http.ResponseWriter, req *http.Request)
 	switch {
 	case util.IsKubeletLeaseReq(req):
 		p.handleKubeletLease(rw, req)
+	case util.IsKubeletGetNodeReq(req):
+		if p.localProxy != nil {
+			p.localProxy.ServeHTTP(rw, req)
+		} else {
+			p.loadBalancer.ServeHTTP(rw, req)
+		}
 	case util.IsEventCreateRequest(req):
 		p.eventHandler(rw, req)
 	case util.IsPoolScopedResouceListWatchRequest(req):

pkg/yurthub/proxy/util/util.go

@@ -331,7 +331,7 @@
 					}
 				} else if info.Verb == "get" {
 					query := req.URL.Query()
-					if str, _ := query["timeout"]; len(str) > 0 {
+					if str := query["timeout"]; len(str) > 0 {
 						if t, err := time.ParseDuration(str[0]); err == nil {
 							if t > time.Duration(getAndListTimeoutReduce)*time.Second {
 								timeout = t - time.Duration(getAndListTimeoutReduce)*time.Second
@@ -415,6 +415,18 @@
 	return true
 }
 
+// IsKubeletGetNodeReq judge whether the request is a get node request from kubelet
+func IsKubeletGetNodeReq(req *http.Request) bool {
+	ctx := req.Context()
+	if comp, ok := util.ClientComponentFrom(ctx); !ok || comp != "kubelet" {
+		return false
+	}
+	if info, ok := apirequest.RequestInfoFrom(ctx); !ok || info.Resource != "nodes" || info.Verb != "get" {
+		return false
+	}
+	return true
+}
+
 // WriteObject write object to response writer
 func WriteObject(statusCode int, obj runtime.Object, w http.ResponseWriter, req *http.Request) error {
 	ctx := req.Context()
@@ -503,10 +515,6 @@
 	}
 
 	streamingEncoder := streaming.NewEncoder(framer.NewFrameWriter(rw), streamingSerializer)
-	if err != nil {
-		klog.Errorf("ReListWatchReq %s failed with error = %s", util.ReqString(req), err.Error())
-		return
-	}
 
 	outEvent := &metav1.WatchEvent{
 		Type: string(watch.Error),