如何用Python获取Google Cloud Run服务的CPU与内存使用率以构建资源追踪器？

一、psutil 失效原因 #

Cloud Run 采用容器化弹性运行模式，容器内的 psutil 无法获取宿主机层面的 CPU 分配统计（Cloud Run 的 CPU 是按需动态调度的），因此会出现 CPU 使用率始终显示 0% 的问题，该方案不适用于 Cloud Run 环境。

二、优化 Google Monitoring API 实现 #

通过官方 Monitoring API 可以准确获取 Cloud Run 的资源使用指标，以下是优化后的实现方案：

核心调整点 #

• 使用正确的指标类型：
  • CPU 使用率：run.googleapis.com/container/cpu/utilization（返回百分比值）
  • 内存消耗：run.googleapis.com/container/memory/usage（返回字节数，需转换为 MB）
• 过滤特定 Cloud Run 服务：避免获取所有服务的指标数据
• 调整时间区间与聚合策略：获取最新的近实时指标

完整代码示例 #

from google.cloud import monitoring_v3
from google.cloud.monitoring_v3 import types
import datetime

# 初始化客户端
client = monitoring_v3.MetricServiceClient()
PROJECT_ID = "你的项目ID"
SERVICE_NAME = "你的Cloud Run服务名"
REGION = "你的服务部署区域（如us-central1）"
project_name = client.common_project_path(PROJECT_ID)

def get_cloud_run_resource_usage():
    resource_tracker = {}
    
    # 获取CPU使用率（百分比）
    cpu_metric_type = "run.googleapis.com/container/cpu/utilization"
    now = datetime.datetime.now(datetime.timezone.utc)  # 使用UTC时间避免时区问题
    interval = types.TimeInterval({
        "end_time": {"seconds": int(now.timestamp())},
        "start_time": {"seconds": int((now - datetime.timedelta(minutes=1)).timestamp())},
    })
    aggregation = types.Aggregation({
        "alignment_period": {"seconds": 30},
        "per_series_aligner": monitoring_v3.Aggregation.Aligner.ALIGN_LAST,  # 取最新值
    })
    # 过滤特定服务和区域
    filter_str = f'metric.type="{cpu_metric_type}" AND resource.labels.service_name="{SERVICE_NAME}" AND resource.labels.location="{REGION}"'
    
    request = types.ListTimeSeriesRequest({
        "name": project_name,
        "filter": filter_str,
        "interval": interval,
        "view": monitoring_v3.ListTimeSeriesRequest.TimeSeriesView.FULL,
        "aggregation": aggregation,
    })
    
    cpu_response = client.list_time_series(request=request)
    for series in cpu_response:
        # 提取最新的CPU使用率值（百分比）
        cpu_value = series.points[-1].value.double_value
        resource_tracker["CPU Consumed (%)"] = round(cpu_value * 100, 2)  # 转换为百分比并保留两位小数
    
    # 获取内存消耗（转换为MB）
    memory_metric_type = "run.googleapis.com/container/memory/usage"
    filter_str = f'metric.type="{memory_metric_type}" AND resource.labels.service_name="{SERVICE_NAME}" AND resource.labels.location="{REGION}"'
    
    request = types.ListTimeSeriesRequest({
        "name": project_name,
        "filter": filter_str,
        "interval": interval,
        "view": monitoring_v3.ListTimeSeriesRequest.TimeSeriesView.FULL,
        "aggregation": aggregation,
    })
    
    memory_response = client.list_time_series(request=request)
    for series in memory_response:
        # 字节转MB，保留两位小数
        memory_bytes = series.points[-1].value.int64_value
        memory_mb = round(memory_bytes / (1024 * 1024), 6)
        resource_tracker["Memory Consumed (MB)"] = memory_mb
    
    return {"Resources Tracker": resource_tracker}

# 调用示例
if __name__ == "__main__":
    usage_data = get_cloud_run_resource_usage()
    print(usage_data)

关键注意事项 #

1. 权限配置：确保运行代码的服务账号拥有 roles/monitoring.viewer 角色权限，否则无法读取监控指标
2. 指标延迟：Monitoring API 的指标存在约 1-2 分钟的延迟，无法获取完全实时的数据
3. 时区处理：代码中使用 UTC 时间避免时区偏差导致的指标查询错误

内容的提问来源于stack exchange，提问作者lapiceroazul4