Changeset c3dcf48 for fedd

Timestamp:

Dec 4, 2008 9:49:39 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:

416292f

Parents:

2ac63f7d

Message:

Snapshot of new state-based allocation

Location:

fedd/federation

Files:

• access.py (modified) (8 diffs)
• allocate_project.py (modified) (9 diffs)

fedd/federation/access.py

@@ -80 +80 @@
         self.projects = { }
         self.keys = { }
+        self.types = { }
         self.allocation = { }
         self.state = { 
             'projects': self.projects,
             'allocation' : self.allocation,
-            'keys' : self.keys
+            'keys' : self.keys,
+            'types': self.types
         }
         self.log = logging.getLogger("fedd.access")
@@ -331 +333 @@
                 self.projects = self.state['projects']
                 self.keys = self.state['keys']
+                self.types = self.state['types']
 
                 self.log.debug("[read_state]: Read state from %s" % \
@@ -653 +656 @@
             self.state_lock.acquire()
             self.allocation[aid] = { }
+            try:
+                pname = ap['project']['name']['localname']
+            except KeyError:
+                pname = None
+
             if dyn[1]:
-                try:
-                    pname = ap['project']['name']['localname']
-                except KeyError:
+                if not pname:
                     self.state_lock.release()
                     raise service_error(service_error.internal,
@@ -663 +669 @@
                 else: self.projects[pname] = 1
                 self.allocation[aid]['project'] = pname
+
+            if ap.has_key('resources'):
+                if not pname:
+                    self.state_lock.release()
+                    raise service_error(service_error.internal,
+                            "Misformed allocation response?")
+                self.allocation[aid]['types'] = set()
+                nodes = ap['resources'].get('node', [])
+                for n in nodes:
+                    for h in n.get('hardware', []):
+                        if self.types.has_key((pname, h)):
+                            self.types[(pname, h)] += 1
+                        else:
+                            self.types[(pname, h)] = 1
+                        self.allocation[aid]['types'].add((pname,h))
+
 
             self.allocation[aid]['keys'] = [ ]
@@ -680 +702 @@
                         "Misformed allocation response?")
 
+
             self.allocation[aid]['owners'] = owners
             self.write_state()
@@ -741 +764 @@
             del_users = { }
             del_project = None
+            del_types = set()
+
             if self.allocation.has_key(aid):
                 self.log.debug("Found allocation for %s" %aid)
@@ -759 +784 @@
                         del_project = pname
                         del self.projects[pname]
+
+                if self.allocation[aid].has_key('types'):
+                    for t in self.allocation[aid]['types']:
+                        self.types[t] -= 1
+                        if self.types[t] == 0:
+                            if not del_project: del_project = t[0]
+                            del_types.add(t[1])
+                            del self.types[t]
 
                 del self.allocation[aid]
@@ -776 +809 @@
                     if users: 
                         msg['project']['user'] = users
+                    if len(del_types) > 0:
+                        msg['resources'] = { 'node': \
+                                [ {'hardware': [ h ] } for h in del_types ]\
+                            }
                     if self.allocate_project.release_project:
                         msg = { 'ReleaseProjectRequestBody' : msg}

fedd/federation/allocate_project.py

@@ -6 +6 @@
 import string
 import subprocess
+import threading
+import pickle
 import tempfile
 
@@ -87 +89 @@
             self.allocation_level = self.none
 
+        self.state = { 
+                'keys': set(), 
+                'types': set(),
+                'projects': set(),
+                'users': set(),
+                }
+        self.state_filename = config.get('allocate', 'allocation_state')
+        self.state_lock = threading.Lock()
+        self.read_state()
+
         access_db = config.get("allocate", "accessdb")
         if access_db:
@@ -127 +139 @@
         self.xmlrpc_services = { }
 
+    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.
+        """
+        if self.state_filename:
+            try:
+                f = open(self.state_filename, "r")
+                self.state = pickle.load(f)
+                self.log.debug("[allocation]: Read state from %s" % \
+                        self.state_filename)
+            except IOError, e:
+                self.log.warning(("[allocation]: No saved state: " +\
+                        "Can't open %s: %s") % (self.state_filename, e))
+            except EOFError, e:
+                self.log.warning(("[allocation]: " +\
+                        "Empty or damaged state file: %s:") % \
+                        self.state_filename)
+            except pickle.UnpicklingError, e:
+                self.log.warning(("[allocation]: No saved state: " + \
+                        "Unpickling failed: %s") % e)
+            # These should all be in the picked representation, but make sure
+            if not self.state.has_key('keys'): self.state['keys'] = set()
+            if not self.state.has_key('types'): self.state['types'] = set()
+            if not self.state.has_key('projects'):
+                self.state['projects'] = set()
+            if not self.state.has_key('users'): self.state['users'] = set()
+
+    def write_state(self):
+        if self.state_filename:
+            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 random_string(self, s, n=3):
         """Append n random ASCII characters to s and return the string"""
@@ -149 +204 @@
         f.write("</%s>\n" % root)
         f.close()
+
+    def run_cmd(self, cmd, log_prefix='allocate'):
+        """
+        Run the command passed in.  Cmd is a list containing the words of the
+        command.  Return the exit value from the subprocess - that is 0 on
+        success.  On an error running the command -  python or OS error, raise
+        a service exception.
+        """
+        self.log.debug("[%s]: %s" % (log_prefix, ' '.join(cmd)))
+        if not self.debug:
+            try:
+                return subprocess.call(cmd)
+            except OSError, e:
+                raise service_error(service_error.internal,
+                        "Static project subprocess creation error "+ \
+                                "[%s] (%s)" %  (cmd[0], e.strerror))
+        else:
+            return 0
+
+    def confirm_key(self, user, key):
+        """
+        Call run_cmd to comfirm the key.  Return a boolean rather
+        than the subprocess code.
+        """
+        return self.run_cmd((self.wap, self.confirmkey, '-C',  
+            '-u', user, '-k', key)) ==0
+
+    def add_key(self, user, key):
+        """
+        Call run_cmd to add the key.  Return a boolean rather
+        than the subprocess code.
+        """
+        return self.run_cmd((self.wap, self.addpubkey, '-u', user,
+            '-k', key)) == 0
+
+    def remove_key(self, user, key):
+        """
+        Call run_cmd to remove the key.  Return a boolean rather
+        than the subprocess code.
+        """
+        return self.run_cmd((self.wap, self.addpubkey, '-R', '-u', user, 
+            '-k', key)) == 0
+
+    def confirm_access(self, project, type):
+        """
+        Call run_cmd to comfirm the key.  Return a boolean rather
+        than the subprocess code.
+        """
+        return self.run_cmd((self.wap, self.grantnodetype, '-C',  
+            '-p', project, type)) ==0
+
+    def add_access(self, project, type):
+        """
+        Call run_cmd to add the key.  Return a boolean rather
+        than the subprocess code.
+        """
+        return self.run_cmd((self.wap, self.grantnodetype, 
+            '-p', project, type)) == 0
+
+    def remove_access(self, project, type):
+        """
+        Call run_cmd to remove the key.  Return a boolean rather
+        than the subprocess code.
+        """
+
+        return self.run_cmd((self.wap, self.grantnodetype, '-R', 
+            '-p', project, type)) == 0
+
+    def add_project(self, project, projfile):
+        """
+        Create a project using run_cmd.  This is two steps, and assumes that
+        the relevant XML files are in place and correct.  Make the return value
+        boolean.  Note that if a new user is specified in the XML, that user is
+        created on success.
+        """
+
+        if self.run_cmd((self.wap, self.newproj, projfile)) == 0:
+            return self.run_cmd((self.wap, self.mkproj, project)) ==0
+        else:
+            return False
+
+    def remove_project(self, project):
+        """
+        Call run_cmd to remove the project.  Make the return value boolean.
+        """
+
+        return self.run_cmd(self.wap, self.rmproj, project) == 0
+
+   
+    def add_user(self, name, param_file, project):
+        """
+        Create a user and link them to the given project.  Similar to
+        add_project, this requires a two step approach.  Returns True on success
+        False on failure.
+        """
+
+        if self.run_cmd((self.wap, self.newuser, param_file)) == 0:
+           return self.run_cmd((self.wap, self.user_to_project,
+               user, project)) == 0
+        else:
+           return False
+
+    def remove_user(self, user):
+        """
+        Call run_cmd to remove the user.  Make the return value boolean.
+        """
+
+        return self.run_cmd(self.wap, self.rmuser, user) == 0
+
 
 
@@ -269 +433 @@
         
 
-        # Write out the files
-        self.write_attr_xml(cuf, "user", create_user_fields)
-        self.write_attr_xml(suf, "user", service_user_fields)
-        self.write_attr_xml(pf, "project", proj_fields)
-
-        # Generate the commands (only grantnodetype's are dynamic)
-        cmds = [
-                (self.wap, self.newproj, projfile),
-                (self.wap, self.mkproj, name),
-                (self.wap, self.newuser, service_userfile),
-                (self.wap, self.user_to_project, uname['serviceAccess'], name),
-                ]
-
-        # Add commands to grant access to any resources in the request.  The
-        # list comprehension pulls out the hardware types in the node entries
-        # in the resources list.
-        if resources.has_key('node'):
-            for nt in [ h for n in resources['node']\
-                    if n.has_key('hardware') for h in n['hardware'] ] :
-                if self.allocation_level >= self.confirm_keys:
-                    cmds.append((self.wap, self.grantnodetype, '-p', pname, nt))
-
-
-        # Create the projects
-        rc = 0
-        for cmd in cmds:
-            self.log.debug("[dynamic_project]: %s" % ' '.join(cmd))
-            if not self.debug:
-                try:
-                    rc = subprocess.call(cmd)
-                except OSerror, e:
-                    raise service_error(service_error.internal,
-                            "Dynamic project subprocess creation error "+ \
-                                    "[%s] (%s)" %  (cmd[1], e.strerror))
-
-            if rc != 0: 
-                raise service_error(service_error.internal,
-                        "Dynamic project subprocess error " +\
-                                "[%s] (%d)" % (cmd[1], rc))
-        # Clean up tempfiles
-        #os.unlink(create_userfile)
-        #os.unlink(service_userfile)
-        #os.unlink(projfile)
+
+        added_projects = [ ]
+        added_users = [ ]
+        added_types = [ ]
+
+        self.state_lock.acquire()
+        try:
+            # Write out the files
+            self.write_attr_xml(cuf, "user", create_user_fields)
+            self.write_attr_xml(suf, "user", service_user_fields)
+            self.write_attr_xml(pf, "project", proj_fields)
+            try:
+                if self.add_project(name, projfile):
+                    # add_project adds a user as well in this case
+                    added_projects.append(name)
+                    added_users.append(uname['createExperiment'])
+                    self.state['projects'].add(name)
+                    self.state['users'].add(uname['createExperiment'])
+
+                    if self.add_user(uname['serviceAccess'], 
+                            service_userfile, name):
+                        added_users.append(uname['serviceAccess'])
+                        self.state['users'].add(uname['serviceAccess'])
+                    else:
+                        raise service_error("Unable to create user %s" % \
+                                uname['serviceAccess'])
+                else:
+                    raise service_error("Unable to create project/user %s/%s" % \
+                            (name, uname['experimentCreation']))
+
+                nodes = resources.get('node', [])
+                # Grant access to restricted resources.  This is simpler than
+                # the corresponding loop from static_project because this is a
+                # clean slate.
+                for nt in [ h for n in nodes\
+                        if n.has_key('hardware')\
+                            for h in n['hardware'] ] :
+                    if self.add_access(name, nt):
+                        self.state['types'].add((name, nt))
+                        added_types.append((name, nt))
+                    else:
+                        raise service_error(service_error.internal,
+                                "Failed to add access for %s to %s"\
+                                        % (name, nt))
+            except service_error, e:
+                # Something failed.  Back out the partial allocation as
+                # completely as possible and re-raise the error.
+                for p, t in added_types:
+                    self.state['types'].discard((p, t))
+                    try:
+                        self.remove_access(p, t)
+                    except service_error:
+                        pass
+                for u in added_users:
+                    self.state['users'].discard(u)
+                    try:
+                        self.remove_user(u)
+                    except service_error:
+                        pass
+
+                for p in added_projects:
+                    self.state['projects'].discard(p)
+                    try:
+                        self.remove_project(p)
+                    except service_error:
+                        pass
+                self.state_lock.release()
+                raise e
+        finally:
+            # Clean up tempfiles
+            os.unlink(create_userfile)
+            os.unlink(service_userfile)
+            os.unlink(projfile)
+
         rv = {\
             'project': {\
@@ -336 +531 @@
         proper resources and users have correct keys.  Add them if necessary.
         """
-
-        cmds =  []
-
         # Internal calls do not have a fedid parameter (i.e., local calls on
         # behalf of already vetted fedids)
@@ -355 +547 @@
             raise service_error(service_error.req, "Badly formed request")
 
-
-        for u in users:
-            try:
-                name = u['userID']['localname']
-            except KeyError:
-                raise service_error(service_error.req, "Badly formed user")
-            for sk in [ k['sshPubkey'] for k in u.get('access', []) \
-                    if k.has_key('sshPubkey')]:
-                if self.allocation_level >= self.dynamic_keys:
-                    cmds.append((self.wap, self.addpubkey, '-r', \
-                            '-u', name, '-k', sk))
-                elif self.allocation_level >= self.confirm_keys:
-                    cmds.append((self.wap, self.confirmkey, '-C', \
-                            '-u', name, '-k', sk))
+        added_keys = [ ]
+        added_types = [ ]
+        # Keep track of changes made to the system
+        self.state_lock.acquire()
+
+        try:
+            for u in users:
+                try:
+                    name = u['userID']['localname']
+                except KeyError:
+                    raise service_error(service_error.req, "Badly formed user")
+                for sk in [ k['sshPubkey'] for k in u.get('access', []) \
+                        if k.has_key('sshPubkey')]:
+                    if self.allocation_level >=self.confirm_keys:
+                        key_ok = self.confirm_key(name, sk)
+                        if not key_ok:
+                            if self.allocation_level >= self.dynamic_keys:
+                                if self.add_key(name, sk):
+                                    self.state['keys'].add((name, sk))
+                                    added_keys.append((name, sk))
+                                else:
+                                    raise service_error(service_error.internal,
+                                            "Failed to add key for %s" % name)
+                            else:
+                                raise service_error(service_error.internal,
+                                        "Failed to confirm key for %s" % name)
+                    else:
+                        self.log.warning("[static_project] no checking of " + \
+                            "static keys")
+
+            # Grant access to any resources in the request.  The
+            # list comprehension pulls out the hardware types in the node
+            # entries in the resources list.  The access module knows to
+            # only send resources that are restricted and needed by the
+            # project.
+            nodes = resources.get('node', [])
+            for nt in [ h for n in nodes\
+                    if n.has_key('hardware')\
+                        for h in n['hardware'] ] :
+                if self.allocation_level >= self.confirm_keys:
+                    access_ok = self.confirm_access(pname, nt)
+                    if not access_ok:
+                        if self.allocation_level >= self.dynamic_keys:
+                            if self.add_access(pname, nt):
+                                self.state['types'].add((pname, nt))
+                                added_types.append((pname, nt))
+                            else:
+                                raise service_error(service_error.internal,
+                                        "Failed to add access for %s to %s"\
+                                                % (pname, nt))
+                        else:
+                            raise service_error(service_error.internal,
+                                    "Failed to confirm access for %s to %s"\
+                                            % (pname, nt))
                 else:
                     self.log.warning("[static_project] no checking of " + \
-                            "static keys")
-        
-
-        # Add commands to grant access to any resources in the request.  The
-        # list comprehension pulls out the hardware types in the node entries
-        # in the resources list.
-        if resources.has_key('node'):
-            for nt in [ h for n in resources['node']\
-                    if n.has_key('hardware') for h in n['hardware'] ] :
-                if self.allocation_level >= self.confirm_keys:
-                    cmds.append((self.wap, self.grantnodetype, '-p', pname, nt))
-
-        # Run the commands 
-        rc = 0
-        for cmd in cmds:
-            self.log.debug("[static_project]: %s" % ' '.join(cmd))
-            if not self.debug:
+                        "node access")
+        except service_error, e:
+            # Do our best to clean up partial allocation and reraise the
+            # error.  Do our best to make sure that both allocation state and
+            # testbed state is restored.
+            for u, k in added_keys:
+                self.state['keys'].discard((u, k))
                 try:
-                    rc = subprocess.call(cmd)
-                except OSError, e:
-                    raise service_error(service_error.internal,
-                            "Static project subprocess creation error "+ \
-                                    "[%s] (%s)" %  (cmd[0], e.strerror))
-
-            if rc != 0: 
-                raise service_error(service_error.internal,
-                        "Static project subprocess error " +\
-                                "[%s] (%d)" % (cmd[0], rc))
-
-        return { 'project': req['StaticProjectRequestBody']['project']}
+                    self.remove_key(u, k)
+                except service_error:
+                    pass
+            for p, t in added_types:
+                self.state['types'].discard((p, t))
+                try:
+                    self.remove_access(p, t)
+                except service_error:
+                    pass
+            self.state_lock.release()
+            raise e
+        # All is well, save state and release the lock
+        self.write_state()
+        self.state_lock.release()
+        # return { 'project': req['StaticProjectRequestBody']['project']}
+        return req['StaticProjectRequestBody']
 
     def release_project(self, req, fedid=None):
@@ -415 +641 @@
             raise service_error(service_error.access, "Access Denied")
 
-        cmds = []
         pname = None
         users = []
+        nodes = [ ]
+
+        print req
 
         try:
@@ -425 +653 @@
             if req['ReleaseProjectRequestBody']['project'].has_key('user'):
                 users = req['ReleaseProjectRequestBody']['project']['user']
+            if req['ReleaseProjectRequestBody'].has_key('resources'):
+                nodes = req['ReleaseProjectRequestBody']\
+                        ['resources'].get('node', [])
         except KeyError:
             raise service_error(service_error.req, "Badly formed request")
 
-        if pname and pname not in self.fixed_projects and \
-                self.allocation_level >= self.dynamic_projects:
-            cmds.append((self.wap, self.rmproj, pname))
-
-        for u in users:
-            try:
-                name = u['userID']['localname']
-            except KeyError:
-                raise service_error(service_error.req, "Badly formed user")
-            if self.allocation_level >= self.dynamic_projects and \
-                    name not in self.fixed_users:
-                cmds.append((self.wap, self.rmuser, name))
-            else:
-                for sk in [ k['sshPubkey'] for k in u.get('access', []) \
-                        if k.has_key('sshPubkey')]:
-                    if (name.rstrip(), sk.rstrip()) not in self.fixed_keys:
-                        if self.allocation_level >= self.dynamic_keys:
-                            cmds.append((self.wap, self.addpubkey, '-R', '-r', \
-                                    '-u', name, '-k', sk))
-
-        # Run the commands 
-        rc = 0
-        for cmd in cmds:
-            self.log.debug("[release_project]: %s" % ' '.join(cmd))
-            if not self.debug:
+        if nodes and not pname:
+            raise service_error(service_error.req, 
+                    "Badly formed request (nodes without project)")
+
+        self.state_lock.acquire()
+        try:
+            for nt in [ h for n in nodes if n.has_key('hardware')\
+                    for h in n['hardware'] ] :
+                if (pname, nt ) in self.state['types']:
+                    self.remove_access(pname, nt)
+                    self.state['types'].discard((pname, nt))
+
+            for u in users:
                 try:
-                    rc = subprocess.call(cmd)
-                except OSError, e:
-                    raise service_error(service_error.internal,
-                            "Release project subprocess creation error "+ \
-                                    "[%s] (%s)" %  (cmd[0], e.strerror))
-
-            if rc != 0: 
-                raise service_error(service_error.internal,
-                        "Release project subprocess error " +\
-                                "[%s] (%d)" % (cmd[0], rc))
+                    name = u['userID']['localname']
+                except KeyError:
+                    raise service_error(service_error.req, "Badly formed user")
+                if name in self.state['users']:
+                    # If we created this user, discard the user, keys and all
+                    self.remove_user(name)
+                    self.state['users'].discard(name)
+                else:
+                    # If not, just strip any keys we added
+                    for sk in [ k['sshPubkey'] for k in u.get('access', []) \
+                            if k.has_key('sshPubkey')]:
+                        if (name, sk) in self.state['keys']:
+                            self.remove_key(name, sk)
+                            self.state['keys'].discard((name, sk))
+            if pname in self.state['projects']:
+                self.remove_project(pname)
+                self.state['projects'].discard(pname)
+
+        except service_error, e:
+            self.write_state()
+            self.state_lock.release()
+            raise e
+        self.write_state()
+        self.state_lock.release()
 
         return { 'project': req['ReleaseProjectRequestBody']['project']}