Сегодня Kubernetes — это основное средство для оркестрации контейнеров на рынке, поэтому их тестирование занимает особую позицию в перечне задач. Большинство тривиальных тестов можно запустить через команду kubectl, либо фреймворк Sonobuoy для тестирования сертификации версий.

Однако для сложных интеграционных тестов, которые завязаны на Kubernetes API, необходимо реализовать что-то свое. Я воспользовался клиентской библиотекой Python для Kubernetes, которая позволяет работать со всеми прелестями его интерфейса, соединил ее с PyTest и API нашего продукта — Managed Kubernetes. Что из этого вышло, показываю в статье.

Дисклеймер: статья будет полезна тем, кто имел опыт работы с 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. Его мы используем для инициализации клиента.

Работа фреймворка на практике

Более подробно работу нашего фреймворка мы разберем на примере кейса с проверкой работы Admission Controllers и наличием валидных Feature Gates внутри кластера.


В данном примере мы разберем тест, в котором мы обновляем 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
   """
  #Проверка, что все поды находятся в статусе Running
  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, а это не всегда удобно.

Какие решения для тестирования кластеров используете вы? Поделитесь своим опытом и вопросами в комментариях!

Другие интересные материалы

• Как мы автоматизировали тестирование OpenStack с помощью Rally и Tempest
• Kubernetes: большой обзор технологии, установка и настройка
• Ферма мобильных устройств — новый способ тестирования Android-приложений