Changeset d81971a for fedd

Timestamp:

Nov 17, 2008 6:19:40 PM (16 years ago)

Author:

Ted Faber <faber@…>

Branches:

axis_example, compt_changes, info-ops, master, version-1.30, version-2.00, version-3.01, version-3.02

Children:

7583a62

Parents:

afa43a8

Message:

checkpoint of the resource management stuff

Location:

fedd

Files:

• fedd_access.py (modified) (8 diffs)
• fedd_bindings.wsdl (modified) (1 diff)
• fedd_experiment_control.py (modified) (6 diffs)
• fedd_messages.wsdl (modified) (2 diffs)
• fedd_types.xsd (modified) (1 diff)

fedd/fedd_access.py

@@ -14 +14 @@
 import string
 import copy
+import pickle
 
 from fedd_access_project import access_project
@@ -224 +225 @@
         self.restricted = [ ]
         self.fedid_category = { }
+        self.projects = { }
+        self.keys = { }
+        self.allocation = { }
+        self.state = { 
+            'projects': self.projects,
+            'allocation' : self.allocation,
+            'keys' : self.keys
+        }
         self.fedid_default = "testbed"
         if config.has_option("access", "accessdb"):
@@ -230 +239 @@
             self.read_trust(config.get("access", "trustdb"))
 
+        self.state_filename = config.get("access", "access_state", "")
         self.log = logging.getLogger("fedd.access")
+        self.read_state()
+
 
         # Certs are promoted from the generic to the specific, so without a
@@ -279 +291 @@
                 RequestAccessRequestMessage.typecode,\
                 self.RequestAccess, RequestAccessResponseMessage,\
-                "RequestAccessResponseBody")\
+                "RequestAccessResponseBody"), \
+            'ReleaseAccess': make_soap_handler(\
+                ReleaseAccessRequestMessage.typecode,\
+                self.ReleaseAccess, ReleaseAccessResponseMessage,\
+                "ReleaseAccessResponseBody")\
             }
         self.xmlrpc_services =  {\
             'RequestAccess': make_xmlrpc_handler(\
-                self.RequestAccess, "RequestAccessResponseBody")\
+                self.RequestAccess, "RequestAccessResponseBody"),\
+            'ReleaseAccess': make_xmlrpc_handler(\
+                self.ReleaseAccess, "ReleaseAccessResponseBody")\
             }
 
@@ -326 +344 @@
         else:
             return None
+
+    def write_state(self):
+        try:
+            f = open(self.state_filename, 'w')
+            pickle.dump(self.state, f)
+        except IOError, e:
+            self.log.error("Can't write file %s: %s" % \
+                    (self.state_filename, e))
+        except pickle.PicklingError, e:
+            self.log.error("Pickling problem: %s" % e)
+        except TypeError, e:
+            self.log.error("Pickling problem (TypeError): %s" % e)
+
+
+    def read_state(self):
+        """
+        Read a new copy of access state.  Old state is overwritten.
+
+        State format is a simple pickling of the state dictionary.
+        """
+        try:
+            f = open(self.state_filename, "r")
+            self.state = pickle.load(f)
+
+            self.allocation = self.state['allocation']
+            self.projects = self.state['projects']
+            self.keys = self.state['keys']
+
+            self.log.debug("[read_state]: Read state from %s" % \
+                    self.state_filename)
+        except IOError, e:
+            self.log.warning("[read_state]: No saved state: Can't open %s: %s"\
+                    % (self.state_filename, e))
+        except EOFError, e:
+            self.log.warning("[read_state]: Empty or damaged state file: %s:"\
+                    % self.state_filename)
+        except pickle.UnpicklingError, e:
+            self.log.warning(("[read_state]: No saved state: " + \
+                    "Unpickling failed: %s") % e)
+
 
     def proxy_xmlrpc_request(self, dt, req):
@@ -649 +707 @@
                     ap = self.allocate_project.dynamic_project(preq)
                 else:
-                    # XXX ssh key additions
                     preq = {'StaticProjectRequestBody' : \
                             { 'project': \
@@ -677 +734 @@
                 raise service_error(service_error.req, 
                         "SSH access parameters required")
-
+            # keep track of what's been added
+            if req['allocID'].has_key('fedid'):
+                aid = unicode(req['allocId']['fedid'])
+            elif req['allocID'].has_key('localname'):
+                aid = req['allocID']['localname']
+            else:
+                raise service_error(service_error.req, "Bad allocation ID")
+
+            self.allocation[aid] = { }
+            if dyn[1]:
+                try:
+                    pname = ap['project']['name']['localname']
+                except KeyError:
+                    raise service_error(service_error.internal,
+                            "Misformed allocation response?")
+                if self.projects.has_key(pname): self.projects[pname] += 1
+                else: self.projects[pname] = 1
+                self.allocation[aid]['project'] = pname
+
+            self.allocation[aid]['keys'] = [ ]
+
+            try:
+                for u in ap['project']['user']:
+                    uname = u['userID']['localname']
+                    for k in [ k['sshPubkey'] for k in u['access'] \
+                            if k.has_key('sshPubkey') ]:
+                        kv = "%s:%s" % (uname, k)
+                        if self.keys.has_key(kv): self.keys[kv] += 1
+                        else: self.keys[kv] = 1
+                        self.allocation[aid]['keys'].append((uname, k))
+            except KeyError:
+                raise service_error(service_error.internal,
+                        "Misformed allocation response?")
+
+            self.write_state()
             resp = self.build_response(req['allocID'], ap)
             return resp
@@ -707 +798 @@
                             service_error.proxy,
                             "Undefined fault from proxy??");
+
+    def ReleaseAccess(self, req, fid):
+        # The dance to get into the request body
+        if req.has_key('ReleaseAccessRequestBody'):
+            req = req['ReleaseAccessRequestBody']
+        else:
+            raise service_error(service_error.req, "No request!?")
+
+        print req
+        try:
+            if req['allocID'].has_key('localname'):
+                aid = req['allocID']['localname']
+            elif req['allocID'].has_key('fedid'):
+                aid = unicode(req['allocID']['fedid'])
+            else:
+                raise service_error(service_error.req,
+                        "Only localnames and fedids are understood")
+        except KeyError:
+            raise service_error(service_error.req, "Badly formed request")
+
+        # If we know this allocation, reduce the reference counts and remove
+        # the local allocations.  Otherwise report an error.  If there is an
+        # allocation to delete, del_users will be a dictonary of sets where the
+        # key is the user that owns the keys in the set.  We use a set to avoid
+        # duplicates.  del_project is just the name of any dynamic project to
+        # delete.
+        del_users = { }
+        del_project = None
+        if self.allocation.has_key(aid):
+            for k in self.allocation[aid]['keys']:
+                kk = "%s:%s" % k
+                self.keys[kk] -= 1
+                if self.keys[kk] == 0:
+                    if not del_users.has_key(k[0]):
+                        del_users[k[0]] = set()
+                    del_users[k[0]].add(k[1])
+                    del self.keys[kk]
+
+            if self.allocation[aid].has_key('project'):
+                pname = self.allocation[aid]['project']
+                self.projects[pname] -= 1
+                if self.projects[pname] == 0:
+                    del_project = pname
+                    del self.projects[pname]
+
+            del self.allocation[aid]
+            # If we actually have resources to deallocate, prepare the call.
+            if del_project or del_users:
+                msg = { 'project': { }}
+                if del_project:
+                    msg['project']['name']['localname'] =  del_project
+                users = [ ]
+                for u in del_users.keys():
+                    users.append({ 'userID': { 'localname': u },\
+                        'access' :  \
+                                [ {'sshPubkey' : s } for s in del_users[u]]\
+                    })
+                if users: 
+                    msg['project']['user'] = users
+                print msg
+            self.write_state()
+            return { 'allocID': req['allocID'] } 
+        else:
+            raise service_error(service_error.req, "No such allocation")
+
+
+

fedd/fedd_bindings.wsdl

@@ -28 +28 @@
         <output>
           <soap:body use="encoded" parts="tns:RequestAccessResponseBody"
+            namespace="http://www.isi.edu/faber/fedd.wsdl"
+            encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"/>
+        </output>
+        <fault>
+          <soap:fault use="encoded"  name="tns:FeddFault"
+            namespace="http://www.isi.edu/faber/fedd.wsdl"
+            encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"/>
+        </fault>
+      </operation>
+      <operation name="ReleaseAccess">
+        <documentation>
+          The bindings of this operation are straightforward SOAP RPC 1.1.
+        </documentation>
+        <soap:operation soapAction="ReleaseAccess"/> 
+        <input>
+          <soap:body use="encoded" parts="tns:ReleaseAccessRequestBody"
+            namespace="http://www.isi.edu/faber/fedd.wsdl"
+            encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"/>
+        </input>
+        <output>
+          <soap:body use="encoded" parts="tns:ReleaseAccessResponseBody"
             namespace="http://www.isi.edu/faber/fedd.wsdl"
             encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"/>

fedd/fedd_experiment_control.py

@@ -869 +869 @@
                 "eventserver": e['eventServer'],
                 "project": unpack_id(p['name']),
-                "emulab" : e
+                "emulab" : e,
+                "allocID" : r['allocID'],
                 }
         # Make the testbed name be the label the user applied
@@ -883 +884 @@
                     tbparam[tb][key]= a['value']
         
+    def release_access(self, tb, aid):
+        """
+        Release access to testbed through fedd
+        """
+
+        uri = self.tbmap.get(tb, None)
+        if not uri:
+            raise service_error(serice_error.server_config, 
+                    "Unknown testbed: %s" % tb)
+
+        # The basic request
+        req = { 'allocID' : aid }
+        
+        # No retry loop here.  Proxy servers must correctly authenticate
+        # themselves without help
+
+        try:
+            ctx = fedd_ssl_context(self.cert_file, 
+                    self.trusted_certs, password=self.cert_pwd)
+        except SSL.SSLError:
+            raise service_error(service_error.server_config, 
+                    "Server certificates misconfigured")
+
+        loc = feddServiceLocator();
+        port = loc.getfeddPortType(uri,
+                transport=M2Crypto.httpslib.HTTPSConnection, 
+                transdict={ 'ssl_context' : ctx })
+
+        # Reconstruct the full request message
+        msg = ReleaseAccessRequestMessage()
+        msg.set_element_ReleaseAccessRequestBody(
+                pack_soap(msg, "ReleaseAccessRequestBody", req))
+
+        try:
+            resp = port.ReleaseAccess(msg)
+        except ZSI.ParseException, e:
+            raise service_error(service_error.req,
+                    "Bad format message (XMLRPC??): %s" % str(e))
+        except ZSI.FaultException, e:
+            resp = e.fault.detail[0]
+
+        # better error coding
+
+
+
     def remote_splitter(self, uri, desc, master):
 
@@ -1510 +1556 @@
                     'name': [ { 'localname' : eid} ],\
                     'emulab': tbparams[k]['emulab'],\
+                    'allocID' : tbparams[k]['allocID'],\
                     'master' : k == master,\
                 }
@@ -1789 +1836 @@
                     domain = fed['emulab']['domain']
                     host  = "%s%s" % (fed['emulab']['ops'], domain)
+                    aid = fed['allocID']
                 except KeyError, e:
                     continue
@@ -1797 +1845 @@
                         'host': host,\
                         'eid': eid,\
+                        'aid': aid,\
                     }
             self.state_lock.release()
@@ -1804 +1853 @@
                 self.stop_segment(tb, tbparams[tb]['eid'], tbparams)
 
-            # Remove teh terminated experiment
+            # release the allocations
+            for tb in tbparams.keys():
+                self.release_access(tb, tbparams[tb]['aid'])
+
+            # Remove the terminated experiment
             self.state_lock.acquire()
             for id in ids:

fedd/fedd_messages.wsdl

@@ -19 +19 @@
   </message>
 
+  <message name="ReleaseAccessRequestMessage">
+    <part name="ReleaseAccessRequestBody" type="xsd1:releaseRequestType"/>
+  </message>
+
+  <message name="ReleaseAccessResponseMessage">
+    <part name="ReleaseAccessResponseBody" type="xsd1:releaseResponseType"/>
+  </message>
   <message name="CreateRequestMessage">
     <part name="CreateRequestBody" type="xsd1:createRequestType"/>
@@ -77 +84 @@
       <fault name="FeddFault" message="tns:FaultMessage"/>
     </operation>
+    <operation name="ReleaseAccess">
+      <documentation>
+        Release an allocation of access to a testbed. This informs the testbed
+        that it is no longer necessary to allow the access mthods negotiated by
+        a RequestAccess cal.
+      </documentation>
+      <input message="tns:ReleaseAccessRequestMessage"/>
+      <output message="tns:ReleaseAccessResponseMessage"/>
+      <fault name="FeddFault" message="tns:FaultMessage"/>
+    </operation>
     <operation name="Create">
       <documentation>

fedd/fedd_types.xsd

@@ -395 +395 @@
       <xsd:element name="until" type="xsd:dateTime" minOccurs="0" 
         maxOccurs="1"/>
+    </xsd:sequence>
+  </xsd:complexType>
+
+  <xsd:complexType name="releaseRequestType">
+    <xsd:annotation>
+      <xsd:documentation>
+        A request to release the access rights allocated by an earlier 
+        RequestAccess call.
+      </xsd:documentation>
+    </xsd:annotation>
+    <xsd:sequence>
+      <xsd:element name="allocID" type="tns:IDType"/>
+    </xsd:sequence>
+  </xsd:complexType>
+
+  <xsd:complexType name="releaseResponseType">
+    <xsd:annotation>
+      <xsd:documentation>
+        Indication that the access has been terminated.
+      </xsd:documentation>
+    </xsd:annotation>
+    <xsd:sequence>
+      <xsd:element name="allocID" type="tns:IDType"/>
     </xsd:sequence>
   </xsd:complexType>