8341: Fix up debug labels. Avoid deadlock after exceptions in thread.

diff --git a/tools/crunchstat-summary/crunchstat_summary/reader.py b/tools/crunchstat-summary/crunchstat_summary/reader.py
index 049b48f0636089a5354e16550fa5daa9877b9227..564e8208f6c03cc337d1c65b9990622aa3a7eaee 100644 (file)
--- a/tools/crunchstat-summary/crunchstat_summary/reader.py
+++ b/tools/crunchstat-summary/crunchstat_summary/reader.py
@@ -17,6 +17,10 @@ class CollectionReader(object):
                 "collection {} has {} files; need exactly one".format(
                     collection_id, len(filenames)))
         self._reader = collection.open(filenames[0])
+        self._label = "{}/{}".format(collection_id, filenames[0])
+
+    def __str__(self):
+        return self._label
 
     def __iter__(self):
         return iter(self._reader)
@@ -27,33 +31,38 @@ class LiveLogReader(object):
 
     def __init__(self, job_uuid):
         logger.debug('load stderr events for job %s', job_uuid)
-        self._filters = [
-            ['object_uuid', '=', job_uuid],
-            ['event_type', '=', 'stderr']]
-        self._label = job_uuid
+        self.job_uuid = job_uuid
+
+    def __str__(self):
+        return self.job_uuid
 
     def _get_all_pages(self):
         got = 0
         last_id = 0
-        while True:
-            page = arvados.api().logs().index(
-                limit=1000,
-                order=['id asc'],
-                filters=self._filters + [['id','>',str(last_id)]],
-            ).execute(num_retries=2)
-            got += len(page['items'])
-            logger.debug(
-                '%s: received %d of %d log events',
-                self._label, got,
-                got + page['items_available'] - len(page['items']))
-            for i in page['items']:
-                for line in i['properties']['text'].split('\n'):
-                    self._queue.put(line+'\n')
-                last_id = i['id']
-            if (len(page['items']) == 0 or
-                len(page['items']) >= page['items_available']):
-                break
-        self._queue.put(self.EOF)
+        filters = [
+            ['object_uuid', '=', self.job_uuid],
+            ['event_type', '=', 'stderr']]
+        try:
+            while True:
+                page = arvados.api().logs().index(
+                    limit=1000,
+                    order=['id asc'],
+                    filters=filters + [['id','>',str(last_id)]],
+                ).execute(num_retries=2)
+                got += len(page['items'])
+                logger.debug(
+                    '%s: received %d of %d log events',
+                    self.job_uuid, got,
+                    got + page['items_available'] - len(page['items']))
+                for i in page['items']:
+                    for line in i['properties']['text'].split('\n'):
+                        self._queue.put(line+'\n')
+                    last_id = i['id']
+                if (len(page['items']) == 0 or
+                    len(page['items']) >= page['items_available']):
+                    break
+        finally:
+            self._queue.put(self.EOF)
 
     def __iter__(self):
         self._queue = Queue.Queue()
@@ -65,5 +74,6 @@ class LiveLogReader(object):
     def next(self):
         line = self._queue.get()
         if line is self.EOF:
+            self._thread.join()
             raise StopIteration
         return line

diff --git a/tools/crunchstat-summary/crunchstat_summary/summarizer.py b/tools/crunchstat-summary/crunchstat_summary/summarizer.py
index 70b85f84f084eec1ba26c30d1d5623d9f2cc52c7..a8d3fd0e99e628e7842b97148a37f040f1089d05 100644 (file)
--- a/tools/crunchstat-summary/crunchstat_summary/summarizer.py
+++ b/tools/crunchstat-summary/crunchstat_summary/summarizer.py
@@ -52,10 +52,10 @@ class Summarizer(object):
         # constructor will overwrite this with something useful.
         self.existing_constraints = {}
 
-        logger.debug("%s: logdata %s", self.label, repr(logdata))
+        logger.debug("%s: logdata %s", self.label, logdata)
 
     def run(self):
-        logger.debug("%s: parsing log data", self.label)
+        logger.debug("%s: parsing logdata %s", self.label, self._logdata)
         for line in self._logdata:
             m = re.search(r'^\S+ \S+ \d+ (?P<seq>\d+) job_task (?P<task_uuid>\S+)$', line)
             if m: