GCP Metrics CLI

gcp-metrics-cli.py

import argparse
import datetime
import logging
from google.api import metric_pb2
from google.cloud import monitoring_v3
from google.protobuf import timestamp_pb2
import google.auth

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")


def get_current_project():
    _, project_id = google.auth.default()
    if not project_id:
        raise ValueError("Failed to get current project from Google Cloud credentials")
    return project_id


def write_custom_metric(project_id, metric_type, value, end_time, metric_kind, interval=60):
    client = monitoring_v3.MetricServiceClient()
    project_name = f"projects/{project_id}"

    series = monitoring_v3.TimeSeries()
    series.metric.type = metric_type
    series.resource.type = "global"
    series.resource.labels["project_id"] = project_id

    series.metric_kind = (
        metric_pb2.MetricDescriptor.MetricKind.CUMULATIVE
        if metric_kind.upper() == "CUMULATIVE"
        else metric_pb2.MetricDescriptor.MetricKind.GAUGE
    )

    point = monitoring_v3.Point(value=monitoring_v3.TypedValue(double_value=value))
    end_time_utc = end_time.astimezone(datetime.timezone.utc)
    end_seconds = int(end_time_utc.timestamp())

    if series.metric_kind == metric_pb2.MetricDescriptor.MetricKind.CUMULATIVE:
        start_time_utc = end_time_utc - datetime.timedelta(seconds=interval)
        start_seconds = int(start_time_utc.timestamp())
        time_interval = monitoring_v3.TimeInterval(
            start_time=timestamp_pb2.Timestamp(seconds=start_seconds),
            end_time=timestamp_pb2.Timestamp(seconds=end_seconds),
        )
        logging.info(f"CUMULATIVE metric interval: {start_time_utc.isoformat()} to {end_time_utc.isoformat()}")
    else:
        time_interval = monitoring_v3.TimeInterval(end_time=timestamp_pb2.Timestamp(seconds=end_seconds))
        logging.info(f"GAUGE metric end time: {end_time_utc.isoformat()}")

    point.interval = time_interval
    series.points = [point]

    client.create_time_series(name=project_name, time_series=[series])
    logging.info(f"Successfully wrote data point to {metric_type} with value {value}")


def delete_custom_metric(project_id, metric_type):
    client = monitoring_v3.MetricServiceClient()
    descriptor_name = f"projects/{project_id}/metricDescriptors/{metric_type}"
    client.delete_metric_descriptor(name=descriptor_name)
    logging.info(f"Deleted metric descriptor {descriptor_name}")


def main():
    parser = argparse.ArgumentParser(description="Manage custom metrics in Google Cloud Monitoring")
    subparsers = parser.add_subparsers(dest="action", required=True)

    write_parser = subparsers.add_parser("write", help="Write a custom metric")
    write_parser.add_argument("--project", type=str, help="Google Cloud project ID")
    write_parser.add_argument("--metric", type=str, required=True, help="Custom metric name")
    write_parser.add_argument("--value", type=float, required=True, help="Metric value")
    write_parser.add_argument(
        "--type",
        type=str,
        default="GAUGE",
        choices=["GAUGE", "CUMULATIVE"],
        help="Metric kind",
    )
    write_parser.add_argument(
        "--interval",
        type=int,
        default=60,
        help="Interval in seconds (for CUMULATIVE metrics)",
    )

    delete_parser = subparsers.add_parser("delete", help="Delete a custom metric")
    delete_parser.add_argument("--project", type=str, help="Google Cloud project ID")
    delete_parser.add_argument("--metric", type=str, required=True, help="Custom metric name")

    args = parser.parse_args()
    project_id = args.project or get_current_project()

    if args.action == "write":
        write_custom_metric(
            project_id,
            args.metric,
            args.value,
            datetime.datetime.now(datetime.timezone.utc),
            args.type,
            args.interval,
        )
    else:
        delete_custom_metric(project_id, args.metric)


if __name__ == "__main__":
    main()