Skip to main content

Advanced guide

Helper functions for mount/unmount verification​

For production use, consider implementing helper functions to wait for operations to complete:

import time
import cgc.sdk.volume as volume

def wait_for_volume_mount(volume_name, compute_name, timeout=60):
    """Wait for volume to show as mounted"""
    print(f"Waiting for volume '{volume_name}' to be mounted...")
    start_time = time.time()

    while time.time() - start_time < timeout:
        response = volume.volume_list()
        if response['code'] == 200:
            for vol in response['details']['volume_list']:
                if vol['name'] == volume_name:
                    for mount in vol.get('mounted_to', []):
                        if compute_name in mount:
                            print("✓ Volume mount confirmed!")
                            return True
        time.sleep(3)

    print("⚠ Mount verification timed out")
    return False

def wait_for_volume_unmount(volume_name, timeout=30):
    """Wait for volume to show as unmounted"""
    print(f"Waiting for volume '{volume_name}' to be unmounted...")
    start_time = time.time()

    while time.time() - start_time < timeout:
        response = volume.volume_list()
        if response['code'] == 200:
            for vol in response['details']['volume_list']:
                if vol['name'] == volume_name:
                    if not vol.get('mounted_to', []):
                        print("✓ Volume unmount confirmed!")
                        return True
        time.sleep(2)

    print("⚠ Unmount verification timed out")
    return False

# Usage example
response = volume.volume_mount("my-volume", "my-compute")
if response['code'] == 200:
    if wait_for_volume_mount("my-volume", "my-compute"):
        print("Volume is ready for use!")

Volume status monitoring​

Volume can be either bound or not yet ready for usage.

def monitor_volumes():
    """Monitor status of all volumes"""
    try:
        response = volume.volume_list()
        if response['code'] != 200:
            print("Failed to list volumes")
            return

        volumes = response['details']['volume_list']
        for vol in volumes:
            name = vol['name']
            status = vol['status']
            size = vol['size'].replace('Gi', '')  # Remove 'Gi' suffix
            storage_type = vol['disks_type']
            mounted_to = vol['mounted_to']

            status_symbol = "✓" if status.lower() == "bound" else "○"
            mount_info = f", Mounted: {len(mounted_to)} resources" if mounted_to else ", Not mounted"
            print(f"{status_symbol} {name}: {status} ({size}GB, {storage_type}{mount_info})")

    except Exception as e:
        print(f"Error monitoring volumes: {e}")

Common issues​

Volume creation fails​

# Check available storage classes
classes = volume.get_available_storage_classes()
print("Available storage classes:", classes)

# Use default storage class
default = volume.get_default_storage_class()
volume.volume_create("my-volume", 10, default)

Volume not becoming ready​

# Check volume status manually
response = volume.volume_list()
for vol in response['details']['volumes']:
    if vol['name'] == 'my-volume':
        print(f"Volume status: {vol['status']}")
        print(f"Volume details: {vol}")

Mount failures​

# Ensure volume is ready before mounting
if volume.volume_ready("my-volume"):
    volume.volume_mount("my-volume", "my-compute")
else:
    print("Volume not ready for mounting")

Force operations when needed​

# Force delete stuck volume
volume.volume_delete("stuck-volume", force=True)

# Force unmount stuck volume
volume.volume_umount("stuck-volume", force=True)