X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=sfa%2Fmanagers%2Faggregate_manager_eucalyptus.py;h=e62381a12470b493aa8db2a36ad27fcd4e0b8ed1;hb=52b4cb21f33f380f55d467709b1b9bc8ba84bd57;hp=0a69d2d67acba1a2e0bbe77751cc278df582016e;hpb=7e5f036a09b01bcd29cf0d2e16bfec9ac2f01435;p=sfa.git diff --git a/sfa/managers/aggregate_manager_eucalyptus.py b/sfa/managers/aggregate_manager_eucalyptus.py index 0a69d2d6..e62381a1 100644 --- a/sfa/managers/aggregate_manager_eucalyptus.py +++ b/sfa/managers/aggregate_manager_eucalyptus.py @@ -10,7 +10,7 @@ from boto.ec2.regioninfo import RegionInfo from boto.exception import EC2ResponseError from ConfigParser import ConfigParser from xmlbuilder import XMLBuilder -from xml.etree import ElementTree as ET +from lxml import etree as ET from sqlobject import * import sys @@ -23,6 +23,11 @@ import os # cloud = {} +## +# The location of the RelaxNG schema. +# +EUCALYPTUS_RSPEC_SCHEMA='/etc/sfa/eucalyptus.rng' + ## # A representation of an Eucalyptus instance. This is a support class # for instance <-> slice mapping. @@ -42,8 +47,13 @@ class EucaInstance(SQLObject): # @param botoConn A connection to Eucalyptus. # def reserveInstance(self, botoConn): - print >>sys.stderr, 'Reserving an instance. image: %s, kernel: %s, ramdisk: %s, type: %s, key: %s' % \ - (self.image_id, self.kernel_id, self.ramdisk_id, self.inst_type, self.key_pair) + print >>sys.stderr, 'Reserving an instance: image: %s, kernel: ' \ + '%s, ramdisk: %s, type: %s, key: %s' % \ + (self.image_id, self.kernel_id, self.ramdisk_id, + self.inst_type, self.key_pair) + + # XXX The return statement is for testing. REMOVE in production + #return try: reservation = botoConn.run_instances(self.image_id, @@ -78,7 +88,7 @@ def init_server(): configParser.read(['/etc/sfa/eucalyptus_aggregate.conf', 'eucalyptus_aggregate.conf']) if len(configParser.sections()) < 1: print >>sys.stderr, 'No cloud defined in the config file' - raise 'Cannot find cloud definition in configuration file.' + raise Exception('Cannot find cloud definition in configuration file.') # Only read the first section. cloudSec = configParser.sections()[0] @@ -106,6 +116,12 @@ def init_server(): Slice.createTable(ifNotExists=True) EucaInstance.createTable(ifNotExists=True) + # Make sure the schema exists. + if not os.path.exists(EUCALYPTUS_RSPEC_SCHEMA): + err = 'Cannot location schema at %s' % EUCALYPTUS_RSPEC_SCHEMA + print >>sys.stderr, err + raise Exception(err) + ## # Creates a connection to Eucalytpus. This function is inspired by # the make_connection() in Euca2ools. @@ -157,31 +173,76 @@ class EucaRSpecBuilder(object): # @param cloud A dictionary containing data about a # cloud (ex. clusters, ip) def __init__(self, cloud): - self.eucaRSpec = XMLBuilder() + self.eucaRSpec = XMLBuilder(format = True, tab_step = " ") self.cloudInfo = cloud ## - # Creates the ClusterSpec stanza. + # Creates a request stanza. + # + # @param num The number of instances to create. + # @param image The disk image id. + # @param kernel The kernel image id. + # @param keypair Key pair to embed. + # @param ramdisk Ramdisk id (optional). + # + def __requestXML(self, num, image, kernel, keypair, ramdisk = ''): + xml = self.eucaRSpec + with xml.request: + with xml.instances: + xml << str(num) + with xml.kernel_image(id=kernel): + xml << '' + if ramdisk == '': + with xml.ramdisk: + xml << '' + else: + with xml.ramdisk(id=ramdisk): + xml << '' + with xml.disk_image(id=image): + xml << '' + with xml.keypair: + xml << keypair + + ## + # Creates the cluster stanza. # # @param clusters Clusters information. # def __clustersXML(self, clusters): + cloud = self.cloudInfo xml = self.eucaRSpec + for cluster in clusters: instances = cluster['instances'] - with xml.ClusterSpec(id=cluster['name'], ip=cluster['ip']): - for inst in instances: - with xml.Node(instanceType=inst[0]): - with xml.FreeSlot: - xml << str(inst[1]) - with xml.MaxAllow: - xml << str(inst[2]) - with xml.NumCore: - xml << str(inst[3]) - with xml.Mem: - xml << str(inst[4]) - with xml.DiskSpace(unit='GB'): - xml << str(inst[5]) + with xml.cluster(id=cluster['name']): + with xml.ipv4: + xml << cluster['ip'] + with xml.vm_types: + for inst in instances: + with xml.vm_type(name=inst[0]): + with xml.free_slots: + xml << str(inst[1]) + with xml.max_instances: + xml << str(inst[2]) + with xml.cores: + xml << str(inst[3]) + with xml.memory(unit='MB'): + xml << str(inst[4]) + with xml.disk_space(unit='GB'): + xml << str(inst[5]) + if inst[0] == 'm1.small': + self.__requestXML(1, 'emi-88760F45', 'eki-F26610C6', 'cortex') + if 'instances' in cloud and inst[0] in cloud['instances']: + existingEucaInstances = cloud['instances'][inst[0]] + with xml.euca_instances: + for eucaInst in existingEucaInstances: + with xml.euca_instance(id=eucaInst['id']): + with xml.state: + xml << eucaInst['state'] + with xml.public_dns: + xml << eucaInst['public_dns'] + with xml.keypair: + xml << eucaInst['key'] ## # Creates the Images stanza. @@ -190,14 +251,14 @@ class EucaRSpecBuilder(object): # def __imagesXML(self, images): xml = self.eucaRSpec - with xml.Images: + with xml.images: for image in images: - with xml.Image(id=image.id): - with xml.Type: + with xml.image(id=image.id): + with xml.type: xml << image.type - with xml.Arch: + with xml.arch: xml << image.architecture - with xml.State: + with xml.state: xml << image.state with xml.location: xml << image.location @@ -209,9 +270,9 @@ class EucaRSpecBuilder(object): # def __keyPairsXML(self, keypairs): xml = self.eucaRSpec - with xml.KeyPairs: + with xml.keypairs: for key in keypairs: - with xml.Key: + with xml.keypair: xml << key.name ## @@ -224,25 +285,13 @@ class EucaRSpecBuilder(object): xml = self.eucaRSpec cloud = self.cloudInfo - with xml.RSpec(name='eucalyptus'): - with xml.Capacity: - with xml.CloudSpec(id=cloud['name'], ip=cloud['ip']): - self.__keyPairsXML(cloud['keypairs']) - self.__imagesXML(cloud['images']) - self.__clustersXML(cloud['clusters']) - with xml.Request: - with xml.CloudSpec(id=cloud['name'], ip=cloud['ip']): - with xml.Credential(type='X509'): - xml << 'cred' - with xml.Node(instanceType='m1.small', number='1'): - with xml.Kernel: - xml << 'eki-F26610C6' - with xml.Ramdisk: - xml << '' - with xml.DiskImage: - xml << 'emi-88760F45' - with xml.Key: - xml << 'cortex' + with xml.RSpec(type='eucalyptus'): + with xml.cloud(id=cloud['name']): + with xml.ipv4: + xml << cloud['ip'] + self.__keyPairsXML(cloud['keypairs']) + self.__imagesXML(cloud['images']) + self.__clustersXML(cloud['clusters']) return str(xml) ## @@ -308,6 +357,41 @@ def get_rspec(api, xrn, origin_hrn): # Key Pairs keyPairs = conn.get_all_key_pairs() cloud['keypairs'] = keyPairs + + if hrn: + instanceId = [] + instances = [] + + # Get the instances that belong to the given slice from sqlite3 + # XXX use getOne() in production because the slice's hrn is supposed + # to be unique. For testing, uniqueness is turned off in the db. + theSlice = list(Slice.select(Slice.q.slice_hrn == hrn))[-1] + for instance in theSlice.instances: + instanceId.append(instance.instance_id) + + # Get the information about those instances using their ids. + if len(instanceId) > 0: + reservations = conn.get_all_instances(instanceId) + else: + reservations = [] + for reservation in reservations: + for instance in reservation.instances: + instances.append(instance) + + # Construct a dictory for the EucaRSpecBuilder + instancesDict = {} + for instance in instances: + instList = instancesDict.setdefault(instance.instance_type, []) + instInfoDict = {} + + instInfoDict['id'] = instance.id + instInfoDict['public_dns'] = instance.public_dns_name + instInfoDict['state'] = instance.state + instInfoDict['key'] = instance.key_name + + instList.append(instInfoDict) + cloud['instances'] = instancesDict + except EC2ResponseError, ec2RespErr: errTree = ET.fromstring(ec2RespErr.body) errMsgE = errTree.find('.//Message') @@ -315,6 +399,11 @@ def get_rspec(api, xrn, origin_hrn): rspec = EucaRSpecBuilder(cloud).toXML() + # Remove the instances records so next time they won't + # show up. + if 'instances' in cloud: + del cloud['instances'] + return rspec """ @@ -329,50 +418,75 @@ def create_slice(api, xrn, xml): print >>sys.stderr, 'Error: Cannot create a connection to Eucalyptus' return False + # Validate RSpec + schemaXML = ET.parse(EUCALYPTUS_RSPEC_SCHEMA) + rspecValidator = ET.RelaxNG(schemaXML) + rspecXML = ET.XML(xml) + 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. - # XXX: For testing purposes, I'll just create the slice. - #s = Slice.select(Slice.q.slice_hrn == hrn).getOne(None) - #if s is None: - s = Slice(slice_hrn = hrn) - - # Process the RSpec - rspecXML = RSpec(xml) - rspecDict = rspecXML.toDict() - request = rspecDict['RSpec']['Request'] - for cloudSpec in request: - cloudSpec = cloudSpec['CloudSpec'] - for cloudReqInfo in cloudSpec: - for nodeReq in cloudReqInfo['Node']: - instKernel = nodeReq['Kernel'][0] - instDiskImg = nodeReq['DiskImage'][0] - instRamDisk = nodeReq['Ramdisk'][0] - instKey = nodeReq['Key'][0] - instType = nodeReq['instanceType'] - numInst = int(nodeReq['number']) - - # Ramdisk is optional. - if isinstance(instRamDisk, dict): - instRamDisk = 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) - eucaInst.reserveInstance(conn) + 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: + print >>sys.stderr, 'Instance %s will be terminated' % inst + dbInst = EucaInstance.select(EucaInstance.q.instance_id == inst).getOne(None) + dbInst.destroySelf() + conn.terminate_instances(pendingRmInst) + + # Process new instance requests + requests = rspecXML.findall('.//request') + for req in requests: + vmTypeElement = req.getparent() + instType = vmTypeElement.get('name') + numInst = int(req.find('instances').text) + instKernel = req.find('kernel_image').get('id') + instDiskImg = req.find('disk_image').get('id') + instKey = req.find('keypair').text + + ramDiskElement = req.find('ramdisk') + ramDiskAttr = ramDiskElement.attrib + if 'id' in ramDiskAttr: + instRamDisk = ramDiskAttr['id'] + else: + instRamDisk = 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) + eucaInst.reserveInstance(conn) return True def main(): init_server() - r = RSpec() - r.parseFile(sys.argv[1]) - rspec = r.toDict() - create_slice(None,'planetcloud.pc.test',rspec) - #rspec = get_rspec('euca', 'hrn:euca', 'oring_hrn') + + theRSpec = None + with open(sys.argv[1]) as xml: + theRSpec = xml.read() + create_slice(None, 'planetcloud.pc.test', theRSpec) + + #rspec = get_rspec('euca', 'planetcloud.pc.test', 'planetcloud.pc.marcoy') #print rspec if __name__ == "__main__":