aggregator_submit

README

聚合提交服务说明（aggregator_submit）

本目录实现的是「用户流量聚合 + 定时上报」的服务，核心文件：

• service.go：服务入口、worker、定时器、调用提交接口 SubmitWithAgent
• model.go：数据结构与并发控制（双缓冲：active / sealed）

下面重点说明为什么要用双缓冲以及完整的数据流动过程。

---

一、整体设计目标

• 不丢数据：任何已产生的流量要么已经上报，要么一定还保存在内存/磁盘中等待上报。
• 尽量不重复：失败重试时，最多会对同一批数据重复上报（可以通过服务端幂等批次 ID 抹平）。
• 并发安全：支持多个 worker 并发 AddRequest。
• 与持久化兼容：saveData / loadData 的数据格式不变化。

---

二、双缓冲核心概念：active / sealed

对每个 (nodeType, registerID)，内部维护一份 RegisterBuffers：

RegisterBuffers (for 某个 registerID)

 +----------------------------+
 |        RegisterBuffers     |
 +----------------------------+
 |                            |
 |   +----------+             |
 |   | active   |  正在累加   |  <- AddRequest/Update 写入
 |   +----------+             |
 |                            |
 |   +----------+             |
 |   | sealed   |  已封口     |  -> submitData 提交
 |   +----------+             |
 +----------------------------+

• active：当前正在写入的新流量。
• sealed：已经封口、等待（或正在）提交的流量。

---

三、关键数据结构（简化版）

NodeTypeDataMap
  └── internalMap: map[nodeType]*RequestDataMap

RequestDataMap
  └── internalMap: map[registerID]*RegisterBuffers

RegisterBuffers
  ├── active: *UserDataMap    // 正在写
  └── sealed: *UserDataMap    // 等待/重试提交

UserDataMap
  └── internalMap: map[userID]*TrafficData

• TrafficData：单个用户的流量聚合（Up/Down/Count）。
• NodeTypeDataMap.Update()：入口；所有 AddRequest 最终都会走到这里，把数据写入对应 RegisterBuffers.active。

---

四、时间轴：一次完整的提交周期

以某个 registerID = R1 为例，看从 T0 到 T2 的整个过程。

1. T0 ~ T1：收集阶段（只动 active）

时间轴 →
T0                 T1
|------------------|
  业务 AddRequest 不停到达

状态示意：

R1:
  active: [A1, A2, A3, ...]   // 本周期累加的流量
  sealed: [S_old]             // 上一轮未成功提交的旧数据（可能为空）

业务调用链：

AddRequest(...)
  -> Service.dataChan
    -> startProcessing()
      -> NodeTypeDataMap.Update(...)
        -> RequestDataMap.Update(...)
          -> RegisterBuffers.active.Update(...)

2. T1：定时器触发 submitData()，调用 SealAndGet()

service.go 中的 submitData()：

dataCopy := s.dataMap.SealAndGet()

SealAndGet() 内部对每个 (nodeType, registerID) 做：

1. activeData := active.RetrieveAndReset()
2. sealed.AddFromMap(activeData)
3. snapshot := sealed.RetrieveWithoutReset()
4. 把 snapshot 放入返回的 dataCopy 中

对于 R1，状态从：

调用前：
  active: [A1, A2, A3, ...]
  sealed: [S_old]

调用后：
  active: []                          // 被 ResetAll，重新变空
  sealed: [S_old + A1 + A2 + A3 ...]  // 所有要提交的数据都到 sealed 里

本轮要提交的快照 = sealed 的一份拷贝

重要点：

• 从这一刻起，本轮要提交的数据已经冻结在 sealed 中。
• 接下来的新流量会写入新的 active，不会再影响本轮 sealed 的内容。

3. T1 ~ T2：对 sealed 的快照执行 HTTP 提交

submitData() 遍历 dataCopy：

err := s.api.SubmitWithAgent(registerID, api.NodeType(nodeType), submitData)

此时 R1 的状态：

R1:
  active: [T1 之后新产生的流量，不断累加]
  sealed: [本轮要提交的数据，保持不变]

---

五、提交成功 vs 提交失败的处理

情况 A：提交成功

submitData() 中：

if err == nil {
    if successfulSubmissions[nodeType] == nil {
        successfulSubmissions[nodeType] = make(map[string]bool)
    }
    successfulSubmissions[nodeType][registerID] = true
}
...
s.dataMap.ClearSealed(successfulSubmissions)

ClearSealed() 对 R1：

R1:
  sealed.ResetAll()        // 清空 sealed，表示这批数据已成功上报
  active: 保持不变         // T1 之后新累加的数据还在 active 里

下一轮定时器再次 SealAndGet()：

active 中（T1~T_next）的新流量会被移动到 sealed
上一轮 sealed 已是空，不会重复上报

情况 B：提交失败

submitData() 中：

if err != nil {
    log.Errorf("submit error:%v\n", err)
    // 不记录到 successfulSubmissions，后面也不会 ClearSealed
}

因此 R1：

R1:
  active: 继续累加 T1 之后的新流量
  sealed: 保持不变（仍保存“此次没成功提交的那一批数据”）

下一轮 SealAndGet()：

1. activeData := active.RetrieveAndReset()   // T1~T_next 的新流量
2. sealed.AddFromMap(activeData)            // sealed = 上次失败的 + 本轮新 seal 的
3. 对新的 sealed 做快照再提交

这样保证：

• 失败不会丢数据：sealed 始终保存所有“截至上一次 seal 为止、尚未成功上报”的数据。
• 提交成功才清理：只有当 server 端确认成功时，对应 sealed 才会被 ClearSealed 清空。

---

六、与 saveData / loadData 的兼容性

saveData：仍然是「当前所有未完全上报的总流量快照」

service.go：

// 获取当前内存中的数据（不重置）
dataToSave := s.dataMap.RetrieveWithoutReset()

RetrieveWithoutReset() 的语义：

• 对每个 (nodeType, registerID)：
  • 取 sealed 的快照；
  • 取 active 的快照；
  • 两者按用户维度合并成一个 UserTrafficMap。
• 组合成一个 NodeTypeTrafficMap 返回。

因此，磁盘上的 JSON 格式与原来完全一致，表示：

截至当前时刻，所有尚未成功上报的流量（包含 sealed + active）。

loadData：恢复为 active，再按正常流程走

service.go：

for nodeType, requestMap := range data {
    for registerID, userMap := range requestMap {
        for userID, trafficData := range userMap {
            s.dataMap.Update(nodeType, registerID, userID, trafficData)
        }
    }
}

• Update() 只写入 active，sealed 初始为空。
• 重启后第一次 SealAndGet()：
  • 会把这些恢复出来的 active 数据 seal 进 sealed；
  • 然后按双缓冲流程正常提交。

这保证了：

• 持久化格式不变；
• 重启后之前未上报的数据会被重新上报一次（幂等需要服务端处理），但不会被静默丢弃。

---

七、与 SubmitWithAgent 的配合

SubmitWithAgent 本身只是一个 HTTP 请求封装：

• 不做裁剪、采样或「减半」等运算；
• 接收到的 userTraffic 数组就是 sealed 中的那一份快照。

结合双缓冲：

• 所有要提交的数据都通过 sealed 统一发给 SubmitWithAgent；
• 提交成功才会清空 sealed；
• 提交失败时仅保留 sealed、等待下次重试。

如需进一步提升幂等性，可以在请求体里增加一个批次 ID（如时间戳或自增序号），让服务端对 (register_id, batch_id) 做幂等处理，避免失败重试导致的重复计费。

Documentation

Index

• Constants
• type BatchEntry
• type NodeTypeDataMap
  • func NewNodeTypeDataMap() *NodeTypeDataMap
  • func (ntdm *NodeTypeDataMap) ExportState() *SavedState
  • func (ntdm *NodeTypeDataMap) ImportLegacy(data NodeTypeTrafficMap)
  • func (ntdm *NodeTypeDataMap) ImportState(state *SavedState)
  • func (ntdm *NodeTypeDataMap) IncrementBatchRetry(nodeType, registerID, batchID string) int
  • func (ntdm *NodeTypeDataMap) Pending() map[string]map[string][]*PendingBatch
  • func (ntdm *NodeTypeDataMap) RemoveBatch(nodeType, registerID, batchID string)
  • func (ntdm *NodeTypeDataMap) ResetAllBatchRetries()
  • func (ntdm *NodeTypeDataMap) RetrieveWithoutReset() NodeTypeTrafficMap
  • func (ntdm *NodeTypeDataMap) SealIdle(batchSize int, gen func(registerID string) string)
  • func (ntdm *NodeTypeDataMap) Update(nodeType string, registerID string, userID int, data *TrafficData)
• type NodeTypeTrafficMap
• type PendingBatch
• type RegisterBuffers
• type RequestDataMap
  • func NewRequestDataMap() *RequestDataMap
  • func (rdm *RequestDataMap) RetrieveWithoutReset() RequestTrafficMap
  • func (rdm *RequestDataMap) Update(registerID string, userID int, data *TrafficData)
• type RequestTrafficMap
• type SavedRegisterState
• type SavedState
• type Service
  • func NewService(api *api.Client, submitInterval time.Duration, bufferSize int, workerNum int, ...) (*Service, error)
  • func (s *Service) AddRequest(nodeType string, registerID string, userID int, data *TrafficData)
  • func (s *Service) Destroy() error
  • func (s *Service) Init() error
  • func (s *Service) Start() error
  • func (s *Service) Stop() error
  • func (s *Service) String() string
• type TrafficData
• type UserDataMap
  • func NewUserDataMap() *UserDataMap
  • func (udm *UserDataMap) RetrieveAndReset() UserTrafficMap
  • func (udm *UserDataMap) RetrieveWithoutReset() UserTrafficMap
  • func (udm *UserDataMap) Update(userID int, data *TrafficData)
• type UserTrafficMap

Constants

const (
	ServiceName           = "srv_aggregator_submit"
	DefaultBufferSize     = 2048
	DefaultWorkerNum      = 0
	DefaultSubmitInterval = 5 * time.Minute
	DefaultBatchSize      = 5000             // 每批次最大提交用户数，约400KB，避免请求体过大
	DefaultMaxRetries     = 3                // 最大重试次数，超过后丢弃数据
	DefaultSubmitTimeout  = 30 * time.Second // 单次提交超时，兜底防止请求卡死
)

const SaveFormatVersion = 2

SaveFormatVersion is the on-disk persistence format version. Version 2 introduced the frozen-batch model (active buffer + sealed batches with stable batch IDs). Files without a version field are treated as legacy (plain traffic, loaded into the active buffer).

Types

type BatchEntry

type BatchEntry struct {
	UID  int    `json:"uid"`
	Up   uint64 `json:"up"`
	Down uint64 `json:"down"`
	N    uint64 `json:"n"`
}

BatchEntry is one user's frozen traffic inside a sealed batch.

type NodeTypeDataMap

type NodeTypeDataMap struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewNodeTypeDataMap

func NewNodeTypeDataMap() *NodeTypeDataMap

func (*NodeTypeDataMap) ExportState

func (ntdm *NodeTypeDataMap) ExportState() *SavedState

ExportState snapshots the full buffer state for persistence.

func (*NodeTypeDataMap) ImportLegacy

func (ntdm *NodeTypeDataMap) ImportLegacy(data NodeTypeTrafficMap)

ImportLegacy loads pre-v2 persisted traffic (plain nodeType/registerID/user) into the active buffer. Such data is re-sealed with fresh batch IDs.

func (*NodeTypeDataMap) ImportState

func (ntdm *NodeTypeDataMap) ImportState(state *SavedState)

ImportState restores buffer state produced by ExportState.

func (*NodeTypeDataMap) IncrementBatchRetry

func (ntdm *NodeTypeDataMap) IncrementBatchRetry(nodeType, registerID, batchID string) int

IncrementBatchRetry bumps and returns the retry counter for a batch.

func (*NodeTypeDataMap) Pending

func (ntdm *NodeTypeDataMap) Pending() map[string]map[string][]*PendingBatch

Pending returns all in-flight batches as nodeType -> registerID -> batches.

func (*NodeTypeDataMap) RemoveBatch

func (ntdm *NodeTypeDataMap) RemoveBatch(nodeType, registerID, batchID string)

RemoveBatch drops a delivered/discarded batch.

func (*NodeTypeDataMap) ResetAllBatchRetries

func (ntdm *NodeTypeDataMap) ResetAllBatchRetries()

ResetAllBatchRetries clears retry counters on every in-flight batch.

func (*NodeTypeDataMap) RetrieveWithoutReset

func (ntdm *NodeTypeDataMap) RetrieveWithoutReset() NodeTypeTrafficMap

RetrieveWithoutReset returns merged active + sealed traffic, for inspection.

func (*NodeTypeDataMap) SealIdle

func (ntdm *NodeTypeDataMap) SealIdle(batchSize int, gen func(registerID string) string)

SealIdle freezes idle active buffers into new batch generations across all nodeTypes/registerIDs. gen produces a stable batch_id for each new batch.

func (*NodeTypeDataMap) Update

func (ntdm *NodeTypeDataMap) Update(nodeType string, registerID string, userID int, data *TrafficData)

type NodeTypeTrafficMap

type NodeTypeTrafficMap map[string]RequestTrafficMap

type PendingBatch

type PendingBatch struct {
	BatchID string       `json:"batch_id"`
	Entries []BatchEntry `json:"entries"`
	Retries int          `json:"retries"`
}

PendingBatch is an immutable, sealed chunk of traffic awaiting delivery.

BatchID is generated once at seal time and reused on every retry so the server can deduplicate idempotently (it dedupes by batch_id). Entries never change after sealing: new traffic for the same users accumulates in the active buffer and forms a different batch in a later generation.

type RegisterBuffers

type RegisterBuffers struct {
	// contains filtered or unexported fields
}

RegisterBuffers maintains the double-buffer for one registerID:

• active: new traffic still accumulating
• sealed: at most one frozen generation of batches awaiting delivery

Single-generation invariant: while sealed is non-empty the active buffer is NOT sealed again. New traffic keeps accumulating in active and is only sealed once the current generation has fully drained. This keeps each batch's content frozen so its stable batch_id stays valid across retries.

type RequestDataMap

type RequestDataMap struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewRequestDataMap

func NewRequestDataMap() *RequestDataMap

func (*RequestDataMap) RetrieveWithoutReset

func (rdm *RequestDataMap) RetrieveWithoutReset() RequestTrafficMap

RetrieveWithoutReset returns the merged active + sealed traffic per user without mutating state. Used for inspection and tests.

func (*RequestDataMap) Update

func (rdm *RequestDataMap) Update(registerID string, userID int, data *TrafficData)

Update writes new traffic into the active buffer.

type RequestTrafficMap

type RequestTrafficMap map[string]UserTrafficMap

type SavedRegisterState

type SavedRegisterState struct {
	Active UserTrafficMap  `json:"active"`
	Sealed []*PendingBatch `json:"sealed"`
}

SavedRegisterState is the persisted form of one registerID's buffers.

type SavedState

type SavedState struct {
	Version int                                       `json:"version"`
	Data    map[string]map[string]*SavedRegisterState `json:"data"`
}

SavedState is the versioned on-disk format. Data is nodeType -> registerID -> state.

type Service

type Service struct {
	// contains filtered or unexported fields
}

func NewService

func NewService(api *api.Client, submitInterval time.Duration, bufferSize int, workerNum int, dataDir string, port int) (*Service, error)

func (*Service) AddRequest

func (s *Service) AddRequest(nodeType string, registerID string, userID int, data *TrafficData)

func (*Service) Destroy

func (s *Service) Destroy() error

func (*Service) Init

func (s *Service) Init() error

func (*Service) Start

func (s *Service) Start() error

func (*Service) Stop

func (s *Service) Stop() error

func (*Service) String

func (s *Service) String() string

type TrafficData

type TrafficData struct {
	Up   uint64
	Down uint64
	N    uint64
}

type UserDataMap

type UserDataMap struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

UserDataMap is the active buffer: a mutable map accumulating new traffic.

func NewUserDataMap

func NewUserDataMap() *UserDataMap

func (*UserDataMap) RetrieveAndReset

func (udm *UserDataMap) RetrieveAndReset() UserTrafficMap

RetrieveAndReset returns the non-zero traffic and replaces each retrieved slot with a fresh pooled zero value, leaving the buffer ready for new traffic.

func (*UserDataMap) RetrieveWithoutReset

func (udm *UserDataMap) RetrieveWithoutReset() UserTrafficMap

RetrieveWithoutReset returns a value copy of the non-zero traffic without mutating the buffer.

func (*UserDataMap) Update

func (udm *UserDataMap) Update(userID int, data *TrafficData)

type UserTrafficMap

type UserTrafficMap map[int]*TrafficData