Сегодня 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:
class KubernetesSelectelClient:
def __init__(self):
self.steps = ClusterSteps()
def export_kubeconfig(self, cluster_id):
config.load_kube_config(config_file=os.environ.get("KUBECONFIG", self.steps.get_kubeconfig_file(
cluster_id=cluster_id)))
def get_client(self, cluster_id):
self.export_kubeconfig(cluster_id=cluster_id)
v1 = client
return v1
Для инициализации клиента необходимо было экспортировать kubeconfig, который мы получали после деплоя кластера. Для этого мы реализовали метод export_kubeconfig, внутри которого используется другой метод — get_kubeconfig_file:
def get_kubeconfig_file(self, cluster_id):
file_path = '{}.yaml'.format(os.getenv("CLUSTER_NAME", default="selectel"))
with open(file_path, "w") as kubeconfig:
with contextlib.redirect_stdout(kubeconfig):
response = self.client.get_request(self.client.url_with_cluster_id(cluster_id) + '/kubeconfig')
print(response.text)
return file_path
Файл kubeconfig мы получаем после деплоя кластера с помощью GET‑запроса:
/clusters/{cluster_id}/kubeconfig
Используя этот запрос, мы скачиваем kubeconfig — он нужен для взаимодействия с кластером через Kubernetes API.
В file_path мы записываем название файла, которое соответствует переменной окружения CLUSTER_NAME. Выводом функции get_kubeconfig_file является файл, в который записан результат нашего API-запроса по получению kubeconfig. Его мы используем для инициализации kubclient.
Работа фреймворка на практике
Более подробно работу нашего фреймворка мы разберем на примере кейса с проверкой работы Admission Controllers и наличием валидных Feature Gates внутри кластера.
Что такое Feature Gates и Admission Controllers
Feature Gates — это опция, с которой должен быть запущен необходимый компонент Kubernetes. C Features Gates доступны дополнительные возможности для компонента kube-apiserver.
Чтобы активировать Feature Gates, необходимо при создании или обновлении кластера указать в виде списка названия необходимых дополнений. Далее kube-apiserver будет запущен или перезапущен с опцией –feature-gates=… и заданными дополнениями. Если значение, которую вы передаете для опции, недоступно в текущей версии Kubernetes, будет выведена соответствующая ошибка.
Admission Controllers (контроллеры доступа) позволяют добавить дополнительные опции в работу Kubernetes для изменения или валидации объектов при запросах к Kubernetes API. Если в результате работы контроллера запрос отклоняется, вместе с ним не срабатывает весь запрос к API-серверу, а конечный пользователь получает ошибку.
Чтобы активировать контроллеры доступа, при создании или кластера необходимо указать названия контроллеров в виде списка. После этого kube-apiserver будет запущен или перезапущен с опцией –enable-admission-plugins и заданными контроллерами доступа. Если значение, которую вы передаете для опции, недоступно в текущей версии Kubernetes, будет выведена соответствующая ошибка.
В данном примере мы разберем тест, в котором мы обновляем Feature Gates и Admission Controllers и проверяем корректность нашего апдейта как со стороны Managed Kubernetes API, так и со стороны Kubernetes API.
@pytest.mark.regression
def test_update_admissions_controllers_and_feature_gates(clusters_steps, kubernetes_steps, create_cluster):
"""**Scenario:** Update admissions_controllers and feature_gates in cluster.
**Steps:**
#. Create cluster (POST /clusters)
#. Check that pods running
#. Update admissions_controllers and feature_gates
#. Check that cluster has status "ACTIVE"
#. Check admissions_controllers
#. Check feature_gates
#. Check feature_gates in kubernetes
# Create pod "test-image-pull-policy"
#. Check that pods running
#. Check image pull policy in kubernetes
#. Delete cluster
"""
#Проверка, что все поды находятся в статусе раннинг
kubernetes_steps.check_pods_after_deployment(cluster_id=create_cluster)
#Обновление кластера
clusters_steps.update_cluster(status=CLUSTER_STATUS_ACTIVE, data=ClusterSteps.data_for_update_cluster,
cluster_id=create_cluster)
#Получение admissions_controllers из API
admissions_controllers = clusters_steps.get_admissions_controllers(cluster_id=create_cluster)
#Получение feature_gates из API
feature_gates = clusters_steps.get_feature_gates_in_api(cluster_id=create_cluster)
#Проверка соответствия admissions_controllers внутри API
assert_that(admissions_controllers, equal_to(
ClusterSteps.data_for_update_cluster['cluster']['kubernetes_options']['admission_controllers'][0]))
#Проверка соответствия feature_gates внутри API
assert_that(feature_gates, equal_to(
ClusterSteps.data_for_update_cluster['cluster']['kubernetes_options']['feature_gates'][0]))
#Проверка наличия feature_gates через Kubernetes API kubernetes_steps.check_feature_gates_in_container(cluster_id=create_cluster,
feature_gate=feature_gates)
#Cоздание пода
pod = kubernetes_steps.create_pod(cluster_id=create_cluster,
file_name="test-image-pull-policy.yaml")
#Проверка, что все поды в запущены
kubernetes_steps.check_pods_after_deployment(cluster_id=create_cluster)
#Получение пода image_pull_policy
image_pull_policy = kubernetes_steps.get_image_pull_policy(pod_name=pod.metadata.name,
cluster_id=create_cluster)
#Проверка соответствия image_pull_policy c подом image_pull_policy, переданным при обновлении кластера
assert_that(image_pull_policy, equal_to(config.IMAGE_PULL_POLICY_ALWAYS))
Первым шагом после деплоя кластера мы проверяем (метод check_pods_after_deployment), что все поды находятся в статусе Running. Это необходимо, чтобы убедиться, что кластер находится в рабочем состоянии. Далее вызывается метод update_cluster из нашего API, в котором мы передаем значения admission_controller и feature_gate.
data_for_update_cluster = {"cluster": {"enable_autorepair": True,
"enable_patch_version_auto_upgrade": True,
"kubernetes_options": {"admission_controllers": ["AlwaysPullImages"],
"feature_gates": ["HonorPVReclaimPolicy"]},
"maintenance_window_start": "01:00:00"}
}
После обновления кластера мы проверяем admission_controllers и feature_gate на уровне API. После этого в методе check_feature_gates_in_container мы запускаем проверку соответствующего feature_gate в кластере.
def check_feature_gates_in_container(self, cluster_id, feature_gate):
self.set_cluster_client(cluster_id=cluster_id)
time.sleep(config.SMALL_TIMEOUT_FOR_RUNNING_TEST_SEC)
assert_that(self.client.CoreV1Api().read_namespaced_pod(name=self.get_name_kube_proxy_pod(cluster_id=cluster_id),
namespace="kube-system").spec.containers[0].command,
has_item("--feature-gates={}=true".format(feature_gate)))
В метод check_feature_gates_in_container мы передаем аргумент — значение feature_gate, которому в тесте присвоено значение в clusters_steps:
feature_gates = clusters_steps.get_feature_gates_in_api(cluster_id=create_cluster)
Таким образом, мы передаем в функцию check_feature_gates_in_container значение feature_gate, которое получили из нашего API. Далее внутри assert мы вызываем метод read_namespaced_pod, который отдает нам список значений feature_gate:
self.client.CoreV1Api().read_namespaced_pod(name=self.get_name_kube_proxy_pod(cluster_id=cluster_id),namespace="kube-system").spec.containers[0].command
['/usr/local/bin/kube-proxy', '--config=/var/lib/kube-proxy/config.conf', '--hostname-override=$(NODE_NAME)', '--feature-gates=HonorPVReclaimPolicy=true']
В списке отображается значение feature_gate, которое мы передали в функцию из API. Соответственно, мы проверили наличие feature_gate внутри кластера через сам Kubernetes API.
Получить admission_controller таким образом у нас не получится, потому что Python Kubernetes Client при чтении пода этот параметр не отдает. Поэтому здесь нужно немного сымпровизировать.
Мы, например, создали отдельный под, который описан внутри файла test-image-pull-policy.yaml. Его мы используем для прогона теста на корректную работу admission_controller — принцип этого теста будет описан ниже.
apiVersion: v1
kind: Pod
metadata:
name: test-image-pull-policy
spec:
containers:
- name: nginx
image: nginx:1.13
imagePullPolicy: IfNotPresent
Инициализация пода реализована в методе create_pod:
pod = kubernetes_steps.create_pod(cluster_id=create_cluster, file_name="test-image-pull-policy.yaml")
def create_pod(self, cluster_id, file_name):
self.set_cluster_client(cluster_id=cluster_id)
with open(path.join(path.dirname(__file__), "{}".format(file_name))) as f:
dep = yaml.safe_load(f)
k8s_apps_v1 = self.client.CoreV1Api()
pod = k8s_apps_v1.create_namespaced_pod(
body=dep, namespace='default')
return pod
C помощью менеджера контекста мы открываем файл, считываем его, переводим в формат словаря и передаем в качестве аргумента body в функцию create_namespaced_pod, которая создает необходимый под.
Далее мы снова проверяем, что все поды находятся в статусе Running. А после создания и проверки — получаем их image_pull_policy. Притом, что admission_controller при обновлении нашего кластера был передан AlwaysPullImages. Соответственно, imagePullPolicy нашего пода должен поменяться сразу с IfNotPresent (его мы указали при создании пода), на AlwaysPullImages — параметр, который мы изменили на шаге апдейта кластера.
Эта проверка выполнена в последнем assert:
assert_that(image_pull_policy, equal_to(config.IMAGE_PULL_POLICY_ALWAYS))
Готово — нам удалось протестировать кластер с помощью фреймворка и убедиться в том, что функционал feature_gates и admission_controllers работает корректно.
Заключение
Так с помощью самописного фреймворка на Python у нас есть возможность гибко тестировать кластеры Kubernetes и запускать итерационные сценарии, связанные со сторонними API.
Пакет PyTest нам также позволяет работать с динамическими фикстурами всех уровней и использовать их возможности — финализаторы, области видимости, объект request и другое. Также мы можем гибко параметризировать тесты и проставлять метки (marks) для управления пайплайном тестирования.
Библиотека python-kubernetes позволяет использовать набор готовых открытых классов и работы с Kubernetes. Из минусов можно отметить только отсутствие версионирования: часто приходится самостоятельно подбирать необходимую функцию для работы со своей версией Kubernetes, а это не всегда удобно.
