added getCredentialFromReistry() method
[sfa.git] / geni / slicemgr.py
index 8b9e925..9138923 100644 (file)
@@ -6,27 +6,30 @@ import time
 from geni.util.geniserver import *
 from geni.util.geniclient import *
 from geni.util.cert import *
+from geni.util.credential import Credential
 from geni.util.trustedroot import *
 from geni.util.excep import *
 from geni.util.misc import *
 from geni.util.config import Config
+from geni.util.rspec import Rspec
+from geni.util.specdict import *
+from geni.util.storage import SimpleStorage
 
 class SliceMgr(GeniServer):
 
     hrn = None
-    key_file = None
-    cert_file = None
-    components_file = None
-    slices_file = None    
-    components_ttl = None
-    components = []
-    slices = []    
-    policy = {}
+    nodes_ttl = None
+    nodes = None
+    slices = None
+    policy = None
+    aggregates = None
     timestamp = None
     threshold = None    
     shell = None
-    aggregates = {}
-    
+    registry = None
+    key_file = None
+    cert_file = None
+    credential = None 
   
     ##
     # Create a new slice manager object.
@@ -36,27 +39,81 @@ class SliceMgr(GeniServer):
     # @param key_file private key filename of registry
     # @param cert_file certificate filename containing public key (could be a GID file)     
 
-    def __init__(self, ip, port, key_file, cert_file, config = "/usr/share/geniwrapper/util/geni_config"):
-        GeniServer.__init__(ip, port, key_file, cert_file)
+    def __init__(self, ip, port, key_file, cert_file, config = "/usr/share/geniwrapper/geni/util/geni_config"):
+        GeniServer.__init__(self, ip, port, key_file, cert_file)
         self.key_file = key_file
         self.cert_file = cert_file
-        self.conf = Config(config)
-        basedir = self.conf.GENI_BASE_DIR + os.sep
-        server_basedir = basedir + os.sep + "plc" + os.sep
-        self.hrn = conf.GENI_INTERFACE_HRN
-    
+        self.config = Config(config)
+        self.basedir = self.config.GENI_BASE_DIR + os.sep
+        self.server_basedir = self.basedir + os.sep + "geni" + os.sep
+        self.hrn = self.config.GENI_INTERFACE_HRN    
+        self.time_format = "%Y-%m-%d %H:%M:%S"
+        
         # Get list of aggregates this sm talks to
-        aggregates_file = server_basedir + os.sep + 'aggregates'
-        self.load_aggregates(aggregates_file) 
-        self.components_file = os.sep.join([server_basedir, 'components', 'slicemgr.' + hrn + '.comp'])
-        self.slices_file = os.sep.join([server_basedir, 'components', 'slicemgr' + hrn + '.slices'])
-        self.timestamp_file = os.sep.join([server_basedir, 'components', 'slicemgr' + hrn + '.timestamp']) 
-        self.components_ttl = components_ttl
-        self.policy['whitelist'] = []
-        self.policy['blacklist'] = []
-        self.connect()
-
-    def load_aggregates(self, aggregates_file):
+        # XX do we use simplestorage to maintain this file manually?
+        aggregates_file = self.server_basedir + os.sep + 'aggregates'
+        self.aggregates = SimpleStorage(aggregates_file)
+        
+        nodes_file = os.sep.join([self.server_basedir, 'smgr.' + self.hrn + '.components'])
+        self.nodes = SimpleStorage(nodes_file)
+        self.nodes.load()
+        
+        slices_file = os.sep.join([self.server_basedir, 'smgr.' + self.hrn + '.slices'])
+        self.slices = SimpleStorage(slices_file)
+        self.slices.load()
+
+        policy_file = os.sep.join([self.server_basedir, 'smgr.' + self.hrn + '.policy'])
+        self.policy = SimpleStorage(policy_file, {'whitelist': [], 'blacklist': []})
+        self.policy.load()
+
+        # How long before we refresh nodes cache  
+        self.nodes_ttl = 1
+
+        self.connectRegistry()
+        self.loadCredential()
+        self.connectAggregates(aggregates_file)
+
+
+    def loadCredential(self):
+        """
+        Attempt to load credential from file if it exists. If it doesnt get
+        credential from registry.
+        """
+        
+        # see if this file exists
+        ma_cred_filename = self.server_basedir + os.sep + "smgr." + self.hrn + ".sa.cred"
+        try:
+            self.credential = Credential(filename = ma_cred_filename)
+        except IOError:
+            self.credential = self.getCrednetialFromRegistry()
+            
+        
+    def getCredentialFromRegistry(self):
+        """
+        Get our current credential from the registry.
+        """
+        # get self credential
+        self_cred_filename = self.server_basedir + os.sep + "smgr." + self.hrn + ".cred"
+        self_cred = self.registry.get_credential(None, 'ma', self.hrn)
+        self_cred.save_to_file(self_cred_filename, save_parents=True)
+
+        # get ma credential
+        ma_cred_filename = self.server_basedir + os.sep + "smgr." + self.hrn + ".sa.cred"
+        ma_cred = self.registry.get_credential(self_cred, 'sa', self.hrn)
+        ma_cred.save_to_file(ma_cred_filename, save_parents=True)
+        return ma_cred        
+
+    def connectRegistry(self):
+        """
+        Connect to the registry
+        """
+        # connect to registry using GeniClient
+        address = self.config.GENI_REGISTRY_HOSTNAME
+        port = self.config.GENI_REGISTRY_PORT
+        url = 'http://%(address)s:%(port)s' % locals()
+        self.registry = GeniClient(url, self.key_file, self.cert_file)
+
+    def connectAggregates(self, aggregates_file):
         """
         Get info about the aggregates available to us from file and create 
         an xmlrpc connection to each. If any info is invalid, skip it. 
@@ -67,12 +124,13 @@ class SliceMgr(GeniServer):
             lines = f.readlines()
             f.close()
         except: raise 
-    
+        
         for line in lines:
             # Skip comments
-            if line.strip.startswith("#"):
+            if line.strip().startswith("#"):
                 continue
-            agg_info = line.split("\t").split(" ")
+            line = line.replace("\t", " ").replace("\n", "").replace("\r", "").strip()
+            agg_info = line.split(" ")
         
             # skip invalid info
             if len(agg_info) != 3:
@@ -80,10 +138,9 @@ class SliceMgr(GeniServer):
 
             # create xmlrpc connection using GeniClient
             hrn, address, port = agg_info[0], agg_info[1], agg_info[2]
-            url = 'https://%(address)s:%(port)s' % locals()
+            url = 'http://%(address)s:%(port)s' % locals()
             self.aggregates[hrn] = GeniClient(url, self.key_file, self.cert_file)
 
-
     def item_hrns(self, items):
         """
         Take a list of items (components or slices) and return a dictionary where
@@ -120,213 +177,226 @@ class SliceMgr(GeniServer):
         """
         Update the cached list of nodes.
         """
-        print "refreshing"
     
+        # convert and threshold to ints
+        if self.nodes.has_key('timestamp') and self.nodes['timestamp']:
+            hr_timestamp = self.nodes['timestamp']
+            timestamp = datetime.datetime.fromtimestamp(time.mktime(time.strptime(hr_timestamp, self.time_format)))
+            hr_threshold = self.nodes['threshold']
+            threshold = datetime.datetime.fromtimestamp(time.mktime(time.strptime(hr_threshold, self.time_format)))
+        else:
+            timestamp = datetime.datetime.now()
+            hr_timestamp = timestamp.strftime(self.time_format)
+            delta = datetime.timedelta(hours=self.nodes_ttl)
+            threshold = timestamp + delta
+            hr_threshold = threshold.strftime(self.time_format)
+
+        start_time = int(timestamp.strftime("%s"))
+        end_time = int(threshold.strftime("%s"))
+        duration = end_time - start_time
+
         aggregates = self.aggregates.keys()
-        all_nodes = []
-        all_slices = []
+        rspecs = {}
+        networks = []
+        rspec = Rspec()
         for aggregate in aggregates:
             try:
-                # resolve components hostnames
-                nodes = self.aggregates[aggregate].get_components()
-                all_nodes.extend(nodes)    
-                # update timestamp and threshold
-                self.timestamp = datetime.datetime.now()
-                delta = datetime.timedelta(hours=self.components_ttl)
-                self.threshold = self.timestamp + delta 
+                # get the rspec from the aggregate
+                agg_rspec = self.aggregates[aggregate].list_nodes(self.credential)
+                # extract the netspec from each aggregates rspec
+                rspec.parseString(agg_rspec)
+                networks.extend([{'NetSpec': rspec.getDictsByTagName('NetSpec')}])
             except:
                 # XX print out to some error log
-                pass    
-   
-        self.components = all_nodes
-        f = open(self.components_file, 'w')
-        f.write(str(self.components))
-        f.close()
-        f = open(self.timestamp_file, 'w')
-        f.write(str(self.threshold))
-        f.close()
-    def load_components(self):
-        """
-        Read cached list of nodes and slices.
-        """
-        print "loading nodes"
-        # Read component list from cached file 
-        if os.path.exists(self.components_file):
-            f = open(self.components_file, 'r')
-            self.components = eval(f.read())
-            f.close()
-    
-        time_format = "%Y-%m-%d %H:%M:%S"
-        if os.path.exists(self.timestamp_file):
-            f = open(self.timestamp_file, 'r')
-            timestamp = str(f.read()).split(".")[0]
-            self.timestamp = datetime.datetime.fromtimestamp(time.mktime(time.strptime(timestamp, time_format)))
-            delta = datetime.timedelta(hours=self.components_ttl)
-            self.threshold = self.timestamp + delta
-            f.close()
+                print "Error calling list nodes at aggregate %s" % aggregate
+                raise    
+  
+        # create the rspec dict
+        resources = {'networks': networks, 'start_time': start_time, 'duration': duration}
+        resourceDict = {'Rspec': resources} 
+        # convert rspec dict to xml
+        rspec.parseDict(resourceDict)
+       
+        # filter according to policy
+        rspec.filter('NodeSpec', 'name', blacklist=self.policy['blacklist'], whitelist=self.policy['whitelist'])
+
+        # update timestamp and threshold
+        timestamp = datetime.datetime.now()
+        hr_timestamp = timestamp.strftime(self.time_format)
+        delta = datetime.timedelta(hours=self.nodes_ttl)
+        threshold = timestamp + delta
+        hr_threshold = threshold.strftime(self.time_format)
+        
+        nodedict = {'rspec': rspec.toxml(),
+                    'timestamp': hr_timestamp,
+                    'threshold':  hr_threshold}
+
+        self.nodes = SimpleStorage(self.nodes.db_filename, nodedict)
+        self.nodes.write()
 
     def load_policy(self):
         """
         Read the list of blacklisted and whitelisted nodes.
         """
-        whitelist = []
-        blacklist = []
-        if os.path.exists(self.whitelist_file):
-            f = open(self.whitelist_file, 'r')
-            lines = f.readlines()
-            f.close()
-            for line in lines:
-                line = line.strip().replace(" ", "").replace("\n", "")
-                whitelist.extend(line.split(","))
-
-
-        if os.path.exists(self.blacklist_file):
-            f = open(self.blacklist_file, 'r')
-            lines = f.readlines()
-            f.close()
-            for line in lines:
-                line = line.strip().replace(" ", "").replace("\n", "")
-                blacklist.extend(line.split(","))
-
-        self.policy['whitelist'] = whitelist
-        self.policy['blacklist'] = blacklist
+        self.policy.load()
  
     def load_slices(self):
         """
-         Read current slice instantiation states.
-        """
-        print "loading slices"
-        if os.path.exists(self.slices_file):
-            f = open(self.components_file, 'r')
-            self.slices = eval(f.read())
-            f.close()    
-
-    def write_slices(self):
-        """
-        Write current slice instantiations to file.
+        Read current slice instantiation states.
         """
-        print "writing slices"
-        f = open(self.slices_file, 'w')
-        f.write(str(self.slices))
-        f.close()
+        self.slices.load()
 
 
-    def get_components(self):
+    def getNodes(self, format = 'rspec'):
         """
         Return a list of components managed by this slice manager.
         """
         # Reload components list
-        now = datetime.datetime.now()
-        #self.load_components()
-        if not self.threshold or not self.timestamp or now > self.threshold:
+        if not self.nodes.has_key('threshold') or not self.nodes['threshold'] or not self.nodes.has_key('timestamp') or not self.nodes['timestamp']:
             self.refresh_components()
-        elif now < self.threshold and not self.components: 
-            self.load_components()
-        return self.components
+        else:
+            now = datetime.datetime.now()
+            threshold = datetime.datetime.fromtimestamp(time.mktime(time.strptime(self.nodes['threshold'], self.time_format)))
+            if  now > threshold:
+                self.refresh_components()
+        return self.nodes[format]
    
      
-    def get_slices(self):
+    def getSlices(self):
         """
         Return a list of instnatiated managed by this slice manager.
         """
-        now = datetime.datetime.now()
-        #self.load_components()
-        if not self.threshold or not self.timestamp or now > self.threshold:
-            self.refresh_components()
-        elif now < self.threshold and not self.slices:
-            self.load_components()
-        return self.slices
-
-    def get_slivers(self, hrn):
-        """
-        Return the list of slices instantiated at the specified component.
-        """
-
-        # hrn is assumed to be a component hrn
-        if hrn not in self.slices:
-            raise RecordNotFound(hrn)
-    
-        return self.slices[hrn]
+        slice_hrns = []
+        for aggregate in self.aggregates:
+            try:
+                slices = self.aggregates[aggregate].list_slices(self.credential)
+                slice_hrns.extend(slices)
+            except:
+                raise
+                # print to some error log
+                pass
 
-    def get_rspec(self, hrn, type):
-        #rspec = Rspec()
-        if type in ['node']:
-            nodes = self.shell.GetNodes(self.auth)
-        elif type in ['slice']:
-            slices = self.shell.GetSlices(self.auth)
-        elif type in ['aggregate']:
-            pass
+        return slice_hrns
 
-    def get_resources(self, slice_hrn):
+    def getResources(self, slice_hrn):
         """
         Return the current rspec for the specified slice.
         """
-        slicename = hrn_to_plcslicename(slice_hrn)
-        rspec = self.get_rspec(slicenamem, 'slice' )
-        
-        return rspec
+
+        # request this slices state from all known aggregates
+        rspec = Rspec()
+        rspecdicts = []
+        networks = []
+        for hrn in self.aggregates.keys():
+            # check if the slice has resources at this hrn
+            slice_resources = self.aggregates[hrn].get_resources(self.credential, slice_hrn)
+            rspec.parseString(slice_resources)
+            networks.extend({'NetSpec': rspec.getDictsByTagName('NetSpec')})
+            
+        # merge all these rspecs into one
+        start_time = int(datetime.datetime.now().strftime("%s"))
+        end_time = start_time
+        duration = end_time - start_time
+    
+        resources = {'networks': networks, 'start_time': start_time, 'duration': duration}
+        resourceDict = {'Rspec': resources}
+        # convert rspec dict to xml
+        rspec.parseDict(resourceDict)
+        # save this slices resources
+        #self.slices[slice_hrn] = rspec.toxml()
+        #self.slices.write()
+         
+        return rspec.toxml()
  
-    def create_slice(self, slice_hrn, rspec, attributes):
+    def createSlice(self, cred, slice_hrn, rspec):
         """
         Instantiate the specified slice according to whats defined in the rspec.
         """
+
+        # save slice state locally
+        # we can assume that spec object has been validated so its safer to
+        # save this instead of the unvalidated rspec the user gave us
+        rspec = Rspec()
+        tempspec = Rspec()
+        rspec.parseString(rspec)
+
+        self.slices[slice_hrn] = rspec.toxml()
+        self.slices.write()
+
+        # extract network list from the rspec and create a separate
+        # rspec for each network
         slicename = self.hrn_to_plcslicename(slice_hrn)
-        #spec = Rspec(rspec)
-        node_hrns = []
-        #for netspec in spec['networks]:
-        #    networkname = netspec['name']
-        #    nodespec = spec['networks']['nodes']
-        #    nodes = [nspec['name'] for nspec in nodespec]
-        #    node_hrns = [networkname + node for node in nodes]
-        #    
-        self.db.AddSliceToNodes(slice_hrn, node_hrns)
+        specDict = rspec.toDict()
+        start_time = specDict['start_time']
+        end_time = specDict['end_time']
+
+        rspecs = {}
+        # only attempt to extract information about the aggregates we know about
+        for hrn in self.aggregates.keys():
+            netspec = spec.getDictByTagNameValue('NetSpec', hrn)
+            if netspec:
+                # creat a plc dict 
+                resources = {'start_time': star_time, 'end_time': end_time, 'networks': netspec}
+                resourceDict = {'Rspec': resources}
+                tempspec.parseDict(resourceDict)
+                rspecs[hrn] = tempspec.toxml()
+
+        # notify the aggregates
+        for hrn in self.rspecs.keys():
+            self.aggregates[hrn].createSlice(self.credential, rspecs[hrn])
+            
         return 1
+
+    def updateSlice(self, slice_hrn, rspec, attributes = []):
+        """
+        Update the specifed slice
+        """
+        self.create_slice(slice_hrn, rspec, attributes)
     
-    def delete_slice_(self, slice_hrn):
+    def deleteSlice_(self, slice_hrn):
         """
         Remove this slice from all components it was previouly associated with and 
         free up the resources it was using.
         """
-        self.db.DeleteSliceFromNodes(self.auth, slicename, self.components)
+        # XX need to get the correct credential
+        cred = self.credential
+        
+        if self.slices.has_key(slice_hrn):
+            self.slices.pop(slice_hrn)
+            self.slices.write()
+
+        for hrn in self.aggregates.keys():
+            self.aggregates[hrn].deleteSlice(cred, slice_hrn)
+
         return 1
 
-    def start_slice(self, slice_hrn):
+    def startSlice(self, slice_hrn):
         """
         Stop the slice at plc.
         """
-        slicename = hrn_to_plcslicename(slice_hrn)
-        slices = self.shell.GetSlices(self.auth, {'name': slicename}, ['slice_id'])
-        if not slices:
-            raise RecordNotFound(slice_hrn)
-        slice_id = slices[0]
-        atrribtes = self.shell.GetSliceAttributes({'slice_id': slice_id, 'name': 'enabled'}, ['slice_attribute_id'])
-        attribute_id = attreibutes[0] 
-        self.shell.UpdateSliceAttribute(self.auth, attribute_id, "1" )
+        cred = self.credential
+
+        for hrn in self.aggregates.keys():
+            self.aggregates[hrn].startSlice(cred, slice_hrn)
         return 1
 
-    def stop_slice(self, slice_hrn):
+    def stopSlice(self, slice_hrn):
         """
         Stop the slice at plc
         """
-        slicename = hrn_to_plcslicename(slice_hrn)
-        slices = self.shell.GetSlices(self.auth, {'name': slicename}, ['slice_id'])
-        if not slices:
-            raise RecordNotFound(slice_hrn)
-        slice_id = slices[0]
-        atrribtes = self.shell.GetSliceAttributes({'slice_id': slice_id, 'name': 'enabled'}, ['slice_attribute_id'])
-        attribute_id = attreibutes[0]
-        self.shell.UpdateSliceAttribute(self.auth, attribute_id, "0")
+        cred = self.credential
+        for hrn in self.aggregates.keys():
+            self.aggregates[hrn].startSlice(cred, slice_hrn)
         return 1
 
-    def reset_slice(self, slice_hrn):
+    def resetSlice(self, slice_hrn):
         """
         Reset the slice
         """
-        slicename = self.hrn_to_plcslicename(slice_hrn)
+        # XX not yet implemented
         return 1
 
-    def get_policy(self):
+    def getPolicy(self):
         """
         Return the policy of this slice manager.
         """
@@ -339,55 +409,57 @@ class SliceMgr(GeniServer):
 ## Server methods here for now
 ##############################
 
-    def nodes(self):
-        return self..get_components()
+    def list_nodes(self, cred):
+        self.decode_authentication(cred, 'listnodes')
+        return self.getNodes()
 
-    def slices(self):
-        return self.get_slices()
+    def list_slices(self, cred):
+        self.decode_authentication(cred, 'listslices')
+        return self.getSlices()
 
-    def resources(self, cred, hrn):
-        self.decode_authentication(cred, 'info')
-        self.verify_object_belongs_to_me(hrn)
+    def get_resources(self, cred, hrn):
+        self.decode_authentication(cred, 'listnodes')
+        return self.getResources(hrn)
 
-        return self.get_resources(hrn)
+    def get_ticket(self, cred, hrn, rspec):
+        self.decode_authentication(cred, 'getticket')
+        return self.getTicket(hrn, rspec)
 
-    def create(self, cred, hrn, rspec):
-        self.decode_authentication(cred, 'embed')
-        self.verify_object_belongs_to_me(hrn, rspec)
-        return self.create(hrn)
+    def get_policy(self, cred):
+        self.decode_authentication(cred, 'getpolicy')
+        return self.getPolicy()
 
-    def delete(self, cred, hrn):
-        self.decode_authentication(cred, 'embed')
-        self.verify_object_belongs_to_me(hrn)
-        return self.delete_slice(hrn)
+    def create_slice(self, cred, hrn, rspec):
+        self.decode_authentication(cred, 'creatslice')
+        return self.createSlice(cred, hrn, rspec)
 
-    def start(self, cred, hrn):
-        self.decode_authentication(cred, 'control')
-        return self.start(hrn)
+    def delete_slice(self, cred, hrn):
+        self.decode_authentication(cred, 'deleteslice')
+        return self.deleteSlice(hrn)
 
-    def stop(self, cred, hrn):
-        self.decode_authentication(cred, 'control')
-        return self.stop(hrn)
+    def start_slice(self, cred, hrn):
+        self.decode_authentication(cred, 'startslice')
+        return self.startSlice(hrn)
 
-    def reset(self, cred, hrn):
-        self.decode_authentication(cred, 'control')
-        return self.reset(hrn)
+    def stop_slice(self, cred, hrn):
+        self.decode_authentication(cred, 'stopslice')
+        return self.stopSlice(hrn)
 
-    def policy(self, cred):
-        self.decode_authentication(cred, 'info')
-        return self.get_policy()
+    def reset_slice(self, cred, hrn):
+        self.decode_authentication(cred, 'resetslice')
+        return self.resetSlice(hrn)
 
     def register_functions(self):
         GeniServer.register_functions(self)
 
         # Aggregate interface methods
-        self.server.register_function(self.components)
-        self.server.register_function(self.slices)
-        self.server.register_function(self.resources)
-        self.server.register_function(self.create)
-        self.server.register_function(self.delete)
-        self.server.register_function(self.start)
-        self.server.register_function(self.stop)
-        self.server.register_function(self.reset)
-        self.server.register_function(self.policy)
+        self.server.register_function(self.list_nodes)
+        self.server.register_function(self.list_slices)
+        self.server.register_function(self.get_resources)
+        self.server.register_function(self.get_policy)
+        self.server.register_function(self.create_slice)
+        self.server.register_function(self.delete_slice)
+        self.server.register_function(self.start_slice)
+        self.server.register_function(self.stop_slice)
+        self.server.register_function(self.reset_slice)