resources/libraries/python/KubernetesUtils.py

   1 # Copyright (c) 2018 Cisco and/or its affiliates.
   2 # Licensed under the Apache License, Version 2.0 (the "License");
   3 # you may not use this file except in compliance with the License.
   4 # You may obtain a copy of the License at:
   5 #
   6 #     http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS,
  10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11 # See the License for the specific language governing permissions and
  12 # limitations under the License.
  13
  14 """Library to control Kubernetes kubectl."""
  15
  16 from time import sleep
  17
  18 from resources.libraries.python.Constants import Constants
  19 from resources.libraries.python.topology import NodeType
  20 from resources.libraries.python.ssh import SSH, exec_cmd_no_error
  21 from resources.libraries.python.CpuUtils import CpuUtils
  22 from resources.libraries.python.VppConfigGenerator import VppConfigGenerator
  23
  24 __all__ = ["KubernetesUtils"]
  25
  26 # Maximum number of retries to check if PODs are running or deleted.
  27 MAX_RETRY = 48
  28
  29 class KubernetesUtils(object):
  30     """Kubernetes utilities class."""
  31
  32     def __init__(self):
  33         """Initialize KubernetesUtils class."""
  34         pass
  35
  36     @staticmethod
  37     def load_docker_image_on_node(node, image_path):
  38         """Load Docker container image from file on node.
  39
  40         :param node: DUT node.
  41         :param image_path: Container image path.
  42         :type node: dict
  43         :type image_path: str
  44         :raises RuntimeError: If loading image failed on node.
  45         """
  46         command = 'docker load -i {image_path}'.\
  47             format(image_path=image_path)
  48         message = 'Failed to load Docker image on {node}.'.\
  49             format(node=node['host'])
  50         exec_cmd_no_error(node, command, timeout=240, sudo=True,
  51                           message=message)
  52
  53         command = "docker rmi $(sudo docker images -f 'dangling=true' -q)".\
  54             format(image_path=image_path)
  55         message = 'Failed to clean Docker images on {node}.'.\
  56             format(node=node['host'])
  57         try:
  58             exec_cmd_no_error(node, command, timeout=240, sudo=True,
  59                               message=message)
  60         except RuntimeError:
  61             pass
  62
  63     @staticmethod
  64     def load_docker_image_on_all_duts(nodes, image_path):
  65         """Load Docker container image from file on all DUTs.
  66
  67         :param nodes: Topology nodes.
  68         :param image_path: Container image path.
  69         :type nodes: dict
  70         :type image_path: str
  71         """
  72         for node in nodes.values():
  73             if node['type'] == NodeType.DUT:
  74                 KubernetesUtils.load_docker_image_on_node(node, image_path)
  75
  76     @staticmethod
  77     def setup_kubernetes_on_node(node):
  78         """Set up Kubernetes on node.
  79
  80         :param node: DUT node.
  81         :type node: dict
  82         :raises RuntimeError: If Kubernetes setup failed on node.
  83         """
  84         ssh = SSH()
  85         ssh.connect(node)
  86
  87         cmd = '{dir}/{lib}/k8s_setup.sh deploy_calico'\
  88             .format(dir=Constants.REMOTE_FW_DIR,
  89                     lib=Constants.RESOURCES_LIB_SH)
  90         (ret_code, _, _) = ssh.exec_command(cmd, timeout=240)
  91         if int(ret_code) != 0:
  92             raise RuntimeError('Failed to setup Kubernetes on {node}.'
  93                                .format(node=node['host']))
  94
  95         KubernetesUtils.wait_for_kubernetes_pods_on_node(node,
  96                                                          nspace='kube-system')
  97
  98     @staticmethod
  99     def setup_kubernetes_on_all_duts(nodes):
 100         """Set up Kubernetes on all DUTs.
 101
 102         :param nodes: Topology nodes.
 103         :type nodes: dict
 104         """
 105         for node in nodes.values():
 106             if node['type'] == NodeType.DUT:
 107                 KubernetesUtils.setup_kubernetes_on_node(node)
 108
 109     @staticmethod
 110     def destroy_kubernetes_on_node(node):
 111         """Destroy Kubernetes on node.
 112
 113         :param node: DUT node.
 114         :type node: dict
 115         :raises RuntimeError: If destroying Kubernetes failed.
 116         """
 117         ssh = SSH()
 118         ssh.connect(node)
 119
 120         cmd = '{dir}/{lib}/k8s_setup.sh destroy'\
 121             .format(dir=Constants.REMOTE_FW_DIR,
 122                     lib=Constants.RESOURCES_LIB_SH)
 123         (ret_code, _, _) = ssh.exec_command(cmd, timeout=120)
 124         if int(ret_code) != 0:
 125             raise RuntimeError('Failed to destroy Kubernetes on {node}.'
 126                                .format(node=node['host']))
 127
 128     @staticmethod
 129     def destroy_kubernetes_on_all_duts(nodes):
 130         """Destroy Kubernetes on all DUTs.
 131
 132         :param nodes: Topology nodes.
 133         :type nodes: dict
 134         """
 135         for node in nodes.values():
 136             if node['type'] == NodeType.DUT:
 137                 KubernetesUtils.destroy_kubernetes_on_node(node)
 138
 139     @staticmethod
 140     def apply_kubernetes_resource_on_node(node, yaml_file, **kwargs):
 141         """Apply Kubernetes resource on node.
 142
 143         :param node: DUT node.
 144         :param yaml_file: YAML configuration file.
 145         :param kwargs: Key-value pairs to replace in YAML template.
 146         :type node: dict
 147         :type yaml_file: str
 148         :type kwargs: dict
 149         :raises RuntimeError: If applying Kubernetes template failed.
 150         """
 151         ssh = SSH()
 152         ssh.connect(node)
 153
 154         fqn_file = '{tpl}/{yaml}'.format(tpl=Constants.RESOURCES_TPL_K8S,
 155                                          yaml=yaml_file)
 156         with open(fqn_file, 'r') as src_file:
 157             stream = src_file.read()
 158             data = reduce(lambda a, kv: a.replace(*kv), kwargs.iteritems(),
 159                           stream)
 160             cmd = 'cat <<EOF | kubectl apply -f - \n{data}\nEOF'.format(
 161                 data=data)
 162             (ret_code, _, _) = ssh.exec_command_sudo(cmd)
 163             if int(ret_code) != 0:
 164                 raise RuntimeError('Failed to apply Kubernetes template {yaml} '
 165                                    'on {node}.'.format(yaml=yaml_file,
 166                                                        node=node['host']))
 167
 168     @staticmethod
 169     def apply_kubernetes_resource_on_all_duts(nodes, yaml_file, **kwargs):
 170         """Apply Kubernetes resource on all DUTs.
 171
 172         :param nodes: Topology nodes.
 173         :param yaml_file: YAML configuration file.
 174         :param kwargs: Key-value pairs to replace in YAML template.
 175         :type nodes: dict
 176         :type yaml_file: str
 177         :type kwargs: dict
 178         """
 179         for node in nodes.values():
 180             if node['type'] == NodeType.DUT:
 181                 KubernetesUtils.apply_kubernetes_resource_on_node(node,
 182                                                                   yaml_file,
 183                                                                   **kwargs)
 184
 185     @staticmethod
 186     def create_kubernetes_cm_from_file_on_node(node, nspace, name, **kwargs):
 187         """Create Kubernetes ConfigMap from file on node.
 188
 189         :param node: DUT node.
 190         :param nspace: Kubernetes namespace.
 191         :param name: ConfigMap name.
 192         :param kwargs: Named parameters.
 193         :type node: dict
 194         :type nspace: str
 195         :type name: str
 196         :param kwargs: dict
 197         :raises RuntimeError: If creating Kubernetes ConfigMap failed.
 198         """
 199         ssh = SSH()
 200         ssh.connect(node)
 201
 202         nspace = '-n {nspace}'.format(nspace=nspace) if nspace else ''
 203
 204         from_file = '{0}'.format(' '.join('--from-file={0}={1} '\
 205             .format(key, kwargs[key]) for key in kwargs))
 206
 207         cmd = 'kubectl create {nspace} configmap {name} {from_file}'\
 208             .format(nspace=nspace, name=name, from_file=from_file)
 209         (ret_code, _, _) = ssh.exec_command_sudo(cmd)
 210         if int(ret_code) != 0:
 211             raise RuntimeError('Failed to create Kubernetes ConfigMap '
 212                                'on {node}.'.format(node=node['host']))
 213
 214     @staticmethod
 215     def create_kubernetes_cm_from_file_on_all_duts(nodes, nspace, name,
 216                                                    **kwargs):
 217         """Create Kubernetes ConfigMap from file on all DUTs.
 218
 219         :param nodes: Topology nodes.
 220         :param nspace: Kubernetes namespace.
 221         :param name: ConfigMap name.
 222         :param kwargs: Named parameters.
 223         :type nodes: dict
 224         :type nspace: str
 225         :type name: str
 226         :param kwargs: dict
 227         """
 228         for node in nodes.values():
 229             if node['type'] == NodeType.DUT:
 230                 KubernetesUtils.create_kubernetes_cm_from_file_on_node(node,
 231                                                                        nspace,
 232                                                                        name,
 233                                                                        **kwargs)
 234
 235     @staticmethod
 236     def delete_kubernetes_resource_on_node(node, nspace, name=None,
 237                                            rtype='po,cm,deploy,rs,rc,svc'):
 238         """Delete Kubernetes resource on node.
 239
 240         :param node: DUT node.
 241         :param nspace: Kubernetes namespace.
 242         :param rtype: Kubernetes resource type.
 243         :param name: Name of resource (Default: all).
 244         :type node: dict
 245         :type nspace: str
 246         :type rtype: str
 247         :type name: str
 248         :raises RuntimeError: If retrieving or deleting Kubernetes resource
 249             failed.
 250         """
 251         ssh = SSH()
 252         ssh.connect(node)
 253
 254         name = '{name}'.format(name=name) if name else '--all'
 255         nspace = '-n {nspace}'.format(nspace=nspace) if nspace else ''
 256
 257         cmd = 'kubectl delete {nspace} {rtype} {name}'\
 258             .format(nspace=nspace, rtype=rtype, name=name)
 259         (ret_code, _, _) = ssh.exec_command_sudo(cmd, timeout=120)
 260         if int(ret_code) != 0:
 261             raise RuntimeError('Failed to delete Kubernetes resources '
 262                                'on {node}.'.format(node=node['host']))
 263
 264         cmd = 'kubectl get {nspace} pods --no-headers'\
 265             .format(nspace=nspace)
 266         for _ in range(MAX_RETRY):
 267             (ret_code, stdout, stderr) = ssh.exec_command_sudo(cmd)
 268             if int(ret_code) != 0:
 269                 raise RuntimeError('Failed to retrieve Kubernetes resources on '
 270                                    '{node}.'.format(node=node['host']))
 271             if name == '--all':
 272                 ready = False
 273                 for line in stderr.splitlines():
 274                     if 'No resources found.' in line:
 275                         ready = True
 276                 if ready:
 277                     break
 278             else:
 279                 ready = False
 280                 for line in stdout.splitlines():
 281                     try:
 282                         state = line.split()[1].split('/')
 283                         ready = True if 'Running' in line and\
 284                             state == state[::-1] else False
 285                         if not ready:
 286                             break
 287                     except (ValueError, IndexError):
 288                         ready = False
 289                 if ready:
 290                     break
 291             sleep(5)
 292         else:
 293             raise RuntimeError('Failed to delete Kubernetes resources on '
 294                                '{node}.'.format(node=node['host']))
 295
 296     @staticmethod
 297     def delete_kubernetes_resource_on_all_duts(nodes, nspace, name=None,
 298                                                rtype='po,cm,deploy,rs,rc,svc'):
 299         """Delete all Kubernetes resource on all DUTs.
 300
 301         :param nodes: Topology nodes.
 302         :param nspace: Kubernetes namespace.
 303         :param rtype: Kubernetes resource type.
 304         :param name: Name of resource.
 305         :type nodes: dict
 306         :type nspace: str
 307         :type rtype: str
 308         :type name: str
 309         """
 310         for node in nodes.values():
 311             if node['type'] == NodeType.DUT:
 312                 KubernetesUtils.delete_kubernetes_resource_on_node(node, nspace,
 313                                                                    name, rtype)
 314
 315     @staticmethod
 316     def describe_kubernetes_resource_on_node(node, nspace):
 317         """Describe all Kubernetes PODs in namespace on node.
 318
 319         :param node: DUT node.
 320         :param nspace: Kubernetes namespace.
 321         :type node: dict
 322         :type nspace: str
 323         """
 324         ssh = SSH()
 325         ssh.connect(node)
 326
 327         nspace = '-n {nspace}'.format(nspace=nspace) if nspace else ''
 328
 329         cmd = 'kubectl describe {nspace} all'.format(nspace=nspace)
 330         ssh.exec_command_sudo(cmd)
 331
 332     @staticmethod
 333     def describe_kubernetes_resource_on_all_duts(nodes, nspace):
 334         """Describe all Kubernetes PODs in namespace on all DUTs.
 335
 336         :param nodes: Topology nodes.
 337         :param nspace: Kubernetes namespace.
 338         :type nodes: dict
 339         :type nspace: str
 340         """
 341         for node in nodes.values():
 342             if node['type'] == NodeType.DUT:
 343                 KubernetesUtils.describe_kubernetes_resource_on_node(node,
 344                                                                      nspace)
 345
 346     @staticmethod
 347     def get_kubernetes_logs_on_node(node, nspace):
 348         """Get Kubernetes logs from all PODs in namespace on node.
 349
 350         :param node: DUT node.
 351         :param nspace: Kubernetes namespace.
 352         :type node: dict
 353         :type nspace: str
 354         """
 355         ssh = SSH()
 356         ssh.connect(node)
 357
 358         nspace = '-n {nspace}'.format(nspace=nspace) if nspace else ''
 359
 360         cmd = "for p in $(kubectl get pods {nspace} -o jsonpath="\
 361             "'{{.items[*].metadata.name}}'); do echo $p; kubectl logs "\
 362             "{nspace} $p; done".format(nspace=nspace)
 363         ssh.exec_command(cmd)
 364
 365         cmd = "kubectl exec {nspace} etcdv3 -- etcdctl --endpoints "\
 366             "\"localhost:22379\" get \"/\" --prefix=true".format(nspace=nspace)
 367         ssh.exec_command(cmd)
 368
 369     @staticmethod
 370     def get_kubernetes_logs_on_all_duts(nodes, nspace):
 371         """Get Kubernetes logs from all PODs in namespace on all DUTs.
 372
 373         :param nodes: Topology nodes.
 374         :param nspace: Kubernetes namespace.
 375         :type nodes: dict
 376         :type nspace: str
 377         """
 378         for node in nodes.values():
 379             if node['type'] == NodeType.DUT:
 380                 KubernetesUtils.get_kubernetes_logs_on_node(node, nspace)
 381
 382     @staticmethod
 383     def wait_for_kubernetes_pods_on_node(node, nspace):
 384         """Wait for Kubernetes PODs to become ready on node.
 385
 386         :param node: DUT node.
 387         :param nspace: Kubernetes namespace.
 388         :type node: dict
 389         :type nspace: str
 390         :raises RuntimeError: If Kubernetes PODs are not in Running state.
 391         """
 392         ssh = SSH()
 393         ssh.connect(node)
 394
 395         nspace = '-n {nspace}'.format(nspace=nspace) if nspace \
 396             else '--all-namespaces'
 397
 398         cmd = 'kubectl get {nspace} pods --no-headers' \
 399             .format(nspace=nspace)
 400         for _ in range(MAX_RETRY):
 401             (ret_code, stdout, _) = ssh.exec_command_sudo(cmd)
 402             if int(ret_code) == 0:
 403                 ready = False
 404                 for line in stdout.splitlines():
 405                     try:
 406                         state = line.split()[1].split('/')
 407                         ready = True if 'Running' in line and \
 408                             state == state[::-1] else False
 409                         if not ready:
 410                             break
 411                     except (ValueError, IndexError):
 412                         ready = False
 413                 if ready:
 414                     break
 415             sleep(5)
 416         else:
 417             raise RuntimeError('Kubernetes PODs are not running on {node}.'
 418                                .format(node=node['host']))
 419
 420     @staticmethod
 421     def wait_for_kubernetes_pods_on_all_duts(nodes, nspace):
 422         """Wait for Kubernetes to become ready on all DUTs.
 423
 424         :param nodes: Topology nodes.
 425         :param nspace: Kubernetes namespace.
 426         :type nodes: dict
 427         :type nspace: str
 428         """
 429         for node in nodes.values():
 430             if node['type'] == NodeType.DUT:
 431                 KubernetesUtils.wait_for_kubernetes_pods_on_node(node, nspace)
 432
 433     @staticmethod
 434     def set_kubernetes_pods_affinity_on_node(node):
 435         """Set affinity for all Kubernetes PODs except VPP on node.
 436
 437         :param node: DUT node.
 438         :type node: dict
 439         """
 440         ssh = SSH()
 441         ssh.connect(node)
 442
 443         cmd = '{dir}/{lib}/k8s_setup.sh affinity_non_vpp'\
 444             .format(dir=Constants.REMOTE_FW_DIR,
 445                     lib=Constants.RESOURCES_LIB_SH)
 446         ssh.exec_command(cmd)
 447
 448     @staticmethod
 449     def set_kubernetes_pods_affinity_on_all_duts(nodes):
 450         """Set affinity for all Kubernetes PODs except VPP on all DUTs.
 451
 452         :param nodes: Topology nodes.
 453         :type nodes: dict
 454         """
 455         for node in nodes.values():
 456             if node['type'] == NodeType.DUT:
 457                 KubernetesUtils.set_kubernetes_pods_affinity_on_node(node)
 458
 459     @staticmethod
 460     def create_kubernetes_vswitch_startup_config(**kwargs):
 461         """Create Kubernetes VSWITCH startup configuration.
 462
 463         :param kwargs: Key-value pairs used to create configuration.
 464         :param kwargs: dict
 465         """
 466         smt_used = CpuUtils.is_smt_enabled(kwargs['node']['cpuinfo'])
 467
 468         cpuset_cpus = \
 469             CpuUtils.cpu_slice_of_list_per_node(node=kwargs['node'],
 470                                                 cpu_node=kwargs['cpu_node'],
 471                                                 skip_cnt=2,
 472                                                 cpu_cnt=kwargs['phy_cores'],
 473                                                 smt_used=smt_used)
 474         cpuset_main = \
 475             CpuUtils.cpu_slice_of_list_per_node(node=kwargs['node'],
 476                                                 cpu_node=kwargs['cpu_node'],
 477                                                 skip_cnt=1,
 478                                                 cpu_cnt=1,
 479                                                 smt_used=smt_used)
 480
 481         # Create config instance
 482         vpp_config = VppConfigGenerator()
 483         vpp_config.set_node(kwargs['node'])
 484         vpp_config.add_unix_cli_listen(value='0.0.0.0:5002')
 485         vpp_config.add_unix_nodaemon()
 486         vpp_config.add_heapsize('4G')
 487         vpp_config.add_ip_heap_size('4G')
 488         vpp_config.add_ip6_heap_size('4G')
 489         vpp_config.add_ip6_hash_buckets('2000000')
 490         if not kwargs['jumbo']:
 491             vpp_config.add_dpdk_no_multi_seg()
 492         vpp_config.add_dpdk_no_tx_checksum_offload()
 493         vpp_config.add_dpdk_dev_default_rxq(kwargs['rxq_count_int'])
 494         vpp_config.add_dpdk_dev(kwargs['if1'], kwargs['if2'])
 495         vpp_config.add_buffers_per_numa(kwargs['buffers_per_numa'])
 496         # We will pop first core from list to be main core
 497         vpp_config.add_cpu_main_core(str(cpuset_main.pop(0)))
 498         # if this is not only core in list, the rest will be used as workers.
 499         if cpuset_cpus:
 500             corelist_workers = ','.join(str(cpu) for cpu in cpuset_cpus)
 501             vpp_config.add_cpu_corelist_workers(corelist_workers)
 502         vpp_config.write_config(filename=kwargs['filename'])
 503
 504     @staticmethod
 505     def create_kubernetes_vnf_startup_config(**kwargs):
 506         """Create Kubernetes VNF startup configuration.
 507
 508         :param kwargs: Key-value pairs used to create configuration.
 509         :param kwargs: dict
 510         """
 511         smt_used = CpuUtils.is_smt_enabled(kwargs['node']['cpuinfo'])
 512         skip_cnt = kwargs['cpu_skip'] + (kwargs['i'] - 1) * \
 513             (kwargs['phy_cores'] - 1)
 514         cpuset_cpus = \
 515             CpuUtils.cpu_slice_of_list_per_node(node=kwargs['node'],
 516                                                 cpu_node=kwargs['cpu_node'],
 517                                                 skip_cnt=skip_cnt,
 518                                                 cpu_cnt=kwargs['phy_cores']-1,
 519                                                 smt_used=smt_used)
 520         cpuset_main = \
 521             CpuUtils.cpu_slice_of_list_per_node(node=kwargs['node'],
 522                                                 cpu_node=kwargs['cpu_node'],
 523                                                 skip_cnt=1,
 524                                                 cpu_cnt=1,
 525                                                 smt_used=smt_used)
 526         # Create config instance
 527         vpp_config = VppConfigGenerator()
 528         vpp_config.set_node(kwargs['node'])
 529         vpp_config.add_unix_cli_listen(value='0.0.0.0:5002')
 530         vpp_config.add_unix_nodaemon()
 531         # We will pop first core from list to be main core
 532         vpp_config.add_cpu_main_core(str(cpuset_main.pop(0)))
 533         # if this is not only core in list, the rest will be used as workers.
 534         if cpuset_cpus:
 535             corelist_workers = ','.join(str(cpu) for cpu in cpuset_cpus)
 536             vpp_config.add_cpu_corelist_workers(corelist_workers)
 537         vpp_config.add_plugin('disable', 'dpdk_plugin.so')
 538         vpp_config.write_config(filename=kwargs['filename'])