Merge pull request #42 from helen-frank/instancetype-list

feat: instancetype add list

Author: jwcesign

go.mod

@@ -117,7 +117,7 @@ require (
 	k8s.io/klog v1.0.0 // indirect
 	k8s.io/klog/v2 v2.130.1
 	k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340 // indirect
-	knative.dev/pkg v0.0.0-20230712131115-7051d301e7f4 // indirect
+	knative.dev/pkg v0.0.0-20230712131115-7051d301e7f4
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
 )

pkg/cache/cache.go

@@ -0,0 +1,35 @@
+/*
+Copyright 2024 The CloudPilot AI Authors.
+
+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.
+*/
+
+package cache
+
+import "time"
+
+const (
+	// KubernetesVersionTTL is the time before the detected Kubernetes version is removed from cache,
+	// to be re-detected next time it is needed.
+	KubernetesVersionTTL = 15 * time.Minute
+	// UnavailableOfferingsTTL is the time before offerings that were marked as unavailable
+	// are removed from the cache and are available for launch again
+	UnavailableOfferingsTTL = 3 * time.Minute
+
+	// DefaultCleanupInterval triggers cache cleanup (lazy eviction) at this interval.
+	DefaultCleanupInterval = 1 * time.Minute
+	// UnavailableOfferingsCleanupInterval triggers cache cleanup (lazy eviction) at this interval.
+	// We drop the cleanup interval down for the ICE cache to get quicker reactivity to offerings
+	// that become available after they get evicted from the cache
+	UnavailableOfferingsCleanupInterval = time.Second * 10
+)

pkg/cache/unavailableofferings.go

@@ -0,0 +1,96 @@
+/*
+Copyright 2024 The CloudPilot AI Authors.
+
+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.
+*/
+
+package cache
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"time"
+
+	"github.com/patrickmn/go-cache"
+	"knative.dev/pkg/logging"
+	"sigs.k8s.io/karpenter/pkg/apis/v1beta1"
+)
+
+var (
+	spotKey = key("", "", v1beta1.CapacityTypeSpot)
+)
+
+// UnavailableOfferings stores any offerings that return ICE (insufficient capacity errors) when
+// attempting to launch the capacity. These offerings are ignored as long as they are in the cache on
+// GetInstanceTypes responses
+type UnavailableOfferings struct {
+	// key: <capacityType>:<instanceType>:<zone>, value: struct{}{}
+	cache  *cache.Cache
+	SeqNum uint64
+}
+
+func NewUnavailableOfferingsWithCache(c *cache.Cache) *UnavailableOfferings {
+	uo := &UnavailableOfferings{
+		cache:  c,
+		SeqNum: 0,
+	}
+	uo.cache.OnEvicted(func(_ string, _ interface{}) {
+		atomic.AddUint64(&uo.SeqNum, 1)
+	})
+	return uo
+}
+
+func NewUnavailableOfferings() *UnavailableOfferings {
+	return NewUnavailableOfferingsWithCache(
+		cache.New(UnavailableOfferingsTTL, UnavailableOfferingsCleanupInterval))
+}
+
+// IsUnavailable returns true if the offering appears in the cache
+func (u *UnavailableOfferings) IsUnavailable(instanceType, zone, capacityType string) bool {
+	if capacityType == v1beta1.CapacityTypeSpot {
+		if _, found := u.cache.Get(spotKey); found {
+			return true
+		}
+	}
+	_, found := u.cache.Get(key(instanceType, zone, capacityType))
+	return found
+}
+
+// MarkUnavailableWithTTL allows us to mark an offering unavailable with a custom TTL
+func (u *UnavailableOfferings) MarkUnavailableWithTTL(ctx context.Context, unavailableReason, instanceType, zone, capacityType string, ttl time.Duration) {
+	// even if the key is already in the cache, we still need to call Set to extend the cached entry's TTL
+	logging.FromContext(ctx).With(
+		"unavailable", unavailableReason,
+		"instance-type", instanceType,
+		"zone", zone,
+		"capacity-type", capacityType,
+		"ttl", ttl).Debugf("removing offering from offerings")
+	u.cache.Set(key(instanceType, zone, capacityType), struct{}{}, ttl)
+	atomic.AddUint64(&u.SeqNum, 1)
+}
+
+// MarkUnavailable communicates recently observed temporary capacity shortages in the provided offerings
+func (u *UnavailableOfferings) MarkUnavailable(ctx context.Context, unavailableReason, instanceType, zone, capacityType string) {
+	u.MarkUnavailableWithTTL(ctx, unavailableReason, instanceType, zone, capacityType, UnavailableOfferingsTTL)
+}
+
+func (u *UnavailableOfferings) Flush() {
+	u.cache.Flush()
+	atomic.AddUint64(&u.SeqNum, 1)
+}
+
+// key returns the cache key for all offerings in the cache
+func key(instanceType string, zone string, capacityType string) string {
+	return fmt.Sprintf("%s:%s:%s", capacityType, instanceType, zone)
+}

pkg/cache/unavailableofferings_test.go

@@ -0,0 +1,60 @@
+/*
+Copyright 2024 The CloudPilot AI Authors.
+
+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.
+*/
+
+package cache
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/patrickmn/go-cache"
+)
+
+func TestUnavailableOfferings(t *testing.T) {
+	// create a new cache with a short TTL
+	c := cache.New(time.Second, time.Second)
+	u := NewUnavailableOfferingsWithCache(c)
+
+	// test that an offering is not marked as unavailable initially
+	if u.IsUnavailable("NV16as_v4", "westus", "spot") {
+		t.Error("Offering should not be marked as unavailable initially")
+	}
+
+	// mark the offering as unavailable
+	u.MarkUnavailableWithTTL(context.TODO(), "test reason", "NV16as_v4", "westus", "spot", time.Second)
+
+	// test that the offering is now marked as unavailable
+	if !u.IsUnavailable("NV16as_v4", "westus", "spot") {
+		t.Error("Offering should be marked as unavailable after being marked as such")
+	}
+
+	// wait for the cache entry to expire
+	time.Sleep(time.Second)
+
+	// test that the offering is no longer marked as unavailable
+	if u.IsUnavailable("NV16as_v4", "westus", "spot") {
+		t.Error("Offering should not be marked as unavailable after cache entry has expired")
+	}
+}
+
+func TestUnavailableOfferings_KeyGeneration(t *testing.T) {
+	expectedKey := "spot:NV16as_v4:westus"
+	key := key("NV16as_v4", "westus", "spot")
+	if key != expectedKey {
+		t.Errorf("Expected key to be %s, but got %s", expectedKey, key)
+	}
+}

pkg/operator/options/options.go

@@ -25,6 +25,8 @@ import (
 
 	coreoptions "sigs.k8s.io/karpenter/pkg/operator/options"
 	"sigs.k8s.io/karpenter/pkg/utils/env"
+
+	"github.com/cloudpilot-ai/karpenter-provider-alicloud/pkg/utils"
 )
 
 func init() {
@@ -34,15 +36,17 @@ func init() {
 type optionsKey struct{}
 
 type Options struct {
-	ClusterCABundle string
-	ClusterName     string
-	ClusterEndpoint string
+	ClusterCABundle         string
+	ClusterName             string
+	ClusterEndpoint         string
+	VMMemoryOverheadPercent float64
 }
 
 func (o *Options) AddFlags(fs *coreoptions.FlagSet) {
 	fs.StringVar(&o.ClusterCABundle, "cluster-ca-bundle", env.WithDefaultString("CLUSTER_CA_BUNDLE", ""), "Cluster CA bundle for nodes to use for TLS connections with the API server. If not set, this is taken from the controller's TLS configuration.")
 	fs.StringVar(&o.ClusterName, "cluster-name", env.WithDefaultString("CLUSTER_NAME", ""), "[REQUIRED] The kubernetes cluster name for resource discovery.")
 	fs.StringVar(&o.ClusterEndpoint, "cluster-endpoint", env.WithDefaultString("CLUSTER_ENDPOINT", ""), "The external kubernetes cluster endpoint for new nodes to connect with. If not specified, will discover the cluster endpoint using DescribeCluster API.")
+	fs.Float64Var(&o.VMMemoryOverheadPercent, "vm-memory-overhead-percent", utils.WithDefaultFloat64("VM_MEMORY_OVERHEAD_PERCENT", 0.075), "The VM memory overhead as a percent that will be subtracted from the total memory for all instance types.")
 }
 
 func (o *Options) Parse(fs *coreoptions.FlagSet, args ...string) error {