Freeで試せるCisco DNAC APIを使ってトラブルシューティングに役立つツールを作ってみよう

#まずは
(^ω^)/ﾟ･:【 HAPPY BIRTHDAY to 自分 】:･ﾟ＼(^ω^)
なんだかんだで、print(0x10*3)歳なんですよ。。。これがやりたいだけで、今日という、この日を選んでしまいました。

#はじめに

この記事はシスコの同志による Cisco Systems Japan Advent Calendar 2021（一枚目） の 24 日目として投稿しています。

2021年版(1枚目): https://qiita.com/advent-calendar/2021/cisco
2021年版(2枚目): https://qiita.com/advent-calendar/2021/cisco2

2017年版: https://qiita.com/advent-calendar/2017/cisco
2018年版: https://qiita.com/advent-calendar/2018/cisco
2019年版: https://qiita.com/advent-calendar/2019/cisco
2020年版: https://qiita.com/advent-calendar/2020/cisco
2020年版(1枚目): https://qiita.com/advent-calendar/2020/cisco
2020年版(2枚目): https://qiita.com/advent-calendar/2020/cisco2

#何がやりたい？
Cisco DNA CenterのAPIを使ってみよう。というだけです。

そうは言っても、何が大変かというと、「どうやって、APIを試すための環境を用意するの？」の一言に尽きると思います。

大丈夫です。下記の通り、DNACを自由に試すことができる環境(サンドボックス)が用意されています。
https://sandboxdnac.cisco.com/
Username : devnetuser
Password : Cisco123!

一般的にAPIを試すとなると、何かのAPIが動作するのを確認してそこで終了満足しがちですが、折角なので、APIを使ったトラブルシューティングに役立ちそうなツールを作ってみよう。
というのがゴールになります。

それでは、最初にツールを実行した結果を説明しますと、ここで紹介するサンプルコードを実行$ python3 show_diff_stats.pyすると、下記のグラフが表示されます。
「なんやねんこれこのグラフは何？」という感じですが、サンドボックスで用意されているスイッチのインタフェースカウンタで増加している値を表示しています。

##トラブルシューティングで必要なのは？
まず、その機器で何が起こっているのかを知ることだと考えています。show loggingで確認できるようなメッセージも重要ですし、**「どのようなカウンタの値が増加しているのか。」**というのも非常に重要だと考えています。

例えば、「この機器でパケットのドロップが発生しているのでは。」という問題があったとします。基本的に、ネットワーク機器は、理由もなくパケットをドロップすることはありません。パケットを破棄する際には、何らかのカウンタが増加します。

どのコマンドでドロップや問題となる値を確認するのか、ということも重要ですが、もし、「今、問題が発生している。」のであれば、その事象の原因を示すカウンタも増加しているはずです。

そうなると、繰り返し、同じログを取得して、どこに問題がありそうなのかを比較して確認するのが面倒くさい大変です。
そこで、そのカウンタの増加をグラフで表示してみよう。というのがここで作成するツールです。

#ツールについて
このツールでは、大きく分けて、3つの部分から成り立っています。

1. Command Runner ==> DNACが窓口となり、DNACが任意の機器にアクセスして、コマンドを実行して、その結果を取得する。
2. Genie ==> コマンドの実行結果を解析して、キーと値のJSON形式に変換する。
3. Matplotlib ==> グラフを描画する。

show stasを実行して、初回の値を保存します。

##Command Runner
このツールはDNAC APIからでも利用できますが、DNACにも標準のツールとして用意されています。ブラウザからDNACにログインするだけで、GUI画面上で簡単に実行することができますので、是非、こちらも試してみてください。（下記は、show interfaceの実行結果です。）

API経由でコマンドを実行すると言っても、DNAC APIにコマンドを送るだけで結果が返ってくる。という程、簡単ではありません。
コマンドを実行して、その結果を取得するまでには、下記の手順を踏む必要があります。

1. Device idを指定して、コマンドを実行 => Taskが生成され、Task idが得られる。
2. Taskが完了 => 結果として、File idが得られる。
3. File idを指定して、コマンドの結果をダウンロードする。

###Device idについて一言
DNACから見て、どの機器にアクセスするのかを指示するには、ホスト名ではなく、Device idが必要です。このツールでは、基本的に同一のホスト名を複数の機器に割り当てていないという前提で、ホスト名に一致するDevice idを取得しています。

##Genie
Genieは、pyATSの一部となります。
pyATSと言えば、この方でしょう。私もこのサイトを参考にさせていただきました。
https://ccieojisan.net/post-2221/

##Matplotlib
こちらの詳細についても、他で説明されているサイトをご参照してください。下記のようなサンプルを探して、色々とお試しいただくのが良いかと思います。

sample.py

import numpy as np
import matplotlib.pyplot as plt
 
left = np.array([1, 2, 3, 4, 5])
height = np.array([100, 200, 300, 400, 500])
plt.bar(left, height)
plt.show()

#ツールの実行
動作環境としては、MacとUbuntu上で動作することを確認しています。Ubuntuの場合は、エラーが出ましたので、対処方法などの詳細につきましては、GitHubの方をご参照ください。

##ダウンロード&実行
GitHubから、下記のコマンドを実行することでダウンロードすることができます。

git clone https://github.com/eiuemura/cisco-dnac-show-diff-stats

cd cisco-dnac-show-diff-stats
pip install -r requirements.txt
python3 show_diff_stats.py

##ソースコード

show_diff_stats.py

#! /usr/bin/env python3
# -*- coding: utf-8 -*-

import sys
import json
import requests
import urllib3
import time
from requests.auth import HTTPBasicAuth
from pprint import pprint
from genie.conf.base import Device
import numpy as np
import matplotlib.pyplot as plt

# https://sandboxdnac.cisco.com/
DNAC_HOST = "sandboxdnac.cisco.com"
DNAC_USER = "devnetuser"
DNAC_PASS = "Cisco123!"
DNAC_DEVICE_HOSTNAME = "leaf1.abc.inc"
DNAC_TIMEOUT = 0

# Silence the insecure warning due to SSL Certificate
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class dnacApiClass:
    def __init__(self):
        pass

    def auth(self, dnacHost, dnacUser, dnacPass):
        # --------------------------------------------------------------------------------
        # Authentication API
        # https://developer.cisco.com/docs/dna-center/#!cisco-dna-center-2-1-2-x-api-api-authentication-authentication-api
        # --------------------------------------------------------------------------------
        url = "https://"+dnacHost+"/api/system/v1/auth/token"
        headers = {
            'Content-Type': "application/json",
            }
        payload = ""
        response = requests.post(url, auth = HTTPBasicAuth(dnacUser, dnacPass), headers = headers, data = payload, verify = False)
        if response.status_code == 200:
            dnacToken = json.loads(response.text)
            dnacToken = dnacToken["Token"]
            return(dnacToken)
        else:
            print(">>> Error Cisco DNA-C 'Auth' Failed <<<")
            print("HTTP response status code : {}".format(response.status_code))
            return False

    def get_all_device_id(self, dnacToken):
        # --------------------------------------------------------------------------------
        # Get Device List
        # https://developer.cisco.com/docs/dna-center/#!cisco-dna-center-2-1-2-x-api-api-devices-get-device-list
        # --------------------------------------------------------------------------------
        url = "https://{}/api/v1/network-device".format(DNAC_HOST)
        headers = {
            'x-auth-token': dnacToken,
            'Content-Type': "application/json",
            'Accept': "application/json"
            }
        all_device_response = requests.get(url, headers = headers, verify = False)
        device_list = all_device_response.json()
        for device in device_list['response']:
            print("DEBUG : Hostname = [{}] , id = [{}]".format(device['hostname'], device['id']))

    def get_device_id_from_hostname(self, device_name, dnacToken):
        # --------------------------------------------------------------------------------
        # Get Device List
        # https://developer.cisco.com/docs/dna-center/#!cisco-dna-center-2-1-2-x-api-api-devices-get-device-list
        # --------------------------------------------------------------------------------
        url = "https://{}/api/v1/network-device".format(DNAC_HOST)
        headers = {
            'x-auth-token': dnacToken,
            'Content-Type': "application/json",
            'Accept': "application/json"
            }
        all_device_response = requests.get(url, headers = headers, verify = False)
        device_list = all_device_response.json()
        for device in device_list['response']:
            # For the purpose of debugging.
            # print("DEBUG : Hostname = [{}] , id = [{}]").format((device)['hostname'], (device)['id'])
            # print(device)
            if device['hostname'] == device_name:
                return device['id']
        print(">>> Error Cisco DNA-C 'Hostname was not found.' <<<")
        return False

    def run_command_and_get_task_id(self, deviceId, deviceCmd, dnacToken):
        # --------------------------------------------------------------------------------
        # Run Read Only Commands On Devices To Get Their Real Time Configuration
        # https://developer.cisco.com/docs/dna-center/#!cisco-dna-center-2-1-2-x-api-api-command-runner-run-read-only-commands-on-devices-to-get-their-real-time-configuration
        # --------------------------------------------------------------------------------
        url = "https://{}/api/v1/network-device-poller/cli/read-request".format(DNAC_HOST)
        headers = {
            'x-auth-token': dnacToken,
            'Content-Type': "application/json",
            'Accept': "application/json"
            }
        payload = {
            "commands": [deviceCmd],
            "deviceUuids": [deviceId],
            "timeout": DNAC_TIMEOUT
            }
        response = requests.post(url, headers = headers, data = json.dumps(payload), verify = False)
        response_json = response.json()
        try:
            task_id = response_json['response']['taskId']
            return task_id
        except:
            print(">>> Error Cisco DNA-C 'Run Read Only Commands' Failed <<<")
            return False

    def get_file_id(self, taskId, dnacToken):
        # --------------------------------------------------------------------------------
        # Get Tasks
        # https://developer.cisco.com/docs/dna-center/#!cisco-dna-center-2-1-2-x-api-api-task-get-tasks
        # --------------------------------------------------------------------------------
        url = "https://{}/api/v1/task/{}".format(DNAC_HOST, taskId)
        headers = {
            'x-auth-token': dnacToken,
            'Content-Type': "application/json",
            'Accept': "application/json"
            }
        while True:
            try:
                task_response = requests.get(url, headers = headers, verify = False)
                task_json = task_response.json()
                file_info = json.loads(task_json['response']['progress'])
                file_id = file_info['fileId']
                return file_id
            except:
                if task_json['response']['isError'] is True:
                    print(">>> Error Cisco DNA-C 'Get Task' Failed <<<")
                    return False
                pass
            finally:
                time.sleep(1)

    def download_file(self, fileId, deviceCmd, dnacToken):
        # --------------------------------------------------------------------------------
        # Download A File By File Id
        # https://developer.cisco.com/docs/dna-center/#!cisco-dna-center-2-1-2-x-api-api-file-download-a-file-by-file-id
        # --------------------------------------------------------------------------------
        url_file = "https://{}/api/v1/file/{}".format(DNAC_HOST, fileId)
        headers = {
            'x-auth-token': dnacToken,
            'Content-Type': "application/json",
            'Accept': "application/json"
            }
        response = requests.get(url_file, headers = headers, verify = False, stream = True)
        response_json = response.json()
        try:
            command_response = response_json[0]['commandResponses']['SUCCESS'][deviceCmd]
        except:
            command_response = response_json[0]['commandResponses']['FAILURE'][deviceCmd]
            print(">>> Error Cisco DNA-C 'Download File' Failed <<<")
            print(command_response)
            return False
        return command_response


if __name__ == '__main__':
    dnacApi = dnacApiClass()

    # >>> DNACにアクセスするためのトークンを取得 <<<
    dnacToken = dnacApi.auth(DNAC_HOST, DNAC_USER, DNAC_PASS)
    if dnacToken is False:
        quit()

    # >>> 全てのホスト名とデバイスIDを表示(デバッグ用) <<<
    dnacApi.get_all_device_id(dnacToken)

    # >>> ホスト名からデバイスIDを検索して取得 <<<
    deviceId = dnacApi.get_device_id_from_hostname(DNAC_DEVICE_HOSTNAME, dnacToken)
    if deviceId is False:
        quit()

    # >>> show interfaces コマンドを実行して結果を取得 <<<
    DNAC_CMD =  "show interfaces"
    taskId = dnacApi.run_command_and_get_task_id(deviceId, DNAC_CMD, dnacToken)
    if taskId is False:
        quit()
    fileId = dnacApi.get_file_id(taskId, dnacToken)
    if fileId is False:
        quit()
    result_cmd = dnacApi.download_file(fileId, DNAC_CMD, dnacToken)
    if result_cmd is False:
        quit()
    print(result_cmd)

    # >>> Genieでコマンドの出力結果をパースしてJSONに変換 <<<
    try:
        device = Device(DNAC_DEVICE_HOSTNAME, os='iosxe')
        device.custom.setdefault("abstraction", {})["order"] = ["os"]
        result_json = device.parse(DNAC_CMD, output=result_cmd)
    except:
        print(">>> Error Genie parser Failed <<<")
        print(sys.exc_info())
        quit()
    pprint(result_json)

    # >>> show interfacesの結果から、UP/UPのインタフェースだけを抽出してリストに追加 <<<
    up_interfaces = []
    for temp_interface in result_json:
        if result_json[temp_interface]['line_protocol'] == 'up' and result_json[temp_interface]['oper_status'] == 'up':
            up_interfaces.append(temp_interface)

    # >>> show interface stats コマンドを実行して結果を取得 <<<
    DNAC_CMD =  "show interface stats"
    taskId = dnacApi.run_command_and_get_task_id(deviceId, DNAC_CMD, dnacToken)
    if taskId is False:
        quit()
    fileId = dnacApi.get_file_id(taskId, dnacToken)
    if fileId is False:
        quit()
    result_cmd = dnacApi.download_file(fileId, DNAC_CMD, dnacToken)
    if result_cmd is False:
        quit()
    print(result_cmd)

    # >>> Genieでコマンドの出力結果をパースしてJSONに変換 <<<
    try:
        device = Device(DNAC_DEVICE_HOSTNAME, os='iosxe')
        device.custom.setdefault("abstraction", {})["order"] = ["os"]
        result_json = device.parse(DNAC_CMD, output=result_cmd)
    except:
        print(">>> Error Genie parser Failed <<<")
        print(sys.exc_info())
        quit()
    pprint(result_json)

    # >>> 基準(初回の値)となるstats情報を取得して保存 <<<
    base_counter = []
    for temp_interface in result_json:
        if temp_interface in up_interfaces:
            temp_list = []
            temp_list.append(temp_interface)
            temp_list.append(result_json[temp_interface]['switching_path']['total']['pkts_in'])
            temp_list.append(result_json[temp_interface]['switching_path']['total']['pkts_out'])
            base_counter.append(temp_list)

    # >>> Matplotlibで描画領域としてのFigureオブジェクトを作成 <<<
    fig = plt.figure(tight_layout=True)

    # >>> UPしているインタフェースの数だけ、サブプロットを作成 <<<
    axes = fig.subplots(len(up_interfaces),1)
    
    # >>> 各サブプロットにタイトルとして、インタフェース名を指定 <<<
    for i in range(0, len(up_interfaces)):
        axes[i].set_title(up_interfaces[i])
 
    # >>> グラフを30回更新して終了 <<<
    for i_loop in range(0,30):

        # >>> show interface stats コマンドを実行して結果を取得 <<<
        DNAC_CMD =  "show interface stats"
        taskId = dnacApi.run_command_and_get_task_id(deviceId, DNAC_CMD, dnacToken)
        if taskId is False:
            quit()
        fileId = dnacApi.get_file_id(taskId, dnacToken)
        if fileId is False:
            quit()
        result_cmd = dnacApi.download_file(fileId, DNAC_CMD, dnacToken)
        if result_cmd is False:
            quit()
        print(result_cmd)

        # >>> Genieでコマンドの出力結果をパースしてJSONに変換 <<<
        try:
            device = Device(DNAC_DEVICE_HOSTNAME, os='iosxe')
            device.custom.setdefault("abstraction", {})["order"] = ["os"]
            result_json = device.parse(DNAC_CMD, output=result_cmd)
        except:
            print(">>> Error Genie parser Failed <<<")
            print(sys.exc_info())
            quit()
        pprint(result_json)

        # >>> 最新のstats情報を取得して保存 <<<
        latest_counter = []
        for temp_interface in result_json:
            if temp_interface in up_interfaces:
                temp_list = []
                temp_list.append(temp_interface)
                temp_list.append(result_json[temp_interface]['switching_path']['total']['pkts_in'])
                temp_list.append(result_json[temp_interface]['switching_path']['total']['pkts_out'])
                latest_counter.append(temp_list)

        # >>> 基準のstatsと最新のstats情報を比較してグラフを描画 <<<
        for i in range(0, len(up_interfaces)):
            diff_total_in = int(latest_counter[i][1]) - int(base_counter[i][1])
            diff_total_out = int(latest_counter[i][2]) - int(base_counter[i][2])
            height = np.array([diff_total_out, diff_total_in])
            height_max = max([diff_total_out, diff_total_in])

            # >>> 棒グラフの幅が100を超える場合は、その値を棒グラフ幅の最大値として指定 <<<
            if height_max >= 100:
                axes[i].set_xlim(0, height_max)
            else:
                # >>> 棒グラフの幅が100を超えるない場合は、100を棒グラフ幅の最大値として指定 <<<
                axes[i].set_xlim(0, 100)
            x_bar = np.array(["Total Pkts Out", "Total Pkts In"])
            axes[i].barh(x_bar, height, color=['#aa4c8f','#1e50a2'], edgecolor=['r','b'])

        # >>> グラフを表示して、指定された時間を待つ <<<
        plt.pause(0.1)

    # >>> 最後にグラフを表示した状態で停止するため、スクリプトを終了する場合はグラフウィンドウを閉じる <<<
    plt.show()

#おわりに
本当は、show interfaceなどで表示されるエラーの差分をグラフで表示したかったのですが、サンドボックスの環境だと、エラーカウンタが増加していませんので、グラフで表示するための値も増加しなくなります。そのため、今回のサンプルでは、stats情報の差分を棒グラフで表示するようにしています。

また、Command Runnerの部分だけでも、ログをまとめて取得するようなスクリプトは簡単に作れると思いますので、色々とご活用いただけますと幸いです。

#免責事項
本サイトおよび対応するコメントにおいて表明される意見は、投稿者本人の個人的意見であり、シスコの意見ではありません。本サイトの内容は、情報の提供のみを目的として掲載されており、シスコや他の関係者による推奨や表明を目的としたものではありません。各利用者は、本Webサイトへの掲載により、投稿、リンクその他の方法でアップロードした全ての情報の内容に対して全責任を負い、本Web サイトの利用に関するあらゆる責任からシスコを免責することに同意したものとします。