pythonでプロセス全体をpingでモニタリングするサンプル

特定のホストを再起動させ、プロセス全体をpingでモニタリングするサンプル

import subprocess
import time
import threading
from datetime import datetime

def ping_host(host):
    """
    指定されたホストに対して単一のpingを実行する。
    
    Args:
        host (str): pingを実行するホストのアドレス
    
    Returns:
        tuple: (bool, float) pingが成功したかどうかと応答時間
    """
    try:
        start_time = time.time()
        result = subprocess.run(["ping", "-c", "1", host], 
                                stdout=subprocess.PIPE, 
                                stderr=subprocess.PIPE, 
                                check=True, text=True)
        end_time = time.time()
        response_time = end_time - start_time
        return True, response_time
    except subprocess.CalledProcessError:
        return False, 0

def continuous_ping(host, stop_event):
    """
    指定されたホストに対して継続的にpingを実行し、結果を定期的に出力する。
    stop_eventがセットされるまで続ける。
    
    Args:
        host (str): pingを実行するホストのアドレス
        stop_event (threading.Event): pingを停止するためのイベント
    """
    while not stop_event.is_set():
        success, response_time = ping_host(host)
        current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        if success:
            print(f"[{current_time}] Ping to {host} successful. Response time: {response_time:.3f} seconds")
        else:
            print(f"[{current_time}] Ping to {host} failed")
        time.sleep(1)

def reboot_and_check(host):
    """
    指定されたホストを再起動し、再起動プロセス全体をモニタリングする。
    
    Args:
        host (str): 再起動するホストのアドレス
    """
    # バックグラウンドでのping実行を開始
    stop_event = threading.Event()
    ping_thread = threading.Thread(target=continuous_ping, args=(host, stop_event))
    ping_thread.start()

    print(f"Starting continuous ping to {host}")
    time.sleep(5)  # 再起動前にpingが安定するのを待つ

    # ホストの再起動
    print(f"Rebooting {host}")
    subprocess.run(["ssh", host, "sudo reboot"], check=True)

    # ホストがダウンするまで待機
    print("Waiting for host to go down...")
    while ping_host(host)[0]:
        time.sleep(1)

    # ホストが再びオンラインになるまで待機
    print("Host is down. Waiting for it to come back up...")
    while not ping_host(host)[0]:
        time.sleep(1)

    # ホストがオンラインに戻ったことを確認
    print(f"{host} is back online")
    
    # バックグラウンドのping実行を停止
    stop_event.set()
    ping_thread.join()

if __name__ == "__main__":
    # 再起動したいホストのアドレスを指定
    host = "example.com"
    reboot_and_check(host)

ホストAとホストB間でリンク障害を発生・復旧させ、プロセス全体をpingでモニタリングするサンプル

import subprocess
import time
import threading
from datetime import datetime

def ping_host(host):
    """
    指定されたホストに対して単一のpingを実行する。

    Args:
        host (str): pingを実行するホストのアドレス

    Returns:
        bool: pingが成功した場合はTrue、失敗した場合はFalse
    """
    try:
        result = subprocess.run(["ping", "-c", "1", host], 
                                stdout=subprocess.PIPE, 
                                stderr=subprocess.PIPE, 
                                check=True, text=True)
        return True
    except subprocess.CalledProcessError:
        return False

def continuous_ping(source_host, target_host, stop_event, ping_results):
    """
    指定されたソースホストから対象ホストに対して継続的にpingを実行し、結果を記録する。

    Args:
        source_host (str): pingを実行するソースホストのアドレス
        target_host (str): pingの対象となるホストのアドレス
        stop_event (threading.Event): pingを停止するためのイベント
        ping_results (list): ping結果を格納するリスト
    """
    while not stop_event.is_set():
        success = ping_host(target_host)
        current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        result = "successful" if success else "failed"
        print(f"[{current_time}] Ping from {source_host} to {target_host} {result}")
        ping_results.append(success)
        time.sleep(1)

def check_port_status(host, interface):
    """
    指定されたホストのインターフェースの状態を確認する。

    Args:
        host (str): 確認対象のホストのアドレス
        interface (str): 確認対象のインターフェース名

    Returns:
        bool: インターフェースがUPの場合はTrue、それ以外はFalse
    """
    try:
        result = subprocess.run(["ssh", host, f"ip link show {interface}"], 
                                stdout=subprocess.PIPE, 
                                stderr=subprocess.PIPE, 
                                check=True, text=True)
        return "UP" in result.stdout
    except subprocess.CalledProcessError:
        return False

def test_ping(ping_results, expected_result, test_description):
    """
    pingの結果をテストし、結果を出力する。

    Args:
        ping_results (list): ping結果のリスト
        expected_result (bool): 期待される結果（True: 成功、False: 失敗）
        test_description (str): テストの説明

    Returns:
        bool: テストが成功した場合はTrue、失敗した場合はFalse
    """
    actual_result = ping_results[-1]
    if actual_result == expected_result:
        result_str = "成功" if expected_result else "失敗"
        print(f"テストOK: {test_description}後にpingが{result_str}しています")
        return True
    else:
        result_str = "失敗" if expected_result else "成功"
        print(f"テストNG: {test_description}後もpingが{result_str}しています")
        return False

def test_port_status(host, interface, expected_status, test_description):
    """
    ポートの状態をテストし、結果を出力する。

    Args:
        host (str): 確認対象のホストのアドレス
        interface (str): 確認対象のインターフェース名
        expected_status (bool): 期待されるポートの状態（True: UP、False: DOWN）
        test_description (str): テストの説明

    Returns:
        bool: テストが成功した場合はTrue、失敗した場合はFalse
    """
    actual_status = check_port_status(host, interface)
    if actual_status == expected_status:
        status_str = "Up" if expected_status else "Down"
        print(f"テストOK: {test_description}のポートが{status_str}しています")
        return True
    else:
        status_str = "Down" if expected_status else "Up"
        print(f"テストNG: {test_description}のポートが{status_str}していません")
            return False

def simulate_link_failure(source_host, target_host, interface, failure_duration):
    """
    リンク障害をシミュレートし、その間のネットワーク状態をテストする。

    Args:
        source_host (str): pingを実行するソースホストのアドレス
        target_host (str): pingの対象となるホストのアドレス
        interface (str): 障害を発生させるインターフェース名
        failure_duration (int): リンク障害の継続時間（秒）
    """
    stop_event = threading.Event()
    ping_results = []
    ping_thread = threading.Thread(target=continuous_ping, args=(source_host, target_host, stop_event, ping_results))
    ping_thread.start()

    print(f"1. Starting continuous ping from {source_host} to {target_host}")
    time.sleep(5)  # 障害発生前にpingが安定するのを待つ

    print(f"2. Simulating link failure between {source_host} and {target_host}")
    subprocess.run(["ssh", source_host, f"sudo ip link set dev {interface} down"], check=True)

    time.sleep(5)  # 障害が反映されるのを待つ

    print("3. Testing if ping fails after link failure")
    test_ping(ping_results, False, "リンク切断")

    print("4. Checking if the port is down in the logs")
    test_port_status(source_host, interface, False, "障害箇所")

    print(f"5. Restoring link between {source_host} and {target_host}")
    subprocess.run(["ssh", source_host, f"sudo ip link set dev {interface} up"], check=True)

    time.sleep(5)  # リンクが復旧するのを待つ

    print("6. Testing if ping succeeds after link restoration")
    test_ping(ping_results, True, "リンク復旧")

    print("7. Checking if the port is up in the logs")
    test_port_status(source_host, interface, True, "障害箇所")

    stop_event.set()
    ping_thread.join()

if __name__ == "__main__":
    # テスト設定
    source_host = "192.168.1.100"  # pingを実行するソース機器のIPアドレス
    target_host = "192.168.1.200"  # pingの対象となる機器のIPアドレス
    interface = "eth0"  # 障害を発生させるインターフェース
    failure_duration = 30  # リンク障害の継続時間（秒）

    # リンク障害シミュレーションの実行
    simulate_link_failure(source_host, target_host, interface, failure_duration)