openrvdas.logger.readers.logfile_reader
No module-level documentation available.
1#!/usr/bin/env python3 2import json 3import logging 4import time 5 6try: 7 import parse 8 PARSE_INSTALLED = True 9except ImportError: 10 PARSE_INSTALLED = False 11 12from logger.utils import timestamp # noqa: E402 13from logger.utils.das_record import DASRecord # noqa: E402 14from logger.readers.text_file_reader import TextFileReader # noqa: E402 15from logger.readers.reader import TimestampedReader # noqa: E402 16 17 18################################################################################ 19# Open and read single-line records from one or more text files. 20class LogfileReader(TimestampedReader): 21 """ 22 Read lines from one or more text files. Sequentially open all 23 files that match the file_spec. 24 25 Expect that each line will either be a string prefixed by a timestamp that 26 follows the time_format parameter (ISO8601 by default) or a line of JSON 27 encoding a DASRecord. 28 29 If line is a string prefixed by a timestamp, return the string. If JSON, 30 return the DASRecord encoded by the JSON string. 31 """ 32 ############################ 33 def __init__(self, filebase=None, tail=False, refresh_file_spec=False, 34 retry_interval=0.1, interval=0, use_timestamps=False, 35 time_acceleration_factor=1.0, 36 record_format=None, 37 time_format=timestamp.TIME_FORMAT, 38 date_format=timestamp.DATE_FORMAT, 39 eol=None, quiet=False, **kwargs): 40 """ 41 ``` 42 filebase Possibly wildcarded string specifying files to be opened. 43 Special case: if file_spec is None, read from stdin. 44 45 tail If False, return None upon reaching end of last file; if 46 True, block upon reaching EOF of last file and wait for 47 more records. 48 49 refresh_file_spec 50 If True, refresh the search for matching filenames when 51 reaching last EOF to see if any new matching files have 52 appeared in the interim. 53 54 retry_interval 55 If tail and/or refresh_file_spec are True, how long to 56 wait before looking to see if any new records or files 57 have shown up. 58 59 interval 60 How long to sleep between returning records. In general 61 this should be zero except for debugging purposes. 62 63 use_timestamps 64 If True, use the timestamps from the log file to determine 65 at what interval each record should be emitted. 66 67 time_acceleration_factor 68 When use_timestamps is True, multiplies the time intervals 69 between records by this factor. Values greater than 1.0 will 70 speed up playback, values between 0 and 1.0 will slow it down. 71 Default is 1.0 (normal speed). 72 73 record_format 74 If specified, a custom record format to use for extracting 75 timestamp and record. The default is '{timestamp:ti} {record}'. 76 77 eol Optional character by which to recognize the end of a record 78 79 quiet - if not False, don't complain when unable to parse a record. 80 81 ``` 82 Note that the order in which files are opened will probably be in 83 alphanumeric by filename, but this is not strictly enforced and 84 depends on how glob returns them. 85 """ 86 super().__init__(**kwargs) 87 88 if not PARSE_INSTALLED: 89 raise ImportError('LogfileReader requires Python "parse" module; ' 90 'please run "pip install parse"') 91 if interval and use_timestamps: 92 raise ValueError('Can not specify both "interval" and "use_timestamps"') 93 94 self.filebase = filebase 95 self.use_timestamps = use_timestamps 96 97 # Validate time_acceleration_factor 98 if time_acceleration_factor is None: 99 raise ValueError("time_acceleration_factor must be a number") 100 if not isinstance(time_acceleration_factor, (int, float)): 101 raise ValueError("time_acceleration_factor must be a number") 102 if time_acceleration_factor <= 0: 103 raise ValueError("time_acceleration_factor must be greater than zero") 104 self.time_acceleration_factor = time_acceleration_factor 105 106 self.record_format = record_format or '{timestamp:ti} {record}' 107 self.compiled_record_format = parse.compile(self.record_format) 108 self.date_format = date_format 109 self.time_format = time_format 110 self.tail = tail 111 self.refresh_file_spec = refresh_file_spec 112 self.eol = eol 113 self.quiet = quiet 114 115 # If use_timestamps, we need to keep track of our last_read to 116 # know how long to sleep 117 self.last_timestamp = 0 118 self.last_read = 0 119 120 self._first_msec_timestamp = None 121 self.prev_record = None 122 123 # If they give us a filebase, add wildcard to match its suffixes; 124 # otherwise, we'll pass on the empty string to TextFileReader so 125 # that it uses stdin. NOTE: we should really use a pattern that 126 # echoes timestamp.DATE_FORMAT, e.g. 127 # DATE_FORMAT_WILDCARD = '????-??-??' 128 self.file_spec = filebase + '*' if filebase else None 129 self.reader = TextFileReader(file_spec=self.file_spec, 130 tail=tail, 131 refresh_file_spec=refresh_file_spec, 132 retry_interval=retry_interval, 133 interval=interval, eol=eol) 134 135 ############################ 136 def read(self): 137 """ 138 Return the next line in the file(s), or None if there are no more 139 records (as opposed to '' if the next record is a blank line). To test 140 EOF you'll need to test 141 142 if record is None: 143 no more records... 144 145 rather than simply 146 147 if not record: 148 could be EOF or simply an empty next line 149 """ 150 151 # NOTE: It feels like we should check here that the reader's 152 # current file really does match our logfile name format... 153 while True: 154 record = self.reader.read() 155 if not record: # None means we're out of records 156 return record 157 158 # If we've got a record and we're not using timestamps, we're 159 # done - just return it. 160 if not self.use_timestamps: 161 self.prev_record = record 162 # We need this in case the next call is seek_time() or 163 # read_time_range(). This is less expensive than parsing every 164 # timestamp and keeping self.last_timestamp, but an 165 # alternative might be to implement read_previous(), which 166 # would be expensive but which could be called only when 167 # actually needed. 168 169 # Check whether this is a JSON-encoded DASRecord. If so, return 170 # as a DASRecord; otherwise, just return as string. Yes, this 171 # adds overhead, but our assumption is that the throughput on 172 # a LogfileReader is going to be pretty low. 173 try: 174 das_record = DASRecord(json_str=record) 175 return das_record 176 except json.JSONDecodeError: 177 return record 178 179 # If we're here, we're going to be doling out records according to the 180 # differences in their timestamps. 181 182 # Try to parse the timestamp off the front. If we succeed, grab the 183 # timestamp and break out of loop. 184 try: 185 parsed_record = self.compiled_record_format.parse(record).named 186 ts = parsed_record['timestamp'].timestamp() 187 break 188 # We had a problem parsing as a timestamped string. 189 except (KeyError, ValueError, AttributeError): 190 pass 191 192 # Try parsing as JSON DASRecord. If we succeed, grab the 193 # timestamp and break out of loop. 194 try: 195 record = DASRecord(json_str=record) 196 ts = record.timestamp 197 break 198 except json.JSONDecodeError: 199 pass 200 201 # If we're here, we failed to parse the record. Complain, if appropriate, 202 # then spit out it out without updating timestamps. 203 if not self.quiet: 204 logging.warning('Unable to parse record into DASRecord') 205 logging.warning(f'Unable to parse record into "{self.record_format}"') 206 logging.warning(f'Record: "{record}"') 207 return record 208 209 # If this is not our first read, figure out how long we need to wait 210 # for our next one. 211 if self.last_read > 0: 212 # If here, we've got a record and a timestamp and are intending to 213 # use it. Figure out how long we should sleep before returning it. 214 desired_interval = ts - self.last_timestamp 215 216 # Apply the time acceleration factor to the desired interval 217 desired_interval = desired_interval / self.time_acceleration_factor 218 219 now = timestamp.timestamp() 220 actual_interval = now - self.last_read 221 logging.debug('Desired interval %f, actual %f; sleeping %f', 222 desired_interval, actual_interval, 223 max(0, desired_interval - actual_interval)) 224 time.sleep(max(0, desired_interval - actual_interval)) 225 226 self.last_timestamp = ts 227 self.last_read = timestamp.timestamp() 228 229 self.prev_record = record 230 return record 231 232 ############################ 233 def _read_until(self, desired_time_msec): 234 while True: 235 record = self.reader.read() 236 if record is None: 237 return 238 self.prev_record = record 239 if self._get_msec_timestamp(record) >= desired_time_msec: 240 self.reader.seek(-1, 'current') 241 return 242 243 ############################ 244 def _reset(self): 245 self.reader.seek(0, 'start') 246 247 ############################ 248 def _get_msec_timestamp(self, record): 249 time_str = record.split(' ', 1)[0] 250 return timestamp.timestamp(time_str, time_format=self.time_format) * 1000 251 252 ############################ 253 def _peek_msec(self): 254 record = self.reader.read() 255 if record is None: 256 return None 257 self.reader.seek(-1, 'current') 258 return self._get_msec_timestamp(record) 259 260 ############################ 261 # Note: this will change the file position if necessary, and should not be used 262 # except where that behavior is appropriate. 263 def _get_first_msec_timestamp(self): 264 if self._first_msec_timestamp is None: 265 self._reset() 266 record = self.reader.read() 267 if record is None: 268 return None 269 self._first_msec_timestamp = self._get_msec_timestamp(record) 270 return self._first_msec_timestamp 271 272 ############################ 273 def seek_time(self, offset=0, origin='current'): 274 """ 275 Behavior is intended to mimic file seek() behavior but with 276 respect to timestamps. 277 After calling this, the next record read will be the first record 278 whose timestamp is the same as or later than the requested time; 279 if no such record is found, it will read to the end. 280 Exception: if the records are not in exact chronological order, 281 records appearing before the current record but with a later 282 timestamp might be missed. 283 284 Args: 285 offset: offset in msec relative to origin 286 origin: 'start', 'current' or 'end' 287 288 Returns: 289 Requested time in msec, i.e. timestamp of (T0 + offset), 290 where T0 = timestamp(first record) if origin = 'start' 291 = timestamp(next record) if origin = 'current' and next record is not None 292 = timestamp(last record) if origin = 'current' and next record is None 293 = timestamp(last record) if origin = 'end' 294 Returns None if no timestamps were found 295 """ 296 if self.filebase is None: 297 raise ValueError('seek_time() not allowed on stdin') 298 299 # TODO: Maybe these are OK, as long as 'end' is defined as the point where 300 # read() returns None for the first time. 301 if self.tail and origin == 'end': 302 raise ValueError('tail=True incompatible with origin == "end"') 303 if self.refresh_file_spec and origin == 'end': 304 raise ValueError('refresh_file_spec=True incompatible with origin == "end"') 305 306 if origin == 'start': 307 if offset < 0: 308 raise ValueError("Can't back up past earliest record") 309 first_timestamp = self._get_first_msec_timestamp() 310 if first_timestamp is None: 311 return None 312 desired_time = first_timestamp + offset 313 if self.prev_record is None: 314 self._reset() 315 else: 316 prev_timestamp = self._get_msec_timestamp(self.prev_record) 317 if prev_timestamp >= desired_time: 318 self._reset() 319 self._read_until(desired_time) 320 return desired_time 321 322 elif origin == 'current': 323 next_timestamp = self._peek_msec() 324 curr_timestamp = next_timestamp or self._get_msec_timestamp(self.prev_record) 325 if curr_timestamp is None: 326 return None 327 desired_time = curr_timestamp + offset 328 if offset == 0: 329 return desired_time 330 if offset < 0: 331 self._reset() 332 self._read_until(desired_time) 333 return desired_time 334 335 elif origin == 'end': 336 while self.read() is not None: 337 pass 338 if self.prev_record is None: 339 return None 340 end_timestamp = self._get_msec_timestamp(self.prev_record) 341 desired_time = end_timestamp + offset 342 if offset < 0: 343 self._reset() 344 self._read_until(desired_time) 345 return desired_time 346 347 else: 348 raise ValueError('Unknown origin value: "%s"' % origin) 349 350 ############################ 351 # Read a range of records beginning with timestamp start 352 # milliseconds, and ending *before* timestamp stop milliseconds. 353 def read_time_range(self, start=None, stop=None): 354 if self.filebase is None: 355 raise ValueError('read_time_range() not allowed on stdin') 356 357 # TODO: Is this needed? stop=None would be OK unless records are 358 # being written faster than they're being read. 359 if stop is None: 360 if self.tail: 361 raise ValueError('tail=True incompatible with stop=None') 362 if self.refresh_file_spec: 363 raise ValueError('refresh_file_spec=True incompatible with stop=None') 364 365 if start is None: 366 starting_offset = 0 367 else: 368 starting_offset = start - self._get_first_msec_timestamp() 369 370 self.seek_time(starting_offset, 'start') 371 records = [] 372 while True: 373 record = self.read() 374 if record is None: 375 break 376 if stop and self._get_msec_timestamp(record) >= stop: 377 break 378 records.append(record) 379 return records
class
LogfileReader(logger.readers.reader.TimestampedReader):
21class LogfileReader(TimestampedReader): 22 """ 23 Read lines from one or more text files. Sequentially open all 24 files that match the file_spec. 25 26 Expect that each line will either be a string prefixed by a timestamp that 27 follows the time_format parameter (ISO8601 by default) or a line of JSON 28 encoding a DASRecord. 29 30 If line is a string prefixed by a timestamp, return the string. If JSON, 31 return the DASRecord encoded by the JSON string. 32 """ 33 ############################ 34 def __init__(self, filebase=None, tail=False, refresh_file_spec=False, 35 retry_interval=0.1, interval=0, use_timestamps=False, 36 time_acceleration_factor=1.0, 37 record_format=None, 38 time_format=timestamp.TIME_FORMAT, 39 date_format=timestamp.DATE_FORMAT, 40 eol=None, quiet=False, **kwargs): 41 """ 42 ``` 43 filebase Possibly wildcarded string specifying files to be opened. 44 Special case: if file_spec is None, read from stdin. 45 46 tail If False, return None upon reaching end of last file; if 47 True, block upon reaching EOF of last file and wait for 48 more records. 49 50 refresh_file_spec 51 If True, refresh the search for matching filenames when 52 reaching last EOF to see if any new matching files have 53 appeared in the interim. 54 55 retry_interval 56 If tail and/or refresh_file_spec are True, how long to 57 wait before looking to see if any new records or files 58 have shown up. 59 60 interval 61 How long to sleep between returning records. In general 62 this should be zero except for debugging purposes. 63 64 use_timestamps 65 If True, use the timestamps from the log file to determine 66 at what interval each record should be emitted. 67 68 time_acceleration_factor 69 When use_timestamps is True, multiplies the time intervals 70 between records by this factor. Values greater than 1.0 will 71 speed up playback, values between 0 and 1.0 will slow it down. 72 Default is 1.0 (normal speed). 73 74 record_format 75 If specified, a custom record format to use for extracting 76 timestamp and record. The default is '{timestamp:ti} {record}'. 77 78 eol Optional character by which to recognize the end of a record 79 80 quiet - if not False, don't complain when unable to parse a record. 81 82 ``` 83 Note that the order in which files are opened will probably be in 84 alphanumeric by filename, but this is not strictly enforced and 85 depends on how glob returns them. 86 """ 87 super().__init__(**kwargs) 88 89 if not PARSE_INSTALLED: 90 raise ImportError('LogfileReader requires Python "parse" module; ' 91 'please run "pip install parse"') 92 if interval and use_timestamps: 93 raise ValueError('Can not specify both "interval" and "use_timestamps"') 94 95 self.filebase = filebase 96 self.use_timestamps = use_timestamps 97 98 # Validate time_acceleration_factor 99 if time_acceleration_factor is None: 100 raise ValueError("time_acceleration_factor must be a number") 101 if not isinstance(time_acceleration_factor, (int, float)): 102 raise ValueError("time_acceleration_factor must be a number") 103 if time_acceleration_factor <= 0: 104 raise ValueError("time_acceleration_factor must be greater than zero") 105 self.time_acceleration_factor = time_acceleration_factor 106 107 self.record_format = record_format or '{timestamp:ti} {record}' 108 self.compiled_record_format = parse.compile(self.record_format) 109 self.date_format = date_format 110 self.time_format = time_format 111 self.tail = tail 112 self.refresh_file_spec = refresh_file_spec 113 self.eol = eol 114 self.quiet = quiet 115 116 # If use_timestamps, we need to keep track of our last_read to 117 # know how long to sleep 118 self.last_timestamp = 0 119 self.last_read = 0 120 121 self._first_msec_timestamp = None 122 self.prev_record = None 123 124 # If they give us a filebase, add wildcard to match its suffixes; 125 # otherwise, we'll pass on the empty string to TextFileReader so 126 # that it uses stdin. NOTE: we should really use a pattern that 127 # echoes timestamp.DATE_FORMAT, e.g. 128 # DATE_FORMAT_WILDCARD = '????-??-??' 129 self.file_spec = filebase + '*' if filebase else None 130 self.reader = TextFileReader(file_spec=self.file_spec, 131 tail=tail, 132 refresh_file_spec=refresh_file_spec, 133 retry_interval=retry_interval, 134 interval=interval, eol=eol) 135 136 ############################ 137 def read(self): 138 """ 139 Return the next line in the file(s), or None if there are no more 140 records (as opposed to '' if the next record is a blank line). To test 141 EOF you'll need to test 142 143 if record is None: 144 no more records... 145 146 rather than simply 147 148 if not record: 149 could be EOF or simply an empty next line 150 """ 151 152 # NOTE: It feels like we should check here that the reader's 153 # current file really does match our logfile name format... 154 while True: 155 record = self.reader.read() 156 if not record: # None means we're out of records 157 return record 158 159 # If we've got a record and we're not using timestamps, we're 160 # done - just return it. 161 if not self.use_timestamps: 162 self.prev_record = record 163 # We need this in case the next call is seek_time() or 164 # read_time_range(). This is less expensive than parsing every 165 # timestamp and keeping self.last_timestamp, but an 166 # alternative might be to implement read_previous(), which 167 # would be expensive but which could be called only when 168 # actually needed. 169 170 # Check whether this is a JSON-encoded DASRecord. If so, return 171 # as a DASRecord; otherwise, just return as string. Yes, this 172 # adds overhead, but our assumption is that the throughput on 173 # a LogfileReader is going to be pretty low. 174 try: 175 das_record = DASRecord(json_str=record) 176 return das_record 177 except json.JSONDecodeError: 178 return record 179 180 # If we're here, we're going to be doling out records according to the 181 # differences in their timestamps. 182 183 # Try to parse the timestamp off the front. If we succeed, grab the 184 # timestamp and break out of loop. 185 try: 186 parsed_record = self.compiled_record_format.parse(record).named 187 ts = parsed_record['timestamp'].timestamp() 188 break 189 # We had a problem parsing as a timestamped string. 190 except (KeyError, ValueError, AttributeError): 191 pass 192 193 # Try parsing as JSON DASRecord. If we succeed, grab the 194 # timestamp and break out of loop. 195 try: 196 record = DASRecord(json_str=record) 197 ts = record.timestamp 198 break 199 except json.JSONDecodeError: 200 pass 201 202 # If we're here, we failed to parse the record. Complain, if appropriate, 203 # then spit out it out without updating timestamps. 204 if not self.quiet: 205 logging.warning('Unable to parse record into DASRecord') 206 logging.warning(f'Unable to parse record into "{self.record_format}"') 207 logging.warning(f'Record: "{record}"') 208 return record 209 210 # If this is not our first read, figure out how long we need to wait 211 # for our next one. 212 if self.last_read > 0: 213 # If here, we've got a record and a timestamp and are intending to 214 # use it. Figure out how long we should sleep before returning it. 215 desired_interval = ts - self.last_timestamp 216 217 # Apply the time acceleration factor to the desired interval 218 desired_interval = desired_interval / self.time_acceleration_factor 219 220 now = timestamp.timestamp() 221 actual_interval = now - self.last_read 222 logging.debug('Desired interval %f, actual %f; sleeping %f', 223 desired_interval, actual_interval, 224 max(0, desired_interval - actual_interval)) 225 time.sleep(max(0, desired_interval - actual_interval)) 226 227 self.last_timestamp = ts 228 self.last_read = timestamp.timestamp() 229 230 self.prev_record = record 231 return record 232 233 ############################ 234 def _read_until(self, desired_time_msec): 235 while True: 236 record = self.reader.read() 237 if record is None: 238 return 239 self.prev_record = record 240 if self._get_msec_timestamp(record) >= desired_time_msec: 241 self.reader.seek(-1, 'current') 242 return 243 244 ############################ 245 def _reset(self): 246 self.reader.seek(0, 'start') 247 248 ############################ 249 def _get_msec_timestamp(self, record): 250 time_str = record.split(' ', 1)[0] 251 return timestamp.timestamp(time_str, time_format=self.time_format) * 1000 252 253 ############################ 254 def _peek_msec(self): 255 record = self.reader.read() 256 if record is None: 257 return None 258 self.reader.seek(-1, 'current') 259 return self._get_msec_timestamp(record) 260 261 ############################ 262 # Note: this will change the file position if necessary, and should not be used 263 # except where that behavior is appropriate. 264 def _get_first_msec_timestamp(self): 265 if self._first_msec_timestamp is None: 266 self._reset() 267 record = self.reader.read() 268 if record is None: 269 return None 270 self._first_msec_timestamp = self._get_msec_timestamp(record) 271 return self._first_msec_timestamp 272 273 ############################ 274 def seek_time(self, offset=0, origin='current'): 275 """ 276 Behavior is intended to mimic file seek() behavior but with 277 respect to timestamps. 278 After calling this, the next record read will be the first record 279 whose timestamp is the same as or later than the requested time; 280 if no such record is found, it will read to the end. 281 Exception: if the records are not in exact chronological order, 282 records appearing before the current record but with a later 283 timestamp might be missed. 284 285 Args: 286 offset: offset in msec relative to origin 287 origin: 'start', 'current' or 'end' 288 289 Returns: 290 Requested time in msec, i.e. timestamp of (T0 + offset), 291 where T0 = timestamp(first record) if origin = 'start' 292 = timestamp(next record) if origin = 'current' and next record is not None 293 = timestamp(last record) if origin = 'current' and next record is None 294 = timestamp(last record) if origin = 'end' 295 Returns None if no timestamps were found 296 """ 297 if self.filebase is None: 298 raise ValueError('seek_time() not allowed on stdin') 299 300 # TODO: Maybe these are OK, as long as 'end' is defined as the point where 301 # read() returns None for the first time. 302 if self.tail and origin == 'end': 303 raise ValueError('tail=True incompatible with origin == "end"') 304 if self.refresh_file_spec and origin == 'end': 305 raise ValueError('refresh_file_spec=True incompatible with origin == "end"') 306 307 if origin == 'start': 308 if offset < 0: 309 raise ValueError("Can't back up past earliest record") 310 first_timestamp = self._get_first_msec_timestamp() 311 if first_timestamp is None: 312 return None 313 desired_time = first_timestamp + offset 314 if self.prev_record is None: 315 self._reset() 316 else: 317 prev_timestamp = self._get_msec_timestamp(self.prev_record) 318 if prev_timestamp >= desired_time: 319 self._reset() 320 self._read_until(desired_time) 321 return desired_time 322 323 elif origin == 'current': 324 next_timestamp = self._peek_msec() 325 curr_timestamp = next_timestamp or self._get_msec_timestamp(self.prev_record) 326 if curr_timestamp is None: 327 return None 328 desired_time = curr_timestamp + offset 329 if offset == 0: 330 return desired_time 331 if offset < 0: 332 self._reset() 333 self._read_until(desired_time) 334 return desired_time 335 336 elif origin == 'end': 337 while self.read() is not None: 338 pass 339 if self.prev_record is None: 340 return None 341 end_timestamp = self._get_msec_timestamp(self.prev_record) 342 desired_time = end_timestamp + offset 343 if offset < 0: 344 self._reset() 345 self._read_until(desired_time) 346 return desired_time 347 348 else: 349 raise ValueError('Unknown origin value: "%s"' % origin) 350 351 ############################ 352 # Read a range of records beginning with timestamp start 353 # milliseconds, and ending *before* timestamp stop milliseconds. 354 def read_time_range(self, start=None, stop=None): 355 if self.filebase is None: 356 raise ValueError('read_time_range() not allowed on stdin') 357 358 # TODO: Is this needed? stop=None would be OK unless records are 359 # being written faster than they're being read. 360 if stop is None: 361 if self.tail: 362 raise ValueError('tail=True incompatible with stop=None') 363 if self.refresh_file_spec: 364 raise ValueError('refresh_file_spec=True incompatible with stop=None') 365 366 if start is None: 367 starting_offset = 0 368 else: 369 starting_offset = start - self._get_first_msec_timestamp() 370 371 self.seek_time(starting_offset, 'start') 372 records = [] 373 while True: 374 record = self.read() 375 if record is None: 376 break 377 if stop and self._get_msec_timestamp(record) >= stop: 378 break 379 records.append(record) 380 return records
Read lines from one or more text files. Sequentially open all files that match the file_spec.
Expect that each line will either be a string prefixed by a timestamp that follows the time_format parameter (ISO8601 by default) or a line of JSON encoding a DASRecord.
If line is a string prefixed by a timestamp, return the string. If JSON, return the DASRecord encoded by the JSON string.
LogfileReader( filebase=None, tail=False, refresh_file_spec=False, retry_interval=0.1, interval=0, use_timestamps=False, time_acceleration_factor=1.0, record_format=None, time_format='%Y-%m-%dT%H:%M:%S.%fZ', date_format='%Y-%m-%d', eol=None, quiet=False, **kwargs)
34 def __init__(self, filebase=None, tail=False, refresh_file_spec=False, 35 retry_interval=0.1, interval=0, use_timestamps=False, 36 time_acceleration_factor=1.0, 37 record_format=None, 38 time_format=timestamp.TIME_FORMAT, 39 date_format=timestamp.DATE_FORMAT, 40 eol=None, quiet=False, **kwargs): 41 """ 42 ``` 43 filebase Possibly wildcarded string specifying files to be opened. 44 Special case: if file_spec is None, read from stdin. 45 46 tail If False, return None upon reaching end of last file; if 47 True, block upon reaching EOF of last file and wait for 48 more records. 49 50 refresh_file_spec 51 If True, refresh the search for matching filenames when 52 reaching last EOF to see if any new matching files have 53 appeared in the interim. 54 55 retry_interval 56 If tail and/or refresh_file_spec are True, how long to 57 wait before looking to see if any new records or files 58 have shown up. 59 60 interval 61 How long to sleep between returning records. In general 62 this should be zero except for debugging purposes. 63 64 use_timestamps 65 If True, use the timestamps from the log file to determine 66 at what interval each record should be emitted. 67 68 time_acceleration_factor 69 When use_timestamps is True, multiplies the time intervals 70 between records by this factor. Values greater than 1.0 will 71 speed up playback, values between 0 and 1.0 will slow it down. 72 Default is 1.0 (normal speed). 73 74 record_format 75 If specified, a custom record format to use for extracting 76 timestamp and record. The default is '{timestamp:ti} {record}'. 77 78 eol Optional character by which to recognize the end of a record 79 80 quiet - if not False, don't complain when unable to parse a record. 81 82 ``` 83 Note that the order in which files are opened will probably be in 84 alphanumeric by filename, but this is not strictly enforced and 85 depends on how glob returns them. 86 """ 87 super().__init__(**kwargs) 88 89 if not PARSE_INSTALLED: 90 raise ImportError('LogfileReader requires Python "parse" module; ' 91 'please run "pip install parse"') 92 if interval and use_timestamps: 93 raise ValueError('Can not specify both "interval" and "use_timestamps"') 94 95 self.filebase = filebase 96 self.use_timestamps = use_timestamps 97 98 # Validate time_acceleration_factor 99 if time_acceleration_factor is None: 100 raise ValueError("time_acceleration_factor must be a number") 101 if not isinstance(time_acceleration_factor, (int, float)): 102 raise ValueError("time_acceleration_factor must be a number") 103 if time_acceleration_factor <= 0: 104 raise ValueError("time_acceleration_factor must be greater than zero") 105 self.time_acceleration_factor = time_acceleration_factor 106 107 self.record_format = record_format or '{timestamp:ti} {record}' 108 self.compiled_record_format = parse.compile(self.record_format) 109 self.date_format = date_format 110 self.time_format = time_format 111 self.tail = tail 112 self.refresh_file_spec = refresh_file_spec 113 self.eol = eol 114 self.quiet = quiet 115 116 # If use_timestamps, we need to keep track of our last_read to 117 # know how long to sleep 118 self.last_timestamp = 0 119 self.last_read = 0 120 121 self._first_msec_timestamp = None 122 self.prev_record = None 123 124 # If they give us a filebase, add wildcard to match its suffixes; 125 # otherwise, we'll pass on the empty string to TextFileReader so 126 # that it uses stdin. NOTE: we should really use a pattern that 127 # echoes timestamp.DATE_FORMAT, e.g. 128 # DATE_FORMAT_WILDCARD = '????-??-??' 129 self.file_spec = filebase + '*' if filebase else None 130 self.reader = TextFileReader(file_spec=self.file_spec, 131 tail=tail, 132 refresh_file_spec=refresh_file_spec, 133 retry_interval=retry_interval, 134 interval=interval, eol=eol)
filebase Possibly wildcarded string specifying files to be opened.
Special case: if file_spec is None, read from stdin.
tail If False, return None upon reaching end of last file; if
True, block upon reaching EOF of last file and wait for
more records.
refresh_file_spec
If True, refresh the search for matching filenames when
reaching last EOF to see if any new matching files have
appeared in the interim.
retry_interval
If tail and/or refresh_file_spec are True, how long to
wait before looking to see if any new records or files
have shown up.
interval
How long to sleep between returning records. In general
this should be zero except for debugging purposes.
use_timestamps
If True, use the timestamps from the log file to determine
at what interval each record should be emitted.
time_acceleration_factor
When use_timestamps is True, multiplies the time intervals
between records by this factor. Values greater than 1.0 will
speed up playback, values between 0 and 1.0 will slow it down.
Default is 1.0 (normal speed).
record_format
If specified, a custom record format to use for extracting
timestamp and record. The default is '{timestamp:ti} {record}'.
eol Optional character by which to recognize the end of a record
quiet - if not False, don't complain when unable to parse a record.
Note that the order in which files are opened will probably be in alphanumeric by filename, but this is not strictly enforced and depends on how glob returns them.
def
read(self):
137 def read(self): 138 """ 139 Return the next line in the file(s), or None if there are no more 140 records (as opposed to '' if the next record is a blank line). To test 141 EOF you'll need to test 142 143 if record is None: 144 no more records... 145 146 rather than simply 147 148 if not record: 149 could be EOF or simply an empty next line 150 """ 151 152 # NOTE: It feels like we should check here that the reader's 153 # current file really does match our logfile name format... 154 while True: 155 record = self.reader.read() 156 if not record: # None means we're out of records 157 return record 158 159 # If we've got a record and we're not using timestamps, we're 160 # done - just return it. 161 if not self.use_timestamps: 162 self.prev_record = record 163 # We need this in case the next call is seek_time() or 164 # read_time_range(). This is less expensive than parsing every 165 # timestamp and keeping self.last_timestamp, but an 166 # alternative might be to implement read_previous(), which 167 # would be expensive but which could be called only when 168 # actually needed. 169 170 # Check whether this is a JSON-encoded DASRecord. If so, return 171 # as a DASRecord; otherwise, just return as string. Yes, this 172 # adds overhead, but our assumption is that the throughput on 173 # a LogfileReader is going to be pretty low. 174 try: 175 das_record = DASRecord(json_str=record) 176 return das_record 177 except json.JSONDecodeError: 178 return record 179 180 # If we're here, we're going to be doling out records according to the 181 # differences in their timestamps. 182 183 # Try to parse the timestamp off the front. If we succeed, grab the 184 # timestamp and break out of loop. 185 try: 186 parsed_record = self.compiled_record_format.parse(record).named 187 ts = parsed_record['timestamp'].timestamp() 188 break 189 # We had a problem parsing as a timestamped string. 190 except (KeyError, ValueError, AttributeError): 191 pass 192 193 # Try parsing as JSON DASRecord. If we succeed, grab the 194 # timestamp and break out of loop. 195 try: 196 record = DASRecord(json_str=record) 197 ts = record.timestamp 198 break 199 except json.JSONDecodeError: 200 pass 201 202 # If we're here, we failed to parse the record. Complain, if appropriate, 203 # then spit out it out without updating timestamps. 204 if not self.quiet: 205 logging.warning('Unable to parse record into DASRecord') 206 logging.warning(f'Unable to parse record into "{self.record_format}"') 207 logging.warning(f'Record: "{record}"') 208 return record 209 210 # If this is not our first read, figure out how long we need to wait 211 # for our next one. 212 if self.last_read > 0: 213 # If here, we've got a record and a timestamp and are intending to 214 # use it. Figure out how long we should sleep before returning it. 215 desired_interval = ts - self.last_timestamp 216 217 # Apply the time acceleration factor to the desired interval 218 desired_interval = desired_interval / self.time_acceleration_factor 219 220 now = timestamp.timestamp() 221 actual_interval = now - self.last_read 222 logging.debug('Desired interval %f, actual %f; sleeping %f', 223 desired_interval, actual_interval, 224 max(0, desired_interval - actual_interval)) 225 time.sleep(max(0, desired_interval - actual_interval)) 226 227 self.last_timestamp = ts 228 self.last_read = timestamp.timestamp() 229 230 self.prev_record = record 231 return record
Return the next line in the file(s), or None if there are no more records (as opposed to '' if the next record is a blank line). To test EOF you'll need to test
if record is None: no more records...
rather than simply
if not record: could be EOF or simply an empty next line
def
seek_time(self, offset=0, origin='current'):
274 def seek_time(self, offset=0, origin='current'): 275 """ 276 Behavior is intended to mimic file seek() behavior but with 277 respect to timestamps. 278 After calling this, the next record read will be the first record 279 whose timestamp is the same as or later than the requested time; 280 if no such record is found, it will read to the end. 281 Exception: if the records are not in exact chronological order, 282 records appearing before the current record but with a later 283 timestamp might be missed. 284 285 Args: 286 offset: offset in msec relative to origin 287 origin: 'start', 'current' or 'end' 288 289 Returns: 290 Requested time in msec, i.e. timestamp of (T0 + offset), 291 where T0 = timestamp(first record) if origin = 'start' 292 = timestamp(next record) if origin = 'current' and next record is not None 293 = timestamp(last record) if origin = 'current' and next record is None 294 = timestamp(last record) if origin = 'end' 295 Returns None if no timestamps were found 296 """ 297 if self.filebase is None: 298 raise ValueError('seek_time() not allowed on stdin') 299 300 # TODO: Maybe these are OK, as long as 'end' is defined as the point where 301 # read() returns None for the first time. 302 if self.tail and origin == 'end': 303 raise ValueError('tail=True incompatible with origin == "end"') 304 if self.refresh_file_spec and origin == 'end': 305 raise ValueError('refresh_file_spec=True incompatible with origin == "end"') 306 307 if origin == 'start': 308 if offset < 0: 309 raise ValueError("Can't back up past earliest record") 310 first_timestamp = self._get_first_msec_timestamp() 311 if first_timestamp is None: 312 return None 313 desired_time = first_timestamp + offset 314 if self.prev_record is None: 315 self._reset() 316 else: 317 prev_timestamp = self._get_msec_timestamp(self.prev_record) 318 if prev_timestamp >= desired_time: 319 self._reset() 320 self._read_until(desired_time) 321 return desired_time 322 323 elif origin == 'current': 324 next_timestamp = self._peek_msec() 325 curr_timestamp = next_timestamp or self._get_msec_timestamp(self.prev_record) 326 if curr_timestamp is None: 327 return None 328 desired_time = curr_timestamp + offset 329 if offset == 0: 330 return desired_time 331 if offset < 0: 332 self._reset() 333 self._read_until(desired_time) 334 return desired_time 335 336 elif origin == 'end': 337 while self.read() is not None: 338 pass 339 if self.prev_record is None: 340 return None 341 end_timestamp = self._get_msec_timestamp(self.prev_record) 342 desired_time = end_timestamp + offset 343 if offset < 0: 344 self._reset() 345 self._read_until(desired_time) 346 return desired_time 347 348 else: 349 raise ValueError('Unknown origin value: "%s"' % origin)
Behavior is intended to mimic file seek() behavior but with respect to timestamps. After calling this, the next record read will be the first record whose timestamp is the same as or later than the requested time; if no such record is found, it will read to the end. Exception: if the records are not in exact chronological order, records appearing before the current record but with a later timestamp might be missed.
Args: offset: offset in msec relative to origin origin: 'start', 'current' or 'end'
Returns: Requested time in msec, i.e. timestamp of (T0 + offset), where T0 = timestamp(first record) if origin = 'start' = timestamp(next record) if origin = 'current' and next record is not None = timestamp(last record) if origin = 'current' and next record is None = timestamp(last record) if origin = 'end' Returns None if no timestamps were found
def
read_time_range(self, start=None, stop=None):
354 def read_time_range(self, start=None, stop=None): 355 if self.filebase is None: 356 raise ValueError('read_time_range() not allowed on stdin') 357 358 # TODO: Is this needed? stop=None would be OK unless records are 359 # being written faster than they're being read. 360 if stop is None: 361 if self.tail: 362 raise ValueError('tail=True incompatible with stop=None') 363 if self.refresh_file_spec: 364 raise ValueError('refresh_file_spec=True incompatible with stop=None') 365 366 if start is None: 367 starting_offset = 0 368 else: 369 starting_offset = start - self._get_first_msec_timestamp() 370 371 self.seek_time(starting_offset, 'start') 372 records = [] 373 while True: 374 record = self.read() 375 if record is None: 376 break 377 if stop and self._get_msec_timestamp(record) >= stop: 378 break 379 records.append(record) 380 return records