Библиотеки Python в работе системного администратора Linux

Расширенные примеры

1. Мониторинг в реальном времени с psutil и asyncio

Пример асинхронного сбора данных о системе каждые 5 секунд.

import asyncio
import psutil

async def monitor():
    while True:
        cpu = psutil.cpu_percent(interval=1)
        mem = psutil.virtual_memory().percent
        disk = psutil.disk_usage('/').percent
        print(f"CPU: {cpu}% | RAM: {mem}% | Disk: {disk}%" )
        await asyncio.sleep(5)

if __name__ == '__main__':
    asyncio.run(monitor())

CPU: 15.0% | RAM: 52.3% | Disk: 68.1%
CPU: 10.2% | RAM: 51.8% | Disk: 68.1%
...

2. Асинхронное выполнение shell-команд с subprocess и asyncio

Запуск команды с таймаутом и обработкой ошибок.

import asyncio

async def run_cmd(cmd, timeout=10):
    proc = await asyncio.create_subprocess_shell(
        cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
        return proc.returncode, stdout.decode(), stderr.decode()
    except asyncio.TimeoutError:
        proc.kill()
        return -1, '', 'Timeout'

async def main():
    code, out, err = await run_cmd('ping -c 4 google.com', timeout=5)
    print(f"Exit code: {code}")
    print(out if code == 0 else err)

if __name__ == '__main__':
    asyncio.run(main())

Exit code: 0
PING google.com (142.250.185.78) 56(84) bytes of data.
64 bytes from ...
...

3. Поиск файлов с pathlib и glob

Рекурсивный поиск всех файлов с расширением .log в /var/log.

from pathlib import Path

p = Path('/var/log')
for log_file in p.rglob('*.log'):
    if log_file.stat().st_size > 10_000_000:  # больше 10 MB
        print(f"Большой лог: {log_file} - {log_file.stat().st_size / 1e6:.2f} MB")

Большой лог: /var/log/syslog - 15.34 MB
Большой лог: /var/log/kern.log - 12.78 MB

4. Чтение /proc/cpuinfo для детальной информации о процессоре

Парсинг /proc/cpuinfo без использования сторонних библиотек.

def get_cpu_info():
    with open('/proc/cpuinfo', 'r') as f:
        lines = f.readlines()
    cpu_info = {}
    current_key = None
    for line in lines:
        if ':' in line:
            key, value = line.split(':', 1)
            key = key.strip()
            value = value.strip()
            if key in ['model name', 'cpu cores', 'siblings', 'cache size']:
                cpu_info[key] = value
    return cpu_info

info = get_cpu_info()
print(f"Модель: {info.get('model name')}")
print(f"Ядер: {info.get('cpu cores')}")
print(f"Потоков: {info.get('siblings')}")
print(f"Кэш: {info.get('cache size')}")

Модель: Intel(R) Core(TM) i7-8700K CPU @ 3.70GHz
Ядер: 6
Потоков: 12
Кэш: 12288 KB

5. Удаленное выполнение команд через SSH с Paramiko

Подключение к удаленному хосту и выполнение команды.

import paramiko

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('192.168.1.100', username='admin', password='secret')

stdin, stdout, stderr = ssh.exec_command('uptime')
print(stdout.read().decode())
ssh.close()

 14:30:15 up 30 days,  2:15,  1 user,  load average: 0.08, 0.03, 0.01

6. Создание и применение YAML-шаблонов для конфигурации

Чтение шаблона и подстановка переменных.

import yaml
from string import Template

with open('template.yaml', 'r') as f:
    template_str = f.read()

tmpl = Template(template_str)
filled = tmpl.safe_substitute(hostname='server01', port=8080)

config = yaml.safe_load(filled)
print(config)

template.yaml:

server:
  host: ${hostname}
  port: ${port}

{'server': {'host': 'server01', 'port': 8080}}

7. Ротация логов с помощью RotatingFileHandler

Реализация логирования с ротацией по размеру.

import logging
from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler('/var/log/myapp.log', maxBytes=1024*1024, backupCount=5)
logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s %(levelname)s %(message)s',
                    handlers=[handler])

for i in range(1000):
    logging.info(f'Тестовая запись {i}')

После превышения 1 MB создаётся myapp.log.1, myapp.log.2 и т.д.

8. Остановка и удаление всех контейнеров Docker по определенному образу

import docker

client = docker.from_env()
for container in client.containers.list(all=True):
    if 'myapp:latest' in container.image.tags:
        container.stop()
        container.remove()
        print(f"Контейнер {container.short_id} удалён")

9. Сниффинг и анализ пакетов с фильтрацией (scapy)

Захват пакетов DNS и вывод запросов.

from scapy.all import sniff, DNSQR, IP

def dns_callback(pkt):
    if pkt.haslayer(DNSQR):
        domain = pkt[DNSQR].qname.decode()
        print(f"DNS запрос: {domain} от {pkt[IP].src}")

sniff(filter='udp port 53', prn=dns_callback, count=20)

DNS запрос: google.com. от 192.168.1.10
DNS запрос: example.org. от 192.168.1.11