sfa/managers/aggregate_manager_eucalyptus.py

   1 from __future__ import with_statement
   2
   3 import sys
   4 import os, errno
   5 import logging
   6 import datetime
   7 from multiprocessing import Process
   8 from time import sleep
   9
  10 import boto
  11 from boto.ec2.regioninfo import RegionInfo
  12 from boto.exception import EC2ResponseError
  13 from ConfigParser import ConfigParser
  14 from xmlbuilder import XMLBuilder
  15 from lxml import etree as ET
  16 from sqlobject import *
  17
  18 from sfa.util.faults import *
  19 from sfa.util.xrn import urn_to_hrn, Xrn
  20 from sfa.util.plxrn import hrn_to_pl_slicename, slicename_to_hrn
  21 from sfa.util.callids import Callids
  22 from sfa.util.sfalogging import logger
  23 from sfa.util.version import version_core
  24
  25 from sfa.trust.credential import Credential
  26
  27 from sfa.server.sfaapi import SfaApi
  28
  29 from sfa.plc.aggregate import Aggregate
  30 from sfa.plc.slices import *
  31 from sfa.rspecs.sfa_rspec import sfa_rspec_version
  32
  33
  34 ##
  35 # The data structure used to represent a cloud.
  36 # It contains the cloud name, its ip address, image information,
  37 # key pairs, and clusters information.
  38 #
  39 cloud = {}
  40
  41 ##
  42 # The location of the RelaxNG schema.
  43 #
  44 EUCALYPTUS_RSPEC_SCHEMA='/etc/sfa/eucalyptus.rng'
  45
  46 ##
  47 # Meta data of an instance.
  48 #
  49 class Meta(SQLObject):
  50     instance   = SingleJoin('EucaInstance')
  51     state      = StringCol(default = 'new')
  52     pub_addr   = StringCol(default = None)
  53     pri_addr   = StringCol(default = None)
  54     start_time = DateTimeCol(default = None)
  55
  56 ##
  57 # A representation of an Eucalyptus instance. This is a support class
  58 # for instance <-> slice mapping.
  59 #
  60 class EucaInstance(SQLObject):
  61     instance_id = StringCol(unique=True, default=None)
  62     kernel_id   = StringCol()
  63     image_id    = StringCol()
  64     ramdisk_id  = StringCol()
  65     inst_type   = StringCol()
  66     key_pair    = StringCol()
  67     slice       = ForeignKey('Slice')
  68     meta        = ForeignKey('Meta')
  69
  70     ##
  71     # Contacts Eucalyptus and tries to reserve this instance.
  72     #
  73     # @param botoConn A connection to Eucalyptus.
  74     # @param pubKeys A list of public keys for the instance.
  75     #
  76     def reserveInstance(self, botoConn, pubKeys):
  77         logger = logging.getLogger('EucaAggregate')
  78         logger.info('Reserving an instance: image: %s, kernel: ' \
  79                     '%s, ramdisk: %s, type: %s, key: %s' % \
  80                     (self.image_id, self.kernel_id, self.ramdisk_id,
  81                     self.inst_type, self.key_pair))
  82
  83         # XXX The return statement is for testing. REMOVE in production
  84         #return
  85
  86         try:
  87             reservation = botoConn.run_instances(self.image_id,
  88                                                  kernel_id = self.kernel_id,
  89                                                  ramdisk_id = self.ramdisk_id,
  90                                                  instance_type = self.inst_type,
  91                                                  key_name  = self.key_pair,
  92                                                  user_data = pubKeys)
  93             for instance in reservation.instances:
  94                 self.instance_id = instance.id
  95
  96         # If there is an error, destroy itself.
  97         except EC2ResponseError, ec2RespErr:
  98             errTree = ET.fromstring(ec2RespErr.body)
  99             msg = errTree.find('.//Message')
 100             logger.error(msg.text)
 101             self.destroySelf()
 102
 103 ##
 104 # A representation of a PlanetLab slice. This is a support class
 105 # for instance <-> slice mapping.
 106 #
 107 class Slice(SQLObject):
 108     slice_hrn = StringCol()
 109     #slice_index = DatabaseIndex('slice_hrn')
 110     instances = MultipleJoin('EucaInstance')
 111
 112 ##
 113 # Initialize the aggregate manager by reading a configuration file.
 114 #
 115 def init_server():
 116     logger = logging.getLogger('EucaAggregate')
 117     fileHandler = logging.FileHandler('/var/log/euca.log')
 118     fileHandler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
 119     logger.addHandler(fileHandler)
 120     fileHandler.setLevel(logging.DEBUG)
 121     logger.setLevel(logging.DEBUG)
 122
 123     configParser = ConfigParser()
 124     configParser.read(['/etc/sfa/eucalyptus_aggregate.conf', 'eucalyptus_aggregate.conf'])
 125     if len(configParser.sections()) < 1:
 126         logger.error('No cloud defined in the config file')
 127         raise Exception('Cannot find cloud definition in configuration file.')
 128
 129     # Only read the first section.
 130     cloudSec = configParser.sections()[0]
 131     cloud['name'] = cloudSec
 132     cloud['access_key'] = configParser.get(cloudSec, 'access_key')
 133     cloud['secret_key'] = configParser.get(cloudSec, 'secret_key')
 134     cloud['cloud_url']  = configParser.get(cloudSec, 'cloud_url')
 135     cloudURL = cloud['cloud_url']
 136     if cloudURL.find('https://') >= 0:
 137         cloudURL = cloudURL.replace('https://', '')
 138     elif cloudURL.find('http://') >= 0:
 139         cloudURL = cloudURL.replace('http://', '')
 140     (cloud['ip'], parts) = cloudURL.split(':')
 141
 142     # Create image bundles
 143     images = getEucaConnection().get_all_images()
 144     cloud['images'] = images
 145     cloud['imageBundles'] = {}
 146     for i in images:
 147         if i.type != 'machine' or i.kernel_id is None: continue
 148         name = os.path.dirname(i.location)
 149         detail = {'imageID' : i.id, 'kernelID' : i.kernel_id, 'ramdiskID' : i.ramdisk_id}
 150         cloud['imageBundles'][name] = detail
 151
 152     # Initialize sqlite3 database and tables.
 153     dbPath = '/etc/sfa/db'
 154     dbName = 'euca_aggregate.db'
 155
 156     if not os.path.isdir(dbPath):
 157         logger.info('%s not found. Creating directory ...' % dbPath)
 158         os.mkdir(dbPath)
 159
 160     conn = connectionForURI('sqlite://%s/%s' % (dbPath, dbName))
 161     sqlhub.processConnection = conn
 162     Slice.createTable(ifNotExists=True)
 163     EucaInstance.createTable(ifNotExists=True)
 164     Meta.createTable(ifNotExists=True)
 165
 166     # Start the update process to keep track of the meta data
 167     # about Eucalyptus instance.
 168     Process(target=updateMeta).start()
 169
 170     # Make sure the schema exists.
 171     if not os.path.exists(EUCALYPTUS_RSPEC_SCHEMA):
 172         err = 'Cannot location schema at %s' % EUCALYPTUS_RSPEC_SCHEMA
 173         logger.error(err)
 174         raise Exception(err)
 175
 176 ##
 177 # Creates a connection to Eucalytpus. This function is inspired by
 178 # the make_connection() in Euca2ools.
 179 #
 180 # @return A connection object or None
 181 #
 182 def getEucaConnection():
 183     global cloud
 184     accessKey = cloud['access_key']
 185     secretKey = cloud['secret_key']
 186     eucaURL   = cloud['cloud_url']
 187     useSSL    = False
 188     srvPath   = '/'
 189     eucaPort  = 8773
 190     logger    = logging.getLogger('EucaAggregate')
 191
 192     if not accessKey or not secretKey or not eucaURL:
 193         logger.error('Please set ALL of the required environment ' \
 194                      'variables by sourcing the eucarc file.')
 195         return None
 196
 197     # Split the url into parts
 198     if eucaURL.find('https://') >= 0:
 199         useSSL  = True
 200         eucaURL = eucaURL.replace('https://', '')
 201     elif eucaURL.find('http://') >= 0:
 202         useSSL  = False
 203         eucaURL = eucaURL.replace('http://', '')
 204     (eucaHost, parts) = eucaURL.split(':')
 205     if len(parts) > 1:
 206         parts = parts.split('/')
 207         eucaPort = int(parts[0])
 208         parts = parts[1:]
 209         srvPath = '/'.join(parts)
 210
 211     return boto.connect_ec2(aws_access_key_id=accessKey,
 212                             aws_secret_access_key=secretKey,
 213                             is_secure=useSSL,
 214                             region=RegionInfo(None, 'eucalyptus', eucaHost),
 215                             port=eucaPort,
 216                             path=srvPath)
 217
 218 ##
 219 # Returns a string of keys that belong to the users of the given slice.
 220 # @param sliceHRN The hunman readable name of the slice.
 221 # @return sting()
 222 #
 223 # This method is no longer needed because the user keys are passed into
 224 # CreateSliver
 225 #
 226 def getKeysForSlice(api, sliceHRN):
 227     logger   = logging.getLogger('EucaAggregate')
 228     cred     = api.getCredential()
 229     registry = api.registries[api.hrn]
 230     keys     = []
 231
 232     # Get the slice record
 233     records = registry.Resolve(sliceHRN, cred)
 234     if not records:
 235         logging.warn('Cannot find any record for slice %s' % sliceHRN)
 236         return []
 237
 238     # Find who can log into this slice
 239     persons = records[0]['persons']
 240
 241     # Extract the keys from persons records
 242     for p in persons:
 243         sliceUser = registry.Resolve(p, cred)
 244         userKeys = sliceUser[0]['keys']
 245         keys += userKeys
 246
 247     return '\n'.join(keys)
 248
 249 ##
 250 # A class that builds the RSpec for Eucalyptus.
 251 #
 252 class EucaRSpecBuilder(object):
 253     ##
 254     # Initizes a RSpec builder
 255     #
 256     # @param cloud A dictionary containing data about a
 257     #              cloud (ex. clusters, ip)
 258     def __init__(self, cloud):
 259         self.eucaRSpec = XMLBuilder(format = True, tab_step = "  ")
 260         self.cloudInfo = cloud
 261
 262     ##
 263     # Creates a request stanza.
 264     #
 265     # @param num The number of instances to create.
 266     # @param image The disk image id.
 267     # @param kernel The kernel image id.
 268     # @param keypair Key pair to embed.
 269     # @param ramdisk Ramdisk id (optional).
 270     #
 271     def __requestXML(self, num, image, kernel, keypair, ramdisk = ''):
 272         xml = self.eucaRSpec
 273         with xml.request:
 274             with xml.instances:
 275                 xml << str(num)
 276             with xml.kernel_image(id=kernel):
 277                 xml << ''
 278             if ramdisk == '':
 279                 with xml.ramdisk:
 280                     xml << ''
 281             else:
 282                 with xml.ramdisk(id=ramdisk):
 283                     xml << ''
 284             with xml.disk_image(id=image):
 285                 xml << ''
 286             with xml.keypair:
 287                 xml << keypair
 288
 289     ##
 290     # Creates the cluster stanza.
 291     #
 292     # @param clusters Clusters information.
 293     #
 294     def __clustersXML(self, clusters):
 295         cloud = self.cloudInfo
 296         xml = self.eucaRSpec
 297
 298         for cluster in clusters:
 299             instances = cluster['instances']
 300             with xml.cluster(id=cluster['name']):
 301                 with xml.ipv4:
 302                     xml << cluster['ip']
 303                 with xml.vm_types:
 304                     for inst in instances:
 305                         with xml.vm_type(name=inst[0]):
 306                             with xml.free_slots:
 307                                 xml << str(inst[1])
 308                             with xml.max_instances:
 309                                 xml << str(inst[2])
 310                             with xml.cores:
 311                                 xml << str(inst[3])
 312                             with xml.memory(unit='MB'):
 313                                 xml << str(inst[4])
 314                             with xml.disk_space(unit='GB'):
 315                                 xml << str(inst[5])
 316                             if 'instances' in cloud and inst[0] in cloud['instances']:
 317                                 existingEucaInstances = cloud['instances'][inst[0]]
 318                                 with xml.euca_instances:
 319                                     for eucaInst in existingEucaInstances:
 320                                         with xml.euca_instance(id=eucaInst['id']):
 321                                             with xml.state:
 322                                                 xml << eucaInst['state']
 323                                             with xml.public_dns:
 324                                                 xml << eucaInst['public_dns']
 325
 326     def __imageBundleXML(self, bundles):
 327         xml = self.eucaRSpec
 328         with xml.bundles:
 329             for bundle in bundles.keys():
 330                 with xml.bundle(id=bundle):
 331                     xml << ''
 332
 333     ##
 334     # Creates the Images stanza.
 335     #
 336     # @param images A list of images in Eucalyptus.
 337     #
 338     def __imagesXML(self, images):
 339         xml = self.eucaRSpec
 340         with xml.images:
 341             for image in images:
 342                 with xml.image(id=image.id):
 343                     with xml.type:
 344                         xml << image.type
 345                     with xml.arch:
 346                         xml << image.architecture
 347                     with xml.state:
 348                         xml << image.state
 349                     with xml.location:
 350                         xml << image.location
 351
 352     ##
 353     # Creates the KeyPairs stanza.
 354     #
 355     # @param keypairs A list of key pairs in Eucalyptus.
 356     #
 357     def __keyPairsXML(self, keypairs):
 358         xml = self.eucaRSpec
 359         with xml.keypairs:
 360             for key in keypairs:
 361                 with xml.keypair:
 362                     xml << key.name
 363
 364     ##
 365     # Generates the RSpec.
 366     #
 367     def toXML(self):
 368         logger = logging.getLogger('EucaAggregate')
 369         if not self.cloudInfo:
 370             logger.error('No cloud information')
 371             return ''
 372
 373         xml = self.eucaRSpec
 374         cloud = self.cloudInfo
 375         with xml.RSpec(type='eucalyptus'):
 376             with xml.network(name=cloud['name']):
 377                 with xml.ipv4:
 378                     xml << cloud['ip']
 379                 #self.__keyPairsXML(cloud['keypairs'])
 380                 #self.__imagesXML(cloud['images'])
 381                 self.__imageBundleXML(cloud['imageBundles'])
 382                 self.__clustersXML(cloud['clusters'])
 383         return str(xml)
 384
 385 ##
 386 # A parser to parse the output of availability-zones.
 387 #
 388 # Note: Only one cluster is supported. If more than one, this will
 389 #       not work.
 390 #
 391 class ZoneResultParser(object):
 392     def __init__(self, zones):
 393         self.zones = zones
 394
 395     def parse(self):
 396         if len(self.zones) < 3:
 397             return
 398         clusterList = []
 399         cluster = {}
 400         instList = []
 401
 402         cluster['name'] = self.zones[0].name
 403         cluster['ip']   = self.zones[0].state
 404
 405         for i in range(2, len(self.zones)):
 406             currZone = self.zones[i]
 407             instType = currZone.name.split()[1]
 408
 409             stateString = currZone.state.split('/')
 410             rscString   = stateString[1].split()
 411
 412             instFree      = int(stateString[0])
 413             instMax       = int(rscString[0])
 414             instNumCpu    = int(rscString[1])
 415             instRam       = int(rscString[2])
 416             instDiskSpace = int(rscString[3])
 417
 418             instTuple = (instType, instFree, instMax, instNumCpu, instRam, instDiskSpace)
 419             instList.append(instTuple)
 420         cluster['instances'] = instList
 421         clusterList.append(cluster)
 422
 423         return clusterList
 424
 425 def ListResources(api, creds, options, call_id):
 426     if Callids().already_handled(call_id): return ""
 427     global cloud
 428     # get slice's hrn from options
 429     xrn = options.get('geni_slice_urn', '')
 430     hrn, type = urn_to_hrn(xrn)
 431     logger = logging.getLogger('EucaAggregate')
 432
 433     # get hrn of the original caller
 434     origin_hrn = options.get('origin_hrn', None)
 435     if not origin_hrn:
 436         origin_hrn = Credential(string=creds[0]).get_gid_caller().get_hrn()
 437
 438     conn = getEucaConnection()
 439
 440     if not conn:
 441         logger.error('Cannot create a connection to Eucalyptus')
 442         return 'Cannot create a connection to Eucalyptus'
 443
 444     try:
 445         # Zones
 446         zones = conn.get_all_zones(['verbose'])
 447         p = ZoneResultParser(zones)
 448         clusters = p.parse()
 449         cloud['clusters'] = clusters
 450
 451         # Images
 452         images = conn.get_all_images()
 453         cloud['images'] = images
 454         cloud['imageBundles'] = {}
 455         for i in images:
 456             if i.type != 'machine' or i.kernel_id is None: continue
 457             name = os.path.dirname(i.location)
 458             detail = {'imageID' : i.id, 'kernelID' : i.kernel_id, 'ramdiskID' : i.ramdisk_id}
 459             cloud['imageBundles'][name] = detail
 460
 461         # Key Pairs
 462         keyPairs = conn.get_all_key_pairs()
 463         cloud['keypairs'] = keyPairs
 464
 465         if hrn:
 466             instanceId = []
 467             instances  = []
 468
 469             # Get the instances that belong to the given slice from sqlite3
 470             # XXX use getOne() in production because the slice's hrn is supposed
 471             # to be unique. For testing, uniqueness is turned off in the db.
 472             # If the slice isn't found in the database, create a record for the
 473             # slice.
 474             matchedSlices = list(Slice.select(Slice.q.slice_hrn == hrn))
 475             if matchedSlices:
 476                 theSlice = matchedSlices[-1]
 477             else:
 478                 theSlice = Slice(slice_hrn = hrn)
 479             for instance in theSlice.instances:
 480                 instanceId.append(instance.instance_id)
 481
 482             # Get the information about those instances using their ids.
 483             if len(instanceId) > 0:
 484                 reservations = conn.get_all_instances(instanceId)
 485             else:
 486                 reservations = []
 487             for reservation in reservations:
 488                 for instance in reservation.instances:
 489                     instances.append(instance)
 490
 491             # Construct a dictionary for the EucaRSpecBuilder
 492             instancesDict = {}
 493             for instance in instances:
 494                 instList = instancesDict.setdefault(instance.instance_type, [])
 495                 instInfoDict = {}
 496
 497                 instInfoDict['id'] = instance.id
 498                 instInfoDict['public_dns'] = instance.public_dns_name
 499                 instInfoDict['state'] = instance.state
 500                 instInfoDict['key'] = instance.key_name
 501
 502                 instList.append(instInfoDict)
 503             cloud['instances'] = instancesDict
 504
 505     except EC2ResponseError, ec2RespErr:
 506         errTree = ET.fromstring(ec2RespErr.body)
 507         errMsgE = errTree.find('.//Message')
 508         logger.error(errMsgE.text)
 509
 510     rspec = EucaRSpecBuilder(cloud).toXML()
 511
 512     # Remove the instances records so next time they won't
 513     # show up.
 514     if 'instances' in cloud:
 515         del cloud['instances']
 516
 517     return rspec
 518
 519 """
 520 Hook called via 'sfi.py create'
 521 """
 522 def CreateSliver(api, slice_xrn, creds, xml, users, call_id):
 523     if Callids().already_handled(call_id): return ""
 524
 525     global cloud
 526     logger = logging.getLogger('EucaAggregate')
 527     logger.debug("In CreateSliver")
 528
 529     aggregate = Aggregate(api)
 530     slices = Slices(api)
 531     (hrn, type) = urn_to_hrn(slice_xrn)
 532     peer = slices.get_peer(hrn)
 533     sfa_peer = slices.get_sfa_peer(hrn)
 534     slice_record=None
 535     if users:
 536         slice_record = users[0].get('slice_record', {})
 537
 538     conn = getEucaConnection()
 539     if not conn:
 540         logger.error('Cannot create a connection to Eucalyptus')
 541         return ""
 542
 543     # Validate RSpec
 544     schemaXML = ET.parse(EUCALYPTUS_RSPEC_SCHEMA)
 545     rspecValidator = ET.RelaxNG(schemaXML)
 546     rspecXML = ET.XML(xml)
 547     for network in rspecXML.iterfind("./network"):
 548         if network.get('name') != cloud['name']:
 549             # Throw away everything except my own RSpec
 550             # sfa_logger().error("CreateSliver: deleting %s from rspec"%network.get('id'))
 551             network.getparent().remove(network)
 552     if not rspecValidator(rspecXML):
 553         error = rspecValidator.error_log.last_error
 554         message = '%s (line %s)' % (error.message, error.line)
 555         raise InvalidRSpec(message)
 556
 557     """
 558     Create the sliver[s] (slice) at this aggregate.
 559     Verify HRN and initialize the slice record in PLC if necessary.
 560     """
 561
 562     # ensure site record exists
 563     site = slices.verify_site(hrn, slice_record, peer, sfa_peer)
 564     # ensure slice record exists
 565     slice = slices.verify_slice(hrn, slice_record, peer, sfa_peer)
 566     # ensure person records exists
 567     persons = slices.verify_persons(hrn, slice, users, peer, sfa_peer)
 568
 569     # Get the slice from db or create one.
 570     s = Slice.select(Slice.q.slice_hrn == hrn).getOne(None)
 571     if s is None:
 572         s = Slice(slice_hrn = hrn)
 573
 574     # Process any changes in existing instance allocation
 575     pendingRmInst = []
 576     for sliceInst in s.instances:
 577         pendingRmInst.append(sliceInst.instance_id)
 578     existingInstGroup = rspecXML.findall(".//euca_instances")
 579     for instGroup in existingInstGroup:
 580         for existingInst in instGroup:
 581             if existingInst.get('id') in pendingRmInst:
 582                 pendingRmInst.remove(existingInst.get('id'))
 583     for inst in pendingRmInst:
 584         dbInst = EucaInstance.select(EucaInstance.q.instance_id == inst).getOne(None)
 585         if dbInst.meta.state != 'deleted':
 586             logger.debug('Instance %s will be terminated' % inst)
 587             # Terminate instances one at a time for robustness
 588             conn.terminate_instances([inst])
 589             # Only change the state but do not remove the entry from the DB.
 590             dbInst.meta.state = 'deleted'
 591             #dbInst.destroySelf()
 592
 593     # Process new instance requests
 594     requests = rspecXML.findall(".//request")
 595     if requests:
 596         # Get all the public keys associate with slice.
 597         keys = []
 598         for user in users:
 599             keys += user['keys']
 600             logger.debug("Keys: %s" % user['keys'])
 601         pubKeys = '\n'.join(keys)
 602         logger.debug('Passing the following keys to the instance:\n%s' % pubKeys)
 603     for req in requests:
 604         vmTypeElement = req.getparent()
 605         instType = vmTypeElement.get('name')
 606         numInst  = int(req.find('instances').text)
 607
 608         bundleName = req.find('bundle').text
 609         if not cloud['imageBundles'][bundleName]:
 610             logger.error('Cannot find bundle %s' % bundleName)
 611         bundleInfo = cloud['imageBundles'][bundleName]
 612         instKernel  = bundleInfo['kernelID']
 613         instDiskImg = bundleInfo['imageID']
 614         instRamDisk = bundleInfo['ramdiskID']
 615         instKey     = None
 616
 617         # Create the instances
 618         for i in range(0, numInst):
 619             eucaInst = EucaInstance(slice      = s,
 620                                     kernel_id  = instKernel,
 621                                     image_id   = instDiskImg,
 622                                     ramdisk_id = instRamDisk,
 623                                     key_pair   = instKey,
 624                                     inst_type  = instType,
 625                                     meta       = Meta(start_time=datetime.datetime.now()))
 626             eucaInst.reserveInstance(conn, pubKeys)
 627
 628     # xxx - should return altered rspec
 629     # with enough data for the client to understand what's happened
 630     return xml
 631
 632 ##
 633 # Return information on the IP addresses bound to each slice's instances
 634 #
 635 def dumpInstanceInfo():
 636     logger = logging.getLogger('EucaMeta')
 637     outdir = "/var/www/html/euca/"
 638     outfile = outdir + "instances.txt"
 639
 640     try:
 641         os.makedirs(outdir)
 642     except OSError, e:
 643         if e.errno != errno.EEXIST:
 644             raise
 645
 646     dbResults = Meta.select(
 647         AND(Meta.q.pri_addr != None,
 648             Meta.q.state    == 'running')
 649         )
 650     dbResults = list(dbResults)
 651     f = open(outfile, "w")
 652     for r in dbResults:
 653         instId = r.instance.instance_id
 654         ipaddr = r.pri_addr
 655         hrn = r.instance.slice.slice_hrn
 656         logger.debug('[dumpInstanceInfo] %s %s %s' % (instId, ipaddr, hrn))
 657         f.write("%s %s %s\n" % (instId, ipaddr, hrn))
 658     f.close()
 659
 660 ##
 661 # A separate process that will update the meta data.
 662 #
 663 def updateMeta():
 664     logger = logging.getLogger('EucaMeta')
 665     fileHandler = logging.FileHandler('/var/log/euca_meta.log')
 666     fileHandler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
 667     logger.addHandler(fileHandler)
 668     fileHandler.setLevel(logging.DEBUG)
 669     logger.setLevel(logging.DEBUG)
 670
 671     while True:
 672         sleep(30)
 673
 674         # Get IDs of the instances that don't have IPs yet.
 675         dbResults = Meta.select(
 676                       AND(Meta.q.pri_addr == None,
 677                           Meta.q.state    != 'deleted')
 678                     )
 679         dbResults = list(dbResults)
 680         logger.debug('[update process] dbResults: %s' % dbResults)
 681         instids = []
 682         for r in dbResults:
 683             if not r.instance:
 684                 continue
 685             instids.append(r.instance.instance_id)
 686         logger.debug('[update process] Instance Id: %s' % ', '.join(instids))
 687
 688         # Get instance information from Eucalyptus
 689         conn = getEucaConnection()
 690         vmInstances = []
 691         reservations = conn.get_all_instances(instids)
 692         for reservation in reservations:
 693             vmInstances += reservation.instances
 694
 695         # Check the IPs
 696         instIPs = [ {'id':i.id, 'pri_addr':i.private_dns_name, 'pub_addr':i.public_dns_name}
 697                     for i in vmInstances if i.private_dns_name != '0.0.0.0' ]
 698         logger.debug('[update process] IP dict: %s' % str(instIPs))
 699
 700         # Update the local DB
 701         for ipData in instIPs:
 702             dbInst = EucaInstance.select(EucaInstance.q.instance_id == ipData['id']).getOne(None)
 703             if not dbInst:
 704                 logger.info('[update process] Could not find %s in DB' % ipData['id'])
 705                 continue
 706             dbInst.meta.pri_addr = ipData['pri_addr']
 707             dbInst.meta.pub_addr = ipData['pub_addr']
 708             dbInst.meta.state    = 'running'
 709
 710         dumpInstanceInfo()
 711
 712 def GetVersion(api):
 713     xrn=Xrn(api.hrn)
 714     request_rspec_versions = [dict(sfa_rspec_version)]
 715     ad_rspec_versions = [dict(sfa_rspec_version)]
 716     version_more = {'interface':'aggregate',
 717                     'testbed':'myplc',
 718                     'hrn':xrn.get_hrn(),
 719                     'request_rspec_versions': request_rspec_versions,
 720                     'ad_rspec_versions': ad_rspec_versions,
 721                     'default_ad_rspec': dict(sfa_rspec_version)
 722                     }
 723     return version_core(version_more)
 724
 725 #def main():
 726 #    init_server()
 727 #
 728 #    #theRSpec = None
 729 #    #with open(sys.argv[1]) as xml:
 730 #    #    theRSpec = xml.read()
 731 #    #CreateSliver(None, 'planetcloud.pc.test', theRSpec, 'call-id-cloudtest')
 732 #
 733 #    #rspec = ListResources('euca', 'planetcloud.pc.test', 'planetcloud.pc.marcoy', 'test_euca')
 734 #    #print rspec
 735 #
 736 #    server_key_file = '/var/lib/sfa/authorities/server.key'
 737 #    server_cert_file = '/var/lib/sfa/authorities/server.cert'
 738 #    api = PlcSfaApi(key_file = server_key_file, cert_file = server_cert_file, interface='aggregate')
 739 #    print getKeysForSlice(api, 'gc.gc.test1')
 740 #
 741 #if __name__ == "__main__":
 742 #    main()
 743