PAYATHON 2026

如何按 API key 获取 Google Cloud Endpoints API 用量数据

技术老齐

结论

可以使用 Cloud Monitoring 的 Service Runtime 指标 serviceruntime.googleapis.com/api/request_count 统计月度调用次数，汇总维度如下：

  • 客户：通过独立的 consumer project 标识，也可以使用返回数据中的 credential_id
  • API 服务：resource.labels.service
  • API 方法：resource.labels.method
  • 时间范围：按自然月使用半开区间 [月初, 下月月初)

为每个客户创建独立的 GCP project，并让该客户的 API key 归属于对应 project。请求经过 ESP 或 ESPv2 时，Cloud Endpoints 会将消费方项目、服务和方法写入 Service Runtime 指标。之后可在 API 提供方项目中通过 Cloud Monitoring API 查询并汇总这些数据。

如果统计结果直接用于出账，最好同时通过 Logs Router 将请求日志导出到 BigQuery。Monitoring 适合统计用量和对账，BigQuery 日志则便于保存可审计的计费明细。

数据模型

API key 可能轮换或撤销，也不适合出现在报表中，因此不要将其明文作为账单主键。可以维护以下映射：

consumer_project_id -> client_id

client-a-project -> client_a
client-b-project -> client_b

每个客户项目可以创建一个或多个 API key，所有 key 都应限制为只能调用指定的 Endpoints 服务。如果需要区分同一项目下的不同 key，可以再使用监控数据中的 credential_id。该字段的格式应以项目实际返回的时序标签为准，不能假设它一定是 API key 明文。

账单明细可以采用以下结构：

billing_month
client_id
consumer_project_id
service
method
request_count

例如：

2026-09,client_a,client-a-project,api-esp1.example.com,ApiEsp1,12500
2026-09,client_a,client-a-project,api-esp2.example.com,ApiEsp2Foo,3800
2026-09,client_b,client-b-project,api-esp2.example.com,ApiEsp2Bar,9100

配置流程

1. 为每个客户准备 consumer project

每个客户使用独立项目，并在项目中创建 API key。调用 Endpoints API 时，Service Control 会据此识别消费方项目。

配置时需要确认：

  • consumer project 已启用相应的 Endpoints 服务。
  • API key 已绑定正确的 API restrictions。
  • ESP 或 ESPv2 的 OpenAPI 配置要求提供 API key。
  • 不同客户没有共用同一个未受控的 API key。
  • 提供方项目中的 Service Runtime 指标可以看到该 consumer project。

如果所有客户都使用同一项目中的 key，仅靠 project_id 无法区分客户。这种情况下，需要确认 credential_id 能否满足区分要求，否则应改为每个客户使用独立项目。

2. 确认指标及标签

先查看指标描述，确认当前环境提供了哪些标签：

gcloud monitoring metrics describe \
  serviceruntime.googleapis.com/api/request_count

也可以查询一小段时间的数据，检查各条 time series 的 resource.labels 和 metric.labels。一般需要关注以下字段：

resource.labels.project_id
resource.labels.service
resource.labels.method
resource.labels.version
resource.labels.location
resource.labels.credential_id
metric.labels.response_code
metric.labels.response_code_class
metric.labels.protocol

标签集合可能因受监控资源类型、API 配置和数据来源而异，应以 Monitoring API 的实际返回结果为准。

各字段的用途如下：

  • project_id：识别 consumer project。
  • service：区分 ESP 暴露的 API 服务。
  • method：区分具体操作。
  • credential_id：存在该标签时，可进一步区分调用凭据。
  • 响应码标签：区分成功请求和失败请求。

method 的实际值由 Endpoints 服务配置决定，不一定与 URL path 完全一致。可以为 OpenAPI operation 设置稳定且唯一的 operationId，并在首次采集数据时核对它与监控标签的对应关系。

使用 Cloud Monitoring API 汇总

以下 Python 代码从 API 提供方项目查询一个自然月内的请求数，并按 consumer project、服务、方法和凭据汇总：

from collections import defaultdict
from datetime import datetime, timezone

from google.cloud import monitoring_v3
from google.protobuf.timestamp_pb2 import Timestamp


PRODUCER_PROJECT_ID = "api-producer-project"

# 使用 UTC 半开区间：[2026-09-01 00:00:00, 2026-10-01 00:00:00)
START = datetime(2026, 9, 1, tzinfo=timezone.utc)
END = datetime(2026, 10, 1, tzinfo=timezone.utc)


def to_timestamp(value: datetime) -> Timestamp:
    timestamp = Timestamp()
    timestamp.FromDatetime(value)
    return timestamp


client = monitoring_v3.MetricServiceClient()

interval = monitoring_v3.TimeInterval(
    start_time=to_timestamp(START),
    end_time=to_timestamp(END),
)

filter_expression = """
metric.type = "serviceruntime.googleapis.com/api/request_count"
AND resource.type = "consumed_api"
"""

request = monitoring_v3.ListTimeSeriesRequest(
    name=f"projects/{PRODUCER_PROJECT_ID}",
    filter=filter_expression,
    interval=interval,
    view=monitoring_v3.ListTimeSeriesRequest.TimeSeriesView.FULL,
)

usage = defaultdict(int)

for series in client.list_time_series(request=request):
    resource_labels = dict(series.resource.labels)
    metric_labels = dict(series.metric.labels)

    consumer_project = resource_labels.get("project_id", "")
    service = resource_labels.get("service", "")
    method = resource_labels.get("method", "")
    credential_id = resource_labels.get(
        "credential_id",
        metric_labels.get("credential_id", ""),
    )

    count = sum(point.value.int64_value for point in series.points)

    key = (
        consumer_project,
        credential_id,
        service,
        method,
    )
    usage[key] += count

for key, count in sorted(usage.items()):
    consumer_project, credential_id, service, method = key
    print(
        consumer_project,
        credential_id,
        service,
        method,
        count,
        sep=",",
    )

安装依赖：

python -m pip install google-cloud-monitoring

运行程序的身份需要具备读取提供方项目监控数据的权限，例如包含 monitoring.timeSeries.list 的只读角色。

如果只查询某个 Endpoints 服务，可以在过滤器中增加条件：

filter_expression = """
metric.type = "serviceruntime.googleapis.com/api/request_count"
AND resource.type = "consumed_api"
AND resource.labels.service = "api-esp1.example.com"
"""

生产查询应将示例服务名替换为 Endpoints 部署后实际注册的 service name。

生成客户账单

查询结果还需要与内部客户表关联：

CLIENT_BY_PROJECT = {
    "client-a-project": "client_a",
    "client-b-project": "client_b",
}

for key, count in sorted(usage.items()):
    consumer_project, credential_id, service, method = key
    client_id = CLIENT_BY_PROJECT.get(consumer_project)

    if client_id is None:
        # 未知消费方不应静默计费，应进入异常记录或人工核对流程。
        continue

    print(
        {
            "client_id": client_id,
            "consumer_project_id": consumer_project,
            "credential_id": credential_id,
            "service": service,
            "method": method,
            "request_count": count,
        }
    )

如果一个客户可能使用多个项目，客户与项目之间的映射应支持一对多，避免在计费代码中写死项目 ID。

正式计费时的注意事项

明确一次调用的计算方式

request_count 记录请求次数，具体计费规则仍需自行确定，包括：

  • 是否只统计 2xx 请求。
  • 4xx 参数错误是否收费。
  • 401、403 和配额拒绝是否收费。
  • 5xx 是否从账单中扣除。
  • 客户重试产生的请求是否分别计费。
  • 流式 RPC 按连接次数、消息数还是其他业务量计费。

如果只统计成功请求，应根据实际存在的响应码标签增加过滤条件，或在汇总阶段筛选。不同协议使用的响应码标签可能不同，需要先检查实际数据。

固定月份边界

建议统一使用 UTC，并采用半开区间：

[start_of_month, start_of_next_month)

如果合同按客户所在时区结算，需要先将当地月初转换为 UTC，再查询对应区间。对于实行夏令时的地区，不能简单地将每个月的起点设为 UTC 零点。

处理迟到数据

监控数据不一定在请求发生后立即稳定。月末任务可以先生成暂定结果，等约定的窗口结束后重新查询并封账。等待时间应根据实际观测到的数据延迟确定，不宜将未经验证的固定时长写入计费规则。

保存原始结果

每次结算至少保存以下内容：

  • 查询起止时间。
  • Monitoring filter。
  • 原始 time series 标签和值。
  • API 配置版本。
  • 客户与 consumer project 的映射快照。
  • 最终汇总结果。
  • 账单生成时间和重算版本。

这些记录用于解释某个客户的调用次数为何被计入账单。

对账要求较高时导出日志

Cloud Monitoring 保存指标时序，适合聚合统计，但不能作为逐请求账本。如果需要逐条审计、排除特定请求或长期保存原始记录，应通过 Logs Router 将 Endpoints 或 ESP 请求日志导出到 BigQuery，再按日志中的服务、方法、消费方身份、状态码和时间分组。

日志字段取决于 ESP、ESPv2、部署平台和日志格式。编写 SQL 前应先检查 BigQuery 中的实际 schema，不要假定 API key 位于某个固定的 JSON 字段中。

落地方式

可以分两层统计：

  1. 每天通过 Cloud Monitoring API 拉取 api/request_count，按 consumer project、service 和 method 保存汇总结果。
  2. 每月重新查询完整月份并生成账单，避免直接累加每日任务而造成重复或遗漏。
  3. 同时将请求日志导出到 BigQuery，用于争议处理和抽样核对。
  4. 账单使用内部 client_id，不展示完整 API key。

采用这种方式后，客户轮换 API key 不会影响客户归属，服务和方法维度也可以分别统计 api_esp1、api_esp2_foo 和 api_esp2_bar 的调用量。

备注：内容仅供参考。