実践ネット自動化スクリプト集 - 実践Python：ネットワーク機器の健全性を自動診断するヘルスチェック手法

実践Python：ネットワーク機器の健全性を自動診断するヘルスチェック手法

Tags: Python, ネットワーク自動化, ヘルスチェック, 監視, 状態監視

はじめに

システムエンジニアやインフラエンジニアの皆様にとって、日々運用するネットワーク機器の状態把握は重要なタスクです。機器が正常に稼働しているか、パフォーマンスに問題はないかなどを継続的に確認することは、障害予防や早期復旧に不可欠です。しかし、機器の増加や構成の複雑化に伴い、手動での確認作業は大きな負担となります。

この記事では、Pythonのスキルを活かしてネットワーク機器のヘルスチェックを自動化する方法に焦点を当てます。ネットワーク機器そのものに対する直接的な知識は限定的であっても、Pythonによる開発や自動化の経験をお持ちであれば、効率的にヘルスチェックの仕組みを構築できます。CLIコマンド実行やAPI連携といった具体的な手法とコード例を通して、実践的なヘルスチェック自動化スクリプトの実装方法を解説します。

ネットワーク機器におけるヘルスチェックの重要性

ネットワーク機器におけるヘルスチェックとは、機器の稼働状態や健全性を定期的に、あるいは必要に応じて確認するプロセスです。単に機器が疎通可能であるか（Ping応答があるか）だけでなく、CPU使用率、メモリ使用率、インターフェースの稼働状態、エラーカウンター、ルーティングテーブルの状態、特定プロセスの稼働状況など、より詳細な情報を取得・確認し、異常の兆候を早期に発見することが目的です。

ヘルスチェックを自動化することで、以下のようなメリットが得られます。

作業効率の向上: 手動での確認作業から解放され、より重要な業務に集中できます。
確認漏れの防止: 定期的な自動実行により、確認漏れを防ぎます。
障害の早期発見: 異常の兆候を素早く検知し、大規模な障害への発展を防ぎます。
状態の定量化と可視化: 取得したデータを蓄積・分析することで、機器やネットワーク全体の傾向把握やパフォーマンス監視にも繋がります。
属人化の解消: スクリプトとして共有することで、特定の担当者しかできない作業をなくします。

Pythonによるヘルスチェック自動化の基本アプローチ

Pythonを使ってネットワーク機器のヘルスチェックを自動化する主なアプローチは、以下の二つです。

CLI（コマンドラインインターフェース）ベースのアプローチ:
- SSHなどのプロトコル経由で機器にログインし、ヘルスチェックに関連するコマンドを実行します（例: show version, show cpu usage, show interface statusなど）。
- コマンド実行結果はテキスト形式で返されるため、Pythonでこれをパース（解析）し、必要な情報を抽出・判定します。
- paramikoやnetmikoのようなライブラリがこのアプローチに利用されます。netmikoは様々なベンダーのCLIコマンド実行に対応しており、特に便利です。
API（アプリケーションプログラミングインターフェース）ベースのアプローチ:
- 機器がREST API、NETConf、RESTConfなどのAPIを提供している場合、これらを直接呼び出して構造化されたデータ（JSON, XMLなど）を取得します。
- APIからの応答は最初から構造化されているため、CLI出力のパースに比べてデータの取り扱いが容易です。
- requests（REST API用）、ncclient（NETConf用）などのライブラリが利用されます。

どちらのアプローチを選択するかは、対象機器が提供する機能や自動化の目的によって異なります。比較的新しい機器や、状態を構造化データで取得したい場合にはAPIが適していますが、多くのレガシーな機器ではCLIが唯一の自動化インターフェースとなるため、CLIベースのアプローチも依然として重要です。

CLIベースのヘルスチェック実装例 (Netmikoを使用)

CLIベースのアプローチでは、netmikoライブラリが非常に強力です。様々なネットワークベンダーのCLIに対応しており、機器への接続やコマンド実行を容易に行えます。

ここでは、Netmikoを使ってCisco IOS XE機器からCPU使用率とメモリ使用率を取得し、特定の閾値を超えていないかチェックする簡単なスクリプト例を示します。

まず、netmikoをインストールします。

pip install netmiko

以下のPythonコードは、指定した機器に接続し、show processes cpu sortedおよびshow processes memory sortedコマンドを実行して、CPUとメモリの使用率を取得・判定する例です。

from netmiko import ConnectHandler
import re

# 接続情報（実際の環境に合わせて変更してください）
device = {
    'device_type': 'cisco_ios', # 機器タイプ
    'host': 'your_device_ip_or_hostname',
    'username': 'your_username',
    'password': 'your_password',
    # 'secret': 'your_enable_password', # enableモードに移行する場合
    'port': 22, # デフォルトはSSHの22
}

# 判定閾値
CPU_THRESHOLD = 70 # %
MEMORY_THRESHOLD = 70 # %

def check_health(device_info):
    """
    ネットワーク機器のCPUとメモリ使用率をチェックする関数
    """
    print(f"--- {device_info['host']} のヘルスチェックを開始 ---")
    try:
        # 機器に接続
        with ConnectHandler(**device_info) as net_connect:
            # enableモードに移行（必要な場合）
            # net_connect.enable()

            # CPU使用率の取得と判定
            cpu_output = net_connect.send_command("show processes cpu sorted | include CPU utilization")
            print(f"CLI Output (CPU): {cpu_output.strip()}")
            # 例: "CPU utilization for five seconds: 5%/1%; one minute: 7%; five minutes: 10%"
            # 正規表現で5秒間のCPU使用率を抽出
            cpu_match = re.search(r"five seconds: (\d+)%", cpu_output)
            if cpu_match:
                cpu_usage = int(cpu_match.group(1))
                print(f"Current CPU Usage (5s): {cpu_usage}%")
                if cpu_usage > CPU_THRESHOLD:
                    print(f"[警告] CPU使用率が閾値({CPU_THRESHOLD}%)を超えています！")
                    # ここで通知処理などを追加
                else:
                    print("CPU使用率は正常です。")
            else:
                 print("[エラー] CPU使用率をCLI出力から抽出できませんでした。")


            # メモリ使用率の取得と判定
            # IOSのshow processes memoryは複雑なので、ここでは別のコマンド例 (例: show memory summary) を想定
            # ベンダーやOSによってコマンドと出力形式は異なります。
            # Cisco IOS XEの例: show memory summary
            # Output Example:
            # Total: 4194304, Used: 1234567, Free: 2959737
            memory_output = net_connect.send_command("show memory summary | include ^Total:|^Used:|^Free:")
            print(f"CLI Output (Memory): \n{memory_output.strip()}")

            total_mem_match = re.search(r"^Total: (\d+)", memory_output, re.MULTILINE)
            used_mem_match = re.search(r"^Used: (\d+)", memory_output, re.MULTILINE)

            if total_mem_match and used_mem_match:
                total_mem = int(total_mem_match.group(1))
                used_mem = int(used_mem_match.group(1))
                if total_mem > 0:
                    memory_usage = (used_mem / total_mem) * 100
                    print(f"Current Memory Usage: {memory_usage:.2f}%")
                    if memory_usage > MEMORY_THRESHOLD:
                        print(f"[警告] メモリ使用率が閾値({MEMORY_THRESHOLD}%)を超えています！")
                        # ここで通知処理などを追加
                    else:
                         print("メモリ使用率は正常です。")
                else:
                     print("[エラー] 総メモリ容量が0です。")
            else:
                 print("[エラー] メモリ使用率をCLI出力から抽出できませんでした。コマンド出力形式を確認してください。")

        print(f"--- {device_info['host']} のヘルスチェック完了 ---")

    except Exception as e:
        print(f"[致命的なエラー] {device_info['host']} への接続またはコマンド実行中にエラーが発生しました: {e}")
        # エラー発生時の通知やログ記録

# スクリプト実行
if __name__ == "__main__":
    # 複数の機器を対象とする場合はリストにしてループします
    # devices = [...]
    check_health(device)

コードのポイント:

ConnectHandlerクラスで機器への接続情報を指定します。device_typeはベンダーやOSに合わせて適切に設定します。
send_command()メソッドでCLIコマンドを実行します。
コマンドの実行結果は文字列として返されるため、Pythonの正規表現(reモジュール)などを使って必要な数値を抽出します。CLI出力の形式は機器やOSバージョンによって異なるため、実際の出力に合わせて正規表現やパース処理を調整する必要があります。
抽出した数値を定義した閾値と比較し、異常があればメッセージを表示します。
try...exceptブロックを使って、接続エラーやコマンド実行エラーが発生した場合の例外処理を行っています。これは現場で利用する上で非常に重要です。

CLI出力のパースはネットワーク自動化において最も手間のかかる部分の一つです。複雑なCLI出力のパースには、TextFSMやGenieといったライブラリの利用も検討すると良いでしょう。これらは事前に定義されたテンプレートやパーサーを使って、CLI出力を構造化データに変換してくれます。

APIベースのヘルスチェック実装例 (REST APIを使用)

多くのモダンなネットワーク機器は、状態情報の取得や設定変更のためのAPIを提供しています。APIを利用すると、CLI出力のパースが不要になり、より構造化されたデータを直接扱えるため、スクリプト開発の効率が向上します。ここでは、Pythonのrequestsライブラリを使って、REST APIを提供する機器のヘルスチェックを行う例を示します。具体的なAPIエンドポイントや認証方法はベンダーや機器によって大きく異なりますので、以下のコードは概念的な例として捉えてください。

import requests
import json

# API接続情報（実際の環境に合わせて変更してください）
# 例: Cisco NX-OSvのREST API (NX-API) を想定
api_info = {
    'host': 'your_device_ip_or_hostname',
    'port': 80, # HTTPSの場合は443
    'username': 'your_username',
    'password': 'your_password',
    'use_https': False, # HTTPSを使用するか
    'verify_ssl': False # SSL証明書の検証を行うか (検証は推奨)
}

# ヘルスチェック対象のAPIエンドポイントとチェック内容
# 例: show version コマンドのJSON RPC APIエンドポイント
# Cisco NX-APIでは、POSTリクエストでコマンドと形式を指定することが多い
# ここでは架空のGETエンドポイント例として記述します
HEALTHCHECK_ENDPOINTS = {
    'cpu_status': '/api/v1/monitor/cpu', # CPU使用率を取得するAPI (例)
    'memory_status': '/api/v1/monitor/memory' # メモリ使用率を取得するAPI (例)
    # 実際にはベンダーのAPIドキュメントを参照してください
}

# 判定閾値
CPU_THRESHOLD = 70 # %
MEMORY_THRESHOLD = 70 # %

def check_api_health(api_config):
    """
    ネットワーク機器のAPIを使ってヘルスチェックを行う関数
    """
    protocol = "https" if api_config['use_https'] else "http"
    base_url = f"{protocol}://{api_config['host']}:{api_config['port']}"
    auth = (api_config['username'], api_config['password'])

    print(f"--- {api_config['host']} のAPIヘルスチェックを開始 ---")

    try:
        # CPU使用率のチェック
        cpu_url = f"{base_url}{HEALTHCHECK_ENDPOINTS['cpu_status']}"
        print(f"API Request (CPU): GET {cpu_url}")
        response = requests.get(
            cpu_url,
            auth=auth,
            verify=api_config['verify_ssl'],
            timeout=10 # タイムアウトを設定
        )
        response.raise_for_status() # HTTPステータスコードが200番台以外の場合は例外発生

        cpu_data = response.json()
        print(f"API Response (CPU): {json.dumps(cpu_data, indent=2)}")

        # JSONデータからCPU使用率を抽出・判定 (JSON構造はAPIに依存します)
        # 例: {'status': 'ok', 'usage_percent': 55} というJSONを想定
        if 'usage_percent' in cpu_data:
            cpu_usage = cpu_data['usage_percent']
            print(f"Current CPU Usage: {cpu_usage}%")
            if cpu_usage > CPU_THRESHOLD:
                print(f"[警告] CPU使用率が閾値({CPU_THRESHOLD}%)を超えています！")
                # 通知処理など
            else:
                print("CPU使用率は正常です。")
        else:
             print("[エラー] API応答からCPU使用率を抽出できませんでした。応答構造を確認してください。")

        # メモリ使用率のチェック
        memory_url = f"{base_url}{HEALTHCHECK_ENDPOINTS['memory_status']}"
        print(f"API Request (Memory): GET {memory_url}")
        response = requests.get(
            memory_url,
            auth=auth,
            verify=api_config['verify_ssl'],
            timeout=10
        )
        response.raise_for_status()

        memory_data = response.json()
        print(f"API Response (Memory): {json.dumps(memory_data, indent=2)}")

        # JSONデータからメモリ使用率を抽出・判定 (JSON構造はAPIに依存します)
        # 例: {'status': 'ok', 'memory': {'total': 1000, 'used': 600}} というJSONを想定
        if 'memory' in memory_data and 'total' in memory_data['memory'] and 'used' in memory_data['memory']:
             total_mem = memory_data['memory']['total']
             used_mem = memory_data['memory']['used']
             if total_mem > 0:
                 memory_usage = (used_mem / total_mem) * 100
                 print(f"Current Memory Usage: {memory_usage:.2f}%")
                 if memory_usage > MEMORY_THRESHOLD:
                    print(f"[警告] メモリ使用率が閾値({MEMORY_THRESHOLD}%)を超えています！")
                    # 通知処理など
                 else:
                    print("メモリ使用率は正常です。")
             else:
                  print("[エラー] 総メモリ容量が0です。")
        else:
            print("[エラー] API応答からメモリ使用率を抽出できませんでした。応答構造を確認してください。")

        print(f"--- {api_config['host']} のAPIヘルスチェック完了 ---")

    except requests.exceptions.RequestException as e:
        print(f"[致命的なエラー] {api_config['host']} へのAPIリクエスト中にエラーが発生しました: {e}")
        # エラー発生時の通知やログ記録
    except json.JSONDecodeError:
        print(f"[致命的なエラー] {api_config['host']} からの応答がJSON形式ではありませんでした。")
    except Exception as e:
        print(f"[致命的なエラー] 予期しないエラーが発生しました: {e}")

# スクリプト実行
if __name__ == "__main__":
    check_api_health(api_info)

コードのポイント:

requestsライブラリを使ってHTTP(S)リクエストを送信します。
認証情報やSSL証明書の検証設定は、APIやセキュリティポリシーに合わせて適切に設定します。
APIからの応答は通常JSON形式で返されるため、.json()メソッドでPythonの辞書やリストに変換できます。
取得した構造化データから必要な情報を簡単に抽出できます。CLIパースのような複雑な正規表現は不要です。
.raise_for_status()を呼び出すことで、HTTPステータスコードがエラー（4xx, 5xx）の場合に例外を発生させ、エラーハンドリングを容易にしています。
try...except requests.exceptions.RequestExceptionで、ネットワークエラーやHTTPエラーなどのAPIリクエスト関連の例外を捉えています。

APIベースのアプローチは、データ取得の容易さや信頼性の面で優位ですが、全ての機器が十分なAPIを提供しているわけではない点に注意が必要です。

実践的なヘルスチェック自動化の考慮点

現場で実際にヘルスチェック自動化スクリプトを運用するには、いくつかの考慮点があります。

複数のチェック項目の組み合わせ: CPU, メモリだけでなく、インターフェース状態、ログバッファ、特定のプロセス状態、ルーティング隣接関係など、複数のチェック項目を一つのスクリプトやワークフローに組み合わせることで、より包括的なヘルスチェックが可能です。
エラーハンドリングとリトライ: 接続エラー、コマンド実行エラー、APIエラーなどは必ず発生し得ます。適切に例外処理を行い、必要であれば数回のリトライを試みることで、スクリプトの頑健性を高めます。
認証情報の安全な管理: スクリプト内にユーザー名やパスワードを平文で記述することは避けるべきです。環境変数、秘密鍵、パスワードマネージャー、またはCyberArkのような認証情報管理システムなどを利用して、安全に認証情報を扱ってください。
定期実行とトリガー: ヘルスチェックは定期的に実行することで効果を発揮します。Linuxのcron、Windowsのタスクスケジューラ、JenkinsやGitLab CI/CDのようなCI/CDツール、Ansible Tower/AWXのような自動化プラットフォームなどを使って、スクリプトの定期実行をスケジュールします。特定のイベント（例: Syslogメッセージ）をトリガーとしてヘルスチェックを実行するイベント駆動型のアプローチも考えられます。
結果の通知と記録: ヘルスチェックの結果は、担当者に適切に通知される必要があります。メール、Slack、Microsoft Teamsなどのチャットツール、またはPagerDutyのようなインシデント管理ツールと連携することで、異常発生時に迅速な対応が可能となります。また、チェック結果をファイル、データベース、または時系列データベース（InfluxDBなど）に記録することで、後からトレンド分析やレポート作成に利用できます。
冪等性: 設定変更の自動化ほど厳密ではありませんが、ヘルスチェックにおいても、スクリプトを複数回実行しても副作用がないように考慮することが望ましいです。状態の取得のみを行うスクリプトであれば基本的に冪等性は保たれます。
異なるベンダーへの対応: 複数のベンダーの機器が混在する環境では、ベンダーごとに異なるコマンドやAPIへの対応が必要です。Netmikoのようにマルチベンダーに対応したライブラリを利用したり、Nornirのような自動化フレームワークでインベントリ情報に基づいて処理を切り替えたりするなどの手法があります。

自動化基盤やIaCとの連携

Pythonで実装したヘルスチェックは、単体で実行するだけでなく、既存のインフラ自動化基盤やIaC（Infrastructure as Code）のワークフローに組み込むことで、さらにその価値を高めることができます。

CI/CDパイプラインへの組み込み: 設定変更のデプロイパイプラインにヘルスチェックを組み込み、変更前後の機器状態を自動的に確認することで、変更による影響を早期に検知できます。
監視システムとの連携: Pythonスクリプトで取得したメトリクス（CPU使用率、メモリ使用率など）をPrometheusやZabbixのような監視システムに連携し、グラフ化やアラート発報に活用します。
IaCツールとの連携: AnsibleなどのIaCツールで設定変更を行った後に、Pythonスクリプトでヘルスチェックを実行し、意図した状態になっているかを確認するといった連携パターンも考えられます。

まとめ

この記事では、Pythonを使ったネットワーク機器のヘルスチェック自動化について、CLIベースとAPIベースの具体的な手法とコード例を交えて解説しました。Pythonの豊富なライブラリとプログラミングスキルを活用することで、ネットワーク機器に不慣れな方でも、効果的なヘルスチェックの仕組みを構築することが可能です。

ヘルスチェックの自動化は、ネットワーク運用の効率化、障害の早期発見、そして安定稼働に大きく貢献します。今回紹介した基本的なスクリプトを基に、対象機器や運用要件に合わせてカスタマイズし、実践的な自動化を進めていただければ幸いです。さらに進んだステップとして、ヘルスチェックで収集したデータを活用したトレンド分析や、イベント駆動型の自動応答などにも挑戦してみてください。