import csv
import json
import os
import requests
import sys
import time
from datetime import datetime, timedelta, timezone
def main():
timestr = time.strftime("%m%d%Y-%H%M%S")
days = int(os.environ['DAYS'])
api_key = os.environ['RAFAY_DEFAULT_API_KEY']
partner_api_key = os.environ['PARTNER_API_KEY']
url = os.environ['RAFAY_CONSOLE_URL']
currency = os.environ['CURRENCY']
metrics_row = ["Organization", "Profile Type", "Profile", "Instance", "Usage(h)", "Status", "Billing Rate", "Billing Amount(" + currency + ")"]
filename = "ncp-metrics-" + timestr + ".csv"
filename_sorted = "ncp-metrics-sorted-" + timestr + ".csv"
fd_csv = open(filename, 'w')
csv_writer = csv.writer(fd_csv)
csv_writer.writerow(metrics_row)
# Cache for profile billing data to avoid redundant API calls
billing_cache = {}
# Cache for full profile lists to avoid re-fetching when paginating
profile_cache = {}
v3_headers = {
"accept": "application/json",
"X-API-KEY": api_key,
"Content-Type": "application/json"
}
now_utc = datetime.now(timezone.utc)
current_time_str = get_formatted_utc_timestamp(now_utc)
time_delta = timedelta(days=days)
past_time_utc = now_utc - time_delta
past_time_str = get_formatted_utc_timestamp(past_time_utc)
# Process compute instances
compute_instances = list_compute_instances(v3_headers, url, past_time_str, current_time_str, "compute")
if compute_instances["count"] > 0:
process_instances(compute_instances["instance_usage_data"], v3_headers, partner_api_key, url, currency, "Compute", csv_writer, billing_cache, profile_cache)
# Reset headers for service instances
v3_headers.pop('X-ORGANIZATION-ID', None)
v3_headers.pop('X-API-KEY', None)
v3_headers['X-API-KEY'] = api_key
# Process service instances
service_instances = list_compute_instances(v3_headers, url, past_time_str, current_time_str, "service")
print("service_instances: ", service_instances["count"])
if service_instances["count"] > 0:
process_instances(service_instances["instance_usage_data"], v3_headers, partner_api_key, url, currency, "Service", csv_writer, billing_cache, profile_cache)
fd_csv.close()
sort_csv(filename, filename_sorted, "Organization", False, True)
def fetch_all_profiles(headers, url, org_name, project_name, profile_type, profile_cache):
"""Fetch all profiles with pagination support. Returns list of all profile items."""
# Create cache key for the full profile list (include org_name as API is org-scoped)
cache_key = (org_name, project_name, profile_type)
# Return cached profile list if available
if cache_key in profile_cache:
return profile_cache[cache_key]
all_items = []
limit = 50
offset = 0
# Build base URL
if profile_type == "compute":
base_url = f"https://{url}/apis/paas.envmgmt.io/v1/projects/{project_name}/computeprofiles"
else:
base_url = f"https://{url}/apis/paas.envmgmt.io/v1/projects/{project_name}/serviceprofiles"
# Fetch all pages
while True:
profile_url = f"{base_url}?limit={limit}&offset={offset}"
response = requests.get(profile_url, headers=headers)
if response.status_code == 200:
data = response.json()
items = data.get("items", [])
all_items.extend(items)
# Check if there are more pages to fetch
metadata = data.get("metadata", {})
total_count = metadata.get("count", 0)
# If we've fetched all items or no items returned, break
if len(all_items) >= total_count or len(items) == 0:
break
# Move to next page
offset += limit
else:
print(f"Failed to get profile list. Status Code: {response.status_code}")
# Cache empty list on error to avoid retrying
profile_cache[cache_key] = []
return []
# Cache the full profile list
profile_cache[cache_key] = all_items
return all_items
def profile_billing(headers, partner_api_key, url, org_name, project_name, profile_name, currency, profile_type, billing_cache, profile_cache):
# Create cache key to avoid redundant API calls
cache_key = (org_name, project_name, profile_name, currency, profile_type)
# Return cached value if available
if cache_key in billing_cache:
return billing_cache[cache_key]
# Set headers for API request
headers['X-ORGANIZATION-ID'] = org_name
headers['X-API-KEY'] = partner_api_key
# Fetch all profiles (with pagination support)
all_profiles = fetch_all_profiles(headers, url, org_name, project_name, profile_type, profile_cache)
# Search through all profiles for the matching profile_name
for bill in all_profiles:
if bill["metadata"]["name"] == profile_name:
billing_data = bill["status"]["globalSettings"]["billing"]
if not billing_data:
print("billing_data: ", billing_data)
billing_rate = None
else:
print("billing_data: ", billing_data)
billing_rate = None
if 'dimensions' in billing_data and 'instance' in billing_data['dimensions']:
for biiling_currency in billing_data['ratecard']['instance']:
if biiling_currency['currency'] == currency:
billing_rate = biiling_currency['price']
break
# Cache the result before returning
billing_cache[cache_key] = billing_rate
return billing_rate
# Profile not found in any page
print(f"Profile '{profile_name}' not found in project '{project_name}'")
billing_rate = None
billing_cache[cache_key] = billing_rate
return billing_rate
def process_instances(instances, headers, partner_api_key, url, currency, profile_type_label, csv_writer, billing_cache, profile_cache):
"""Process instances and write to CSV."""
for instance in instances:
billing_rate = profile_billing(
headers, partner_api_key, url,
instance['instance_organization_id'],
instance['instance_project_name'],
instance["profile_name"],
currency,
profile_type_label.lower(),
billing_cache,
profile_cache
)
print(f"Organization: {instance["instance_organization_name"]}")
print(f"Profile: {instance["profile_name"]}")
print(f"Instance: {instance["instance_name"]}")
print(f"Usage: {instance["usage"]}")
print("billing rate: ", billing_rate)
if billing_rate is not None:
usage = float(instance["usage"].rsplit('h')[0])
billing_amount = billing_rate * usage
print("billing amount: ", billing_amount)
else:
billing_amount = 0
if "deleted_at" in instance.keys():
status = "Deleted"
else:
status = "Running"
print(f"\n")
billing_rate_currency = currency + " " + str(billing_rate) + "/h"
# if billing_amount > 0:
csv_writer.writerow([
instance["instance_organization_name"],
profile_type_label,
instance["profile_name"],
instance["instance_name"],
instance["usage"],
status,
billing_rate_currency,
billing_amount
])
def get_formatted_utc_timestamp(dt_object: datetime) -> str:
"""Formats a datetime object into YYYY-MM-DDTHH:MM:SSZ format."""
return dt_object.strftime("%Y-%m-%dT%H:%M:%SZ")
def list_compute_instances(headers, url, start_time, end_time, profile_type):
compute_url = f"https://{url}/apis/billing.envmgmt.io/v1/metrics/partner/instance/kind/{profile_type}/usage?range_from={start_time}&range_to={end_time}&limit=10000"
response = requests.get(compute_url, headers=headers)
if response.status_code == 200:
return response.json()
else:
print(f"Failed to get compute instances. Status Code: {response.status_code}")
return response.json()
def sort_csv(input_file: str, output_file: str, sort_column_name: str, is_numeric: bool = False, reverse: bool = False):
print(f"--- Sorting CSV File ---")
print(f"Reading from '{input_file}', sorting by column '{sort_column_name}'...")
try:
with open(input_file, mode='r', newline='') as infile:
reader = csv.reader(infile)
header = next(reader)
try:
sort_column_index = header.index(sort_column_name)
except ValueError:
print(f"Error: Column '{sort_column_name}' not found in the CSV header.")
return
data = list(reader)
if is_numeric:
sorted_data = sorted(data, key=lambda row: float(row[sort_column_index]), reverse=reverse)
else:
sorted_data = sorted(data, key=lambda row: row[sort_column_index], reverse=reverse)
with open(output_file, mode='w', newline='') as outfile:
writer = csv.writer(outfile)
writer.writerow(header)
writer.writerows(sorted_data)
print(f"Successfully sorted data and saved to '{output_file}'.")
except FileNotFoundError:
print(f"Error: The file '{input_file}' was not found.")
except Exception as e:
print(f"An unexpected error occurred: {e}")
if __name__ == "__main__":
main()