LAB 62 ‐ AI ‐ Extract capacity related decision making parameters for storage family in CSV & also to object storage - DevWithNehaCloud/Oracle-Cloud-Infrastructure-wiki GitHub Wiki
import oci
import csv
import subprocess
import json
# ============================================================
# OCI CONFIGURATION
# ============================================================
config = oci.config.from_file()
tenancy_id = config["tenancy"]
# ============================================================
# OCI CLIENTS
# ============================================================
identity = oci.identity.IdentityClient(config)
blockstorage = oci.core.BlockstorageClient(config)
object_storage = oci.object_storage.ObjectStorageClient(config)
# ============================================================
# OBJECT STORAGE CONFIGURATION
# ============================================================
# Bucket where CSV will be uploaded
bucket_name = "compute-capacity-inventory"
# Local CSV file
csv_file = "storage_capacity_inventory.csv"
# Object name inside Object Storage
object_name = "storage_capacity_inventory.csv"
# ============================================================
# GET OBJECT STORAGE NAMESPACE
# ============================================================
namespace = object_storage.get_namespace().data
print("\n==============================================")
print(" STORAGE CAPACITY INVENTORY")
print("==============================================")
print(f"Namespace : {namespace}")
print(f"Bucket : {bucket_name}")
# ============================================================
# GET ALL COMPARTMENTS
# ============================================================
compartments = [
{
"id": tenancy_id,
"name": "Tenancy Root"
}
]
try:
compartments_response = oci.pagination.list_call_get_all_results(
identity.list_compartments,
tenancy_id,
compartment_id_in_subtree=True,
access_level="ANY"
)
for compartment in compartments_response.data:
if compartment.lifecycle_state == "ACTIVE":
compartments.append({
"id": compartment.id,
"name": compartment.name
})
except Exception as e:
print(f"\nCompartment scan failed: {e}")
print(
f"\nTotal compartments: "
f"{len(compartments)}"
)
# ============================================================
# GET AVAILABILITY DOMAINS
# ============================================================
try:
availability_domains = identity.list_availability_domains(
tenancy_id
).data
except Exception as e:
print(
f"\nAvailability Domain scan failed: "
f"{e}"
)
availability_domains = []
print(
f"Availability Domains: "
f"{len(availability_domains)}"
)
# ============================================================
# VARIABLES
# ============================================================
rows = []
total_block_volumes = 0
total_boot_volumes = 0
total_buckets = 0
# ============================================================
# FUNCTION: GET OBJECT COUNT USING OCI CLI
# ============================================================
def get_object_count(bucket_name_current):
try:
command = [
"oci",
"os",
"object",
"list",
"--namespace-name",
namespace,
"--bucket-name",
bucket_name_current
]
result = subprocess.run(
command,
capture_output=True,
text=True,
check=True
)
response = json.loads(result.stdout)
objects = response.get("data", [])
return len(objects)
except Exception as e:
print(
f" Object count check failed for "
f"{bucket_name_current}: {e}"
)
return 0
# ============================================================
# SCAN COMPARTMENTS
# ============================================================
for compartment in compartments:
compartment_id = compartment["id"]
compartment_name = compartment["name"]
print(
"\n----------------------------------------------"
)
print(
f"Scanning compartment: "
f"{compartment_name}"
)
# ========================================================
# BLOCK VOLUMES
# ========================================================
block_count = 0
for ad in availability_domains:
try:
block_volumes = oci.pagination.list_call_get_all_results(
blockstorage.list_volumes,
compartment_id=compartment_id,
availability_domain=ad.name
).data
for volume in block_volumes:
if volume.lifecycle_state == "TERMINATED":
continue
block_count += 1
total_block_volumes += 1
size_gb = (
volume.size_in_gbs
if volume.size_in_gbs is not None
else 0
)
vpus = (
volume.vpus_per_gb
if volume.vpus_per_gb is not None
else 0
)
rows.append([
"Block Volume",
volume.display_name or "Unnamed",
compartment_name,
volume.lifecycle_state or "UNKNOWN",
size_gb,
vpus,
volume.availability_domain or ad.name,
0
])
except Exception as e:
print(
f" Block Volume check failed "
f"in {ad.name}: {e}"
)
# ========================================================
# BOOT VOLUMES
# ========================================================
boot_count = 0
for ad in availability_domains:
try:
boot_volumes = oci.pagination.list_call_get_all_results(
blockstorage.list_boot_volumes,
compartment_id=compartment_id,
availability_domain=ad.name
).data
for boot_volume in boot_volumes:
if boot_volume.lifecycle_state == "TERMINATED":
continue
boot_count += 1
total_boot_volumes += 1
size_gb = (
boot_volume.size_in_gbs
if boot_volume.size_in_gbs is not None
else 0
)
vpus = (
boot_volume.vpus_per_gb
if boot_volume.vpus_per_gb is not None
else 0
)
rows.append([
"Boot Volume",
boot_volume.display_name or "Unnamed",
compartment_name,
boot_volume.lifecycle_state or "UNKNOWN",
size_gb,
vpus,
boot_volume.availability_domain or ad.name,
0
])
except Exception as e:
print(
f" Boot Volume check failed "
f"in {ad.name}: {e}"
)
# ========================================================
# OBJECT STORAGE BUCKETS
# ========================================================
bucket_count = 0
try:
print(
" Checking Object Storage buckets..."
)
buckets = oci.pagination.list_call_get_all_results(
object_storage.list_buckets,
namespace_name=namespace,
compartment_id=compartment_id
).data
for bucket in buckets:
bucket_count += 1
total_buckets += 1
current_bucket_name = bucket.name
# ------------------------------------------------
# GET OBJECT COUNT
# ------------------------------------------------
object_count = get_object_count(
current_bucket_name
)
# ------------------------------------------------
# ADD BUCKET TO CSV
# ------------------------------------------------
rows.append([
"Object Storage Bucket",
current_bucket_name,
compartment_name,
"ACTIVE",
0,
0,
"N/A",
object_count
])
print(
f" Bucket found: "
f"{current_bucket_name}"
)
print(
f" Compartment: "
f"{compartment_name}"
)
print(
f" Objects: "
f"{object_count}"
)
except Exception as e:
print(
f" Object Storage bucket check failed: "
f"{e}"
)
# ========================================================
# COMPARTMENT SUMMARY
# ========================================================
print(
f" Block Volumes : "
f"{block_count}"
)
print(
f" Boot Volumes : "
f"{boot_count}"
)
print(
f" Buckets : "
f"{bucket_count}"
)
# ============================================================
# ADD ZERO RECORD IF NO BLOCK VOLUME
# ============================================================
if total_block_volumes == 0:
rows.append([
"Block Volume",
"No Block Volume",
"N/A",
"N/A",
0,
0,
"N/A",
0
])
# ============================================================
# ADD ZERO RECORD IF NO BOOT VOLUME
# ============================================================
if total_boot_volumes == 0:
rows.append([
"Boot Volume",
"No Boot Volume",
"N/A",
"N/A",
0,
0,
"N/A",
0
])
# ============================================================
# ADD ZERO RECORD IF NO BUCKET
# ============================================================
if total_buckets == 0:
rows.append([
"Object Storage Bucket",
"No Bucket",
"N/A",
"N/A",
0,
0,
"N/A",
0
])
# ============================================================
# CREATE CSV
# ============================================================
with open(
csv_file,
"w",
newline="",
encoding="utf-8-sig"
) as file:
writer = csv.writer(file)
writer.writerow([
"Storage Type",
"Storage Name",
"Compartment",
"Lifecycle State",
"Size (GB)",
"VPUs/GB",
"Availability Domain",
"Object Count"
])
writer.writerows(rows)
# ============================================================
# CSV SUMMARY
# ============================================================
print("\n==============================================")
print(" CSV CREATED")
print("==============================================")
print(
f"Block Volumes : "
f"{total_block_volumes}"
)
print(
f"Boot Volumes : "
f"{total_boot_volumes}"
)
print(
f"Buckets : "
f"{total_buckets}"
)
print(
f"Total CSV rows: "
f"{len(rows)}"
)
print(
f"\nCSV file: "
f"{csv_file}"
)
# ============================================================
# UPLOAD CSV TO OBJECT STORAGE
# ============================================================
print("\n==============================================")
print(" UPLOADING CSV TO OBJECT STORAGE")
print("==============================================")
try:
with open(
csv_file,
"rb"
) as file:
object_storage.put_object(
namespace_name=namespace,
bucket_name=bucket_name,
object_name=object_name,
put_object_body=file
)
print(
"\nCSV uploaded successfully!"
)
print(
f"Bucket : "
f"{bucket_name}"
)
print(
f"Object : "
f"{object_name}"
)
except Exception as e:
print(
"\nCSV upload failed!"
)
print(
f"Error: {e}"
)
# ============================================================
# SCRIPT COMPLETED
# ============================================================
print("\n==============================================")
print(" SCRIPT COMPLETED")
print("==============================================")
save and run the script in code editor
python storage_capacity_inventory.py
csv file is also generated you can download it and it is also saved in object storage bucket