DNAC Command Runner Question

[zhenningx]

Hello, I am using the command runner code here: https://github.com/CiscoDevNet/dne-dna-code/blob/master/intro-dnac/04_Cmd_Runner/cmd_runner.py

It can get the command output but it finishes with an error as below. If I am trying to run this with multiple switches, it can only get the output for the first switch and then stops with the error. Any ideas? Thanks!

hostname id
S1743 1025e9d4-45ce-46b2-85ef-def4a9d9b990
executing ios command --> show ver | inc RELEASE
Command runner Initiated! Task ID --> bbbab842-8c68-42ab-9e3d-e800c754b763
Retrieving Path Trace Results....
File ID --> 63a0f84d-4bc9-42a8-98df-1baab25c454b
{'SUCCESS': {'show ver | inc RELEASE': 'show ver | inc RELEASE\nCisco IOS Software [Fuji], Catalyst L3 Switch Software (CAT9K_IOSXE), Version 16.9.8, RELEASE SOFTWARE (fc4)\nBOOTLDR: System Bootstrap, Version 16.9.1r[FC3], RELEASE SOFTWARE (P)\nS1743#'}, 'FAILURE': {}, 'BLACKLISTED': {}}
Traceback (most recent call last):

File "C:\Python Codes\command_runner_bulk_device.py", line 112, in <module>
get_device_list()

File "C:\Python Codes\command_runner_bulk_device.py", line 63, in get_device_list
initiate_cmd_runner(token,device['id']) # initiate command runner

File "C:\Python Codes\command_runner_bulk_device.py", line 82, in initiate_cmd_runner
get_task_info(task_id, token)

File "C:\Python Codes\command_runner_bulk_device.py", line 98, in get_task_info
get_task_info(task_id, token)

File "C:\Python Codes\command_runner_bulk_device.py", line 98, in get_task_info
get_task_info(task_id, token)

File "C:\Python Codes\command_runner_bulk_device.py", line 98, in get_task_info
get_task_info(task_id, token)

File "C:\Python Codes\command_runner_bulk_device.py", line 98, in get_task_info
get_task_info(task_id, token)

File "C:\Python Codes\command_runner_bulk_device.py", line 98, in get_task_info
get_task_info(task_id, token)

File "C:\Python Codes\command_runner_bulk_device.py", line 98, in get_task_info
get_task_info(task_id, token)

File "C:\Python Codes\command_runner_bulk_device.py", line 98, in get_task_info
get_task_info(task_id, token)

File "C:\Python Codes\command_runner_bulk_device.py", line 98, in get_task_info
get_task_info(task_id, token)

File "C:\Python Codes\command_runner_bulk_device.py", line 98, in get_task_info
get_task_info(task_id, token)

File "C:\Python Codes\command_runner_bulk_device.py", line 99, in get_task_info
get_cmd_output(token, file_id)

File "C:\Python Codes\command_runner_bulk_device.py", line 108, in get_cmd_output
print(cmd_result.json()[0]["commandResponses"])

KeyError: 0

[balaji.bandi]

i would edit the python file check those lines. also check the requirements to run this environment in main GIT folder:

https://github.com/CiscoDevNet/dne-dna-code

[zhenningx]

I checked the code. To me it looks like just the code does not finish property.

If I add a line sys.exit() to the end of get_task_info(task_id, token) method, it would finish nicely for one device query:

def get_task_info(task_id, token):
url = "https://{}/api/v1/task/{}".format(DNAC_URL, task_id)
hdr = {'x-auth-token': token, 'content-type' : 'application/json'}
task_result = requests.get(url, headers=hdr, verify=False)
file_id = task_result.json()['response']['progress']
if "fileId" in file_id:
unwanted_chars = '{"}'
for char in unwanted_chars:
file_id = file_id.replace(char, '')
file_id = file_id.split(':')
file_id = file_id[1]
print("File ID --> ", file_id)
else: # keep checking for task completion
get_task_info(task_id, token)
get_cmd_output(token, file_id)
sys.exit()

But I cannot do this for multiple devices query. Any other suggestions why this code does not end properly without adding sys.exit()?

Thanks!

[bigevilbeard]

@zhenningx i used the sandboxdnac2.cisco.com (i am not sure what you are using here). When i first ran this the output was

dne-dna-code/intro-dnac/04_Cmd_Runner on  master [!?] via  v3.8.0 (new_venv) took 45s
❯ python cmd_runner.py
hostname                 id
leaf1.test.com           a25646c1-5c3d-4f18-b158-0f689a255a03
leaf2.test.com           beb47905-7b80-4e6e-b352-a04098cf79db
spine1.abc.in            23cad7df-acbf-41c0-beec-11d4f66364ad
Traceback (most recent call last):
  File "cmd_runner.py", line 100, in <module>
    get_device_list()
  File "cmd_runner.py", line 53, in get_device_list
    print("{0:25}{1:25}".format(device['hostname'], device['id']))
TypeError: unsupported format string passed to NoneType.__format__

I update line 53 with an f string

    device_list = resp.json()
    print("{0:25}{1:25}".format("hostname", "id"))
    for device in device_list['response']:
        print(f"{device['hostname'], device['id']}")
        # print("{0:25}{1:25}".format(device['hostname'], device['id']))
    initiate_cmd_runner(token) # initiate command runner

This now printing

dne-dna-code/intro-dnac/04_Cmd_Runner on  master [!?] via  v3.8.0 (new_venv)
❯ python cmd_runner.py
hostname                 id
('leaf1.test.com', 'a25646c1-5c3d-4f18-b158-0f689a255a03')
('leaf2.test.com', 'beb47905-7b80-4e6e-b352-a04098cf79db')
('spine1.abc.in', '23cad7df-acbf-41c0-beec-11d4f66364ad')
(None, '1e5db4ef-a711-4802-8b10-3571048d96c5')
Copy/Past a device ID here:

Hope this helps.

[zhenningx]

Thanks but that did not help. The issue happens after I copy and paste an id to run the command.

And sorry I forgot that I made one more change to the code. The code in the Github does work without errors, but it prints too many unnecessary outputs. I only want to print out the command response, so I changed line 93 to be: print(cmd_result.json()[0]["commandResponses"])

Like this:

def get_cmd_output(token,file_id):
url = "https://{}/api/v1/file/{}".format(DNAC_URL, file_id)
hdr = {'x-auth-token': token, 'content-type': 'application/json'}
cmd_result = requests.get(url, headers=hdr, verify=False)
#print(json.dumps(cmd_result.json(), indent=4, sort_keys=True))
print(cmd_result.json()[0]["commandResponses"])

So basically instead of dump all the cmd_result.json, I only want the commandResponses. This does print out only the info I want, but it ends with the errors above. Any suggestions why this change would cause those errors?

Thanks!

[bigevilbeard]

@zhenningx i would suggest looking at the SDK for this https://blogs.cisco.com/developer/network-automation-cisco-dna-center-sdk-2

Hope this helps.

[zhenningx]

Thanks. The SDK solution is much simpler than the previous way. I run the code but how do I retrieve the information from the cmd_output? It looks like cmd_output is a class, not an object.

cmd_output
<urllib3.response.HTTPResponse object at 0x000001DF5F781988>

type(cmd_output)
<class 'urllib3.response.HTTPResponse'>

Thanks.

[bigevilbeard]

@zhenningx yes this is a class for CommandRunner, this wraps the DNA Center Command Runner API and exposes the API as native Python methods that return native python objects, here is the link to the docs. Hope this helps.

https://dnacentersdk.readthedocs.io/en/latest/api/api.html#dnacentersdk.api.v1_3_1.command_runner.CommandRunner

[zhenningx]

Sorry still have no idea what to do next. How do I get the output of the command with this returned class? What can I do with this class?

And where is this method defined: get_all_keywords_of_clis_accepted()

It is not part of the cmd_output class:

dir(cmd_output)
['CONTENT_DECODERS', 'DECODER_ERROR_CLASSES', 'REDIRECT_STATUSES', '__abstractmethods__', '__class__', '__del__', '__delattr__', '__dict__', '__dir__', '__doc__', '__enter__', '__eq__', '__exit__', '__format__', '__ge__', '__getattribute__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__iter__', '__le__', '__lt__', '__module__', '__ne__', '__new__', '__next__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', '_abc_impl', '_body', '_checkClosed', '_checkReadable', '_checkSeekable', '_checkWritable', '_connection', '_decode', '_decoder', '_error_catcher', '_flush_decoder', '_fp', '_fp_bytes_read', '_handle_chunk', '_init_decoder', '_init_length', '_original_response', '_pool', '_request_url', '_update_chunk_length', 'auto_close', 'chunk_left', 'chunked', 'close', 'closed', 'connection', 'data', 'decode_content', 'drain_conn', 'enforce_content_length', 'fileno', 'flush', 'from_httplib', 'get_redirect_location', 'getheader', 'getheaders', 'geturl', 'headers', 'info', 'isatty', 'isclosed', 'length_remaining', 'msg', 'read', 'read_chunked', 'readable', 'readinto', 'readline', 'readlines', 'reason', 'release_conn', 'retries', 'seek', 'seekable', 'status', 'stream', 'strict', 'supports_chunked_reads', 'tell', 'truncate', 'version', 'writable', 'writelines']

Thanks.

[zhenningx]

The blog says "Retrieve the results and save them to file", but it is missing the "save them to file" part. How do I save the results to a file?

Thanks.

[bigevilbeard]

@zhenningx please find sample below

#!/usr/bin/env python

"""
Summary:
    Example use case of DNACENTERSDK module
    ref: https://dnacentersdk.readthedocs.io/en/latest/api/intro.html


Usage:
    
"""
import os
import time
from datetime import date
from argparse import ArgumentParser
from pathlib import Path
from getpass import getpass, getuser
import base64
import urllib3
urllib3.disable_warnings()

# create logging structure
import logging

# Additional packages
if __name__ != "__main__":
    from dnacentersdk import DNACenterAPI, ApiError
from tabulate import tabulate
import yaml

logger = logging.getLogger(__name__)
logger.setLevel(logging.WARN)
formatter = logging.Formatter(
    "%(asctime)s::%(name)s::-%(module)s::-: %(funcName)s -  %(levelname)s :: - %(message)s"
)
# setting stream handler to print to STDOUT
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)

# create log file along with printing to console
# _file_name = Path(__file__)
# log_file_name = str(_file_name.stem) + '_' + date.isoformat(date.today() + ".log"

log_file_name = "all_logging_" + date.isoformat(date.today()) + ".log"

file_handler = logging.FileHandler(log_file_name)
file_handler.setLevel(logging.WARN)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)


def get_creds_base64():
    """
    Create Base64 creds from the user input and set ENV varaible
    """
    if os.getenv('DNA_CENTER_ENCODED_AUTH'):
        return os.getenv('DNA_CENTER_ENCODED_AUTH')
    user = input('USERNAME : ')
    passw = getpass(prompt='PASSWORD : ')
    os.environ["DNA_CENTER_USERNAME"] = user
    os.environ["DNA_CENTER_PASSWORD"] = passw
    userPass = user + ":" + passw
    os.environ["DNA_CENTER_ENCODED_AUTH"] = base64.b64encode(userPass.encode()).decode()
    return base64.b64encode(userPass.encode()).decode()


def set_env_var_dnac(env_var_file):
    """
    Use yml file to set enviroment variables to use DNACENTERSDK

    Args:
        env_var_file (filepath): Path to file with DNACENTERSDK environment variables
    """
    logger.debug("Function start with args:")
    local_args = locals()
    logger.debug("Function args : ")
    for arg in local_args.items():
        logger.debug(arg)
        logger.debug("*" * 20)
    # Get creds to do something
    get_creds_base64()
    with open(env_var_file, "r") as fp:
        env_vars = yaml.safe_load(fp)
    for key, val in env_vars.items():
        os.environ[key.strip()] = str(val)
    # env vals set for script
    logger.info(" Starting DNAC script with :")
    logger.info(" Authorization : {}".format(os.environ.get("DNA_CENTER_ENCODED_AUTH")))
    logger.info(" Version : {}".format(os.environ.get("DNA_CENTER_VERSION")))
    logger.info(" Base URL : {}".format(os.environ.get("DNA_CENTER_BASE_URL")))
    logger.info(
        "cert verifciation : {}".format(str(os.environ.get("DNA_CENTER_VERIFY")))
    )


def get_device_list(api, *args, **kwargs):
    """
    summary :
        gets LIst of devices on the DNAC

    """

    # bit of house keeping like printing a log msg
    logger.debug("Function start with args:")
    for param in locals().items():
        logger.debug("arg = {}".format(param))
    # start coding here
    try:
        devices = api.devices.get_device_list()
        # devices = api.devices.get_device_list(platform_id="C9300-48U")
    except (ApiError) as err:
        logger.error("XXXX  API Call failed XXXX")
        logger.error(err)
    # Print the devices in the response
    table = {}
    for dev in devices.response:
        print(
            dev.hostname,
            dev.managementIpAddress,
            dev.platformId,
            dev.softwareType,
            dev.softwareVersion,
        )
        table.update(
            {
                dev.instanceUuid: [
                    dev.instanceUuid,
                    dev.hostname,
                    dev.managementIpAddress,
                    dev.platformId,
                    dev.softwareType,
                    dev.softwareVersion,
                ]
            }
        )

    # print in table format
    print(
        tabulate(
            list(table.values()),
            headers=[
                "UUID",
                "Hostname",
                "Mgmt_IP",
                "Platform",
                "sw_type",
                "sw_version",
            ],
        )
    )
    return table


# -- end of method --#


def get_command(api, list_cmd, list_uuid):
    """get the list of commands to be run on the list of devices (UUIDs)
    retrun command output as a dict {uuid : [cmd1_out, .. cmdn_out]}

    Args:
        api ([type]): [description]
        list_cmd ([type]): [description]
        list_uuid ([type]): [description]
    """
    # post commands and get taskIDs
    get_cmd_taskid = api.command_runner.run_read_only_commands_on_devices(
        commands=list_cmd, deviceUuids=list_uuid, timeout=5
    )
    cmd_taskid = get_cmd_taskid.response.taskId
    print("taskId : ", cmd_taskid)
    result = wait_on_task(api, cmd_taskid)
    if result:
        print("This is the file to get {}", result)
        return result
    else:
        print("Task Failed")
        return None


def wait_on_task(api, taskid):
    """Check the taskid for progress
    Returm response on successful completion
    Raise exception if task is in error (isError == True)

    Args:
        api ([DNACAPI]): [dnacentersdk api object]
        taskid ([type]): [task id obtained after post to DNAC]

    """
    count = 0
    SLEEP = 1
    TIMEOUT = 10
    while True:
        get_progress = api.task.get_task_by_id(taskid)
        print(get_progress.response)
        status = get_progress.response.progress
        err_status = get_progress.response.isError
        print("status of taskid {} : {}".format(taskid, status))
        if err_status == True:
            print(" Task in error")
            return None
        if get_progress.response.endTime:
            print("Task completed")
            return status
        if count == TIMEOUT:
            return None
        count += 1
        time.sleep(SLEEP)


def my_new_script(api):
    """This function does the actual work intended
    1. makes a list of devices and their types
    2. get "show version " from routers and swtiches
    3. print hostname - software version to file


    Args:
        api ([type]): [description]
    """
    # get Device UUID
    dict_dev_uuid = get_device_list(api)
    CMD = ["show clock"]
    list_uuid = list(dict_dev_uuid.keys())
    cmd_out = get_command(api, CMD, list_uuid)
    print(cmd_out)


def main():
    """
    summary:
        The main function is called if this module is executed by as
            ./script_template.py <cmd line args>
        * Parses command line args
        * Calls my_function() with args from cmd line
        * can print module usage help
    """
    # parse command line args
    parser = ArgumentParser()
    # positinal args ( Mandatory args)
    # parser.add_argument(
    #    "arg1", type=str, help=" expecting a string type arg. Change code as you need"
    # )
    # setting up logging to DEBUG level
    parser.add_argument(
        "--verbose",
        "-v",
        action="store_true",
        help="set logging:  DEBUG ",
    )
    # set logging to DEBUG based on CMD line args
    cmd_args = parser.parse_args()
    if cmd_args.verbose:
        logger.setLevel(logging.DEBUG)
        # set level for all handles
        for hdler in logger.handlers:
            hdler.setLevel(logging.DEBUG)
        logger.debug("Loggig set to DEBUG")
        logger.warning("Logging set to DEBUG")

    # set env variables for dnacentersdk
    script_path = Path(__file__).resolve().parent
    env_var_file = script_path / "sdk_env_vars.yml"
    # import dnacentersdk after setting up the env variables
    set_env_var_dnac(env_var_file)
    from dnacentersdk import DNACenterAPI, ApiError

    api = DNACenterAPI()
    # call my_function with the required args:
    my_new_script(api)


# Check to see if scirpt run by itself or called as module
if __name__ == "__main__":
    main()

## ----- End of script --##'

[zhenningx]

@bigevilbeardI tried your code and it is same issue. I got the result of the file ID, but still the same question how to retrieve this file and save locally? I assume this file contains the result of the command?

This is the file to get {} {"fileId":"963f9ddd-e430-4bad-ad96-dda8918b95c3"}
{"fileId":"963f9ddd-e430-4bad-ad96-dda8918b95c3"}

The code prints out the fileID as above, but did not print out the result of the command run.

Same as the solution by the blog you mentioned which returns a cmd_output class but this is not the command outputs that the user needs.

How can I get the command outputs? Thanks.

[bigevilbeard · Accepted solution]

Hello @zhenningx i would suggest raising this as an issue on the code https://github.com/cisco-en-programmability/dnacentersdk

[zhenningx · Accepted solution]

I have submitted an issue: https://github.com/cisco-en-programmability/dnacentersdk/issues/50

Thanks.

[zhenningx]

This issue has been resolved: https://github.com/cisco-en-programmability/dnacentersdk/issues/50

Thanks for all the helps!