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:
- Manually handcraft the JSON payload to APIs as documented in Getting started with Xi IoT API
- 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.
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
pushd xiiot_apipython3 setup.py installpopdpython3 -c “import xiiot_api”retval=$?if [[ “$retval” -ne “0” ]] ; then exit $retvalfi
pushd xiiot_apipython setup.py installpopdpython -c “import xiiot_api”retval=$?if [[ “$retval” -ne “0” ]] ; then exit $retvalfi
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:
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_responseexcept 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.
- 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!