fix: 完善多云同步验证与实例统计
- 修复华为云签名、账单字段与区域映射缓存 - 按真实云实例更新账号服务器数并补充并发测试
This commit is contained in:
1 parent
132fe9ba97
commit
9b895efee3
6 files changed
+226
-64
No files matched your search
+26
-12
@@ -8,15 +8,20 @@ import (
|
||||
"net/http"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// HuaweiProvider 通过华为云 AK/SK 签名 API 同步项目区域与 ECS 实例。 // Logan.Li
|
||||
type HuaweiProvider struct{ accessKey, secretKey string }
|
||||
type HuaweiProvider struct {
|
||||
accessKey, secretKey string
|
||||
regionMu sync.RWMutex
|
||||
regionByProject map[string]string
|
||||
}
|
||||
|
||||
// NewHuaweiProvider 创建华为云 Provider。 // Logan.Li
|
||||
func NewHuaweiProvider(accessKey, secretKey string) *HuaweiProvider {
|
||||
return &HuaweiProvider{accessKey, secretKey}
|
||||
return &HuaweiProvider{accessKey: accessKey, secretKey: secretKey, regionByProject: make(map[string]string)}
|
||||
}
|
||||
|
||||
// Name 返回华为云厂商标识。 // Logan.Li
|
||||
@@ -32,7 +37,11 @@ func (p *HuaweiProvider) signedGet(rawURL, _ string) ([]byte, error) {
|
||||
host := u.URL.Host
|
||||
canonicalHeaders := "host:" + host + "\n" + "x-sdk-date:" + now + "\n"
|
||||
signedHeaders := "host;x-sdk-date"
|
||||
canonicalRequest := "GET\n" + u.URL.EscapedPath() + "\n" + u.URL.Query().Encode() + "\n" + canonicalHeaders + "\n" + signedHeaders + "\n" + sha256Hex("")
|
||||
canonicalURI := u.URL.EscapedPath()
|
||||
if !strings.HasSuffix(canonicalURI, "/") {
|
||||
canonicalURI += "/"
|
||||
}
|
||||
canonicalRequest := "GET\n" + canonicalURI + "\n" + u.URL.Query().Encode() + "\n" + canonicalHeaders + "\n" + signedHeaders + "\n" + sha256Hex("")
|
||||
stringToSign := "SDK-HMAC-SHA256\n" + now + "\n" + sha256Hex(canonicalRequest)
|
||||
signature := hex.EncodeToString(hmacSHA256([]byte(p.secretKey), stringToSign))
|
||||
u.Header.Set("X-Sdk-Date", now)
|
||||
@@ -68,28 +77,33 @@ func (p *HuaweiProvider) ListRegions() ([]CloudRegion, error) {
|
||||
return nil, fmt.Errorf("解析华为云区域响应失败: %w", err)
|
||||
}
|
||||
regions := []CloudRegion{}
|
||||
regionByProject := make(map[string]string, len(result.Projects))
|
||||
for _, project := range result.Projects {
|
||||
if project.ID == "" || project.Name == "" {
|
||||
continue
|
||||
}
|
||||
regions = append(regions, CloudRegion{RegionID: project.ID, LocalName: project.Name, Endpoint: "https://ecs." + project.Name + ".myhuaweicloud.com", Status: "active"})
|
||||
regionByProject[project.ID] = project.Name
|
||||
}
|
||||
p.regionMu.Lock()
|
||||
p.regionByProject = regionByProject
|
||||
p.regionMu.Unlock()
|
||||
sort.Slice(regions, func(i, j int) bool { return regions[i].LocalName < regions[j].LocalName })
|
||||
return regions, nil
|
||||
}
|
||||
|
||||
// ListInstances 获取指定华为云项目中的 ECS 实例。 // Logan.Li
|
||||
func (p *HuaweiProvider) ListInstances(projectID string) ([]CloudInstance, error) {
|
||||
regions, err := p.ListRegions()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
regionName := ""
|
||||
for _, region := range regions {
|
||||
if region.RegionID == projectID {
|
||||
regionName = region.LocalName
|
||||
break
|
||||
p.regionMu.RLock()
|
||||
regionName := p.regionByProject[projectID]
|
||||
p.regionMu.RUnlock()
|
||||
if regionName == "" {
|
||||
if _, err := p.ListRegions(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
p.regionMu.RLock()
|
||||
regionName = p.regionByProject[projectID]
|
||||
p.regionMu.RUnlock()
|
||||
}
|
||||
if regionName == "" {
|
||||
return nil, fmt.Errorf("未找到华为云项目 %s", projectID)
|
||||
|
||||
@@ -11,6 +11,28 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
type huaweiBillRecord struct {
|
||||
ServiceTypeCode string `json:"service_type_code"`
|
||||
ServiceTypeName string `json:"service_type_name"`
|
||||
CloudServiceType string `json:"cloud_service_type"`
|
||||
CloudServiceTypeName string `json:"cloud_service_type_name"`
|
||||
ResourceTypeName string `json:"resource_type_name"`
|
||||
RegionCode string `json:"region_code"`
|
||||
Region string `json:"region"`
|
||||
RegionName string `json:"region_name"`
|
||||
Amount json.RawMessage `json:"amount"`
|
||||
ConsumeAmount json.RawMessage `json:"consume_amount"`
|
||||
OfficialAmount json.RawMessage `json:"official_amount"`
|
||||
DiscountAmount json.RawMessage `json:"discount_amount"`
|
||||
}
|
||||
|
||||
type huaweiBillPage struct {
|
||||
TotalCount int `json:"total_count"`
|
||||
Currency string `json:"currency"`
|
||||
Records []huaweiBillRecord `json:"records"`
|
||||
FeeRecords []huaweiBillRecord `json:"fee_records"`
|
||||
}
|
||||
|
||||
// FetchDailyBill 通过华为云 BSS 费用流水接口拉取单日账单。 // Logan.Li
|
||||
func (p *HuaweiProvider) FetchDailyBill(billingDate string) ([]CloudBillingRecord, error) {
|
||||
if _, err := time.Parse("2006-01-02", billingDate); err != nil {
|
||||
@@ -21,10 +43,11 @@ func (p *HuaweiProvider) FetchDailyBill(billingDate string) ([]CloudBillingRecor
|
||||
region string
|
||||
}
|
||||
aggregation := map[aggregationKey]float64{}
|
||||
currency := "CNY"
|
||||
const pageSize = 1000
|
||||
for offset := 0; ; offset += pageSize {
|
||||
query := url.Values{
|
||||
"cycle": {billingDate[:7]},
|
||||
"bill_cycle": {billingDate[:7]},
|
||||
"bill_date_begin": {billingDate},
|
||||
"bill_date_end": {billingDate},
|
||||
"offset": {strconv.Itoa(offset)},
|
||||
@@ -34,35 +57,27 @@ func (p *HuaweiProvider) FetchDailyBill(billingDate string) ([]CloudBillingRecor
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var response struct {
|
||||
TotalCount int `json:"total_count"`
|
||||
FeeRecords []struct {
|
||||
CloudServiceType string `json:"cloud_service_type"`
|
||||
CloudServiceTypeName string `json:"cloud_service_type_name"`
|
||||
ResourceTypeName string `json:"resource_type_name"`
|
||||
Region string `json:"region"`
|
||||
RegionName string `json:"region_name"`
|
||||
Amount json.RawMessage `json:"amount"`
|
||||
OfficialAmount json.RawMessage `json:"official_amount"`
|
||||
DiscountAmount json.RawMessage `json:"discount_amount"`
|
||||
} `json:"fee_records"`
|
||||
page, err := decodeHuaweiBillPage(body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err = json.Unmarshal(body, &response); err != nil {
|
||||
return nil, fmt.Errorf("解析华为云 BSS 响应失败: %w", err)
|
||||
if page.Currency != "" {
|
||||
currency = page.Currency
|
||||
}
|
||||
for _, item := range response.FeeRecords {
|
||||
cost := huaweiBillAmount(item.Amount)
|
||||
if cost == 0 {
|
||||
cost = huaweiBillAmount(item.OfficialAmount) - huaweiBillAmount(item.DiscountAmount)
|
||||
}
|
||||
records := page.Records
|
||||
if len(records) == 0 {
|
||||
records = page.FeeRecords
|
||||
}
|
||||
for _, item := range records {
|
||||
cost := huaweiBillRecordCost(item)
|
||||
if cost == 0 {
|
||||
continue
|
||||
}
|
||||
service := firstNonEmpty(item.CloudServiceTypeName, item.ResourceTypeName, item.CloudServiceType, "其他")
|
||||
region := firstNonEmpty(item.RegionName, item.Region, "全局")
|
||||
service := firstNonEmpty(item.ServiceTypeName, item.CloudServiceTypeName, item.ResourceTypeName, item.ServiceTypeCode, item.CloudServiceType, "其他")
|
||||
region := firstNonEmpty(item.RegionName, item.RegionCode, item.Region, "全局")
|
||||
aggregation[aggregationKey{service: service, region: region}] += cost
|
||||
}
|
||||
if len(response.FeeRecords) < pageSize || (response.TotalCount > 0 && offset+pageSize >= response.TotalCount) {
|
||||
if len(records) < pageSize || (page.TotalCount > 0 && offset+pageSize >= page.TotalCount) {
|
||||
break
|
||||
}
|
||||
}
|
||||
@@ -80,13 +95,33 @@ func (p *HuaweiProvider) FetchDailyBill(billingDate string) ([]CloudBillingRecor
|
||||
for _, key := range keys {
|
||||
records = append(records, CloudBillingRecord{
|
||||
BillingDate: billingDate, Provider: "huawei", Cost: math.Round(aggregation[key]*100) / 100,
|
||||
Currency: "CNY", Service: key.service, Region: key.region,
|
||||
Currency: currency, Service: key.service, Region: key.region,
|
||||
Description: "云 API 自动同步(华为云 BSS)",
|
||||
})
|
||||
}
|
||||
return records, nil
|
||||
}
|
||||
|
||||
// decodeHuaweiBillPage 兼容华为云 records 与旧版 fee_records 响应。 // Logan.Li
|
||||
func decodeHuaweiBillPage(body []byte) (huaweiBillPage, error) {
|
||||
var page huaweiBillPage
|
||||
if err := json.Unmarshal(body, &page); err != nil {
|
||||
return page, fmt.Errorf("解析华为云 BSS 响应失败: %w", err)
|
||||
}
|
||||
return page, nil
|
||||
}
|
||||
|
||||
// huaweiBillRecordCost 依次读取实际消费、流水金额和官网价减优惠。 // Logan.Li
|
||||
func huaweiBillRecordCost(item huaweiBillRecord) float64 {
|
||||
if cost := huaweiBillAmount(item.ConsumeAmount); cost != 0 {
|
||||
return cost
|
||||
}
|
||||
if cost := huaweiBillAmount(item.Amount); cost != 0 {
|
||||
return cost
|
||||
}
|
||||
return huaweiBillAmount(item.OfficialAmount) - huaweiBillAmount(item.DiscountAmount)
|
||||
}
|
||||
|
||||
// huaweiBillAmount 兼容华为云金额字段的字符串和数字格式。 // Logan.Li
|
||||
func huaweiBillAmount(raw json.RawMessage) float64 {
|
||||
value := strings.Trim(strings.TrimSpace(string(raw)), "\"")
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
package cloud
|
||||
|
||||
import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestDecodeHuaweiBillPageRecords(t *testing.T) {
|
||||
body := []byte(`{
|
||||
"currency":"CNY",
|
||||
"total_count":1,
|
||||
"records":[{
|
||||
"service_type_name":"弹性云服务器",
|
||||
"region_name":"华南-广州",
|
||||
"consume_amount":"12.34"
|
||||
}]
|
||||
}`)
|
||||
page, err := decodeHuaweiBillPage(body)
|
||||
if err != nil {
|
||||
t.Fatalf("解析华为云账单失败: %v", err)
|
||||
}
|
||||
if page.TotalCount != 1 || page.Currency != "CNY" || len(page.Records) != 1 {
|
||||
t.Fatalf("账单分页字段错误: %+v", page)
|
||||
}
|
||||
record := page.Records[0]
|
||||
if record.ServiceTypeName != "弹性云服务器" || record.RegionName != "华南-广州" {
|
||||
t.Fatalf("账单维度字段错误: %+v", record)
|
||||
}
|
||||
if cost := huaweiBillRecordCost(record); cost != 12.34 {
|
||||
t.Fatalf("实际消费金额错误: %v", cost)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecodeHuaweiBillPageFeeRecordsCompatibility(t *testing.T) {
|
||||
body := []byte(`{"total_count":1,"fee_records":[{"official_amount":20,"discount_amount":3}]}`)
|
||||
page, err := decodeHuaweiBillPage(body)
|
||||
if err != nil {
|
||||
t.Fatalf("解析旧版华为云账单失败: %v", err)
|
||||
}
|
||||
if len(page.FeeRecords) != 1 {
|
||||
t.Fatalf("旧版 fee_records 未解析: %+v", page)
|
||||
}
|
||||
if cost := huaweiBillRecordCost(page.FeeRecords[0]); cost != 17 {
|
||||
t.Fatalf("官网价减优惠金额错误: %v", cost)
|
||||
}
|
||||
}
|
||||
@@ -33,24 +33,6 @@ func NewCloudAccountHandler(db *gorm.DB) *CloudAccountHandler {
|
||||
func (h *CloudAccountHandler) ListCloudAccounts(c *gin.Context) {
|
||||
var accounts []model.CloudAccount
|
||||
h.db.Find(&accounts)
|
||||
// 服务器数来自真实云账号关联,避免重复导入后被“本次新增数”覆盖。 // Logan.Li
|
||||
type accountServerCount struct {
|
||||
CloudAccountID uint
|
||||
Total int
|
||||
}
|
||||
var serverCounts []accountServerCount
|
||||
h.db.Model(&model.Server{}).
|
||||
Select("cloud_account_id, COUNT(*) AS total").
|
||||
Where("cloud_account_id IS NOT NULL").Group("cloud_account_id").Scan(&serverCounts)
|
||||
countByAccount := make(map[uint]int, len(serverCounts))
|
||||
for _, item := range serverCounts {
|
||||
countByAccount[item.CloudAccountID] = item.Total
|
||||
}
|
||||
for index := range accounts {
|
||||
if total, ok := countByAccount[accounts[index].ID]; ok {
|
||||
accounts[index].ServerCount = total
|
||||
}
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, model.Response{
|
||||
Code: 200,
|
||||
|
||||
@@ -3,6 +3,7 @@ package controller
|
||||
import (
|
||||
"net/http"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"ezops/cloud"
|
||||
@@ -73,22 +74,60 @@ func (h *CloudSyncHandler) SyncRegions(c *gin.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
// 更新账号的 region_count 和 last_sync_at
|
||||
h.db.Model(&account).Updates(map[string]interface{}{
|
||||
// 并发统计所有可用区域实例数,卡片展示云账号真实资源量。 // Logan.Li
|
||||
serverCount, countedRegions := countCloudInstances(provider, cloudRegions)
|
||||
updates := map[string]interface{}{
|
||||
"region_count": len(cloudRegions),
|
||||
"last_sync_at": time.Now(),
|
||||
})
|
||||
}
|
||||
if countedRegions > 0 {
|
||||
updates["server_count"] = serverCount
|
||||
}
|
||||
// 更新账号区域、实例数量和同步时间。 // Logan.Li
|
||||
h.db.Model(&account).Updates(updates)
|
||||
|
||||
c.JSON(http.StatusOK, model.Response{
|
||||
Code: 200,
|
||||
Message: "同步成功",
|
||||
Data: gin.H{
|
||||
"synced": synced,
|
||||
"total": len(cloudRegions),
|
||||
"synced": synced,
|
||||
"total": len(cloudRegions),
|
||||
"server_count": serverCount,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
// countCloudInstances 以固定并发统计各区域实例数量,避免串行请求过慢。 // Logan.Li
|
||||
func countCloudInstances(provider cloud.Provider, regions []cloud.CloudRegion) (total int, succeeded int) {
|
||||
type countResult struct {
|
||||
count int
|
||||
ok bool
|
||||
}
|
||||
results := make(chan countResult, len(regions))
|
||||
semaphore := make(chan struct{}, 4)
|
||||
var waitGroup sync.WaitGroup
|
||||
for _, region := range regions {
|
||||
regionID := region.RegionID
|
||||
waitGroup.Add(1)
|
||||
go func() {
|
||||
defer waitGroup.Done()
|
||||
semaphore <- struct{}{}
|
||||
defer func() { <-semaphore }()
|
||||
instances, err := provider.ListInstances(regionID)
|
||||
results <- countResult{count: len(instances), ok: err == nil}
|
||||
}()
|
||||
}
|
||||
waitGroup.Wait()
|
||||
close(results)
|
||||
for result := range results {
|
||||
if result.ok {
|
||||
total += result.count
|
||||
succeeded++
|
||||
}
|
||||
}
|
||||
return total, succeeded
|
||||
}
|
||||
|
||||
// GetAccountRegions 获取指定云账号的区域列表(按 provider 过滤)
|
||||
// GET /cloud-accounts/:id/regions
|
||||
func (h *CloudSyncHandler) GetAccountRegions(c *gin.Context) {
|
||||
@@ -308,11 +347,6 @@ func (h *CloudSyncHandler) ImportInstances(c *gin.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
// 保存该账号当前已关联的资产总数,而不是本次新增数。 // Logan.Li
|
||||
var serverCount int64
|
||||
h.db.Model(&model.Server{}).Where("cloud_account_id = ?", account.ID).Count(&serverCount)
|
||||
h.db.Model(&account).UpdateColumn("server_count", serverCount)
|
||||
|
||||
c.JSON(http.StatusOK, model.Response{
|
||||
Code: 200,
|
||||
Message: "导入完成",
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
package controller
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"ezops/cloud"
|
||||
)
|
||||
|
||||
type countInstancesProvider struct {
|
||||
counts map[string]int
|
||||
failures map[string]bool
|
||||
activeCalls atomic.Int32
|
||||
maxCalls atomic.Int32
|
||||
}
|
||||
|
||||
func (p *countInstancesProvider) Name() string { return "fake" }
|
||||
|
||||
func (p *countInstancesProvider) ListRegions() ([]cloud.CloudRegion, error) { return nil, nil }
|
||||
|
||||
func (p *countInstancesProvider) ListInstances(regionID string) ([]cloud.CloudInstance, error) {
|
||||
active := p.activeCalls.Add(1)
|
||||
defer p.activeCalls.Add(-1)
|
||||
for {
|
||||
maximum := p.maxCalls.Load()
|
||||
if active <= maximum || p.maxCalls.CompareAndSwap(maximum, active) {
|
||||
break
|
||||
}
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
if p.failures[regionID] {
|
||||
return nil, errors.New("区域请求失败")
|
||||
}
|
||||
return make([]cloud.CloudInstance, p.counts[regionID]), nil
|
||||
}
|
||||
|
||||
func TestCountCloudInstances(t *testing.T) {
|
||||
provider := &countInstancesProvider{
|
||||
counts: map[string]int{"a": 2, "b": 3, "c": 4, "d": 5, "e": 6},
|
||||
failures: map[string]bool{"c": true},
|
||||
}
|
||||
regions := []cloud.CloudRegion{{RegionID: "a"}, {RegionID: "b"}, {RegionID: "c"}, {RegionID: "d"}, {RegionID: "e"}}
|
||||
total, succeeded := countCloudInstances(provider, regions)
|
||||
if total != 16 || succeeded != 4 {
|
||||
t.Fatalf("实例统计错误: total=%d succeeded=%d", total, succeeded)
|
||||
}
|
||||
if maximum := provider.maxCalls.Load(); maximum > 4 {
|
||||
t.Fatalf("并发数超过限制: %d", maximum)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user