Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _get_volume_placement(self, vol_id):
"""
Returns the placement of a volume (or None, if it cannot be determined)
"""
try:
vol = self.ec2_conn.get_all_volumes(volume_ids=[vol_id])
except EC2ResponseError as ec2e:
bioblend.log.error("EC2ResponseError querying for volume {0}: {1}"
.format(vol_id, ec2e))
vol = None
if vol:
return vol[0].zone
else:
bioblend.log.error("Requested placement of a volume '%s' that does not exist.", vol_id)
return None
# V2 format.
for fs in [fs for fs in pd['filesystems'] if fs.get(
'kind', None) == 'volume' and 'ids' in fs]:
# All volumes must be in the same zone
vol_id = fs['ids'][0]
response['placement'] = self._get_volume_placement(
vol_id)
# No need to continue to iterate through
# filesystems, if we found one with a volume.
break
except Exception as exc:
response['error'] = ("Exception while finding placement for "
"cluster '{0}'. This can indicate malformed "
"instance data. Or that this method is "
"broken: {1}".format(cluster_name, exc))
bioblend.log.error(response['error'])
response['placement'] = None
else:
bioblend.log.debug("Insufficient info about cluster {0} to get placement."
.format(cluster_name))
return response
# attempt auto allocation of floating IP
if rs[0].instances[0].private_ip_address and not public_ip:
self.assign_floating_ip(ec2_conn, rs[0].instances[0])
cm_url = "http://{dns}/cloud".format(dns=public_ip)
# Wait until the CloudMan URL is accessible to return the
# data
if self._checkURL(cm_url) is True:
state['instance_state'] = inst_state
state['placement'] = rs[0].instances[0].placement
else:
state['instance_state'] = 'booting'
else:
state['instance_state'] = inst_state
except Exception as e:
err = "Problem updating instance '%s' state: %s" % (instance_id, e)
bioblend.log.error(err)
state['error'] = err
return state
"""
ms = self.get_machine_status()
# Check if the machine is running and update IP and state
self.vm_status = ms.get('instance_state', None)
self.vm_error = ms.get('error', None)
public_ip = ms.get('public_ip', None)
# Update url if we don't have it or is different than what we have
if not self.url and (public_ip and self.url != public_ip):
self._set_url(public_ip)
# See if the cluster has been initialized
if self.vm_status == 'running' or self.url:
ct = self.get_cluster_type()
if ct.get('cluster_type', None):
self.initialized = True
if self.vm_error:
bioblend.log.error(self.vm_error)