Files
Damien Grisonnet 4f01955a87 kubelet/cadvisor: disable container discovery when using CRI stats
When the container runtime provides stats via CRI, cAdvisor only needs
to track the root container for node-level resource and filesystem
metrics. Set DisableContainerDiscovery on the cAdvisor manager to skip
plugin initialization, cgroup watchers, and subcontainer scanning in
that mode.

Signed-off-by: Damien Grisonnet <dgrisonn@redhat.com>
2026-07-15 14:51:05 +02:00

1152 lines
36 KiB
Go

// 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
}