Source code for pycloudlib.gce.instance

# This file is part of pycloudlib. See LICENSE file for license information.
"""GCE instance."""

from time import sleep

import googleapiclient.discovery
from googleapiclient.errors import HttpError

from pycloudlib.gce.util import raise_on_error, get_credentials
from pycloudlib.instance import BaseInstance

[docs]class GceInstance(BaseInstance): """GCE backed instance.""" _type = 'gce'
[docs] def __init__( self, key_pair, instance_id, project, zone, credentials_path, *, name=None ): """Set up the instance. Args: key_pair: A KeyPair for SSH interactions instance_id: Id returned when creating the instance project: Project instance was created in zone: Zone instance was created in """ if project is None or zone is None: raise ValueError("kwargs 'project' and 'zone' are required. " "Project: {}, Zone: {}".format(project, zone)) super().__init__(key_pair) self.instance_id = instance_id self._name = name self.project = project = zone self._ip = None credentials = get_credentials(credentials_path) self.instance = 'compute', 'v1', cache_discovery=False, credentials=credentials ).instances()
def __repr__(self): """Create string representation of class.""" return '{}(instance_id={})'.format( self.__class__.__name__, self.instance_id, ) @property def id(self): """Return the instance id.""" return self.instance_id @property def name(self): """Return the instance name.""" if not self._name: result = self.instance.get( project=self.project,, instance=self.instance_id, ).execute() self._name = result['name'] return self._name @property def ip(self): """Return IP address of instance.""" if not self._ip: self._ip = self._get_ip() return self._ip def _get_ip(self): result = self.instance.get( project=self.project,, instance=self.instance_id, ).execute() ip = result['networkInterfaces'][0]['accessConfigs'][0]['natIP'] return ip
[docs] def delete(self, wait=True): """Delete the instance. Args: wait: wait for instance to be deleted """ response = self.instance.delete( project=self.project,, instance=self.instance_id ).execute() raise_on_error(response) if wait: self.wait_for_delete()
[docs] def restart(self, wait=True, **kwargs): """Restart the instance. Args: wait: wait for the instance to be fully started """ self.shutdown() self.start()
[docs] def shutdown(self, wait=True, **kwargs): """Shutdown the instance. Args: wait: wait for the instance to shutdown """ response = self.instance.stop( project=self.project,, instance=self.instance_id ).execute() raise_on_error(response) if wait: self.wait_for_stop()
[docs] def start(self, wait=True): """Start the instance. Args: wait: wait for the instance to start. """ response = self.instance.start( project=self.project,, instance=self.instance_id ).execute() raise_on_error(response) if wait: self.wait()
def _wait_for_instance_start(self): """Wait for instance to be up.""" self._wait_for_status('RUNNING') self._ip = self._get_ip()
[docs] def wait_for_delete(self, sleep_seconds=300): """Wait for instance to be deleted.""" for _ in range(sleep_seconds): try: result = self.instance.get( project=self.project,, instance=self.instance_id ).execute() if result['status'] == 'TERMINATED': break except HttpError as e: # Sometimes URL just 404s once deleted if e.resp.status == 404: break raise e else: raise Exception( 'Instance not terminated after {} seconds. ' 'Check GCE console.'.format(sleep_seconds) )
[docs] def wait_for_stop(self): """Wait for instance stop.""" self._wait_for_status('TERMINATED')
def _wait_for_status(self, status, sleep_seconds=300): response = None for _ in range(sleep_seconds): response = self.instance.get( project=self.project,, instance=self.instance_id ).execute() if response['status'] == status: break sleep(1) else: raise Exception( 'Expected {} state, but found {} after waiting {} seconds. ' 'Check GCE console for more details. \n' 'Status message: {}'.format( status, response['status'], sleep_seconds, response['statusMessage'] ) )