Rails API Usage Tracking 与 Metrics 数据结构设计

明确结论

不要把 usage 数据存进 Flow 的多层 Hash，也不建议在 Flow 记录上持续更新计数。

更合适的设计是：

1. 用独立的时间分桶表保存聚合计数，例如每个 Flow 每小时一条记录。
2. 通过数据库的原子操作递增计数，避免并发请求造成数据丢失。
3. 如果计费需要审计、去重，或者以后需要分析 IP、地理位置等维度，再增加一张不可变的原始事件表。
4. 流量较高时，可以用 Redis 暂存计数并批量落库，但数据库中的聚合结果或原始事件仍应是可恢复、可审计的数据来源。

“每小时创建一个 object instance 并递增 integer field”这个方向是合理的。它本质上是一张时间序列 rollup 表，比嵌套 Hash 更适合范围查询、求和、索引和数据保留管理。

为什么不应继续使用多层 Hash

把年份、月份、日期和小时嵌套在一个字段中，会产生不少问题：

• 数据库很难利用普通索引执行时间范围查询。
• 按用户、多个 Flow 或全站聚合时，需要加载并解析大量 Hash。
• 每次更新都可能重写整个字段。
• 多个请求同时更新同一个 Flow 时，容易发生 lost update。
• 很难增加 status_code、API endpoint、region 等分析维度。
• 不便删除过期的小时数据，也不便生成日级、月级汇总。
• 单条 Flow 记录会持续膨胀，最终成为写入热点。

JSON/JSONB 适合存储结构不固定的附加属性，不适合承担核心计数和时间范围统计。

推荐的数据结构

可以建立一张 flow_usage_buckets 表，每个时间桶保存调用次数：

class CreateFlowUsageBuckets < ActiveRecord::Migration[7.0]
  def change
    create_table :flow_usage_buckets do |t|
      t.references :flow, null: false, foreign_key: true
      t.references :user, null: false, foreign_key: true
      t.datetime :bucket_start, null: false
      t.string :granularity, null: false, default: "hour"
      t.bigint :calls_count, null: false, default: 0

      t.timestamps
    end

    add_index :flow_usage_buckets,
              [:flow_id, :granularity, :bucket_start],
              unique: true,
              name: "idx_flow_usage_unique_bucket"

    add_index :flow_usage_buckets,
              [:user_id, :bucket_start]

    add_index :flow_usage_buckets, :bucket_start
  end
end

对应模型：

class FlowUsageBucket < ApplicationRecord
  belongs_to :flow
  belongs_to :user

  validates :bucket_start, presence: true
  validates :granularity, inclusion: { in: %w[minute hour day month] }
  validates :calls_count,
            numericality: { only_integer: true, greater_than_or_equal_to: 0 }
end

如果目前只保留小时粒度，可以暂时省略 granularity，让唯一索引和查询更简单：

add_index :flow_usage_buckets,
          [:flow_id, :bucket_start],
          unique: true

从关系上看，user_id 可以通过 flow.user_id 获取，因此它是一个冗余字段。不过，把它放进 bucket 表能明显简化按用户和时间统计的查询，代价是必须保证它与 flow.user_id 一致。如果 Flow 可能转移给其他用户，还要提前确定历史 usage 应归原用户还是当前用户。

使用原子 UPSERT 递增

不要使用“先查询当前值，再加一并保存”的写法：

bucket = FlowUsageBucket.find_or_create_by!(...)
bucket.update!(calls_count: bucket.calls_count + 1)

两个并发请求可能读到相同的旧值，最后只记录一次。应该让数据库通过原子 UPSERT 完成递增。

以 PostgreSQL 为例，可以封装一个 recorder：

class ApiUsageRecorder
  def self.record!(flow:, occurred_at: Time.current)
    bucket_start = occurred_at.utc.beginning_of_hour
    now = Time.current

    sql = ApplicationRecord.send(
      :sanitize_sql_array,
      [
        <<~SQL,
          INSERT INTO flow_usage_buckets
            (flow_id, user_id, bucket_start, granularity, calls_count, created_at, updated_at)
          VALUES
            (?, ?, ?, 'hour', 1, ?, ?)
          ON CONFLICT (flow_id, granularity, bucket_start)
          DO UPDATE SET
            calls_count = flow_usage_buckets.calls_count + 1,
            updated_at = EXCLUDED.updated_at
        SQL
        flow.id,
        flow.user_id,
        bucket_start,
        now,
        now
      ]
    )

    ApplicationRecord.connection.execute(sql)
  end
end

调用方式：

ApiUsageRecorder.record!(flow: flow)

这段代码依赖 PostgreSQL 的 ON CONFLICT。如果使用 MySQL，应改用对应的 INSERT ... ON DUPLICATE KEY UPDATE，具体 SQL 需要根据实际数据库调整。

Rails 的 upsert_all 也可以创建或覆盖记录，但“发生冲突时，把数据库中的现有值原子加一”并不是所有 Rails 与数据库组合都能通过同一套高层 API 清楚表达。对于计数器，显式 SQL 通常更直接，也更容易确认并发语义。

三类查询的实现

查询某个 Flow 的小时 usage

from = 7.days.ago.beginning_of_hour
to   = Time.current.end_of_hour

usage = FlowUsageBucket
  .where(
    flow_id: flow.id,
    granularity: "hour",
    bucket_start: from...to
  )
  .order(:bucket_start)
  .pluck(:bucket_start, :calls_count)

数据库只会返回实际有请求的时间桶。绘制连续曲线时，可以在展示层补齐缺失的小时，并把计数设为 0。

如果使用 PostgreSQL，也可以通过 generate_series 在 SQL 中补齐时间点。是否值得这样处理，要看图表的数据量和查询复杂度。

查询某用户某月全部 Flows 的总调用量

month = Time.zone.parse("2026-08-01")
range = month.beginning_of_month...month.next_month.beginning_of_month

total = FlowUsageBucket
  .where(
    user_id: user.id,
    granularity: "hour",
    bucket_start: range
  )
  .sum(:calls_count)

如果 bucket 表不保存 user_id，可以通过 flows 关联查询：

total = FlowUsageBucket
  .joins(:flow)
  .where(
    flows: { user_id: user.id },
    granularity: "hour",
    bucket_start: range
  )
  .sum(:calls_count)

前一种写法查询更简单，后一种写法避免了数据冗余。选择时需要考虑查询频率、数据量，以及 Flow 的归属是否可能变化。

查询给定时间段的全站 usage

total = FlowUsageBucket
  .where(
    granularity: "hour",
    bucket_start: from...to
  )
  .sum(:calls_count)

按小时生成全站趋势：

series = FlowUsageBucket
  .where(
    granularity: "hour",
    bucket_start: from...to
  )
  .group(:bucket_start)
  .order(:bucket_start)
  .sum(:calls_count)

计费数据不能只考虑“计数”

如果 usage 会直接决定账单，一个可变计数器通常不够。还需要明确以下问题：

• 客户端重试是否会重复计费。
• 请求执行到一半失败时是否计费。
• 统计写入失败后能否重放。
• 如何把某笔账单追溯到具体请求。
• 是否应排除内部重试、健康检查和测试流量。
• 迟到事件应计入请求发生月份还是处理月份。
• 删除 Flow 后是否仍需保留账单记录。

更稳妥的做法是记录不可变的 usage event：

class CreateApiUsageEvents < ActiveRecord::Migration[7.0]
  def change
    create_table :api_usage_events do |t|
      t.references :flow, null: true, foreign_key: { on_delete: :nullify }
      t.references :user, null: false, foreign_key: true
      t.string :request_id, null: false
      t.datetime :occurred_at, null: false
      t.string :endpoint
      t.integer :status_code
      t.inet :ip_address
      t.jsonb :metadata, null: false, default: {}

      t.timestamps
    end

    add_index :api_usage_events, :request_id, unique: true
    add_index :api_usage_events, [:flow_id, :occurred_at]
    add_index :api_usage_events, [:user_id, :occurred_at]
    add_index :api_usage_events, :occurred_at
  end
end

其中：

• request_id 或业务定义的 idempotency_key 用于防止重复计数。
• occurred_at 表示实际发生时间，created_at 表示事件写入时间。
• 常用过滤字段应使用独立列。
• 只有结构不固定、使用频率低的属性才放进 metadata。
• IP 地址可能涉及隐私和合规问题，应限制用途、访问权限和保留周期。

之后可以通过后台任务把事件聚合到小时 bucket。展示报表时查询 bucket，审计和重新计算时使用原始事件。

这种模式的数据量更大，但能提供计费系统所需的可追溯性。可以通过分区、压缩、归档和保留策略控制成本，不应为了减少行数而牺牲账单的正确性。

高流量下的写入架构

对于中等流量，直接向数据库执行原子 UPSERT 通常是最简单、可靠的起点。流量增长后，可以逐步演进为：

API request
    ↓
写入 usage event 或消息队列
    ↓
后台消费者聚合
    ↓
小时 bucket / 日 bucket
    ↓
图表、报表与账单

也可以在 Redis 中使用 INCRBY：

usage:flow:123:2026-08-29T14:00:00Z

然后定期批量写入数据库。不过，使用 Redis 前必须明确：

• 进程崩溃或 Redis 故障时，是否允许丢失计数。
• flush 成功后，如何避免重复落库。
• key 的过期时间。
• 后台任务重试是否具有幂等性。
• Redis 与数据库不一致时如何修复。

因此，Redis 更适合作为写入缓冲或实时计数层。如果没有持久化和恢复方案，不应把它作为计费的唯一依据。

分钟级或更细粒度如何选择

时间桶越细，记录越多，但图表也越灵活。可以按实际展示和计费需求选择：

• 只展示长期趋势时，小时 bucket 通常足够。
• 展示短时间实时流量时，可以使用分钟 bucket。
• 需要请求级审计时，保存 event，同时生成分钟或小时 rollup。
• 长期保存时，可以只保留最近一段时间的分钟数据，之后保留小时或日汇总。

以一条持续活跃的 Flow 为例：

• 小时粒度每年最多约 8,760 条 bucket。
• 分钟粒度每年最多约 525,600 条 bucket。
• 原始事件数量与实际请求数相同。

实际记录数还取决于是否为空桶创建记录。通常没有必要为没有请求的时间段预建记录。

日级和月级汇总

小时表增长到一定规模后，可以再建立日汇总表：

flow_usage_hourly
flow_usage_daily

最近七天的图表读取小时表，月度账单和年度报表读取日表。日表由可重试的后台任务根据小时表生成，并通过唯一索引保证幂等性。

不要把小时表和基于这些小时数据生成的日表放在一起求和，否则会重复统计。每个查询都必须明确使用哪一种粒度。

账单结算后，建议生成独立且不可变的账单明细或 usage statement，不要在每次打开历史账单时重新查询当前 rollup。迟到事件、数据修正和业务规则变化都可能使重新计算的结果发生变化。

时区与月份边界

存储和分桶建议统一使用 UTC：

bucket_start = occurred_at.utc.beginning_of_hour

不过，“用户的某个月”可能是按其账单时区定义的。查询时，应先根据账单时区计算月份边界，再转换成 UTC：

zone = ActiveSupport::TimeZone[user.billing_time_zone]

local_start = zone.local(2026, 8, 1)
local_end   = local_start.next_month

range = local_start.utc...local_end.utc

不要假设每个月或每天都有固定的小时数。使用夏令时的地区，一天可能有 23 小时或 25 小时。

索引、分区与数据保留

索引应按照实际查询模式设计，不必为每一列都添加索引。这个场景通常需要以下核心索引：

UNIQUE (flow_id, granularity, bucket_start)
(user_id, bucket_start)
(bucket_start)

原始事件表达到较大规模后，可以按 occurred_at 做月度分区，以便进行范围扫描和删除过期分区。是否需要分区，应以实际数据量和查询计划为依据。过早引入分区会增加迁移和运维的复杂度。

还需要制定清晰的数据保留策略，例如：

• 原始请求事件保留若干月。
• 分钟数据只保留较短时间。
• 小时或日汇总长期保留。
• 已结算的账单明细按照财务和合规要求保留。

具体期限取决于业务、合同和适用法规，不能只由技术实现决定。

是否需要 gem

这类需求的重点不在于缺少某个 Rails gem，而在于数据模型、并发更新、幂等性和保留策略是否清楚。

图表聚合可以考虑使用 groupdate 等时间分组类 gem，但它只能简化 GROUP BY 查询，不能代替 usage 事件、原子计数或计费审计设计。数据库监控方案也不等同于面向客户的 billing ledger。

如果规模继续扩大，可以评估专门的时间序列或分析系统，例如 PostgreSQL 分区表、TimescaleDB、ClickHouse，或者托管的计量计费服务。不过，对大多数早期 Rails 应用来说，PostgreSQL 加独立 bucket 表已经足够，后续迁移路径也很清楚。

实施建议

可以按风险从低到高逐步实施：

1. 停止写入 Flow#daily_calls，建立小时 bucket 表。
2. 使用唯一索引和数据库 UPSERT，实现并发安全的计数。
3. 将统计查询改为基于 bucket_start 的范围查询。
4. 明确 UTC、账单时区，以及失败请求和重试请求的计费规则。
5. 如果数据用于真实计费，增加带有唯一 request_id 的不可变 usage event。
6. 通过后台任务生成小时、日级 rollup，并核对事件总数与汇总总数。
7. 等数据增长后，再引入 Redis 缓冲、表分区或专门的分析存储。

如果 usage 仅用于图表展示，小时 bucket 足以解决问题。如果 usage 会产生账单，建议采用“不可变事件 + 可重建汇总 + 已结算账单快照”的结构。这样既便于查询，也能在出现重复请求、迟到数据或统计错误时核对和修复。

备注：内容仅供参考。