2.상세분석

// syncLoop is the main loop for processing changes. It watches for changes from
// three channels (file, apiserver, and http) and creates a union of them. For
// any new change seen, will run a sync against desired state and running state. If
// no changes are seen to the configuration, will synchronize the last known desired
// state every sync-frequency seconds. Never returns.
func (kl *Kubelet) syncLoop(ctx context.Context, updates <-chan kubetypes.PodUpdate, handler SyncHandler) {
    klog.InfoS("Starting kubelet main sync loop")
    ...

    for {
        if err := kl.runtimeState.runtimeErrors(); err != nil {
            klog.ErrorS(err, "Skipping pod synchronization")
            // exponential backoff
            time.Sleep(duration)
            duration = time.Duration(math.Min(float64(max), factor*float64(duration)))
            continue
        }
        // reset backoff if we have a success
        duration = base

        kl.syncLoopMonitor.Store(kl.clock.Now())
        if !kl.syncLoopIteration(ctx, updates, handler, syncTicker.C, housekeepingTicker.C, plegCh) {
            break
        }
        kl.syncLoopMonitor.Store(kl.clock.Now())
    }
}

func (kl *Kubelet) syncLoopIteration(ctx context.Context, configCh <-chan kubetypes.PodUpdate, handler SyncHandler,
    syncCh <-chan time.Time, housekeepingCh <-chan time.Time, plegCh <-chan *pleg.PodLifecycleEvent) bool {
    select {
    case u, open := <-configCh:
        // Update from a config source; dispatch it to the right handler
        // callback.
        if !open {
            klog.ErrorS(nil, "Update channel is closed, exiting the sync loop")
            return false
        }

        switch u.Op {
        case kubetypes.ADD:
            klog.V(2).InfoS("SyncLoop ADD", "source", u.Source, "pods", klog.KObjSlice(u.Pods))
            // After restarting, kubelet will get all existing pods through
            // ADD as if they are new pods. These pods will then go through the
            // admission process and *may* be rejected. This can be resolved
            // once we have checkpointing.
            handler.HandlePodAdditions(u.Pods)
        ...
    return true
}

func (kl *Kubelet) HandlePodAdditions(pods []*v1.Pod) {
    ...
    for _, pod := range pods {
        existingPods := kl.podManager.GetPods()
        // Always add the pod to the pod manager. Kubelet relies on the pod
        // manager as the source of truth for the desired state. If a pod does
        // not exist in the pod manager, it means that it has been deleted in
        // the apiserver and no action (other than cleanup) is required.
        kl.podManager.AddPod(pod)

        pod, mirrorPod, wasMirror := kl.podManager.GetPodAndMirrorPod(pod)
        if wasMirror {
            if pod == nil {
                klog.V(2).InfoS("Unable to find pod for mirror pod, skipping", "mirrorPod", klog.KObj(mirrorPod), "mirrorPodUID", mirrorPod.UID)
                continue
            }
            kl.podWorkers.UpdatePod(UpdatePodOptions{
                Pod:        pod,
                MirrorPod:  mirrorPod,
                UpdateType: kubetypes.SyncPodUpdate,
                StartTime:  start,
            })
            continue
        }

        // Only go through the admission process if the pod is not requested
        // for termination by another part of the kubelet. If the pod is already
        // using resources (previously admitted), the pod worker is going to be
        // shutting it down. If the pod hasn't started yet, we know that when
        // the pod worker is invoked it will also avoid setting up the pod, so
        // we simply avoid doing any work.
        // We also do not try to admit the pod that is already in terminated state.
        if !kl.podWorkers.IsPodTerminationRequested(pod.UID) && !podutil.IsPodPhaseTerminal(pod.Status.Phase) {
            // We failed pods that we rejected, so activePods include all admitted
            // pods that are alive.
            activePods := kl.filterOutInactivePods(existingPods)

            if utilfeature.DefaultFeatureGate.Enabled(features.InPlacePodVerticalScaling) {
                // To handle kubelet restarts, test pod admissibility using AllocatedResources values
                // (for cpu & memory) from checkpoint store. If found, that is the source of truth.
                podCopy := pod.DeepCopy()
                kl.updateContainerResourceAllocation(podCopy)

                // Check if we can admit the pod; if not, reject it.
                if ok, reason, message := kl.canAdmitPod(activePods, podCopy); !ok {
                    kl.rejectPod(pod, reason, message)
                    continue
                }
                // For new pod, checkpoint the resource values at which the Pod has been admitted
                if err := kl.statusManager.SetPodAllocation(podCopy); err != nil {
                    //TODO(vinaykul,InPlacePodVerticalScaling): Can we recover from this in some way? Investigate
                    klog.ErrorS(err, "SetPodAllocation failed", "pod", klog.KObj(pod))
                }
            } else {
                // Check if we can admit the pod; if not, reject it.
                if ok, reason, message := kl.canAdmitPod(activePods, pod); !ok {
                    kl.rejectPod(pod, reason, message)
                    continue
                }
            }
        }
        kl.podWorkers.UpdatePod(UpdatePodOptions{
            Pod:        pod,
            MirrorPod:  mirrorPod,
            UpdateType: kubetypes.SyncPodCreate,
            StartTime:  start,
        })
    }
}

func (kl *Kubelet) canAdmitPod(pods []*v1.Pod, pod *v1.Pod) (bool, string, string) {
    // the kubelet will invoke each pod admit handler in sequence
    // if any handler rejects, the pod is rejected.
    // TODO: move out of disk check into a pod admitter
    // TODO: out of resource eviction should have a pod admitter call-out
    attrs := &lifecycle.PodAdmitAttributes{Pod: pod, OtherPods: pods}
    if utilfeature.DefaultFeatureGate.Enabled(features.InPlacePodVerticalScaling) {
        // Use allocated resources values from checkpoint store (source of truth) to determine fit
        otherPods := make([]*v1.Pod, 0, len(pods))
        for _, p := range pods {
            op := p.DeepCopy()
            kl.updateContainerResourceAllocation(op)

            otherPods = append(otherPods, op)
        }
        attrs.OtherPods = otherPods
    }
    for _, podAdmitHandler := range kl.admitHandlers {
        if result := podAdmitHandler.Admit(attrs); !result.Admit {
            return false, result.Reason, result.Message
        }
    }

    return true, "", ""
}

// NewMainKubelet instantiates a new Kubelet object along with all the required internal modules.
// No initialization of Kubelet and its modules should happen here.
func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
    kubeDeps *Dependencies,
    crOptions *config.ContainerRuntimeOptions,
    hostname string,
    hostnameOverridden bool,
    nodeName types.NodeName,
    nodeIPs []net.IP,
    providerID string,
    cloudProvider string,
    certDirectory string,
    rootDirectory string,
    podLogsDirectory string,
    imageCredentialProviderConfigFile string,
    imageCredentialProviderBinDir string,
    registerNode bool,
    registerWithTaints []v1.Taint,
    allowedUnsafeSysctls []string,
    experimentalMounterPath string,
    kernelMemcgNotification bool,
    experimentalNodeAllocatableIgnoreEvictionThreshold bool,
    minimumGCAge metav1.Duration,
    maxPerPodContainerCount int32,
    maxContainerCount int32,
    registerSchedulable bool,
    keepTerminatedPodVolumes bool,
    nodeLabels map[string]string,
    nodeStatusMaxImages int32,
    seccompDefault bool,
) (*Kubelet, error) {
    ctx := context.Background()
    logger := klog.TODO()

    ...

    // setup eviction manager
    evictionManager, evictionAdmitHandler := eviction.NewManager(klet.resourceAnalyzer, evictionConfig,
        killPodNow(klet.podWorkers, kubeDeps.Recorder), klet.imageManager, klet.containerGC, kubeDeps.Recorder, nodeRef, klet.clock, kubeCfg.LocalStorageCapacityIsolation)

    klet.evictionManager = evictionManager
    klet.admitHandlers.AddPodAdmitHandler(evictionAdmitHandler)

    // Safe, allowed sysctls can always be used as unsafe sysctls in the spec.
    // Hence, we concatenate those two lists.
    safeAndUnsafeSysctls := append(sysctl.SafeSysctlAllowlist(), allowedUnsafeSysctls...)
    sysctlsAllowlist, err := sysctl.NewAllowlist(safeAndUnsafeSysctls)
    if err != nil {
        return nil, err
    }
    klet.admitHandlers.AddPodAdmitHandler(sysctlsAllowlist)

    // enable active deadline handler
    activeDeadlineHandler, err := newActiveDeadlineHandler(klet.statusManager, kubeDeps.Recorder, klet.clock)
    if err != nil {
        return nil, err
    }
    klet.AddPodSyncLoopHandler(activeDeadlineHandler)
    klet.AddPodSyncHandler(activeDeadlineHandler)

    klet.admitHandlers.AddPodAdmitHandler(klet.containerManager.GetAllocateResourcesPodAdmitHandler())

    criticalPodAdmissionHandler := preemption.NewCriticalPodAdmissionHandler(klet.GetActivePods, killPodNow(klet.podWorkers, kubeDeps.Recorder), kubeDeps.Recorder)
    klet.admitHandlers.AddPodAdmitHandler(lifecycle.NewPredicateAdmitHandler(klet.getNodeAnyWay, criticalPodAdmissionHandler, klet.containerManager.UpdatePluginResources))
    // apply functional Option's
    for _, opt := range kubeDeps.Options {
        opt(klet)
    }

    if sysruntime.GOOS == "linux" {
        // AppArmor is a Linux kernel security module and it does not support other operating systems.
        klet.appArmorValidator = apparmor.NewValidator()
        klet.softAdmitHandlers.AddPodAdmitHandler(lifecycle.NewAppArmorAdmitHandler(klet.appArmorValidator))
    }

    leaseDuration := time.Duration(kubeCfg.NodeLeaseDurationSeconds) * time.Second
    renewInterval := time.Duration(float64(leaseDuration) * nodeLeaseRenewIntervalFraction)
    klet.nodeLeaseController = lease.NewController(
        klet.clock,
        klet.heartbeatClient,
        string(klet.nodeName),
        kubeCfg.NodeLeaseDurationSeconds,
        klet.onRepeatedHeartbeatFailure,
        renewInterval,
        string(klet.nodeName),
        v1.NamespaceNodeLease,
        util.SetNodeOwnerFunc(klet.heartbeatClient, string(klet.nodeName)))

    // setup node shutdown manager
    shutdownManager, shutdownAdmitHandler := nodeshutdown.NewManager(&nodeshutdown.Config{
        Logger:                           logger,
        ProbeManager:                     klet.probeManager,
        Recorder:                         kubeDeps.Recorder,
        NodeRef:                          nodeRef,
        GetPodsFunc:                      klet.GetActivePods,
        KillPodFunc:                      killPodNow(klet.podWorkers, kubeDeps.Recorder),
        SyncNodeStatusFunc:               klet.syncNodeStatus,
        ShutdownGracePeriodRequested:     kubeCfg.ShutdownGracePeriod.Duration,
        ShutdownGracePeriodCriticalPods:  kubeCfg.ShutdownGracePeriodCriticalPods.Duration,
        ShutdownGracePeriodByPodPriority: kubeCfg.ShutdownGracePeriodByPodPriority,
        StateDirectory:                   rootDirectory,
    })
    klet.shutdownManager = shutdownManager
    klet.usernsManager, err = userns.MakeUserNsManager(klet)
    if err != nil {
        return nil, fmt.Errorf("create user namespace manager: %w", err)
    }
    klet.admitHandlers.AddPodAdmitHandler(shutdownAdmitHandler)

    // Finally, put the most recent version of the config on the Kubelet, so
    // people can see how it was configured.
    klet.kubeletConfiguration = *kubeCfg

    // Generating the status funcs should be the last thing we do,
    // since this relies on the rest of the Kubelet having been constructed.
    klet.setNodeStatusFuncs = klet.defaultNodeStatusFuncs()

    return klet, nil
}

func (cm *containerManagerImpl) GetAllocateResourcesPodAdmitHandler() lifecycle.PodAdmitHandler {
    return cm.topologyManager
}

type containerManagerImpl struct {
    sync.RWMutex
    cadvisorInterface cadvisor.Interface
    mountUtil         mount.Interface
    NodeConfig
    status Status
    // External containers being managed.
    systemContainers []*systemContainer
    // Tasks that are run periodically
    periodicTasks []func()
    // Holds all the mounted cgroup subsystems
    subsystems *CgroupSubsystems
    nodeInfo   *v1.Node
    // Interface for cgroup management
    cgroupManager CgroupManager
    // Capacity of this node.
    capacity v1.ResourceList
    // Capacity of this node, including internal resources.
    internalCapacity v1.ResourceList
    // Absolute cgroupfs path to a cgroup that Kubelet needs to place all pods under.
    // This path include a top level container for enforcing Node Allocatable.
    cgroupRoot CgroupName
    // Event recorder interface.
    recorder record.EventRecorder
    // Interface for QoS cgroup management
    qosContainerManager QOSContainerManager
    // Interface for exporting and allocating devices reported by device plugins.
    deviceManager devicemanager.Manager
    // Interface for CPU affinity management.
    cpuManager cpumanager.Manager
    // Interface for memory affinity management.
    memoryManager memorymanager.Manager
    // Interface for Topology resource co-ordination
    topologyManager topologymanager.Manager
    // Interface for Dynamic Resource Allocation management.
    draManager dra.Manager
}

// Manager interface provides methods for Kubelet to manage pod topology hints
type Manager interface {
    // PodAdmitHandler is implemented by Manager
    lifecycle.PodAdmitHandler
    // AddHintProvider adds a hint provider to manager to indicate the hint provider
    // wants to be consulted with when making topology hints
    AddHintProvider(HintProvider)
    // AddContainer adds pod to Manager for tracking
    AddContainer(pod *v1.Pod, container *v1.Container, containerID string)
    // RemoveContainer removes pod from Manager tracking
    RemoveContainer(containerID string) error
    // Store is the interface for storing pod topology hints
    Store
}

type manager struct {
    //Topology Manager Scope
    scope Scope
}

...

func (m *manager) GetAffinity(podUID string, containerName string) TopologyHint {
    return m.scope.GetAffinity(podUID, containerName)
}

func (m *manager) GetPolicy() Policy {
    return m.scope.GetPolicy()
}

func (m *manager) AddHintProvider(h HintProvider) {
    m.scope.AddHintProvider(h)
}

func (m *manager) AddContainer(pod *v1.Pod, container *v1.Container, containerID string) {
    m.scope.AddContainer(pod, container, containerID)
}

func (m *manager) RemoveContainer(containerID string) error {
    return m.scope.RemoveContainer(containerID)
}

func (m *manager) Admit(attrs *lifecycle.PodAdmitAttributes) lifecycle.PodAdmitResult {
    klog.InfoS("Topology Admit Handler", "podUID", attrs.Pod.UID, "podNamespace", attrs.Pod.Namespace, "podName", attrs.Pod.Name)
    metrics.TopologyManagerAdmissionRequestsTotal.Inc()

    startTime := time.Now()
    podAdmitResult := m.scope.Admit(attrs.Pod)
    metrics.TopologyManagerAdmissionDuration.Observe(float64(time.Since(startTime).Milliseconds()))

    return podAdmitResult
}

// PodAdmitHandler is notified during pod admission.
type PodAdmitHandler interface {
    // Admit evaluates if a pod can be admitted.
    Admit(attrs *PodAdmitAttributes) PodAdmitResult
}

func (m *manager) Admit(attrs *lifecycle.PodAdmitAttributes) lifecycle.PodAdmitResult {
    klog.InfoS("Topology Admit Handler", "podUID", attrs.Pod.UID, "podNamespace", attrs.Pod.Namespace, "podName", attrs.Pod.Name)
    metrics.TopologyManagerAdmissionRequestsTotal.Inc()

    startTime := time.Now()
    podAdmitResult := m.scope.Admit(attrs.Pod)
    metrics.TopologyManagerAdmissionDuration.Observe(float64(time.Since(startTime).Milliseconds()))

    return podAdmitResult
}

// Scope interface for Topology Manager
type Scope interface {
    Name() string
    GetPolicy() Policy
    Admit(pod *v1.Pod) lifecycle.PodAdmitResult
    // AddHintProvider adds a hint provider to manager to indicate the hint provider
    // wants to be consoluted with when making topology hints
    AddHintProvider(h HintProvider)
    // AddContainer adds pod to Manager for tracking
    AddContainer(pod *v1.Pod, container *v1.Container, containerID string)
    // RemoveContainer removes pod from Manager tracking
    RemoveContainer(containerID string) error
    // Store is the interface for storing pod topology hints
    Store
}

type scope struct {
    mutex sync.Mutex
    name  string
    // Mapping of a Pods mapping of Containers and their TopologyHints
    // Indexed by PodUID to ContainerName
    podTopologyHints podTopologyHints
    // The list of components registered with the Manager
    hintProviders []HintProvider
    // Topology Manager Policy
    policy Policy
    // Mapping of (PodUid, ContainerName) to ContainerID for Adding/Removing Pods from PodTopologyHints mapping
    podMap containermap.ContainerMap
}

func (s *scope) Name() string {
    return s.name
}

func (s *scope) getTopologyHints(podUID string, containerName string) TopologyHint {
    s.mutex.Lock()
    defer s.mutex.Unlock()
    return s.podTopologyHints[podUID][containerName]
}

func (s *scope) setTopologyHints(podUID string, containerName string, th TopologyHint) {
    s.mutex.Lock()
    defer s.mutex.Unlock()

    if s.podTopologyHints[podUID] == nil {
        s.podTopologyHints[podUID] = make(map[string]TopologyHint)
    }
    s.podTopologyHints[podUID][containerName] = th
}

func (s *scope) GetAffinity(podUID string, containerName string) TopologyHint {
    return s.getTopologyHints(podUID, containerName)
}

func (s *scope) GetPolicy() Policy {
    return s.policy
}

func (s *scope) AddHintProvider(h HintProvider) {
    s.hintProviders = append(s.hintProviders, h)
}

// It would be better to implement this function in topologymanager instead of scope
// but topologymanager do not track mapping anymore
func (s *scope) AddContainer(pod *v1.Pod, container *v1.Container, containerID string) {
    s.mutex.Lock()
    defer s.mutex.Unlock()

    s.podMap.Add(string(pod.UID), container.Name, containerID)
}

// It would be better to implement this function in topologymanager instead of scope
// but topologymanager do not track mapping anymore
func (s *scope) RemoveContainer(containerID string) error {
    s.mutex.Lock()
    defer s.mutex.Unlock()

    klog.InfoS("RemoveContainer", "containerID", containerID)
    // Get the podUID and containerName associated with the containerID to be removed and remove it
    podUIDString, containerName, err := s.podMap.GetContainerRef(containerID)
    if err != nil {
        return nil
    }
    s.podMap.RemoveByContainerID(containerID)

    // In cases where a container has been restarted, it's possible that the same podUID and
    // containerName are already associated with a *different* containerID now. Only remove
    // the TopologyHints associated with that podUID and containerName if this is not true
    if _, err := s.podMap.GetContainerID(podUIDString, containerName); err != nil {
        delete(s.podTopologyHints[podUIDString], containerName)
        if len(s.podTopologyHints[podUIDString]) == 0 {
            delete(s.podTopologyHints, podUIDString)
        }
    }

    return nil
}

type containerScope struct {
    scope
}

// Ensure containerScope implements Scope interface
var _ Scope = &containerScope{}

// NewContainerScope returns a container scope.
func NewContainerScope(policy Policy) Scope {
    return &containerScope{
        scope{
            name:             containerTopologyScope,
            podTopologyHints: podTopologyHints{},
            policy:           policy,
            podMap:           containermap.NewContainerMap(),
        },
    }
}

func (s *containerScope) Admit(pod *v1.Pod) lifecycle.PodAdmitResult {
    for _, container := range append(pod.Spec.InitContainers, pod.Spec.Containers...) {
        bestHint, admit := s.calculateAffinity(pod, &container)
        klog.InfoS("Best TopologyHint", "bestHint", bestHint, "pod", klog.KObj(pod), "containerName", container.Name)

        if !admit {
            metrics.TopologyManagerAdmissionErrorsTotal.Inc()
            return admission.GetPodAdmitResult(&TopologyAffinityError{})
        }
        klog.InfoS("Topology Affinity", "bestHint", bestHint, "pod", klog.KObj(pod), "containerName", container.Name)
        s.setTopologyHints(string(pod.UID), container.Name, bestHint)

        err := s.allocateAlignedResources(pod, &container)
        if err != nil {
            metrics.TopologyManagerAdmissionErrorsTotal.Inc()
            return admission.GetPodAdmitResult(err)
        }
    }
    return admission.GetPodAdmitResult(nil)
}

func (s *containerScope) accumulateProvidersHints(pod *v1.Pod, container *v1.Container) []map[string][]TopologyHint {
    var providersHints []map[string][]TopologyHint

    for _, provider := range s.hintProviders {
        // Get the TopologyHints for a Container from a provider.
        hints := provider.GetTopologyHints(pod, container)
        providersHints = append(providersHints, hints)
        klog.InfoS("TopologyHints", "hints", hints, "pod", klog.KObj(pod), "containerName", container.Name)
    }
    return providersHints
}

func (s *containerScope) calculateAffinity(pod *v1.Pod, container *v1.Container) (TopologyHint, bool) {
    providersHints := s.accumulateProvidersHints(pod, container)
    bestHint, admit := s.policy.Merge(providersHints)
    klog.InfoS("ContainerTopologyHint", "bestHint", bestHint)
    return bestHint, admit
}

type podScope struct {
    scope
}

// Ensure podScope implements Scope interface
var _ Scope = &podScope{}

// NewPodScope returns a pod scope.
func NewPodScope(policy Policy) Scope {
    return &podScope{
        scope{
            name:             podTopologyScope,
            podTopologyHints: podTopologyHints{},
            policy:           policy,
            podMap:           containermap.NewContainerMap(),
        },
    }
}

func (s *podScope) Admit(pod *v1.Pod) lifecycle.PodAdmitResult {
    bestHint, admit := s.calculateAffinity(pod)
    klog.InfoS("Best TopologyHint", "bestHint", bestHint, "pod", klog.KObj(pod))
    if !admit {
        metrics.TopologyManagerAdmissionErrorsTotal.Inc()
        return admission.GetPodAdmitResult(&TopologyAffinityError{})
    }

    for _, container := range append(pod.Spec.InitContainers, pod.Spec.Containers...) {
        klog.InfoS("Topology Affinity", "bestHint", bestHint, "pod", klog.KObj(pod), "containerName", container.Name)
        s.setTopologyHints(string(pod.UID), container.Name, bestHint)

        err := s.allocateAlignedResources(pod, &container)
        if err != nil {
            metrics.TopologyManagerAdmissionErrorsTotal.Inc()
            return admission.GetPodAdmitResult(err)
        }
    }
    return admission.GetPodAdmitResult(nil)
}

func (s *podScope) accumulateProvidersHints(pod *v1.Pod) []map[string][]TopologyHint {
    var providersHints []map[string][]TopologyHint

    for _, provider := range s.hintProviders {
        // Get the TopologyHints for a Pod from a provider.
        hints := provider.GetPodTopologyHints(pod)
        providersHints = append(providersHints, hints)
        klog.InfoS("TopologyHints", "hints", hints, "pod", klog.KObj(pod))
    }
    return providersHints
}

func (s *podScope) calculateAffinity(pod *v1.Pod) (TopologyHint, bool) {
    providersHints := s.accumulateProvidersHints(pod)
    bestHint, admit := s.policy.Merge(providersHints)
    klog.InfoS("PodTopologyHint", "bestHint", bestHint)
    return bestHint, admit
}

func NewContainerManager(mountUtil mount.Interface, cadvisorInterface cadvisor.Interface, nodeConfig NodeConfig, failSwapOn bool, recorder record.EventRecorder, kubeClient clientset.Interface) (ContainerManager, error) {
    subsystems, err := GetCgroupSubsystems()
    ...

    cm := &containerManagerImpl{
        cadvisorInterface:   cadvisorInterface,
        mountUtil:           mountUtil,
        NodeConfig:          nodeConfig,
        subsystems:          subsystems,
        cgroupManager:       cgroupManager,
        capacity:            capacity,
        internalCapacity:    internalCapacity,
        cgroupRoot:          cgroupRoot,
        recorder:            recorder,
        qosContainerManager: qosContainerManager,
    }

    cm.topologyManager, err = topologymanager.NewManager(
        machineInfo.Topology,
        nodeConfig.TopologyManagerPolicy,
        nodeConfig.TopologyManagerScope,
        nodeConfig.TopologyManagerPolicyOptions,
    )

    if err != nil {
        return nil, err
    }

    klog.InfoS("Creating device plugin manager")
    cm.deviceManager, err = devicemanager.NewManagerImpl(machineInfo.Topology, cm.topologyManager)
    if err != nil {
        return nil, err
    }
    cm.topologyManager.AddHintProvider(cm.deviceManager)

    // initialize DRA manager
    if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.DynamicResourceAllocation) {
        klog.InfoS("Creating Dynamic Resource Allocation (DRA) manager")
        cm.draManager, err = dra.NewManagerImpl(kubeClient, nodeConfig.KubeletRootDir, nodeConfig.NodeName)
        if err != nil {
            return nil, err
        }
    }

    // Initialize CPU manager
    cm.cpuManager, err = cpumanager.NewManager(
        nodeConfig.CPUManagerPolicy,
        nodeConfig.CPUManagerPolicyOptions,
        nodeConfig.CPUManagerReconcilePeriod,
        machineInfo,
        nodeConfig.NodeAllocatableConfig.ReservedSystemCPUs,
        cm.GetNodeAllocatableReservation(),
        nodeConfig.KubeletRootDir,
        cm.topologyManager,
    )
    if err != nil {
        klog.ErrorS(err, "Failed to initialize cpu manager")
        return nil, err
    }
    cm.topologyManager.AddHintProvider(cm.cpuManager)

    if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.MemoryManager) {
        cm.memoryManager, err = memorymanager.NewManager(
            nodeConfig.ExperimentalMemoryManagerPolicy,
            machineInfo,
            cm.GetNodeAllocatableReservation(),
            nodeConfig.ExperimentalMemoryManagerReservedMemory,
            nodeConfig.KubeletRootDir,
            cm.topologyManager,
        )
        if err != nil {
            klog.ErrorS(err, "Failed to initialize memory manager")
            return nil, err
        }
        cm.topologyManager.AddHintProvider(cm.memoryManager)
    }

    return cm, nil
}

func NewContainerManager(mountUtil mount.Interface, cadvisorInterface cadvisor.Interface, nodeConfig NodeConfig, failSwapOn bool, recorder record.EventRecorder, kubeClient clientset.Interface) (ContainerManager, error) {
    ...

    cm := &containerManagerImpl{
        cadvisorInterface:   cadvisorInterface,
        mountUtil:           mountUtil,
        NodeConfig:          nodeConfig,
        subsystems:          subsystems,
        cgroupManager:       cgroupManager,
        capacity:            capacity,
        internalCapacity:    internalCapacity,
        cgroupRoot:          cgroupRoot,
        recorder:            recorder,
        qosContainerManager: qosContainerManager,
    }

    cm.topologyManager, err = topologymanager.NewManager(
        machineInfo.Topology,
        nodeConfig.TopologyManagerPolicy,
        nodeConfig.TopologyManagerScope,
        nodeConfig.TopologyManagerPolicyOptions,
    )

    if err != nil {
        return nil, err
    }

    klog.InfoS("Creating device plugin manager")
    cm.deviceManager, err = devicemanager.NewManagerImpl(machineInfo.Topology, cm.topologyManager)
    if err != nil {
        return nil, err
    }
    cm.topologyManager.AddHintProvider(cm.deviceManager)

    // Initialize CPU manager
    cm.cpuManager, err = cpumanager.NewManager(
        nodeConfig.CPUManagerPolicy,
        nodeConfig.CPUManagerPolicyOptions,
        nodeConfig.CPUManagerReconcilePeriod,
        machineInfo,
        nodeConfig.NodeAllocatableConfig.ReservedSystemCPUs,
        cm.GetNodeAllocatableReservation(),
        nodeConfig.KubeletRootDir,
        cm.topologyManager,
    )
    if err != nil {
        klog.ErrorS(err, "Failed to initialize cpu manager")
        return nil, err
    }
    cm.topologyManager.AddHintProvider(cm.cpuManager)

    if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.MemoryManager) {
        cm.memoryManager, err = memorymanager.NewManager(
            nodeConfig.ExperimentalMemoryManagerPolicy,
            machineInfo,
            cm.GetNodeAllocatableReservation(),
            nodeConfig.ExperimentalMemoryManagerReservedMemory,
            nodeConfig.KubeletRootDir,
            cm.topologyManager,
        )
        if err != nil {
            klog.ErrorS(err, "Failed to initialize memory manager")
            return nil, err
        }
        cm.topologyManager.AddHintProvider(cm.memoryManager)
    }

    return cm, nil
}

// Manager interface provides methods for Kubelet to manage pod topology hints
type Manager interface {
    // PodAdmitHandler is implemented by Manager
    lifecycle.PodAdmitHandler
    // AddHintProvider adds a hint provider to manager to indicate the hint provider
    // wants to be consulted with when making topology hints
    AddHintProvider(HintProvider)
    // AddContainer adds pod to Manager for tracking
    AddContainer(pod *v1.Pod, container *v1.Container, containerID string)
    // RemoveContainer removes pod from Manager tracking
    RemoveContainer(containerID string) error
    // Store is the interface for storing pod topology hints
    Store
}

type manager struct {
    //Topology Manager Scope
    scope Scope
}

...

// NewManager creates a new TopologyManager based on provided policy and scope
func NewManager(topology []cadvisorapi.Node, topologyPolicyName string, topologyScopeName string, topologyPolicyOptions map[string]string) (Manager, error) {
    // When policy is none, the scope is not relevant, so we can short circuit here.
    if topologyPolicyName == PolicyNone {
        klog.InfoS("Creating topology manager with none policy")
        return &manager{scope: NewNoneScope()}, nil
    }

    opts, err := NewPolicyOptions(topologyPolicyOptions)
    if err != nil {
        return nil, err
    }

    klog.InfoS("Creating topology manager with policy per scope", "topologyPolicyName", topologyPolicyName, "topologyScopeName", topologyScopeName, "topologyPolicyOptions", opts)

    numaInfo, err := NewNUMAInfo(topology, opts)
    if err != nil {
        return nil, fmt.Errorf("cannot discover NUMA topology: %w", err)
    }

    if topologyPolicyName != PolicyNone && len(numaInfo.Nodes) > maxAllowableNUMANodes {
        return nil, fmt.Errorf("unsupported on machines with more than %v NUMA Nodes", maxAllowableNUMANodes)
    }

    var policy Policy
    switch topologyPolicyName {

    case PolicyBestEffort:
        policy = NewBestEffortPolicy(numaInfo, opts)

    case PolicyRestricted:
        policy = NewRestrictedPolicy(numaInfo, opts)

    case PolicySingleNumaNode:
        policy = NewSingleNumaNodePolicy(numaInfo, opts)

    default:
        return nil, fmt.Errorf("unknown policy: \"%s\"", topologyPolicyName)
    }

    var scope Scope
    switch topologyScopeName {

    case containerTopologyScope:
        scope = NewContainerScope(policy)

    case podTopologyScope:
        scope = NewPodScope(policy)

    default:
        return nil, fmt.Errorf("unknown scope: \"%s\"", topologyScopeName)
    }

    manager := &manager{
        scope: scope,
    }

    return manager, nil
}