Fixed a bug that showed instances not associated with the slice
[sfa.git] / sfa / managers / aggregate_manager_eucalyptus.py
index 0a69d2d..e62381a 100644 (file)
@@ -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__":