Merge pull request #87913 from cheftako/master

Add code to fix kubelet/metrics memory issue.

pkg/kubelet/server/server.go

@@ -89,6 +89,7 @@ type Server struct {
 	auth                       AuthInterface
 	host                       HostInterface
 	restfulCont                containerInterface
+	metricsBuckets             map[string]bool
 	resourceAnalyzer           stats.ResourceAnalyzer
 	redirectContainerStreaming bool
 }
@@ -225,6 +226,7 @@ func NewServer(
 		resourceAnalyzer:           resourceAnalyzer,
 		auth:                       auth,
 		restfulCont:                &filteringContainer{Container: restful.NewContainer()},
+		metricsBuckets:             make(map[string]bool),
 		redirectContainerStreaming: redirectContainerStreaming,
 	}
 	if auth != nil {
@@ -280,14 +282,32 @@ func (s *Server) InstallAuthFilter() {
 	})
 }
 
+// addMetricsBucketMatcher adds a regexp matcher and the relevant bucket to use when
+// it matches. Please be aware this is not thread safe and should not be used dynamically
+func (s *Server) addMetricsBucketMatcher(bucket string) {
+	s.metricsBuckets[bucket] = true
+}
+
+// getMetricBucket find the appropriate metrics reporting bucket for the given path
+func (s *Server) getMetricBucket(path string) string {
+	root := getURLRootPath(path)
+	if s.metricsBuckets[root] == true {
+		return root
+	}
+	return "Invalid path"
+}
+
 // InstallDefaultHandlers registers the default set of supported HTTP request
 // patterns with the restful Container.
 func (s *Server) InstallDefaultHandlers(enableCAdvisorJSONEndpoints bool) {
+	s.addMetricsBucketMatcher("healthz")
 	healthz.InstallHandler(s.restfulCont,
 		healthz.PingHealthz,
 		healthz.LogHealthz,
 		healthz.NamedCheck("syncloop", s.syncLoopHealthCheck),
 	)
+
+	s.addMetricsBucketMatcher("pods")
 	ws := new(restful.WebService)
 	ws.
 		Path("/pods").
@@ -297,7 +317,14 @@ func (s *Server) InstallDefaultHandlers(enableCAdvisorJSONEndpoints bool) {
 		Operation("getPods"))
 	s.restfulCont.Add(ws)
 
+	s.addMetricsBucketMatcher("stats")
 	s.restfulCont.Add(stats.CreateHandlers(statsPath, s.host, s.resourceAnalyzer, enableCAdvisorJSONEndpoints))
+
+	s.addMetricsBucketMatcher("metrics")
+	s.addMetricsBucketMatcher("metrics/cadvisor")
+	s.addMetricsBucketMatcher("metrics/probes")
+	s.addMetricsBucketMatcher("metrics/resource/v1alpha1")
+	s.addMetricsBucketMatcher("metrics/resource")
 	//lint:ignore SA1019 https://github.com/kubernetes/enhancements/issues/1206
 	s.restfulCont.Handle(metricsPath, legacyregistry.Handler())
 
@@ -321,12 +348,14 @@ func (s *Server) InstallDefaultHandlers(enableCAdvisorJSONEndpoints bool) {
 	)
 
 	// deprecated endpoint which will be removed in release 1.20.0+.
+	s.addMetricsBucketMatcher("metrics/resource/v1alpha1")
 	v1alpha1ResourceRegistry := compbasemetrics.NewKubeRegistry()
 	v1alpha1ResourceRegistry.CustomMustRegister(stats.NewPrometheusResourceMetricCollector(s.resourceAnalyzer, v1alpha1.Config()))
 	s.restfulCont.Handle(path.Join(resourceMetricsPath, v1alpha1.Version),
 		compbasemetrics.HandlerFor(v1alpha1ResourceRegistry, compbasemetrics.HandlerOpts{ErrorHandling: compbasemetrics.ContinueOnError}),
 	)
 
+	s.addMetricsBucketMatcher("metrics/resource")
 	resourceRegistry := compbasemetrics.NewKubeRegistry()
 	resourceRegistry.CustomMustRegister(collectors.NewResourceMetricsCollector(s.resourceAnalyzer))
 	s.restfulCont.Handle(resourceMetricsPath,
@@ -335,6 +364,7 @@ func (s *Server) InstallDefaultHandlers(enableCAdvisorJSONEndpoints bool) {
 
 	// prober metrics are exposed under a different endpoint
 
+	s.addMetricsBucketMatcher("metrics/probes")
 	p := compbasemetrics.NewKubeRegistry()
 	_ = compbasemetrics.RegisterProcessStartTime(p.Register)
 	p.MustRegister(prober.ProberResults)
@@ -342,6 +372,7 @@ func (s *Server) InstallDefaultHandlers(enableCAdvisorJSONEndpoints bool) {
 		compbasemetrics.HandlerFor(p, compbasemetrics.HandlerOpts{ErrorHandling: compbasemetrics.ContinueOnError}),
 	)
 
+	s.addMetricsBucketMatcher("spec")
 	if enableCAdvisorJSONEndpoints {
 		ws := new(restful.WebService)
 		ws.
@@ -361,6 +392,7 @@ const pprofBasePath = "/debug/pprof/"
 func (s *Server) InstallDebuggingHandlers(criHandler http.Handler) {
 	klog.Infof("Adding debug handlers to kubelet server.")
 
+	s.addMetricsBucketMatcher("run")
 	ws := new(restful.WebService)
 	ws.
 		Path("/run")
@@ -372,6 +404,7 @@ func (s *Server) InstallDebuggingHandlers(criHandler http.Handler) {
 		Operation("getRun"))
 	s.restfulCont.Add(ws)
 
+	s.addMetricsBucketMatcher("exec")
 	ws = new(restful.WebService)
 	ws.
 		Path("/exec")
@@ -389,6 +422,7 @@ func (s *Server) InstallDebuggingHandlers(criHandler http.Handler) {
 		Operation("getExec"))
 	s.restfulCont.Add(ws)
 
+	s.addMetricsBucketMatcher("attach")
 	ws = new(restful.WebService)
 	ws.
 		Path("/attach")
@@ -406,6 +440,7 @@ func (s *Server) InstallDebuggingHandlers(criHandler http.Handler) {
 		Operation("getAttach"))
 	s.restfulCont.Add(ws)
 
+	s.addMetricsBucketMatcher("portForward")
 	ws = new(restful.WebService)
 	ws.
 		Path("/portForward")
@@ -423,6 +458,7 @@ func (s *Server) InstallDebuggingHandlers(criHandler http.Handler) {
 		Operation("getPortForward"))
 	s.restfulCont.Add(ws)
 
+	s.addMetricsBucketMatcher("logs")
 	ws = new(restful.WebService)
 	ws.
 		Path(logsPath)
@@ -435,6 +471,7 @@ func (s *Server) InstallDebuggingHandlers(criHandler http.Handler) {
 		Param(ws.PathParameter("logpath", "path to the log").DataType("string")))
 	s.restfulCont.Add(ws)
 
+	s.addMetricsBucketMatcher("containerLogs")
 	ws = new(restful.WebService)
 	ws.
 		Path("/containerLogs")
@@ -443,8 +480,10 @@ func (s *Server) InstallDebuggingHandlers(criHandler http.Handler) {
 		Operation("getContainerLogs"))
 	s.restfulCont.Add(ws)
 
+	s.addMetricsBucketMatcher("configz")
 	configz.InstallHandler(s.restfulCont)
 
+	s.addMetricsBucketMatcher("debug")
 	handlePprofEndpoint := func(req *restful.Request, resp *restful.Response) {
 		name := strings.TrimPrefix(req.Request.URL.Path, pprofBasePath)
 		switch name {
@@ -460,7 +499,6 @@ func (s *Server) InstallDebuggingHandlers(criHandler http.Handler) {
 			pprof.Index(resp, req.Request)
 		}
 	}
-
 	// Setup pprof handlers.
 	ws = new(restful.WebService).Path(pprofBasePath)
 	ws.Route(ws.GET("/{subpath:*}").To(func(req *restful.Request, resp *restful.Response) {
@@ -473,6 +511,7 @@ func (s *Server) InstallDebuggingHandlers(criHandler http.Handler) {
 	s.restfulCont.Handle("/debug/flags/v", routes.StringFlagPutHandler(logs.GlogSetter))
 
 	// The /runningpods endpoint is used for testing only.
+	s.addMetricsBucketMatcher("runningpods")
 	ws = new(restful.WebService)
 	ws.
 		Path("/runningpods/").
@@ -482,6 +521,7 @@ func (s *Server) InstallDebuggingHandlers(criHandler http.Handler) {
 		Operation("getRunningPods"))
 	s.restfulCont.Add(ws)
 
+	s.addMetricsBucketMatcher("cri")
 	if criHandler != nil {
 		s.restfulCont.Handle("/cri/", criHandler)
 	}
@@ -493,6 +533,14 @@ func (s *Server) InstallDebuggingDisabledHandlers() {
 		http.Error(w, "Debug endpoints are disabled.", http.StatusMethodNotAllowed)
 	})
 
+	s.addMetricsBucketMatcher("run")
+	s.addMetricsBucketMatcher("exec")
+	s.addMetricsBucketMatcher("attach")
+	s.addMetricsBucketMatcher("portForward")
+	s.addMetricsBucketMatcher("containerLogs")
+	s.addMetricsBucketMatcher("runningpods")
+	s.addMetricsBucketMatcher("pprof")
+	s.addMetricsBucketMatcher("logs")
 	paths := []string{
 		"/run/", "/exec/", "/attach/", "/portForward/", "/containerLogs/",
 		"/runningpods/", pprofBasePath, logsPath}
@@ -809,10 +857,10 @@ func (s *Server) getPortForward(request *restful.Request, response *restful.Resp
 	proxyStream(response.ResponseWriter, request.Request, url)
 }
 
-// trimURLPath trims a URL path.
+// getURLRootPath trims a URL path.
 // For paths in the format of "/metrics/xxx", "metrics/xxx" is returned;
 // For all other paths, the first part of the path is returned.
-func trimURLPath(path string) string {
+func getURLRootPath(path string) string {
 	parts := strings.SplitN(strings.TrimPrefix(path, "/"), "/", 3)
 	if len(parts) == 0 {
 		return path
@@ -860,7 +908,7 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		serverType = "readwrite"
 	}
 
-	method, path := req.Method, trimURLPath(req.URL.Path)
+	method, path := req.Method, s.getMetricBucket(req.URL.Path)
 
 	longRunning := strconv.FormatBool(isLongRunningRequest(path))
 

pkg/kubelet/server/server_test.go

@@ -1471,6 +1471,58 @@ func TestCRIHandler(t *testing.T) {
 	assert.Equal(t, query, fw.criHandler.RequestReceived.URL.RawQuery)
 }
 
+func TestMetricBuckets(t *testing.T) {
+	tests := map[string]struct {
+		url    string
+		bucket string
+	}{
+		"healthz endpoint":                {url: "/healthz", bucket: "healthz"},
+		"attach":                          {url: "/attach/podNamespace/podID/containerName", bucket: "attach"},
+		"attach with uid":                 {url: "/attach/podNamespace/podID/uid/containerName", bucket: "attach"},
+		"configz":                         {url: "/configz", bucket: "configz"},
+		"containerLogs":                   {url: "/containerLogs/podNamespace/podID/containerName", bucket: "containerLogs"},
+		"cri":                             {url: "/cri/", bucket: "cri"},
+		"cri with sub":                    {url: "/cri/foo", bucket: "cri"},
+		"debug v flags":                   {url: "/debug/flags/v", bucket: "debug"},
+		"pprof with sub":                  {url: "/debug/pprof/subpath", bucket: "debug"},
+		"exec":                            {url: "/exec/podNamespace/podID/containerName", bucket: "exec"},
+		"exec with uid":                   {url: "/exec/podNamespace/podID/uid/containerName", bucket: "exec"},
+		"healthz":                         {url: "/healthz/", bucket: "healthz"},
+		"healthz log sub":                 {url: "/healthz/log", bucket: "healthz"},
+		"healthz ping":                    {url: "/healthz/ping", bucket: "healthz"},
+		"healthz sync loop":               {url: "/healthz/syncloop", bucket: "healthz"},
+		"logs":                            {url: "/logs/", bucket: "logs"},
+		"logs with path":                  {url: "/logs/logpath", bucket: "logs"},
+		"metrics":                         {url: "/metrics", bucket: "metrics"},
+		"metrics cadvisor sub":            {url: "/metrics/cadvisor", bucket: "metrics/cadvisor"},
+		"metrics probes sub":              {url: "/metrics/probes", bucket: "metrics/probes"},
+		"metrics resource v1alpha1":       {url: "/metrics/resource/v1alpha1", bucket: "metrics/resource"},
+		"metrics resource sub":            {url: "/metrics/resource", bucket: "metrics/resource"},
+		"pods":                            {url: "/pods/", bucket: "pods"},
+		"portForward":                     {url: "/portForward/podNamespace/podID", bucket: "portForward"},
+		"portForward with uid":            {url: "/portForward/podNamespace/podID/uid", bucket: "portForward"},
+		"run":                             {url: "/run/podNamespace/podID/containerName", bucket: "run"},
+		"run with uid":                    {url: "/run/podNamespace/podID/uid/containerName", bucket: "run"},
+		"runningpods":                     {url: "/runningpods/", bucket: "runningpods"},
+		"spec":                            {url: "/spec/", bucket: "spec"},
+		"stats":                           {url: "/stats/", bucket: "stats"},
+		"stats container sub":             {url: "/stats/container", bucket: "stats"},
+		"stats summary sub":               {url: "/stats/summary", bucket: "stats"},
+		"stats containerName with uid":    {url: "/stats/namespace/podName/uid/containerName", bucket: "stats"},
+		"stats containerName":             {url: "/stats/podName/containerName", bucket: "stats"},
+		"invalid path":                    {url: "/junk", bucket: "Invalid path"},
+		"invalid path starting with good": {url: "/healthzjunk", bucket: "Invalid path"},
+	}
+	fw := newServerTest()
+	defer fw.testHTTPServer.Close()
+
+	for _, test := range tests {
+		path := test.url
+		bucket := test.bucket
+		require.Equal(t, fw.serverUnderTest.getMetricBucket(path), bucket)
+	}
+}
+
 func TestDebuggingDisabledHandlers(t *testing.T) {
 	fw := newServerTestWithDebug(false, false, nil)
 	defer fw.testHTTPServer.Close()
@@ -1544,6 +1596,6 @@ func TestTrimURLPath(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		assert.Equal(t, test.expected, trimURLPath(test.path), fmt.Sprintf("path is: %s", test.path))
+		assert.Equal(t, test.expected, getURLRootPath(test.path), fmt.Sprintf("path is: %s", test.path))
 	}
 }