1214181e78
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 16m2s
- 轮询步骤异步模式下只调网关并发布 card.ingest 入库消息，入库消费者固定并发 30、按采集 ID 去重、可重试错误不限次数退避重投 - 采集记录经 card.collection 攒批写入，失败整批重投，新增 (observation_id, observed_at) 唯一索引 - 网络、实名观测增加乱序保护；轮询读查询走缓存，套餐状态查询各自计时且失败不缓存 - 入库积压告警与反压；系统配置 polling.ingest.mode 默认 sync 可回退；测试环境 nsqd 消息全部落盘 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
90 lines
3.9 KiB
Go
90 lines
3.9 KiB
Go
package constants
|
||
|
||
import (
|
||
"fmt"
|
||
"time"
|
||
)
|
||
|
||
// 轮询观测入库异步化（add-polling-async-ingest）：调网关与入库分离，入库单独发消息、单独消费。
|
||
|
||
// 入库 NSQ topic。
|
||
const (
|
||
// NSQCardIngestTopic 轮询观测入库消息（卡状态、流量、实名）。
|
||
NSQCardIngestTopic = "card.ingest"
|
||
// NSQCardCollectionTopic 流量采集记录消息（成功表与失败表），由批量消费者攒批写入。
|
||
NSQCardCollectionTopic = "card.collection"
|
||
)
|
||
|
||
// 入库模式（受控系统配置 polling.ingest.mode）。
|
||
const (
|
||
PollingIngestModeSync = "sync" // 轮询步骤在卡任务内同步入库（回退模式）
|
||
PollingIngestModeAsync = "async" // 轮询步骤只调网关并发布入库消息
|
||
)
|
||
|
||
// 入库相关受控系统配置键（module = polling）。
|
||
const (
|
||
SystemConfigPollingIngestMode = "polling.ingest.mode"
|
||
SystemConfigPollingIngestAlertDepth = "polling.ingest.alert_depth"
|
||
SystemConfigPollingIngestPauseDepth = "polling.ingest.pause_depth"
|
||
)
|
||
|
||
// 入库运行默认值。
|
||
const (
|
||
PollingIngestDefaultConcurrency = 30
|
||
PollingIngestDefaultTimeout = 30 * time.Second
|
||
PollingCollectionDefaultBatchSize = 200
|
||
PollingCollectionFlushInterval = time.Second
|
||
PollingCollectionMaxInFlight = 500
|
||
PollingIngestDefaultAlertDepth = 200000
|
||
PollingIngestDefaultPauseDepth = 500000
|
||
PollingIngestDoneTTL = 48 * time.Hour
|
||
PollingIngestErrorLogFromAttempt = 10
|
||
PollingIngestLagAlert = 10 * time.Minute
|
||
PollingIngestMonitorInterval = time.Minute
|
||
PollingIngestConnReserve = 20
|
||
PollingIngestCollectionWriters = 1
|
||
PollingFlowEvaluatedFallbackTTL = 30 * time.Minute
|
||
PollingPriorityIndexRebuildEvery = 30 * time.Second
|
||
PollingPriorityIndexRecentTTL = 2 * time.Minute
|
||
PollingPriorityIndexReadyTTL = 2 * time.Minute
|
||
PollingScopeCacheTTL = 30 * time.Second
|
||
PollingCarrierResetDayCacheTTL = time.Minute
|
||
PollingTierQueryTimeout = 2 * time.Second
|
||
PollingIngestSyncCollectionTimeout = 10 * time.Second
|
||
)
|
||
|
||
// PollingIngestRetryDelays 是入库可重试错误的延迟重投序列；超过序列长度后取最后一项，不设次数上限。
|
||
var PollingIngestRetryDelays = []time.Duration{
|
||
5 * time.Second, 15 * time.Second, 30 * time.Second, time.Minute, 2 * time.Minute, 5 * time.Minute,
|
||
}
|
||
|
||
// RedisPollingIngestDoneKey 入库已处理标记（String，TTL 48h），按采集 ID 去重。
|
||
func RedisPollingIngestDoneKey(observationID string) string {
|
||
return "polling:ingest:done:" + observationID
|
||
}
|
||
|
||
// RedisPollingFlowEvaluatedKey 流量入库产生增量或跨月后的评估标记（String，TTL=套餐检查间隔），套餐步骤据此跳过。
|
||
func RedisPollingFlowEvaluatedKey(cardID uint) string {
|
||
return fmt.Sprintf("polling:flow_evaluated:%d", cardID)
|
||
}
|
||
|
||
// RedisPollingIngestPausedKey 入库积压超过反压阈值时置位（String），分发器暂停领取新卡任务。
|
||
const RedisPollingIngestPausedKey = "polling:ingest:paused"
|
||
|
||
// 有活动优先轮询项的卡索引：member = "card_id:task_type"。
|
||
const (
|
||
// RedisPollingPriorityActiveKey 活动优先项索引（SET）。
|
||
RedisPollingPriorityActiveKey = "polling:priority:active"
|
||
// RedisPollingPriorityRecentKey 最近入队的优先项（SET，短 TTL），重建索引时并入，避免重建期间漏掉新项。
|
||
RedisPollingPriorityRecentKey = "polling:priority:recent"
|
||
// RedisPollingPriorityReadyKey 索引已由重建任务建立（String，TTL）；不存在时轮询步骤回退查库。
|
||
RedisPollingPriorityReadyKey = "polling:priority:ready"
|
||
// RedisPollingPriorityRebuildKey 重建过程使用的临时 key。
|
||
RedisPollingPriorityRebuildKey = "polling:priority:active:rebuild"
|
||
)
|
||
|
||
// PollingPriorityIndexMember 返回活动优先项索引成员。
|
||
func PollingPriorityIndexMember(cardID uint, taskType string) string {
|
||
return fmt.Sprintf("%d:%s", cardID, taskType)
|
||
}
|