Calculate total bucket usage by project

Here is another script that can be used to calculate the total bucket usage per project from the cloud monitoring metrics.

import time
from google.cloud import monitoring_v3
from google.protobuf.duration_pb2 import Duration

def format_bytes(bytes_value):
    """Converts bytes to a human-readable format."""
    if bytes_value is None:
        return "N/A"
    
    bytes_value = float(bytes_value)
    units = ['Bytes', 'KiB', 'MiB', 'GiB', 'TiB', 'PiB']
    i = 0
    size = bytes_value
    while size >= 1024 and i < len(units) - 1: 
        size /= 1024
        i += 1
    
    return f"{size:.2f} {units[i]}"

def get_total_gcs_storage_for_project(project_id):
    """
    Fetches the total_bytes metric, summed across ALL GCS buckets in a project.
    
    Args:
        project_id (str): Your Google Cloud project ID.
    
    Returns:
        float: The total size of all buckets in bytes, or None if no data is found.
    """
    client = monitoring_v3.MetricServiceClient()
    
    # FIX 1: Use manual path construction for older library versions
    project_name = f"projects/{project_id}"
    
    # 1. Define the Time Interval (last 2 days)
    now = int(time.time()) # Get current time as integer seconds
    
    # FIX 2: Instantiate TimeInterval and TimeStamp components directly
    interval = monitoring_v3.TimeInterval(
        end_time={'seconds': now},  # TimeStamp using dict for seconds
        start_time={'seconds': now - 2 * 86400} 
    )

    # 2. Define the Metric Filter
    metric_filter = (
        'metric.type = "storage.googleapis.com/storage/total_bytes" '
        'AND resource.type = "gcs_bucket"'
    )
    
    # 3. Define Aggregation for Summing All Buckets
    aggregation = monitoring_v3.Aggregation(
        # FIX 3: Instantiate Duration object manually
        alignment_period=Duration(seconds=86400), # 1 day
        
        per_series_aligner=monitoring_v3.Aggregation.Aligner.ALIGN_MAX,
        cross_series_reducer=monitoring_v3.Aggregation.Reducer.REDUCE_SUM,
        group_by_fields=[],
    )

    # 4. List Time Series
    try:
        results = client.list_time_series(
            request={
                "name": project_name,
                "filter": metric_filter,
                "interval": interval,
                "aggregation": aggregation,
                "view": monitoring_v3.ListTimeSeriesRequest.TimeSeriesView.FULL,
            }
        )
    except Exception as e:
        print(f"Error querying Cloud Monitoring: {e}")
        return None

    # Process the result (should contain only one time series with one point)
    for result in results:
        if result.points:
            total_bytes = result.points[0].value.double_value
            return total_bytes
            
    return None

# --- Configuration ---
GCP_PROJECT_ID = "PROJECT_ID"  # <-- CHANGE THIS
# ---------------------

if __name__ == "__main__":
    total_size_bytes = get_total_gcs_storage_for_project(GCP_PROJECT_ID)
    
    print(f"Project ID: {GCP_PROJECT_ID}")
    print("---")
    
    if total_size_bytes is not None and total_size_bytes >= 0:
        print(f"Total GCS Storage Used (Raw Bytes): {int(total_size_bytes)}")
        print(f"Total GCS Storage Used (Human-readable): {format_bytes(total_size_bytes)}")
    else:
        print("Could not retrieve total storage metric. Check the project ID, permissions, or if Cloud Monitoring is enabled.")

Tnx