What insight will you get after reading this blog:
- 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 needs 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 then 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.
- Install swagger-codegen-cli
- Install Java. You can find the steps on how to install Java in the official Java documentation.
- 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). Please see this Real Python article for OS-specific installation instructions.
- Generate Xi IoT API client code – step 1
- 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 – step 2
- 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 – python 3:
pushd xiiot_api python3 setup.py install popd python3 -c "import xiiot_api" retval=$? if [[ "$retval" -ne "0" ]] ; then exit $retval fi
- Install and test the Xi IoT python client SDK – python 2.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 programmatically deploy an application on one or more edges?
Call an API to create an application. For a working sample, look at the code in the following link:
First, login to the Nutanix Xi IoT Cloud Management Console. This location provides both the console and the cloud endpoint.
You need a local account to log on. See Getting Started with Xi IoT APIs for information describing how to create a local account.
The following 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 app...") createDocumentResponseV2 = app_api_wrapper.create_application(app_name, app_manifest, project_id, edge_ids) logging.info("Created application with id: %s", createDocumentResponseV2.id) logging.info("Deleting application with id: %s...", createDocumentResponseV2.id) deleteDocumentResponseV2 = app_api_wrapper.delete_application(createDocumentResponseV2.id) logging.info("Deleted application with id: %s", deleteDocumentResponseV2.id)
How Can I Programmatically Get the Status of Application(s) Running on Edges?
The following shows how to get the 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("application_status_list_v2 API output: %s" % api_response) # returns ApplicationStatusListPayload object return api_response except ApiException as e: raise
Sample code that calls above status functions
logging.info("Getting app status for app id: %s...", createDocumentResponseV2.id) applicationStatusListPayload = app_api_wrapper.get_applications_statuses(createDocumentResponseV2.id) logging.info("Received app status for app id: %s", createDocumentResponseV2.id)
All sample code above can be downloaded from the Nutanix GitHub repo.
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 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 stage.
- 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 you!!