Changeset 5ae3857 for fedd/federation/experiment_control.py

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

File:

• fedd/federation/experiment_control.py (modified) (5 diffs)

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")