yardstick/orchestrator/kubernetes.py

   1 ##############################################################################
   2 # Copyright (c) 2017 Huawei Technologies Co.,Ltd.
   3 #
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9
  10 import copy
  11 import re
  12
  13 from oslo_serialization import jsonutils
  14
  15 from yardstick.common import constants
  16 from yardstick.common import exceptions
  17 from yardstick.common import kubernetes_utils as k8s_utils
  18 from yardstick.common import utils
  19
  20
  21 class ContainerObject(object):
  22
  23     SSH_MOUNT_PATH = '/tmp/.ssh/'
  24     IMAGE_DEFAULT = 'openretriever/yardstick'
  25     COMMAND_DEFAULT = '/bin/bash'
  26     RESOURCES = ('requests', 'limits')
  27     PORT_OPTIONS = ('containerPort', 'hostIP', 'hostPort', 'name', 'protocol')
  28
  29     def __init__(self, name, ssh_key, **kwargs):
  30         self._name = name
  31         self._ssh_key = ssh_key
  32         self._image = kwargs.get('image', self.IMAGE_DEFAULT)
  33         self._command = [kwargs.get('command', self.COMMAND_DEFAULT)]
  34         self._args = kwargs.get('args', [])
  35         self._volume_mounts = kwargs.get('volumeMounts', [])
  36         self._security_context = kwargs.get('securityContext')
  37         self._env = kwargs.get('env', [])
  38         self._resources = kwargs.get('resources', {})
  39         self._ports = kwargs.get('ports', [])
  40
  41     def _create_volume_mounts(self):
  42         """Return all "volumeMounts" items per container"""
  43         volume_mounts_items = [self._create_volume_mounts_item(vol)
  44                                for vol in self._volume_mounts]
  45         ssh_vol = {'name': self._ssh_key,
  46                    'mountPath': self.SSH_MOUNT_PATH}
  47         volume_mounts_items.append(self._create_volume_mounts_item(ssh_vol))
  48         return volume_mounts_items
  49
  50     @staticmethod
  51     def _create_volume_mounts_item(volume_mount):
  52         """Create a "volumeMounts" item"""
  53         return {'name': volume_mount['name'],
  54                 'mountPath': volume_mount['mountPath'],
  55                 'readOnly': volume_mount.get('readOnly', False)}
  56
  57     def get_container_item(self):
  58         """Create a "container" item"""
  59         container_name = '{}-container'.format(self._name)
  60         container = {'args': self._args,
  61                      'command': self._command,
  62                      'image': self._image,
  63                      'name': container_name,
  64                      'volumeMounts': self._create_volume_mounts()}
  65         if self._security_context:
  66             container['securityContext'] = self._security_context
  67         if self._env:
  68             container['env'] = []
  69             for env in self._env:
  70                 container['env'].append({'name': env['name'],
  71                                          'value': env['value']})
  72         if self._ports:
  73             container['ports'] = []
  74             for port in self._ports:
  75                 if 'containerPort' not in port.keys():
  76                     raise exceptions.KubernetesContainerPortNotDefined(
  77                         port=port)
  78                 _port = {port_option: value for port_option, value
  79                          in port.items() if port_option in self.PORT_OPTIONS}
  80                 container['ports'].append(_port)
  81         if self._resources:
  82             container['resources'] = {}
  83             for res in (res for res in self._resources if
  84                         res in self.RESOURCES):
  85                 container['resources'][res] = self._resources[res]
  86         return container
  87
  88
  89 class ReplicationControllerObject(object):
  90
  91     SSHKEY_DEFAULT = 'yardstick_key'
  92     RESTART_POLICY = ('Always', 'OnFailure', 'Never')
  93     TOLERATIONS_KEYS = ('key', 'value', 'effect', 'operator')
  94
  95     def __init__(self, name, **kwargs):
  96         super(ReplicationControllerObject, self).__init__()
  97         parameters = copy.deepcopy(kwargs)
  98         self.name = name
  99         self.node_selector = parameters.pop('nodeSelector', {})
 100         self.ssh_key = parameters.pop('ssh_key', self.SSHKEY_DEFAULT)
 101         self._volumes = parameters.pop('volumes', [])
 102         self._security_context = parameters.pop('securityContext', None)
 103         self._networks = parameters.pop('networks', [])
 104         self._tolerations = parameters.pop('tolerations', [])
 105         self._restart_policy = parameters.pop('restartPolicy', 'Always')
 106         if self._restart_policy not in self.RESTART_POLICY:
 107             raise exceptions.KubernetesWrongRestartPolicy(
 108                 rpolicy=self._restart_policy)
 109
 110         containers = parameters.pop('containers', None)
 111         if containers:
 112             self._containers = [
 113                 ContainerObject(self.name, self.ssh_key, **container)
 114                 for container in containers]
 115         else:
 116             self._containers = [
 117                 ContainerObject(self.name, self.ssh_key, **parameters)]
 118
 119         self.template = {
 120             "apiVersion": "v1",
 121             "kind": "ReplicationController",
 122             "metadata": {
 123                 "name": ""
 124             },
 125             "spec": {
 126                 "replicas": 1,
 127                 "template": {
 128                     "metadata": {
 129                         "labels": {"app": name}
 130                     },
 131                     "spec": {
 132                         "containers": [],
 133                         "volumes": [],
 134                         "nodeSelector": {},
 135                         "restartPolicy": self._restart_policy,
 136                         "tolerations": []
 137                     }
 138                 }
 139             }
 140         }
 141
 142         self._change_value_according_name(name)
 143         self._add_containers()
 144         self._add_node_selector()
 145         self._add_volumes()
 146         self._add_security_context()
 147         self._add_networks()
 148         self._add_tolerations()
 149
 150     @property
 151     def networks(self):
 152         return self._networks
 153
 154     def get_template(self):
 155         return self.template
 156
 157     def _change_value_according_name(self, name):
 158         utils.set_dict_value(self.template, 'metadata.name', name)
 159
 160         utils.set_dict_value(self.template,
 161                              'spec.template.metadata.labels.app',
 162                              name)
 163
 164     def _add_containers(self):
 165         containers = [container.get_container_item()
 166                       for container in self._containers]
 167         utils.set_dict_value(self.template,
 168                              'spec.template.spec.containers',
 169                              containers)
 170
 171     def _add_node_selector(self):
 172         utils.set_dict_value(self.template,
 173                              'spec.template.spec.nodeSelector',
 174                              self.node_selector)
 175
 176     def _add_volumes(self):
 177         """Add "volume" items to container specs, including the SSH one"""
 178         volume_items = [self._create_volume_item(vol) for vol in self._volumes]
 179         volume_items.append(self._create_ssh_key_volume())
 180         utils.set_dict_value(self.template,
 181                              'spec.template.spec.volumes',
 182                              volume_items)
 183
 184     def _create_ssh_key_volume(self):
 185         """Create a "volume" item of type "configMap" for the SSH key"""
 186         return {'name': self.ssh_key,
 187                 'configMap': {'name': self.ssh_key}}
 188
 189     @staticmethod
 190     def _create_volume_item(volume):
 191         """Create a "volume" item"""
 192         volume = copy.deepcopy(volume)
 193         name = volume.pop('name')
 194         for key in (k for k in volume if k in k8s_utils.get_volume_types()):
 195             type_name = key
 196             type_data = volume[key]
 197             break
 198         else:
 199             raise exceptions.KubernetesTemplateInvalidVolumeType(volume=volume)
 200
 201         return {'name': name,
 202                 type_name: type_data}
 203
 204     def _add_security_context(self):
 205         if self._security_context:
 206             utils.set_dict_value(self.template,
 207                                  'spec.template.spec.securityContext',
 208                                  self._security_context)
 209
 210     def _add_networks(self):
 211         networks = []
 212         for net in self._networks:
 213             networks.append({'name': net})
 214
 215         if not networks:
 216             return
 217
 218         annotations = {'networks': jsonutils.dumps(networks)}
 219         utils.set_dict_value(self.template,
 220                              'spec.template.metadata.annotations',
 221                              annotations)
 222
 223     def _add_tolerations(self):
 224         tolerations = []
 225         for tol in self._tolerations:
 226             tolerations.append({k: tol[k] for k in tol
 227                                 if k in self.TOLERATIONS_KEYS})
 228
 229         tolerations = ([{'operator': 'Exists'}] if not tolerations
 230                        else tolerations)
 231         utils.set_dict_value(self.template,
 232                              'spec.template.spec.tolerations',
 233                              tolerations)
 234
 235
 236 class ServiceNodePortObject(object):
 237
 238     MANDATORY_PARAMETERS = {'port', 'name'}
 239     NAME_REGEX = re.compile(r'^[a-z0-9]([-a-z0-9]*[a-z0-9])?$')
 240
 241     def __init__(self, name, **kwargs):
 242         """Service kind "NodePort" object
 243
 244         :param name: (string) name of the Service
 245         :param kwargs: (dict) node_ports -> (list) port, name, targetPort,
 246                                                    nodePort
 247         """
 248         self._name = '{}-service'.format(name)
 249         self.template = {
 250             'metadata': {'name': '{}-service'.format(name)},
 251             'spec': {
 252                 'type': 'NodePort',
 253                 'ports': [],
 254                 'selector': {'app': name}
 255             }
 256         }
 257
 258         self._add_port(22, 'ssh', protocol='TCP')
 259         node_ports = copy.deepcopy(kwargs.get('node_ports', []))
 260         for port in node_ports:
 261             if not self.MANDATORY_PARAMETERS.issubset(port.keys()):
 262                 missing_parameters = ', '.join(
 263                     str(param) for param in
 264                     (self.MANDATORY_PARAMETERS - set(port.keys())))
 265                 raise exceptions.KubernetesServiceObjectDefinitionError(
 266                     missing_parameters=missing_parameters)
 267             port_number = port.pop('port')
 268             name = port.pop('name')
 269             if not self.NAME_REGEX.match(name):
 270                 raise exceptions.KubernetesServiceObjectNameError(name=name)
 271             self._add_port(port_number, name, **port)
 272
 273     def _add_port(self, port, name, protocol=None, targetPort=None,
 274                   nodePort=None):
 275         _port = {'port': port,
 276                  'name': name}
 277         if protocol:
 278             _port['protocol'] = protocol
 279         if targetPort:
 280             _port['targetPort'] = targetPort
 281         if nodePort:
 282             _port['nodePort'] = nodePort
 283         self.template['spec']['ports'].append(_port)
 284
 285     def create(self):
 286         k8s_utils.create_service(self.template)
 287
 288     def delete(self):
 289         k8s_utils.delete_service(self._name)
 290
 291
 292 class CustomResourceDefinitionObject(object):
 293
 294     MANDATORY_PARAMETERS = {'name'}
 295
 296     def __init__(self, ctx_name, **kwargs):
 297         if not self.MANDATORY_PARAMETERS.issubset(kwargs):
 298             missing_parameters = ', '.join(
 299                 str(param) for param in
 300                 (self.MANDATORY_PARAMETERS - set(kwargs)))
 301             raise exceptions.KubernetesCRDObjectDefinitionError(
 302                 missing_parameters=missing_parameters)
 303
 304         singular = kwargs['name']
 305         plural = singular + 's'
 306         kind = singular.title()
 307         version = kwargs.get('version', 'v1')
 308         scope = kwargs.get('scope', constants.SCOPE_NAMESPACED)
 309         group = ctx_name + '.com'
 310         self._name = metadata_name = plural + '.' + group
 311
 312         self._template = {
 313             'metadata': {
 314                 'name': metadata_name
 315             },
 316             'spec': {
 317                 'group': group,
 318                 'version': version,
 319                 'scope': scope,
 320                 'names': {'plural': plural,
 321                           'singular': singular,
 322                           'kind': kind}
 323             }
 324         }
 325
 326     def create(self):
 327         k8s_utils.create_custom_resource_definition(self._template)
 328
 329     def delete(self):
 330         k8s_utils.delete_custom_resource_definition(self._name)
 331
 332
 333 class NetworkObject(object):
 334
 335     MANDATORY_PARAMETERS = {'plugin', 'args'}
 336     KIND = 'Network'
 337
 338     def __init__(self, name, **kwargs):
 339         if not self.MANDATORY_PARAMETERS.issubset(kwargs):
 340             missing_parameters = ', '.join(
 341                 str(param) for param in
 342                 (self.MANDATORY_PARAMETERS - set(kwargs)))
 343             raise exceptions.KubernetesNetworkObjectDefinitionError(
 344                 missing_parameters=missing_parameters)
 345
 346         self._name = name
 347         self._plugin = kwargs['plugin']
 348         self._args = kwargs['args']
 349         self._crd = None
 350         self._template = None
 351         self._group = None
 352         self._version = None
 353         self._plural = None
 354         self._scope = None
 355
 356     @property
 357     def crd(self):
 358         if self._crd:
 359             return self._crd
 360         crd = k8s_utils.get_custom_resource_definition(self.KIND)
 361         if not crd:
 362             raise exceptions.KubernetesNetworkObjectKindMissing()
 363         self._crd = crd
 364         return self._crd
 365
 366     @property
 367     def group(self):
 368         if self._group:
 369             return self._group
 370         self._group = self.crd.spec.group
 371         return self._group
 372
 373     @property
 374     def version(self):
 375         if self._version:
 376             return self._version
 377         self._version = self.crd.spec.version
 378         return self._version
 379
 380     @property
 381     def plural(self):
 382         if self._plural:
 383             return self._plural
 384         self._plural = self.crd.spec.names.plural
 385         return self._plural
 386
 387     @property
 388     def scope(self):
 389         if self._scope:
 390             return self._scope
 391         self._scope = self.crd.spec.scope
 392         return self._scope
 393
 394     @property
 395     def template(self):
 396         """"Network" object template
 397
 398         This template can be rendered only once the CRD "Network" is created in
 399         Kubernetes. This function call must be delayed until the creation of
 400         the CRD "Network".
 401         """
 402         if self._template:
 403             return self._template
 404
 405         self._template = {
 406             'apiVersion': '{}/{}'.format(self.group, self.version),
 407             'kind': self.KIND,
 408             'metadata': {
 409                 'name': self._name
 410             },
 411             'plugin': self._plugin,
 412             'args': self._args
 413         }
 414         return self._template
 415
 416     def create(self):
 417         k8s_utils.create_network(self.scope, self.group, self.version,
 418                                  self.plural, self.template)
 419
 420     def delete(self):
 421         k8s_utils.delete_network(self.scope, self.group, self.version,
 422                                  self.plural, self._name)
 423
 424
 425 class KubernetesTemplate(object):
 426
 427     def __init__(self, name, context_cfg):
 428         """KubernetesTemplate object initialization
 429
 430         :param name: (str) name of the Kubernetes context
 431         :param context_cfg: (dict) context definition
 432         """
 433         context_cfg = copy.deepcopy(context_cfg)
 434         servers_cfg = context_cfg.pop('servers', {})
 435         crd_cfg = context_cfg.pop('custom_resources', [])
 436         networks_cfg = context_cfg.pop('networks', {})
 437         self.name = name
 438         self.ssh_key = '{}-key'.format(name)
 439
 440         self.rcs = {self._get_rc_name(rc): cfg
 441                     for rc, cfg in servers_cfg.items()}
 442         self.rc_objs = [ReplicationControllerObject(
 443             rc, ssh_key=self.ssh_key, **cfg) for rc, cfg in self.rcs.items()]
 444         self.service_objs = [ServiceNodePortObject(rc, **cfg)
 445                              for rc, cfg in self.rcs.items()]
 446         self.crd = [CustomResourceDefinitionObject(self.name, **crd)
 447                     for crd in crd_cfg]
 448         self.network_objs = [NetworkObject(net_name, **net_data)
 449                              for net_name, net_data in networks_cfg.items()]
 450         self.pods = []
 451
 452     def _get_rc_name(self, rc_name):
 453         return '{}-{}'.format(rc_name, self.name)
 454
 455     def get_rc_pods(self):
 456         resp = k8s_utils.get_pod_list()
 457         self.pods = [p.metadata.name for p in resp.items for s in self.rcs
 458                      if p.metadata.name.startswith(s)]
 459
 460         return self.pods
 461
 462     def get_rc_by_name(self, rc_name):
 463         """Returns a ``ReplicationControllerObject``, searching by name"""
 464         for rc in (rc for rc in self.rc_objs if rc.name == rc_name):
 465             return rc