kubelet没有container metric指标排查过程

I0120 04:32:48.926910  660625 factory.go:220] Factory "containerd" was unable to handle container "/kubepods.slice/kubepods-besteffort.slice"
I0120 04:32:48.926917  660625 factory.go:45] /kubepods.slice/kubepods-besteffort.slice not handled by systemd handler
I0120 04:32:48.926919  660625 factory.go:220] Factory "systemd" was unable to handle container "/kubepods.slice/kubepods-besteffort.slice"
I0120 04:32:48.926923  660625 factory.go:220] Factory "docker" was unable to handle container "/kubepods.slice/kubepods-besteffort.slice"
I0120 04:32:48.926927  660625 factory.go:216] Using factory "raw" for container "/kubepods.slice/kubepods-besteffort.slice"
I0120 04:32:48.926989  660625 container.go:527] Start housekeeping for container "/kubepods.slice/kubepods-burstable.slice/kubepods-burstable-pod993e5f25_9257_43f8_80da_db8185e7338c.slice/docker-3328875e5f0215a060e04e9ecd5dc33c88b21713cde00a4cf7899562125d4e48.scope"
I0120 04:32:48.927134  660625 manager.go:987] Added container: "/kubepods.slice/kubepods-besteffort.slice" (aliases: [], namespace: "")
I0120 04:32:48.927293  660625 handler.go:325] Added event &{/kubepods.slice/kubepods-besteffort.slice 2021-12-28 09:48:34.799412935 +0000 UTC containerCreation {<nil>}}
I0120 04:32:48.927341  660625 container.go:527] Start housekeeping for container "/kubepods.slice/kubepods-besteffort.slice"
I0120 04:32:48.943313  660625 factory.go:220] Factory "containerd" was unable to handle container "/system.slice/kubelet.service"
I0120 04:32:48.943318  660625 factory.go:45] /system.slice/kubelet.service not handled by systemd handler
I0120 04:32:48.943320  660625 factory.go:220] Factory "systemd" was unable to handle container "/system.slice/kubelet.service"
I0120 04:32:48.943324  660625 factory.go:220] Factory "docker" was unable to handle container "/system.slice/kubelet.service"
I0120 04:32:48.943329  660625 factory.go:216] Using factory "raw" for container "/system.slice/kubelet.service"
I0120 04:32:48.943639  660625 manager.go:987] Added container: "/system.slice/kubelet.service" (aliases: [], namespace: "")
I0120 04:32:48.943886  660625 handler.go:325] Added event &{/system.slice/kubelet.service 2021-12-28 09:48:34.672413802 +0000 UTC containerCreation {<nil>}}

func (m *manager) createContainerLocked(containerName string, watchSource watcher.ContainerWatchSource) error {
	namespacedName := namespacedContainerName{
		Name: containerName,
	}

	// Check that the container didn't already exist.
	if _, ok := m.containers[namespacedName]; ok {
		return nil
	}

	handler, accept, err := container.NewContainerHandler(containerName, watchSource, m.inHostNamespace)
	if err != nil {
		return err
	}
	if !accept {
		// ignoring this container.
		klog.V(4).Infof("ignoring container %q", containerName)
		return nil
	}
	collectorManager, err := collector.NewCollectorManager()
	if err != nil {
		return err
	}

	logUsage := *logCadvisorUsage && containerName == m.cadvisorContainer
	cont, err := newContainerData(containerName, m.memoryCache, handler, logUsage, collectorManager, m.maxHousekeepingInterval, m.allowDynamicHousekeeping, clock.RealClock{})
	if err != nil {
		return err
	}
	....
	// Start the container's housekeeping.
	return cont.Start()

// Create a new ContainerHandler for the specified container.
func NewContainerHandler(name string, watchType watcher.ContainerWatchSource, inHostNamespace bool) (ContainerHandler, bool, error) {
	factoriesLock.RLock()
	defer factoriesLock.RUnlock()

	// Create the ContainerHandler with the first factory that supports it.
	for _, factory := range factories[watchType] {
		canHandle, canAccept, err := factory.CanHandleAndAccept(name)
		if err != nil {
			klog.V(4).Infof("Error trying to work out if we can handle %s: %v", name, err)
		}
		if canHandle {
			if !canAccept {
				klog.V(3).Infof("Factory %q can handle container %q, but ignoring.", factory, name)
				return nil, false, nil
			}
			klog.V(3).Infof("Using factory %q for container %q", factory, name)
			handle, err := factory.NewContainerHandler(name, inHostNamespace)
			return handle, canAccept, err
		}
		klog.V(4).Infof("Factory %q was unable to handle container %q", factory, name)
	}

	return nil, false, fmt.Errorf("no known factory can handle creation of container")
}

// The raw factory can handle any container. If --docker_only is set to true, non-docker containers are ignored except for "/" and those whitelisted by raw_cgroup_prefix_whitelist flag.
func (f *rawFactory) CanHandleAndAccept(name string) (bool, bool, error) {
	if name == "/" {
		return true, true, nil
	}
	if *dockerOnly && f.rawPrefixWhiteList[0] == "" {
		return true, false, nil
	}
	for _, prefix := range f.rawPrefixWhiteList { //这里的f.rawPrefixWhiteList为[]string{"/kubepods.slice", "/system.slice/kubelet.service", "/system.slice/docker.service"}
		if strings.HasPrefix(name, prefix) {
			return true, true, nil
		}
	}
	return true, false, nil
}

func (cd *containerData) housekeepingTick(timer <-chan time.Time, longHousekeeping time.Duration) bool {
	select {
	case <-cd.stop:
		// Stop housekeeping when signaled.
		return false
	case finishedChan := <-cd.onDemandChan:
		// notify the calling function once housekeeping has completed
		defer close(finishedChan)
	case <-timer:
	}
	start := cd.clock.Now()
	err := cd.updateStats()
	if err != nil {
		if cd.allowErrorLogging() {
			klog.Warningf("Failed to update stats for container \"%s\": %s", cd.info.Name, err)
		}
	}

func (cd *containerData) updateStats() error {
	stats, statsErr := cd.handler.GetStats()
	if statsErr != nil {
		// Ignore errors if the container is dead.
		if !cd.handler.Exists() {
			return nil
		}

		// Stats may be partially populated, push those before we return an error.
		statsErr = fmt.Errorf("%v, continuing to push stats", statsErr)
	}
	if stats == nil {
		return statsErr
	}

func (s *CpusetGroup) GetStats(path string, stats *cgroups.Stats) error {
	var err error

	stats.CPUSetStats.CPUs, err = getCpusetStat(path, "cpuset.cpus")
	if err != nil && !errors.Is(err, os.ErrNotExist) {
		return err
	}
	.....
}

func getCpusetStat(path string, filename string) ([]uint16, error) {
	var extracted []uint16
	fileContent, err := fscommon.GetCgroupParamString(path, filename)
	if err != nil {
		return extracted, err
	}
	if len(fileContent) == 0 {
		return extracted, fmt.Errorf("%s found to be empty", filepath.Join(path, filename))
	}

func (c *PrometheusCollector) collectContainersInfo(ch chan<- prometheus.Metric) {
	containers, err := c.infoProvider.GetRequestedContainersInfo("/", c.opts)
	if err != nil {
		c.errors.Set(1)
		klog.Warningf("Couldn't get containers: %s", err)
		return
	}
	rawLabels := map[string]struct{}{}
	for _, container := range containers {
		for l := range c.containerLabelsFunc(container) {
			rawLabels[l] = struct{}{}
		}
	}
    ....
    // Container spec
		desc := prometheus.NewDesc("container_start_time_seconds", "Start time of the container since unix epoch in seconds.", labels, nil)
		ch <- prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, float64(cont.Spec.CreationTime.Unix()), values...)

		if cont.Spec.HasCpu {
			desc = prometheus.NewDesc("container_spec_cpu_period", "CPU period of the container.", labels, nil)
			ch <- prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, float64(cont.Spec.Cpu.Period), values...)
			if cont.Spec.Cpu.Quota != 0 {
				desc = prometheus.NewDesc("container_spec_cpu_quota", "CPU quota of the container.", labels, nil)
				ch <- prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, float64(cont.Spec.Cpu.Quota), values...)
			}
			desc := prometheus.NewDesc("container_spec_cpu_shares", "CPU share of the container.", labels, nil)
			ch <- prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, float64(cont.Spec.Cpu.Limit), values...)

		}
    ....

func (m *manager) GetRequestedContainersInfo(containerName string, options v2.RequestOptions) (map[string]*info.ContainerInfo, error) {
	containers, err := m.getRequestedContainers(containerName, options)
	if err != nil {
		return nil, err
	}
	var errs partialFailure
	containersMap := make(map[string]*info.ContainerInfo)
	query := info.ContainerInfoRequest{
		NumStats: options.Count,
	}
	for name, data := range containers {
		info, err := m.containerDataToContainerInfo(data, &query)
		if err != nil {
			errs.append(name, "containerDataToContainerInfo", err)
		}
		containersMap[name] = info
	}
	return containersMap, errs.OrNil()
}

func (m *manager) containerDataToContainerInfo(cont *containerData, query *info.ContainerInfoRequest) (*info.ContainerInfo, error) {
	// Get the info from the container.
	cinfo, err := cont.GetInfo(true)
	if err != nil {
		return nil, err
	}

	stats, err := m.memoryCache.RecentStats(cinfo.Name, query.Start, query.End, query.NumStats)
	if err != nil {
		return nil, err
	}

	// Make a copy of the info for the user.
	ret := &info.ContainerInfo{
		ContainerReference: cinfo.ContainerReference,
		Subcontainers:      cinfo.Subcontainers,
		Spec:               m.getAdjustedSpec(cinfo),
		Stats:              stats,
	}
	return ret, nil
}

// ErrDataNotFound is the error resulting if failed to find a container in memory cache.
var ErrDataNotFound = errors.New("unable to find data in memory cache")

func (c *InMemoryCache) RecentStats(name string, start, end time.Time, maxStats int) ([]*info.ContainerStats, error) {
	var cstore *containerCache
	var ok bool
	err := func() error {
		c.lock.RLock()
		defer c.lock.RUnlock()
		if cstore, ok = c.containerCacheMap[name]; !ok {
			return ErrDataNotFound
		}
		return nil
	}()
	if err != nil {
		return nil, err
	}

	return cstore.RecentStats(start, end, maxStats)
}

// 递归的创建dir父目录
// 如果递归路径下面的"cpuset.cpus"和"cpuset.mems"文件的内容为空，则设置为其父目录里对应文件的值
// 创建dir目录
// cgroup里设置了CpusetCpus和CpusetMems值，则写入dir目录下"cpuset.cpus"和"cpuset.mems"文件
// 如果path路径下面的"cpuset.cpus"和"cpuset.mems"文件的内容为空，则设置为父目录里对应文件的值
// 将pid写入（覆盖）到dir下的"cgroup.procs"文件，最多重试5次
func (s *CpusetGroup) ApplyDir(dir string, cgroup *configs.Cgroup, pid int) error {
	// This might happen if we have no cpuset cgroup mounted.
	// Just do nothing and don't fail.
	if dir == "" {
		return nil
	}
	mountInfo, err := ioutil.ReadFile("/proc/self/mountinfo")
	if err != nil {
		return err
	}
	// 从mountinfo中获取最长的匹配dir的挂载路径的父目录
	root := filepath.Dir(cgroups.GetClosestMountpointAncestor(dir, string(mountInfo)))
	// 'ensureParent' start with parent because we don't want to
	// explicitly inherit from parent, it could conflict with
	// 'cpuset.cpu_exclusive'.
	// 递归的创建dir父目录
	// 如果递归路径下面的"cpuset.cpus"和"cpuset.mems"文件的内容为空，则设置为其父目录里对应文件的值
	if err := s.ensureParent(filepath.Dir(dir), root); err != nil {
		return err
	}
	if err := os.MkdirAll(dir, 0755); err != nil {
		return err
	}
	// We didn't inherit cpuset configs from parent, but we have
	// to ensure cpuset configs are set before moving task into the
	// cgroup.
	// The logic is, if user specified cpuset configs, use these
	// specified configs, otherwise, inherit from parent. This makes
	// cpuset configs work correctly with 'cpuset.cpu_exclusive', and
	// keep backward compatibility.
	// cgroup里设置了CpusetCpus和CpusetMems值，则写入dir目录下"cpuset.cpus"和"cpuset.mems"文件
	// 如果path路径下面的"cpuset.cpus"和"cpuset.mems"文件的内容为空，则设置为父目录里对应文件的值
	if err := s.ensureCpusAndMems(dir, cgroup); err != nil {
		return err
	}

	// because we are not using d.join we need to place the pid into the procs file
	// unlike the other subsystems
	// 将pid写入（覆盖）到dir下的"cgroup.procs"文件，最多重试5次
	// 如果dir目录下"cpuset.cpus"和"cpuset.mems"文件为空，则写入dir下的"cgroup.procs"文件会报错"write error: No space left on device"
	return cgroups.WriteCgroupProc(dir, pid)
}

func (s *CpusetGroup) Apply(d *cgroupData) error {
	// 返回cpuset的cgroup绝对路径，比如"/sys/fs/cgroup/cpuset/system.slice/kubelet.service"
	dir, err := d.path("cpuset")
	if err != nil && !cgroups.IsNotFound(err) {
		return err
	}
	return s.ApplyDir(dir, d.config, d.pid)
}

// 如果manager不为空，则会将kubelet进程移动到manager.Cgroups.Name为路径的cgroup，否则跳过
// 然后设置kubelet的oom_score_adj为-999
// 如果manager不为空，则会将dockerd进程移动到manager.Cgroups.Name为路径的cgroup，否则跳过
// 然后设置dockerd的oom_score_adj为-999
func ensureProcessInContainerWithOOMScore(pid int, oomScoreAdj int, manager *fs.Manager) error {
	if runningInHost, err := isProcessRunningInHost(pid); err != nil {
		// Err on the side of caution. Avoid moving the docker daemon unless we are able to identify its context.
		return err
	} else if !runningInHost {
		// Process is running inside a container. Don't touch that.
		klog.V(2).Infof("pid %d is not running in the host namespaces", pid)
		return nil
	}

	var errs []error
	if manager != nil {
		// 获得kubelt进程的cgroup路径，比如/system.slice/kubelet.service
		cont, err := getContainer(pid)
		if err != nil {
			errs = append(errs, fmt.Errorf("failed to find container of PID %d: %v", pid, err))
		}

		// kubelet进程的cgroup路径与manager.Cgroups.Name（ 即cm.KubeletCgroupsName，配置了--kubelet-cgroups或配置文件KubeletCgroups）不一致，则创建指定的cgroup路径，并设置各个子系统的属性值
		if cont != manager.Cgroups.Name {
			// 创建各个cgroup系统目录，并将pid加入到这个cgroup中
			// 其中cpuset、cpu、memory有特殊处理，具体看vendor\github.com\opencontainers\runc\libcontainer\cgroups\fs\apply_raw.go
			err = manager.Apply(pid)
			if err != nil {
				errs = append(errs, fmt.Errorf("failed to move PID %d (in %q) to %q: %v", pid, cont, manager.Cgroups.Name, err))
			}
		}
	}

	// Also apply oom-score-adj to processes
	oomAdjuster := oom.NewOOMAdjuster()
	klog.V(5).Infof("attempting to apply oom_score_adj of %d to pid %d", oomScoreAdj, pid)
	// 将oomScoreAdj值写入到/proc/<pid>/oom_score_adj
	if err := oomAdjuster.ApplyOOMScoreAdj(pid, oomScoreAdj); err != nil {
		klog.V(3).Infof("Failed to apply oom_score_adj %d for pid %d: %v", oomScoreAdj, pid, err)
		errs = append(errs, fmt.Errorf("failed to apply oom score %d to PID %d: %v", oomScoreAdj, pid, err))
	}
	return utilerrors.NewAggregate(errs)
}