Skip to content

Commit

Permalink
Merge pull request #87913 from cheftako/master
Browse files Browse the repository at this point in the history 
Add code to fix kubelet/metrics memory issue.
  • Loading branch information
@k8s-ci-robot
k8s-ci-robot committed Feb 8, 2020
2 parents 0f13c5c + 9802bfc commit b3ba969
Show file tree
Hide file tree
Showing 2 changed files with 105 additions and 5 deletions.
56 changes: 52 additions & 4 deletions pkg/kubelet/server/server.go
View file
Expand Up @@ -89,6 +89,7 @@ type Server struct {
auth AuthInterface
host HostInterface
restfulCont containerInterface
metricsBuckets map[string]bool
resourceAnalyzer stats.ResourceAnalyzer
redirectContainerStreaming bool
}
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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").
Expand All @@ -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())

Expand All @@ -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,
Expand All @@ -335,13 +364,15 @@ 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)
s.restfulCont.Handle(proberMetricsPath,
compbasemetrics.HandlerFor(p, compbasemetrics.HandlerOpts{ErrorHandling: compbasemetrics.ContinueOnError}),
)

s.addMetricsBucketMatcher("spec")
if enableCAdvisorJSONEndpoints {
ws := new(restful.WebService)
ws.
Expand All @@ -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")
Expand All @@ -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")
Expand All @@ -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")
Expand All @@ -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")
Expand All @@ -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)
Expand All @@ -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")
Expand All @@ -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 {
Expand All @@ -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) {
Expand All @@ -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/").
Expand All @@ -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)
}
Expand All @@ -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}
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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))

Expand Down
54 changes: 53 additions & 1 deletion pkg/kubelet/server/server_test.go
View file
Expand Up @@ -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()
Expand Down Expand Up @@ -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))
}
}

0 comments on commit b3ba969

Please sign in to comment.