如何用Python的Cloud Monitoring API创建基于其他指标的自动更新时间序列指标？

实现步骤 #

1. 前置准备 #

• 启用Google Cloud Monitoring API：在GCP控制台的API库中搜索并启用Cloud Monitoring API。
• 安装Python客户端库：

  pip install google-cloud-monitoring
  

• 认证：本地开发可执行gcloud auth application-default login完成认证；生产环境建议使用服务账号密钥，设置环境变量GOOGLE_APPLICATION_CREDENTIALS指向密钥文件路径。

2. 核心实现逻辑 #

要实现每小时计算MetricA与MetricB的和并写入自定义指标MetricC，需完成查询现有指标数据、计算聚合值、写入自定义指标三个关键操作，同时需要定时触发该逻辑才能实现自动更新。

示例代码

以下是完整的Python代码示例，包含查询、计算和写入逻辑：

from google.cloud import monitoring_v3
from datetime import datetime, timedelta
import time

def calculate_and_write_custom_metric(project_id, metric_a_name, metric_b_name, custom_metric_name):
    client = monitoring_v3.MetricServiceClient()
    project_name = f"projects/{project_id}"

    # 定义查询时间区间（最近1小时）与聚合规则
    now = datetime.utcnow()
    interval = monitoring_v3.TimeInterval(
        end_time={"seconds": int(time.mktime(now.timetuple()))},
        start_time={"seconds": int(time.mktime((now - timedelta(hours=1)).timetuple()))},
    )
    aggregation = monitoring_v3.Aggregation(
        alignment_period={"seconds": 3600},  # 按1小时对齐时间序列
        per_series_aligner=monitoring_v3.Aggregation.Aligner.ALIGN_SUM,  # 单序列内按小时求和
        cross_series_reducer=monitoring_v3.Aggregation.Reducer.REDUCE_SUM,  # 多序列结果求和
    )

    # 查询MetricA的小时聚合值
    metric_a_filter = f'metric.type = "{metric_a_name}"'
    a_results = client.list_time_series(
        request={
            "name": project_name,
            "filter": metric_a_filter,
            "interval": interval,
            "view": monitoring_v3.ListTimeSeriesRequest.TimeSeriesView.FULL,
            "aggregation": aggregation,
        }
    )
    metric_a_value = 0.0
    for series in a_results:
        for point in series.points:
            metric_a_value = point.value.double_value

    # 查询MetricB的小时聚合值
    metric_b_filter = f'metric.type = "{metric_b_name}"'
    b_results = client.list_time_series(
        request={
            "name": project_name,
            "filter": metric_b_filter,
            "interval": interval,
            "view": monitoring_v3.ListTimeSeriesRequest.TimeSeriesView.FULL,
            "aggregation": aggregation,
        }
    )
    metric_b_value = 0.0
    for series in b_results:
        for point in series.points:
            metric_b_value = point.value.double_value

    # 计算MetricC的值
    metric_c_value = metric_a_value + metric_b_value

    # 写入自定义指标MetricC
    series = monitoring_v3.TimeSeries()
    series.metric.type = custom_metric_name
    series.resource.type = "global"  # 根据实际关联资源调整，如"gce_instance"
    series.resource.labels["project_id"] = project_id

    point = series.points.add()
    point.value.double_value = metric_c_value
    point.interval.end_time.seconds = int(time.mktime(now.timetuple()))

    client.create_time_series(name=project_name, time_series=[series])
    print(f"Successfully wrote MetricC value: {metric_c_value} at {now}")

# 调用示例
if __name__ == "__main__":
    PROJECT_ID = "your-gcp-project-id"
    METRIC_A = "compute.googleapis.com/instance/cpu/utilization"  # 替换为你的MetricA类型
    METRIC_B = "compute.googleapis.com/instance/disk/read_bytes_count"  # 替换为你的MetricB类型
    CUSTOM_METRIC = "custom.googleapis.com/metrics/metric_c"  # 自定义指标需符合格式规范
    calculate_and_write_custom_metric(PROJECT_ID, METRIC_A, METRIC_B, CUSTOM_METRIC)

3. 实现定时自动更新 #

Cloud Monitoring API不会自动执行计算与写入操作，需手动设置定时触发机制：

• 本地/自建服务器：使用Linux的cron或Windows的任务计划程序，配置每小时执行一次上述脚本。
• GCP云端部署：将代码打包为Cloud Function，通过Cloud Scheduler创建每小时触发的HTTP请求调用该函数，这是生产环境更稳定的方案。

关于自动更新的说明 #

自定义指标MetricC不会自动在Cloud Monitoring中更新，必须通过定时执行计算与写入逻辑，才能生成每小时的时间序列数据。Cloud Monitoring仅负责存储和展示你写入的指标数据，若需动态计算衍生指标可使用MQL查询，但MQL结果是实时计算的，不会生成持久化的自定义时间序列。

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