# Copyright 2020 PerfKitBenchmarker Authors. All rights reserved. # # 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. """Utilities for working with IBM Cloud resources.""" import json import logging import os import threading import time from absl import flags from perfkitbenchmarker import errors from perfkitbenchmarker.providers.ibmcloud import ibmcloud_manager from perfkitbenchmarker.providers.ibmcloud import util FLAGS = flags.FLAGS ONE_HOUR = 3600 class Enum(tuple): __getattr__ = tuple.index class States(Enum): AVAILABLE = 'available' RUNNING = 'running' STOPPED = 'stopped' ATTACHED = 'attached' class IbmAPICommand: """A IBM Cloud rest api command. Attributes: args: list of strings. Positional args to pass, typically specifying an operation to perform (e.g. ['image', 'list'] to list available images). flags: a dict mapping flag name string to flag value. Flags to pass to ibm cloud rest api. If a provided value is True, the flag is passed without a value. If a provided value is a list, the flag is passed multiple times, once with each value in the list. ibmcloud_auth_token: Auth token for cmd. ibmcloud_auth_token_time: Time when auth token was acquired. user_data: user data info. vpc_id: vpc_id - passed in flags. sg_id: sg_id - passed in flags. zone: zone - passed in flags. cidr: cidr - passed in flags. name: name - passed in flags. prefix: prefix - passed in flags. items: items - passed in flags capacity: capacity - passed in flags. iops: iops - passed in flags. profile: profile - passed in flags. encryption_key: encryption_key - passed in flags. resource_group: resource_group - passed in flags. volume: volume - passed in flags. pubkey: pubkey - passed in flags. imageid: imageid - passed in flags. target: target - passed in flags. vnic_id: vnic_id - passed in flags. fip_id: fip_id - passed in flags. instance_id: instance_id - passed in flags. subnet: subnet - passed in flags. image_id: image_id - passed in flags. image_name: image_name - passed in flags. resource_id: resource_id - passed in flags. data_encryption_key: data encryption key for account. boot_encryption_key: boot encryption key for account. """ gen: ibmcloud_manager.IbmCloud = None gen_instmgr: ibmcloud_manager.InstanceManager gen_imgmgr: ibmcloud_manager.ImageManager gen_keymgr: ibmcloud_manager.KeyManager gen_subnetmgr: ibmcloud_manager.SubnetManager gen_fipmgr: ibmcloud_manager.FipManager gen_volumemgr: ibmcloud_manager.VolumeManager gen_vpcmgr: ibmcloud_manager.VPCManager gen_sgmgr: ibmcloud_manager.SGManager ibmcloud_account_id = None ibmcloud_apikey = None ibmcloud_auth_token = None ibmcloud_auth_token_time = 0 _lock = threading.Lock() def __init__(self, *args, **kwargs): """Initializes a IbmAPICommand with the provided args and flags. Args: *args: sequence of strings. Positional args to pass to rest api calls, typically specifying an operation to perform (e.g. ['image', 'list'] to list available images). **kwargs: resource - A IBM Cloud resource of type BaseResource. """ self._CheckEnvironment() if ( self.gen is None or time.time() - self.ibmcloud_auth_token_time >= ONE_HOUR ): IbmAPICommand.gen = ibmcloud_manager.IbmCloud( account=self.ibmcloud_account_id, apikey=self.ibmcloud_apikey, verbose=False, version='v1', silent=True, force=False, ) IbmAPICommand.gen_instmgr = ibmcloud_manager.InstanceManager( IbmAPICommand.gen ) IbmAPICommand.gen_imgmgr = ibmcloud_manager.ImageManager( IbmAPICommand.gen ) IbmAPICommand.gen_fipmgr = ibmcloud_manager.FipManager(IbmAPICommand.gen) IbmAPICommand.gen_keymgr = ibmcloud_manager.KeyManager(IbmAPICommand.gen) IbmAPICommand.gen_subnetmgr = ibmcloud_manager.SubnetManager( IbmAPICommand.gen ) IbmAPICommand.gen_volumemgr = ibmcloud_manager.VolumeManager( IbmAPICommand.gen ) IbmAPICommand.gen_vpcmgr = ibmcloud_manager.VPCManager(IbmAPICommand.gen) IbmAPICommand.gen_sgmgr = ibmcloud_manager.SGManager(IbmAPICommand.gen) self.ibmcloud_auth_token = self.GetToken() self.ibmcloud_auth_token_time = time.time() self.args = args self.flags = kwargs self.user_data = None # self.sgid = None @property def vpc_id(self): return self.flags['vpcid'] @property def sg_id(self): return self.flags['sgid'] @property def zone(self): return self.flags['zone'] @property def cidr(self): return self.flags['cidr'] @property def name(self): return self.flags['name'] @property def prefix(self): return self.flags['prefix'] @property def items(self): return self.flags['items'] @property def capacity(self): return self.flags['capacity'] @property def iops(self): return self.flags['iops'] @property def profile(self): return self.flags['profile'] @property def encryption_key(self): return self.flags['encryption_key'] @property def resource_group(self): return self.flags['resource_group'] @property def volume(self): return self.flags['volume'] @property def pubkey(self): return self.flags['pubkey'] @property def image_id(self): return self.flags['imageid'] @property def target(self): return self.flags['target'] @property def vnic_id(self): return self.flags['vnicid'] @property def fip_id(self): return self.flags['fipid'] @property def instance_id(self): return self.flags['instanceid'] @property def subnet(self): return self.flags['subnet'] @property def image_name(self): return self.flags['image_name'] @property def resource_id(self): return self.flags['id'] def _CheckEnvironment(self): """Get the information needed to authenticate on IBM Cloud. Check for the config if env variables for apikey is not set. """ if not os.environ.get('IBMCLOUD_ENDPOINT'): raise errors.Config.InvalidValue( 'PerfKit Benchmarker on IBM Cloud requires that the ' 'environment variable IBMCLOUD_ENDPOINT is set.' ) account = util.Account( os.environ.get('IBMCLOUD_ACCOUNT_ID'), os.environ.get('IBMCLOUD_APIKEY'), None, ) if not account.name or not account.apikey: accounts = os.environ.get('IBMCLOUD_ACCOUNTS') # read from config file if accounts and accounts: config = util.ReadConfig(accounts) benchmark = config[FLAGS.benchmarks[0]][0] # first account account = util.Account(benchmark[0], benchmark[1], benchmark[2]) logging.info('account_id: %s', account.name) if FLAGS.ibmcloud_datavol_encryption_key: # if this flag is set self.data_encryption_key = account.enckey logging.info('KP key to use, data: %s', self.data_encryption_key) if FLAGS.ibmcloud_bootvol_encryption_key: # use same key as data self.boot_encryption_key = account.enckey logging.info('KP key to use, boot: %s', self.boot_encryption_key) if not account.name or not account.apikey: raise errors.Config.InvalidValue( 'PerfKit Benchmarker on IBM Cloud requires that the ' 'environment variables IBMCLOUD_ACCOUNT_ID and IBMCLOUD_APIKEY ' 'are correctly set.' ) def GetToken(self): """Returns user token.""" return self.gen.GetToken() def CreateSgRules(self): """Creates default security group rules needed for vms to communicate.""" self.gen_sgmgr.CreateRule( self.sg_id, 'inbound', 'ipv4', '0.0.0.0/0', 'icmp', None ) self.gen_sgmgr.CreateRule( self.sg_id, 'inbound', 'ipv4', '0.0.0.0/0', 'tcp', 22 ) self.gen_sgmgr.CreateRule( self.sg_id, 'inbound', 'ipv4', '0.0.0.0/0', 'tcp', None, port_min=1024, port_max=50000, ) self.gen_sgmgr.CreateRule( self.sg_id, 'inbound', 'ipv4', '0.0.0.0/0', 'udp', None, port_min=1024, port_max=50000, ) self.gen_sgmgr.CreateRule( self.sg_id, 'inbound', 'ipv4', '0.0.0.0/0', 'tcp', 443 ) def GetSecurityGroupId(self): """Returns default security group id.""" logging.info('Looking up existing default sg') prefix = self.prefix resp = self.gen_sgmgr.List() if resp and 'security_groups' in resp: for item in resp['security_groups']: # find the one with matching vpc name if item['vpc']['name'].startswith(prefix): return item['id'] else: logging.error('Failed to retrieve security group id: %s', resp) return None def CreatePrefix(self): """Creates address prefix on vpc needed to attach subnets. Flags: vpcid: ibm cloud vpc id. zone: name of the zone within ibm network. cidr: ip range that this prefix will cover. name: name of this prefix. Returns: The json representation of the created address prefix. """ return self.gen_vpcmgr.CreatePrefix( self.vpc_id, self.zone, self.cidr, name=self.name ) def CreateSubnet(self): """Creates a subnet on the vpc. Flags: vpcid: ibm cloud vpc id. zone: name of the zone within ibm network. cidr: ip range for the subnet. name: name of this subnet. Returns: The json representation of the created subnet. """ kwargs = {} kwargs['name'] = self.name kwargs['zone'] = self.zone return self.gen_subnetmgr.Create(self.cidr, self.vpc_id, **kwargs) def DeleteResource(self): """Deletes a resource based on items set in the flags. Flags: items: type of resource to delete. id: matching id to delete. """ data_mgr = self.gen_subnetmgr if self.items == 'vpcs': data_mgr = self.gen_vpcmgr elif self.items == 'keys': data_mgr = self.gen_keymgr resp = data_mgr.List() if resp and self.items in resp: for item in resp[self.items]: item_id = item.get('id') if item_id and item_id == self.resource_id: try: data_mgr.Delete(item_id) logging.info('Deleted %s, id: %s', self.items, item_id) except Exception: # pylint: disable=broad-except pass else: logging.info('No items found to delete: %s', self.items) def GetResource(self): """Returns id of the found resource matching the resource type and prefix. Flags: items: type of resource. zone: name of the zone within ibm network, for subnets. preifx: a label matching the resource name. Returns: The id of the first matching resource. """ logging.info( 'Looking up existing %s matching prefix: %s, zone: %s', self.items, self.prefix, self.zone, ) data_mgr = self.gen_subnetmgr if self.items == 'vpcs': data_mgr = self.gen_vpcmgr elif self.items == 'keys': data_mgr = self.gen_keymgr # instead of returning a list, we just return first matching item resourceid = None resp = data_mgr.List() if resp and self.items in resp: for item in resp[self.items]: itemid = item.get('id') name = item.get('name') if itemid and name and name.startswith(self.prefix): if self.items == 'subnets': if self.zone == item.get('zone')['name']: resourceid = itemid # find the id that matches the zone name break else: # for vpc and key resourceid = itemid # # for vpcs, just get the first matching break logging.info('Resource found, id: %s', resourceid) else: logging.error('Failed to retrieve %s, no data returned', self.items) return resourceid def ListSubnetsExtra(self): """Returns a list of extra subnets, this is not used for regular runs. Flags: zone: name of the zone within ibm network, for subnets. preifx: a label matching the resource name. Returns: List of ids of the additional subnets matching the prefix and zone. """ items = 'subnets' logging.info('Looking up extra subnets for zone: %s', self.zone) data_mgr = self.gen_subnetmgr subnets = {} resp = data_mgr.List() if resp and items in resp: for item in resp[items]: itemid = item.get('id') name = item.get('name') ipv4_cidr_block = item.get('ipv4_cidr_block') # extra subnets start with 'sxs' if itemid and name and name.startswith(self.prefix): if self.zone == item.get('zone')['name']: logging.info('Found extra subnet, name: %s, id: %s', name, itemid) names = name.split(util.DELIMITER) if names[0] not in subnets: subnets[names[0]] = {} subnets[names[0]]['id'] = itemid # needed to create route later subnets[names[0]]['ipv4_cidr_block'] = ipv4_cidr_block else: logging.error('Failed to retrieve extra subnets, no data returned') return subnets def CreateVolume(self): """Creates an external disk volume. Flags: name: name of the new volume. capacity: size of the volume in GB. iops: desired iops on the volume. profile: profile name of the volume, usually custom for large volumes encryption_key: optional, encrytion key resource_group: optional, resource group id to assign the volume to. Returns: json representation of the created volume. """ kwargs = {} kwargs['name'] = self.name kwargs['capacity'] = self.capacity kwargs['iops'] = self.iops kwargs['profile'] = self.profile if 'encryption_key' in self.flags: kwargs['encryption_key'] = self.encryption_key if 'resource_group' in self.flags: kwargs['resource_group'] = self.resource_group return json.dumps(self.gen_volumemgr.Create(self.zone, **kwargs)) def DeleteVolume(self): """Deletes volume.""" return self.gen_volumemgr.Delete(self.volume) def ShowVolume(self): """Shows volume in json.""" return json.dumps(self.gen_volumemgr.Show(self.volume), indent=2) def CreateVpc(self): """Creates a vpc. Flags: name: name of the vpc to create. Returns: The id of the created vpc. """ vpcid = None resp = self.gen_vpcmgr.Create(self.name) if resp: vpcid = resp.get('id') sgid = resp.get('default_security_group')['id'] logging.info('Created vpc: %s, sgid: %s', vpcid, sgid) self.flags['sgid'] = sgid self.CreateSgRules() logging.info('Created sg rules for icmp, tcp 22/443, tcp/udp 1024-50000') return vpcid def CreateKey(self): """Creates a ssh key in ibm cloud. Flags: name: name of the key to create. pubkey: public key to use. Returns: The id of the created ssh key. """ kwargs = {} kwargs['name'] = self.flags['name'] resp = self.gen_keymgr.Create(self.pubkey, 'rsa', **kwargs) return resp.get('id') if resp else None def CreateInstance(self): """Creates a vm. Flags: name: name of the vm. imageid: id of the image to use. profile: machine type for the vm. vpcid: id of the vpc. user_data: this is set if windows optional: optional args added to kwargs. Returns: The json representation of the created vm. """ kwargs = {} for arg in [ 'key', 'zone', 'subnet', 'networks', 'encryption_key', 'iops', 'capacity', 'resource_group', ]: if arg in self.flags: kwargs[arg] = self.flags[arg] return json.dumps( self.gen_instmgr.Create( self.name, self.image_id, self.profile, self.vpc_id, user_data=self.user_data, **kwargs ) ) def InstanceGetPrimaryVnic(self): """Finds the primary network vnic on vm. Flags: instanceid: id of the vm. Returns: The found vnic id. """ resp = self.gen_instmgr.Show(self.instance_id) if ( 'primary_network_interface' in resp and 'href' in resp['primary_network_interface'] ): items = resp['primary_network_interface']['href'].split('/') logging.info('vnic found: %s', items[len(items) - 1]) return items[len(items) - 1] logging.error('no href in instance: %s', self.instance_id) return None def InstanceFipCreate(self): """Creates a FIP on vm. Flags: name: name of the fip. Returns: json represenation of the created fip. """ kwargs = {} kwargs['name'] = self.name return json.dumps(self.gen_fipmgr.Create(None, self.target, **kwargs)) def InstanceFipDelete(self): """Deletes FIP on vm matching the id.""" self.gen_fipmgr.Delete(self.fip_id) def _GetStatus(self, resp): """Returns instance status in the given resp.""" status = None if not resp: logging.error('http response is None') return status if 'status' in resp: status = resp['status'] else: logging.error('No status found in resp: %s', resp) return status def InstanceStatus(self): """Returns instance status matching the instance id.""" resp = self.gen_instmgr.Show(self.instance_id) status = self._GetStatus(resp) logging.info('Instance status: %s, %s', self.instance_id, status) if status != States.RUNNING: endtime = time.time() + FLAGS.ibmcloud_timeout status = self._GetStatus(self.gen_instmgr.Show(self.instance_id)) while status != States.RUNNING and time.time() < endtime: time.sleep(FLAGS.ibmcloud_polling_delay) status = self._GetStatus(self.gen_instmgr.Show(self.instance_id)) logging.info( 'Checking instance status: %s, %s', self.instance_id, status ) return status def InstanceStart(self): """Starts the instance matching instance id and returns final status.""" resp = self.gen_instmgr.Start(self.flags['instanceid']) status = self._GetStatus(resp) logging.info( 'Instance start issued: %s, %s', self.flags['instanceid'], status ) if status != States.RUNNING: endtime = time.time() + FLAGS.ibmcloud_timeout status = self._GetStatus(self.gen_instmgr.Show(self.flags['instanceid'])) while status != States.RUNNING and time.time() < endtime: time.sleep(FLAGS.ibmcloud_polling_delay) status = self._GetStatus( self.gen_instmgr.Show(self.flags['instanceid']) ) logging.info( 'Checking instance start status: %s, %s', self.flags['instanceid'], status, ) return status def InstanceStop(self): """Stops instance matching the instance id and returns final status.""" resp = self.gen_instmgr.Stop(self.instance_id, force=False) logging.info('Instance stop issued: %s', self.instance_id) # force stop again resp = self.gen_instmgr.Stop(self.instance_id, force=True) status = self._GetStatus(resp) logging.info('Instance stop force issued: %s, %s', self.instance_id, status) if status != States.STOPPED: endtime = time.time() + FLAGS.ibmcloud_timeout status = self._GetStatus(self.gen_instmgr.Show(self.instance_id)) while status != States.STOPPED and time.time() < endtime: time.sleep(FLAGS.ibmcloud_polling_delay * 2) status = self._GetStatus(self.gen_instmgr.Show(self.instance_id)) logging.info( 'Checking instance stop status: %s, %s', self.instance_id, status ) return status def InstanceDelete(self): """Deletes instance matching the instance id.""" self.gen_instmgr.Delete(self.instance_id) def InstanceList(self): """Lists instances and returns json objects of all found instances.""" return self.gen_instmgr.List() def InstanceShow(self): """Returns the json respresentation of the matching instance id.""" return self.gen_instmgr.Show(self.instance_id) def InstanceInitializationShow(self): """Returns the json format of the config used to initialize instance id.""" return self.gen_instmgr.ShowInitialization(self.instance_id) def InstanceVnicCreate(self): """Creates a vnic on the instance. Flags: instanceid: id of the vm. name: name of the vnic. subnet: subnet id. Returns: id of the created vnic and assigned private ip address. """ ip_addr = None resp = self.gen_instmgr.CreateVnic(self.instance_id, self.name, self.subnet) status = self._GetStatus(resp) vnicid = resp['id'] port_speed = resp['port_speed'] logging.info( 'Vnic created, vnicid: %s, port speed: %s, vnic status: %s', vnicid, port_speed, status, ) if status != States.AVAILABLE: endtime = time.time() + FLAGS.ibmcloud_timeout resp = self.gen_instmgr.ShowVnic(self.instance_id, vnicid) status = self._GetStatus(resp) while status != States.AVAILABLE and time.time() < endtime: time.sleep(FLAGS.ibmcloud_polling_delay) resp = self.gen_instmgr.ShowVnic(self.instance_id, vnicid) status = self._GetStatus(resp) logging.info('Checking instance vnic status: %s', status) if status == States.AVAILABLE: ip_addr = resp.get('primary_ipv4_address') logging.info('primary_ipv4_address: %s', ip_addr) assert status == States.AVAILABLE return vnicid, ip_addr def InstanceVnicShow(self): """GETs instance vnic. Flags: instanceid: id of the vm. vnicid: id of the vnic. Returns: json representation of the found vnic. """ return json.dumps( self.gen_instmgr.ShowVnic(self.instance_id, self.vnic_id), indent=2 ) def InstanceVnicList(self): """Returns a list of all vnics on instance. Flags: instanceid: id of the vm. Returns: json objects of all vnics on the instance. """ return json.dumps(self.gen_instmgr.ListVnics(self.instance_id), indent=2) def InstanceCreateVolume(self): """Attaches a volume to instance. Flags: instanceid: id of the vm. name: name of the volume on the vm. volume: id of the volume to attach to the vm. True: always true, delete the volume when instance is deleted. Returns: json representation of the attached volume on the instance. """ return json.dumps( self.gen_instmgr.CreateVolume( self.instance_id, self.name, self.volume, True ), indent=2, ) def InstanceDeleteVolume(self): """Deletes the volume on the vm matching instanceid and volume.""" return json.dumps( self.gen_instmgr.DeleteVolume(self.instance_id, self.volume), indent=2 ) def InstanceShowVolume(self): """Returns json representation of the volume matching instanceid.""" for item in self.gen_instmgr.ListVolumes(self.instance_id)[ 'volume_attachments' ]: if item['volume'].get('id') == self.volume: return json.dumps( self.gen_instmgr.ShowVolume(self.instance_id, item.get('id')), indent=2, ) def ImageList(self): """Returns json objects of all available images.""" return self.gen_imgmgr.List(account=self.ibmcloud_account_id) def GetImageId(self): """Returns image id that matches the image name.""" resp = self.gen_imgmgr.List()['images'] if resp is not None: for image in resp: if ( image['name'].startswith(self.image_name) and image['status'] == States.AVAILABLE ): return image['id'] return None def ImageShow(self): """Returns json object of the image matching the id.""" return self.gen_imgmgr.Show(self.image_id)