Add k8s metallb test and sampleservice class
Change-Id: Ibfae133cc0841f802e93b290c87d411ef69d1a25
Related-PROD: PROD-21474
diff --git a/tcp_tests/managers/k8smanager.py b/tcp_tests/managers/k8smanager.py
index 38521c7..409df4f 100644
--- a/tcp_tests/managers/k8smanager.py
+++ b/tcp_tests/managers/k8smanager.py
@@ -16,6 +16,7 @@
import time
from uuid import uuid4
+import requests
import yaml
from devops.helpers import helpers
@@ -361,25 +362,19 @@
return [self._K8SManager__underlay.host_by_node_name(node_name=v)
for pillar in k8s_masters_fqdn for k, v in pillar.items()]
- def kubectl_run(self, name, image, port):
- with self.__underlay.remote(
- node_name=self.ctl_host) as remote:
- result = remote.check_call(
- "kubectl run {0} --image={1} --port={2}".format(
- name, image, port
- )
- )
- return result
+ def kubectl_run(self, name, image, port, replicas=None):
+ cmd = "kubectl run {0} --image={1} --port={2}".format(
+ name, image, port)
+ if replicas is not None:
+ cmd += " --replicas={}".format(replicas)
+ return self.__underlay.check_call(cmd=cmd, node_name=self.ctl_host)
- def kubectl_expose(self, resource, name, port, type):
- with self.__underlay.remote(
- node_name=self.ctl_host) as remote:
- result = remote.check_call(
- "kubectl expose {0} {1} --port={2} --type={3}".format(
- resource, name, port, type
- )
- )
- return result
+ def kubectl_expose(self, resource, name, port, type, target_name=None):
+ cmd = "kubectl expose {0} {1} --port={2} --type={3}".format(
+ resource, name, port, type)
+ if target_name is not None:
+ cmd += " --name={}".format(target_name)
+ return self.__underlay.check_call(cmd=cmd, node_name=self.ctl_host)
def kubectl_annotate(self, resource, name, annotation):
with self.__underlay.remote(
@@ -391,14 +386,11 @@
)
return result
- def get_svc_ip(self, name, namespace='kube-system'):
- with self.__underlay.remote(
- node_name=self.ctl_host) as remote:
- result = remote.check_call(
- "kubectl get svc {0} -n {1} | "
- "awk '{{print $3}}' | tail -1".format(name, namespace)
- )
- return result['stdout'][0].strip()
+ def get_svc_ip(self, name, namespace='kube-system', external=False):
+ cmd = "kubectl get svc {0} -n {1} | awk '{{print ${2}}}' | tail -1".\
+ format(name, namespace, 4 if external else 3)
+ result = self.__underlay.check_call(cmd, node_name=self.ctl_host)
+ return result['stdout'][0].strip()
@retry(300, exception=DevopsCalledProcessError)
def nslookup(self, host, src):
@@ -717,3 +709,42 @@
tgt="I@kubernetes:control:enabled:True",
pillar="_param:cluster_vip_address")[0]
return [vip for minion_id, vip in ctl_vip_pillar.items()][0]
+
+ def get_sample_deployment(self, name, **kwargs):
+ return K8SSampleDeployment(self, name, **kwargs)
+
+
+class K8SSampleDeployment:
+ def __init__(self, manager, name, replicas=2,
+ image='gcr.io/google-samples/node-hello:1.0', port=8080):
+ self.manager = manager
+ self.name = name
+ self.image = image
+ self.port = port
+ self.replicas = replicas
+
+ def run(self):
+ self.manager.kubectl_run(self.name, self.image, self.port,
+ replicas=self.replicas)
+
+ def expose(self, service_type='ClusterIP', target_name=None):
+ self.manager.kubectl_expose(
+ 'deployment', self.name, self.port, service_type, target_name)
+
+ def get_svc_ip(self, external=False):
+ return self.manager.get_svc_ip(self.name, namespace='default',
+ external=external)
+
+ def curl(self, external=False):
+ url = "http://{0}:{1}".format(
+ self.get_svc_ip(external=external), self.port)
+ if external:
+ return requests.get(url).text
+ else:
+ return self.manager.curl(url)
+
+ def is_service_available(self, external=False):
+ return "Hello Kubernetes!" in self.curl(external=external)
+
+ def wait_for_ready(self):
+ return self.manager.wait_deploy_ready(self.name)
diff --git a/tcp_tests/settings_oslo.py b/tcp_tests/settings_oslo.py
index 45b871d..541ca34 100644
--- a/tcp_tests/settings_oslo.py
+++ b/tcp_tests/settings_oslo.py
@@ -344,6 +344,8 @@
help="", default='coredns'),
ct.Cfg('kubernetes_coredns_enabled', ct.Boolean(),
help="", default=False),
+ ct.Cfg('kubernetes_metallb_enabled', ct.Boolean(),
+ help="", default=False),
ct.Cfg('kubelet_fail_on_swap', ct.Boolean(),
help="", default=False)
]
diff --git a/tcp_tests/tests/system/test_k8s_actions.py b/tcp_tests/tests/system/test_k8s_actions.py
index 2ebc010..1d28a5f 100644
--- a/tcp_tests/tests/system/test_k8s_actions.py
+++ b/tcp_tests/tests/system/test_k8s_actions.py
@@ -13,7 +13,6 @@
# under the License.
import pytest
-import time
from tcp_tests import logger
from tcp_tests import settings
@@ -98,24 +97,13 @@
availability, run conformance
"""
- deployment_name = 'test-dep-chain-upgrade'
-
show_step(5)
- k8s_deployed.kubectl_run(
- deployment_name, 'gcr.io/google-samples/node-hello:1.0', '8080')
- k8s_deployed.kubectl_expose(
- 'deployment', deployment_name, '8080', 'ClusterIP')
- sample_service_ip = k8s_deployed.get_svc_ip(deployment_name, 'default')
- k8s_deployed.wait_deploy_ready(deployment_name)
+ sample = k8s_deployed.get_sample_deployment('test-dep-chain-upgrade')
+ sample.run()
+ sample.expose()
+ sample.wait_for_ready()
- # workaround for PROD-20720
- time.sleep(30)
-
- def check_is_test_service_available():
- assert "Hello Kubernetes!" in k8s_deployed.curl(
- "http://{}:{}".format(sample_service_ip, 8080))
-
- check_is_test_service_available()
+ assert sample.is_service_available()
show_step(6)
k8s_deployed.run_conformance(log_out="k8s_conformance.log")
@@ -127,8 +115,63 @@
k8s_deployed.update_k8s_images(version)
LOG.info("Checking test service availability")
- check_is_test_service_available()
+ assert sample.is_service_available()
LOG.info("Running conformance on {} version".format(version))
log_name = "k8s_conformance_{}.log".format(version)
k8s_deployed.run_conformance(log_out=log_name, raise_on_err=False)
+
+ @pytest.mark.grap_versions
+ @pytest.mark.fail_snapshot
+ def test_k8s_metallb(self, show_step, config, k8s_deployed):
+ """Enable metallb in cluster and do basic tests
+
+ Scenario:
+ 1. Setup Kubernetes cluster with enabled metallb
+ 2. Check that metallb pods created in metallb-system namespace
+ 3. Run 5 sample deployments
+ 4. Expose deployments with type=LoadBalancer
+ 5. Check services availability from outside of cluster
+ 6. Run conformance
+ 7. Check services availability from outside of cluster
+ """
+ show_step(1)
+ if not config.k8s_deploy.kubernetes_metallb_enabled:
+ pytest.skip("Test requires metallb addon enabled")
+
+ show_step(2)
+ pods = k8s_deployed.api.pods.list(namespace="metallb-system")
+
+ def is_pod_exists_with_prefix(prefix):
+ for pod in pods:
+ if pod.name.startswith(prefix) and pod.phase == 'Running':
+ return True
+ return False
+
+ assert is_pod_exists_with_prefix("controller")
+ assert is_pod_exists_with_prefix("speaker")
+
+ show_step(3)
+ samples = []
+ for i in range(5):
+ name = 'test-dep-metallb-{}'.format(i)
+ sample = k8s_deployed.get_sample_deployment(name)
+ sample.run()
+ samples.append(sample)
+
+ show_step(4)
+ for sample in samples:
+ sample.expose('LoadBalancer')
+ for sample in samples:
+ sample.wait_for_ready()
+
+ show_step(5)
+ for sample in samples:
+ assert sample.is_service_available(external=True)
+
+ show_step(6)
+ k8s_deployed.run_conformance()
+
+ show_step(7)
+ for sample in samples:
+ assert sample.is_service_available(external=True)