Python获取Google Cloud GCE实例监控指标遇属性错误求助

我是Google Cloud新手，希望通过Python获取GCE实例的最新网络收发字节数、CPU使用率（类似截图中的数据）。查阅官方文档后仍无法实现，于是尝试编写了如下代码，但运行时出现错误：AttributeError: 'MetricServiceClient' object has no attribute 'project_path'。以下是我尝试的代码及报错信息，恳请协助解决：

代码：

from google.cloud import monitoring_v3
from google.oauth2 import service_account
import datetime

# 配置凭据
credentials = service_account.Credentials.from_service_account_file(
    's.json'
)

# 配置查询
client = monitoring_v3.MetricServiceClient(credentials=credentials)
project_id = '00000000000'
start_time = datetime.datetime.utcnow() - datetime.timedelta(minutes=60)
end_time = datetime.datetime.utcnow()

query = (
    f'fetch '
    f'compute.googleapis.com/instance/network/sent_bytes_count, '
    f'compute.googleapis.com/instance/network/received_bytes_count '
    f'where '
    f'resource.type = "gce_instance" and '
    f'resource.label.instance_name = "your-instance-name" '
    f'and metric.type = "compute.googleapis.com/instance/network/sent_bytes_count" '
    f'or metric.type = "compute.googleapis.com/instance/network/received_bytes_count" '
    f'and timestamp >= "{start_time.isoformat()}Z" and timestamp <= "{end_time.isoformat()}Z" '
    f'order by value desc '
    f'limit 5 '
    f'align_rate()'
)

# 执行查询
results = client.list_time_series(
    request={
        "name": client.project_path(project_id),
        "filter": query,
        "interval_start_time": start_time,
        "interval_end_time": end_time,
        "view": monitoring_v3.ListTimeSeriesRequest.TimeSeriesView.FULL,
    }
)

# 打印结果
for result in results:
    print(f"{result.metric.labels['instance_name']}:")
    for point in result.points:
        value = point.value.double_value
        if 'sent' in result.metric.type:
            print(f"    Bytes enviados: {value}")
        else:
            print(f"    Bytes recebidos: {value}")

报错信息：

35 query = (
     36     f'fetch '
     37     f'compute.googleapis.com/instance/network/sent_bytes_count, '
   (...)
     47     f'align_rate()'
     48 )
     50 # 执行查询
     51 results = client.list_time_series(
     52     request={
---> 53         "name": client.project_path(project_id),
     54         "filter": query,
     55         "interval_start_time": start_time,
     56         "interval_end_time": end_time,
     57         "view": monitoring_v3.ListTimeSeriesRequest.TimeSeriesView.FULL,
     58     }
     59 )
     61 # 打印结果
     62 for result in results:

AttributeError: 'MetricServiceClient' object has no attribute 'project_path'

错误核心原因 #

新版本的Google Cloud Monitoring Python客户端库（v3）已移除project_path方法，需直接构造项目资源名称字符串projects/{project_id}。此外你的代码还存在其他几个问题：

• list_time_series接口不支持MQL查询（即你写的fetch...语法），需使用监控过滤器语法
• 时间参数需转换为Protobuf的Timestamp类型，不能直接传入Python datetime对象
• 未包含CPU使用率的指标查询

修正后的完整代码 #

from google.cloud import monitoring_v3
from google.oauth2 import service_account
import datetime
from google.protobuf.timestamp_pb2 import Timestamp

# 配置凭据
credentials = service_account.Credentials.from_service_account_file(
    's.json'
)

# 初始化客户端
client = monitoring_v3.MetricServiceClient(credentials=credentials)
project_id = '00000000000'
instance_name = 'your-instance-name'  # 替换为你的实例名称

# 设置时间范围（最近60分钟）
start_time = datetime.datetime.utcnow() - datetime.timedelta(minutes=60)
end_time = datetime.datetime.utcnow()

# 转换时间为Protobuf Timestamp格式
start_ts = Timestamp()
start_ts.FromDatetime(start_time)
end_ts = Timestamp()
end_ts.FromDatetime(end_time)

# 构造监控过滤器：筛选GCE实例的网络收发字节数、CPU使用率指标
filter_str = (
    f'resource.type = "gce_instance" AND '
    f'resource.label.instance_name = "{instance_name}" AND '
    f'(metric.type = "compute.googleapis.com/instance/network/sent_bytes_count" OR '
    f'metric.type = "compute.googleapis.com/instance/network/received_bytes_count" OR '
    f'metric.type = "compute.googleapis.com/instance/cpu/utilization")'
)

# 执行时间序列查询
results = client.list_time_series(
    request={
        "name": f"projects/{project_id}",  # 直接构造项目资源名称
        "filter": filter_str,
        "interval": {
            "start_time": start_ts,
            "end_time": end_ts
        },
        "view": monitoring_v3.ListTimeSeriesRequest.TimeSeriesView.FULL,
        "aggregation": {
            "alignment_period": {"seconds": 60},  # 按1分钟对齐数据
            "per_series_aligner": monitoring_v3.Aggregation.Aligner.ALIGN_RATE,  # 计算速率（对应网络字节数的每秒数值）
            "cross_series_reducer": monitoring_v3.Aggregation.Reducer.REDUCE_NONE
        }
    }
)

# 解析并打印结果
for result in results:
    metric_type = result.metric.type
    instance_name = result.resource.labels['instance_name']
    print(f"\n实例: {instance_name}")
    
    # 获取最新的数据点
    if result.points:
        latest_point = result.points[-1]
        value = latest_point.value.double_value
        
        if 'sent_bytes_count' in metric_type:
            print(f"网络发送速率: {round(value, 2)} 字节/秒")
        elif 'received_bytes_count' in metric_type:
            print(f"网络接收速率: {round(value, 2)} 字节/秒")
        elif 'cpu/utilization' in metric_type:
            print(f"CPU使用率: {round(value * 100, 2)}%")

关键修改说明 #

1. 替换project_path：用f"projects/{project_id}"直接构造项目资源名称，替代已废弃的client.project_path方法。
2. 修正查询语法：使用监控过滤器语法替代MQL，确保list_time_series接口能正确解析。
3. 时间格式转换：将Python datetime转换为Protobuf Timestamp对象，符合接口参数要求。
4. 添加CPU使用率指标：新增compute.googleapis.com/instance/cpu/utilization指标查询，满足你的需求。
5. 数据聚合配置：通过aggregation参数设置按1分钟对齐，并计算速率（网络字节数需要转换为每秒速率才有意义），CPU使用率直接取最新值。

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