Changeset 5ae3857

Timestamp:

Sep 6, 2009 2:15:52 PM (15 years ago)

Author:

Ted Faber <faber@…>

Branches:

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

Children:

2b7d768

Parents:

1da6a23

Message:

terminate works

Location:

fedd/federation

Files:

• access.py (modified) (6 diffs)
• experiment_control.py (modified) (5 diffs)

fedd/federation/access.py

@@ -67 +67 @@
         self.certdir = config.get("access","certdir")
         self.ssh_privkey_file = config.get("access","ssh_privkey_file")
+        self.create_debug = config.getboolean("access", "create_debug")
+        print self.create_debug
 
         self.attrs = { }
@@ -116 +118 @@
             'ReleaseAccess': soap_handler("ReleaseAccess", self.ReleaseAccess),
             'StartSegment': soap_handler("StartSegment", self.StartSegment),
+            'TerminateSegment': soap_handler("TerminateSegment", self.TerminateSegment),
             }
         self.xmlrpc_services =  {\
@@ -122 +125 @@
             'ReleaseAccess': xmlrpc_handler('ReleaseAccess',
                 self.ReleaseAccess),
-            'StartSegment': xmlrpc_handler('StartSegment',
-                self.StartSegment),
+            'StartSegment': xmlrpc_handler("StartSegment", self.StartSegment),
+            'TerminateSegment': xmlrpc_handler('TerminateSegment',
+                self.TerminateSegment),
             }
 
@@ -1083 +1087 @@
     class stop_segment(proxy_emulab_segment):
         def __init__(self, log=None, keyfile=None, debug=False):
-            experiment_control_local.emulab_segment.__init__(self,
+            access.proxy_emulab_segment.__init__(self,
                     log=log, keyfile=keyfile, debug=debug)
 
-        def __call__(self, tb, eid, tbparams):
+        def __call__(self, parent, user, pid, eid):
             """
             Stop a sub experiment by calling swapexp on the federant
             """
-            user = tbparams[tb]['user']
-            host = "%s%s" % (tbparams[tb]['host'], tbparams[tb]['domain'])
-            pid = tbparams[tb]['project']
-
-            self.log.info("[stop_segment]: Stopping %s on %s" % (eid, tb))
+            host = "%s%s" % (parent.ops, parent.domain)
+            self.log.info("[stop_segment]: Stopping %s" % eid)
             rv = False
             try:
                 # Clean out tar files: we've gone over quota in the past
-                self.ssh_cmd(user, host, "rm -rf /proj/%s/rpms/%s" % (pid, eid))
-                self.ssh_cmd(user, host, "rm -rf /proj/%s/tarfiles/%s" % \
-                        (pid, eid))
+                self.ssh_cmd(user, host, "rm -rf /proj/%s/software/%s" % \
+                        (pid, eid))
                 rv = self.ssh_cmd(user, host,
                         "/usr/testbed/bin/swapexp -w %s %s out" % (pid, eid))
@@ -1335 +1335 @@
                 proj = self.allocation[aid].get('sproject', None)
             user = self.allocation[aid].get('user', None)
+            self.allocation[aid]['experiment'] = ename
+            self.write_state()
         self.state_lock.release()
 
@@ -1351 +1353 @@
         self.generate_ns2(topo, expfile, 
                 "/proj/%s/software/%s/" % (proj, ename), master)
-        starter = self.start_segment(keyfile=self.ssh_privkey_file, debug=False)
+        starter = self.start_segment(keyfile=self.ssh_privkey_file, debug=self.create_debug)
         starter(self, ename, proj, user, expfile, tmpdir)
 
         return { 'allocID': req['allocID'] }
+
+    def TerminateSegment(self, req, fid):
+        print "In terminate"
+        try:
+            req = req['TerminateSegmentRequestBody']
+        except KeyError:
+            raise service_error(server_error.req, "Badly formed request")
+
+        auth_attr = req['allocID']['fedid']
+        aid = "%s" % auth_attr
+        attrs = req.get('fedAttr', [])
+        print "in terminate %s" % auth_attr
+        if not self.auth.check_attribute(fid, auth_attr):
+            print "access denied"
+            raise service_error(service_error.access, "Access denied")
+
+        self.state_lock.acquire()
+        if self.allocation.has_key(aid):
+            proj = self.allocation[aid].get('project', None)
+            if not proj: 
+                proj = self.allocation[aid].get('sproject', None)
+            user = self.allocation[aid].get('user', None)
+            ename = self.allocation[aid].get('experiment', None)
+        self.state_lock.release()
+
+        if not proj:
+            raise service_error(service_error.internal, 
+                    "Can't find project for %s" % aid)
+
+        if not user:
+            raise service_error(service_error.internal, 
+                    "Can't find creation user for %s" % aid)
+        if not ename:
+            raise service_error(service_error.internal, 
+                    "Can't find experiment name for %s" % aid)
+        stopper = self.stop_segment(keyfile=self.ssh_privkey_file, debug=self.create_debug)
+        stopper(self, user, proj, ename)
+        print { 'allocID': req['allocID'] }
+        return { 'allocID': req['allocID'] }

fedd/federation/experiment_control.py

@@ -197 +197 @@
     call_ReleaseAccess = service_caller('ReleaseAccess')
     call_StartSegment = service_caller('StartSegment')
+    call_TerminateSegment = service_caller('TerminateSegment')
     call_Ns2Split = service_caller('Ns2Split')
 
@@ -350 +351 @@
                 'MultiInfo': soap_handler('MultiInfo', self.get_multi_info),
                 'Terminate': soap_handler('Terminate', 
-                    self.terminate_experiment),
+                    self.new_terminate_experiment),
         }
 
@@ -360 +361 @@
                 'MultiInfo': xmlrpc_handler('MultiInfo', self.get_multi_info),
                 'Terminate': xmlrpc_handler('Terminate',
-                    self.terminate_experiment),
+                    self.new_terminate_experiment),
         }
 
@@ -2166 +2167 @@
                     self.trusted_certs)
             print r
+            return True
+
+
+
+    class new_terminate_segment:
+        def __init__(self, debug=False, log=None, cert_file=None,
+                cert_pwd=None, trusted_certs=None, caller=None):
+            self.log = log
+            self.debug = debug
+            self.cert_file = cert_file
+            self.cert_pwd = cert_pwd
+            self.trusted_certs = None
+            self.caller = caller
+
+        def __call__(self, uri, aid ):
+            print "in terminate_segment: %s" % aid
+            req = {
+                    'allocID': aid , 
+                }
+            r = self.caller(uri, req, self.cert_file, self.cert_pwd,
+                    self.trusted_certs)
             return True
 
@@ -3236 +3258 @@
             self.state_lock.release()
             raise service_error(service_error.req, "No saved state")
+
+    def new_terminate_experiment(self, req, fid):
+        """
+        Swap this experiment out on the federants and delete the shared
+        information
+        """
+        tbparams = { }
+        req = req.get('TerminateRequestBody', None)
+        if not req:
+            raise service_error(service_error.req,
+                    "Bad request format (no TerminateRequestBody)")
+        force = req.get('force', False)
+        exp = req.get('experiment', None)
+        if exp:
+            if exp.has_key('fedid'):
+                key = exp['fedid']
+                keytype = "fedid"
+            elif exp.has_key('localname'):
+                key = exp['localname']
+                keytype = "localname"
+            else:
+                raise service_error(service_error.req, "Unknown lookup type")
+        else:
+            raise service_error(service_error.req, "No request?")
+
+        self.check_experiment_access(fid, key)
+
+        dealloc_list = [ ]
+
+
+        # Create a logger that logs to the dealloc_list as well as to the main
+        # log file.
+        dealloc_log = logging.getLogger('fedd.experiment_control.%s' % key)
+        h = logging.StreamHandler(self.list_log(dealloc_list))
+        # XXX: there should be a global one of these rather than repeating the
+        # code.
+        h.setFormatter(logging.Formatter("%(asctime)s %(name)s %(message)s",
+                    '%d %b %y %H:%M:%S'))
+        dealloc_log.addHandler(h)
+
+        self.state_lock.acquire()
+        fed_exp = self.state.get(key, None)
+
+        if fed_exp:
+            # This branch of the conditional holds the lock to generate a
+            # consistent temporary tbparams variable to deallocate experiments.
+            # It releases the lock to do the deallocations and reacquires it to
+            # remove the experiment state when the termination is complete.
+
+            # First make sure that the experiment creation is complete.
+            status = fed_exp.get('experimentStatus', None)
+
+            if status:
+                if status in ('starting', 'terminating'):
+                    if not force:
+                        self.state_lock.release()
+                        raise service_error(service_error.partial, 
+                                'Experiment still being created or destroyed')
+                    else:
+                        self.log.warning('Experiment in %s state ' % status + \
+                                'being terminated by force.')
+            else:
+                # No status??? trouble
+                self.state_lock.release()
+                raise service_error(service_error.internal,
+                        "Experiment has no status!?")
+
+            ids = []
+            #  experimentID is a list of dicts that are self-describing
+            #  identifiers.  This finds all the fedids and localnames - the
+            #  keys of self.state - and puts them into ids.
+            for id in fed_exp.get('experimentID', []):
+                if id.has_key('fedid'): ids.append(id['fedid'])
+                if id.has_key('localname'): ids.append(id['localname'])
+
+            # Collect the allocation/segment ids
+            for fed in fed_exp.get('federant', []):
+                try:
+                    print "looking at %s" % fed
+                    tb = fed['emulab']['project']['testbed']['localname']
+                    aid = fed['allocID']
+                except KeyError, e:
+                    print "Key error: %s" %e
+                    continue
+                tbparams[tb] = aid
+            fed_exp['experimentStatus'] = 'terminating'
+            if self.state_filename: self.write_state()
+            self.state_lock.release()
+
+            # Stop everyone.  NB, wait_for_all waits until a thread starts and
+            # then completes, so we can't wait if nothing starts.  So, no
+            # tbparams, no start.
+            if len(tbparams) > 0:
+                thread_pool = self.thread_pool(self.nthreads)
+                for tb in tbparams.keys():
+                    # Create and start a thread to stop the segment
+                    thread_pool.wait_for_slot()
+                    uri = self.tbmap.get(tb, None)
+                    t  = self.pooled_thread(\
+                            target=self.new_terminate_segment(log=dealloc_log,
+                                cert_file=self.cert_file, 
+                                cert_pwd=self.cert_pwd,
+                                trusted_certs=self.trusted_certs,
+                                caller=self.call_TerminateSegment),
+                            args=(uri, tbparams[tb]), name=tb,
+                            pdata=thread_pool, trace_file=self.trace_file)
+                    t.start()
+                # Wait for completions
+                thread_pool.wait_for_all_done()
+
+            # release the allocations (failed experiments have done this
+            # already, and starting experiments may be in odd states, so we
+            # ignore errors releasing those allocations
+            try: 
+                for tb in tbparams.keys():
+                    self.release_access(tb, tbparams[tb])
+            except service_error, e:
+                if status != 'failed' and not force:
+                    raise e
+
+            # Remove the terminated experiment
+            self.state_lock.acquire()
+            for id in ids:
+                if self.state.has_key(id): del self.state[id]
+
+            if self.state_filename: self.write_state()
+            self.state_lock.release()
+
+            return { 
+                    'experiment': exp , 
+                    'deallocationLog': "".join(dealloc_list),
+                    }
+        else:
+            # Don't forget to release the lock
+            self.state_lock.release()
+            raise service_error(service_error.req, "No saved state")