Skip to content

K
k3s
Commit bb1920ed authored by Jing Xu's avatar Jing Xu
Browse files
Options

Fix issues for local storage allocatable feature

This PR fixes the following issues: 1. Use ResourceStorageScratch instead of ResourceStorage API to represent local storage capacity 2. In eviction manager, use container manager instead of node provider (kubelet) to retrieve the node capacity and reserved resources. Node provider (kubelet) has a feature gate so that storagescratch information may not be exposed if feature gate is not set. On the other hand, container manager has all the capacity and allocatable resource information.
parent 9e223539
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 64 additions and 43 deletions
cmd/kubelet/app/server.go
View file @ bb1920ed
......@@ -942,7 +942,12 @@ func parseResourceList(m componentconfig.ConfigurationMap) (v1.ResourceList, err
if q.Sign() == -1 {
return nil, fmt.Errorf("resource quantity for %q cannot be negative: %v", k, v)
}
rl[v1.ResourceName(k)] = q
// storage specified in configuration map is mapped to ResourceStorageScratch API
if v1.ResourceName(k) == v1.ResourceStorage {
rl[v1.ResourceStorageScratch] = q
} else {
rl[v1.ResourceName(k)] = q
}
default:
return nil, fmt.Errorf("cannot reserve %q resource", k)
}
......
pkg/kubelet/cm/node_container_manager.go
View file @ bb1920ed
......@@ -217,6 +217,10 @@ func hardEvictionReservation(thresholds []evictionapi.Threshold, capacity v1.Res
memoryCapacity := capacity[v1.ResourceMemory]
value := evictionapi.GetThresholdQuantity(threshold.Value, &memoryCapacity)
ret[v1.ResourceMemory] = *value
case evictionapi.SignalNodeFsAvailable:
storageCapacity := capacity[v1.ResourceStorageScratch]
value := evictionapi.GetThresholdQuantity(threshold.Value, &storageCapacity)
ret[v1.ResourceStorageScratch] = *value
}
}
return ret
......
pkg/kubelet/eviction/eviction_manager.go
View file @ bb1920ed
......@@ -148,11 +148,11 @@ func (m *managerImpl) Admit(attrs *lifecycle.PodAdmitAttributes) lifecycle.PodAd
}
// Start starts the control loop to observe and response to low compute resources.
func (m *managerImpl) Start(diskInfoProvider DiskInfoProvider, podFunc ActivePodsFunc, podCleanedUpFunc PodCleanedUpFunc, nodeProvider NodeProvider, monitoringInterval time.Duration) {
func (m *managerImpl) Start(diskInfoProvider DiskInfoProvider, podFunc ActivePodsFunc, podCleanedUpFunc PodCleanedUpFunc, capacityProvider CapacityProvider, monitoringInterval time.Duration) {
// start the eviction manager monitoring
go func() {
for {
if evictedPods := m.synchronize(diskInfoProvider, podFunc, nodeProvider); evictedPods != nil {
if evictedPods := m.synchronize(diskInfoProvider, podFunc, capacityProvider); evictedPods != nil {
glog.Infof("eviction manager: pods %s evicted, waiting for pod to be cleaned up", format.Pods(evictedPods))
m.waitForPodsCleanup(podCleanedUpFunc, evictedPods)
} else {
......@@ -211,7 +211,7 @@ func startMemoryThresholdNotifier(thresholds []evictionapi.Threshold, observatio
// synchronize is the main control loop that enforces eviction thresholds.
// Returns the pod that was killed, or nil if no pod was killed.
func (m *managerImpl) synchronize(diskInfoProvider DiskInfoProvider, podFunc ActivePodsFunc, nodeProvider NodeProvider) []*v1.Pod {
func (m *managerImpl) synchronize(diskInfoProvider DiskInfoProvider, podFunc ActivePodsFunc, capacityProvider CapacityProvider) []*v1.Pod {
// if we have nothing to do, just return
thresholds := m.config.Thresholds
if len(thresholds) == 0 {
......@@ -233,7 +233,7 @@ func (m *managerImpl) synchronize(diskInfoProvider DiskInfoProvider, podFunc Act
activePods := podFunc()
// make observations and get a function to derive pod usage stats relative to those observations.
observations, statsFunc, err := makeSignalObservations(m.summaryProvider, nodeProvider, activePods, *m.dedicatedImageFs)
observations, statsFunc, err := makeSignalObservations(m.summaryProvider, capacityProvider, activePods, *m.dedicatedImageFs)
if err != nil {
glog.Errorf("eviction manager: unexpected err: %v", err)
return nil
......@@ -248,7 +248,7 @@ func (m *managerImpl) synchronize(diskInfoProvider DiskInfoProvider, podFunc Act
err = startMemoryThresholdNotifier(m.config.Thresholds, observations, false, func(desc string) {
glog.Infof("soft memory eviction threshold crossed at %s", desc)
// TODO wait grace period for soft memory limit
m.synchronize(diskInfoProvider, podFunc, nodeProvider)
m.synchronize(diskInfoProvider, podFunc, capacityProvider)
})
if err != nil {
glog.Warningf("eviction manager: failed to create hard memory threshold notifier: %v", err)
......@@ -256,7 +256,7 @@ func (m *managerImpl) synchronize(diskInfoProvider DiskInfoProvider, podFunc Act
// start hard memory notification
err = startMemoryThresholdNotifier(m.config.Thresholds, observations, true, func(desc string) {
glog.Infof("hard memory eviction threshold crossed at %s", desc)
m.synchronize(diskInfoProvider, podFunc, nodeProvider)
m.synchronize(diskInfoProvider, podFunc, capacityProvider)
})
if err != nil {
glog.Warningf("eviction manager: failed to create soft memory threshold notifier: %v", err)
......
pkg/kubelet/eviction/helpers.go
View file @ bb1920ed
......@@ -658,16 +658,11 @@ func (a byEvictionPriority) Less(i, j int) bool {
}
// makeSignalObservations derives observations using the specified summary provider.
func makeSignalObservations(summaryProvider stats.SummaryProvider, nodeProvider NodeProvider, pods []*v1.Pod, withImageFs bool) (signalObservations, statsFunc, error) {
func makeSignalObservations(summaryProvider stats.SummaryProvider, capacityProvider CapacityProvider, pods []*v1.Pod, withImageFs bool) (signalObservations, statsFunc, error) {
summary, err := summaryProvider.Get()
if err != nil {
return nil, nil, err
}
node, err := nodeProvider.GetNode()
if err != nil {
return nil, nil, err
}
// build the function to work against for pod stats
statsFunc := cachedStatsFunc(summary.Pods)
// build an evaluation context for current eviction signals
......@@ -714,8 +709,12 @@ func makeSignalObservations(summaryProvider stats.SummaryProvider, nodeProvider
}
}
}
if memoryAllocatableCapacity, ok := node.Status.Allocatable[v1.ResourceMemory]; ok {
memoryAllocatableAvailable := memoryAllocatableCapacity.Copy()
nodeCapacity := capacityProvider.GetCapacity()
allocatableReservation := capacityProvider.GetNodeAllocatableReservation()
memoryAllocatableCapacity, memoryAllocatableAvailable, exist := getResourceAllocatable(nodeCapacity, allocatableReservation, v1.ResourceMemory)
if exist {
for _, pod := range summary.Pods {
mu, err := podMemoryUsage(pod)
if err == nil {
......@@ -724,12 +723,12 @@ func makeSignalObservations(summaryProvider stats.SummaryProvider, nodeProvider
}
result[evictionapi.SignalAllocatableMemoryAvailable] = signalObservation{
available: memoryAllocatableAvailable,
capacity: memoryAllocatableCapacity.Copy(),
capacity: memoryAllocatableCapacity,
}
}
if storageScratchAllocatableCapacity, ok := node.Status.Allocatable[v1.ResourceStorage]; ok {
storageScratchAllocatable := storageScratchAllocatableCapacity.Copy()
storageScratchCapacity, storageScratchAllocatable, exist := getResourceAllocatable(nodeCapacity, allocatableReservation, v1.ResourceStorageScratch)
if exist {
for _, pod := range pods {
podStat, ok := statsFunc(pod)
if !ok {
......@@ -754,13 +753,25 @@ func makeSignalObservations(summaryProvider stats.SummaryProvider, nodeProvider
}
result[evictionapi.SignalAllocatableNodeFsAvailable] = signalObservation{
available: storageScratchAllocatable,
capacity: storageScratchAllocatableCapacity.Copy(),
capacity: storageScratchCapacity,
}
}
return result, statsFunc, nil
}
func getResourceAllocatable(capacity v1.ResourceList, reservation v1.ResourceList, resourceName v1.ResourceName) (*resource.Quantity, *resource.Quantity, bool) {
if capacity, ok := capacity[resourceName]; ok {
allocate := capacity.Copy()
if reserved, exists := reservation[resourceName]; exists {
allocate.Sub(reserved)
}
return capacity.Copy(), allocate, true
}
glog.Errorf("Could not find capacity information for resource %v", resourceName)
return nil, nil, false
}
// thresholdsMet returns the set of thresholds that were met independent of grace period
func thresholdsMet(thresholds []evictionapi.Threshold, observations signalObservations, enforceMinReclaim bool) []evictionapi.Threshold {
results := []evictionapi.Threshold{}
......
pkg/kubelet/eviction/helpers_test.go
View file @ bb1920ed
......@@ -782,12 +782,12 @@ func TestMakeSignalObservations(t *testing.T) {
fakeStats.Pods = append(fakeStats.Pods, newPodStats(pod, containerWorkingSetBytes))
}
res := quantityMustParse("5Gi")
nodeProvider := newMockNodeProvider(v1.ResourceList{v1.ResourceMemory: *res})
capacityProvider := newMockCapacityProvider(v1.ResourceList{v1.ResourceMemory: *quantityMustParse("5Gi")}, v1.ResourceList{v1.ResourceMemory: *quantityMustParse("0Gi")})
// Allocatable thresholds are always 100%. Verify that Threshold == Capacity.
if res.CmpInt64(int64(allocatableMemoryCapacity)) != 0 {
t.Errorf("Expected Threshold %v to be equal to value %v", res.Value(), allocatableMemoryCapacity)
}
actualObservations, statsFunc, err := makeSignalObservations(provider, nodeProvider, pods, false)
actualObservations, statsFunc, err := makeSignalObservations(provider, capacityProvider, pods, false)
if err != nil {
t.Errorf("Unexpected err: %v", err)
}
......
pkg/kubelet/eviction/types.go
View file @ bb1920ed
......@@ -53,7 +53,7 @@ type Config struct {
// Manager evaluates when an eviction threshold for node stability has been met on the node.
type Manager interface {
// Start starts the control loop to monitor eviction thresholds at specified interval.
Start(diskInfoProvider DiskInfoProvider, podFunc ActivePodsFunc, podCleanedUpFunc PodCleanedUpFunc, nodeProvider NodeProvider, monitoringInterval time.Duration)
Start(diskInfoProvider DiskInfoProvider, podFunc ActivePodsFunc, podCleanedUpFunc PodCleanedUpFunc, capacityProvider CapacityProvider, monitoringInterval time.Duration)
// IsUnderMemoryPressure returns true if the node is under memory pressure.
IsUnderMemoryPressure() bool
......@@ -68,10 +68,12 @@ type DiskInfoProvider interface {
HasDedicatedImageFs() (bool, error)
}
// NodeProvider is responsible for providing the node api object describing this node
type NodeProvider interface {
// GetNode returns the node info for this node
GetNode() (*v1.Node, error)
// CapacityProvider is responsible for providing the resource capacity and reservation information
type CapacityProvider interface {
// GetCapacity returns the amount of compute resources tracked by container manager available on the node.
GetCapacity() v1.ResourceList
// GetNodeAllocatable returns the amount of compute resources that have to be reserved from scheduling.
GetNodeAllocatableReservation() v1.ResourceList
}
// ImageGC is responsible for performing garbage collection of unused images.
......
pkg/kubelet/kubelet.go
View file @ bb1920ed
......@@ -1250,7 +1250,7 @@ func (kl *Kubelet) initializeRuntimeDependentModules() {
glog.Fatalf("Failed to start cAdvisor %v", err)
}
// eviction manager must start after cadvisor because it needs to know if the container runtime has a dedicated imagefs
kl.evictionManager.Start(kl.cadvisor, kl.GetActivePods, kl.podResourcesAreReclaimed, kl, evictionMonitoringPeriod)
kl.evictionManager.Start(kl.cadvisor, kl.GetActivePods, kl.podResourcesAreReclaimed, kl.containerManager, evictionMonitoringPeriod)
}
// Run starts the kubelet reacting to config updates
......
plugin/pkg/scheduler/algorithm/predicates/predicates_test.go
View file @ bb1920ed
......@@ -80,24 +80,24 @@ var (
func makeResources(milliCPU, memory, nvidiaGPUs, pods, opaqueA, storage int64) v1.NodeResources {
return v1.NodeResources{
Capacity: v1.ResourceList{
v1.ResourceCPU: *resource.NewMilliQuantity(milliCPU, resource.DecimalSI),
v1.ResourceMemory: *resource.NewQuantity(memory, resource.BinarySI),
v1.ResourcePods: *resource.NewQuantity(pods, resource.DecimalSI),
v1.ResourceNvidiaGPU: *resource.NewQuantity(nvidiaGPUs, resource.DecimalSI),
opaqueResourceA: *resource.NewQuantity(opaqueA, resource.DecimalSI),
v1.ResourceStorage: *resource.NewQuantity(storage, resource.BinarySI),
v1.ResourceCPU: *resource.NewMilliQuantity(milliCPU, resource.DecimalSI),
v1.ResourceMemory: *resource.NewQuantity(memory, resource.BinarySI),
v1.ResourcePods: *resource.NewQuantity(pods, resource.DecimalSI),
v1.ResourceNvidiaGPU: *resource.NewQuantity(nvidiaGPUs, resource.DecimalSI),
opaqueResourceA: *resource.NewQuantity(opaqueA, resource.DecimalSI),
v1.ResourceStorageScratch: *resource.NewQuantity(storage, resource.BinarySI),
},
}
}
func makeAllocatableResources(milliCPU, memory, nvidiaGPUs, pods, opaqueA, storage int64) v1.ResourceList {
return v1.ResourceList{
v1.ResourceCPU: *resource.NewMilliQuantity(milliCPU, resource.DecimalSI),
v1.ResourceMemory: *resource.NewQuantity(memory, resource.BinarySI),
v1.ResourcePods: *resource.NewQuantity(pods, resource.DecimalSI),
v1.ResourceNvidiaGPU: *resource.NewQuantity(nvidiaGPUs, resource.DecimalSI),
opaqueResourceA: *resource.NewQuantity(opaqueA, resource.DecimalSI),
v1.ResourceStorage: *resource.NewQuantity(storage, resource.BinarySI),
v1.ResourceCPU: *resource.NewMilliQuantity(milliCPU, resource.DecimalSI),
v1.ResourceMemory: *resource.NewQuantity(memory, resource.BinarySI),
v1.ResourcePods: *resource.NewQuantity(pods, resource.DecimalSI),
v1.ResourceNvidiaGPU: *resource.NewQuantity(nvidiaGPUs, resource.DecimalSI),
opaqueResourceA: *resource.NewQuantity(opaqueA, resource.DecimalSI),
v1.ResourceStorageScratch: *resource.NewQuantity(storage, resource.BinarySI),
}
}
......
plugin/pkg/scheduler/schedulercache/node_info.go
View file @ bb1920ed
......@@ -97,7 +97,7 @@ func (r *Resource) Add(rl v1.ResourceList) {
r.NvidiaGPU += rQuant.Value()
case v1.ResourcePods:
r.AllowedPodNumber += int(rQuant.Value())
case v1.ResourceStorage:
case v1.ResourceStorageScratch:
r.StorageScratch += rQuant.Value()
case v1.ResourceStorageOverlay:
r.StorageOverlay += rQuant.Value()
......@@ -116,7 +116,7 @@ func (r *Resource) ResourceList() v1.ResourceList {
v1.ResourceNvidiaGPU: *resource.NewQuantity(r.NvidiaGPU, resource.DecimalSI),
v1.ResourcePods: *resource.NewQuantity(int64(r.AllowedPodNumber), resource.BinarySI),
v1.ResourceStorageOverlay: *resource.NewQuantity(r.StorageOverlay, resource.BinarySI),
v1.ResourceStorage: *resource.NewQuantity(r.StorageScratch, resource.BinarySI),
v1.ResourceStorageScratch: *resource.NewQuantity(r.StorageScratch, resource.BinarySI),
}
for rName, rQuant := range r.OpaqueIntResources {
result[rName] = *resource.NewQuantity(rQuant, resource.DecimalSI)
......
test/e2e_node/local_storage_allocatable_eviction_test.go
View file @ bb1920ed
......@@ -52,8 +52,7 @@ var _ = framework.KubeDescribe("LocalStorageAllocatableEviction [Slow] [Serial]
diskReserve = uint64(0.8 * diskAvail / 1000000) // Reserve 0.8 * disk Capacity for kube-reserved scratch storage
maxDisk := 10000000 // Set dd command to read and write up to 10MB at a time
count := uint64(0.8 * diskAvail / float64(maxDisk))
command := fmt.Sprintf("dd if=/dev/urandom of=dummy bs=%d count=%d; sleep 0.5; while true; do sleep 5; done", maxDisk, count)
command := fmt.Sprintf("dd if=/dev/urandom of=dummy bs=%d count=%d; while true; do sleep 5; done", maxDisk, count)
podTestSpecs = []podTestSpec{
{
evictionPriority: 1, // This pod should be evicted before the innocent pod
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment