1 ##############################################################################
2 # Copyright (c) 2017 Huawei Technologies Co.,Ltd.
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 ##############################################################################
11 from kubernetes import client
12 from kubernetes import config
13 from kubernetes.client.rest import ApiException
15 from yardstick.common import constants as consts
17 LOG = logging.getLogger(__name__)
18 LOG.setLevel(logging.DEBUG)
21 def get_core_api(): # pragma: no cover
23 config.load_kube_config(config_file=consts.K8S_CONF_FILE)
25 LOG.exception('config file not found')
28 return client.CoreV1Api()
31 def get_node_list(**kwargs): # pragma: no cover
32 core_v1_api = get_core_api()
34 return core_v1_api.list_node(**kwargs)
36 LOG.exception('Get node list failed')
40 def create_service(template,
43 **kwargs): # pragma: no cover
44 # pylint: disable=unused-argument
45 core_v1_api = get_core_api()
46 metadata = client.V1ObjectMeta(**template.get('metadata', {}))
48 ports = [client.V1ServicePort(**port) for port in
49 template.get('spec', {}).get('ports', [])]
50 template['spec']['ports'] = ports
51 spec = client.V1ServiceSpec(**template.get('spec', {}))
53 service = client.V1Service(metadata=metadata, spec=spec)
56 core_v1_api.create_namespaced_service('default', service)
58 LOG.exception('Create Service failed')
62 def delete_service(name,
64 **kwargs): # pragma: no cover
65 core_v1_api = get_core_api()
67 body = client.V1DeleteOptions()
68 core_v1_api.delete_namespaced_service(name, namespace, body, **kwargs)
70 LOG.exception('Delete Service failed')
73 def get_service_list(namespace='default', **kwargs):
74 core_v1_api = get_core_api()
76 return core_v1_api.list_namespaced_service(namespace, **kwargs)
78 LOG.exception('Get Service list failed')
82 def get_service_by_name(name): # pragma: no cover
83 service_list = get_service_list()
84 return next((s.spec for s in service_list.items if s.metadata.name == name), None)
87 def create_replication_controller(template,
90 **kwargs): # pragma: no cover
91 # pylint: disable=unused-argument
92 core_v1_api = get_core_api()
94 core_v1_api.create_namespaced_replication_controller(namespace,
98 LOG.exception('Create replication controller failed')
102 def delete_replication_controller(name,
105 **kwargs): # pragma: no cover
106 # pylint: disable=unused-argument
107 core_v1_api = get_core_api()
108 body = kwargs.get('body', client.V1DeleteOptions())
109 kwargs.pop('body', None)
111 core_v1_api.delete_namespaced_replication_controller(name,
116 LOG.exception('Delete replication controller failed')
123 **kwargs): # pragma: no cover
124 # pylint: disable=unused-argument
125 core_v1_api = get_core_api()
126 body = kwargs.get('body', client.V1DeleteOptions())
127 kwargs.pop('body', None)
129 core_v1_api.delete_namespaced_pod(name,
134 LOG.exception('Delete pod failed')
140 **kwargs): # pragma: no cover
141 core_v1_api = get_core_api()
143 resp = core_v1_api.read_namespaced_pod(name, namespace, **kwargs)
145 LOG.exception('Read pod failed')
151 def read_pod_status(name, namespace='default', **kwargs): # pragma: no cover
152 # pylint: disable=unused-argument
153 return read_pod(name).status.phase
156 def create_config_map(name,
160 **kwargs): # pragma: no cover
161 # pylint: disable=unused-argument
162 core_v1_api = get_core_api()
163 metadata = client.V1ObjectMeta(name=name)
164 body = client.V1ConfigMap(data=data, metadata=metadata)
166 core_v1_api.create_namespaced_config_map(namespace, body, **kwargs)
168 LOG.exception('Create config map failed')
172 def delete_config_map(name,
175 **kwargs): # pragma: no cover
176 # pylint: disable=unused-argument
177 core_v1_api = get_core_api()
178 body = kwargs.get('body', client.V1DeleteOptions())
179 kwargs.pop('body', None)
181 core_v1_api.delete_namespaced_config_map(name,
186 LOG.exception('Delete config map failed')
190 def get_pod_list(namespace='default'): # pragma: no cover
191 core_v1_api = get_core_api()
193 return core_v1_api.list_namespaced_pod(namespace=namespace)
195 LOG.exception('Get pod list failed')
199 def get_pod_by_name(name): # pragma: no cover
200 pod_list = get_pod_list()
201 return next((n for n in pod_list.items if n.metadata.name.startswith(name)), None)
204 def get_volume_types():
205 """Return the "volume" types supported by the current API"""
206 return [vtype for vtype in client.V1Volume.attribute_map.values()