Артем Минасян

Сегодня Kubernetes — это основное средство для оркестрации контейнеров на рынке, поэтому их тестирование занимает особую позицию в перечне задач. Большинство тривиальных тестов можно запустить через команду kubectl, либо фреймворк Sonobuoy для тестирования сертификации версий. Однако для сложных интеграционных тестов, которые завязаны на Kubernetes API, необходимо реализовать что-то свое. Я воспользовался клиентской библиотекой Python для Kubernetes, которая позволяет работать со всеми прелестями его интерфейса, и соединил ее с PyTest. Что из этого вышло, показываю в статье. Небольшая предыстория Одним из основных направлений продуктовой разработки в нашей компании является Managed Kubernetes, который позволяет работать с отказоустойчивыми и автомасштабируемыми кластерами Kubernetes. По приходу в компанию я познакомился с Kubernetes API, которое нужно было использовать для тестирования работы кластеров. Но в силу того, что для работы с кластером наши клиенты используют возможности Managed Kubernetes API, нужно было проводить интеграционное тестирование, которое включало бы проверку как со стороны Kubernetes API, так и со стороны Managed Kubernetes API. Самым удобным вариантом в нашей ситуации стала связка PyTest, Python и kubernetes-client, которую мы используем для тестирования. Было необходимо написать фреймворк для тестирования работы продукта, так как в процессе разработки достаточно часто появлялся новый функционал. Его работа могла аффектить функционирование нашего Managed Kubernetes API. Также эта задача выпала на мой испытательный срок — поэтому разработкой фреймворка занялся я. Подробнее о решении Для того, чтобы обращаться через Kubernetes API к нашему кластеру, нужно экспортировать kubeconfig, который генерируется в процессе деплоя кластера. Его мы можем просто скачать из панели управления кластером. Но так как мы хотели автоматизировать интеграционные тесты, экспорт kubeconfig нужно было делать через специальные запросы. Разберемся по порядку. Первым делом нужно было подготовить основу — инициализацию клиента, для которой мы создали скрипт kube_api.py: Для инициализации клиента необходимо было экспортировать kubeconfig, который мы получали после деплоя кластера. Для этого мы реализовали метод export_kubeconfig, внутри которого используется другой метод — get_kubeconfig_file: Файл kubeconfig мы получаем после деплоя кластера с помощью GET‑запроса: Используя этот запрос, мы скачиваем kubeconfig — он нужен для взаимодействия с кластером через Kubernetes API. В file_path мы записываем название файла, которое соответствует переменной окружения CLUSTER_NAME. Выводом функции get_kubeconfig_file является файл, в который записан результат нашего API-запроса по получению kubeconfig. Его мы используем для инициализации kubclient. Работа фреймворка на практике Более подробно работу нашего фреймворка мы разберем на примере кейса с проверкой работы Admission Controllers и наличием валидных Feature Gates внутри кластера. В данном примере мы разберем тест, в котором мы обновляем Feature Gates и Admission Controllers и проверяем корректность нашего апдейта как со стороны Managed Kubernetes API, так и со стороны Kubernetes API. Первым шагом после деплоя кластера мы проверяем (метод check_pods_after_deployment), что все поды находятся в статусе Running. Это необходимо, чтобы убедиться, что кластер находится в рабочем состоянии. Далее вызывается метод update_cluster из нашего API, в котором мы передаем значения admission_controller и feature_gate. После обновления кластера мы проверяем admission_controllers и feature_gate на уровне API. После этого в методе check_feature_gates_in_container мы запускаем проверку соответствующего feature_gate в кластере. В метод check_feature_gates_in_container мы передаем аргумент — значение feature_gate, которому в тесте присвоено значение в clusters_steps: Таким образом, мы передаем в функцию check_feature_gates_in_container значение feature_gate, которое получили из нашего API. Далее внутри assert мы вызываем метод read_namespaced_pod, который отдает нам список значений feature_gate: В списке отображается значение feature_gate, которое мы передали в функцию из API. Соответственно, мы проверили наличие feature_gate внутри кластера через сам Kubernetes API. Получить admission_controller таким образом у нас не получится, потому что Python Kubernetes Client при чтении пода этот параметр не отдает. Поэтому здесь нужно немного сымпровизировать. Мы, например, создали отдельный под, который описан внутри файла test-image-pull-policy.yaml. Его мы используем для прогона теста на корректную работу admission_controller — принцип этого теста будет описан ниже. Инициализация пода реализована в методе create_pod: C помощью менеджера контекста мы открываем файл, считываем его, переводим в формат словаря и передаем в качестве аргумента body в функцию create_namespaced_pod, которая создает необходимый под. Далее мы снова проверяем, что все поды находятся в статусе Running. А после создания и проверки — получаем их image_pull_policy. Притом, что admission_controller при обновлении нашего кластера был передан AlwaysPullImages. Соответственно, imagePullPolicy нашего пода должен поменяться сразу с IfNotPresent (его мы указали при создании пода), на AlwaysPullImages — параметр, который мы изменили на шаге апдейта кластера. Эта проверка выполнена в последнем assert: Готово — нам удалось протестировать кластер с помощью фреймворка и убедиться в том, что функционал feature_gates и admission_controllers работает корректно. Заключение Так с помощью самописного фреймворка на Python у нас есть возможность гибко тестировать кластеры Kubernetes и запускать итерационные сценарии, связанные со сторонними API. Пакет PyTest нам также позволяет работать с динамическими фикстурами всех уровней и использовать их возможности — финализаторы, области видимости, объект request и другое. Также мы можем гибко параметризировать тесты и проставлять метки (marks) для управления пайплайном тестирования. Библиотека python-kubernetes позволяет использовать набор готовых открытых классов и работы с Kubernetes. Из минусов можно отметить только отсутствие версионирования: часто приходится самостоятельно подбирать необходимую функцию для работы со своей версией Kubernetes, а это не всегда удобно.

Артем Минасян

Материалы автора

Как тестировать кластеры Kubernetes с помощью Python