Source code for

"""Openstack cloud type."""
import base64

import openstack

from 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 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) = 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 except AttributeError as e: raise Exception( 'No network found named {}'.format( ) 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,, networks=networks,, user_data=user_data, wait=wait, **kwargs, ) instance = OpenstackInstance( key_pair=self.key_pair,, 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),, wait=True ) return
[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 = 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, ) )