MiFaZhan
256 lines
7.6 KiB
Go
256 lines
7.6 KiB
Go
package upstreamimpl
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"os"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"testing"
|
||
"time"
|
||
|
||
"git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/model"
|
||
"git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/model/entity"
|
||
)
|
||
|
||
// 真实上游时延测量（默认跳过）。
|
||
//
|
||
// 目的：实测移动 OneLink「批量提交返回 jobId」与「按 jobId 查到终态」各自的耗时，
|
||
// 用于校准平台超时与补查节奏。
|
||
//
|
||
// 仅在 APN_REAL_CARRIER_TEST=1 时执行。
|
||
//
|
||
// APN_REAL_CREDS_FILE=<path> 两行：appid / appsecret
|
||
// APN_REAL_BASE_URL=<url> 默认 https://api.iot.10086.cn/v5
|
||
// APN_REAL_APN=CMIOT
|
||
// APN_REAL_BATCHES=<b1>|<b2>|... 每批：operType,iccid:msisdn,iccid:msisdn,...
|
||
// APN_REAL_POLL_INTERVAL=15s 查询间隔
|
||
// APN_REAL_MAX_WAIT=15m 最长等待终态
|
||
func TestRealCarrierApnLatencyMeasure(t *testing.T) {
|
||
if strings.TrimSpace(os.Getenv("APN_REAL_CARRIER_TEST")) != "1" {
|
||
t.Skip("真实上游测量未启用：设置 APN_REAL_CARRIER_TEST=1 才执行")
|
||
}
|
||
appID, appSecret, err := loadRealCarrierCreds()
|
||
if err != nil {
|
||
t.Fatalf("读取凭证失败：%v", err)
|
||
}
|
||
baseURL := strings.TrimSpace(os.Getenv("APN_REAL_BASE_URL"))
|
||
if baseURL == "" {
|
||
baseURL = "https://api.iot.10086.cn/v5"
|
||
}
|
||
apn := strings.TrimSpace(os.Getenv("APN_REAL_APN"))
|
||
if apn == "" {
|
||
apn = "CMIOT"
|
||
}
|
||
batches, operTypes, err := parseRealBatches(os.Getenv("APN_REAL_BATCHES"))
|
||
if err != nil {
|
||
t.Fatalf("解析批次失败：%v", err)
|
||
}
|
||
pollInterval := 15 * time.Second
|
||
if raw := strings.TrimSpace(os.Getenv("APN_REAL_POLL_INTERVAL")); raw != "" {
|
||
if d, derr := time.ParseDuration(raw); derr == nil && d > 0 {
|
||
pollInterval = d
|
||
}
|
||
}
|
||
maxWait := 15 * time.Minute
|
||
if raw := strings.TrimSpace(os.Getenv("APN_REAL_MAX_WAIT")); raw != "" {
|
||
if d, derr := time.ParseDuration(raw); derr == nil && d > 0 {
|
||
maxWait = d
|
||
}
|
||
}
|
||
|
||
channel := &sSimYD{
|
||
channelConfig: &model.SimChannelConfigItemCache{Channel: &entity.TSimChannels{ChanKey: "real-carrier-measure", AppKey: appID}},
|
||
tokenProvider: &realCarrierToken{appID: appID, appSecret: appSecret, baseURL: baseURL},
|
||
apiConfig: &oneLinkAPIConfig{BaseURL: baseURL, Timeout: 60 * time.Second},
|
||
}
|
||
|
||
type batchOutcome struct {
|
||
index int
|
||
operType string
|
||
cards int
|
||
elapsed time.Duration
|
||
jobID string
|
||
failed string
|
||
}
|
||
start := time.Now()
|
||
ctx := context.Background()
|
||
results := make([]batchOutcome, len(batches))
|
||
var wg sync.WaitGroup
|
||
for i := range batches {
|
||
wg.Add(1)
|
||
go func(i int) {
|
||
defer wg.Done()
|
||
outcome := batchOutcome{index: i, operType: operTypes[i], cards: len(batches[i])}
|
||
begin := time.Now()
|
||
items, callErr := changeSimApnFunctionYD(ctx, channel, &model.SimCardOperateChangeApnInput{
|
||
SimKeys: batches[i],
|
||
ReqData: model.SimCardOperateChangeApnInputItem{
|
||
OperType: operTypes[i],
|
||
ApnName: apn,
|
||
SubmitOnly: true,
|
||
OperationID: fmt.Sprintf("measure-%d-%d", i, begin.UnixNano()),
|
||
},
|
||
})
|
||
outcome.elapsed = time.Since(begin)
|
||
switch {
|
||
case callErr != nil:
|
||
outcome.failed = callErr.Error()
|
||
case len(items) == 0:
|
||
outcome.failed = "空返回"
|
||
default:
|
||
outcome.jobID = items[0].ResData.OrderNum
|
||
for _, item := range items {
|
||
if item == nil {
|
||
continue
|
||
}
|
||
if item.ResData.OrderNum != "" && item.ResData.OrderNum != outcome.jobID {
|
||
outcome.failed = "同批出现多个jobId"
|
||
}
|
||
if !item.ResData.Pending && !item.ResData.Uncertain {
|
||
outcome.failed = fmt.Sprintf("提交即终态 code=%d msg=%s", item.Code, item.Message)
|
||
}
|
||
}
|
||
}
|
||
results[i] = outcome
|
||
}(i)
|
||
}
|
||
wg.Wait()
|
||
submitWall := time.Since(start)
|
||
|
||
fmt.Printf("MEASURE_SUBMIT batches=%d cards=%d wall=%s\n", len(batches), totalRealCards(batches), submitWall)
|
||
for _, r := range results {
|
||
fmt.Printf("MEASURE_SUBMIT_BATCH idx=%d operType=%s cards=%d elapsed=%s jobId=%s failed=%s\n",
|
||
r.index, r.operType, r.cards, r.elapsed, r.jobID, r.failed)
|
||
}
|
||
|
||
// 汇总待查任务
|
||
pendingJobs := map[string][]*model.SimKey{}
|
||
for _, r := range results {
|
||
if r.jobID != "" {
|
||
pendingJobs[r.jobID] = batches[r.index]
|
||
}
|
||
}
|
||
terminalAt := map[string]time.Duration{}
|
||
fmt.Printf("MEASURE_JOBS distinct=%d %v\n", len(pendingJobs), sortedJobIDs(pendingJobs))
|
||
|
||
deadline := start.Add(maxWait)
|
||
for len(terminalAt) < len(pendingJobs) && time.Now().Before(deadline) {
|
||
time.Sleep(pollInterval)
|
||
elapsed := time.Since(start)
|
||
for jobID, keys := range pendingJobs {
|
||
if _, done := terminalAt[jobID]; done {
|
||
continue
|
||
}
|
||
items, err := changeSimApnFunctionYD(ctx, channel, &model.SimCardOperateChangeApnInput{
|
||
SimKeys: keys,
|
||
ReqData: model.SimCardOperateChangeApnInputItem{QueryJobID: jobID},
|
||
})
|
||
if err != nil {
|
||
fmt.Printf("MEASURE_QUERY_ERR jobId=%s at=%s err=%v\n", jobID, elapsed.Round(time.Second), err)
|
||
continue
|
||
}
|
||
allTerminal := len(items) > 0
|
||
for _, item := range items {
|
||
if item == nil || item.ResData.Pending {
|
||
allTerminal = false
|
||
break
|
||
}
|
||
}
|
||
if allTerminal {
|
||
terminalAt[jobID] = elapsed
|
||
fmt.Printf("MEASURE_TERMINAL jobId=%s at=%s\n", jobID, elapsed.Round(time.Second))
|
||
for _, item := range items {
|
||
fmt.Printf("MEASURE_CARD jobId=%s iccid=%s msisdn=%s code=%d uncertain=%v message=%s\n",
|
||
jobID, item.SimKey.ICCID, item.SimKey.MSISDN, item.Code, item.ResData.Uncertain, item.Message)
|
||
}
|
||
} else {
|
||
fmt.Printf("MEASURE_PENDING jobId=%s at=%s note=%s\n", jobID, elapsed.Round(time.Second), firstMessage(items))
|
||
}
|
||
}
|
||
}
|
||
|
||
fmt.Printf("MEASURE_SUMMARY\n")
|
||
for _, jobID := range sortedJobIDs(pendingJobs) {
|
||
if at, ok := terminalAt[jobID]; ok {
|
||
fmt.Printf("MEASURE_RESULT jobId=%s terminalAt=%s\n", jobID, at.Round(time.Second))
|
||
} else {
|
||
fmt.Printf("MEASURE_RESULT jobId=%s terminalAt=NONE within %s\n", jobID, maxWait)
|
||
}
|
||
}
|
||
}
|
||
|
||
func totalRealCards(batches [][]*model.SimKey) int {
|
||
total := 0
|
||
for _, b := range batches {
|
||
total += len(b)
|
||
}
|
||
return total
|
||
}
|
||
|
||
func sortedJobIDs(jobs map[string][]*model.SimKey) []string {
|
||
ids := make([]string, 0, len(jobs))
|
||
for id := range jobs {
|
||
ids = append(ids, id)
|
||
}
|
||
sort.Strings(ids)
|
||
return ids
|
||
}
|
||
|
||
func firstMessage(items model.SimCardOperateChangeApnOutput) string {
|
||
for _, item := range items {
|
||
if item != nil {
|
||
return item.Message
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// parseRealBatches 解析 APN_REAL_BATCHES：批之间用 | 分隔，每批首项为 operType，其余为 iccid:msisdn。
|
||
func parseRealBatches(raw string) ([][]*model.SimKey, []string, error) {
|
||
raw = strings.TrimSpace(raw)
|
||
if raw == "" {
|
||
return nil, nil, fmt.Errorf("未设置 APN_REAL_BATCHES")
|
||
}
|
||
batches := make([][]*model.SimKey, 0)
|
||
operTypes := make([]string, 0)
|
||
for _, batchRaw := range strings.Split(raw, "|") {
|
||
batchRaw = strings.TrimSpace(batchRaw)
|
||
if batchRaw == "" {
|
||
continue
|
||
}
|
||
fields := strings.Split(batchRaw, ",")
|
||
operType := strings.TrimSpace(fields[0])
|
||
if _, err := strconv.Atoi(operType); err != nil {
|
||
return nil, nil, fmt.Errorf("批次首项应为 operType：%q", batchRaw)
|
||
}
|
||
keys := make([]*model.SimKey, 0, len(fields)-1)
|
||
for _, card := range fields[1:] {
|
||
card = strings.TrimSpace(card)
|
||
if card == "" {
|
||
continue
|
||
}
|
||
parts := strings.Split(card, ":")
|
||
if len(parts) < 2 {
|
||
return nil, nil, fmt.Errorf("卡项格式应为 iccid:msisdn：%q", card)
|
||
}
|
||
keys = append(keys, &model.SimKey{
|
||
ChanKey: "real-carrier-measure",
|
||
ICCID: strings.TrimSpace(parts[0]),
|
||
MSISDN: strings.TrimSpace(parts[1]),
|
||
})
|
||
}
|
||
if len(keys) == 0 {
|
||
continue
|
||
}
|
||
batches = append(batches, keys)
|
||
operTypes = append(operTypes, operType)
|
||
}
|
||
if len(batches) == 0 {
|
||
return nil, nil, fmt.Errorf("批次为空")
|
||
}
|
||
return batches, operTypes, nil
|
||
}
|