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