added cache for instances

update error log content

added a boilerplate header

change paramater value to specStr

remove blank line

check if specified the node groups

change autoScalingGroups to autoScalingGroupCache

Author: wangshiqi308

cluster-autoscaler/cloudprovider/huaweicloud/huaweicloud_auto_scaling_groups.go

@@ -0,0 +1,97 @@
+/*
+Copyright 2020 The Kubernetes 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 huaweicloud
+
+import (
+	"sync"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/klog/v2"
+)
+
+type autoScalingGroupCache struct {
+	registeredAsgs           []*AutoScalingGroup
+	instanceToAsg            map[string]*AutoScalingGroup
+	cacheMutex               sync.Mutex
+	instancesNotInManagedAsg map[string]struct{}
+}
+
+func newAutoScalingGroupCache() *autoScalingGroupCache {
+	registry := &autoScalingGroupCache{
+		instanceToAsg:            make(map[string]*AutoScalingGroup),
+		instancesNotInManagedAsg: make(map[string]struct{}),
+	}
+
+	return registry
+}
+
+// Register asg in HuaweiCloud Manager.
+func (asg *autoScalingGroupCache) Register(autoScalingGroup *AutoScalingGroup) {
+	asg.cacheMutex.Lock()
+	defer asg.cacheMutex.Unlock()
+
+	asg.registeredAsgs = append(asg.registeredAsgs, autoScalingGroup)
+}
+
+// FindForInstance returns AsgConfig of the given Instance
+func (asg *autoScalingGroupCache) FindForInstance(instanceId string, csm *cloudServiceManager) (*AutoScalingGroup, error) {
+	asg.cacheMutex.Lock()
+	defer asg.cacheMutex.Unlock()
+	if config, found := asg.instanceToAsg[instanceId]; found {
+		return config, nil
+	}
+	if _, found := asg.instancesNotInManagedAsg[instanceId]; found {
+		return nil, nil
+	}
+	if err := asg.regenerateCache(csm); err != nil {
+		return nil, err
+	}
+	if config, found := asg.instanceToAsg[instanceId]; found {
+		return config, nil
+	}
+	// instance does not belong to any configured ASG
+	asg.instancesNotInManagedAsg[instanceId] = struct{}{}
+	return nil, nil
+}
+
+func (asg *autoScalingGroupCache) generateCache(csm *cloudServiceManager) {
+	go wait.Forever(func() {
+		asg.cacheMutex.Lock()
+		defer asg.cacheMutex.Unlock()
+		if err := asg.regenerateCache(csm); err != nil {
+			klog.Errorf("failed to do regenerating ASG cache,because of %s", err.Error())
+		}
+	}, time.Hour)
+}
+
+func (asg *autoScalingGroupCache) regenerateCache(csm *cloudServiceManager) error {
+	newCache := make(map[string]*AutoScalingGroup)
+
+	for _, registeredAsg := range asg.registeredAsgs {
+		instances, err := csm.GetInstances(registeredAsg.groupID)
+		if err != nil {
+			return err
+		}
+		for _, instance := range instances {
+			newCache[instance.Id] = registeredAsg
+		}
+	}
+
+	asg.instanceToAsg = newCache
+	return nil
+}

cluster-autoscaler/cloudprovider/huaweicloud/huaweicloud_cloud_provider.go

@@ -24,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
 	"k8s.io/autoscaler/cluster-autoscaler/config"
+	"k8s.io/autoscaler/cluster-autoscaler/config/dynamic"
 	"k8s.io/autoscaler/cluster-autoscaler/utils/errors"
 	klog "k8s.io/klog/v2"
 )
@@ -54,6 +55,11 @@ type huaweicloudCloudProvider struct {
 }
 
 func newCloudProvider(opts config.AutoscalingOptions, do cloudprovider.NodeGroupDiscoveryOptions, rl *cloudprovider.ResourceLimiter) *huaweicloudCloudProvider {
+	if do.AutoDiscoverySpecified() {
+		klog.Errorf("only support static discovery scaling group in huaweicloud for now")
+		return nil
+	}
+
 	cloudConfig, err := readConf(opts.CloudConfig)
 	if err != nil {
 		klog.Errorf("failed to read cloud configuration. error: %v", err)
@@ -65,17 +71,24 @@ func newCloudProvider(opts config.AutoscalingOptions, do cloudprovider.NodeGroup
 	}
 
 	csm := newCloudServiceManager(cloudConfig)
-	sgs, err := csm.ListScalingGroups()
-	if err != nil {
-		klog.Errorf("failed to list scaling groups. error: %v", err)
-		return nil
-	}
 
-	return &huaweicloudCloudProvider{
+	hcp := &huaweicloudCloudProvider{
 		cloudServiceManager: csm,
 		resourceLimiter:     rl,
-		autoScalingGroup:    sgs,
+		autoScalingGroup:    make([]AutoScalingGroup, 0),
+	}
+
+	if len(do.NodeGroupSpecs) <= 0 {
+		klog.Error("no auto scaling group specified")
+		return nil
+	}
+	err = hcp.buildAsgs(do.NodeGroupSpecs)
+	if err != nil {
+		klog.Errorf("failed to build auto scaling groups. error: %v", err)
+		return nil
 	}
+
+	return hcp
 }
 
 // Name returns the name of the cloud provider.
@@ -111,25 +124,7 @@ func (hcp *huaweicloudCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudpr
 		return nil, fmt.Errorf("provider id missing from node: %s", node.Name)
 	}
 
-	hcp.lock.RLock()
-	defer hcp.lock.RUnlock()
-
-	for i := range hcp.autoScalingGroup {
-		instances, err := hcp.autoScalingGroup[i].Nodes()
-		if err != nil {
-			klog.Warningf("failed to list instances from scaling group: %s, error: %v", hcp.autoScalingGroup[i].groupName, err)
-			return nil, err
-		}
-
-		for j := range instances {
-			if instanceID == instances[j].Id {
-				pinnedGroup := hcp.autoScalingGroup[i]
-				return &pinnedGroup, nil
-			}
-		}
-	}
-
-	return nil, fmt.Errorf("no node group found")
+	return hcp.cloudServiceManager.GetAsgForInstance(instanceID)
 }
 
 // Pricing returns pricing model for this cloud provider or error if not available. Not implemented.
@@ -176,6 +171,38 @@ func (hcp *huaweicloudCloudProvider) Refresh() error {
 	return nil
 }
 
+func (hcp *huaweicloudCloudProvider) buildAsgs(specs []string) error {
+	asgs, err := hcp.cloudServiceManager.ListScalingGroups()
+	if err != nil {
+		klog.Errorf("failed to list scaling groups, because of %s", err.Error())
+		return err
+	}
+
+	for _, spec := range specs {
+		if err := hcp.addNodeGroup(spec, asgs); err != nil {
+			klog.Warningf("failed to add node group to huaweicloud provider with spec: %s", spec)
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (hcp *huaweicloudCloudProvider) addNodeGroup(spec string, asgs []AutoScalingGroup) error {
+	asg, err := buildAsgFromSpec(spec, asgs, hcp.cloudServiceManager)
+	if err != nil {
+		klog.Errorf("failed to build ASG from spec, because of %s", err.Error())
+		return err
+	}
+	hcp.addAsg(asg)
+	return nil
+}
+
+func (hcp *huaweicloudCloudProvider) addAsg(asg *AutoScalingGroup) {
+	hcp.autoScalingGroup = append(hcp.autoScalingGroup, *asg)
+	hcp.cloudServiceManager.RegisterAsg(asg)
+}
+
 // BuildHuaweiCloud is called by the autoscaler/cluster-autoscaler/builder to build a huaweicloud cloud provider.
 func BuildHuaweiCloud(opts config.AutoscalingOptions, do cloudprovider.NodeGroupDiscoveryOptions, rl *cloudprovider.ResourceLimiter) cloudprovider.CloudProvider {
 	if len(opts.CloudConfig) == 0 {
@@ -184,3 +211,23 @@ func BuildHuaweiCloud(opts config.AutoscalingOptions, do cloudprovider.NodeGroup
 
 	return newCloudProvider(opts, do, rl)
 }
+
+func buildAsgFromSpec(specStr string, asgs []AutoScalingGroup, manager CloudServiceManager) (*AutoScalingGroup, error) {
+	spec, err := dynamic.SpecFromString(specStr, true)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse node group spec: %v", err)
+	}
+
+	for _, asg := range asgs {
+		if asg.groupName == spec.Name {
+			return &AutoScalingGroup{
+				cloudServiceManager: manager,
+				groupName:           asg.groupName,
+				groupID:             asg.groupID,
+				minInstanceNumber:   asg.minInstanceNumber,
+				maxInstanceNumber:   asg.maxInstanceNumber,
+			}, nil
+		}
+	}
+	return nil, fmt.Errorf("no auto scaling group found, spec: %s", spec.Name)
+}

cluster-autoscaler/cloudprovider/huaweicloud/huaweicloud_service_manager.go

@@ -62,6 +62,12 @@ type AutoScalingService interface {
 	// IncreaseSizeInstance wait until instance number is updated.
 	IncreaseSizeInstance(groupID string, delta int) error
 
+	// GetAsgForInstance returns auto scaling group for the given instance.
+	GetAsgForInstance(instanceID string) (*AutoScalingGroup, error)
+
+	// RegisterAsg registers auto scaling group to manager
+	RegisterAsg(asg *AutoScalingGroup)
+
 	// DeleteScalingInstances is used to delete instances from auto scaling group by instanceIDs.
 	DeleteScalingInstances(groupID string, instanceIds []string) error
 
@@ -86,6 +92,7 @@ type cloudServiceManager struct {
 	cloudConfig      *CloudConfig
 	getECSClientFunc func() *huaweicloudsdkecs.EcsClient
 	getASClientFunc  func() *huaweicloudsdkas.AsClient
+	asgs             *autoScalingGroupCache
 }
 
 type asgTemplate struct {
@@ -99,11 +106,24 @@ type asgTemplate struct {
 }
 
 func newCloudServiceManager(cloudConfig *CloudConfig) *cloudServiceManager {
-	return &cloudServiceManager{
+	csm := &cloudServiceManager{
 		cloudConfig:      cloudConfig,
 		getECSClientFunc: cloudConfig.getECSClient,
 		getASClientFunc:  cloudConfig.getASClient,
+		asgs:             newAutoScalingGroupCache(),
 	}
+
+	csm.asgs.generateCache(csm)
+
+	return csm
+}
+
+func (csm *cloudServiceManager) GetAsgForInstance(instanceID string) (*AutoScalingGroup, error) {
+	return csm.asgs.FindForInstance(instanceID, csm)
+}
+
+func (csm *cloudServiceManager) RegisterAsg(asg *AutoScalingGroup) {
+	csm.asgs.Register(asg)
 }
 
 // DeleteServers deletes a group of server by ID.