vendor/github.com/google/cadvisor/lib/manager/manager.go - master - githubmirror/kubernetes

vendor/github.com/google/cadvisor/lib/manager/manager.go
1 151 строка36 KB
Damien Grisonnet
kubelet/cadvisor: disable container discovery when using CRI stats
15 июл 2026, 15:51
15 июл 2026, 15:514f01955
// Copyright 2014 Google Inc. All Rights Reserved.
//
// 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.

//go:build linux

// Manager of cAdvisor-monitored containers.
package manager

import (
	"flag"
	"fmt"
	"os"
	"path"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/google/cadvisor/lib/cache/memory"
	"github.com/google/cadvisor/lib/container"
	"github.com/google/cadvisor/lib/container/raw"
	"github.com/google/cadvisor/lib/fs"
	"github.com/google/cadvisor/lib/machine"
	info "github.com/google/cadvisor/lib/model"
	"github.com/google/cadvisor/lib/stats"
	"github.com/google/cadvisor/lib/utils/sysfs"
	"github.com/google/cadvisor/lib/version"
	"github.com/google/cadvisor/lib/watcher"

	"github.com/opencontainers/cgroups"

	"k8s.io/klog/v2"
	"k8s.io/utils/clock"
)

var globalHousekeepingInterval = flag.Duration("global_housekeeping_interval", 1*time.Minute, "Interval between global housekeepings")
var updateMachineInfoInterval = flag.Duration("update_machine_info_interval", 5*time.Minute, "Interval between machine info updates.")

// log_cadvisor_usage is retained as a registered no-op: the per-container usage
// logging path was removed (C6), but the kubelet pins this flag by name in
// cmd/kubelet/app/options/globalflags_linux.go, so it must keep resolving.
var _ = flag.Bool("log_cadvisor_usage", false, "Whether to log the usage of the cAdvisor container")
var eventStorageAgeLimit = flag.String("event_storage_age_limit", "default=24h", "Max length of time for which to store events (per type). Value is a comma separated list of key values, where the keys are event types (e.g.: creation, oom) or \"default\" and the value is a duration. Default is applied to all non-specified event types")
var eventStorageEventLimit = flag.String("event_storage_event_limit", "default=100000", "Max number of events to store (per type). Value is a comma separated list of key values, where the keys are event types (e.g.: creation, oom) or \"default\" and the value is an integer. Default is applied to all non-specified event types")

// application_metrics_count_limit is registered here so the kubelet's global
// flag of the same name resolves at startup. The full binary's collector
// factory reads its value via ApplicationMetricsCountLimit; the kubelet runs no
// collectors, so it does not.
var applicationMetricsCountLimit = flag.Int("application_metrics_count_limit", 100, "Max number of application metrics to store (per container)")

// ApplicationMetricsCountLimit exposes the application_metrics_count_limit flag
// value so the root binary can build its collectors without re-registering the
// flag.
func ApplicationMetricsCountLimit() int { return *applicationMetricsCountLimit }

// The namespace under which aliases are unique.
const (
	DockerNamespace = "docker"
	PodmanNamespace = "podman"
)

var HousekeepingConfigFlags = HousekeepingConfig{
	flag.Duration("max_housekeeping_interval", 60*time.Second, "Largest interval to allow between container housekeepings"),
	flag.Bool("allow_dynamic_housekeeping", true, "Whether to allow the housekeeping interval to be dynamic"),
	false,
}

// The Manager interface defines operations for starting a manager and getting
// container and machine information.
type Manager interface {
	// Start the manager. Calling other manager methods before this returns
	// may produce undefined behavior.
	Start() error

	// Stops the manager.
	Stop() error

	// Get V2 information about a container.
	// Recursive (subcontainer) requests are best-effort, and may return a partial result alongside an
	// error in the partial failure case.
	GetContainerInfoV2(containerName string, options info.RequestOptions) (map[string]info.ContainerInfo, error)

	// Get info for all requested containers based on the request options.
	GetRequestedContainersInfo(containerName string, options info.RequestOptions) (map[string]*info.ContainerInfo, error)

	// Get information about the machine.
	GetMachineInfo() (*info.MachineInfo, error)

	// Get version information about different components we depend on.
	GetVersionInfo() (*info.VersionInfo, error)

	// Get filesystem information for the filesystem that contains the given directory
	GetDirFsInfo(dir string) (info.FsInfo, error)

	// Get filesystem information for a given label.
	// Returns information for all global filesystems if label is empty.
	GetFsInfo(label string) ([]info.FsInfo, error)

	// DebugInfo returns debug information about the manager's tracked containers.
	DebugInfo() map[string][]string

	// The methods below back the full cAdvisor binary's v1/v2 REST API and web
	// UI. They are pure queries over the in-memory container registry and add no
	// dependencies; the kubelet does not call them. See queries.go.

	// Get information about a container.
	GetContainerInfo(containerName string, query *info.ContainerInfoRequest) (*info.ContainerInfo, error)

	// Get information about all subcontainers of the specified container (includes self).
	SubcontainersInfo(containerName string, query *info.ContainerInfoRequest) ([]*info.ContainerInfo, error)

	// Get information about the docker containers (by container name) of an instance.
	AllDockerContainers(query *info.ContainerInfoRequest) (map[string]info.ContainerInfo, error)

	// Get information about the docker container with the specified name.
	DockerContainer(dockerName string, query *info.ContainerInfoRequest) (info.ContainerInfo, error)

	// Get information about the podman containers (by container name) of an instance.
	AllPodmanContainers(query *info.ContainerInfoRequest) (map[string]info.ContainerInfo, error)

	// Get information about the podman container with the specified name.
	PodmanContainer(containerName string, query *info.ContainerInfoRequest) (info.ContainerInfo, error)

	// Get the specs for a container, possibly with subcontainers.
	GetContainerSpec(containerName string, options info.RequestOptions) (map[string]info.ContainerSpec, error)

	// Get filesystem information for the filesystem identified by the given UUID.
	GetFsInfoByFsUUID(uuid string) (info.FsInfo, error)

	// Returns true if the named container exists.
	Exists(containerName string) bool

	// Get the derived (rolling-window percentile) stats for a container.
	GetDerivedStats(containerName string, options info.RequestOptions) (map[string]info.DerivedStats, error)

	// Get the list of processes running in a container.
	GetProcessList(containerName string, options info.RequestOptions) ([]info.ProcessInfo, error)

	// SetEventSink wires the sink that receives container lifecycle events. The
	// full binary injects its events manager; the kubelet does not call this (no
	// events are emitted). See events.go.
	SetEventSink(sink EventSink)
}

// Housekeeping configuration for the manager
type HousekeepingConfig = struct {
	Interval     *time.Duration
	AllowDynamic *bool
	// When set to true, the manager will not discover or watch for new
	// containers. Only the root container ("/") is created at startup;
	// other cgroup paths are created on demand when first queried.
	DisableContainerDiscovery bool
}

// New takes a memory storage and returns a new manager.
func New(memoryCache *memory.InMemoryCache, sysfs sysfs.SysFs, HousekeepingConfig HousekeepingConfig, includedMetricsSet container.MetricSet, rawContainerCgroupPathPrefixWhiteList, containerEnvMetadataWhiteList []string, perfEventsFile string, resctrlInterval time.Duration) (Manager, error) {
	if memoryCache == nil {
		return nil, fmt.Errorf("manager requires memory storage")
	}

	// Detect the container we are running on.
	selfContainer := "/"
	var err error
	// Avoid using GetOwnCgroupPath on cgroup v2 as it is not supported by libcontainer
	if !cgroups.IsCgroup2UnifiedMode() {
		selfContainer, err = cgroups.GetOwnCgroup("cpu")
		if err != nil {
			return nil, err
		}
		klog.V(2).Infof("cAdvisor running in container: %q", selfContainer)
	}

	context := fs.Context{}

	if err := container.InitializeFSContext(&context); err != nil {
		return nil, err
	}

	fsInfo, err := fs.NewFsInfo(context)
	if err != nil {
		return nil, err
	}

	// If cAdvisor was started with host's rootfs mounted, assume that its running
	// in its own namespaces.
	inHostNamespace := false
	if _, err := os.Stat("/rootfs/proc"); os.IsNotExist(err) {
		inHostNamespace = true
	}

	// Register for new subcontainers.
	eventsChannel := make(chan watcher.ContainerEvent, 16)

	newManager := &manager{
		quitChannels:                          make([]chan error, 0, 2),
		memoryCache:                           memoryCache,
		fsInfo:                                fsInfo,
		sysFs:                                 sysfs,
		cadvisorContainer:                     selfContainer,
		inHostNamespace:                       inHostNamespace,
		startupTime:                           time.Now(),
		maxHousekeepingInterval:               *HousekeepingConfig.Interval,
		allowDynamicHousekeeping:              *HousekeepingConfig.AllowDynamic,
		disableContainerDiscovery:             HousekeepingConfig.DisableContainerDiscovery,
		includedMetrics:                       includedMetricsSet,
		containerWatchers:                     []watcher.ContainerWatcher{},
		eventsChannel:                         eventsChannel,
		rawContainerCgroupPathPrefixWhiteList: rawContainerCgroupPathPrefixWhiteList,
		containerEnvMetadataWhiteList:         containerEnvMetadataWhiteList,
	}

	machineInfo, err := machine.Info(sysfs, fsInfo, inHostNamespace)
	if err != nil {
		return nil, err
	}
	newManager.machineInfo = *machineInfo
	klog.V(1).Infof("Machine: %+v", newManager.machineInfo)

	// Collector managers default to Noop; the full cAdvisor binary registers
	// real perf/resctrl factories via plugins.go. The kubelet leaves them nil.
	newManager.perfManager = &stats.NoopManager{}
	if PerfManagerFactory != nil {
		newManager.perfManager, err = PerfManagerFactory(perfEventsFile, machineInfo.Topology)
		if err != nil {
			return nil, err
		}
	}
	newManager.resctrlManager = &stats.NoopResctrlManager{}
	if ResctrlManagerFactory != nil {
		if rm, rerr := ResctrlManagerFactory(resctrlInterval, machineInfo.CPUVendorID, inHostNamespace); rerr != nil {
			klog.V(4).Infof("Cannot gather resctrl metrics: %v", rerr)
		} else {
			newManager.resctrlManager = rm
		}
	}

	versionInfo, err := getVersionInfo()
	if err != nil {
		return nil, err
	}
	klog.V(1).Infof("Version: %+v", *versionInfo)

	return newManager, nil
}

// A namespaced container name.
type namespacedContainerName struct {
	// The namespace of the container. Can be empty for the root namespace.
	Namespace string

	// The name of the container in this namespace.
	Name string
}

// containerMap is a type-safe wrapper around sync.Map for storing containerData
// keyed by namespacedContainerName.
type containerMap struct {
	m sync.Map
}

// Load returns the containerData for the given name, or nil if not found.
func (c *containerMap) Load(name namespacedContainerName) (*containerData, bool) {
	v, ok := c.m.Load(name)
	if !ok {
		return nil, false
	}
	return v.(*containerData), true
}

// Store stores the containerData for the given name.
func (c *containerMap) Store(name namespacedContainerName, data *containerData) {
	c.m.Store(name, data)
}

// Delete removes the containerData for the given name.
func (c *containerMap) Delete(name namespacedContainerName) {
	c.m.Delete(name)
}

// Range calls f for each container in the map. If f returns false, iteration stops.
func (c *containerMap) Range(f func(name namespacedContainerName, data *containerData) bool) {
	c.m.Range(func(key, value any) bool {
		return f(key.(namespacedContainerName), value.(*containerData))
	})
}

type manager struct {
	containers                containerMap
	memoryCache               *memory.InMemoryCache
	fsInfo                    fs.FsInfo
	sysFs                     sysfs.SysFs
	machineMu                 sync.RWMutex // protects machineInfo
	machineInfo               info.MachineInfo
	quitChannels              []chan error
	cadvisorContainer         string
	inHostNamespace           bool
	startupTime               time.Time
	maxHousekeepingInterval   time.Duration
	allowDynamicHousekeeping  bool
	disableContainerDiscovery bool
	includedMetrics           container.MetricSet
	containerWatchers         []watcher.ContainerWatcher
	eventsChannel             chan watcher.ContainerEvent
	// List of raw container cgroup path prefix whitelist.
	rawContainerCgroupPathPrefixWhiteList []string
	// List of container env prefix whitelist, the matched container envs would be collected into metrics as extra labels.
	containerEnvMetadataWhiteList []string

	// Collector managers for perf_event / resctrl. Default to Noop; the full
	// cAdvisor binary injects real implementations via plugins.go.
	perfManager    stats.Manager
	resctrlManager stats.ResctrlManager

	// eventSink, if set by the full binary via SetEventSink, receives container
	// lifecycle and OOM events. nil for the kubelet (no event machinery). See
	// events.go.
	eventSink EventSink
}

// Start the container manager.
func (m *manager) Start() error {
	if !m.disableContainerDiscovery {
		m.containerWatchers = container.InitializePlugins(m, m.fsInfo, m.includedMetrics)
	}

	err := raw.Register(m, m.fsInfo, m.includedMetrics, m.rawContainerCgroupPathPrefixWhiteList)
	if err != nil {
		klog.Errorf("Registration of the raw container factory failed: %v", err)
	}

	if !m.disableContainerDiscovery {
		rawWatcher, err := raw.NewRawContainerWatcher(m.includedMetrics)
		if err != nil {
			return err
		}
		m.containerWatchers = append(m.containerWatchers, rawWatcher)
	}

	// If there are no factories, don't start any housekeeping and serve the information we do have.
	if !container.HasFactories() {
		return nil
	}

	// Create root and then recover all containers.
	err = m.createContainer("/", watcher.Raw)
	if err != nil {
		return err
	}

	if !m.disableContainerDiscovery {
		klog.V(2).Infof("Starting recovery of all containers")
		err = m.detectSubcontainers("/")
		if err != nil {
			return err
		}
		klog.V(2).Infof("Recovery completed")

		// Watch for new container.
		quitWatcher := make(chan error)
		err = m.watchForNewContainers(quitWatcher)
		if err != nil {
			return err
		}
		m.quitChannels = append(m.quitChannels, quitWatcher)

		// Look for new containers in the main housekeeping thread.
		quitGlobalHousekeeping := make(chan error)
		m.quitChannels = append(m.quitChannels, quitGlobalHousekeeping)
		go m.globalHousekeeping(quitGlobalHousekeeping)
	} else {
		klog.V(2).Infof("Container discovery disabled; only root container will be tracked")
	}

	quitUpdateMachineInfo := make(chan error)
	m.quitChannels = append(m.quitChannels, quitUpdateMachineInfo)
	go m.updateMachineInfo(quitUpdateMachineInfo)

	return nil
}

func (m *manager) Stop() error {
	defer m.destroyCollectors()
	// Stop and wait on all quit channels.
	for i, c := range m.quitChannels {
		// Send the exit signal and wait on the thread to exit (by closing the channel).
		c <- nil
		err := <-c
		if err != nil {
			// Remove the channels that quit successfully.
			m.quitChannels = m.quitChannels[i:]
			return err
		}
	}
	m.quitChannels = make([]chan error, 0, 2)
	return nil
}

func (m *manager) destroyCollectors() {
	m.containers.Range(func(_ namespacedContainerName, container *containerData) bool {
		if container == nil {
			return true
		}
		container.perfCollector.Destroy()
		container.resctrlCollector.Destroy()
		return true
	})
}

func (m *manager) updateMachineInfo(quit chan error) {
	ticker := time.NewTicker(*updateMachineInfoInterval)
	for {
		select {
		case <-ticker.C:
			info, err := machine.Info(m.sysFs, m.fsInfo, m.inHostNamespace)
			if err != nil {
				klog.Errorf("Could not get machine info: %v", err)
				break
			}
			m.machineMu.Lock()
			m.machineInfo = *info
			m.machineMu.Unlock()
			klog.V(5).Infof("Update machine info: %+v", *info)
		case <-quit:
			ticker.Stop()
			quit <- nil
			return
		}
	}
}

func (m *manager) globalHousekeeping(quit chan error) {
	// Long housekeeping is either 100ms or half of the housekeeping interval.
	longHousekeeping := 100 * time.Millisecond
	if *globalHousekeepingInterval/2 < longHousekeeping {
		longHousekeeping = *globalHousekeepingInterval / 2
	}

	ticker := time.NewTicker(*globalHousekeepingInterval)
	for {
		select {
		case t := <-ticker.C:
			start := time.Now()

			// Check for new containers.
			err := m.detectSubcontainers("/")
			if err != nil {
				klog.Errorf("Failed to detect containers: %s", err)
			}

			// Log if housekeeping took too long.
			duration := time.Since(start)
			if duration >= longHousekeeping {
				klog.V(3).Infof("Global Housekeeping(%d) took %s", t.Unix(), duration)
			}
		case <-quit:
			// Quit if asked to do so.
			quit <- nil
			klog.Infof("Exiting global housekeeping thread")
			return
		}
	}
}

func (m *manager) getAdjustedSpec(cinfo *containerInfo) info.ContainerSpec {
	spec := cinfo.Spec

	// Set default value to an actual value
	if spec.HasMemory {
		// Memory.Limit is 0 means there's no limit
		if spec.Memory.Limit == 0 {
			m.machineMu.RLock()
			spec.Memory.Limit = uint64(m.machineInfo.MemoryCapacity)
			m.machineMu.RUnlock()
		}
	}
	return spec
}

func (m *manager) GetContainerInfoV2(containerName string, options info.RequestOptions) (map[string]info.ContainerInfo, error) {
	containers, err := m.getRequestedContainers(containerName, options)
	if err != nil {
		return nil, err
	}

	var errs partialFailure
	var nilTime time.Time // Ignored.

	infos := make(map[string]info.ContainerInfo, len(containers))
	for name, container := range containers {
		result := info.ContainerInfo{}
		cinfo, err := container.GetInfo(false)
		if err != nil {
			errs.append(name, "GetInfo", err)
			infos[name] = result
			continue
		}
		result.Spec = m.getAdjustedSpec(cinfo)
		result.ContainerReference = cinfo.ContainerReference

		stats, err := m.memoryCache.RecentStats(name, nilTime, nilTime, options.Count)
		if err != nil {
			errs.append(name, "RecentStats", err)
			infos[name] = result
			continue
		}

		statsOut := make([]*info.ContainerStats, len(stats))
		var lastStat *info.ContainerStats
		for i, s := range stats {
			cp := *s
			if cinfo.Spec.HasCpu {
				if ci, err := info.InstCpuStats(lastStat, s); err == nil {
					cp.CpuInst = ci
				}
			}
			statsOut[i] = &cp
			lastStat = s
		}
		result.Stats = statsOut
		infos[name] = result
	}

	return infos, 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
}

func (m *manager) getContainer(containerName string) (*containerData, error) {
	cont, ok := m.containers.Load(namespacedContainerName{Name: containerName})
	if ok {
		return cont, nil
	}

	if m.disableContainerDiscovery {
		if err := m.createContainer(containerName, watcher.Raw); err != nil {
			return nil, fmt.Errorf("on-demand creation of container %q failed: %v", containerName, err)
		}
		cont, ok = m.containers.Load(namespacedContainerName{Name: containerName})
		if ok {
			return cont, nil
		}
	}

	return nil, fmt.Errorf("unknown container %q", containerName)
}

func (m *manager) getSubcontainers(containerName string) map[string]*containerData {
	matchedName := path.Join(containerName, "/")
	containersMap := make(map[string]*containerData)

	// Get all the unique subcontainers of the specified container
	m.containers.Range(func(_ namespacedContainerName, cont *containerData) bool {
		if cont == nil {
			return true
		}
		name := cont.info.Name
		if name == containerName || strings.HasPrefix(name, matchedName) {
			containersMap[name] = cont
		}
		return true
	})
	return containersMap
}

func (m *manager) getAllNamespacedContainers(ns string) map[string]*containerData {
	containers := make(map[string]*containerData)

	// Get containers in a namespace.
	m.containers.Range(func(name namespacedContainerName, cont *containerData) bool {
		if cont == nil {
			return true
		}
		if name.Namespace == ns {
			containers[cont.info.Name] = cont
		}
		return true
	})
	return containers
}

func (m *manager) namespacedContainer(containerName string, ns string) (*containerData, error) {
	// Check for the container in the namespace.
	if cont, ok := m.containers.Load(namespacedContainerName{Namespace: ns, Name: containerName}); ok {
		return cont, nil
	}

	// Look for container by short prefix name if no exact match found.
	var cont *containerData
	var err error
	m.containers.Range(func(name namespacedContainerName, c *containerData) bool {
		if name.Namespace == ns && strings.HasPrefix(name.Name, containerName) {
			if cont == nil {
				cont = c
			} else {
				err = fmt.Errorf("unable to find container in %q namespace. Container %q is not unique", ns, containerName)
				return false // stop iteration
			}
		}
		return true
	})

	if err != nil {
		return nil, err
	}

	if cont == nil {
		return nil, fmt.Errorf("unable to find container %q in %q namespace", containerName, ns)
	}

	return cont, nil
}

func (m *manager) GetRequestedContainersInfo(containerName string, options info.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 {
			if err == memory.ErrDataNotFound {
				klog.V(4).Infof("Error getting data for container %s because of race condition", name)
				continue
			}
			errs.append(name, "containerDataToContainerInfo", err)
		}
		containersMap[name] = info
	}
	return containersMap, errs.OrNil()
}

func (m *manager) getRequestedContainers(containerName string, options info.RequestOptions) (map[string]*containerData, error) {
	containersMap := make(map[string]*containerData)
	switch options.IdType {
	case info.TypeName:
		if !options.Recursive {
			cont, err := m.getContainer(containerName)
			if err != nil {
				return containersMap, err
			}
			containersMap[cont.info.Name] = cont
		} else {
			containersMap = m.getSubcontainers(containerName)
			if len(containersMap) == 0 {
				return containersMap, fmt.Errorf("unknown container: %q", containerName)
			}
		}
	case info.TypeDocker, info.TypePodman:
		namespace := map[string]string{
			info.TypeDocker: DockerNamespace,
			info.TypePodman: PodmanNamespace,
		}[options.IdType]
		if !options.Recursive {
			containerName = strings.TrimPrefix(containerName, "/")
			cont, err := m.namespacedContainer(containerName, namespace)
			if err != nil {
				return containersMap, err
			}
			containersMap[cont.info.Name] = cont
		} else {
			if containerName != "/" {
				return containersMap, fmt.Errorf("invalid request for %s container %q with subcontainers", options.IdType, containerName)
			}
			containersMap = m.getAllNamespacedContainers(namespace)
		}
	default:
		return containersMap, fmt.Errorf("invalid request type %q", options.IdType)
	}
	if options.MaxAge != nil {
		// update stats for all containers in containersMap
		var waitGroup sync.WaitGroup
		waitGroup.Add(len(containersMap))
		for _, container := range containersMap {
			go func(cont *containerData) {
				cont.OnDemandHousekeeping(*options.MaxAge)
				waitGroup.Done()
			}(container)
		}
		waitGroup.Wait()
	}
	return containersMap, nil
}

func (m *manager) GetDirFsInfo(dir string) (info.FsInfo, error) {
	device, err := m.fsInfo.GetDirFsDevice(dir)
	if err != nil {
		return info.FsInfo{}, fmt.Errorf("failed to get device for dir %q: %v", dir, err)
	}
	return m.getFsInfoByDeviceName(device.Device)
}

func (m *manager) GetFsInfo(label string) ([]info.FsInfo, error) {
	var empty time.Time
	// Get latest data from filesystems hanging off root container.
	stats, err := m.memoryCache.RecentStats("/", empty, empty, 1)
	if err != nil {
		return nil, err
	}
	dev := ""
	if len(label) != 0 {
		dev, err = m.fsInfo.GetDeviceForLabel(label)
		if err != nil {
			return nil, err
		}
	}
	fsInfo := []info.FsInfo{}
	for i := range stats[0].Filesystem {
		fs := stats[0].Filesystem[i]
		if len(label) != 0 && fs.Device != dev {
			continue
		}
		mountpoint, err := m.fsInfo.GetMountpointForDevice(fs.Device)
		if err != nil {
			return nil, err
		}
		labels, err := m.fsInfo.GetLabelsForDevice(fs.Device)
		if err != nil {
			return nil, err
		}

		fi := info.FsInfo{
			Timestamp:  stats[0].Timestamp,
			Device:     fs.Device,
			Mountpoint: mountpoint,
			Capacity:   fs.Limit,
			Usage:      fs.Usage,
			Available:  fs.Available,
			Labels:     labels,
		}
		if fs.HasInodes {
			fi.Inodes = &fs.Inodes
			fi.InodesFree = &fs.InodesFree
		}
		fsInfo = append(fsInfo, fi)
	}
	return fsInfo, nil
}

func (m *manager) GetMachineInfo() (*info.MachineInfo, error) {
	m.machineMu.RLock()
	defer m.machineMu.RUnlock()
	return m.machineInfo.Clone(), nil
}

func (m *manager) DebugInfo() map[string][]string {
	debugInfo := container.DebugInfo()

	// Get unique containers.
	conts := make(map[*containerData]struct{})
	m.containers.Range(func(_ namespacedContainerName, cont *containerData) bool {
		if cont != nil {
			conts[cont] = struct{}{}
		}
		return true
	})

	// List containers.
	lines := make([]string, 0, len(conts))
	for cont := range conts {
		lines = append(lines, cont.info.Name)
		if cont.info.Namespace != "" {
			lines = append(lines, fmt.Sprintf("\tNamespace: %s", cont.info.Namespace))
		}

		if len(cont.info.Aliases) != 0 {
			lines = append(lines, "\tAliases:")
			for _, alias := range cont.info.Aliases {
				lines = append(lines, fmt.Sprintf("\t\t%s", alias))
			}
		}
	}

	debugInfo["Managed containers"] = lines
	return debugInfo
}

func (m *manager) GetVersionInfo() (*info.VersionInfo, error) {
	// TODO: Consider caching this and periodically updating.  The VersionInfo may change if
	// the docker daemon is started after the cAdvisor client is created.  Caching the value
	// would be helpful so we would be able to return the last known docker version if
	// docker was down at the time of a query.
	return getVersionInfo()
}

// Create a container.
func (m *manager) createContainer(containerName string, watchSource watcher.ContainerWatchSource) error {
	namespacedName := namespacedContainerName{
		Name: containerName,
	}

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

	handler, accept, err := container.NewContainerHandler(containerName, watchSource, m.containerEnvMetadataWhiteList, m.inHostNamespace)
	if err != nil {
		return err
	}
	if !accept {
		// ignoring this container.
		klog.V(4).Infof("ignoring container %q", containerName)
		return nil
	}
	cont, err := newContainerData(containerName, m.memoryCache, handler, m.maxHousekeepingInterval, m.allowDynamicHousekeeping, clock.RealClock{})
	if err != nil {
		return err
	}

	if m.includedMetrics.Has(container.PerfMetrics) {
		perfCgroupPath, err := handler.GetCgroupPath("perf_event")
		if err != nil {
			klog.Warningf("Error getting perf_event cgroup path: %q", err)
		} else {
			cont.perfCollector, err = m.perfManager.GetCollector(perfCgroupPath)
			if err != nil {
				klog.Errorf("Perf event metrics will not be available for container %q: %v", containerName, err)
			}
		}
	}
	if m.includedMetrics.Has(container.ResctrlMetrics) {
		m.machineMu.RLock()
		noOfNUMA := len(m.machineInfo.Topology)
		m.machineMu.RUnlock()
		cont.resctrlCollector, err = m.resctrlManager.GetCollector(containerName, func() ([]string, error) {
			pids, perr := handler.ListProcesses(container.ListSelf)
			if perr != nil {
				return nil, perr
			}
			ss := make([]string, len(pids))
			for i, p := range pids {
				ss[i] = strconv.Itoa(p)
			}
			return ss, nil
		}, noOfNUMA)
		if err != nil {
			klog.V(4).Infof("resctrl metrics will not be available for container %s: %s", cont.info.Name, err)
		}
	}

	// Application-metrics collectors (binary-only; the kubelet leaves the factory
	// nil). The readFile closure lets the collector factory read config files
	// (declared via container labels) from inside the container.
	if CollectorManagerFactory != nil {
		cm, cerr := CollectorManagerFactory(handler, func(p string) ([]byte, error) {
			return cont.ReadFile(p, m.inHostNamespace)
		})
		if cerr != nil {
			klog.V(4).Infof("Failed to set up application-metrics collectors for %q: %v", containerName, cerr)
		} else if cm != nil {
			cont.collectorManager = cm
			// Refresh the spec so custom-metric definitions appear in it.
			if specErr := cont.updateSpec(); specErr != nil {
				klog.V(4).Infof("Failed to refresh spec after wiring collectors for %q: %v", containerName, specErr)
			}
		}
	}

	// Add the container name and all its aliases. The aliases must be within the namespace of the factory.
	m.containers.Store(namespacedName, cont)
	for _, alias := range cont.info.Aliases {
		m.containers.Store(namespacedContainerName{
			Namespace: cont.info.Namespace,
			Name:      alias,
		}, cont)
	}

	klog.V(3).Infof("Added container: %q (aliases: %v, namespace: %q)", containerName, cont.info.Aliases, cont.info.Namespace)

	// Surface the container-creation event to the sink (no-op if unset).
	if m.eventSink != nil {
		var creationTime time.Time
		if spec, specErr := cont.handler.GetSpec(); specErr == nil {
			creationTime = spec.CreationTime
		}
		m.addEvent(&info.Event{
			ContainerName: cont.info.Name,
			Timestamp:     creationTime,
			EventType:     info.EventContainerCreation,
		})
	}

	// Start the container's housekeeping.
	return cont.Start()
}

func (m *manager) destroyContainer(containerName string) error {
	namespacedName := namespacedContainerName{
		Name: containerName,
	}
	cont, ok := m.containers.Load(namespacedName)
	if !ok {
		// Already destroyed, done.
		return nil
	}

	exitCode, err := cont.handler.GetExitCode()
	if err != nil {
		klog.V(4).Infof("Could not retrieve exit code for container %q: %v (using -1)", containerName, err)
		exitCode = -1
	}

	err = cont.Stop()
	if err != nil {
		return err
	}

	// Remove the container from our records (and all its aliases).
	m.containers.Delete(namespacedName)
	for _, alias := range cont.info.Aliases {
		m.containers.Delete(namespacedContainerName{
			Namespace: cont.info.Namespace,
			Name:      alias,
		})
	}
	klog.V(3).Infof("Destroyed container: %q (aliases: %v, namespace: %q, exit_code: %d)", containerName, cont.info.Aliases, cont.info.Namespace, exitCode)

	// Surface the container-deletion event to the sink (no-op if unset),
	// carrying the exit code (the /events API and its tests rely on it).
	m.addEvent(&info.Event{
		ContainerName: cont.info.Name,
		Timestamp:     time.Now(),
		EventType:     info.EventContainerDeletion,
		EventData: info.EventData{
			ContainerDeletion: &info.ContainerDeletionEventData{
				ExitCode: exitCode,
			},
		},
	})

	return nil
}

// Detect all containers that have been added or deleted from the specified container.
func (m *manager) getContainersDiff(containerName string) (added []info.ContainerReference, removed []info.ContainerReference, err error) {
	// Get all subcontainers recursively.
	cont, ok := m.containers.Load(namespacedContainerName{Name: containerName})
	if !ok {
		return nil, nil, fmt.Errorf("failed to find container %q while checking for new containers", containerName)
	}
	allContainers, err := cont.handler.ListContainers(container.ListRecursive)

	if err != nil {
		return nil, nil, err
	}
	allContainers = append(allContainers, info.ContainerReference{Name: containerName})

	// Determine which were added and which were removed.
	allContainersSet := make(map[string]*containerData)
	m.containers.Range(func(name namespacedContainerName, cont *containerData) bool {
		if cont == nil {
			return true
		}
		// Only add the canonical name.
		if cont.info.Name == name.Name {
			allContainersSet[name.Name] = cont
		}
		return true
	})

	// Added containers
	for _, c := range allContainers {
		delete(allContainersSet, c.Name)
		_, ok := m.containers.Load(namespacedContainerName{Name: c.Name})
		if !ok {
			added = append(added, c)
		}
	}

	// Removed ones are no longer in the container listing.
	for _, d := range allContainersSet {
		removed = append(removed, d.info.ContainerReference)
	}

	return
}

// Detect the existing subcontainers and reflect the setup here.
func (m *manager) detectSubcontainers(containerName string) error {
	added, removed, err := m.getContainersDiff(containerName)
	if err != nil {
		return err
	}

	// Add the new containers.
	for _, cont := range added {
		err = m.createContainer(cont.Name, watcher.Raw)
		if err != nil {
			klog.Errorf("Failed to create existing container: %s: %s", cont.Name, err)
		}
	}

	// Remove the old containers.
	for _, cont := range removed {
		err = m.destroyContainer(cont.Name)
		if err != nil {
			klog.Errorf("Failed to destroy existing container: %s: %s", cont.Name, err)
		}
	}

	return nil
}

// Watches for new containers started in the system. Runs forever unless there is a setup error.
func (m *manager) watchForNewContainers(quit chan error) error {
	watched := make([]watcher.ContainerWatcher, 0)
	for _, watcher := range m.containerWatchers {
		err := watcher.Start(m.eventsChannel)
		if err != nil {
			for _, w := range watched {
				stopErr := w.Stop()
				if stopErr != nil {
					klog.Warningf("Failed to stop wacher %v with error: %v", w, stopErr)
				}
			}
			return err
		}
		watched = append(watched, watcher)
	}

	// There is a race between starting the watch and new container creation so we do a detection before we read new containers.
	err := m.detectSubcontainers("/")
	if err != nil {
		return err
	}

	// Listen to events from the container handler.
	go func() {
		for {
			select {
			case event := <-m.eventsChannel:
				switch {
				case event.EventType == watcher.ContainerAdd:
					switch event.WatchSource {
					default:
						err = m.createContainer(event.Name, event.WatchSource)
					}
				case event.EventType == watcher.ContainerDelete:
					err = m.destroyContainer(event.Name)
				}
				if err != nil {
					klog.Warningf("Failed to process watch event %+v: %v", event, err)
				}
			case <-quit:
				var errs partialFailure

				// Stop processing events if asked to quit.
				for i, watcher := range m.containerWatchers {
					err := watcher.Stop()
					if err != nil {
						errs.append(fmt.Sprintf("watcher %d", i), "Stop", err)
					}
				}

				if len(errs) > 0 {
					quit <- errs
				} else {
					quit <- nil
					klog.Infof("Exiting thread watching subcontainers")
					return
				}
			}
		}
	}()
	return nil
}

func (m *manager) getFsInfoByDeviceName(deviceName string) (info.FsInfo, error) {
	mountPoint, err := m.fsInfo.GetMountpointForDevice(deviceName)
	if err != nil {
		return info.FsInfo{}, fmt.Errorf("failed to get mount point for device %q: %v", deviceName, err)
	}
	infos, err := m.GetFsInfo("")
	if err != nil {
		return info.FsInfo{}, err
	}
	for _, info := range infos {
		if info.Mountpoint == mountPoint {
			return info, nil
		}
	}
	return info.FsInfo{}, fmt.Errorf("cannot find filesystem info for device %q", deviceName)
}

func getVersionInfo() (*info.VersionInfo, error) {

	kernelVersion := machine.KernelVersion()
	osVersion := machine.ContainerOsVersion()

	return &info.VersionInfo{
		KernelVersion:      kernelVersion,
		ContainerOsVersion: osVersion,
		CadvisorVersion:    version.Info["version"],
		CadvisorRevision:   version.Info["revision"],
	}, nil
}

// Helper for accumulating partial failures.
type partialFailure []string

func (f *partialFailure) append(id, operation string, err error) {
	*f = append(*f, fmt.Sprintf("[%q: %s: %s]", id, operation, err))
}

func (f partialFailure) Error() string {
	return fmt.Sprintf("partial failures: %s", strings.Join(f, ", "))
}

func (f partialFailure) OrNil() error {
	if len(f) == 0 {
		return nil
	}
	return f
}