Changeset f07fa49

Timestamp:

Sep 8, 2009 5:57:00 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:

40dd8c1

Parents:

430e98d

Message:

better logging and cleanup

Location:

fedd/federation

Files:

• access.py (modified) (5 diffs)
• experiment_control.py (modified) (11 diffs)
• list_log.py (added)

fedd/federation/access.py

@@ -22 +22 @@
 
 import topdl
+import list_log
 import httplib
 import tempfile
@@ -782 +783 @@
                         msg = { 'ReleaseProjectRequestBody' : msg}
                         self.allocate_project.release_project(msg)
+                # And remove the access cert
+                cf = "%s/%s.pem" % (self.certdir, aid)
+                self.log.debug("Removing %s" % cf)
+                os.remove(cf)
                 return { 'allocID': req['allocID'] } 
             else:
@@ -1340 +1345 @@
                 user = self.allocation[aid].get('user', None)
                 self.allocation[aid]['experiment'] = ename
+                self.allocation[aid]['log'] = [ ]
+                # Create a logger that logs to the experiment's state object as
+                # well as to the main log file.
+                alloc_log = logging.getLogger('fedd.access.%s' % ename)
+                h = logging.StreamHandler(
+                        list_log.list_log(self.allocation[aid]['log']))
+                # 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'))
+                alloc_log.addHandler(h)
                 self.write_state()
             self.state_lock.release()
@@ -1357 +1374 @@
             self.generate_ns2(topo, expfile, 
                     "/proj/%s/software/%s/" % (proj, ename), master)
+
             starter = self.start_segment(keyfile=self.ssh_privkey_file, 
-                    debug=self.create_debug)
+                    debug=self.create_debug, log=alloc_log)
             rv = starter(self, ename, proj, user, expfile, tmpdir)
         except service_error, e:
@@ -1373 +1391 @@
 
         if rv:
-            return { 'allocID': req['allocID'] }
+            # Grab the log (this is some anal locking, but better safe than
+            # sorry)
+            self.state_lock.acquire()
+            logv = "".join(self.allocation[aid]['log'])
+            self.state_lock.release()
+
+            return { 'allocID': req['allocID'], 'allocationLog': logv }
         elif err:
             raise service_error(service_error.federant,

fedd/federation/experiment_control.py

@@ -30 +30 @@
 
 import topdl
+import list_log
 from ip_allocator import ip_allocator
 from ip_addr import ip_addr
@@ -49 +50 @@
 
     class ssh_cmd_timeout(RuntimeError): pass
-
-    class list_log:
-        """
-        Provide an interface that lets logger.StreamHandler s write to a list
-        of strings.
-        """
-        def __init__(self, l=[]):
-            """
-            Link to an existing list or just create a log
-            """
-            self.ll = l
-            self.lock = Lock()
-        def write(self, str):
-            """
-            Add the string to the log.  Lock for consistency.
-            """
-            self.lock.acquire()
-            self.ll.append(str)
-            self.lock.release()
-
-        def flush(self):
-            """
-            No-op that StreamHandlers expect
-            """
-            pass
-
     
     class thread_pool:
@@ -221 +196 @@
         self.thread_with_rv = experiment_control_local.pooled_thread
         self.thread_pool = experiment_control_local.thread_pool
-        self.list_log = experiment_control_local.list_log
+        self.list_log = list_log.list_log
 
         self.cert_file = config.get("experiment_control", "cert_file")
@@ -924 +899 @@
     class start_segment:
         def __init__(self, debug=False, log=None, testbed="", cert_file=None,
-                cert_pwd=None, trusted_certs=None, caller=None):
+                cert_pwd=None, trusted_certs=None, caller=None,
+                log_collector=None):
             self.log = log
             self.debug = debug
@@ -932 +908 @@
             self.caller = caller
             self.testbed = testbed
+            self.log_collector = log_collector
 
         def __call__(self, uri, aid, topo, master, attrs=None):
@@ -947 +924 @@
                 r = self.caller(uri, req, self.cert_file, self.cert_pwd,
                         self.trusted_certs)
+                if r.has_key('StartSegmentResponseBody'):
+                    lval = r['StartSegmentResponseBody'].get('allocationLog',
+                            None)
+                    if lval and self.log_collector:
+                        for line in  lval.splitlines(True):
+                            self.log_collector.write(line)
+                else:
+                    raise service_error(service_error.internal, 
+                            "Bad response!?: %s" %r)
                 return True
             except service_error, e:
@@ -981 +967 @@
 
     def allocate_resources(self, allocated, master, eid, expid, expcert, 
-            tbparams, topo, tmpdir, alloc_log=None, attrs=None):
+            tbparams, topo, tmpdir, alloc_log=None, log_collector=None, 
+            attrs=None):
         started = { }           # Testbeds where a sub-experiment started 
                                 # successfully
@@ -1014 +1001 @@
                         cert_pwd=self.cert_pwd,
                         trusted_certs=self.trusted_certs,
-                        caller=self.call_StartSegment), 
+                        caller=self.call_StartSegment,
+                        log_collector=log_collector), 
                     args=(uri, aid, topo[tb], False, attrs), name=tb,
                     pdata=thread_pool, trace_file=self.trace_file)
@@ -1041 +1029 @@
                     testbed=master, cert_file=self.cert_file,
                     cert_pwd=self.cert_pwd, trusted_certs=self.trusted_certs,
-                    caller=self.call_StartSegment)
+                    caller=self.call_StartSegment,
+                    log_collector=log_collector)
             if not starter(uri, aid, topo[master], True, attrs):
                 failed.append(master)
@@ -1678 +1667 @@
         # to the main log file.
         alloc_log = logging.getLogger('fedd.experiment_control.%s' % eid)
-        h = logging.StreamHandler(self.list_log(self.state[eid]['log']))
+        alloc_collector = self.list_log(self.state[eid]['log'])
+        h = logging.StreamHandler(alloc_collector)
         # XXX: there should be a global one of these rather than repeating the
         # code.
@@ -1712 +1702 @@
         t  = Thread(target=self.allocate_resources,
                 args=(allocated, master, eid, expid, expcert, tbparams, 
-                    topo, tmpdir, alloc_log, attrs),
+                    topo, tmpdir, alloc_log, alloc_collector, attrs),
                 name=eid)
         t.start()