Рубрики
Без рубрики

Тестирование интеграции Kubernetes

Краткое тестирование трубопроводов инфраструктуры имеет важное значение для обеспечения минимальной регрессии … Теги с Куберовщиками, интеграцией, Python, Ruby.

Резюме

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

Использовать дело

Для теста я создал образец флэк-приложения, развернутое на EKS, которые выйдут на S3 и прихватывают файл под названием «Test». Манифест Flask-App определяет готовность зонда для обеспечения конечной точки/тестирования (которая захватывает файл с S3) может успешно стрелять. Наша тестовая набор Inspec будет обеспечивать развертывание POD и запускается. Pytest Suite проверит компоненты K8S, такие как IAM авторизация через наш зонд готовности, сети, и имеет возможность расширяться до конца для окончания тестов в будущем.

Если развертывание этого набора кода Убедитесь, что заменить любые с соответствующим значением для вашей среды.

Технологии

Инспекта

Несмотря на то, что предложения с открытым исходным кодом, такие как инспектива- k8s. , Подкладка SDK был написан компанией, которая закрыла свои двери в 2019 году. В зависимости от поддержанного бегуна для вашего кластера создания, вы также можете столкнуться с несоответствием зависимостей, создавая непередательное выполнение, см. Сухие типы Овладение . В конечном итоге можно, но для начала необходимы несколько обходных путей.

Чтобы полностью использовать Inspec, вам необходимо сначала развернуть настройку приложения, а также Установите поезд Кубернаны Отказ Альтернативно, работает из BGEESAMAN’S INSPEC-K8S-RINNER Контейнер Docker дает вам хорошее стартовое место. После установки вы можете начать тестировать различные компоненты в пределах вашего пространства имен.

Ниже приведены некоторые образцы ресурсов, которые вы можете использовать для тестирования.

### Test your pods exists, and all are running
control "k8s-app-validate" do
  impact 1.0
  title "Validate K8s test Application"
  desc "The k8s-app test app should exist and be running"

  ### Test various namespaces exist
  describe k8sobject(api: 'v1', type: 'namespaces', name: 'default') do
      it { should exist }
  end
  ### Test for every pod in deployment the pod exists, and is running
  k8sobjects(api: 'v1', type: 'pods', namespace: 'default', labelSelector: 'app=flask').items.each do |pod|
    describe "#{pod.namespace}/#{pod.name} pod" do
      subject { k8sobject(api: 'v1', type: 'pods', namespace: pod.namespace, name: pod.name) }
      it { should exist }
      it { should be_running }
    end
  end
  ### Test your service exists
  describe k8sobjects(api: 'v1', type: 'services', namespace: 'default', labelSelector: 'app=flask') do
    it { should exist }
  end
end

Питиш

Паров имеет рамки под названием Kubetest , который будет развернуть конфигурацию к вашему кластеру в уникальном пространстве имен. Эта структура предназначена для развертывания и тестирования и не обязательно тестирования существующей инфраструктуры, которая существует. Одним из преимуществ этой рамки является ли узы, чеки готовности, тогда как Inspec не сделал.

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

Существуют некоторые ограничения в поддержанных объектах API, см. API Ресурсы .

from boto3 import client
from time import time

# Create And Test Service Account
def create_sa(kube, modifier):
    """
    A helper function to create service account
    """
    sa = kube.load_serviceaccount("configs/sa.yaml")
    account_id = client('sts').get_caller_identity()["Account"]
    role_arn = f"arn:aws:iam::{account_id}:role/test_role"
    sa.obj.metadata.annotations['eks.amazonaws.com/role-arn'] = role_arn
    return sa

def test_create_sa(kube, modifier):
    """
    A function to test the creation of a service account
    Goal: This will test the ability to interface with the k8s cli
    """
    sa = create_sa(kube, modifier)
    kube.create(sa)
    assert sa.is_ready()
# Create and Test Deployment
def create_deployment(kube, sa, modifier):
    """
    A helper function to create a deployment object
    """
    account_id = client('sts').get_caller_identity()["Account"]

    deployment = kube.load_deployment('configs/deployment.yaml')
    repository = f"{account_id}.dkr.ecr.us-east-1.amazonaws.com/k8s-test"
    deployment.obj.spec.template.spec.containers[0].image = repository
    deployment.obj.spec.template.spec.service_account_name = sa.obj.metadata.name
    deployment.obj.spec.template.spec.containers[0].env[1].value = f"quackenbush-test-bucket"
    return deployment

def test_deployment(kube, modifier):
    """
    A function to test the creation of a deployment.
    Goals: If a pod becomes ready, that means it can successful connect to SSM
    """
    sa = create_sa(kube, modifier)
    kube.create(sa)

    deployment = create_deployment(kube, sa, modifier)
    kube.create(deployment)

    timeout = time() + 60 # 60 seconds
    while not deployment.is_ready():
        if time() > timeout:
            # Fail early
            assert deployment.is_ready()

    pods = deployment.get_pods()

    for pod in pods:
        pod.wait_until_containers_start(timeout=60)
        timeout = time() + 300 # 5 Minutes
        while not pod.is_ready():
            if time() > timeout:
                break

        assert pod.is_ready()

Идти

Другими популярными рамками, которые могут быть выполнены, является терратест. Эта кодовая база может быть расширена, чтобы использовать те же рабочие процессы, что и раньше, где у нас есть наш контейнер связи с S3. Однако для простоты я продемонстрирую, просто спинут простой веб-стручок, а затем туннелирующий трафик для проверки здоровья.

package test

import (
    "crypto/tls"
    "fmt"
    "path/filepath"
    "testing"
    "time"

    http_helper "github.com/gruntwork-io/terratest/modules/http-helper"
    "github.com/gruntwork-io/terratest/modules/k8s"
)

func TestKubernetesDeployment(t *testing.T) {
    t.Run("test", func(t *testing.T) {
        validateK8s(t)
    })
}
func validateK8s(t *testing.T) {
    kubeResourcePath, _ := filepath.Abs("./configs/")
    podName := "static-web"

    options := k8s.NewKubectlOptions("", "", "default")
    k8s.KubectlApply(t, options, kubeResourcePath)

    defer k8s.KubectlDelete(t, options, kubeResourcePath)
    k8s.WaitUntilPodAvailable(t, options, podName, 30, 5*time.Second)

    // Create Tunnel and validate success
    tunnel := k8s.NewTunnel(options, k8s.ResourceTypePod, podName, 0, 80)
    defer tunnel.Close()
    tunnel.ForwardPort(t)
    tlsConfig := tls.Config{}
    http_helper.HttpGetWithRetryWithCustomValidation(
        t,
        fmt.Sprintf("http://%s", tunnel.Endpoint()),
        &tlsConfig,
        10,
        5*time.Second,
        validateHealth,
    )
}

func validateHealth(statusCode int, body string) bool {
    if statusCode != 200 {
        return false
    }
    // Were going to ignore body, but we could do some test there too
    return true
}

Оригинал: “https://dev.to/danquack/kubernetes-integration-testing-283d”