Merge pull request #63406 from derekwaynecarr/label-pod-cgroups

Automatic merge from submit-queue (batch tested with PRs 60200, 63623, 63406). If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

Apply pod name and namespace labels for pod cgroup for cadvisor metrics

**What this PR does / why we need it**:
1. Enable Prometheus users to determine usage by pod name and namespace for pod cgroup sandbox.
1. Label cAdvisor metrics for pod cgroups by pod name and namespace.
1. Aligns with kubelet stats summary endpoint pod cpu and memory stats.

**Special notes for your reviewer**:
This provides parity with the summary API enhancements done here:
https://github.com/kubernetes/kubernetes/pull/55969

**Release note**:
```release-note
Apply pod name and namespace labels to pod cgroup in cAdvisor metrics
```

pkg/kubelet/cm/BUILD

@@ -188,6 +188,7 @@ go_test(
             "container_manager_linux_test.go",
             "helpers_linux_test.go",
             "node_container_manager_test.go",
+            "pod_container_manager_linux_test.go",
         ],
         "//conditions:default": [],
     }),
@@ -200,6 +201,7 @@ go_test(
             "//vendor/github.com/stretchr/testify/require:go_default_library",
             "//vendor/k8s.io/api/core/v1:go_default_library",
             "//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
+            "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
         ],
         "//conditions:default": [],
     }),

pkg/kubelet/cm/pod_container_manager_linux.go

@@ -184,6 +184,31 @@ func (m *podContainerManagerImpl) ReduceCPULimits(podCgroup CgroupName) error {
 	return m.cgroupManager.ReduceCPULimits(podCgroup)
 }
 
+// IsPodCgroup returns true if the literal cgroupfs name corresponds to a pod
+func (m *podContainerManagerImpl) IsPodCgroup(cgroupfs string) (bool, types.UID) {
+	// convert the literal cgroupfs form to the driver specific value
+	cgroupName := m.cgroupManager.CgroupName(cgroupfs)
+	qosContainersList := [3]CgroupName{m.qosContainersInfo.BestEffort, m.qosContainersInfo.Burstable, m.qosContainersInfo.Guaranteed}
+	basePath := ""
+	for _, qosContainerName := range qosContainersList {
+		// a pod cgroup is a direct child of a qos node, so check if its a match
+		if len(cgroupName) == len(qosContainerName)+1 {
+			basePath = cgroupName[len(qosContainerName)]
+		}
+	}
+	if basePath == "" {
+		return false, types.UID("")
+	}
+	if !strings.HasPrefix(basePath, podCgroupNamePrefix) {
+		return false, types.UID("")
+	}
+	parts := strings.Split(basePath, podCgroupNamePrefix)
+	if len(parts) != 2 {
+		return false, types.UID("")
+	}
+	return true, types.UID(parts[1])
+}
+
 // GetAllPodsFromCgroups scans through all the subsystems of pod cgroups
 // Get list of pods whose cgroup still exist on the cgroup mounts
 func (m *podContainerManagerImpl) GetAllPodsFromCgroups() (map[types.UID]CgroupName, error) {
@@ -278,3 +303,7 @@ func (m *podContainerManagerNoop) ReduceCPULimits(_ CgroupName) error {
 func (m *podContainerManagerNoop) GetAllPodsFromCgroups() (map[types.UID]CgroupName, error) {
 	return nil, nil
 }
+
+func (m *podContainerManagerNoop) IsPodCgroup(cgroupfs string) (bool, types.UID) {
+	return false, types.UID("")
+}

pkg/kubelet/cm/pod_container_manager_linux_test.go

@@ -0,0 +1,125 @@
+// +build linux
+
+/*
+Copyright 2016 The Kubernetes 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 cm
+
+import (
+	"strings"
+	"testing"
+
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+func TestIsCgroupPod(t *testing.T) {
+	qosContainersInfo := QOSContainersInfo{
+		Guaranteed: RootCgroupName,
+		Burstable:  NewCgroupName(RootCgroupName, strings.ToLower(string(v1.PodQOSBurstable))),
+		BestEffort: NewCgroupName(RootCgroupName, strings.ToLower(string(v1.PodQOSBestEffort))),
+	}
+	podUID := types.UID("123")
+	testCases := []struct {
+		input          CgroupName
+		expectedResult bool
+		expectedUID    types.UID
+	}{
+		{
+			input:          RootCgroupName,
+			expectedResult: false,
+			expectedUID:    types.UID(""),
+		},
+		{
+			input:          NewCgroupName(qosContainersInfo.Guaranteed),
+			expectedResult: false,
+			expectedUID:    types.UID(""),
+		},
+		{
+			input:          NewCgroupName(qosContainersInfo.Guaranteed, GetPodCgroupNameSuffix(podUID)),
+			expectedResult: true,
+			expectedUID:    podUID,
+		},
+		{
+			input:          NewCgroupName(qosContainersInfo.Guaranteed, GetPodCgroupNameSuffix(podUID), "container.scope"),
+			expectedResult: false,
+			expectedUID:    types.UID(""),
+		},
+		{
+			input:          NewCgroupName(qosContainersInfo.Burstable),
+			expectedResult: false,
+			expectedUID:    types.UID(""),
+		},
+		{
+			input:          NewCgroupName(qosContainersInfo.Burstable, GetPodCgroupNameSuffix(podUID)),
+			expectedResult: true,
+			expectedUID:    podUID,
+		},
+		{
+			input:          NewCgroupName(qosContainersInfo.Burstable, GetPodCgroupNameSuffix(podUID), "container.scope"),
+			expectedResult: false,
+			expectedUID:    types.UID(""),
+		},
+		{
+			input:          NewCgroupName(qosContainersInfo.BestEffort),
+			expectedResult: false,
+			expectedUID:    types.UID(""),
+		},
+		{
+			input:          NewCgroupName(qosContainersInfo.BestEffort, GetPodCgroupNameSuffix(podUID)),
+			expectedResult: true,
+			expectedUID:    podUID,
+		},
+		{
+			input:          NewCgroupName(qosContainersInfo.BestEffort, GetPodCgroupNameSuffix(podUID), "container.scope"),
+			expectedResult: false,
+			expectedUID:    types.UID(""),
+		},
+		{
+			input:          NewCgroupName(RootCgroupName, "system"),
+			expectedResult: false,
+			expectedUID:    types.UID(""),
+		},
+		{
+			input:          NewCgroupName(RootCgroupName, "system", "kubelet"),
+			expectedResult: false,
+			expectedUID:    types.UID(""),
+		},
+	}
+	for _, cgroupDriver := range []string{"cgroupfs", "systemd"} {
+		pcm := &podContainerManagerImpl{
+			cgroupManager:     NewCgroupManager(nil, cgroupDriver),
+			enforceCPULimits:  true,
+			qosContainersInfo: qosContainersInfo,
+		}
+		for _, testCase := range testCases {
+			// give the right cgroup structure based on driver
+			cgroupfs := testCase.input.ToCgroupfs()
+			if cgroupDriver == "systemd" {
+				cgroupfs = testCase.input.ToSystemd()
+			}
+			// check if this is a pod or not with the literal cgroupfs input
+			result, resultUID := pcm.IsPodCgroup(cgroupfs)
+			if result != testCase.expectedResult {
+				t.Errorf("Unexpected result for driver: %v, input: %v, expected: %v, actual: %v", cgroupDriver, testCase.input, testCase.expectedResult, result)
+			}
+			if resultUID != testCase.expectedUID {
+				t.Errorf("Unexpected result for driver: %v, input: %v, expected: %v, actual: %v", cgroupDriver, testCase.input, testCase.expectedUID, resultUID)
+			}
+
+		}
+	}
+}

pkg/kubelet/cm/pod_container_manager_stub.go

@@ -49,3 +49,7 @@ func (m *podContainerManagerStub) ReduceCPULimits(_ CgroupName) error {
 func (m *podContainerManagerStub) GetAllPodsFromCgroups() (map[types.UID]CgroupName, error) {
 	return nil, nil
 }
+
+func (m *podContainerManagerStub) IsPodCgroup(cgroupfs string) (bool, types.UID) {
+	return false, types.UID("")
+}

pkg/kubelet/cm/types.go

@@ -124,4 +124,7 @@ type PodContainerManager interface {
 
 	// GetAllPodsFromCgroups enumerates the set of pod uids to their associated cgroup based on state of cgroupfs system.
 	GetAllPodsFromCgroups() (map[types.UID]CgroupName, error)
+
+	// IsPodCgroup returns true if the literal cgroupfs name corresponds to a pod
+	IsPodCgroup(cgroupfs string) (bool, types.UID)
 }

pkg/kubelet/kubelet_getters.go

@@ -174,6 +174,16 @@ func (kl *Kubelet) GetPodByName(namespace, name string) (*v1.Pod, bool) {
 	return kl.podManager.GetPodByName(namespace, name)
 }
 
+// GetPodByCgroupfs provides the pod that maps to the specified cgroup, as well
+// as whether the pod was found.
+func (kl *Kubelet) GetPodByCgroupfs(cgroupfs string) (*v1.Pod, bool) {
+	pcm := kl.containerManager.NewPodContainerManager()
+	if result, podUID := pcm.IsPodCgroup(cgroupfs); result {
+		return kl.podManager.GetPodByUID(podUID)
+	}
+	return nil, false
+}
+
 // GetHostname Returns the hostname as the kubelet sees it.
 func (kl *Kubelet) GetHostname() string {
 	return kl.hostname

pkg/kubelet/server/server.go

@@ -278,7 +278,7 @@ func (s *Server) InstallDefaultHandlers() {
 
 	// cAdvisor metrics are exposed under the secured handler as well
 	r := prometheus.NewRegistry()
-	r.MustRegister(metrics.NewPrometheusCollector(prometheusHostAdapter{s.host}, containerPrometheusLabels))
+	r.MustRegister(metrics.NewPrometheusCollector(prometheusHostAdapter{s.host}, containerPrometheusLabelsFunc(s.host)))
 	s.restfulCont.Handle(cadvisorMetricsPath,
 		promhttp.HandlerFor(r, promhttp.HandlerOpts{ErrorHandling: promhttp.ContinueOnError}),
 	)
@@ -825,31 +825,40 @@ func (a prometheusHostAdapter) GetMachineInfo() (*cadvisorapi.MachineInfo, error
 	return a.host.GetCachedMachineInfo()
 }
 
-// containerPrometheusLabels maps cAdvisor labels to prometheus labels.
-func containerPrometheusLabels(c *cadvisorapi.ContainerInfo) map[string]string {
-	// Prometheus requires that all metrics in the same family have the same labels,
-	// so we arrange to supply blank strings for missing labels
-	var name, image, podName, namespace, containerName string
-	if len(c.Aliases) > 0 {
-		name = c.Aliases[0]
+func containerPrometheusLabelsFunc(s stats.StatsProvider) metrics.ContainerLabelsFunc {
+	// containerPrometheusLabels maps cAdvisor labels to prometheus labels.
+	return func(c *cadvisorapi.ContainerInfo) map[string]string {
+		// Prometheus requires that all metrics in the same family have the same labels,
+		// so we arrange to supply blank strings for missing labels
+		var name, image, podName, namespace, containerName string
+		if len(c.Aliases) > 0 {
+			name = c.Aliases[0]
+		}
+		image = c.Spec.Image
+		if v, ok := c.Spec.Labels[kubelettypes.KubernetesPodNameLabel]; ok {
+			podName = v
+		}
+		if v, ok := c.Spec.Labels[kubelettypes.KubernetesPodNamespaceLabel]; ok {
+			namespace = v
+		}
+		if v, ok := c.Spec.Labels[kubelettypes.KubernetesContainerNameLabel]; ok {
+			containerName = v
+		}
+		// Associate pod cgroup with pod so we have an accurate accounting of sandbox
+		if podName == "" && namespace == "" {
+			if pod, found := s.GetPodByCgroupfs(c.Name); found {
+				podName = pod.Name
+				namespace = pod.Namespace
+			}
+		}
+		set := map[string]string{
+			metrics.LabelID:    c.Name,
+			metrics.LabelName:  name,
+			metrics.LabelImage: image,
+			"pod_name":         podName,
+			"namespace":        namespace,
+			"container_name":   containerName,
+		}
+		return set
 	}
-	image = c.Spec.Image
-	if v, ok := c.Spec.Labels[kubelettypes.KubernetesPodNameLabel]; ok {
-		podName = v
-	}
-	if v, ok := c.Spec.Labels[kubelettypes.KubernetesPodNamespaceLabel]; ok {
-		namespace = v
-	}
-	if v, ok := c.Spec.Labels[kubelettypes.KubernetesContainerNameLabel]; ok {
-		containerName = v
-	}
-	set := map[string]string{
-		metrics.LabelID:    c.Name,
-		metrics.LabelName:  name,
-		metrics.LabelImage: image,
-		"pod_name":         podName,
-		"namespace":        namespace,
-		"container_name":   containerName,
-	}
-	return set
 }

pkg/kubelet/server/server_test.go

@@ -166,10 +166,10 @@ func (fk *fakeKubelet) StreamingConnectionIdleTimeout() time.Duration {
 }
 
 // Unused functions
-func (_ *fakeKubelet) GetNode() (*v1.Node, error)   { return nil, nil }
-func (_ *fakeKubelet) GetNodeConfig() cm.NodeConfig { return cm.NodeConfig{} }
-func (_ *fakeKubelet) GetPodCgroupRoot() string     { return "" }
-
+func (_ *fakeKubelet) GetNode() (*v1.Node, error)                       { return nil, nil }
+func (_ *fakeKubelet) GetNodeConfig() cm.NodeConfig                     { return cm.NodeConfig{} }
+func (_ *fakeKubelet) GetPodCgroupRoot() string                         { return "" }
+func (_ *fakeKubelet) GetPodByCgroupfs(cgroupfs string) (*v1.Pod, bool) { return nil, false }
 func (fk *fakeKubelet) ListVolumesForPod(podUID types.UID) (map[string]volume.Volume, bool) {
 	return map[string]volume.Volume{}, true
 }

pkg/kubelet/server/stats/handler.go

@@ -84,6 +84,10 @@ type StatsProvider interface {
 
 	// GetPodCgroupRoot returns the literal cgroupfs value for the cgroup containing all pods
 	GetPodCgroupRoot() string
+
+	// GetPodByCgroupfs provides the pod that maps to the specified cgroup literal, as well
+	// as whether the pod was found.
+	GetPodByCgroupfs(cgroupfs string) (*v1.Pod, bool)
 }
 
 type handler struct {

pkg/kubelet/server/stats/testing/mock_stats_provider.go

@@ -64,6 +64,12 @@ func (_m *StatsProvider) GetCgroupStats(cgroupName string, updateStats bool) (*v
 	return r0, r1, r2
 }
 
+// GetPodByCgroupfs provides the pod that maps to the specified cgroup, as well
+// as whether the pod was found.
+func (_m *StatsProvider) GetPodByCgroupfs(cgroupfs string) (*corev1.Pod, bool) {
+	return nil, false
+}
+
 // GetContainerInfo provides a mock function with given fields: podFullName, uid, containerName, req
 func (_m *StatsProvider) GetContainerInfo(podFullName string, uid types.UID, containerName string, req *v1.ContainerInfoRequest) (*v1.ContainerInfo, error) {
 	ret := _m.Called(podFullName, uid, containerName, req)