"""Openstack cloud type."""
import base64
import openstack
from pycloudlib.cloud import BaseCloud
from pycloudlib.config import ConfigFile
from pycloudlib.openstack.instance import OpenstackInstance
[docs]class Openstack(BaseCloud):
"""Openstack cloud class."""
_type = 'openstack'
[docs] def __init__(
self, tag, timestamp_suffix=True, config_file: ConfigFile = None, *,
network=None
):
"""Initialize the connection to openstack.
Requires valid pre-configured environment variables or clouds.yaml.
See https://docs.openstack.org/python-openstackclient/pike/configuration/index.html
Args:
tag: Name of instance
timestamp_suffix: bool set True to append a timestamp suffix to the
tag
config_file: path to pycloudlib configuration file
network: Name of the network to use (from openstack network list)
""" # noqa: E501
super().__init__(tag, timestamp_suffix, config_file)
self.network = network or self.config['network']
self._openstack_keypair = None
self.conn = openstack.connect()
[docs] def delete_image(self, image_id):
"""Delete an image.
Args:
image_id: string, id of the image to delete
"""
self.conn.delete_image(image_id, wait=True)
[docs] def released_image(self, release, **kwargs):
"""Not supported for openstack."""
raise Exception(
'Obtaining released image for a release is not supported on '
'Openstack because we have no guarantee of what images will be '
'available for any particular openstack setup.'
)
[docs] def daily_image(self, release, **kwargs):
"""Not supported for openstack."""
raise Exception(
'Obtaining daily image for a release is not supported on '
'Openstack because we have no guarantee of what images will be '
'available for any particular openstack setup.'
)
[docs] def image_serial(self, image_id):
"""Find the image serial of the latest daily image for a particular release.
Args:
image_id: string, Ubuntu image id
Returns:
string, serial of latest image
"""
raise NotImplementedError
[docs] def get_instance(self, instance_id) -> OpenstackInstance:
"""Get an instance by id.
Args:
instance_id: ID of instance to get
Returns:
An instance object to use to manipulate the instance further.
"""
return OpenstackInstance(
key_pair=self.key_pair,
instance_id=instance_id,
network_id=self._get_network_id()
)
def _get_network_id(self):
try:
return self.conn.network.find_network(self.network).id
except AttributeError as e:
raise Exception(
'No network found named {}'.format(self.network)
) from e
[docs] def launch(self, image_id, instance_type='m1.small', user_data='',
wait=True, **kwargs) -> OpenstackInstance:
"""Launch an instance.
Args:
image_id: string, image ID to use for the instance
instance_type: string, type (flavor) of instance to create
user_data: used by cloud-init to run custom scripts/configuration
wait: wait for instance to be live
**kwargs: dictionary of other arguments to pass to launch
Returns:
An instance object to use to manipulate the instance further.
"""
network_id = self._get_network_id()
networks = [{'uuid': network_id}]
if not self._openstack_keypair:
self._openstack_keypair = self._get_openstack_keypair()
if user_data:
user_data = base64.b64encode(user_data.encode()).decode()
else:
user_data = ''
flavor = self.conn.compute.find_flavor(instance_type)
if flavor is None:
raise Exception(
'No Openstack flavor found named {}. Please pass a valid '
'Openstack flavor as the `instance_type` when calling '
'launch.'.format(instance_type)
)
instance = self.conn.compute.create_server(
name=self.tag,
image_id=image_id,
flavor_id=flavor.id,
networks=networks,
key_name=self._openstack_keypair.name,
user_data=user_data,
wait=wait,
**kwargs,
)
instance = OpenstackInstance(
key_pair=self.key_pair,
instance_id=instance.id,
network_id=network_id,
connection=self.conn,
)
if wait:
instance.wait()
return instance
[docs] def snapshot(self, instance, clean=True, **kwargs):
"""Snapshot an instance and generate an image from it.
Args:
instance: Instance to snapshot
clean: run instance clean method before taking snapshot
Returns:
An image id
"""
if clean:
instance.clean()
instance.shutdown()
image = self.conn.create_image_snapshot(
'{}-snapshot'.format(self.tag),
instance.server.id,
wait=True
)
return image.id
[docs] def use_key(self, public_key_path, private_key_path=None, name=None):
"""Use an existing key.
Args:
public_key_path: path to the public key to upload
private_key_path: path to the private key
name: name to reference key by
"""
super().use_key(public_key_path, private_key_path, name)
self._openstack_keypair = self._get_openstack_keypair()
def _get_openstack_keypair(self):
"""Get openstack keypair corresponding to this instances keypair.
When creating an openstack instance, a keypair (maintained in
openstack) must be created first. This method gets or creates
the openstack keypair corresponding to the keypair already created
for this cloud instance.
"""
name = self.key_pair.name
public_key_content = self.key_pair.public_key_content
openstack_keypair = self.conn.get_keypair(name)
if not openstack_keypair:
# If the openstack keypair doesn't exist, create it
return self.conn.create_keypair(
name,
public_key_content
)
if public_key_content == openstack_keypair.public_key:
return openstack_keypair
raise Exception(
"An openstack keypair with name {name} already exists, but its"
" public key doesn't match the public key passed in.\n"
"{name} key: {openstack_key}\n"
"Passed in key: {passed_key}".format(
name=name,
openstack_key=openstack_keypair.public_key,
passed_key=public_key_content,
)
)