1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from openstack.cloud import openstackcloud
class AcceleratorCloudMixin(openstackcloud._OpenStackCloudMixin):
def list_deployables(self, filters=None):
"""List all available deployables.
:param filters: (optional) dict of filter conditions to push down
:returns: A list of accelerator ``Deployable`` objects.
"""
# Translate None from search interface to empty {} for kwargs below
if not filters:
filters = {}
return list(self.accelerator.deployables(**filters))
def list_devices(self, filters=None):
"""List all devices.
:param filters: (optional) dict of filter conditions to push down
:returns: A list of accelerator ``Device`` objects.
"""
# Translate None from search interface to empty {} for kwargs below
if not filters:
filters = {}
return list(self.accelerator.devices(**filters))
def list_device_profiles(self, filters=None):
"""List all device_profiles.
:param filters: (optional) dict of filter conditions to push down
:returns: A list of accelerator ``DeviceProfile`` objects.
"""
# Translate None from search interface to empty {} for kwargs below
if not filters:
filters = {}
return list(self.accelerator.device_profiles(**filters))
def create_device_profile(self, attrs):
"""Create a device_profile.
:param attrs: The info of device_profile to be created.
:returns: An accelerator ``DeviceProfile`` objects.
"""
return self.accelerator.create_device_profile(**attrs)
def delete_device_profile(self, name_or_id, filters):
"""Delete a device_profile.
:param name_or_id: The name or uuid of the device profile to be
deleted.
:param filters: dict of filter conditions to push down
:returns: True if delete succeeded, False otherwise.
"""
device_profile = self.accelerator.get_device_profile(
name_or_id,
filters,
)
if device_profile is None:
self.log.debug(
"device_profile %s not found for deleting",
name_or_id,
)
return False
self.accelerator.delete_device_profile(device_profile=device_profile)
return True
def list_accelerator_requests(self, filters=None):
"""List all accelerator_requests.
:param filters: (optional) dict of filter conditions to push down
:returns: A list of accelerator ``AcceleratorRequest`` objects.
"""
# Translate None from search interface to empty {} for kwargs below
if not filters:
filters = {}
return list(self.accelerator.accelerator_requests(**filters))
def delete_accelerator_request(self, name_or_id, filters):
"""Delete a accelerator_request.
:param name_or_id: The name or UUID of the accelerator request to
be deleted.
:param filters: dict of filter conditions to push down
:returns: True if delete succeeded, False otherwise.
"""
accelerator_request = self.accelerator.get_accelerator_request(
name_or_id,
filters,
)
if accelerator_request is None:
self.log.debug(
"accelerator_request %s not found for deleting",
name_or_id,
)
return False
self.accelerator.delete_accelerator_request(
accelerator_request=accelerator_request,
)
return True
def create_accelerator_request(self, attrs):
"""Create an accelerator_request.
:param attrs: The info of accelerator_request to be created.
:returns: An accelerator ``AcceleratorRequest`` object.
"""
return self.accelerator.create_accelerator_request(**attrs)
def bind_accelerator_request(self, uuid, properties):
"""Bind an accelerator to VM.
:param uuid: The uuid of the accelerator_request to be binded.
:param properties: The info of VM that will bind the accelerator.
:returns: True if bind succeeded, False otherwise.
"""
accelerator_request = self.accelerator.get_accelerator_request(uuid)
if accelerator_request is None:
self.log.debug(
"accelerator_request %s not found for unbinding", uuid
)
return False
return self.accelerator.update_accelerator_request(uuid, properties)
def unbind_accelerator_request(self, uuid, properties):
"""Unbind an accelerator from VM.
:param uuid: The uuid of the accelerator_request to be unbinded.
:param properties: The info of VM that will unbind the accelerator.
:returns: True if unbind succeeded, False otherwise.
"""
accelerator_request = self.accelerator.get_accelerator_request(uuid)
if accelerator_request is None:
self.log.debug(
"accelerator_request %s not found for unbinding", uuid
)
return False
return self.accelerator.update_accelerator_request(uuid, properties)
|
