Говнокод #1 - Сортируем коробки по uptime с использованием Nornir, Scrapli, Python

Приветствую, друг!

Обещал себе, что нужно тратить больше времени на автоматизацию в 2024-м, пора начать выполнять.

Проблема изучения любого языка программирования - практика. Опытные дядьки скажут - отличная практика - писать код на работе. Выбери себе задачу и попробуй решить. Оно в целом да, но когда садишься думать “а что же такого написать” оказывается это не так и просто.

На досуге в ЛС пришел коллега и спросил “смогешь написать скрипт, который будет собирать с коробок аптайм, сортировать их и выводить в каком-то виде?”

Задача вроде понятна, надо пробежаться по коробкам (например используя nornir+scrapli) собрать вывод в структуру данных (например список/словарь etc), нормализовать аптайм, чтобы его можно было сравнивать, прикрутить сравнение и решить куда вывести полученное.

Ниже постарался описать, что у меня получилось :) Критика и пожелания приветствуется, начнем пока без ООП.

Код писал под SNR-ы, которые так удачно подвернулись в количестве трех штук. По причине различия вывода uptime и его расположение в разных вендорах, код будет отличаться, поэтому, думается мне, логичнее потом все это будет оформить в отдельный класс и наплодить внутри подклассы/методы на каждого вендора, но пока, как я уже сказал выше, без ООП и на одном вендоре.

Если пишите новый инструмент автоматизации, пишите сначала для одной коробки и одного вендора, такой сетап легче тестировать и фокус с задачи не сбивается на всякие межвендорные различия.

Что будем использовать? Nornir, модуль nornir-scrapli и python с его встроенными модулями.

Nornir - это фреймворк для написания Python-скриптов для автоматизации сетевых задач. Он предоставляет удобные средства для управления устройствами в сети, включая выполнение команд, фильтрацию устройств и многое другое.

Если не знакомы с Nornir, я писал о нем в отдельной заметке.

Scrapli - это библиотека Python для взаимодействия с устройствами через протоколы, такие как SSH. Похожа на netmiko, только лучше. Подробно рассказывать пока не буду т.к логика работы скрыта за модулем норнира - nornir-scrapli.

Должен ли сетевик кодить аки разработчик? Скорее нет, чем да, но как минимум, будем придерживаться следующих принципов:

продумываем заранее, что хочется видеть после выполнения с кода;
соблюдаем PEP8;
стараемся держаться подальше от спагетти-кода;
стремимся разделять код на модули, для удобства восприятия;
вносим базовые проверки, чтобы перехватить трейсбэки и пропускать проблемные коробки, вместо “падения” скрипта.

Напишет ли условный ChatGPT этот скрипт? Да, на 90% это будет рабочий код, важно лишь задать верный контекст и снабдить нейронку исходными данными. Останется только адаптировать под задачу и готово.

Есть ли тогда смысл разбирать все это? Мое мнение, конечно стоит. Нейронки без работы не останутся, а ты, друг мой, как инженер, нуждаешься в скиллах, которые дадут тебе хлеб, немного масла и конкурентные преимущества.

Всяческие рекомендации опробовать тот же ChatGPT и использовать его как помощника в изучении. Годится для генерация примеров, объяснения концепций кода, генерация кусков кода, код-ревью и прочие мелочи. Точно будет стойкое желание “а я ща кааак скопирую и пойду курить бамбук”, но это неправильная стратегия. Вот как стоит сделать:

перепиши руками
вникни в происходящее
выпиши неизвестные конструкции и разбери в том же ipython или pdbr по частям
сходи обсудить в группу если требуется мнение со стороны :)

Весь код можно будет найти на гитхабе. Комментарии и PR приветствуются.

Разбираем код

Пройдемся по коду и разберем основные моменты и нюансы.

import re

from nornir import InitNornir
from nornir.core.filter import F
from nornir_scrapli.tasks import (
    send_command,
)
from nornir.core.exceptions import NornirExecutionError

Здесь мы импортируем необходимые библиотеки и модули. re используется для работы с регулярными выражениями, а Nornir и Scrapli - для автоматизации задач сетевого взаимодействия, что есть что:

InitNornir - иницилизирует сам Nornir;
F - функция продвинутой фильтрации
send_command - отправка show команд через Scrapli
NornirExecutionError - Nornir перехватит исключение, если какая-либо из задач завершится сбоем.

Код модульный, каждая функция выполняет определенное действие. В дальнейшем с таким кодом будет удобнее работать. Импорты сделали, теперь пробежимся по каждой функции отдельно.

def get_uptime(task):
    '''
    Получает информацию о времени работы устройства.

    :param task: Объект задачи Nornir
    :return: Общее время работы устройства в минутах или None в случае ошибки

    '''
    regex = r'\w+ +(?P<weeks>\d+).+?(?P<days>\d+).+?(?P<hours>\d+).+?(?P<minutes>\d+)' 
    try:
        get_version = task.run(
            task=send_command,
            command="sh ver",
        )

        current_uptime = get_version.result.split('\n')[-1]
        match_uptime = re.search(regex, current_uptime)
        if match_uptime:
            weeks, days, hours, minutes = map(int, match_uptime.groups())
            total_minutes = weeks * 7 * 24 * 60 + days * 24 * 60 + hours * 60 + minutes
            return total_minutes
    except NornirExecutionError as error:
        print(f"Error accessing device {task.host.name}: {str(error)}")
    return None

Функция get_uptime содержит основную логику по сбору uptime с устройств.

Тут мы пишем регулярное выражение и “присваиваем” его переменнойregex. Регуляркой будем искать в строке формата Uptime is 3 weeks, 6 days, 2 hours, 36 minutes, которую вытянем из вывода sh version. Еще с курса Наташи я начал писать регулярки в regex.101, собсна там и продолжаю. Не забываем тыкнуть Python в Flavor.

Для выдергивания вывода sh version используем модуль nornir-scrapli, через отправку команды show version на устройство.

Результат выполнения таски будет хранится в переменной get_version в виде строки, оттуда и вырезаем нужный нам кусок с uptime.

Вывод команды с оборудования, чаще всего, получаем в виде единой строки с пробельными символами, например \t\n\r\f\v:

(Pdbr) result['R1'][1].result
'  SNR-S2982G-24T-POE-E Device, Compiled on May 07 19:33:53 2023\n  sysLocation Building 57/2,Predelnaya st, Ekaterinburg,
Russia\n  CPU Mac f8:f0:82:d3:b8:5e\n  Vlan MAC f8:f0:82:d3:b8:5d\n  SoftWare Version 7.0.3.5(R0241.0594)\n  BootRom Version
7.2.55\n  HardWare Version 1.0.1\n  CPLD Version N/A\n  Serial No.:SW083310MA15000036\n  Copyright (C) 2023 NAG LLC\n  All
rights reserved\n  Last reboot is cold reset.\n  Uptime is 4 weeks, 0 days, 0 hours, 19 minutes'

Нам нужно выгрызть только uptime и затем пробежаться регуляркой, либо можно написать регулярку на весь вывод, тогда не придется приседать с split/strip и пр. методами по работе со строками, выбор за вами. То, что нам интересно:

Uptime is 4 weeks, 0 days, 0 hours, 19 minutes

Т.к это строка, в ход идут популярные методы работы со строками, тут нам подходит split. С его помощью можно разбить строку на несколько элементов, которые соберутся в список. Наш uptime окажется последним элементом в списке, который мы можем гарантированно забрать обращением к списку через индекс [-1]. Получим следующее:

(Pdbr) result['R1'][1].result.split('\n')
[
    'SNR-S2982G-24T-POE-E Device, Compiled on May 07 19:33:53 2023',
    '  sysLocation Building 57/2,Predelnaya st, Ekaterinburg, Russia',
    '  CPU Mac f8:f0:82:d3:b8:5e',
    '  Vlan MAC f8:f0:82:d3:b8:5d',
    '  SoftWare Version 7.0.3.5(R0241.0594)',
    '  BootRom Version 7.2.55',
    '  HardWare Version 1.0.1',
    '  CPLD Version N/A',
    '  Serial No.:SW083310MA15000036',
    '  Copyright (C) 2023 NAG LLC',
    '  All rights reserved',
    '  Last reboot is cold reset.',
    '  Uptime is 4 weeks, 0 days, 0 hours, 19 minutes'
]

Добавим индекс [-1] и получим необходимую строку:

(Pdbr) result['R1'][1].result.strip().split('\n')[-1]
'  Uptime is 4 weeks, 0 days, 0 hours, 19 minutes'

Проверка с if нужна, чтобы исключить ошибки, если совпадения в re, по какой-то причине, не было. А вот если данные успешно найдены, преобразует их в минуты и возвращает общее время работы в минутах.

Тут может быть интересна вот эта строка:

weeks, days, hours, minutes = map(int, match_uptime.groups())

С помощью функции map мы не только распакуем все 4 элемента, которые собрались в регулярке, но и превратим их из строки в целые числа и все это в одной строке. Нам нужны именно числа т.к дальше будут математические операции, что логично.

def sort_devices(device_info):
    return device_info[1]

sort_devices - это функция сортировки устройств по времени работы. Она возвращает время работы для каждого устройства, что позволяет отсортировать устройства по этому критерию. Зачем? Удобно запихнуть в параметр key в функции sorted() который аля фильтр. Уверен, можно запилить лямбда-функцией, но я пока не дорос.

def format_uptime(device_info):
    '''
    Форматирует информацию о времени работы устройства для вывода.

    :param device_info: Кортеж (хост, время работы в минутах)
    :return: Строка с информацией о времени работы

    '''
    host, uptime_minutes = device_info
    
    if not isinstance(uptime_minutes, int):
        print(f"Error: Unexpected data type for uptime_minutes on {host}. Skipping.")
        return None

    weeks, days = divmod(uptime_minutes, 7 * 24 * 60)
    days, hours = divmod(days, 24 * 60)
    hours, minutes = divmod(hours, 60)
    return f"Device {host} uptime is {weeks} weeks, {days} days, {hours} hours, {minutes} minutes."

format_uptime - это функция форматирования uptime устройства. Она использует divmod для разбиения времени в минутах на недели, дни, часы и минуты, а затем возвращает отформатированную строку.

Сначала желательно выполнить проверку на соответствие, что uptime_minutes - целое число, иначе получим ошибку и скрипт прекратит выполнение.

Что такое divmod? divmod - это встроенная функция в Python, которая принимает два аргумента и возвращает пару чисел, представляющих результат целочисленного деления и остаток от деления этих двух чисел. Синтаксис функции divmod выглядит следующим образом:

divmod(a, b)

где a и b - это два числа.

Функция divmod выполняет деление a на b и возвращает кортеж из двух значений: результат целочисленного деления (a // b) И остаток от деления (a % b).

Рассмотрим ее работу на примере недель, дней, часов и минут из uptime:

weeks, days = divmod(uptime_minutes, 7 * 24 * 60) 
days, hours = divmod(days, 24 * 60) 
hours, minutes = divmod(hours, 60)

В данном коде uptime_minutes представляет собой общее время работы устройства в минутах.

weeks, days = divmod(uptime_minutes, 7 * 24 * 60): Эта строка преобразует общее количество минут в количество недель и дней. 7 * 24 * 60 представляет собой количество минут в неделе. divmod(uptime_minutes, 7 * 24 * 60) возвращает два значения: количество недель и остаток в минутах после вычета недель.
days, hours = divmod(days, 24 * 60): Затем оставшиеся дни из предыдущего шага делятся на количество минут в сутках (24 * 60), чтобы вычислить количество дней и количество минут, оставшихся после вычета целых суток.
hours, minutes = divmod(hours, 60): Последний шаг делит оставшееся количество минут на 60, чтобы получить количество часов и количество минут после вычета целых часов.

Так мы собираем uptime в нужный вид и выводим его через print с использованием f-строки.

def collect_devices_info(task):
    '''
    Собирает информацию о времени работы устройств из результатов выполнения задач.

    :param task: Результат выполнения задач Nornir
    :return: Список кортежей (хост, время работы в минутах)
    '''
    devices_info = []
    for host, task_result in task.items():
        uptime_minutes = task_result.result
        if uptime_minutes is not None:
            devices_info.append((host, uptime_minutes))
    return devices_info

Функция collect_devices_info собирает список кортежей из хоста и соответствующего ему uptime.

Почему список кортежей? В данном случае, данные, которые мы собираем (хост и время работы в минутах), образуют простую пару, и нет необходимости в сложной структуре типа словаря с ключами и значениями.

Список кортежей удобно передать функции sorted, которая сама прогонит каждый элемент списка и отсортирует по условию, указанному в аргументе key.

def main():
    '''
    Основная функция скрипта.

    Инициализирует Nornir, выполняет задачу получения времени работы, собирает информацию
    и выводит отсортированный результат. Сортировка убывающая.
    '''
    nr = InitNornir(config_file="./config.yaml")
    snr = nr.filter(F(groups__contains="snr"))
    result = snr.run(get_uptime)

    devices_info = collect_devices_info(result)
    sorted_devices = sorted(devices_info, key=sort_devices, reverse=True)
    for device_info in sorted_devices:
        print(format_uptime(device_info))

Функция main - блок с кодом, где вызываются все остальные, ранее объявленные, функции.

Здесь инициализируем объект Nornir, фильтруем устройства по группе snr, передаем Nornir-у функцию get_uptime в качестве задачи, которая будет выполнена на устройствах только из группы snr. После того, как таска отработает, результат будет хранится в переменной result. Прогоним его через функцию collect_devices_info и получим список кортежей в формате [("R1", 310), (...)]. И уже после переходим к передаче полученного списка в функцию sorted. Останется оформить все в какой-нибудь вывод, для этого прогоним отсортированный список через цикл, внутри которого используем функцию format_uptime, которая сформирует итоговый результат вида:

Device R1 uptime is 3 weeks, 6 days, 15 hours, 29 minutes.
Device R2 uptime is 3 weeks, 1 days, 3 hours, 26 minutes.
Device R3 uptime is 3 weeks, 1 days, 3 hours, 26 minutes.

Ну и под конец конструкция if __name__ == "__main__", которая должна быть в каждом скрипте, если планируете дальше импортировать функции в этом коде как модули в другом коде. В таком случае все, что находится в функции main отработает только тогда, когда код запускается напрямую.

Криво косо, но работает! И это важный момент на начальном этапе. Пробовать, пробовать и еще раз пробовать. Говнокодить придется, это часть процесса обучения. Все через это проходят.

Хочешь обсудить тему?

С вопросами, комментариями и/или замечаниями, приходи в чат или подписывайся на Telegram-канал.