This tutorial will show you how to create a simple workflow to test IP connectivity of multiple IP address with ICMP

Overview

This tutorial describes the steps to implement a workflow that will prompt for a list of IP addresses and test the connectivity to these IP addresses using ping.

Topic addressed

Topic Comment

Workflow

learn how to develop a workflow in Python

Python SDK

learn how to use the Python to leverage the MSactivator™ functionalities

Array variables

learn how to use and manipulate the array variables of a workflow

Context

learn how to read and write data in the context of a workflow instance

Sandbox environment

This tutorial can be fully implemented and tested with the minilab.

Step 1: create a new workflow

Login to the developer portal and create a new workflow with the name "Connectivity Test" and set the workflow language to Python (default is PHP).

conn test 1

Save the workflow and add it to the subtenant you want to use for testing (use "Tyrell Corporation" if don’t want to create a new subtenant)

Select the subtenant and navigate to the workflows under "Automation" > "Workflows" and click on "Connectivity Test"

Step 2: Create the variables

You need to create 2 variables, one for the IP address and one for the status of the ICMP request. Since we want to tests multiple addresses, you need to create an array of variables with one key for the IP and one for the status.

Variable name Type Display name

addresses.0.ip

String

IP

addresses.0.status

String

Ping status

Step 3: Implement the processes

Create process

Add a new process by clicking on the "+" icon in the left menu of the workflow editor. Name this process "Create connectivity test" and set its type to "CREATE".

Select the newly created process and add a task by clicking on the "+" icon on the left, near the code editor and add a task "register addresses"

Select the new task to see it’s content and remove the default code template and replace it by the code below

from msa_sdk.variables import Variables
from msa_sdk.msa_api import MSA_API
from msa_sdk.device import Device
from msa_sdk import util
import json

# this task can be used either in a CREATE or an UPDATE process

# list the variables to expose on the end used web form
dev_var = Variables()
dev_var.add('addresses.0.ip')
dev_var.add('addresses.0.status')
context = Variables.task_call(dev_var)

# get the current process id, useful for logging message to the process log file
process_id = context['SERVICEINSTANCEID']

# create a new variable Device to use the sdk function msa_sdk/device.html#msa_sdk.device.Device.ping
device = Device()

# get the list of IP addresses registered in the UI
# context['addresses'] contains the values entered for the array of variables "addresses"
addresses = context['addresses']

# for each IP addresses
i=0
for address in addresses:
  ip = address['ip']

  # call the ping function with the IP
  # the ping function will call a function implemented in the CoreEngine: the ping will be done from the container msa_sms
  ping_result = device.ping(ip)

  # log the result in the log file process-xx.log
  util.log_to_process_file(process_id, ping_result)

  # get the JSON result as a Python object
  ping_result_json = json.loads(ping_result)

  # update the addresses with the ping status
  num = len(context['addresses'])
  context['addresses'][i] = {}
  context['addresses'][i]['ip'] = ip
  context['addresses'][i]['status'] = ping_result_json['status']
  i += 1


# prepare the exit status of the task
ret = MSA_API.process_content('ENDED', 'IP addresses tested', context, True)
print(ret)

Step 4: test the process

Save the workflow and click on the button "+ Create connectivity test" to execute the process. A user form will prompt you for a list of IP addresses.

conn test 2

Click on "Run" to launch the process execution.

Next Steps

As an evolution of this workflow, you can add a process to update your test. You can simply reuse the same code as the create process, just set the process type to "Update"

You can also add a process to delete your workflow instances by adding a new process with an single empty task.