Quota Limit Dashboard in Oracle Cloud using Logging Analytics

In this blog we will see how we can send the compartment quota limit data to Logging Analytics and create a Dashboard for nice visualisation and reporting.

Incase if you are not familiar with Quotas in Oracle Cloud please refer this doc to understand .

I have used oci python sdk to send limit data to logging analytics. Understand what the code does before running it in your environment.

Pre-requisite :

1. Enable Logging Analytics if not done.
2. Create a log source named Quota and a json type log parser to be associated with the source. This can be achieved by importing the zip file shared here or you can create manually via the OCI console.
3. Create a new log group in Logging Analytics
4. Python script to send the quota details to Logging Analytics. Replace the log_group_id ,LA_NAMESPACE and tenancy_name(which will be at the end of the file) placeholder with the correct values.

import concurrent.futures
import oci
import json
import sys

#By default it expects the config to be present in ~/.oci/config.Specify the absolute path if its in  a different path.
config = oci.config.from_file()
tenancy_id = config['tenancy']

#Replace the placeholder with correct values
log_group_id = "<LA_log_group_ocid>"
LA_NAMESPACE = "<Logging analytics namespace>"

limits_client = oci.limits.LimitsClient(config)
identity_client = oci.identity.IdentityClient(config)
log_analytics_client = oci.log_analytics.LogAnalyticsClient(config)
quotas_client = oci.limits.QuotasClient(config)

#Fetch availability domain names
availability_domains = identity_client.list_availability_domains(compartment_id=tenancy_id).data
availability_domain_name = []
for ad in availability_domains:
    availability_domain_name.append(ad.name)


#To fetch limit_name,service_name and scope details from limit_definitions API
def list_services() -> dict:
    list_limit_definitions_response = oci.pagination.list_call_get_all_results(limits_client.list_limit_definitions,
                                                                               compartment_id=tenancy_id)
    limit_list = {}
    for i in list_limit_definitions_response.data:
        if i.are_quotas_supported and not i.is_deprecated and i.is_resource_availability_supported:
            service_name = i.service_name
            limit_name = i.name
            scope = i.scope_type
            limit_dict = {f"{limit_name}_{service_name}": scope}
            limit_list.update(limit_dict)

    return limit_list

#Function to fetch the compartment name from the quota statements.It can handle below example statements.If you have a different testcase please modify.
#Example1: set quota <servicename> <limitname> to <number> in compartment <compartmentname>
#Example2: set quota <servicename> <limitname> to <number> in compartment <parentcompartmentname>:<childcompartmentname>
#Example3: set quota <servicename> <limitname> to <number> in compartment <compartmentname> where <statements>
def get_string_after_last_colon(input_string):
    if ':' in input_string:
        parts = input_string.rsplit(':', 1)  # Split from the right, only once
        if len(parts) > 1:
            end_string=parts[1].strip()
            return end_string.split()[0]
    elif ":" not in input_string:
        index = input_string.find("compartment")
        if index != -1:
            # Slice the string from the end of "compartment"
            substring = input_string[index + len("compartment"):].strip()
            
            # Split the substring by whitespace and return the first part
            first_string = substring.split()[0] if substring else None
            return first_string
    else:
        return None


#To list compartment names from the quota policy statements
def list_quota_compartment() -> list:
    list_quotas_response = quotas_client.list_quotas(
        compartment_id=tenancy_id,
        lifecycle_state="ACTIVE",
        sort_order="ASC",
        sort_by="NAME")
    compartment_name_list = []
    for i in list_quotas_response.data:
        quota_id = i.id
        get_quota_response = quotas_client.get_quota(
            quota_id=quota_id)
        #If you have written single quota policy per compartment with multiple statement
        #you can improve this function by just taking one statement per quota.
        for statement in get_quota_response.data.statements:
            if "compartment" in statement:
                compartment = get_string_after_last_colon(statement)
                if compartment is not None:
                    compartment_name_list.append(compartment)
    print(set(compartment_name_list))
    return list(set(compartment_name_list))


#To upload data to logging analytics
def upload_to_logginganalytics(data):
    log_analytics_client.upload_log_file(
        namespace_name=LA_NAMESPACE,
        upload_name="Service_Quota",
        log_source_name="Quota",
        opc_meta_loggrpid=log_group_id,
        filename="quota.json",
        content_type="application/octet-stream",
        upload_log_file_body=data)


#To get compartment_ocid from compartment_name.
def list_compartment_id(compartment_name) -> str:
    compartment_ocid = oci.pagination.list_call_get_all_results(identity_client.list_compartments,
                                                                compartment_id=tenancy_id,
                                                                access_level="ACCESSIBLE",
                                                                compartment_id_in_subtree=True,
                                                                sort_order="ASC",
                                                                name=compartment_name,
                                                                lifecycle_state="ACTIVE").data[0].id
    return compartment_ocid


def list_compartments_quota(compartment_name):
    all_quota = []
    compartment_id = list_compartment_id(compartment_name)
    print(compartment_id)
    for name, scope in list_services().items():
        limit_name = name.split('_')[0]
        service_name = name.split('_')[1]
        if scope == "AD":
            for ad_name in availability_domain_name:
                get_resource_availability_response = limits_client.get_resource_availability(
                    service_name=service_name,
                    limit_name=limit_name,
                    availability_domain=ad_name,
                    compartment_id=compartment_id)
                resource_availability_tmp = get_resource_availability_response.data
                if resource_availability_tmp.effective_quota_value is not None and int(
                        resource_availability_tmp.available) != 0:
                    used = resource_availability_tmp.used
                    available = resource_availability_tmp.available
                    quota_value = resource_availability_tmp.effective_quota_value
                    quota_dict = {"Used": used, "Available": available, "Quota": quota_value, "Limit": limit_name,
                                  "Service": service_name, "Compartment": compartment_name, "AD": ad_name}
                    all_quota.append(quota_dict)
        else:
            get_resource_availability_response = limits_client.get_resource_availability(
                service_name=service_name,
                limit_name=limit_name,
                compartment_id=compartment_id)
            resource_availability_tmp = get_resource_availability_response.data
            if resource_availability_tmp.effective_quota_value is not None and int(
                    resource_availability_tmp.available) != 0:
                used = resource_availability_tmp.used
                available = resource_availability_tmp.available
                quota_value = resource_availability_tmp.effective_quota_value
                quota_dict = {"Used": used, "Available": available, "Quota": quota_value, "Limit": limit_name,
                              "Service": service_name, "Compartment": compartment_name}
                all_quota.append(quota_dict)

    print(f"Size of the data:{sys.getsizeof(all_quota)}")
    upload_data = json.dumps(all_quota)
    upload_to_logginganalytics(upload_data)


if __name__ == '__main__':
    compartment_names = list_quota_compartment()
    # This check is there if the quota policy statement is written as compartment <rootcompartmentname>. Substitute the root tenancy name in the below if statement
    if "<tenancy_name>" in compartment_names:
        compartment_names.remove('<tenancy_name>')
    print(compartment_names)
    with concurrent.futures.ThreadPoolExecutor() as executor:
        executor.map(list_compartments_quota, compartment_names)

NOTE: The below is the explanation of the above code by using OCI Gen AI.

This Python script is designed to fetch and upload service quota data from Oracle Cloud Infrastructure (OCI) to Logging Analytics. It retrieves the quota data for each compartment and uploads it to Logging Analytics.

Key Components

1. OCI Client Configuration: The script starts by importing the necessary OCI client libraries and configuring the client using the oci.config.from_file() method. This method reads the configuration from a file (by default, ~/.oci/config) and returns a dictionary containing the configuration details.
2. Service Clients: The script creates instances of the following OCI service clients:

• limits_client: for interacting with the Limits service
• identity_client: for interacting with the Identity service
• log_analytics_client: for interacting with the Logging Analytics service
• quotas_client: for interacting with the Quotas service

Functions: The script defines several functions to perform specific tasks:

• list_services(): retrieves a list of services with their corresponding limit names and scope details
• list_quota_compartment(): retrieves a list of compartment names from quota policy statements
• list_compartment_id(): retrieves the OCID of a compartment given its name
• list_compartments_quota(): retrieves quota data for a given compartment and uploads it to Logging Analytics
• upload_to_logginganalytics(): uploads data to Logging Analytics

Main Execution

The script’s main execution flow is as follows:

1. Retrieves a list of compartment names from quota policy statements using list_quota_compartment().
2. Iterates over the compartment names and retrieves the quota data for each compartment using list_compartments_quota().
3. Uploads the quota data to Logging Analytics using upload_to_logginganalytics().

Concurrency

The script uses the concurrent.futures.ThreadPoolExecutor() creates a pool of worker threads that can be used to execute tasks concurrently.

Notes

• The script assumes that the OCI configuration file is present in the default location (~/.oci/config). If the configuration file is located elsewhere, the script will need to be modified accordingly.
• The script uses placeholder values for the Logging Analytics namespace and log group ID, which should be replaced with the actual values.
• The script assumes that the quota policy statements are in a specific format (e.g., “set quota <servicename> <limitname> to <number> in compartment <compartmentname>”). If the quota policy statements have a different format, the script may need to be modified accordingly.

=============End of Gen AI code explanation =======================

Once the code successfully run you will see the uploads in Logging Analytics like below.
The above code will not work if quota statements are mentioned for deleted compartments.Please remove quota statement related to deleted compartment if any present before running the script.

I had tried with both ProcessPoolExecutor and ThreadPoolExecutor . Based on my testing ThreadPoolExecutor is fast for this task for my test-case where i ran against 11 compartments having quota statements.

Its recommended to run this code in a Linux compute instance in OCI . Running it in local machine might take longer time.

Refer this doc for more info related to python concurrent.futures.

If you write multiple quota statements for a compartment in one quota policy the script can be changed little bit to run little faster as mentioned in the comment. I have mentioned the line of code which needs to be changed.
Instead of looping through multiple statements it will pick the first statement and from that the compartment name can be identified.

#If you have written single quota policy per compartment with multiple statement
        #you can improve this function by just taking one statement per quota.
        for statement in get_quota_response.data.statements:
            if "compartment" in statement:


#ABOVE CODE needs to be CHANGED like below to read only the first statement:
        for statement in [get_quota_response.data.statements[0]]:
            if "compartment" in statement:

Create Label for the Limit Available values so you can alert when its about to reach a limit.

Reference : https://docs.oracle.com/en-us/iaas/logging-analytics/doc/detect-predefined-events-ingest-time.html