Nutanix Cloud Clusters (NC2) delivers the promise of a hybrid multicloud platform designed to run applications in private or multiple public clouds. NC2 operates as an extension of on-prem data centers and provides a hybrid cloud architecture that spans private and public clouds, operated as a single cloud.

This document covers the NC2 version 2 (v2) APIs. These APIs allow you to create and manage Azure and AWS clusters on the cloud. The v2 APIs are compliant with the Open API Specification (OAS3) 3.0 standards.

The NC2 version1 (v1) APIs were mainly used for internal purposes and are not generally available. Nutanix recommends all its customers using v1 APIs to manually migrate to the latest v2 GA APIs.

For more information on NC2, see Nutanix user guide for AWS, and Azure.

This topic serves as a roadmap, guiding users through the process of leveraging these APIs to create a 3-node AWS cluster efficiently.

Before you begin, ensure that Python 3.x is installed on your system. If not, you may need to install Python 3.x before proceeding further with utilizing the NC2 V2 APIs.

# Check python version

!python3 --version

Install the necessary Python packages using a package manager called pip.

pip3 install requests pyjwt pytz

1. requests: This package is commonly used for making HTTP requests in Python.
2. pyjwt: PyJWT is a Python library that provides functionality for encoding and decoding JSON Web Tokens (JWTs).
3. pytz: Pytz is a Python library that provides functionality for working with time zones.

Define variables necessary for specifying authentication credentials and other parameters essential for interacting with the NC2 V2 APIs effectively, such as API_KEY and KEY_ID.

To create a MyNutanix API Key (refer to the NC2 Azure user guide and NC2 AWS user guide).

# Constants and Configurations

API_KEY = 'YOUR_API_KEY'                                       
KEY_ID = 'YOUR_KEY_ID'
ISSUER = 'RANDOM_UUID'                                          # A unique issuer uuid

NC2_BASE_URL = 'https://cloud.nutanix.com/api/v2'
CREATE_CLUSTER_URL = '/clusters/aws'
HIBERNATE_CLUSTER_URL = '/clusters/{cluster_id}/hibernate'
RESUME_CLUSTER_URL = '/clusters/{cluster_id}/resume'
TERMINATE_CLUSTER_URL = '/clusters/{cluster_id}/terminate'

GET_TASK_URL = '/tasks/{task_id}'
MAX_TIMEOUT = 7200                                              # Maximum timeout of 2 hours (in seconds)
POLLING_INTERVAL = 60                                           # Initial polling interval (in seconds)
MYNUTANIX_API_KEYS_URL = 'https://apikeys.nutanix.com'          # MyNutanix API Keys URL

By configuring logs, you can ensure that relevant information and errors are captured systematically, enabling you to monitor the execution of their scripts effectively and troubleshoot any issues that may arise.

# Configure Logs
import logging

logging.basicConfig(
    level=logging.DEBUG, # set log level to debug
    format='%(asctime)s - %(levelname)s - %(message)s',
    # filename='api_interaction.log'                               # Alternatively, write logs to the file if filename param is passed
)

Authentication involves two key steps:

• Create a MyNutanix API Key (refer to the NC2 Azure user guide and NC2 AWS user guide).
• Create a JWT token using the provided API Key and Key ID.

import jwt
import datetime
import time
import hmac
import hashlib
import base64
import pytz
import uuid


# Create a JWT token
def create_jwt():
    """
    Create a JWT token using the provided API key and key ID.

    Returns:
        The JWT token and it's expiry if successful, None otherwise.
    """
    
    try:
        curr_time = datetime.datetime.utcnow()
        payload = {
            "aud": MYNUTANIX_API_KEYS_URL,
            "iat": curr_time,
            "exp": curr_time + datetime.timedelta(seconds=300),     # Nutanix recommends a 5 minute expiry
            "iss": ISSUER,
            "metadata": {
                "reason": "NC2 AWS Cluster Quick Starter"
            },
            "context": {}
        }
        body = "{}".format(KEY_ID)
        signature = base64.b64encode(
            hmac.new(bytes(API_KEY , 'latin-1'),
            bytes(body , 'latin-1'),
            digestmod=hashlib.sha512).digest()
        )
        token = jwt.encode(
            payload,
            signature,
            algorithm='HS512',
            headers={"kid": KEY_ID}
        )
        logging.debug("Token: {}" .format(token))

        return token, payload["exp"]

    except Exception as e:
        logging.error(f"JWT generation failed: {str(e)}")
        return None, None
    
create_jwt()

We will utilize the curl command to execute an API call, fetching the organizations by employing the JWT token that has been recently generated.

# List organizations

!curl --location 'https://cloud.nutanix.com/api/v2/organizations' --header 'Authorization: Bearer {{token}}'

We now proceed with another API call to retrieve the list of cloud accounts within an organization. You can choose any organization from the available options if you have access to multiple organizations. Please note that if five minutes have passed since the creation of the JWT token, you will need to generate a new one.

# List cloud accounts under an organization

!curl --location 'https://cloud.nutanix.com/api/v2/organizations/{{organization_id}}/cloud-accounts' --header 'Authorization: Bearer {{token}}'

ORGANIZATION_ID = "YOUR_ORGANIZATION_ID" # Fetched using the List Organizations API

CLOUD_ACCOUNT_ID = "YOUR_CLOUD_ACCOUNT_ID" # Fetched using the List Cloud Account API

In this section, we will see how to utilize the NC2 V2 APIs for creating a 3-node AWS cluster within a new Virtual Private Cloud (VPC). In this usage example, we demonstrate the creation of a 3-node AWS cluster within a freshly created Virtual Private Cloud (VPC). The API call will return a task when initiating the cluster creation process. It's important to note that long-running NC2 V2 API operations, including cluster creation, termination, hibernation, and resumption, will similarly return a task. To monitor the progress and status of these tasks, users can periodically poll the task endpoint using a GET request.

import requests
import json

def create_cluster(jwt):
    """
    Create an AWS cluster using the provided JWT token.

    Args:
        jwt (pair(jwt_token, jwt_expiry)): The JWT token and it's expiry time.

    Returns:
        The task object for cluster creation if successful, None otherwise.
    """
    try:

        # Define headers with JWT token for authentication
        jwt_token, _ = jwt
        headers = {
            'Authorization': f'Bearer {jwt_token}',
            'Content-Type': 'application/json',
            'Accept': 'application/json',
        }

        logging.debug(headers)
        
        # Below is payload for creating an aws cluster (adjust as needed)
        # More information on the api specification for create aws cluster can be found here,
        # https://www.nutanix.dev/api_references/nc2/#/6lo1uf2bcd4bv-create-aws-cluster
        payload = {
            "data": {
                "use_case": "general",
                "organization_id": ORGANIZATION_ID,
                "name": "nc2-v2-apis-quick-starter",
                "cloud_account_id": CLOUD_ACCOUNT_ID,
                "region": "eu-central-1",                             # eu-central-1 represents Frankfurt region. Given region mentioned for reference. 
                "host_access_ssh_key": "SSH_KEY_NAME",                # Can be generated on AWS -'KEY PAIRS' or UI.
                "license": "aos",
                "aos_version": "6.7",
                "software_tier": "pro",                               
                "capacity": [
                    {
                        "host_type": "z1d.metal",                     # Only for reference. For allowed values check API documentation.
                        "number_of_hosts": 3
                    }
                ],
                "redundancy": {
                    "factor": 2                                        
                },
                "network": {
                    "mode": "new",                                     
                    "availability_zone": "eu-central-1a",               
                    "vpc_cidr": "192.10.0.0/16",                        
                    "management_services_access_policy": {
                        "mode": "restricted",
                        "ip_addresses": [
                            "192.10.0.0/16"
                        ]
                    },
                    "prism_element_access_policy": {
                        "mode": "restricted",   
                        "ip_addresses": [
                            "192.10.0.0/16"
                        ]
                    }
                }
            }
        }

        # Send a POST request to create the cluster
        response = requests.post(NC2_BASE_URL + CREATE_CLUSTER_URL, json=payload, headers=headers)

        if response.status_code == 202:
            return response.json()
        else:
            raise Exception(f"Cluster creation failed with status code {response.status_code}, error {response.text}")

    except Exception as e:
        raise Exception(f"Error creating cluster: {str(e)}")

We will monitor a task by polling it every 60 seconds to check its status. We will keep polling until the task reaches a terminal status or until the maximum timeout of 7200 seconds (2 hours) is reached. Additionally, we'll ensure token validity by checking for expiry and generating a new token if needed.

# Function to poll the status of a task with maximum timeout

terminal_status_list = [
    "done",
    "failed",
    "cancelled"
]

def poll_task_status(jwt, task_id):

    start_time = time.time()
    
    while True:

        # Check if the JWT token is expired and create a new one if needed
        jwt_token, jwt_expiry = jwt
        if jwt_token is not None and datetime.datetime.now() >= jwt_expiry:
            jwt_token, jwt_expiry = create_jwt()
            
        headers = {
            'Authorization': f'Bearer {jwt_token}'
        }

        task_status_url = NC2_BASE_URL + GET_TASK_URL.format(task_id=task_id)
        response = requests.get(task_status_url, headers=headers)
        if response.status_code == 200:
            task_status = response.json()['data']['status']
            logging.info(f"Task status: {task_status}")
            if task_status in terminal_status_list:
                logging.debug(f"Task done!")
                return task_status
        else:
            logging.error(f"Error polling task status with status code: {response.status_code}")

        elapsed_time = time.time() - start_time
        if elapsed_time < MAX_TIMEOUT:
            sleep_time = min(POLLING_INTERVAL, MAX_TIMEOUT - elapsed_time)
            time.sleep(sleep_time)
            logging.debug(f"Sleeping for {sleep_time} ..")
        else:
            raise("Polling reached maximum timeout.")

We will combine these steps to create a 3-node AWS cluster in a new VPC and monitor the task status by polling. Upon completion of the task, we will print the cluster UUID.

# First we create a JWT token
jwt = create_jwt()

logging.debug(jwt)

# Let's create a cluster create task
task = create_cluster(jwt)
if task != None :
    task_id = task['data']['id']

    # Poll the status of the task
    if task_id is not None:
        logging.info(f"Task ID: {task_id}")
        task_status = poll_task_status(jwt, task_id)

    # Check if the task was successful
    if task_status == "success":
        logging.info("Cluster creation successful!")
        logging.info("Cluster ID: {}".format(task.get('cluster_id')))
    else:
        logging.error("Cluster creation failed!")
else:
    logging.error("Cluster creation task could not be created!")

Below are sample functions for hibernating, resuming, and terminating a cluster. Similar to the create cluster API call, these functions also return a task. We can employ the same mechanism used for polling the status to monitor these tasks.

def hibernate_cluster(cluster_id, jwt):
    """
    Hibernate a cluster.

    Args:
        cluster_id (str): The UUID of the cluster to hibernate.

    Returns:
        str: The task ID for hibernating a cluster if request is accepted, None otherwise.
    """
    try:

        # Define headers with JWT token for authentication
        jwt_token, _ = jwt
        headers = {
            'Authorization': f'Bearer {jwt_token}',
            'Content-Type': 'application/json',
        }

        # Send a POST request to hibernate the cluster
        hibernate_payload = {}
        response = requests.post(
            NC2_BASE_URL + HIBERNATE_CLUSTER_URL.format(cluster_id = cluster_id), 
            json=hibernate_payload, 
            headers=headers
        )

        logging.debug(f"Response: {response.json()}")
        if response.status_code >= 200 and response.status_code <= 299:
            # Extract the task ID from the response
            task_id = response.json()['data']['id']
            return task_id

        else:
            logging.error(f"Cluster hibernating failed with status code {response.status_code}")
            return None

    except Exception as e:
        logging.error(f"Error hibernating cluster: {str(e)}")
        return None

def resume_cluster(cluster_id, jwt):
    """
    Resume a cluster.

    Args:
        cluster_id (str): The UUID of the cluster to resume.

    Returns:
        str: The task ID for resuming a cluster if request is accepted, None otherwise.
    """
    try:

        # Define headers with JWT token for authentication
        jwt_token, _ = jwt
        headers = {
            'Authorization': f'Bearer {jwt_token}',
            'Content-Type': 'application/json',
        }

        # Send a POST request to resume the cluster
        resume_payload = {}
        response = requests.post(
            NC2_BASE_URL + RESUME_CLUSTER_URL.format(cluster_id = cluster_id), 
            json=resume_payload, 
            headers=headers
        )

        logging.debug(f"Response: {response.json()}")
        if response.status_code >= 200 and response.status_code <= 299:
            # Extract the task ID from the response
            task_id = response.json()['data']['id']
            return task_id

        else:
            print(f"Cluster resume failed with status code {response.status_code}")
            return None

    except Exception as e:
        print(f"Error resuming cluster: {str(e)}")
        return None

def terminate_cluster(cluster_id, jwt):
    """
    Terminate a cluster.

    Args:
        cluster_id (str): The UUID of the cluster to terminate.

    Returns:
        str: The task ID for terminating a cluster if request is accepted, None otherwise.
    """
    try:

        # Define headers with JWT token for authentication
        jwt_token, _ = jwt
        headers = {
            'Authorization': f'Bearer {jwt_token}',
            'Content-Type': 'application/json',
        }

        # Send a POST request to resume the cluster
        terminate_payload = {}
        response = requests.post(
            NC2_BASE_URL + TERMINATE_CLUSTER_URL.format(cluster_id = cluster_id), 
            json=terminate_payload, 
            headers=headers
        )

        logging.debug(f"Response: {response.json()}")
        if response.status_code >= 200 and response.status_code <= 299:
            # Extract the task ID from the response
            task_id = response.json()['data']['id']
            return task_id

        else:
            print(f"Cluster terminate failed with status code {response.status_code}")
            return None

    except Exception as e:
        print(f"Error terminating cluster: {str(e)}")
        return None

Note: After completing this example, it is recommended to delete the created cluster to prevent any additional costs. Additionally, if the API key was generated solely for this walkthrough, consider deleting it as well.

NC2 uses JSON Web Token (JWT) based bearer authentication. The JWT can be generated using MyNutanix API keys.

For detailed information on how to generate API keys and JWT tokens, refer to the NC2 Azure user guide and NC2 AWS user guide.

It is advisable to keep your JWT short lived, Nutanix recommends a maximum of 5 minutes as the JWT token lifespan.

List of HTTP status codes returned by the NC2 APIs