Posted inPython

Python pyudev Linux: USB接続時のデータ処理を自動化する方法

Python tutorial - IT technology blog
Python tutorial - IT technology blog

周辺機器のイベントを「キャッチ」する3つのアプローチ

Linuxサーバーを管理していると、2TBのログファイルが詰まったHDDを接続した際、システムが自動的に認識・マウントし、即座に処理スクリプトを実行してほしいという場面に遭遇することがあるでしょう。

以下は、管理者がよく検討する3つの一般的な方法です：

  1. シェルスクリプト + Cronjob: 60秒ごとにlsblkを実行する方法。これでは動作が遅く、不要なシステム負荷がかかります。
  2. 純粋な Udev ルール: /etc/udev/rules.d/で直接設定する方法。非常に軽量ですが、デバッグが極めて困難です。udevルール内でAPIを叩いたりJSONをパースしたりする複雑なロジックをシェルスクリプトで書くのは苦行です。
  3. Python + pyudev ライブラリ: これが最適な選択肢です。Linuxのudevシステムの強力さと、Pythonの柔軟な処理能力を兼ね備えています。

なぜ pyudev はシェルスクリプトより優れているのか？

筆者の最初のプロジェクトは、USBからサーバーに数ファイルをコピーするだけのものでした。その時のPythonスクリプトはわずか50行程度。しかし、MD5チェックサムの確認、500GBのデータを日付別に分類、Slackへの通知送信といった要件が増えるにつれ、状況は複雑化しました。

シェルスクリプトを使用すると、メンテナンスは悪夢に変わります。pyudevはデバイスをオブジェクトとして管理することで、この問題を解決します。シリアル番号やベンダーIDによるデバイスのフィルタリングも非常にスマートに行えます。Netlinkを介してカーネルから直接イベントを受信するため、USBがポートに触れた瞬間にPythonが信号を受け取り、遅延は通常10ms以下です。

監視スクリプトの実装

環境構築

まず、ライブラリをインストールする必要があります。UbuntuやCentOSなどのほとんどのディストリビューションには既にlibudevが含まれているため、インストールは非常に簡単です。

pip install pyudev

サンプルコード：プラグアンドプレイイベントの監視

以下のコードは、バックグラウンドで動作する「監視役」として機能します。ストレージデバイス（ブロックデバイス）がシステムに接続されると、詳細情報を出力します。

import pyudev

def monitor_devices():
    context = pyudev.Context()
    monitor = pyudev.Monitor.from_netlink(context)
    monitor.filter_by(subsystem='block')

    print("--- デバイスを待機中... ---")
    
    for device in iter(monitor.poll, None):
        if device.action == 'add':
            print(f"検出: {device.device_node}")
            print(f"ファイルシステム: {device.get('ID_FS_TYPE')}")
            print(f"UUID: {device.get('ID_FS_UUID')}")
        elif device.action == 'remove':
            print(f"デバイスが取り外されました: {device.device_node}")

if __name__ == "__main__":
    monitor_devices()

自動マウントとデータ処理

本番環境では、マウントを実行するために通常subprocessモジュールを使用します。ヒント：常にID_FS_USAGE属性を確認するようにしてください。これにより、スワップパーティションや意図しないシステムパーティションを誤ってマウントすることを防げます。

import pyudev
import subprocess
import os

def handle_device_event(device):
    # デバイスが追加され、かつファイルシステムである場合
    if device.action == 'add' and device.get('ID_FS_USAGE') == 'filesystem':
        node = device.device_node
        label = device.get('ID_FS_LABEL') or "usb_disk"
        mount_path = f"/mnt/external/{label}"
        
        os.makedirs(mount_path, exist_ok=True)
        
        try:
            subprocess.run(["mount", node, mount_path], check=True)
            print(f"{node} を {mount_path} に正常にマウントしました")
            # ここにファイル処理ロジック（例：rsyncによるデータ同期）を追加
        except subprocess.CalledProcessError as e:
            print(f"マウントエラー: {e}")

def main():
    context = pyudev.Context()
    monitor = pyudev.Monitor.from_netlink(context)
    monitor.filter_by(subsystem='block')

    observer = pyudev.MonitorObserver(monitor, handle_device_event)
    observer.start()
    
    try:
        while True:
            import time
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()

if __name__ == "__main__":
    main()

大規模システム運用における実践的なヒント

スクリプトが数十行から数千行に成長するにつれ、サーバーのハングアップを防ぐために注意すべき重要なポイントがあります：

  • 重いタスクを直接実行しない: 100GBのデータを直接コピーしないでください。監視プロセス全体がブロックされてしまいます。タスクはthreadingや、Celery/Redisのようなキューに渡すようにしましょう。
  • 「ホットプラグ」エラーの処理: ユーザーはアンマウントせずにUSBを引き抜くことがよくあります。ファイルの読み書き操作はtry...exceptブロックで厳密に囲む必要があります。
  • 権限管理: 通常ユーザーで実行されるスクリプトはマウント権限がありません。パスワードなしでmount/umountコマンドを実行できるよう、sudoersを設定することをお勧めします。
  • UUIDによる識別: /dev/sdb1のような名前は再起動のたびに変わる可能性があるため、決して信頼しないでください。個別のデバイスを正確に識別するには、ID_FS_UUIDを使用してください。

この手法により、観測ステーションから中央サーバーへのオフラインデータの取り込みプロセスを完全に自動化することができました。このソリューションが、皆さんのDevOpsツールセットをより強力にする一助となれば幸いです。

Share:
Tags: