Posted inPython

Python pyudev on Linux: Automating Data Processing for USB Plug-ins on Servers

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

Three Approaches to “Catching” Peripheral Events

If you’re a Linux system administrator, you’ve likely faced a challenging scenario: How do you ensure that when a 2TB hard drive full of log files is plugged in, the system automatically recognizes it, mounts it to the correct location, and runs a processing script immediately?

Here are three common approaches that admins typically consider:

  1. Shell Script + Cronjob: Running lsblk every 60 seconds. This method is both slow and increases unnecessary system load.
  2. Pure Udev Rules: Configuring directly in /etc/udev/rules.d/. This method is very lightweight but extremely difficult to debug. Writing complex logic, such as calling an API or parsing JSON files using Shell scripts inside udev rules, is a nightmare.
  3. Python + pyudev library: This is the optimal choice. It combines the power of the Linux udev system with the flexible processing capabilities of Python.

Why is pyudev Superior to Shell Scripts?

My first project only required copying a few files from a USB to a server. At that time, the Python script was only about 50 lines long. However, as requirements grew—such as verifying MD5 checksums, categorizing 500GB of data by date, and sending notifications via Slack—things started to get complicated.

If using Shell scripts, maintenance would become a nightmare. pyudev solves this by managing devices as objects. You can filter devices by Serial Number or Vendor ID very cleanly. Since it listens directly from the Kernel via Netlink, Python receives the signal as soon as the USB connector touches the port, with latency usually under 10ms.

Implementing a Monitoring Script in Practice

Environment Setup

First, you need to install the library. Most distributions like Ubuntu or CentOS already have libudev available, so installation is quick and easy.

pip install pyudev

Sample Code: Monitoring Plug & Play Events

The code below acts as a background “sentry”. It prints detailed information as soon as any block device connects to the system.

import pyudev

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

    print("--- Waiting for devices... ---")
    
    for device in iter(monitor.poll, None):
        if device.action == 'add':
            print(f"Detected: {device.device_node}")
            print(f"File System: {device.get('ID_FS_TYPE')}")
            print(f"UUID: {device.get('ID_FS_UUID')}")
        elif device.action == 'remove':
            print(f"Device removed: {device.device_node}")

if __name__ == "__main__":
    monitor_devices()

Automating Mounting and Data Processing

For production environments, I usually use the subprocess module to execute mount commands. A small tip: always check the ID_FS_USAGE attribute. This helps you avoid accidentally mounting swap partitions or unwanted system partitions.

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"Successfully mounted {node} to {mount_path}")
            # Add file processing logic (e.g., rsync data) here
        except subprocess.CalledProcessError as e:
            print(f"Mount error: {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()

Real-World Experience Operating Large Systems

When a script grows from dozens to thousands of lines of code, there are critical issues you need to keep in mind to avoid hanging the server:

  • Never run heavy tasks directly: Never copy 100GB of data directly inside the udev callback function. This will block the entire listening process. Instead, push the task to threading or task queues like Celery/Redis.
  • Handle “hot unplug” errors: Users often pull out the USB without unmounting. You need to wrap file read/write operations in very tight try...except blocks.
  • Permission Management: Scripts running with standard user permissions usually cannot mount. You should configure sudoers to allow the user to execute mount/umount commands without a password.
  • Identify by UUID: Never rely on names like /dev/sdb1 because they can change after every reboot. Use ID_FS_UUID to accurately identify specific devices for each client.

This technique has helped me fully automate the process of offloading offline data from monitoring stations to a central server. I hope this solution makes your DevOps toolkit even more powerful.

Share:
Tags: