現場で使えるPython：ネットワーク機器の状態データ収集と可視化基盤連携
はじめに
システムエンジニアやインフラエンジニアの業務において、サーバーやクラウドインフラの監視、自動化は日常的に行われています。しかし、ネットワーク機器に関しては、CLIによる手動操作や、専用の監視ツールに依存しているケースも少なくありません。特に、開発やIaC（Infrastructure as Code）の文脈でインフラ全体をコードで管理・自動化している環境では、ネットワーク機器の状態もプログラムから効率的に取得し、活用したいというニーズがあります。
本記事では、Pythonを使ってネットワーク機器の現在の状態を示すデータ（インターフェース統計、CPU/メモリ使用率、環境情報など）を自動的に収集し、そのデータを時系列データベースのような外部システムに連携して可視化や分析に活用する方法を紹介します。ネットワーク機器に不慣れでも、Pythonのスキルを活かして、これらのデータを収集・整形し、既存の監視・可視化基盤に統合する実践的なアプローチを提供します。
なぜネットワーク機器の状態データを収集し、外部連携するのか
ネットワーク機器の状態データは、運用において非常に重要です。
- トラブルシューティング: 障害発生時に、特定時点または時系列での機器の状態を確認することで、原因特定の迅速化に役立ちます。
- キャパシティプランニング: インターフェースの使用率やリソース状況の推移を把握し、将来的な増設や設計変更の判断材料とします。
- 設定変更影響確認: 設定変更前後の状態を比較し、意図した通りに動作しているか、パフォーマンスに影響が出ていないかを確認します。
- 現状把握: 定期的に最新の状態を取得することで、インベントリ情報の補完や、設定ドリフト（あるべき設定からの乖離）の兆候を捉えるのに役立ちます。
これらのデータをCLIで手動取得するのは非効率であり、取得タイミングやフォーマットのばらつきといった問題があります。Pythonで自動化することで、定期的・網羅的にデータを収集し、構造化されたデータとして時系列データベースやストレージに保存することで、後続の処理や可視化が容易になります。
Pythonによるネットワーク機器からのデータ収集手法
ネットワーク機器から状態データを取得する方法はいくつかあります。現場でよく使われるのは、主にCLIコマンドの実行結果を取得する方法と、APIを利用する方法です。
1. CLIコマンド実行結果の取得
多くのネットワーク機器は、SSH経由でCLIコマンドを実行し、その出力を取得できます。この方法は汎用性が高く、様々なベンダーや機種で利用可能です。
PythonからSSH経由でコマンドを実行し、出力を取得するには、以下のようなライブラリが便利です。
- paramiko: 純粋なSSHv2プロトコル実装ライブラリ。低レベルな制御が可能ですが、コマンド実行結果のパースなどは別途実装が必要です。
- Netmiko: paramikoをバックエンドとして利用し、様々なネットワーク機器ベンダーのCLI操作を抽象化してくれるライブラリ。プロンプト判定や改ページ処理などを自動で行ってくれるため、非常に簡単にコマンド実行結果を取得できます。
Netmikoを使ったコマンド実行の基本的なコード例を示します。
from netmiko import ConnectHandler
import os
# 接続情報（環境変数や安全な設定ファイルから取得することを推奨）
device = {
'device_type': 'cisco_ios', # ベンダー/OSタイプを指定
'host': os.environ.get('NETWORK_DEVICE_HOST'),
'username': os.environ.get('NETWORK_DEVICE_USER'),
'password': os.environ.get('NETWORK_DEVICE_PASSWORD'),
'port': 22,
}
# 実行したいコマンド
commands = [
'show version',
'show ip interface brief',
'show environment all',
'show interface status',
]
# 接続とコマンド実行
try:
with ConnectHandler(**device) as net_connect:
print(f"Connected to {device['host']}")
output_data = {}
for cmd in commands:
print(f"Executing command: {cmd}")
output = net_connect.send_command(cmd)
output_data[cmd] = output
print(f"--- Output for '{cmd}' ---")
# print(output) # デバッグ表示
print("---------------------------")
# 取得したデータ（output_data）を処理する
print("\nData collection complete.")
# output_data ディクショナリには、コマンドとその実行結果が格納されています。
except Exception as e:
print(f"An error occurred: {e}")
# 注: 本コードは環境変数から認証情報を取得することを想定しています。
# 実際には、パスワードマネージャーやVaultなどを利用して安全に管理してください。
このコードでは、指定したコマンドの出力を output_data ディクショナリに格納しています。しかし、この時点での出力は単なるテキスト文字列です。このままでは分析や連携に不向きなため、次に示すデータ整形（パース）が必要になります。
2. 取得データのパースと構造化
CLIコマンドの出力は、人間が読むためには適していますが、プログラムで扱うには非構造化データであり扱いにくい形式です。これをJSONやPythonの辞書のような構造化データに変換するプロセスを「パース」と呼びます。
パースにはいくつかの方法があります。
- 正規表現: 特定のパターンにマッチさせて必要な情報を抽出します。シンプルですが、出力形式のわずかな変更に弱いという欠点があります。
- TextFSM / TTP: 定義ファイル（テンプレート）に基づいて、CLI出力を構造化データに変換するツールです。様々なベンダー/OSのテンプレートが公開されており、比較的容易に利用できます。
- NAPALM: 複数のベンダー機器に対して、共通のAPIを通じて構造化された情報を取得できるライブラリです。内部的にはNetmikoやNETCONFなどを利用しつつ、ベンダー差分を吸収してくれます。CLI出力のパースも内部的に行っています。
ここでは、Netmikoで取得したCLI出力をTTPを使ってパースする例を示します。
from netmiko import ConnectHandler
from ttp import ttp
import os
# 接続情報（前述の例と同様）
device = {
'device_type': 'cisco_ios',
'host': os.environ.get('NETWORK_DEVICE_HOST'),
'username': os.environ.get('NETWORK_DEVICE_USER'),
'password': os.environ.get('NETWORK_DEVICE_PASSWORD'),
'port': 22,
}
# 実行したいコマンド
command = 'show interface status'
# TTPテンプレート（例: show interface status 用の簡易テンプレート）
# このテンプレートは機器の実際の出力に合わせて記述する必要があります。
# 詳細なテンプレートはTTPのドキュメントやコミュニティを参照してください。
ttp_template = """
interface status:
{{ result | macro('column') }}
Port Name Status Vlan Duplex Speed Type
--------- ------------------ ------------ ---------- ------ ------ ----
{{ port }} {{ name | OR(' ') }} {{ status }} {{ vlan }} {{ duplex }} {{ speed }} {{ type }}
"""
# 接続とコマンド実行、パース
try:
with ConnectHandler(**device) as net_connect:
print(f"Connected to {device['host']}")
print(f"Executing command: {command}")
output = net_connect.send_command(command)
print(f"--- Raw Output for '{command}' ---")
print(output)
print("---------------------------")
# TTPパーサーインスタンスを作成
parser = ttp(data=output, template=ttp_template)
# データをパース
parser.parse()
# パース結果を取得
parsed_data = parser.result(structure="flat_list") # flat_list形式で取得
print("\n--- Parsed Data ---")
import json
print(json.dumps(parsed_data, indent=2))
print("---------------------")
except Exception as e:
print(f"An error occurred: {e}")
# 注: TTPテンプレートは対象機器の出力形式に厳密に合わせる必要があります。
# 上記テンプレートはあくまで例です。
この例では、show interface status コマンドの出力をTTPテンプレートを使ってパースし、リスト形式の構造化データとして取得しています。このように構造化されたデータは、後の処理（フィルタリング、集計、外部システムへの書き込みなど）が格段に容易になります。
3. APIの利用 (RESTConf, NETConf, ベンダー独自API)
最近のネットワーク機器は、CLIだけでなくAPI（Application Programming Interface）を提供しているものも増えています。RESTConfやNETConfは標準化されたネットワーク管理プロトコルであり、これらを利用することで構造化されたデータを直接取得したり、設定変更を行ったりできます。ベンダー独自のREST APIを提供している場合もあります。
APIを利用するメリットは、CLI出力のパースが不要であること、プログラムからの操作に適したデータ形式（JSONやXMLなど）で情報が得られることです。
PythonからAPIを利用するには、requests ライブラリなどが一般的です。RESTConfやNETConfに対応したライブラリ（例: ncclient for NETConf）もあります。
APIを利用したデータ収集の例は、機器やAPIの種類によって大きく異なるため、ここでは一般的なREST APIを利用する際の考慮事項に留めます。
- APIエンドポイントの特定
- 認証方法（Token認証、Basic認証など）の確認
- 取得したい情報に対応するAPIリソースの確認
- APIリクエスト（GETメソッドなど）の実行
- レスポンスデータの解析（JSONやXML）
APIが利用可能な場合は、CLIよりもAPIの利用を検討するのが現代的なアプローチと言えます。ただし、すべての情報がAPIで取得できるとは限らず、依然としてCLIが必要な場面も多いです。
収集したデータを時系列データベースに連携する
構造化されたデータを収集したら、次にこれをどのように活用するかが重要です。監視や可視化の目的で、時系列データを扱うのに適したデータベース（時系列データベース）に書き込むのが一般的なパターンです。
代表的な時系列データベースには、InfluxDB, Prometheus, OpenTSDB などがあります。これらのデータベースには、Pythonからデータを書き込むためのクライアントライブラリが提供されています。
ここでは、収集したインターフェースの統計データ（例: エラーカウント、廃棄パケット数など）をInfluxDBに書き込む場合のコード例を示します。InfluxDBのPythonクライアントライブラリ influxdb-client を使用します。
事前に influxdb-client ライブラリをインストールしておく必要があります。
pip install influxdb-client
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS
import os
import time
from datetime import datetime, timezone
# InfluxDB 接続情報（環境変数から取得することを推奨）
url = os.environ.get("INFLUXDB_URL")
token = os.environ.get("INFLUXDB_TOKEN")
org = os.environ.get("INFLUXDB_ORG")
bucket = os.environ.get("INFLUXDB_BUCKET")
# ネットワーク機器から取得・パースしたデータ（サンプル）
# 実際には前述の方法で動的に取得・パースしたデータを使用します
# ここでは、show interface status や show interface counters の結果を
# パースして得られたようなデータを想定しています。
# 例: [{'port': 'GigabitEthernet1/0/1', 'status': 'connected', ... , 'input_errors': 10, 'output_discards': 5}, ...]
parsed_interface_data = [
{'port': 'GigabitEthernet1/0/1', 'status': 'connected', 'vlan': '10', 'speed': '1000', 'input_errors': 10, 'output_discards': 5},
{'port': 'GigabitEthernet1/0/2', 'status': 'notconnect', 'vlan': '1', 'speed': 'auto', 'input_errors': 0, 'output_discards': 0},
# ... 他のインターフェースデータ
]
# InfluxDBクライアントの初期化
client = InfluxDBClient(url=url, token=token, org=org)
write_api = client.write_api(write_options=SYNCHRONOUS)
print(f"Connecting to InfluxDB at {url}, writing to bucket '{bucket}' in org '{org}'")
try:
points_to_write = []
current_time = datetime.now(timezone.utc) # タイムスタンプはUTCを推奨
for iface_data in parsed_interface_data:
port_name = iface_data.get('port')
if not port_name:
continue # ポート名がないデータはスキップ
# Pointオブジェクトを作成
# Measurement: データの種類（例: interface_metrics）
# Tags: データを識別・フィルタリングするためのキーバリューペア（例: hostname, port, status）
# Fields: 計測値（例: input_errors, output_discards, speed_mbps）
point = Point("interface_metrics") \
.tag("hostname", device.get('host', 'unknown_host')) \
.tag("port", port_name) \
.tag("status", iface_data.get('status', 'unknown')) \
.tag("vlan", iface_data.get('vlan', 'unknown')) \
.field("input_errors", int(iface_data.get('input_errors', 0))) \
.field("output_discards", int(iface_data.get('output_discards', 0))) \
.field("speed_mbps", int(iface_data.get('speed', '0').replace('auto', '0').replace('a-', '0'))) \
.time(current_time, WritePrecision.NS) # ナノ秒精度でタイムスタンプを設定
points_to_write.append(point)
# データをInfluxDBに書き込み
if points_to_write:
write_api.write(bucket=bucket, org=org, record=points_to_write)
print(f"Successfully wrote {len(points_to_write)} data points to InfluxDB.")
else:
print("No data points to write.")
except Exception as e:
print(f"An error occurred while writing to InfluxDB: {e}")
finally:
client.close()
print("InfluxDB client closed.")
# 注: このコードはサンプルデータを使用しており、実際には前述のNetmiko+TTPなどを使って
# ネットワーク機器から取得したデータをパースして `parsed_interface_data` を生成する必要があります。
# また、InfluxDBの接続情報、バケット名、組織名などは適切に設定してください。
# Fieldの値の型に注意し、適切に変換（例: int()）してください。
このスクリプトは、サンプルとして用意した構造化されたインターフェースデータを取得し、各インターフェースの情報（ポート名、ステータスなどをタグとして、エラー数や速度などをフィールドとして）をPointオブジェクトに変換し、InfluxDBに書き込んでいます。InfluxDBにデータが格納されれば、Grafanaなどの可視化ツールと連携して、グラフやダッシュボードを作成し、ネットワーク機器の状態を視覚的に把握できるようになります。
実践的な考慮事項
現場でこれらのスクリプトを利用する際には、いくつかの重要な考慮事項があります。
- 認証情報の安全な管理: ネットワーク機器やデータベースへの接続情報は、コード中に直接記述せず、環境変数、Vault、または専用のパスワードマネージャーなどを利用して安全に管理してください。
- エラーハンドリング: ネットワークの瞬断、機器側の認証失敗、コマンド実行失敗、パースエラー、データベースへの書き込み失敗など、様々なエラーが発生し得ます。適切な例外処理を実装し、エラー発生時にはログ出力や通知を行うようにしてください。
- 複数機器への対応: 大規模な環境では、多数のネットワーク機器から同時にデータを収集する必要があります。単一のスレッドで順次処理すると時間がかかりすぎるため、マルチスレッド、マルチプロセス、または非同期処理（asyncio）を利用して並列化を検討してください。ネットワーク自動化フレームワークのNornirは、並列処理や認証情報管理、インベントリ管理などの機能を提供しており、このような用途に適しています。
- 定期実行: 状態データの収集は、通常定期的に実行します。Linuxのcron、Windowsのタスクスケジューラ、またはJenkinsやGitLab CI/CDなどのCI/CDツールを利用して、スクリプトをスケジュール実行してください。
- 冪等性: データ収集自体は一般的に冪等な操作ですが、収集したデータを基に何らかのアクション（例: 設定変更）を行う場合は、その操作が冪等になるように設計することが重要です。
- 監視・アラートとの連携: 収集したデータを監視システムに取り込むことで、閾値を超えた場合にアラートを発生させるなど、能動的な運用が可能になります。
IaC/CI/CDパイプラインにおける位置づけ
Pythonによるネットワークデータ収集と可視化基盤連携は、インフラ自動化全体のライフサイクルの中で重要な役割を果たします。
- デプロイ後検証: IaCツール（Ansible, Terraformなど）でネットワーク設定変更を行った後、Pythonスクリプトを実行して機器の現在の状態（インターフェースの状態、ルーティングテーブル、ACL設定など）を収集し、期待される状態と比較することで、変更が正しく反映されたか、意図しない影響が出ていないかを確認する自動テストとして組み込むことができます。
- 継続的な状態監視: 定期的に機器の状態を収集し、可視化基盤でトレンドを監視したり、設定管理システム（Ansible Tower/AWXなど）に保管されている設定情報やGolden Configと比較して設定ドリフトを検知したりする仕組みを構築できます。
- パフォーマンス監視の強化: インターフェース統計やリソース使用率の時系列データを収集・可視化することで、ネットワークパフォーマンスの変化を継続的に追跡し、問題の早期発見やキャパシティプランニングに役立てます。
これらの活動をCI/CDパイプラインに組み込むことで、ネットワークの変更管理や運用効率を大幅に向上させることが可能になります。
まとめ
本記事では、Pythonを使ってネットワーク機器から状態データを収集し、パースして構造化し、時系列データベースに連携する一連の手順とコード例を紹介しました。NetmikoによるCLIコマンド実行、TTPによる出力パース、influxdb-clientによるInfluxDBへの書き込みを取り上げましたが、これらはあくまで一例です。対象機器や目的に応じて、NAPALMや各種APIを利用したり、Prometheusや他のデータストアに連携したりすることも可能です。
Pythonスキルを活かすことで、ネットワーク機器のブラックボックス感を減らし、他のインフラコンポーネントと同様にコードによる管理・自動化の範囲に組み込むことができます。収集したデータを活用することで、ネットワーク運用はよりデータに基づいた、効率的かつプロアクティブなものへと進化します。
今後、収集したデータを元にした分析方法や、Nornirを使った大規模環境での並列処理、あるいは特定のベンダーAPIを使った高度な自動化などについても触れていきたいと思います。