Python для автоматизации рутинных операций в Linux

Раздел: Системное администрирование -> Автоматизация

Автоматизация администрирования Linux с помощью Python

Самым надёжным и гибким речением для автоматизации задач системного администрирования на Linux считается модуль subprocess. Он позволяет выполнять произвольные команды оболочки, перехватывать вывод, управлять потоками ошибок и обрабатывать коды завершения. В отличие от устаревшего os.system, subprocess полностью контролирует выполнение и защищает от инъекций команд при правильном использовании списков аргументов.

Как выполнить команду и получить её результат сразу?

Пример ниже запускает ls -la, собирает stdout и stderr:

import subprocess

result = subprocess.run(['ls', '-la', '/home'], capture_output=True, text=True)
print('stdout:', result.stdout)
print('stderr:', result.stderr)
print('returncode:', result.returncode)

создание ботов python (создание ботов на python)

Пояснение: capture_output=True объединяет stdout и stderr, text=True возвращает строки вместо байтов. Код возврата 0 означает успех.

Типичная ошибка: попытка передать целую команду строкой с shell=True без экранирования. Это ведёт к уязвимостям. Лучше передавать аргументы списком или тщательно проверять вводимые данные.

Альтернативные подходы

Как выполнить простую команду без захвата вывода?

Модуль os.system подходит для быстрых однострочников, когда вывод не нужен:

import os

exit_code = os.system('touch /tmp/test_file')
if exit_code == 0:
    print('Файл создан')

автоматизация linux python (автоматизация linux с python)

Цель: минималистичный запуск. Случаи использования: создание файлов, запуск сервисов без отслеживания вывода. Недостаток: нельзя перехватить вывод или обработать ошибки.

При использовании os.system вывод команды идёт напрямую в терминал, что мешает автоматизации. Также отсутствует защита от shell-инъекций.

Как автоматизировать интерактивные сессии (например, ssh с паролем)?

Модуль pexpect эмулирует терминал и отвечает на запросы:

import pexpect

child = pexpect.spawn('ssh user@host')
child.expect('password:')
child.sendline('secret')
child.expect('$')
child.sendline('ls /tmp')
child.expect(pexpect.EOF)
print(child.before.decode())

Пояснение: spawn запускает процесс, expect ждёт определённый вывод, sendline отправляет строку. Подходит для ftp, telnet, ssh с паролями (лучше использовать ключи).

Пароль в скрипте – угроза безопасности. Для SSH надёжнее использовать ключи и библиотеку paramiko. Тайм-ауты expect могут привести к зависанию.

Как удалённо администрировать множество серверов?

Библиотека paramiko предоставляет SSH-клиент для Python:

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='admin', key_filename='/home/user/.ssh/id_rsa')
stdin, stdout, stderr = client.exec_command('uptime')
print(stdout.read().decode())
client.close()

Цель: автоматизация повторяющихся команд на группе серверов. Случаи: сбор логов, развёртывание конфигов, проверка состояния.

Проблема: медленная работа при большом количестве серверов (один поток). Решение – использовать пул потоков или asyncio. Также важно управлять исключениями при недоступности хоста.

Как отслеживать состояние системы и реагировать на события?

Модуль psutil даёт доступ к информации о процессах, памяти, диске:

import psutil

cpu = psutil.cpu_percent(interval=1)
mem = psutil.virtual_memory().percent
if cpu > 80 or mem > 90:
    print('Нагрузка высокая: CPU {}% MEM {}%'.format(cpu, mem))

Пояснение: cpu_percent за интервал, virtual_memory возвращает объект с процентом использования. Служит основой для скриптов мониторинга.

Ошибка: вызов cpu_percent без интервала даёт 0.0. Также на некоторых системах требуются дополнительные права для чтения определенных данных.

Расширенные примеры автоматизации

Автоматическая очистка временных файлов по возрасту

Скрипт удаляет файлы в /tmp, к которым не обращались более 7 дней:

Пример

import os
import time
from pathlib import Path

cutoff = time.time() - 7 * 86400
count = 0
for path in Path('/tmp').iterdir():
    if path.is_file() and path.stat().st_atime < cutoff:
        path.unlink()
        count += 1
print(f'Удалено {count} файлов')

Удалено 12 файлов

Пояснение: используется pathlib для итерации и st_atime для даты последнего доступа. Фильтр можно адаптировать под st_mtime (модификация).

Параллельный сбор метрик с нескольких серверов через SSH

Скрипт с использованием concurrent.futures и paramiko собирает uptime на трёх хостах:

Пример

import paramiko
from concurrent.futures import ThreadPoolExecutor

HOSTS = ['server1', 'server2', 'server3']
USER = 'admin'
KEY = '/home/user/.ssh/id_rsa'

def get_uptime(host):
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        client.connect(host, username=USER, key_filename=KEY)
        stdin, stdout, stderr = client.exec_command('uptime')
        data = stdout.read().decode().strip()
    except Exception as e:
        data = f'Error: {e}'
    finally:
        client.close()
    return host, data

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(get_uptime, h) for h in HOSTS]
    for f in futures:
        host, result = f.result()
        print(f'{host}: {result}')

server1: 10:30:42 up 30 days,  2:17,  0 users,  load average: 0.00, 0.01, 0.05
server2: 10:30:42 up 14 days,  1:00,  0 users,  load average: 0.08, 0.02, 0.01
server3: 10:30:42 up 45 days, 19:18,  0 users,  load average: 0.00, 0.00, 0.00

Пояснение: пул потоков ускоряет обработку. Вместо вывода можно записывать метрики в CSV или InfluxDB.

Мониторинг и уведомление при падении процесса

Скрипт проверяет, запущен ли процесс nginx, и отправляет уведомление через Telegram бота:

Пример

import psutil
import requests

PROCESS_NAME = 'nginx'
BOT_TOKEN = '123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11'
CHAT_ID = '987654321'

def check_process():
    for proc in psutil.process_iter(attrs=['pid', 'name']):
        if proc.info['name'] == PROCESS_NAME:
            return True
    return False

if not check_process():
    msg = f'{PROCESS_NAME} не запущен на {platform.node()}'
    url = f'https://api.telegram.org/bot{BOT_TOKEN}/sendMessage'
    requests.post(url, json={'chat_id': CHAT_ID, 'text': msg})

Пояснение: process_iter перебирает процессы без пауз. Уведомление через Telegram – простой способ alert'инга. Для production стоит добавить повторные проверки.

Автоматическое обновление пакетов через apt

Скрипт запускает sudo apt update и apt upgrade -y, логирует вывод:

Пример

import subprocess
import logging

logging.basicConfig(level=logging.INFO, filename='/var/log/apt-upgrade.log')

def run_apt():
    try:
        logging.info('Запуск apt update')
        subprocess.run(['sudo', 'apt', 'update'], check=True, capture_output=True, text=True)
        logging.info('Запуск apt upgrade -y')
        result = subprocess.run(['sudo', 'apt', 'upgrade', '-y'], check=True, capture_output=True, text=True)
        logging.info(result.stdout)
    except subprocess.CalledProcessError as e:
        logging.error(f'Ошибка: {e.stderr}')

if __name__ == '__main__':
    run_apt()

Пояснение: check=True вызывает исключение при ненулевом коде, что позволяет централизованно обрабатывать ошибки. Логирование помогает отследить проблемы.

Автоматизация Linux с Python - comments

автоматизация linux python (python)