Merge pull request #219 from daimaxiaxie/fix-kubelet-reserved

fix kube reserved

Author: jwcesign

pkg/providers/cluster/ackmanaged.go

@@ -23,6 +23,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"math"
 	"net/http"
 	"regexp"
 	"sort"
@@ -34,7 +35,10 @@ import (
 	"github.com/patrickmn/go-cache"
 	"github.com/samber/lo"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/karpenter/pkg/cloudprovider"
+	"sigs.k8s.io/karpenter/pkg/utils/resources"
 
 	"github.com/cloudpilot-ai/karpenter-provider-alibabacloud/pkg/apis/v1alpha1"
 )
@@ -294,6 +298,57 @@ func (a *ACKManaged) formatTaints(taints []corev1.Taint) string {
 	}), ",")
 }
 
+func (a *ACKManaged) DefaultOverhead(capacity corev1.ResourceList) cloudprovider.InstanceTypeOverhead {
+	// referring to: https://help.aliyun.com/zh/ack/ack-managed-and-ack-dedicated/user-guide/resource-reservation-policy#0f5ffe176df7q
+	// CPU overhead calculation
+	cpuOverHead := calculateCPUOverhead(capacity.Cpu().MilliValue())
+
+	// TODO: In a real environment, the formula does not produce accurate results,
+	// consistently yielding values that are 200MiB larger than expected.
+	// Memory overhead: min(11*pods + 255, memoryMi*0.25)
+	memoryOverHead := int64(math.Min(float64(11*capacity.Pods().Value()+255), float64(capacity.Memory().Value()/1024*1024)*0.25)) + 200
+
+	return cloudprovider.InstanceTypeOverhead{
+		KubeReserved: corev1.ResourceList{
+			corev1.ResourceCPU:    *resource.NewMilliQuantity(cpuOverHead/2, resource.DecimalSI),
+			corev1.ResourceMemory: *resources.Quantity(fmt.Sprintf("%dMi", memoryOverHead/2)),
+		},
+		SystemReserved: corev1.ResourceList{
+			corev1.ResourceCPU:    *resource.NewMilliQuantity(cpuOverHead/2, resource.DecimalSI),
+			corev1.ResourceMemory: *resources.Quantity(fmt.Sprintf("%dMi", memoryOverHead/2)),
+		},
+	}
+}
+
+// thresholds defines CPU overhead thresholds and their corresponding percentages
+var thresholds = [...]struct {
+	cores    int64
+	overhead float64
+}{
+	{1000, 0.06},
+	{2000, 0.01},
+	{3000, 0.005},
+	{4000, 0.005},
+}
+
+func calculateCPUOverhead(cpuM int64) int64 {
+	var cpuOverHead int64
+
+	// Calculate overhead for each threshold
+	for _, t := range thresholds {
+		if cpuM >= t.cores {
+			cpuOverHead += int64(1000 * t.overhead)
+		}
+	}
+
+	// Additional overhead for CPU > 4 cores (0.25%)
+	if cpuM > 4000 {
+		cpuOverHead += int64(float64(cpuM-4000) * 0.0025)
+	}
+
+	return cpuOverHead
+}
+
 type NodeConfig struct {
 	KubeletConfig *ACKKubeletConfig `json:"kubelet_config,omitempty"`
 }

pkg/providers/cluster/ackmanaged_test.go

@@ -17,10 +17,13 @@ limitations under the License.
 package cluster
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/alibabacloud-go/tea/tea"
 	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 
 	"github.com/cloudpilot-ai/karpenter-provider-alibabacloud/pkg/apis/v1alpha1"
 )
@@ -32,3 +35,47 @@ func Test_convertNodeClassKubeletConfigToACKNodeConfig(t *testing.T) {
 	d := convertNodeClassKubeletConfigToACKNodeConfig(kubeletCfg)
 	assert.Equal(t, "eyJrdWJlbGV0X2NvbmZpZyI6eyJtYXhQb2RzIjoxMTB9fQ==", d)
 }
+
+// referring to: https://help.aliyun.com/zh/ack/ack-managed-and-ack-dedicated/user-guide/resource-reservation-policy#0f5ffe176df7q
+func TestDefaultOverhead(t *testing.T) {
+	provider := NewACKManaged("test-cluster", "cn-hangzhou", nil, nil)
+
+	// ECS c7 / 1.28+
+	cases := []struct {
+		name         string
+		cpuCores     int64
+		memoryGi     int64
+		maxPods      int64
+		wantCPUMilli int64
+		wantMemMi    int64
+	}{
+		{"2C4Gi-15pods", 2, 4, 15, 70, 420 + 200},
+		{"4C8Gi-48pods", 4, 8, 48, 80, 982},
+		{"8C16Gi-48pods", 8, 16, 48, 90, 982},
+		{"16C32Gi-213pods", 16, 32, 213, 110, 2598 + 200},
+		{"32C64Gi-213pods", 32, 64, 213, 150, 2598 + 200},
+		{"64C128Gi-213pods", 64, 128, 213, 230, 2598 + 200},
+		{"128C256Gi-423pods", 128, 256, 423, 390, 4908 + 200},
+	}
+
+	for _, tt := range cases {
+		t.Run(tt.name, func(t *testing.T) {
+			capacity := corev1.ResourceList{
+				corev1.ResourceCPU:    *resource.NewQuantity(tt.cpuCores, resource.DecimalSI),
+				corev1.ResourceMemory: resource.MustParse(fmt.Sprintf("%dGi", tt.memoryGi)),
+				corev1.ResourcePods:   *resource.NewQuantity(tt.maxPods, resource.DecimalSI),
+			}
+			overhead := provider.DefaultOverhead(capacity)
+
+			kubeCPU := overhead.KubeReserved.Cpu().MilliValue()
+			sysCPU := overhead.SystemReserved.Cpu().MilliValue()
+			assert.Equal(t, tt.wantCPUMilli, kubeCPU+sysCPU, "total CPU reserved")
+			assert.Equal(t, kubeCPU, sysCPU, "CPU split 50/50")
+
+			kubeMemMi := overhead.KubeReserved.Memory().Value() / (1024 * 1024)
+			sysMemMi := overhead.SystemReserved.Memory().Value() / (1024 * 1024)
+			assert.Equal(t, tt.wantMemMi, kubeMemMi+sysMemMi, "total memory reserved (Mi)")
+			assert.Equal(t, kubeMemMi, sysMemMi, "memory split 50/50")
+		})
+	}
+}

pkg/providers/cluster/custom.go

@@ -19,10 +19,12 @@ package cluster
 import (
 	"context"
 	"encoding/base64"
+	"net/http"
+
 	"github.com/cloudpilot-ai/karpenter-provider-alibabacloud/pkg/apis/v1alpha1"
 	"github.com/samber/lo"
 	corev1 "k8s.io/api/core/v1"
-	"net/http"
+	"sigs.k8s.io/karpenter/pkg/cloudprovider"
 )
 
 const customClusterType = "Custom"
@@ -60,3 +62,7 @@ func (c *Custom) FeatureFlags() FeatureFlags {
 		SupportsENILimitedPodDensity: false,
 	}
 }
+
+func (c *Custom) DefaultOverhead(_ corev1.ResourceList) cloudprovider.InstanceTypeOverhead {
+	return cloudprovider.InstanceTypeOverhead{}
+}

pkg/providers/cluster/types.go

@@ -23,6 +23,7 @@ import (
 	ackclient "github.com/alibabacloud-go/cs-20151215/v5/client"
 	"github.com/patrickmn/go-cache"
 	corev1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/karpenter/pkg/cloudprovider"
 
 	"github.com/cloudpilot-ai/karpenter-provider-alibabacloud/pkg/apis/v1alpha1"
 	alicache "github.com/cloudpilot-ai/karpenter-provider-alibabacloud/pkg/cache"
@@ -58,6 +59,7 @@ type Provider interface {
 	LivenessProbe(*http.Request) error
 	GetSupportedImages(string) ([]Image, error)
 	FeatureFlags() FeatureFlags
+	DefaultOverhead(corev1.ResourceList) cloudprovider.InstanceTypeOverhead
 }
 
 func NewClusterProvider(ctx context.Context, ackClient *ackclient.Client, region string) Provider {

pkg/providers/instancetype/instancetype.go

@@ -195,7 +195,7 @@ func (p *DefaultProvider) List(ctx context.Context, kc *v1alpha1.KubeletConfigur
 		// so that Karpenter is able to cache the set of InstanceTypes based on values that alter the set of instance types
 		// !!! Important !!!
 		offers := p.createOfferings(ctx, *i.InstanceTypeId, zoneData)
-		return NewInstanceType(ctx, i, kc, p.region, nodeClass.Spec.SystemDisk, offers, clusterCNI)
+		return NewInstanceType(ctx, i, kc, p.region, nodeClass.Spec.SystemDisk, offers, clusterCNI, p.clusterProvider)
 	})
 
 	// Filter out nil values

pkg/providers/instancetype/types.go

@@ -39,6 +39,11 @@ import (
 	"github.com/cloudpilot-ai/karpenter-provider-alibabacloud/pkg/providers/imagefamily"
 )
 
+const (
+	MemoryAvailable = "memory.available"
+	NodeFSAvailable = "nodefs.available"
+)
+
 var (
 	instanceTypeScheme = regexp.MustCompile(`^ecs\.([a-z]+)(\-[0-9]+tb)?([0-9]+).*`)
 )
@@ -59,55 +64,10 @@ type ZoneData struct {
 	SpotAvailable bool
 }
 
-func calculateResourceOverhead(pods, cpuM, memoryMi int64) corev1.ResourceList {
-	// referring to: https://help.aliyun.com/zh/ack/ack-managed-and-ack-dedicated/user-guide/resource-reservation-policy#0f5ffe176df7q
-	// CPU overhead calculation
-	cpuOverHead := calculateCPUOverhead(cpuM)
-
-	// TODO: In a real environment, the formula does not produce accurate results,
-	// consistently yielding values that are 200MiB larger than expected.
-	// Memory overhead: min(11*pods + 255, memoryMi*0.25)
-	memoryOverHead := int64(math.Min(float64(11*pods+255), float64(memoryMi)*0.25)) + 200
-
-	return corev1.ResourceList{
-		corev1.ResourceCPU:    *resource.NewMilliQuantity(cpuOverHead, resource.DecimalSI),
-		corev1.ResourceMemory: *resources.Quantity(fmt.Sprintf("%dMi", memoryOverHead)),
-	}
-}
-
-// thresholds defines CPU overhead thresholds and their corresponding percentages
-var thresholds = [...]struct {
-	cores    int64
-	overhead float64
-}{
-	{1000, 0.06},
-	{2000, 0.01},
-	{3000, 0.005},
-	{4000, 0.005},
-}
-
-func calculateCPUOverhead(cpuM int64) int64 {
-	var cpuOverHead int64
-
-	// Calculate overhead for each threshold
-	for _, t := range thresholds {
-		if cpuM >= t.cores {
-			cpuOverHead += int64(1000 * t.overhead)
-		}
-	}
-
-	// Additional overhead for CPU > 4 cores (0.25%)
-	if cpuM > 4000 {
-		cpuOverHead += int64(float64(cpuM-4000) * 0.0025)
-	}
-
-	return cpuOverHead
-}
-
 func NewInstanceType(ctx context.Context,
 	info *ecsclient.DescribeInstanceTypesResponseBodyInstanceTypesInstanceType,
 	kc *v1alpha1.KubeletConfiguration, region string, systemDisk *v1alpha1.SystemDisk,
-	offerings cloudprovider.Offerings, clusterCNI string) *cloudprovider.InstanceType {
+	offerings cloudprovider.Offerings, clusterCNI string, cluster cluster.Provider) *cloudprovider.InstanceType {
 	if offerings == nil {
 		return nil
 	}
@@ -117,16 +77,10 @@ func NewInstanceType(ctx context.Context,
 		Requirements: computeRequirements(info, offerings, region),
 		Offerings:    offerings,
 		Capacity:     computeCapacity(ctx, info, kc.MaxPods, kc.PodsPerCore, systemDisk, clusterCNI),
-		Overhead: &cloudprovider.InstanceTypeOverhead{
-			KubeReserved:      corev1.ResourceList{},
-			SystemReserved:    corev1.ResourceList{},
-			EvictionThreshold: corev1.ResourceList{},
-		},
 	}
 
 	// Follow KubeReserved/SystemReserved/EvictionThreshold will be merged, so we can set only one overhead totally
-	it.Overhead.KubeReserved = calculateResourceOverhead(it.Capacity.Pods().Value(),
-		it.Capacity.Cpu().MilliValue(), extractMemory(info).Value()/MiBByteRatio)
+	it.Overhead = computeOverhead(cluster, it.Capacity, kc)
 	if it.Requirements.Compatible(scheduling.NewRequirements(scheduling.NewRequirement(corev1.LabelOSStable, corev1.NodeSelectorOpIn, string(corev1.Windows)))) == nil {
 		it.Capacity[v1alpha1.ResourcePrivateIPv4Address] = *privateIPv4Address(info)
 	}
@@ -353,3 +307,42 @@ func privateIPv4Address(info *ecsclient.DescribeInstanceTypesResponseBodyInstanc
 func getInstanceBandwidth(info *ecsclient.DescribeInstanceTypesResponseBodyInstanceTypesInstanceType) int32 {
 	return max(lo.FromPtr(info.InstanceBandwidthRx), lo.FromPtr(info.InstanceBandwidthTx))
 }
+
+func computeOverhead(cluster cluster.Provider, capacity corev1.ResourceList, kubeletConfig *v1alpha1.KubeletConfiguration) *cloudprovider.InstanceTypeOverhead {
+	overhead := &cloudprovider.InstanceTypeOverhead{
+		KubeReserved:   corev1.ResourceList{},
+		SystemReserved: corev1.ResourceList{},
+		// ref: https://github.com/kubernetes/kubernetes/blob/master/pkg/kubelet/eviction/defaults_linux.go
+		EvictionThreshold: corev1.ResourceList{
+			corev1.ResourceMemory:           resource.MustParse("100Mi"),
+			corev1.ResourceEphemeralStorage: computeEvictionSignal(*capacity.StorageEphemeral(), "10%"),
+		},
+	}
+
+	defaultOverhead := cluster.DefaultOverhead(capacity)
+	if defaultOverhead.KubeReserved != nil {
+		overhead.KubeReserved = lo.Assign(overhead.KubeReserved, defaultOverhead.KubeReserved)
+	}
+	if defaultOverhead.SystemReserved != nil {
+		overhead.SystemReserved = lo.Assign(overhead.SystemReserved, defaultOverhead.SystemReserved)
+	}
+	if defaultOverhead.EvictionThreshold != nil {
+		overhead.EvictionThreshold = lo.Assign(overhead.EvictionThreshold, defaultOverhead.EvictionThreshold)
+	}
+
+	overhead.KubeReserved = lo.Assign(overhead.KubeReserved, lo.MapEntries(kubeletConfig.KubeReserved, func(k string, v string) (corev1.ResourceName, resource.Quantity) {
+		return corev1.ResourceName(k), resource.MustParse(v)
+	}))
+	overhead.SystemReserved = lo.Assign(overhead.SystemReserved, lo.MapEntries(kubeletConfig.SystemReserved, func(k string, v string) (corev1.ResourceName, resource.Quantity) {
+		return corev1.ResourceName(k), resource.MustParse(v)
+	}))
+	if kubeletConfig.EvictionHard != nil {
+		if v, ok := kubeletConfig.EvictionHard[MemoryAvailable]; ok {
+			overhead.EvictionThreshold[corev1.ResourceMemory] = computeEvictionSignal(*capacity.Memory(), v)
+		}
+		if v, ok := kubeletConfig.EvictionHard[NodeFSAvailable]; ok {
+			overhead.EvictionThreshold[corev1.ResourceEphemeralStorage] = computeEvictionSignal(*capacity.StorageEphemeral(), v)
+		}
+	}
+	return overhead
+}