fix:remove goroutine concurrency as it increased complexity without significantly improving collection speed (#70)

* fix: remove goroutine concurrency as it increased complexity without significantly improving collection speed

* swas using new sdk

* go mod tidy

Author: j3ttt

collector/alicloud/collector/cloudapi/apigateway.go

@@ -17,8 +17,6 @@ package cloudapi
 
 import (
 	"context"
-	"sync"
-
 	cloudapi20160714 "github.com/alibabacloud-go/cloudapi-20160714/v5/client"
 	"github.com/alibabacloud-go/tea/tea"
 	"github.com/cloudrec/alicloud/collector"
@@ -28,8 +26,6 @@ import (
 	"go.uber.org/zap"
 )
 
-const maxWorkers = 10
-
 // GetAPIGatewayResource 返回API Gateway资源定义
 func GetAPIGatewayResource() schema.Resource {
 	return schema.Resource{
@@ -62,30 +58,13 @@ func GetAPIGatewayDetail(ctx context.Context, service schema.ServiceInterface, r
 		return err
 	}
 
-	var wg sync.WaitGroup
-	tasks := make(chan *cloudapi20160714.DescribeApisResponseBodyApiSummarysApiSummary, len(apis))
-
-	// 启动工作协程
-	for i := 0; i < maxWorkers; i++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			for api := range tasks {
-				detail := describeAPIDetail(ctx, client, api)
-				if detail != nil {
-					res <- detail
-				}
-			}
-		}()
-	}
-
-	// 添加任务
 	for _, api := range apis {
-		tasks <- api
+		res <- &APIGatewayDetail{
+			ApiSummary: api,
+			ApiInfo:    describeAPI(ctx, client, api),
+		}
 	}
-	close(tasks)
 
-	wg.Wait()
 	return nil
 }
 
@@ -117,8 +96,7 @@ func listAPIs(ctx context.Context, c *cloudapi20160714.Client) ([]*cloudapi20160
 	return apis, nil
 }
 
-// describeAPIDetail 获取单个API详细信息
-func describeAPIDetail(ctx context.Context, c *cloudapi20160714.Client, api *cloudapi20160714.DescribeApisResponseBodyApiSummarysApiSummary) *APIGatewayDetail {
+func describeAPI(ctx context.Context, c *cloudapi20160714.Client, api *cloudapi20160714.DescribeApisResponseBodyApiSummarysApiSummary) *cloudapi20160714.DescribeApiResponseBody {
 	req := &cloudapi20160714.DescribeApiRequest{}
 	req.ApiId = api.ApiId
 
@@ -128,8 +106,5 @@ func describeAPIDetail(ctx context.Context, c *cloudapi20160714.Client, api *clo
 		return nil
 	}
 
-	return &APIGatewayDetail{
-		ApiSummary: api,
-		ApiInfo:    resp.Body,
-	}
+	return resp.Body
 }

collector/alicloud/collector/ons/instance.go

@@ -17,8 +17,6 @@ package ons
 
 import (
 	"context"
-	"sync"
-
 	"github.com/aliyun/alibaba-cloud-sdk-go/sdk/requests"
 	"github.com/aliyun/alibaba-cloud-sdk-go/services/ons"
 	"github.com/cloudrec/alicloud/collector"
@@ -28,8 +26,6 @@ import (
 	"go.uber.org/zap"
 )
 
-const maxWorkers = 10
-
 // GetInstanceResource returns ONS Instance resource definition
 func GetInstanceResource() schema.Resource {
 	return schema.Resource{
@@ -50,7 +46,7 @@ func GetInstanceResource() schema.Resource {
 // InstanceDetail aggregates resource details
 type InstanceDetail struct {
 	Instance         ons.InstanceVO
-	InstanceBaseInfo ons.InstanceBaseInfo
+	InstanceBaseInfo *ons.InstanceBaseInfo
 }
 
 // GetInstanceDetail gets ONS Instance details
@@ -63,28 +59,13 @@ func GetInstanceDetail(ctx context.Context, service schema.ServiceInterface, res
 		return err
 	}
 
-	var wg sync.WaitGroup
-	tasks := make(chan ons.InstanceVO, len(resources))
-
-	for i := 0; i < maxWorkers; i++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			for resource := range tasks {
-				detail := describeInstanceDetail(ctx, client, resource)
-				if detail != nil {
-					res <- detail
-				}
-			}
-		}()
-	}
-
 	for _, resource := range resources {
-		tasks <- resource
-	}
-	close(tasks)
 
-	wg.Wait()
+		res <- &InstanceDetail{
+			Instance:         resource,
+			InstanceBaseInfo: describeInstance(ctx, client, resource),
+		}
+	}
 	return nil
 }
 
@@ -106,8 +87,8 @@ func listInstances(ctx context.Context, c *ons.Client) ([]ons.InstanceVO, error)
 	return resources, nil
 }
 
-// describeInstanceDetail gets details for a single ONS Instance
-func describeInstanceDetail(ctx context.Context, c *ons.Client, resource ons.InstanceVO) *InstanceDetail {
+// describeInstance gets details for a single ONS Instance
+func describeInstance(ctx context.Context, c *ons.Client, resource ons.InstanceVO) *ons.InstanceBaseInfo {
 	req := ons.CreateOnsInstanceBaseInfoRequest()
 	req.InitWithApiInfo("Ons", "2019-02-14", "OnsInstanceBaseInfo", "ons", "openAPI")
 	req.Method = requests.POST
@@ -116,13 +97,8 @@ func describeInstanceDetail(ctx context.Context, c *ons.Client, resource ons.Ins
 	response, err := c.OnsInstanceBaseInfo(req)
 	if err != nil {
 		log.CtxLogger(ctx).Error("failed to get instance base info", zap.String("instanceId", resource.InstanceId), zap.Error(err))
-		return &InstanceDetail{
-			Instance: resource,
-		}
+		return nil
 	}
 
-	return &InstanceDetail{
-		Instance:         resource,
-		InstanceBaseInfo: response.InstanceBaseInfo,
-	}
+	return &response.InstanceBaseInfo
 }

collector/alicloud/collector/services.go

@@ -60,6 +60,7 @@ import (
 	selectdb20230522 "github.com/alibabacloud-go/selectdb-20230522/v3/client"
 	slb20140515 "github.com/alibabacloud-go/slb-20140515/v4/client"
 	sls20201230 "github.com/alibabacloud-go/sls-20201230/v6/client"
+	swas_open20200601 "github.com/alibabacloud-go/swas-open-20200601/v3/client"
 	tablestore20201209 "github.com/alibabacloud-go/tablestore-20201209/client"
 	util "github.com/alibabacloud-go/tea-utils/v2/service"
 	"github.com/alibabacloud-go/tea/tea"
@@ -84,7 +85,6 @@ import (
 	"github.com/aliyun/alibaba-cloud-sdk-go/services/ons"
 	"github.com/aliyun/alibaba-cloud-sdk-go/services/ram"
 	"github.com/aliyun/alibaba-cloud-sdk-go/services/sgw"
-	"github.com/aliyun/alibaba-cloud-sdk-go/services/swas-open"
 	"github.com/aliyun/alibaba-cloud-sdk-go/services/vod"
 	"github.com/aliyun/alibaba-cloud-sdk-go/services/vpc"
 	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
@@ -185,7 +185,7 @@ type Services struct {
 	ECP             *eds_aic20230930.Client
 	Eflo            *eflo.Client
 	EfloController  *eflo_controller.Client
-	SWAS            *swas_open.Client
+	SWAS            *swas_open20200601.Client
 	Ons             *ons.Client
 	GA              *ga.Client
 	DCDN            *dcdn.Client
@@ -199,8 +199,6 @@ func (s *Services) Clone() schema.ServiceInterface {
 	return &Services{}
 }
 
-
-
 func (s *Services) InitServices(cloudAccountParam schema.CloudAccountParam) (err error) {
 	param := cloudAccountParam.CommonCloudAccountParam
 	s.CloudAccountId = cloudAccountParam.CloudAccountId
@@ -499,7 +497,7 @@ func (s *Services) InitServices(cloudAccountParam schema.CloudAccountParam) (err
 			log.CtxLogger(ctx).Warn("init eci client failed", zap.Error(err))
 		}
 	case SWAS:
-		s.SWAS, err = swas_open.NewClientWithAccessKey(param.Region, param.AK, param.SK)
+		s.SWAS, err = createSWASClient(param.Region, s.Config)
 		if err != nil {
 			log.CtxLogger(ctx).Warn("init swas client failed", zap.Error(err))
 		}
@@ -553,6 +551,12 @@ func (s *Services) InitServices(cloudAccountParam schema.CloudAccountParam) (err
 	return nil
 }
 
+func createSWASClient(region string, config *openapi.Config) (client *swas_open20200601.Client, err error) {
+	config.Endpoint = tea.String("swas." + region + ".aliyuncs.com")
+	client, err = swas_open20200601.NewClient(config)
+	return client, err
+}
+
 func createVPCClient(region string, config *openapi.Config) (client *vpc.Client, err error) {
 	client, err = vpc.NewClientWithAccessKey(region, *config.AccessKeyId, *config.AccessKeySecret)
 	return client, err

collector/alicloud/collector/swas/swas.go

@@ -17,19 +17,15 @@ package swas
 
 import (
 	"context"
-	"sync"
-
-	"github.com/aliyun/alibaba-cloud-sdk-go/sdk/requests"
-	swas "github.com/aliyun/alibaba-cloud-sdk-go/services/swas-open"
+	swas_open20200601 "github.com/alibabacloud-go/swas-open-20200601/v3/client"
+	"github.com/alibabacloud-go/tea/tea"
 	"github.com/cloudrec/alicloud/collector"
 	"github.com/core-sdk/constant"
 	"github.com/core-sdk/log"
 	"github.com/core-sdk/schema"
 	"go.uber.org/zap"
 )
 
-const maxWorkers = 10
-
 // GetInstanceResource returns SWAS instance resource definition
 func GetInstanceResource() schema.Resource {
 	return schema.Resource{
@@ -48,84 +44,78 @@ func GetInstanceResource() schema.Resource {
 
 // Detail aggregates SWAS instance and firewall rule information
 type Detail struct {
-	Instance      *swas.Instance
-	FirewallRules []swas.FirewallRule
+	Instance      *swas_open20200601.ListInstancesResponseBodyInstances
+	FirewallRules []*swas_open20200601.ListFirewallRulesResponseBodyFirewallRules
 }
 
 // ListInstancesResource gets SWAS instance details
 func ListInstancesResource(ctx context.Context, service schema.ServiceInterface, res chan<- any) error {
 	cli := service.(*collector.Services).SWAS
-	req := swas.CreateListInstancesRequest()
-	req.PageSize = requests.NewInteger(50)
-	req.PageNumber = requests.NewInteger(1)
 
-	count := 0
-	for {
-		resp, err := cli.ListInstances(req)
-		if err != nil {
-			log.CtxLogger(ctx).Warn("ListInstances error", zap.Error(err))
-			return err
-		}
+	instances, err := listInstances(ctx, cli)
+	if err != nil {
+		log.CtxLogger(ctx).Warn("ListInstances error", zap.Error(err))
+		return err
+	}
 
-		var wg sync.WaitGroup
-		tasks := make(chan swas.Instance, len(resp.Instances))
-
-		// 启动工作协程
-		for i := 0; i < maxWorkers; i++ {
-			wg.Add(1)
-			go func() {
-				defer wg.Done()
-				for instance := range tasks {
-					d := &Detail{
-						Instance:      &instance,
-						FirewallRules: listFirewallRules(ctx, cli, instance.InstanceId),
-					}
-
-					res <- d
-				}
-			}()
+	for _, instance := range instances {
+		firewallRules := listFirewallRules(ctx, cli, instance.InstanceId)
+		res <- &Detail{
+			Instance:      instance,
+			FirewallRules: firewallRules,
 		}
+	}
+
+	return nil
+}
 
-		// 添加任务
-		for _, instance := range resp.Instances {
-			tasks <- instance
+// listInstances lists all swas instances in a region.
+func listInstances(ctx context.Context, cli *swas_open20200601.Client) (instances []*swas_open20200601.ListInstancesResponseBodyInstances, err error) {
+	request := &swas_open20200601.ListInstancesRequest{
+		PageNumber: tea.Int32(1),
+		PageSize:   tea.Int32(50),
+	}
+
+	count := 0
+	for {
+		resp, err := cli.ListInstances(request)
+		if err != nil {
+			return nil, err
 		}
-		close(tasks)
 
-		wg.Wait()
+		instances = append(instances, resp.Body.Instances...)
 
-		count += len(resp.Instances)
-		if count >= resp.TotalCount || len(resp.Instances) == 0 {
+		count += len(resp.Body.Instances)
+		if count >= int(*resp.Body.TotalCount) || len(resp.Body.Instances) == 0 {
 			break
 		}
-		req.PageNumber = requests.NewInteger(resp.PageNumber + 1)
+		request.PageNumber = tea.Int32(*request.PageNumber + 1)
 	}
 
-	return nil
+	return instances, nil
 }
 
 // listFirewallRules gets firewall rules for a specific instance
-func listFirewallRules(ctx context.Context, client *swas.Client, instanceId string) (firewallRules []swas.FirewallRule) {
-	req := swas.CreateListFirewallRulesRequest()
-	req.InstanceId = instanceId
-	req.PageSize = requests.NewInteger(constant.DefaultPageSize)
-	req.PageNumber = requests.NewInteger(1)
+func listFirewallRules(ctx context.Context, client *swas_open20200601.Client, instanceId *string) (firewallRules []*swas_open20200601.ListFirewallRulesResponseBodyFirewallRules) {
 
 	count := 0
 	for {
-		resp, err := client.ListFirewallRules(req)
+		request := &swas_open20200601.ListFirewallRulesRequest{
+			InstanceId: instanceId,
+		}
+		resp, err := client.ListFirewallRules(request)
 		if err != nil {
 			log.CtxLogger(ctx).Warn("ListFirewallRules error", zap.Error(err))
 			break
 		}
 
-		firewallRules = append(firewallRules, resp.FirewallRules...)
-		count += len(resp.FirewallRules)
+		firewallRules = append(firewallRules, resp.Body.FirewallRules...)
+		count += len(resp.Body.FirewallRules)
 
-		if count >= resp.TotalCount || len(resp.FirewallRules) < constant.DefaultPageSize {
+		if count >= int(*resp.Body.TotalCount) || len(resp.Body.FirewallRules) == 0 {
 			break
 		}
-		req.PageNumber = requests.NewInteger(resp.PageNumber + 1)
+		request.PageNumber = tea.Int32(*request.PageNumber + 1)
 	}
 
 	return firewallRules