openai_profit_control.go 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460
  1. package service
  2. // 分组利润控制(配套 migration 192/193 的 groups.profit_* 字段)。
  3. //
  4. // 定位:利润控制是"候选准入过滤",只决定账号能否进入调度候选池;既有的排序、
  5. // 评分、粘性、熔断、负载均衡在合格账号之间照常工作,本文件不改变它们的行为。
  6. //
  7. // 准入条件:
  8. //
  9. // U(尝试时刻) <= D(pricingAt) × (1 − profit_min_margin − profit_safety_buffer)
  10. //
  11. // - D(用户售价倍率)固定在请求开始的 pricingAt:同一请求的全部 failover 与
  12. // 最终扣费共用同一 D(RecordUsage 的高峰因子同样取 pricingAt),一个请求
  13. // 不会中途变价。D 与计费完全同源:按请求真实计费分组(ctxkey.Group,即
  14. // apiKey 自身分组;composite 请求为父分组)做 ResolveUserGroupRateMultiplier
  15. // (用户-分组覆盖 ?? 分组默认)× Group.PeakMultiplierAt(pricingAt),绝不在
  16. // 用户有覆盖时退回分组默认;开关与 margin/buffer 则始终取被调度
  17. // openai/grok 分组。
  18. // - U(上游成本倍率)取 accounts.rate_multiplier。倍率可以由运营者手工维护,
  19. // 也可以由上游倍率探测同步写回;利润门不再耦合探测协议、新鲜度或账号类型。
  20. // 0 是合法的免费上游倍率;nil、负数、NaN、Inf 属于非法数据并保守拒绝。
  21. //
  22. // 装门点(gate 随 ctx 传播,请求内复用,覆盖等待/重试/failover/抢槽后终检):
  23. // - handler 各文本入口经 WithOpenAIRequestPricingContext 在请求开始统一装门并
  24. // 固定 pricingAt。生图意图只用于能力路由与图片计费,不决定是否装门:混合
  25. // /v1/responses 请求(含声明了生图工具的请求)的 token 计费部分仍受利润门
  26. // 保护,请求体内的任何工具声明都不能作为关门开关。
  27. // - 门范围之外的路径显式携带 WithOpenAIProfitControlSuppressed 标记:独立
  28. // 图片/视频端点与 Grok 媒体(媒体计费另有倍率来源)、count_tokens(不计费)、
  29. // live(按时长计费)。所有装门点(含防御性装门)都尊重该标记。
  30. // - selectAccountWithScheduler 顶部:唯一文本调度入口的防御性装门(ctx 已有
  31. // 同分组门则复用,保证 failover 阈值稳定);requiredImageCapability != ""
  32. // 的专用图片调度不装门。
  33. // - 公开 SelectAccountWithLoadAwareness / SelectAccountByPreviousResponseID:
  34. // 防御性装门,保证不经唯一入口的调用方无法绕过。
  35. // - 选号结果通过 AccountSelectionResult 携带真实生效的门(composite/fallback
  36. // 路由可能解析出与入口分组不同的门),handler 经
  37. // ContextWithSelectionProfitGate 重放后再做抢槽后终检与准入后粘性绑定。
  38. // - 长连接(Responses WS)在每个 turn 开始经 WithOpenAITurnPricingContext
  39. // 重新冻结 pricingAt 并重装门:turn 的准入与计费同源,跨峰谷边界不再共用
  40. // 建连时刻的 D。
  41. //
  42. // 否决点(消费 gate,任何 fallback 都无法把已排除账号重新放回):
  43. // - defaultOpenAIAccountScheduler.isAccountRequestCompatibleReason:候选池
  44. // 过滤 + 调度器内抢槽后终检共用,named reason 进入 openAISelectionFilterStats。
  45. // - isOpenAICompatibleAccountEligibleForRequest:legacy 引擎与 DB recheck 共用。
  46. // - resolveAccountByPreviousResponseIDForCapability:previous_response 粘连
  47. // 两阶段校验;与 quota auto-pause 同语义,跳过复用但不删除绑定。
  48. // - handler 槽位获取后终检(OpenAIProfitControlVeto):快速抢槽与 WaitPlan
  49. // 排队成功后复核,越线则释放槽位、加入本请求排除集重新选号,全池耗尽才
  50. // 返回标准 no available accounts。
  51. //
  52. // 失败语义:分组配置读取失败时放行并告警(fail-open)。这是"配置系统故障时
  53. // 可用性优先"的显式取舍——该异常窗口内利润保证不成立,靠 WARN 与采样观测
  54. // 暴露,绝不把瞬时 DB 抖动放大成全站不可调度。
  55. //
  56. // 可观测性:按分组和平台累计装门/threshold 否决/invalid-rate 否决/终检刷新
  57. // 失败计数,≥5 分钟采样输出一条 Info(profit_control_activity),无逐请求日志。
  58. import (
  59. "context"
  60. "errors"
  61. "fmt"
  62. "log/slog"
  63. "math"
  64. "sync"
  65. "sync/atomic"
  66. "time"
  67. "github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey"
  68. "github.com/Wei-Shaw/sub2api/internal/pkg/timezone"
  69. )
  70. const (
  71. // profitControlRateEpsilon 吸收 decimal(10,4) 落库与浮点乘法的边界误差:
  72. // U 与阈值的相对差在该量级内视为相等(即 U == 阈值 判定为合格)。
  73. profitControlRateEpsilon = 1e-9
  74. // 进入 openAISelectionFilterStats 的排除原因,全排除时出现在
  75. // "no available accounts" 的内部统计摘要里(与 quota_auto_pause 等同通道)。
  76. openAIProfitFilterReasonThreshold = "profit_threshold"
  77. openAIProfitFilterReasonInvalidAccountRate = "profit_invalid_account_rate"
  78. // profitControlActivityLogInterval 是按分组采样输出累计计数的最小间隔。
  79. profitControlActivityLogInterval = 5 * time.Minute
  80. )
  81. type openAIProfitControlGateCtxKey struct{}
  82. // openAIProfitControlSuppressCtxKey 标记本请求显式跳过利润门(独立图片/视频
  83. // 端点、Grok 媒体、count_tokens、live 等利润门范围外流量)。所有装门点看到该
  84. // 标记后一律不装门,防止 service 层防御性装门把边界外流量重新拉回利润过滤。
  85. type openAIProfitControlSuppressCtxKey struct{}
  86. // openAIPricingAtCtxKey 携带请求级定价时刻 pricingAt:门的 D 与 RecordUsage
  87. // 的高峰因子共用,保证一个请求从准入到扣费不中途变价。
  88. type openAIPricingAtCtxKey struct{}
  89. // clampProfitControlThreshold 归一化利润门阈值。Validate/Normalize 已保证
  90. // margin+buffer < 1,这里只对存量脏数据兜底:阈值非有限或为负时按 0 处理
  91. // (等价于只放行免费上游)。装门点与 profit-preview 共用,避免口径漂移。
  92. func clampProfitControlThreshold(threshold float64) float64 {
  93. if math.IsNaN(threshold) || math.IsInf(threshold, 0) || threshold < 0 {
  94. return 0
  95. }
  96. return threshold
  97. }
  98. // profitControlOverThreshold 是"上游倍率越线"的唯一定义:相对 epsilon 吸收
  99. // decimal(10,4) 落库与浮点乘法的边界误差,U == 阈值 判定为合格。
  100. // 线上否决点与 profit-preview 共用,两者不得各自实现。
  101. func profitControlOverThreshold(upstream, threshold float64) bool {
  102. return upstream-threshold > profitControlRateEpsilon*math.Max(1, math.Abs(threshold))
  103. }
  104. // openAIProfitControlGate 是一个请求的利润准入门。除 pricingAt 外全部为预计算
  105. // 标量:候选过滤热路径上每账号只做一次快照解码与一次浮点比较。
  106. type openAIProfitControlGate struct {
  107. // groupID 是门配置来源的被调度分组;请求内按分组复用(failover 阈值稳定),
  108. // composite 等跨分组调度切换分组时重新解析。
  109. groupID int64
  110. // platform 是利润配置所在分组的平台,用于按平台观测门是否真实生效。
  111. platform string
  112. // threshold = D(pricingAt) × (1 − margin − buffer),账号倍率必须 <= 它。
  113. threshold float64
  114. // pricingAt 是本请求的统一定价时刻(D 侧)。
  115. pricingAt time.Time
  116. }
  117. // WithOpenAIRequestPricingContext 在请求开始处装配请求级定价上下文:固定
  118. // pricingAt(返回给调用方,供 RecordUsage 入参共用同一时刻),并按分组安装
  119. // 利润门。ctx 携带 WithOpenAIProfitControlSuppressed 标记(门范围外流量)时
  120. // 只固定 pricingAt、不装门。handler 各文本入口应在选号循环前调用一次。
  121. func (s *OpenAIGatewayService) WithOpenAIRequestPricingContext(ctx context.Context, groupID *int64) (context.Context, time.Time) {
  122. pricingAt := timezone.Now()
  123. ctx = context.WithValue(ctx, openAIPricingAtCtxKey{}, pricingAt)
  124. return s.withOpenAIProfitControlGate(ctx, groupID), pricingAt
  125. }
  126. // WithOpenAIProfitControlSuppressed 标记本请求在利润门范围之外(独立图片/视频
  127. // 端点、Grok 媒体、count_tokens、live)。所有装门点(含 service 层防御性装门)
  128. // 都尊重该标记;它只关闭利润准入过滤,不影响定价上下文与计费。
  129. func WithOpenAIProfitControlSuppressed(ctx context.Context) context.Context {
  130. return context.WithValue(ctx, openAIProfitControlSuppressCtxKey{}, struct{}{})
  131. }
  132. // WithOpenAITurnPricingContext 在长连接(Responses WS)的每个 turn 开始重新
  133. // 冻结 pricingAt 并按当前配置重装利润门,使 turn 的准入与计费同源:峰前建连
  134. // 保活不再让后续 turn 继续按建连时刻的谷价定价。连接可能被调度到与入口分组
  135. // 不同的分组(composite 成员分组),turn 级重装以连接上已装门的调度分组为准;
  136. // 连接从未装门时才回退入口分组。抑制标记下只刷新 pricingAt。
  137. func (s *OpenAIGatewayService) WithOpenAITurnPricingContext(ctx context.Context, groupID *int64) (context.Context, time.Time) {
  138. pricingAt := timezone.Now()
  139. ctx = context.WithValue(ctx, openAIPricingAtCtxKey{}, pricingAt)
  140. if _, suppressed := ctx.Value(openAIProfitControlSuppressCtxKey{}).(struct{}); suppressed {
  141. return ctx, pricingAt
  142. }
  143. if existing, ok := ctx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate); ok && existing != nil {
  144. gid := existing.groupID
  145. groupID = &gid
  146. }
  147. gate := s.resolveOpenAIProfitControlGate(ctx, groupID)
  148. if gate == nil {
  149. // 分组已关门(或配置读取失败 fail-open):清除旧 turn 的门,后续 turn
  150. // 按无门放行,与 HTTP 路径的开关语义一致。
  151. if existing, ok := ctx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate); ok && existing != nil {
  152. return context.WithValue(ctx, openAIProfitControlGateCtxKey{}, (*openAIProfitControlGate)(nil)), pricingAt
  153. }
  154. return ctx, pricingAt
  155. }
  156. openAIProfitControlObserverInstance.recordInstall(gate.groupID, gate.platform, gate.threshold)
  157. return context.WithValue(ctx, openAIProfitControlGateCtxKey{}, gate), pricingAt
  158. }
  159. // openAIPricingAtFromContext 返回请求级定价时刻;未装配(内部调用、非文本
  160. // 入口)时 ok=false,调用方回退 timezone.Now() 保持既有行为。
  161. func openAIPricingAtFromContext(ctx context.Context) (time.Time, bool) {
  162. pricingAt, ok := ctx.Value(openAIPricingAtCtxKey{}).(time.Time)
  163. return pricingAt, ok && !pricingAt.IsZero()
  164. }
  165. // OpenAIPricingAtFromContext 是 handler 侧读取请求级定价时刻的公开入口(未装配
  166. // 时为零值,RecordUsage 对零值回退记录时刻)。供跨函数传递 pricingAt 不便的
  167. // 记录路径直接从请求 ctx 取值。
  168. func OpenAIPricingAtFromContext(ctx context.Context) time.Time {
  169. pricingAt, _ := openAIPricingAtFromContext(ctx)
  170. return pricingAt
  171. }
  172. // withOpenAIProfitControlGate 解析分组利润控制配置;启用时把预计算好的准入门
  173. // 装进 ctx。抑制标记、未启用/非 openai 分组/无法取到分组配置时原样返回 ctx
  174. // (门不存在,全部否决点自动放行,既有行为零变化)。ctx 已有同分组门时直接
  175. // 复用:同一请求的全部 failover 重入共享同一阈值。
  176. func (s *OpenAIGatewayService) withOpenAIProfitControlGate(ctx context.Context, groupID *int64) context.Context {
  177. if _, suppressed := ctx.Value(openAIProfitControlSuppressCtxKey{}).(struct{}); suppressed {
  178. return ctx
  179. }
  180. if groupID != nil {
  181. if existing, ok := ctx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate); ok && existing != nil && existing.groupID == *groupID {
  182. return ctx
  183. }
  184. }
  185. gate := s.resolveOpenAIProfitControlGate(ctx, groupID)
  186. if gate == nil {
  187. // 被调度分组无门(未启用/非 openai/配置读取失败)而 ctx 带着其他分组的
  188. // 请求门时清除之:门配置取被调度分组,父分组阈值不得泄漏到成员分组
  189. //(composite/模型路由等跨分组调度)。typed-nil 覆盖值由否决点按无门放行。
  190. if existing, ok := ctx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate); ok && existing != nil && groupID != nil && existing.groupID != *groupID {
  191. return context.WithValue(ctx, openAIProfitControlGateCtxKey{}, (*openAIProfitControlGate)(nil))
  192. }
  193. return ctx
  194. }
  195. openAIProfitControlObserverInstance.recordInstall(gate.groupID, gate.platform, gate.threshold)
  196. return context.WithValue(ctx, openAIProfitControlGateCtxKey{}, gate)
  197. }
  198. func (s *OpenAIGatewayService) resolveOpenAIProfitControlGate(ctx context.Context, groupID *int64) *openAIProfitControlGate {
  199. if s == nil || groupID == nil || *groupID <= 0 {
  200. return nil
  201. }
  202. // 门配置取被调度分组。直连请求(ctx 认证分组即调度分组,生产绝大多数流量)
  203. // 直接复用 auth cache 分组,热路径零额外查询;composite 父分组路由到成员
  204. // 分组等 ID 不一致场景才回源仓库读取。auth 快照的分组字段完备性由
  205. // GetByKeyForAuth 投影 + 集成测试保证(防投影漏列导致门静默失效)。
  206. var group *Group
  207. if ctxGroup, ok := ctx.Value(ctxkey.Group).(*Group); ok && IsGroupContextValid(ctxGroup) && ctxGroup.ID == *groupID {
  208. group = ctxGroup
  209. } else if s.schedulerSnapshot != nil {
  210. // Lite 读取:门只用平台/倍率/利润/高峰字段,不需要账号计数聚合。
  211. loaded, err := s.schedulerSnapshot.GetGroupByIDLite(ctx, *groupID)
  212. if err != nil {
  213. // fail-open:配置系统故障时可用性优先,该窗口内利润保证不成立,
  214. // 依赖 WARN 暴露;不把瞬时 DB 抖动放大成全站不可调度。
  215. slog.Warn("profit_control_group_load_failed", "group_id", *groupID, "error", err)
  216. return nil
  217. }
  218. group = loaded
  219. }
  220. if group == nil || !group.ProfitControlEnabled ||
  221. (group.Platform != PlatformOpenAI && group.Platform != PlatformGrok) {
  222. return nil
  223. }
  224. pricingAt, ok := openAIPricingAtFromContext(ctx)
  225. if !ok {
  226. pricingAt = timezone.Now()
  227. }
  228. // D 与计费完全同源(RecordUsage 组合):计费永远按 apiKey 自身分组
  229. //(composite 请求即父分组)的"用户覆盖 ?? 分组默认 × 高峰因子"计算,
  230. // 因此优先取认证中间件放入 ctx 的分组;ctx 中无有效分组(内部调用)时
  231. // 退回调度分组组合,直连 openai 分组场景两者等价。
  232. billingGroup := group
  233. if ctxGroup, ok := ctx.Value(ctxkey.Group).(*Group); ok && IsGroupContextValid(ctxGroup) {
  234. billingGroup = ctxGroup
  235. }
  236. downstream := billingGroup.RateMultiplier
  237. if userID, _ := ctx.Value(ctxkey.UserID).(int64); userID > 0 {
  238. downstream = s.ResolveUserGroupRateMultiplier(ctx, userID, billingGroup.ID, billingGroup.RateMultiplier)
  239. }
  240. downstream *= billingGroup.PeakMultiplierAt(pricingAt)
  241. deduction := group.ProfitMinMargin + group.ProfitSafetyBuffer
  242. threshold := clampProfitControlThreshold(downstream * (1 - deduction))
  243. return &openAIProfitControlGate{
  244. groupID: *groupID,
  245. platform: group.Platform,
  246. threshold: threshold,
  247. pricingAt: pricingAt,
  248. }
  249. }
  250. // attachSelectionProfitGate 把调度上下文里生效的利润门记录到选号结果上。门在
  251. // 选号函数内部的局部 ctx 上安装(composite/fallback 路由还可能解析出与入口
  252. // 分组不同的门),不随返回值离开调度栈;结果携带后 handler 才能对"真实过滤了
  253. // 候选的那个门"做抢槽后终检与准入后绑定。
  254. func attachSelectionProfitGate(ctx context.Context, sel *AccountSelectionResult) *AccountSelectionResult {
  255. if sel == nil {
  256. return nil
  257. }
  258. if gate, ok := ctx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate); ok && gate != nil {
  259. sel.profitGate = gate
  260. }
  261. return sel
  262. }
  263. // ContextWithSelectionProfitGate 把选号时真实生效的利润门重放到 ctx 上。
  264. // handler 在拿到选号结果后必须用返回的 ctx 做抢槽后终检
  265. // (ProfitControlVetoLatest / GatewayProfitControlVetoLatest)与准入后粘性
  266. // 绑定,否则这两步会因为看不到调度栈内安装的门而退化为空操作。
  267. func ContextWithSelectionProfitGate(ctx context.Context, sel *AccountSelectionResult) context.Context {
  268. if sel == nil || sel.profitGate == nil {
  269. return ctx
  270. }
  271. if existing, ok := ctx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate); ok && existing == sel.profitGate {
  272. return ctx
  273. }
  274. return context.WithValue(ctx, openAIProfitControlGateCtxKey{}, sel.profitGate)
  275. }
  276. // openAIProfitControlVetoReason 报告利润门是否否决该账号。ctx 中没有门
  277. // (分组未启用利润控制或本请求跳门)或账号为 nil 时一律放行。
  278. func openAIProfitControlVetoReason(ctx context.Context, account *Account) (bool, string) {
  279. gate, _ := ctx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate)
  280. if gate == nil || account == nil {
  281. return false, ""
  282. }
  283. if account.RateMultiplier == nil ||
  284. math.IsNaN(*account.RateMultiplier) ||
  285. math.IsInf(*account.RateMultiplier, 0) ||
  286. *account.RateMultiplier < 0 {
  287. openAIProfitControlObserverInstance.recordVeto(gate.groupID, gate.platform, gate.threshold, openAIProfitFilterReasonInvalidAccountRate)
  288. return true, openAIProfitFilterReasonInvalidAccountRate
  289. }
  290. upstream := *account.RateMultiplier
  291. if profitControlOverThreshold(upstream, gate.threshold) {
  292. openAIProfitControlObserverInstance.recordVeto(gate.groupID, gate.platform, gate.threshold, openAIProfitFilterReasonThreshold)
  293. return true, openAIProfitFilterReasonThreshold
  294. }
  295. return false, ""
  296. }
  297. // OpenAIProfitControlVeto 是 handler 层槽位获取后终检的公开入口:语义与调度
  298. // 内否决点完全一致。ctx 必须是经 WithOpenAIRequestPricingContext 装配过的
  299. // 请求上下文,否则视为无门放行。
  300. func OpenAIProfitControlVeto(ctx context.Context, account *Account) (bool, string) {
  301. return openAIProfitControlVetoReason(ctx, account)
  302. }
  303. // ProfitControlVetoLatest performs the handler-side terminal check after a
  304. // concurrency slot is actually acquired. The latest cached account replaces
  305. // the selection snapshot when available, so a probe/manual rate change during
  306. // wait time cannot pass on a stale pointer.
  307. func (s *OpenAIGatewayService) ProfitControlVetoLatest(ctx context.Context, selected *Account) (*Account, bool, string) {
  308. if s == nil {
  309. return selected, false, ""
  310. }
  311. return profitControlVetoLatest(ctx, selected, s.schedulerSnapshot)
  312. }
  313. // bindOpenAIStickySessionDuringSelection preserves the official eager binding
  314. // behavior for requests without a profit gate. Profit-controlled requests bind
  315. // only after the terminal post-slot check, so an account rejected after a rate
  316. // refresh cannot become the new sticky target.
  317. func (s *OpenAIGatewayService) bindOpenAIStickySessionDuringSelection(ctx context.Context, groupID *int64, sessionHash string, accountID int64) error {
  318. if gatewayProfitControlGateActive(ctx) || preserveOpenAIGuardianParentBinding(ctx, sessionHash) {
  319. return nil
  320. }
  321. return s.BindStickySession(ctx, groupID, sessionHash, accountID)
  322. }
  323. // BindStickySessionAfterProfitAdmission records the terminally admitted
  324. // account. Without a profit gate it preserves the pre-existing eager binding
  325. // behavior at the handler bind points. With a gate it never overwrites a
  326. // different binding that already exists, so a temporarily ineligible account
  327. // remains sticky and becomes eligible again automatically after its rate
  328. // recovers.
  329. func (s *OpenAIGatewayService) BindStickySessionAfterProfitAdmission(ctx context.Context, groupID *int64, sessionHash string, accountID int64) error {
  330. if sessionHash == "" || accountID <= 0 {
  331. return nil
  332. }
  333. if preserveOpenAIGuardianParentBinding(ctx, sessionHash) {
  334. return nil
  335. }
  336. if !gatewayProfitControlGateActive(ctx) {
  337. return s.BindStickySession(ctx, groupID, sessionHash, accountID)
  338. }
  339. existingAccountID, err := s.getStickySessionAccountID(ctx, groupID, sessionHash)
  340. if err != nil && !errors.Is(err, ErrStickySessionNotFound) {
  341. slog.Warn("profit_control_sticky_binding_read_failed", "group_id", derefGroupID(groupID), "account_id", accountID, "error", err)
  342. return nil
  343. }
  344. if existingAccountID > 0 && existingAccountID != accountID {
  345. return nil
  346. }
  347. return s.BindStickySession(ctx, groupID, sessionHash, accountID)
  348. }
  349. // ---- 可观测性:按分组累计计数 + 采样日志(无逐请求输出) ----
  350. //
  351. // 计数按"每次准入评估"累计,而非每请求:粘性层校验与候选池过滤可能对同一账号
  352. // 各评估一次,failover 重入也会再次计数。计数用于确认门在真实流量上生效及否决
  353. // 构成,不能当作精确的请求数或账号数。
  354. type openAIProfitControlGroupStats struct {
  355. installs atomic.Int64
  356. vetoThreshold atomic.Int64
  357. vetoInvalidRate atomic.Int64
  358. refreshFailures atomic.Int64
  359. lastLogUnixMilli atomic.Int64
  360. }
  361. type openAIProfitControlObserver struct {
  362. groups sync.Map // "platform:groupID" -> *openAIProfitControlGroupStats
  363. }
  364. var openAIProfitControlObserverInstance = &openAIProfitControlObserver{}
  365. func profitControlObserverKey(groupID int64, platform string) string {
  366. return platform + ":" + fmt.Sprintf("%d", groupID)
  367. }
  368. func (o *openAIProfitControlObserver) stats(groupID int64, platform string) *openAIProfitControlGroupStats {
  369. key := profitControlObserverKey(groupID, platform)
  370. if v, ok := o.groups.Load(key); ok {
  371. if s, ok := v.(*openAIProfitControlGroupStats); ok {
  372. return s
  373. }
  374. }
  375. v, _ := o.groups.LoadOrStore(key, &openAIProfitControlGroupStats{})
  376. if s, ok := v.(*openAIProfitControlGroupStats); ok {
  377. return s
  378. }
  379. // 不可达:map 中只存 *openAIProfitControlGroupStats;兜底返回独立实例避免 panic。
  380. return &openAIProfitControlGroupStats{}
  381. }
  382. func (o *openAIProfitControlObserver) recordInstall(groupID int64, platform string, threshold float64) {
  383. s := o.stats(groupID, platform)
  384. s.installs.Add(1)
  385. o.maybeLog(groupID, platform, threshold, s)
  386. }
  387. func (o *openAIProfitControlObserver) recordVeto(groupID int64, platform string, threshold float64, reason string) {
  388. s := o.stats(groupID, platform)
  389. switch reason {
  390. case openAIProfitFilterReasonThreshold:
  391. s.vetoThreshold.Add(1)
  392. case openAIProfitFilterReasonInvalidAccountRate:
  393. s.vetoInvalidRate.Add(1)
  394. }
  395. o.maybeLog(groupID, platform, threshold, s)
  396. }
  397. func (o *openAIProfitControlObserver) recordRefreshFailure(groupID int64, platform string, threshold float64) {
  398. s := o.stats(groupID, platform)
  399. s.refreshFailures.Add(1)
  400. o.maybeLog(groupID, platform, threshold, s)
  401. }
  402. // maybeLog 以 CAS 保证同分组 ≥ profitControlActivityLogInterval 才输出一条
  403. // 累计计数 Info;计数为进程内累计值,用于确认门在真实流量上生效及否决构成。
  404. func (o *openAIProfitControlObserver) maybeLog(groupID int64, platform string, threshold float64, s *openAIProfitControlGroupStats) {
  405. now := time.Now().UnixMilli()
  406. last := s.lastLogUnixMilli.Load()
  407. if last != 0 && now-last < profitControlActivityLogInterval.Milliseconds() {
  408. return
  409. }
  410. if !s.lastLogUnixMilli.CompareAndSwap(last, now) {
  411. return
  412. }
  413. slog.Info("profit_control_activity",
  414. "group_id", groupID,
  415. "platform", platform,
  416. "threshold", threshold,
  417. "installs_total", s.installs.Load(),
  418. "veto_threshold_total", s.vetoThreshold.Load(),
  419. "veto_invalid_account_rate_total", s.vetoInvalidRate.Load(),
  420. "refresh_failure_total", s.refreshFailures.Load(),
  421. )
  422. }