perf(vk): cache podmetrics and stats

cmd/virtual-kubelet/root/flag.go

@@ -73,6 +73,10 @@ func InstallFlags(flags *pflag.FlagSet, o *Opts) {
 		"Name of the real load balancer class to use for the actual load balancer")
 	flags.BoolVar(&o.EnableMetrics, "metrics-enabled", false, "Enable the metrics server")
 	flags.StringVar(&o.MetricsAddress, "metrics-address", ":8082", "The address to listen to for metrics requests")
+	flags.DurationVar(&o.MetricsProxyCacheTTL, "metrics-proxy-cache-ttl", o.MetricsProxyCacheTTL,
+		"The time-to-live of the cached responses of the metrics proxy routes (/metrics, /metrics/cadvisor, /metrics/resource, /metrics/probes)")
+	flags.DurationVar(&o.StatsSummaryCacheTTL, "stats-summary-cache-ttl", o.StatsSummaryCacheTTL,
+		"The time-to-live of the cached node stats summary served at /stats/summary")
 	flags.StringVar(&o.HomeAPIServerHost, "home-api-server-host", "",
 		"Home cluster API server HOST, this parameter is optional and required only to override the default values")
 	flags.StringVar(&o.HomeAPIServerPort, "home-api-server-port", "",

cmd/virtual-kubelet/root/http.go

@@ -43,6 +43,7 @@ import (
 	"k8s.io/klog/v2"
 
 	liqov1beta1 "github.com/liqotech/liqo/apis/core/v1beta1"
+	"github.com/liqotech/liqo/pkg/utils/cache"
 	"github.com/liqotech/liqo/pkg/virtualKubelet/reflection/workload"
 	"github.com/liqotech/liqo/pkg/vkMachinery"
 )
@@ -77,7 +78,7 @@ func setupHTTPServer(ctx context.Context, handler workload.PodHandler, localClie
 	mux := http.NewServeMux()
 
 	cl := kubernetes.NewForConfigOrDie(remoteConfig)
-	attachMetricsRoutes(ctx, mux, cl.RESTClient(), cfg.HomeCluster.GetClusterID(), metricsNodeSelector)
+	attachMetricsRoutes(ctx, mux, cl.RESTClient(), cfg.HomeCluster.GetClusterID(), metricsNodeSelector, cfg.MetricsProxyCacheTTL)
 
 	podRoutes := api.PodHandlerConfig{
 		RunInContainer:        handler.Exec,
@@ -193,8 +194,22 @@ func setupHealthServer() {
 	}
 }
 
+const (
+	// metricsProxyErrorCacheTTL is the time-to-live of failed metrics proxy responses. Errors are
+	// cached briefly (much shorter than successful ones), so that a persistently failing scraping
+	// is not retried at each request, while still allowing a prompt recovery once solved.
+	// The time-to-live of successful responses is configurable through the metrics-proxy-cache-ttl flag.
+	metricsProxyErrorCacheTTL = 5 * time.Second
+)
+
+// scrapeResponse is the cached outcome of a proxied scrape: the response body and the status code.
+type scrapeResponse struct {
+	data       []byte
+	statusCode int
+}
+
 func attachMetricsRoutes(ctx context.Context, mux *http.ServeMux, cl rest.Interface,
-	localClusterID liqov1beta1.ClusterID, metricsNodeSelector string) {
+	localClusterID liqov1beta1.ClusterID, metricsNodeSelector string, cacheTTL time.Duration) {
 	// Restrict the scraping to the remote nodes matching the given label selector (i.e., the ones
 	// targeted by this virtual kubelet through its offloading patch), so that node and pod metrics
 	// are not aggregated over the whole remote cluster. An empty selector preserves the aggregated
@@ -204,6 +219,10 @@ func attachMetricsRoutes(ctx context.Context, mux *http.ServeMux, cl rest.Interf
 		query.Add("nodeSelector", metricsNodeSelector)
 	}
 
+	// scrapeCache collapses concurrent and near-duplicate scrapes into a single remote sweep. The
+	// key is the full outgoing URI, hence requests differing in path or query do not alias.
+	scrapeCache := cache.New[scrapeResponse](cacheTTL, metricsProxyErrorCacheTTL)
+
 	handlerFunc := func(w http.ResponseWriter, r *http.Request) {
 		klog.Infof("Received request for %s from %s (user-agent: %q)", r.RequestURI, r.RemoteAddr, r.UserAgent())
 
@@ -221,26 +240,35 @@ func attachMetricsRoutes(ctx context.Context, mux *http.ServeMux, cl rest.Interf
 			uri += "?" + reqQuery.Encode()
 		}
 
-		res := cl.Get().RequestURI(uri).Do(ctx)
-		err := res.Error()
-		if err != nil {
-			klog.Error(err)
-			http.Error(w, "Server Error", http.StatusInternalServerError)
-			return
-		}
+		// ctx is the server-lifetime context (not r.Context()), hence deliberately shared across all
+		// scrapes: the singleflight computation below is therefore never tied to a single client, so
+		// one disconnecting caller cannot abort it for the others waiting on the same key. Keep it
+		// this way: using r.Context() would let the first caller's cancellation poison the shared
+		// result for every waiter.
+		resp, err := scrapeCache.Do(ctx, uri, func(ctx context.Context) (scrapeResponse, error) {
+			res := cl.Get().RequestURI(uri).Do(ctx)
+			if err := res.Error(); err != nil {
+				return scrapeResponse{}, err
+			}
 
-		var statusCode int
-		res.StatusCode(&statusCode)
+			var statusCode int
+			res.StatusCode(&statusCode)
+
+			data, err := res.Raw()
+			if err != nil {
+				return scrapeResponse{}, err
+			}
 
-		data, err := res.Raw()
+			return scrapeResponse{data: data, statusCode: statusCode}, nil
+		})
 		if err != nil {
 			klog.Error(err)
 			http.Error(w, "Server Error", http.StatusInternalServerError)
 			return
 		}
 
-		w.WriteHeader(statusCode)
-		if _, err = w.Write(data); err != nil {
+		w.WriteHeader(resp.statusCode)
+		if _, err = w.Write(resp.data); err != nil {
 			klog.Error(err)
 		}
 	}

cmd/virtual-kubelet/root/http_test.go

@@ -0,0 +1,100 @@
+// Copyright 2019-2026 The Liqo Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package root
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	clientscheme "k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest/fake"
+)
+
+var _ = Describe("Metrics proxy routes", func() {
+	const clusterID = "local-cluster-id"
+
+	var (
+		ctx        context.Context
+		mux        *http.ServeMux
+		calls      atomic.Int64
+		failScrape atomic.Bool
+	)
+
+	BeforeEach(func() {
+		ctx = context.Background()
+		calls.Store(0)
+		failScrape.Store(false)
+
+		cl := &fake.RESTClient{
+			NegotiatedSerializer: clientscheme.Codecs.WithoutConversion(),
+			GroupVersion:         schema.GroupVersion{Group: "metrics.liqo.io", Version: "v1beta1"},
+			Client: fake.CreateHTTPClient(func(*http.Request) (*http.Response, error) {
+				calls.Add(1)
+				if failScrape.Load() {
+					return nil, errors.New("scrape failure")
+				}
+				return &http.Response{
+					StatusCode: http.StatusOK,
+					Header:     http.Header{"Content-Type": []string{"text/plain"}},
+					Body:       io.NopCloser(strings.NewReader("metrics-payload")),
+				}, nil
+			}),
+		}
+
+		mux = http.NewServeMux()
+		attachMetricsRoutes(ctx, mux, cl, clusterID, "", 15*time.Second)
+	})
+
+	doRequest := func(path string) *httptest.ResponseRecorder {
+		rec := httptest.NewRecorder()
+		mux.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, path, http.NoBody))
+		return rec
+	}
+
+	It("should proxy the remote metrics", func() {
+		rec := doRequest("/metrics/resource")
+		Expect(rec.Code).To(Equal(http.StatusOK))
+		Expect(rec.Body.String()).To(Equal("metrics-payload"))
+	})
+
+	It("should serve subsequent requests for the same path from the cache", func() {
+		doRequest("/metrics/resource")
+		doRequest("/metrics/resource")
+		Expect(calls.Load()).To(BeNumerically("==", 1))
+	})
+
+	It("should not alias requests differing in path", func() {
+		doRequest("/metrics/resource")
+		doRequest("/metrics/cadvisor")
+		Expect(calls.Load()).To(BeNumerically("==", 2))
+	})
+
+	It("should cache failures briefly", func() {
+		failScrape.Store(true)
+
+		Expect(doRequest("/metrics/resource").Code).To(Equal(http.StatusInternalServerError))
+		Expect(doRequest("/metrics/resource").Code).To(Equal(http.StatusInternalServerError))
+		Expect(calls.Load()).To(BeNumerically("==", 1))
+	})
+})

cmd/virtual-kubelet/root/opts.go

@@ -44,6 +44,11 @@ const (
 	DefaultListenPort           = 10250
 	DefaultNodePingTimeout      = 1 * time.Second
 	DefaultNodeCheckNetwork     = true
+
+	// DefaultMetricsProxyCacheTTL is the default time-to-live of the cached metrics proxy responses.
+	DefaultMetricsProxyCacheTTL = 10 * time.Second
+	// DefaultStatsSummaryCacheTTL is the default time-to-live of the cached node stats summary.
+	DefaultStatsSummaryCacheTTL = 20 * time.Second
 )
 
 // DefaultReflectorsWorkers contains the default number of workers for each reflected resource.
@@ -127,6 +132,11 @@ type Opts struct {
 	EnableMetrics                   bool
 	MetricsAddress                  string
 
+	// MetricsProxyCacheTTL is the time-to-live of the cached responses of the metrics proxy routes.
+	MetricsProxyCacheTTL time.Duration
+	// StatsSummaryCacheTTL is the time-to-live of the cached node stats summary.
+	StatsSummaryCacheTTL time.Duration
+
 	HomeAPIServerHost string
 	HomeAPIServerPort string
 
@@ -162,6 +172,9 @@ func NewOpts() *Opts {
 		NodePingTimeout:   DefaultNodePingTimeout,
 		NodeCheckNetwork:  DefaultNodeCheckNetwork,
 
+		MetricsProxyCacheTTL: DefaultMetricsProxyCacheTTL,
+		StatsSummaryCacheTTL: DefaultStatsSummaryCacheTTL,
+
 		VirtualKubeletLeaseEnabled:       true,
 		VirtualKubeletLeaseLeaseDuration: 15 * time.Second,
 		VirtualKubeletLeaseRenewDeadline: 10 * time.Second,

cmd/virtual-kubelet/root/root.go

@@ -215,6 +215,8 @@ func runRootCommand(ctx context.Context, c *Opts) error {
 		RemoteRealLoadBalancerClassName: c.RemoteRealLoadBalancerClassName,
 		EnableMetrics:                   c.EnableMetrics,
 
+		StatsSummaryCacheTTL: c.StatsSummaryCacheTTL,
+
 		HomeAPIServerHost: c.HomeAPIServerHost,
 		HomeAPIServerPort: c.HomeAPIServerPort,
 

cmd/virtual-kubelet/root/root_suite_test.go

@@ -0,0 +1,27 @@
+// Copyright 2019-2026 The Liqo Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package root
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestRoot(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Root Suite")
+}