13804: Fix tests

[arvados.git] / tools / crunchstat-summary / crunchstat_summary / reader.py

diff --git a/tools/crunchstat-summary/crunchstat_summary/reader.py b/tools/crunchstat-summary/crunchstat_summary/reader.py
index 52d7e80f58d19da79d07912a215173c05277f1e3..e8f0861be49350eddc232bd8826a4681af3a424f 100644 (file)
--- a/tools/crunchstat-summary/crunchstat_summary/reader.py
+++ b/tools/crunchstat-summary/crunchstat_summary/reader.py
@@ -1,81 +1,102 @@
+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: AGPL-3.0
+
 from __future__ import print_function
 
 import arvados
-import collections
+import Queue
 import threading
+import _strptime
 
 from crunchstat_summary import logger
 
 
 class CollectionReader(object):
     def __init__(self, collection_id):
-        logger.debug('load collection %s', collection_id)
-        collection = arvados.collection.CollectionReader(collection_id)
-        filenames = [filename for filename in collection]
-        if len(filenames) != 1:
-            raise ValueError(
-                "collection {} has {} files; need exactly one".format(
-                    collection_id, len(filenames)))
-        self._reader = collection.open(filenames[0])
+        self._collection_id = collection_id
+        self._label = collection_id
+        self._reader = None
+
+    def __str__(self):
+        return self._label
 
     def __iter__(self):
+        logger.debug('load collection %s', self._collection_id)
+        collection = arvados.collection.CollectionReader(self._collection_id)
+        filenames = [filename for filename in collection]
+        if len(filenames) == 1:
+            filename = filenames[0]
+        else:
+            filename = 'crunchstat.txt'
+        self._label = "{}/{}".format(self._collection_id, filename)
+        self._reader = collection.open(filename)
         return iter(self._reader)
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if self._reader:
+            self._reader.close()
+            self._reader = None
+
 
 class LiveLogReader(object):
+    EOF = None
+
     def __init__(self, job_uuid):
         logger.debug('load stderr events for job %s', job_uuid)
-        self._filters = [
-            ['object_uuid', '=', job_uuid],
+        self.job_uuid = job_uuid
+
+    def __str__(self):
+        return self.job_uuid
+
+    def _get_all_pages(self):
+        got = 0
+        last_id = 0
+        filters = [
+            ['object_uuid', '=', self.job_uuid],
             ['event_type', '=', 'stderr']]
-        self._buffer = collections.deque()
-        self._got = 0
-        self._label = job_uuid
-        self._last_id = 0
-        self._start_getting_next_page()
-
-    def _start_getting_next_page(self):
-        self._thread = threading.Thread(target=self._get_next_page)
+        try:
+            while True:
+                page = arvados.api().logs().index(
+                    limit=1000,
+                    order=['id asc'],
+                    filters=filters + [['id','>',str(last_id)]],
+                    select=['id', 'properties'],
+                ).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()
+        self._thread = threading.Thread(target=self._get_all_pages)
         self._thread.daemon = True
         self._thread.start()
+        return self
 
-    def _get_next_page(self):
-        page = arvados.api().logs().index(
-            limit=1000,
-            order=['id asc'],
-            filters=self._filters + [['id','>',str(self._last_id)]],
-        ).execute()
-        self._got += len(page['items'])
-        logger.debug(
-            '%s: received %d of %d log events',
-            self._label, self._got,
-            self._got + page['items_available'] - len(page['items']))
-        self._page = page
-
-    def _buffer_page(self):
-        """Wait for current worker, copy results to _buffer, start next worker.
-
-        Return True if anything was added to the buffer."""
-        if self._thread is None:
-            return False
-        self._thread.join()
-        self._thread = None
-        page = self._page
-        if len(page['items']) == 0:
-            return False
-        if len(page['items']) < page['items_available']:
-            self._start_getting_next_page()
-        for i in page['items']:
-            for line in i['properties']['text'].split('\n'):
-                self._buffer.append(line)
-            self._last_id = i['id']
-        return True
+    def next(self):
+        line = self._queue.get()
+        if line is self.EOF:
+            self._thread.join()
+            raise StopIteration
+        return line
 
-    def __iter__(self):
+    def __enter__(self):
         return self
 
-    def next(self):
-        if len(self._buffer) == 0:
-            if not self._buffer_page():
-                raise StopIteration
-        return self._buffer.popleft() + '\n'
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        pass