logs

Base log objects

To import...

from dynamite_nsm.services.base import logs

LogFile

__init__(self, log_path, log_sample_size=500, gzip_decode=False) special

A Generic log object

Parameters:

Name	Type	Description	Default
log_path	str	The path to a log file	required
log_sample_size	Optional[int]	The number of most recent entries to include	500
gzip_decode	Optional[bool]	If True, we'll decode the log before reading it in	False

Source code in dynamite_nsm/services/base/logs.py

def __init__(self, log_path: str, log_sample_size: Optional[int] = 500, gzip_decode: Optional[bool] = False):
    """A Generic log object
    Args:
        log_path: The path to a log file
        log_sample_size: The number of most recent entries to include
        gzip_decode: If True, we'll decode the log before reading it in
    """
    if not utilities.is_setup():
        raise exceptions.DynamiteNotSetupError()
    self.log_path = log_path
    self.log_sample_size = log_sample_size
    self.exists = False
    self.current_line = 0
    if gzip_decode and not log_path.endswith('.decoded'):
        decoded_log_path = log_path + '.decoded'
        if not os.path.exists(decoded_log_path):
            with open(decoded_log_path, 'w') as out:
                with gzip.open(log_path, 'rb') as f:
                    line = f.readline().decode('utf-8', errors='ignore')
                    while line:
                        out.write(line)
                        try:
                            line = f.readline().decode('utf-8', errors='ignore')
                        except gzip.BadGzipFile:
                            pass

        self.log_path = decoded_log_path
    linecache.updatecache(self.log_path)
    self.last_line_num = self.find_latest_line_offset()
    if self.last_line_num < self.log_sample_size:
        self.entries = [entry for entry in self.iter_cache(start=1)]
    else:
        self.entries = [entry for entry in self.iter_cache(start=self.last_line_num - self.log_sample_size + 1)]

find_latest_line_offset(self, step=500000)

Relatively fast way of finding the latest offset; algorithm guesses high offset and if over divides the step by half and repeats

Parameters:

Name	Type	Description	Default
step	Optional[int]	The starting step between line offsets	500000

Returns:

Type	Description
int	Most recent line number

Source code in dynamite_nsm/services/base/logs.py

def find_latest_line_offset(self, step: Optional[int] = 500000) -> int:
    """Relatively fast way of finding the latest offset; algorithm guesses high offset and if over divides the step
    by half and repeats
    Args:
        step: The starting step between line offsets
    Returns:
         Most recent line number
    """
    offset = 1
    while step > 0:
        for _ in self.iter_cache(start=offset, step=step):
            offset += step
        step = int(step/2)
        offset -= step
    return offset

iter_cache(self, start=1, step=1)

Relatively Memory efficient method of accessing very large files on disk

Parameters:

Name	Type	Description	Default
start	Optional[int]	The starting line	1
step	Optional[int]	The step between line offsets	1

Returns:

Type	Description
Generator	The line at a particular offset

Source code in dynamite_nsm/services/base/logs.py

def iter_cache(self, start: Optional[int] = 1, step: Optional[int] = 1) -> Generator:
    """Relatively Memory efficient method of accessing very large files on disk
    Args:
        start: The starting line
        step: The step between line offsets
    Returns:
         The line at a particular offset
    """

    i = start
    while True:
        line = linecache.getline(self.log_path, i)
        if line:
            yield line
        else:
            break
        i += step

refresh(self)

Refresh linecache

Returns:

Type	Description
None	None

Source code in dynamite_nsm/services/base/logs.py

def refresh(self) -> None:
    """
    Refresh linecache

    Returns:
        None

    """
    linecache.updatecache(self.log_path)
    if self.last_line_num < self.log_sample_size:
        self.entries = [entry for entry in self.iter_cache(start=1)]
    else:
        self.entries = [entry for entry in self.iter_cache(start=self.last_line_num - self.log_sample_size + 1)]

size(self)

Get the log file size with last offset reached

Returns: A LogFileSize object containing the latest line offset and the total number of log entries available

Source code in dynamite_nsm/services/base/logs.py

def size(self) -> LogFileSize:
    """
    Get the log file size with last offset reached

    Returns: A LogFileSize object containing the latest line offset and the total number of log entries available

    """
    return LogFileSize(self.find_latest_line_offset(), len(self.entries))

LogFileSize

__init__(self, file_line_count, loaded_entries) special

A simple object that represents the latest line offset reached and the number of entries loaded into memory.

Parameters:

Name	Type	Description	Default
file_line_count	int	The offset of the last reached (cached) line	required
loaded_entries	int	The number of entries loaded into memory	required

Source code in dynamite_nsm/services/base/logs.py

def __init__(self, file_line_count: int, loaded_entries: int):
    """
    A simple object that represents the latest line offset reached and the number of entries loaded into memory.
    Args:
        file_line_count: The offset of the last reached (cached) line
        loaded_entries: The number of entries loaded into memory
    """
    self.file_line_count = file_line_count
    self.loaded_entries = loaded_entries