Automated Application Deployment using Nutanix Xi IoT Python SDK

After reading this story, hopefully you will have more insight into Nutanix Xi IoT — especially how it helps you to automate application deployment.

• What is Nutanix Xi IoT?
• What is an Application in Nutanix Xi IoT context?
• Example use case
• What is the Nutanix Xi IoT Python SDK, and How Can I install and Use It?
• How Can I deploy an Application on One or More Edges Programmatically?
• How Can I Programmatically Get the Status of Application(s) Running on Edges?
• How to Create a CI/CD Pipeline of Your application for Xi IoT Infrastructure

What is Nutanix Xi IoT?

Nutanix Xi IoT: An Overview for Developers describes Nutanix Xi IoT and how it can help your organization deploy applications to process sensor data and to implement AI inferencing at the edge.

What is an Application in Nutanix Xi IoT Context?

Nutanix Xi IoT Application Containers for Developers defines applications in detail in the Nutanix Xi IoT context.

Example Use Case

Now that you know about Nutanix Xi IoT and how it will help you to deploy an application for processing your sensor data or AI inferencing applications through a cloud management console user interface, let’s see how can we automate it for thousands of edges.

Imagine a situation where you as an application developer need to upload patches after each fix to an application. And you also want to automatically integrate the same in your CI/CD pipeline so that it cycles through development test, stage test, and production deployment automatically.

Ideally, you would use APIs where you can codify and automate the process. Choose from these two ways to do this:

1. Manually handcraft the JSON payload to APIs as documented in Getting started with Xi IoT API
2. Better — Use an SDK to call functions with parameters without handcrafting the payload., reducing risk and errors

Read on to find out more about choice 2!

What is the Nutanix Xi IoT Python SDK, and How Can I Install and Use It?

The Xi IoT Python SDK is Swagger generated code from Xi IoT’s swagger API JSON file. Here’s how to generate that code.

Prerequisites

Install swagger-codegen-cli

Install Java

You can find the steps on how to install Java here.

Download the swagger-codegen-cli jar file

wget http://central.maven.org/maven2/io/swagger/codegen/v3/swagger-codegen-cli/3.0.8/swagger-codegen-cli-3.0.8.jar -O swagger-codegen-cli.jar

Install python, pip, virtualenv (versions 2.x or 3.x).

You should have python and pip installed. More information is here and is based on your OS.

Generate Xi IoT API client code

Fetch the API Swagger definition JSON file

wget https://iot.nutanix.com/xi_iot_api.json -O xi_iot_api.json

Generate Xi IoT API client code using swagger-codegen-cli

java -jar swagger-codegen-cli.jar generate -i ./xi_iot_api.json \
-l python \
-o ./xiiot_api \
-DpackageName=xiiot_api,projectName=xi_iot_sdk

Install and Test the Xi IoT python Client SDK

For python3:

pushd xiiot_api
python3 setup.py install
popd
python3 -c “import xiiot_api”
retval=$?
if [[ “$retval” -ne “0” ]] ; then
    exit $retval
fi

For python2.x:

pushd xiiot_api
python setup.py install
popd
python -c “import xiiot_api”
retval=$?
if [[ “$retval” -ne “0” ]] ; then
     exit $retval
fi

How Can I deploy an Application on One or More Edges Programmatically?

Call an API to create an application. For a working sample, look at the code in the following link:

https://github.com/nutanix/xi-iot-python-sdk

Log On to the Xi IoT Cloud Management Console

The Xi IoT cloud management console and cloud endpoint are hosted at https://iot.nutanix.com.

You need a local account to log on. See Getting Started with Xi IoT APIs for information describing how to create a local account. This code shows how to log on using this API call in the Xi IoT SDK.

def loginUsingAuthTag(self, xi_iot_endpoint, userEmail, userPwd):
    configuration = xiiot_api.Configuration()
    configuration.host = xi_iot_endpoint
    api_client = xiiot_api.ApiClient(configuration=configuration)
    # create an instance of the API class
    api_instance = xiiot_api.AuthApi(api_client=api_client)
    request = xiiot_api.Credential(userEmail, userPwd)
    try:
        # Lets the user log in.
        api_response = api_instance.login_call_v2(request)
        # Configure API key authorization: BearerToken
        configuration.api_key[‘Authorization’] = api_response.token
        configuration.api_key_prefix[‘Authorization’] = ‘Bearer’
        configuration.debug = False
        return configuration, api_response
    except ApiException as e:
        logging.error(“Exception when calling AuthApi->login_call_v2: %s”, e)
        raise

Call this login function from any API wrapper class as follows:

class ApplicationApiWrapper():
    “””ApplicationApi Test Class”””
    def __init__(self, xi_iot_endpoint, userEmail, userPwd):
        self.configuration, _ = self.loginUsingAuthTag(xi_iot_endpoint, userEmail, userPwd)
        self.authorization = self.configuration.get_api_key_with_prefix(‘Authorization’)
        self.api_client = xiiot_api.ApiClient(configuration=self.configuration)
        self.ApplicationApi = xiiot_api.api.application_api.ApplicationApi(api_client=self.api_client)
        self.AppStatusApi = xiiot_api.api.application_status_api.ApplicationStatusApi(api_client=self.api_client)
        self.ProjectApi = xiiot_api.ProjectApi(api_client=self.api_client)
        self.EdgeApi = xiiot_api.api.edge_api.EdgeApi(api_client=self.api_client)

Create an Application Using the API

def create_application(self, app_name, app_manifest, project_id, edge_ids):
    # ApplicationV2 | Describes the application creation request.
    body = xiiot_api.ApplicationV2(name=app_name, app_manifest=app_manifest, project_id=project_id, edge_ids=edge_ids)
    try:
        api_response = self.ApplicationApi.application_create_v2(body, self.authorization)
        # returns CreateDocumentResponseV2 object
        return api_response
    except ApiException as e:
        raise

Delete an Application Using the API

def delete_application(self, app_id):
    try:
        api_response =    self.ApplicationApi.application_delete_v2(self.authorization, app_id)
        # returns DeleteDocumentResponseV2 object
        return api_response
    except ApiException as e:
        raise

Sample code which calls the above create/delete functions

logging.info(“Creating flask-web-server-blog app…”)
createDocumentResponseV2 = app_api_wrapper.create_application(app_name, app_manifest, project_id, edge_ids)
deleteDocumentResponseV2 = app_api_wrapper.delete_application(createDocumentResponseV2.id)

How Can I Programmatically Get the Status of Application(s) Running on Edges?

To get status of all applications or a specific application:

def get_applications_statuses(self, app_id=None):
try:
    if app_id is not None:
    api_response = self.AppStatusApi.application_status_get_v2(
self.authorization, app_id)
else:
    api_response = self.AppStatusApi.application_status_list_v2(self.authorization)
    logging.info(“API output: %s” % api_response)
    # returns ApplicationStatusListPayload object
    return api_response
except ApiException as e:
    raise

Sample code that calls above status functions:

applicationStatusListPayload = app_api_wrapper.get_applications_statuses(createDocumentResponseV2.id)
logging.info(“Received app status for app id: %s”, createDocumentResponseV2.id)

You can find the above code uploaded here.

How to Create a CI/CD Pipeline of Your Application for Xi IoT Infrastructure

CI/CD is Continuous Integration and Continuous Deployment. If you have an existing CI/CD pipeline in your organization, then you can push the updated application using these API calls to iot.nutanix.com. The application is then deployed on the requested edges by the Xi IoT Infrastructure.

Here’s the general workflow:

• Create an application source code repository (for example, in github) with a Dockerfile and Application code.

Continuous Integration

• For each check-in, build a new container artifact to package the changes with the modified binary.
• Run unit test code as pre-check in tests and, if the tests pass, push the container to the development container registry.

If successful, go to the next step…

Continuous Development Setup Deployment Workflow

• For functional testing, you can create one or more development edges and register them with iot.nutanix.com. Use a script as described here to create an application with these dev edges. Then verify that application gets deployed and works as expected.

If successful go to the next step…

Continuous Stage Deployment Workflow

• For integration testing, you can use existing long-running test edges with the latest binaries and register them with iot.nutanix.com.
• Use a script as described here to create or update an application with these stage edges.
• Verify that the application is deployed and works as expected with all other services. If so, you can be confident that your application is ready to be pushed to production. If something does not work, you can always roll back the deployed application by repeating the entire process again from dev to staging.

If successful, then go to the next step…

Continuous Production Deployment Workflow

• Once stage tests pass, you can safely upload your application to iot.nutanix.com on your production edges.

Please use these steps as an example and feel free to use any variant of them and even include the Canary way if needed to deploy your application. The Xi IoT Platform is flexible enough to accommodate most scenarios.

Hope to see more apps from each of you!