+ logger = logging.getLogger('EucaAggregate')
+
+ conn = getEucaConnection()
+ if not conn:
+ logger.error('Cannot create a connection to Eucalyptus')
+ return ""
+
+ # Validate RSpec
+ schemaXML = ET.parse(EUCALYPTUS_RSPEC_SCHEMA)
+ rspecValidator = ET.RelaxNG(schemaXML)
+ rspecXML = ET.XML(xml)
+ for network in rspecXML.iterfind("./network"):
+ if network.get('id') != cloud['name']:
+ # Throw away everything except my own RSpec
+ # sfa_logger().error("CreateSliver: deleting %s from rspec"%network.get('id'))
+ network.getparent().remove(network)
+ if not rspecValidator(rspecXML):
+ error = rspecValidator.error_log.last_error
+ message = '%s (line %s)' % (error.message, error.line)
+ # XXX: InvalidRSpec is new. Currently, I am not working with Trunk code.
+ #raise InvalidRSpec(message)
+ raise Exception(message)
+
+ # Get the slice from db or create one.
+ s = Slice.select(Slice.q.slice_hrn == hrn).getOne(None)
+ if s is None:
+ s = Slice(slice_hrn = hrn)
+
+ # Process any changes in existing instance allocation
+ pendingRmInst = []
+ for sliceInst in s.instances:
+ pendingRmInst.append(sliceInst.instance_id)
+ existingInstGroup = rspecXML.findall(".//euca_instances")
+ for instGroup in existingInstGroup:
+ for existingInst in instGroup:
+ if existingInst.get('id') in pendingRmInst:
+ pendingRmInst.remove(existingInst.get('id'))
+ for inst in pendingRmInst:
+ logger.debug('Instance %s will be terminated' % inst)
+ dbInst = EucaInstance.select(EucaInstance.q.instance_id == inst).getOne(None)
+ # Only change the state but do not remove the entry from the DB.
+ dbInst.meta.state = 'deleted'
+ #dbInst.destroySelf()
+ conn.terminate_instances(pendingRmInst)
+
+ # Process new instance requests
+ requests = rspecXML.findall(".//request")
+ if requests:
+ # Get all the public keys associate with slice.
+ pubKeys = getKeysForSlice(api, s.slice_hrn)
+ logger.debug('Passing the following keys to the instance:\n%s' % pubKeys)
+ for req in requests:
+ vmTypeElement = req.getparent()
+ instType = vmTypeElement.get('name')
+ numInst = int(req.find('instances').text)
+
+ bundleName = req.find('bundle').text
+ if not cloud['imageBundles'][bundleName]:
+ logger.error('Cannot find bundle %s' % bundleName)
+ bundleInfo = cloud['imageBundles'][bundleName]
+ instKernel = bundleInfo['kernelID']
+ instDiskImg = bundleInfo['imageID']
+ instRamDisk = bundleInfo['ramdiskID']
+ instKey = None
+
+ # Create the instances
+ for i in range(0, numInst):
+ eucaInst = EucaInstance(slice = s,
+ kernel_id = instKernel,
+ image_id = instDiskImg,
+ ramdisk_id = instRamDisk,
+ key_pair = instKey,
+ inst_type = instType,
+ meta = Meta(start_time=datetime.datetime.now()))
+ eucaInst.reserveInstance(conn, pubKeys)
+
+ # xxx - should return altered rspec
+ # with enough data for the client to understand what's happened
+ return xml
+
+##
+# A thread that will update the meta data.
+#
+def updateMeta():
+ logger = logging.getLogger('EucaAggregate')
+ while True:
+ sleep(120)
+
+ # Get IDs of the instances that don't have IPs yet.
+ dbResults = Meta.select(
+ AND(Meta.q.pri_addr == None,
+ Meta.q.state != 'deleted')
+ )
+ dbResults = list(dbResults)
+ logger.debug('[update thread] dbResults: %s' % dbResults)
+ instids = []
+ for r in dbResults:
+ instids.append(r.instance.instance_id)
+ logger.debug('[update thread] Instance Id: %s' % ', '.join(instids))
+
+ # Get instance information from Eucalyptus
+ conn = getEucaConnection()
+ vmInstances = []
+ reservations = conn.get_all_instances(instids)
+ for reservation in reservations:
+ vmInstances += reservation.instances
+
+ # Check the IPs
+ instIPs = [ {'id':i.id, 'pri_addr':i.private_dns_name, 'pub_addr':i.public_dns_name}
+ for i in vmInstances if i.private_dns_name != '0.0.0.0' ]
+ logger.debug('[update thread] IP dict: %s' % str(instIPs))
+
+ # Update the local DB
+ for ipData in instIPs:
+ dbInst = EucaInstance.select(EucaInstance.q.instance_id == ipData['id']).getOne(None)
+ if not dbInst:
+ logger.info('[update thread] Could not find %s in DB' % ipData['id'])
+ continue
+ dbInst.meta.pri_addr = ipData['pri_addr']
+ dbInst.meta.pub_addr = ipData['pub_addr']
+ dbInst.meta.state = 'running'
+
+def GetVersion(api):
+ xrn=Xrn(api.hrn)
+ request_rspec_versions = [dict(sfa_rspec_version)]
+ ad_rspec_versions = [dict(sfa_rspec_version)]
+ version_more = {'interface':'aggregate',
+ 'testbed':'myplc',
+ 'hrn':xrn.get_hrn(),
+ 'request_rspec_versions': request_rspec_versions,
+ 'ad_rspec_versions': ad_rspec_versions,
+ 'default_ad_rspec': dict(sfa_rspec_version)
+ }
+ return version_core(version_more)
git://git.onelab.eu / sfa.git / blobdiff