LAB 63 ‐ AI ‐ Extract capacity related decision making parameters for network family in CSV & also to object storage - DevWithNehaCloud/Oracle-Cloud-Infrastructure-wiki GitHub Wiki
python script
import oci
import csv
import ipaddress
# ============================================================
# OCI NETWORK CAPACITY INVENTORY
# ============================================================
#
# This script:
#
# 1. Uses OCI config file authentication
# 2. Uses us-ashburn-1 region
# 3. Gets all ACTIVE compartments
# 4. Scans all VCNs
# 5. Scans all subnets
# 6. Calculates IP capacity
# 7. Calculates used / available IPs
# 8. Identifies Public / Private subnet
# 9. Gets Route Table
# 10. Gets Security Lists
# 11. Gets DHCP Options
# 12. Gets DNS information
# 13. Generates CSV
# 14. Uploads CSV to Object Storage
#
# Existing Object Storage bucket:
# compute-capacity-inventory
#
# CSV object:
# Network Capacity Inventory.csv
#
# ============================================================
# ============================================================
# OCI AUTHENTICATION
# ============================================================
# Uses OCI config file (~/.oci/config)
config = oci.config.from_file()
# ============================================================
# OCI REGION
# ============================================================
# Region is explicitly defined because the OCI config file
# currently does not contain the "region" key.
region = "us-ashburn-1"
# Add region into config so all OCI clients use this region.
config["region"] = region
# ============================================================
# OCI CLIENTS
# ============================================================
identity = oci.identity.IdentityClient(config)
virtual_network = oci.core.VirtualNetworkClient(config)
object_storage = oci.object_storage.ObjectStorageClient(config)
# ============================================================
# TENANCY OCID
# ============================================================
tenancy_id = config["tenancy"]
# ============================================================
# OBJECT STORAGE CONFIGURATION
# ============================================================
bucket_name = "compute-capacity-inventory"
object_name = "Network Capacity Inventory.csv"
# Get Object Storage namespace automatically
namespace = object_storage.get_namespace().data
# ============================================================
# START MESSAGE
# ============================================================
print("\n============================================================")
print(" OCI NETWORK CAPACITY INVENTORY")
print("============================================================")
print(f"Region : {region}")
print(f"Namespace : {namespace}")
print(f"Bucket : {bucket_name}")
print(f"Object Name : {object_name}")
# ============================================================
# GET ALL COMPARTMENTS
# ============================================================
print("\n------------------------------------------------------------")
print("STEP 1 - Getting Compartments")
print("------------------------------------------------------------")
compartments = [
{
"id": tenancy_id,
"name": "Tenancy Root"
}
]
try:
compartment_response = oci.pagination.list_call_get_all_results(
identity.list_compartments,
tenancy_id,
compartment_id_in_subtree=True,
access_level="ANY"
)
for compartment in compartment_response.data:
if compartment.lifecycle_state == "ACTIVE":
compartments.append({
"id": compartment.id,
"name": compartment.name
})
print(
f"Total compartments scanned: "
f"{len(compartments)}"
)
except Exception as e:
print("\nERROR while getting compartments.")
print(e)
raise
# ============================================================
# COUNTERS
# ============================================================
total_vcns = 0
total_subnets = 0
public_subnets = 0
private_subnets = 0
total_ip_capacity = 0
total_usable_ip_capacity = 0
total_used_ips = 0
total_available_ips = 0
rows = []
# ============================================================
# SCAN ALL COMPARTMENTS
# ============================================================
print("\n------------------------------------------------------------")
print("STEP 2 - Scanning VCNs and Subnets")
print("------------------------------------------------------------")
for compartment in compartments:
compartment_id = compartment["id"]
compartment_name = compartment["name"]
print(
f"\nScanning compartment: "
f"{compartment_name}"
)
# ========================================================
# GET VCNs
# ========================================================
try:
vcns = oci.pagination.list_call_get_all_results(
virtual_network.list_vcns,
compartment_id
).data
except Exception as e:
print(
f"Unable to get VCNs from: "
f"{compartment_name}"
)
print(f"Error: {e}")
continue
# ========================================================
# PROCESS EACH VCN
# ========================================================
for vcn in vcns:
# ----------------------------------------------------
# IGNORE TERMINATED VCN
# ----------------------------------------------------
if vcn.lifecycle_state == "TERMINATED":
continue
total_vcns += 1
# ----------------------------------------------------
# VCN NAME
# ----------------------------------------------------
vcn_name = vcn.display_name
# ----------------------------------------------------
# VCN CIDR
# ----------------------------------------------------
vcn_cidr = ""
if vcn.cidr_blocks:
vcn_cidr = ", ".join(
vcn.cidr_blocks
)
elif vcn.cidr_block:
vcn_cidr = vcn.cidr_block
# ----------------------------------------------------
# VCN DNS LABEL
# ----------------------------------------------------
vcn_dns_label = vcn.dns_label or ""
print(
f" VCN: {vcn_name}"
)
# ====================================================
# GET SUBNETS
# ====================================================
try:
subnets = oci.pagination.list_call_get_all_results(
virtual_network.list_subnets,
compartment_id,
vcn_id=vcn.id
).data
except Exception as e:
print(
f" Unable to get subnets for VCN: "
f"{vcn_name}"
)
print(f" Error: {e}")
continue
# ====================================================
# PROCESS EACH SUBNET
# ====================================================
for subnet in subnets:
# ------------------------------------------------
# IGNORE TERMINATED SUBNET
# ------------------------------------------------
if subnet.lifecycle_state == "TERMINATED":
continue
total_subnets += 1
# ------------------------------------------------
# SUBNET BASIC INFORMATION
# ------------------------------------------------
subnet_name = subnet.display_name
subnet_cidr = subnet.cidr_block or ""
lifecycle_state = subnet.lifecycle_state
subnet_id = subnet.id
# ------------------------------------------------
# PUBLIC / PRIVATE SUBNET
# ------------------------------------------------
if subnet.prohibit_public_ip_on_vnic:
subnet_type = "Private"
private_subnets += 1
else:
subnet_type = "Public"
public_subnets += 1
# =================================================
# IP CAPACITY
# =================================================
total_capacity = 0
usable_capacity = 0
try:
network = ipaddress.ip_network(
subnet_cidr,
strict=False
)
total_capacity = network.num_addresses
# OCI reserves 3 IPv4 addresses
# in an IPv4 subnet.
#
# Usable capacity:
# Total IPs - 3 reserved IPs
if network.version == 4:
usable_capacity = max(
total_capacity - 3,
0
)
else:
usable_capacity = total_capacity
except Exception:
total_capacity = 0
usable_capacity = 0
# =================================================
# USED PRIVATE IPs
# =================================================
used_ips = 0
try:
private_ips = (
oci.pagination.list_call_get_all_results(
virtual_network.list_private_ips,
subnet_id=subnet_id
).data
)
used_ips = len(
private_ips
)
except Exception as e:
print(
f" Unable to get private IPs "
f"for subnet: {subnet_name}"
)
print(
f" Error: {e}"
)
# =================================================
# AVAILABLE IPs
# =================================================
available_ips = max(
usable_capacity - used_ips,
0
)
# =================================================
# UTILIZATION
# =================================================
if usable_capacity > 0:
utilization_percent = round(
(
used_ips /
usable_capacity
) * 100,
2
)
else:
utilization_percent = 0
# =================================================
# ROUTE TABLE
# =================================================
route_table_name = ""
route_table_id = (
subnet.route_table_id or ""
)
if route_table_id:
try:
route_table = (
virtual_network.get_route_table(
route_table_id
).data
)
route_table_name = (
route_table.display_name or ""
)
except Exception:
route_table_name = ""
# =================================================
# SECURITY LISTS
# =================================================
security_list_names = []
security_list_ids = (
subnet.security_list_ids or []
)
for security_list_id in security_list_ids:
try:
security_list = (
virtual_network.get_security_list(
security_list_id
).data
)
security_list_names.append(
security_list.display_name
)
except Exception:
pass
security_lists = ", ".join(
security_list_names
)
# =================================================
# DHCP OPTIONS
# =================================================
dhcp_options_name = ""
dhcp_options_id = (
subnet.dhcp_options_id or ""
)
if dhcp_options_id:
try:
dhcp_options = (
virtual_network.get_dhcp_options(
dhcp_options_id
).data
)
dhcp_options_name = (
dhcp_options.display_name or ""
)
except Exception:
dhcp_options_name = ""
# =================================================
# DNS INFORMATION
# =================================================
subnet_domain_name = (
subnet.subnet_domain_name or ""
)
# =================================================
# IP VERSION
# =================================================
ip_version = "IPv4"
if ":" in subnet_cidr:
ip_version = "IPv6"
# =================================================
# UPDATE COUNTERS
# =================================================
total_ip_capacity += total_capacity
total_usable_ip_capacity += (
usable_capacity
)
total_used_ips += used_ips
total_available_ips += (
available_ips
)
# =================================================
# ADD DATA TO CSV
# =================================================
rows.append([
region,
compartment_name,
vcn_name,
vcn_cidr,
vcn_dns_label,
subnet_name,
subnet_cidr,
ip_version,
subnet_type,
total_capacity,
usable_capacity,
used_ips,
available_ips,
utilization_percent,
route_table_name,
security_lists,
dhcp_options_name,
subnet_domain_name,
lifecycle_state
])
# =================================================
# DISPLAY SUBNET INFORMATION
# =================================================
print(
f" Subnet: {subnet_name} | "
f"{subnet_cidr} | "
f"{subnet_type} | "
f"Used: {used_ips} | "
f"Available: {available_ips}"
)
# ============================================================
# CREATE CSV
# ============================================================
print("\n------------------------------------------------------------")
print("STEP 3 - Creating CSV")
print("------------------------------------------------------------")
csv_file = "Network Capacity Inventory.csv"
with open(
csv_file,
"w",
newline="",
encoding="utf-8-sig"
) as file:
writer = csv.writer(
file,
delimiter=",",
quoting=csv.QUOTE_MINIMAL
)
# ========================================================
# CSV HEADERS
# ========================================================
writer.writerow([
"Region",
"Compartment",
"VCN Name",
"VCN CIDR",
"VCN DNS Label",
"Subnet Name",
"Subnet CIDR",
"IP Version",
"Subnet Type",
"Total IP Capacity",
"Usable IP Capacity",
"Used IPs",
"Available IPs",
"Utilization (%)",
"Route Table",
"Security Lists",
"DHCP Options",
"Subnet Domain Name",
"Lifecycle State"
])
# ========================================================
# WRITE ROWS
# ========================================================
writer.writerows(
rows
)
print(
f"CSV generated successfully: "
f"{csv_file}"
)
# ============================================================
# DISPLAY SUMMARY
# ============================================================
print("\n============================================================")
print(" OCI NETWORK CAPACITY INVENTORY")
print("============================================================")
print(
f"Total VCNs : "
f"{total_vcns}"
)
print(
f"Total Subnets : "
f"{total_subnets}"
)
print(
f"Public Subnets : "
f"{public_subnets}"
)
print(
f"Private Subnets : "
f"{private_subnets}"
)
print(
f"Total IP Capacity : "
f"{total_ip_capacity}"
)
print(
f"Usable IP Capacity : "
f"{total_usable_ip_capacity}"
)
print(
f"Used IPs : "
f"{total_used_ips}"
)
print(
f"Available IPs : "
f"{total_available_ips}"
)
# ============================================================
# OVERALL UTILIZATION
# ============================================================
if total_usable_ip_capacity > 0:
overall_utilization = round(
(
total_used_ips /
total_usable_ip_capacity
) * 100,
2
)
else:
overall_utilization = 0
print(
f"Overall Utilization : "
f"{overall_utilization}%"
)
print(
f"\nCSV generated successfully: "
f"{csv_file}"
)
# ============================================================
# OBJECT STORAGE UPLOAD
# ============================================================
print("\n------------------------------------------------------------")
print("STEP 4 - 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 : {bucket_name}"
)
print(
f"Object : {object_name}"
)
except Exception as e:
print(
"\nObject Storage upload failed."
)
print(
"Please check:"
)
print(
"1. Bucket name"
)
print(
"2. OCI authentication"
)
print(
"3. Object Storage permissions"
)
print(
"4. Object Storage namespace"
)
print(
f"\nError: {e}"
)
# ============================================================
# SCRIPT COMPLETED
# ============================================================
print("\n============================================================")
print(" SCRIPT COMPLETED")
print("============================================================")
print(
"\nNetwork Capacity Inventory completed successfully."
)
print(
f"Local CSV : {csv_file}"
)
print(
f"Object Store : {bucket_name}"
)
print(
f"Object Name : {object_name}"
)
print("============================================================")
to run the script
python network_capacity_inventory.py
verify the csv file in object storage bucket