Merge pull request #56 from jwcesign/main

feat: select the cheapest zone and add user data

Author: helen

examples/workload/inflate.yaml

@@ -3,7 +3,7 @@ kind: Deployment
 metadata:
   name: inflate
 spec:
-  replicas: 2
+  replicas: 1
   selector:
     matchLabels:
       app: inflate
@@ -12,6 +12,13 @@ spec:
       labels:
         app: inflate
     spec:
+      affinity:
+        nodeAffinity:
+          requiredDuringSchedulingIgnoredDuringExecution:
+            nodeSelectorTerms:
+              - matchExpressions:
+                  - key: karpenter.sh/capacity-type
+                    operator: Exists
       securityContext:
         runAsUser: 1000
         runAsGroup: 3000
@@ -21,7 +28,7 @@ spec:
         name: inflate
         resources:
           requests:
-            cpu: "4"
-            memory: 4Gi
+            cpu: 250m
+            memory: 250Mi
         securityContext:
           allowPrivilegeEscalation: false

go.mod

@@ -3,6 +3,7 @@ module github.com/cloudpilot-ai/karpenter-provider-alicloud
 go 1.23
 
 require (
+	github.com/alibabacloud-go/cs-20151215/v5 v5.7.9
 	github.com/alibabacloud-go/darabonba-openapi/v2 v2.0.10
 	github.com/alibabacloud-go/ecs-20140526/v4 v4.26.4
 	github.com/alibabacloud-go/tea v1.2.2

go.sum

@@ -50,6 +50,8 @@ github.com/alibabacloud-go/alibabacloud-gateway-pop v0.0.6/go.mod h1:4EUIoxs/do2
 github.com/alibabacloud-go/alibabacloud-gateway-spi v0.0.4/go.mod h1:sCavSAvdzOjul4cEqeVtvlSaSScfNsTQ+46HwlTL1hc=
 github.com/alibabacloud-go/alibabacloud-gateway-spi v0.0.5 h1:zE8vH9C7JiZLNJJQ5OwjU9mSi4T9ef9u3BURT6LCLC8=
 github.com/alibabacloud-go/alibabacloud-gateway-spi v0.0.5/go.mod h1:tWnyE9AjF8J8qqLk645oUmVUnFybApTQWklQmi5tY6g=
+github.com/alibabacloud-go/cs-20151215/v5 v5.7.9 h1:zQdpRgOhpYQVAXfpywnpz3vMxP+bLUoi2QfaFbg3WU4=
+github.com/alibabacloud-go/cs-20151215/v5 v5.7.9/go.mod h1:1pR8T0PY7DtF/t+yrOnG2aXj/TRGm1KYK7REaKyvPvM=
 github.com/alibabacloud-go/darabonba-array v0.1.0 h1:vR8s7b1fWAQIjEjWnuF0JiKsCvclSRTfDzZHTYqfufY=
 github.com/alibabacloud-go/darabonba-array v0.1.0/go.mod h1:BLKxr0brnggqOJPqT09DFJ8g3fsDshapUD3C3aOEFaI=
 github.com/alibabacloud-go/darabonba-encode-util v0.0.2 h1:1uJGrbsGEVqWcWxrS9MyC2NG0Ax+GpOM5gtupki31XE=

pkg/operator/operator.go

@@ -20,13 +20,16 @@ import (
 	"context"
 	"os"
 
+	ackclient "github.com/alibabacloud-go/cs-20151215/v5/client"
 	ecs "github.com/alibabacloud-go/ecs-20140526/v4/client"
 	vpc "github.com/alibabacloud-go/vpc-20160428/v6/client"
 	"github.com/patrickmn/go-cache"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/karpenter/pkg/operator"
 
 	alicache "github.com/cloudpilot-ai/karpenter-provider-alicloud/pkg/cache"
+	"github.com/cloudpilot-ai/karpenter-provider-alicloud/pkg/operator/options"
+	"github.com/cloudpilot-ai/karpenter-provider-alicloud/pkg/providers/ack"
 	"github.com/cloudpilot-ai/karpenter-provider-alicloud/pkg/providers/imagefamily"
 	"github.com/cloudpilot-ai/karpenter-provider-alicloud/pkg/providers/instance"
 	"github.com/cloudpilot-ai/karpenter-provider-alicloud/pkg/providers/instancetype"
@@ -67,6 +70,12 @@ func NewOperator(ctx context.Context, operator *operator.Operator) (context.Cont
 		log.FromContext(ctx).Error(err, "Failed to create VPC client")
 		os.Exit(1)
 	}
+	ackClient, err := ackclient.NewClient(clientConfig)
+	if err != nil {
+		log.FromContext(ctx).Error(err, "Failed to create ACK client")
+		os.Exit(1)
+	}
+	clusterID := options.FromContext(ctx).ClusterID
 	region := *ecsClient.RegionId
 
 	pricingProvider, err := pricing.NewDefaultProvider(ctx, region)
@@ -80,14 +89,14 @@ func NewOperator(ctx context.Context, operator *operator.Operator) (context.Cont
 	securityGroupProvider := securitygroup.NewDefaultProvider(region, ecsClient, cache.New(alicache.DefaultTTL, alicache.DefaultCleanupInterval))
 	imageProvider := imagefamily.NewDefaultProvider(region, ecsClient, cache.New(alicache.DefaultTTL, alicache.DefaultCleanupInterval))
 	imageResolver := imagefamily.NewDefaultResolver(region, ecsClient, cache.New(alicache.InstanceTypeAvailableDiskTTL, alicache.DefaultCleanupInterval))
+	ackProvider := ack.NewDefaultProvider(clusterID, ackClient)
 
 	instanceProvider := instance.NewDefaultProvider(
-		ctx,
 		region,
-		"",
 		ecsClient,
 		imageResolver,
 		vSwitchProvider,
+		ackProvider,
 	)
 
 	unavailableOfferingsCache := alicache.NewUnavailableOfferings()

pkg/operator/options/options.go

@@ -36,16 +36,14 @@ func init() {
 type optionsKey struct{}
 
 type Options struct {
-	ClusterCABundle         string
 	ClusterName             string
-	ClusterEndpoint         string
+	ClusterID               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.StringVar(&o.ClusterID, "cluster-id", env.WithDefaultString("CLUSTER_ID", ""), "The external kubernetes cluster id for new nodes to connect with.")
 	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.")
 }
 

pkg/operator/options/options_validation.go

@@ -18,34 +18,22 @@ package options
 
 import (
 	"fmt"
-	"net/url"
 
 	"go.uber.org/multierr"
 )
 
 func (o Options) Validate() error {
 	return multierr.Combine(
-		o.validateEndpoint(),
 		o.validateRequiredFields(),
 	)
 }
 
-func (o Options) validateEndpoint() error {
-	if o.ClusterEndpoint == "" {
-		return nil
-	}
-	endpoint, err := url.Parse(o.ClusterEndpoint)
-	// url.Parse() will accept a lot of input without error; make
-	// sure it's a real URL
-	if err != nil || !endpoint.IsAbs() || endpoint.Hostname() == "" {
-		return fmt.Errorf("%q is not a valid cluster-endpoint URL", o.ClusterEndpoint)
-	}
-	return nil
-}
-
 func (o Options) validateRequiredFields() error {
 	if o.ClusterName == "" {
 		return fmt.Errorf("missing field, cluster-name")
 	}
+	if o.ClusterID == "" {
+		return fmt.Errorf("missing field, cluster-id")
+	}
 	return nil
 }

pkg/providers/ack/ack.go

@@ -0,0 +1,83 @@
+/*
+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 ack
+
+import (
+	"context"
+	"encoding/base64"
+	"errors"
+	"fmt"
+	"regexp"
+	"strings"
+
+	ackclient "github.com/alibabacloud-go/cs-20151215/v5/client"
+	"github.com/alibabacloud-go/tea/tea"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+type Provider interface {
+	GetNodeRegisterScript(ctx context.Context, labels map[string]string) (string, error)
+}
+
+type DefaultProvider struct {
+	clusterID string
+	ackClient *ackclient.Client
+}
+
+func NewDefaultProvider(clusterID string, ackClient *ackclient.Client) *DefaultProvider {
+	return &DefaultProvider{
+		clusterID: clusterID,
+		ackClient: ackClient,
+	}
+}
+
+func (p *DefaultProvider) GetNodeRegisterScript(ctx context.Context, labels map[string]string) (string, error) {
+	reqPara := &ackclient.DescribeClusterAttachScriptsRequest{
+		KeepInstanceName: tea.Bool(true),
+	}
+
+	// TODO: Build a cache to store this
+	resp, err := p.ackClient.DescribeClusterAttachScripts(tea.String(p.clusterID), reqPara)
+	if err != nil {
+		log.FromContext(ctx).Error(err, "Failed to get node registration script")
+		return "", err
+	}
+	s := tea.StringValue(resp.Body)
+	if s == "" {
+		err := errors.New("empty node registration script")
+		log.FromContext(ctx).Error(err, "")
+		return "", err
+	}
+
+	return p.resolveUserData(s, labels), nil
+}
+
+func (p *DefaultProvider) resolveUserData(respStr string, labels map[string]string) string {
+	cleanupStr := strings.ReplaceAll(respStr, "\r\n", "")
+
+	labelsFormated := fmt.Sprintf("ack.aliyun.com=%s", p.clusterID)
+	for labelKey, labelValue := range labels {
+		labelsFormated = fmt.Sprintf("%s,%s=%s", labelsFormated, labelKey, labelValue)
+	}
+
+	re := regexp.MustCompile(`--labels\s+\S+`)
+	updatedCommand := re.ReplaceAllString(cleanupStr, "--labels "+labelsFormated)
+
+	finalScript := fmt.Sprintf("#!/bin/bash\n\n%s", updatedCommand)
+
+	return base64.StdEncoding.EncodeToString([]byte(finalScript))
+}

pkg/providers/instance/instance.go

@@ -37,6 +37,7 @@ import (
 
 	"github.com/cloudpilot-ai/karpenter-provider-alicloud/pkg/apis/v1alpha1"
 	"github.com/cloudpilot-ai/karpenter-provider-alicloud/pkg/operator/options"
+	"github.com/cloudpilot-ai/karpenter-provider-alicloud/pkg/providers/ack"
 	"github.com/cloudpilot-ai/karpenter-provider-alicloud/pkg/providers/imagefamily"
 	"github.com/cloudpilot-ai/karpenter-provider-alicloud/pkg/providers/vswitch"
 	"github.com/cloudpilot-ai/karpenter-provider-alicloud/pkg/utils/alierrors"
@@ -57,26 +58,24 @@ type Provider interface {
 }
 
 type DefaultProvider struct {
-	ecsClient       *ecsclient.Client
-	region          string
-	clusterEndpoint string
+	ecsClient *ecsclient.Client
+	region    string
 
-	imageFamily imagefamily.Resolver
-
-	vSwitchProvider vswitch.Provider
+	imageFamilyResolver imagefamily.Resolver
+	vSwitchProvider     vswitch.Provider
+	ackProvider         ack.Provider
 }
 
-func NewDefaultProvider(ctx context.Context, region, clusterEndpoint string, ecsClient *ecsclient.Client,
-	imageFamily imagefamily.Resolver,
-	vSwitchProvider vswitch.Provider) *DefaultProvider {
+func NewDefaultProvider(region string, ecsClient *ecsclient.Client,
+	imageFamilyResolver imagefamily.Resolver, vSwitchProvider vswitch.Provider,
+	ackProvider ack.Provider) *DefaultProvider {
 	return &DefaultProvider{
-		ecsClient:       ecsClient,
-		region:          region,
-		clusterEndpoint: clusterEndpoint,
-
-		imageFamily: imageFamily,
+		ecsClient: ecsClient,
+		region:    region,
 
-		vSwitchProvider: vSwitchProvider,
+		imageFamilyResolver: imageFamilyResolver,
+		vSwitchProvider:     vSwitchProvider,
+		ackProvider:         ackProvider,
 	}
 }
 
@@ -387,7 +386,7 @@ func (p *DefaultProvider) getProvisioningGroup(ctx context.Context, nodeClass *v
 			break
 		}
 
-		vSwitchID := p.getVSwitchID(launchtemplate.InstanceTypes[i], zonalVSwitchs, requirements)
+		vSwitchID := p.getVSwitchID(launchtemplate.InstanceTypes[i], zonalVSwitchs, requirements, capacityType)
 		if vSwitchID == "" {
 			return nil, errors.New("vSwitchID not found")
 		}
@@ -409,6 +408,13 @@ func (p *DefaultProvider) getProvisioningGroup(ctx context.Context, nodeClass *v
 		})
 	}
 
+	labels := lo.Assign(nodeClaim.Labels, map[string]string{karpv1.CapacityTypeLabelKey: capacityType})
+	userData, err := p.ackProvider.GetNodeRegisterScript(ctx, labels)
+	if err != nil {
+		log.FromContext(ctx).Error(err, "Failed to resolve user data for node")
+		return nil, err
+	}
+
 	createAutoProvisioningGroupRequest := &ecsclient.CreateAutoProvisioningGroupRequest{
 		RegionId:                        tea.String(p.region),
 		TotalTargetCapacity:             tea.String("1"),
@@ -420,6 +426,7 @@ func (p *DefaultProvider) getProvisioningGroup(ctx context.Context, nodeClass *v
 		LaunchConfiguration: &ecsclient.CreateAutoProvisioningGroupRequestLaunchConfiguration{
 			ImageId:          tea.String(launchtemplate.ImageID),
 			SecurityGroupIds: launchtemplate.SecurityGroupIds,
+			UserData:         tea.String(userData),
 
 			// TODO: AutoProvisioningGroup is not compatible with SecurityGroupIds, waiting for Aliyun developers to fix it,
 			// so here we only take the first one.
@@ -456,7 +463,12 @@ func (p *DefaultProvider) checkODFallback(nodeClaim *karpv1.NodeClaim, instanceT
 	return nil
 }
 
-func (p *DefaultProvider) getVSwitchID(instanceType *cloudprovider.InstanceType, zonalVSwitchs map[string]*vswitch.VSwitch, reqs scheduling.Requirements) string {
+func (p *DefaultProvider) getVSwitchID(instanceType *cloudprovider.InstanceType,
+	zonalVSwitchs map[string]*vswitch.VSwitch, reqs scheduling.Requirements, capacityType string) string {
+	cheapestVSwitchID := ""
+	cheapestPrice := math.MaxFloat64
+
+	// For different AZ, the spot price may differ. So we need to get the cheapest vSwitch in the zone
 	for i := range instanceType.Offerings {
 		if reqs.Compatible(instanceType.Offerings[i].Requirements, scheduling.AllowUndefinedWellKnownLabels) != nil {
 			continue
@@ -465,9 +477,17 @@ func (p *DefaultProvider) getVSwitchID(instanceType *cloudprovider.InstanceType,
 		if !ok {
 			continue
 		}
-		return vswitch.ID
+		if capacityType == karpv1.CapacityTypeOnDemand {
+			return vswitch.ID
+		}
+
+		if instanceType.Offerings[i].Price < cheapestPrice {
+			cheapestVSwitchID = vswitch.ID
+			cheapestPrice = instanceType.Offerings[i].Price
+		}
 	}
-	return ""
+
+	return cheapestVSwitchID
 }
 
 type LaunchTemplate struct {
@@ -477,12 +497,14 @@ type LaunchTemplate struct {
 	SystemDisk       *v1alpha1.SystemDisk
 }
 
-func (p *DefaultProvider) EnsureAll(ctx context.Context, nodeClass *v1alpha1.ECSNodeClass, nodeClaim *karpv1.NodeClaim, instanceTypes []*cloudprovider.InstanceType, capacityType string, tags map[string]string) ([]*LaunchTemplate, error) {
+func (p *DefaultProvider) EnsureAll(ctx context.Context, nodeClass *v1alpha1.ECSNodeClass,
+	nodeClaim *karpv1.NodeClaim, instanceTypes []*cloudprovider.InstanceType,
+	capacityType string, tags map[string]string) ([]*LaunchTemplate, error) {
 	imageOptions, err := p.resolveImageOptions(ctx, nodeClass, lo.Assign(nodeClaim.Labels, map[string]string{karpv1.CapacityTypeLabelKey: capacityType}), tags)
 	if err != nil {
 		return nil, err
 	}
-	resolvedLaunchTemplates, err := p.imageFamily.Resolve(ctx, nodeClass, nodeClaim, instanceTypes, capacityType, imageOptions)
+	resolvedLaunchTemplates, err := p.imageFamilyResolver.Resolve(ctx, nodeClass, nodeClaim, instanceTypes, capacityType, imageOptions)
 	if err != nil {
 		return nil, err
 	}
@@ -526,11 +548,10 @@ func (p *DefaultProvider) resolveImageOptions(ctx context.Context, nodeClass *v1
 		return nil, fmt.Errorf("no security groups are present in the status")
 	}
 	return &imagefamily.Options{
-		ClusterName:     options.FromContext(ctx).ClusterName,
-		ClusterEndpoint: p.clusterEndpoint,
-		SecurityGroups:  nodeClass.Status.SecurityGroups,
-		Tags:            tags,
-		Labels:          labels,
-		NodeClassName:   nodeClass.Name,
+		ClusterName:    options.FromContext(ctx).ClusterName,
+		SecurityGroups: nodeClass.Status.SecurityGroups,
+		Tags:           tags,
+		Labels:         labels,
+		NodeClassName:  nodeClass.Name,
 	}, nil
 }