1 # Copyright (C) The Arvados Authors. All rights reserved.
3 # SPDX-License-Identifier: AGPL-3.0
11 from crunchstat_summary import logger
14 class CollectionReader(object):
15 def __init__(self, collection_id, api_client=None, collection_object=None):
16 self._collection_id = collection_id
17 self._label = collection_id
19 self._api_client = api_client
20 self._collection = collection_object or arvados.collection.CollectionReader(self._collection_id, api_client=self._api_client)
26 logger.debug('load collection %s', self._collection_id)
28 filenames = [filename for filename in self._collection]
29 # Crunch2 has multiple stats files
30 if len(filenames) > 1:
31 filenames = ['crunchstat.txt', 'arv-mount.txt']
32 for filename in filenames:
34 self._readers.append(self._collection.open(filename, "rt"))
36 logger.warn('Unable to open %s', filename)
37 self._label = "{}/{}".format(self._collection_id, filenames[0])
38 return itertools.chain(*[iter(reader) for reader in self._readers])
43 def __exit__(self, exc_type, exc_val, exc_tb):
45 for reader in self._readers:
51 with self._collection.open("node.json", "rt") as f:
54 logger.warn('Unable to open node.json')
58 class LiveLogReader(object):
61 def __init__(self, job_uuid):
62 self.job_uuid = job_uuid
63 self.event_types = (['stderr'] if '-8i9sb-' in job_uuid else ['crunchstat', 'arv-mount'])
64 logger.debug('load %s events for job %s', self.event_types, self.job_uuid)
69 def _get_all_pages(self):
73 ['object_uuid', '=', self.job_uuid],
74 ['event_type', 'in', self.event_types]]
77 page = arvados.api().logs().list(
80 filters=filters + [['id','>',str(last_id)]],
81 select=['id', 'properties'],
82 ).execute(num_retries=2)
83 got += len(page['items'])
85 '%s: received %d of %d log events',
87 got + page['items_available'] - len(page['items']))
88 for i in page['items']:
89 for line in i['properties']['text'].split('\n'):
90 self._queue.put(line+'\n')
92 if (len(page['items']) == 0 or
93 len(page['items']) >= page['items_available']):
96 self._queue.put(self.EOF)
99 self._queue = queue.Queue()
100 self._thread = threading.Thread(target=self._get_all_pages)
101 self._thread.daemon = True
106 line = self._queue.get()
112 next = __next__ # for Python 2
117 def __exit__(self, exc_type, exc_val, exc_tb):
123 class StubReader(object):
124 def __init__(self, fh):
136 def __exit__(self, exc_type, exc_val, exc_tb):