log_read: Just warn when file shrinks, try reading anyway

[pyi2ncommon] / src / log_read.py

# The software in this package is distributed under the GNU General
# Public License version 2 (with a special exception described below).
#
# A copy of GNU General Public License (GPL) is included in this distribution,
# in the file COPYING.GPL.
#
# As a special exception, if other files instantiate templates or use macros
# or inline functions from this file, or you compile this file and link it
# with other works to produce a work based on this file, this file
# does not by itself cause the resulting work to be covered
# by the GNU General Public License.
#
# However the source code for this file must still be made available
# in accordance with section (3) of the GNU General Public License.
#
# This exception does not invalidate any other reasons why a work based
# on this file might be covered by the GNU General Public License.
#
# Copyright (c) 2016-2018 Intra2net AG <info@intra2net.com>

""" Iterative reading of log files

Basic Functionality (class :py:class:`IterativeReader`):
Runs stat in a loop to find out whether file size has changed. Then reads the
new data and forwards that

.. todo:: Want to also use lsof to find out whether file/pipe/socket was closed,
         so can return from read loop

:py:class:`LineReader` takes output of :py:class:`IterativeReader` and returns
it line-wise as is normal for log files

:py:class:`LogParser` takes those lines and tries to parse them into fields
like date, time, module name, urgency and message.

.. todo:: auto-detect log line layout
"""

import os
import re
from warnings import warn
import os.path
import logging
from .iter_helpers import zip_longest
from .type_helpers import is_str_or_byte, is_file_obj


class LogReadWarning(UserWarning):
    """ warnings issued by classes in this module """
    pass


def true_func(unused_argument_but_that_is_ok):
    """ does nothing, always returns True """
    return True


def check_is_used(some_file_or_handle):
    """ check whether file is being written to

    to be implemented, e.g. using lsof
    """
    raise NotImplementedError()


_create_description_unknown_counter = 0


def create_description(file_obj, file_desc):
    """ create some description for given file-like object / file descriptor

    :param file_obj: file-like object
    :param int file_desc: os-level file descriptor
    :returns: string
    """

    global _create_description_unknown_counter

    try:
        desc = file_obj.name
        if desc:
            return desc
    except AttributeError:
        pass

    if file_desc is not None:
        return 'file{0}'.format(file_desc)
    else:
        _create_description_unknown_counter += 1
        return 'unknown{0}'.format(_create_description_unknown_counter)


#: error message for IterativeReader constructor
_STR_ERR = 'not accepting file name "{0}" since cannot guarantee closing ' \
           'files --> use with open(file_name)!'


class IterativeReader(object):
    """ reads from a given file

    Uses os.stat(file_obj.fileno()).st_size as measure whether file has changed
    or not; Always reads as much data as possible

    Catches most common exceptions in iteration (not constructor)

    Does not care about closing files, so does not accept file names

    This is the base for class :py:class:`LineReader` that just has to
    implement a different :py:meth:`prepare_result` method
    """

    def __init__(self, sources, descs=None, return_when_done=False):
        """ creates a reader; does some basic checks on args

        :param sources: iterable over sources. Sources can be opened file
                        objects or read-opened os-level file descriptors.
                        Calling code has to ensure they are closed properly, so
                        best use this within a "with open(file_name) as
                        file_handle:"-context. If sources is a single file
                        obj/descriptor, both source and desc will be converted
                        to lists of length 1
        :param descs: can be anything of same length as sources. If sources is
                      a single source, then descs is also converted to a list
                      of length 1. If not given (i.e. None), will use
                      :py:func:`create_description` to guess descriptions
        :param bool return_when_done: ignore file_handle if no-one is writing
                                      to it any more. Return from iterator when
                                      all watched files are done (not
                                      implemented yet)
        :raises: OSError when testing fstat on source
        """
        if not sources:
            raise ValueError('need at least some source!')
        elif is_str_or_byte(sources):
            raise ValueError(_STR_ERR.format(sources))
        elif is_file_obj(sources) or isinstance(sources, int):
            source_input = [sources, ]
            desc_input = [descs, ]
        else:
            source_input = sources  # assume some iterable
            desc_input = descs

        # now divide sources into os-level file descriptors for os.fstat,
        # and file objects for read()
        self.file_objs = []
        self.file_descs = []          # file descriptOR, not descriptION
        for source in source_input:
            if is_file_obj(source):
                self.file_objs.append(source)
                self.file_descs.append(source.fileno())
            elif isinstance(source, int):
                self.file_objs.append(os.fdopen(source))
                self.file_descs.append(source)
            elif is_str_or_byte(source):
                raise ValueError(_STR_ERR.format(source))
            else:
                raise ValueError('source {0} is neither file obj nor file '
                                 'descriptor!')

            # try to fstat the new file descriptor just for testing
            os.fstat(self.file_descs[-1])

        # guess descriptions if not given
        if not desc_input:
            self.descriptions = [create_description(obj, file_desc)
                                 for obj, file_desc
                                 in zip(self.file_objs, self.file_descs)]
        else:
            try:
                if len(desc_input) != len(self.file_objs):
                    raise ValueError('need same number of sources and '
                                     'descriptions!')
            except TypeError:
                pass  # desc_input is generator or so

            self.descriptions = []
            for obj, file_desc, description in \
                    zip_longest(self.file_objs, self.file_descs, desc_input):
                if obj is None:
                    raise ValueError('more descriptions than sources!')
                elif description is None:
                    self.descriptions.append(create_description(obj,
                                                                file_desc))
                else:
                    self.descriptions.append(description)

        self.last_sizes = [0 for _ in self.file_objs]
        self.ignore = [False for _ in self.file_objs]

        if return_when_done:
            self.is_used = check_is_used
        else:
            self.is_used = true_func

        for obj, file_desc, description in \
                zip(self.file_objs, self.file_descs, self.descriptions):
            logging.debug('log_read initialized with file descriptor {0}, '
                          'file obj {1}, description "{2}"'
                          .format(file_desc, obj, description))

    def n_sources(self):
        return len(self.file_objs)

    def n_active_sources(self):
        return len(self.ignore) - sum(self.ignore)

    def __iter__(self):
        while True:
            for idx, (obj, file_desc, description, last_size, do_ignore) in \
                    enumerate(zip(self.file_objs, self.file_descs,
                                  self.descriptions, self.last_sizes,
                                  self.ignore)):
                if do_ignore:
                    continue

                # get new file size
                new_size = os.fstat(file_desc).st_size

                # compare to old size
                if new_size == last_size:
                    if not self.is_used(file_desc):
                        warn('no one is writing to {0} / {1} -- '
                             'stop watching it!'
                             .format(file_desc, description),
                             category=LogReadWarning)
                        self.ignore[idx] = True
                else:
                    if new_size < last_size:
                        warn('{0} / {1} has become smaller ({2} --> {3})!'
                             .format(obj, description, last_size, new_size),
                             category=LogReadWarning)
                    try:
                        new_data = obj.read()
                    except OSError as ose:    # includes IOErrors
                        warn('io error reading from {0} / {1}: {2})'
                             .format(obj, description, ose),
                             category=LogReadWarning)
                        new_data = str(ose)
                    except UnicodeDecodeError as ude:
                        warn('unicode error reading from {0} / {1}: {2}'
                             .format(obj, description, ude),
                             category=LogReadWarning)
                        new_data = str(ude)

                    # post-processing
                    to_yield = self.prepare_result(description, new_data, idx)
                    for result in to_yield:
                        yield result

                    # prepare next iteration
                    self.last_sizes[idx] = new_size

    def prepare_result(self, description, data, idx):
        """ from raw new data create some yield-able results

        to be intended for overwriting in sub-classes

        this function is called from __iter__ for each new data that becomes
        available. It has to return some iterable whose entries are yielded
        from iteration over objects of this class.

        It receives the following args:
        - the description of the source
        - the data itself
        - the index of the source
        The result must be an iterable of objects, which are yielded as-is, so
        can have any form

        This base implementation just returns its input in a list, so new data
        is yielded from __iter__ as-is
        """
        return [(description, data), ]


LINE_SPLITTERS = '\n\r'


class LineReader(IterativeReader):
    """ an IterativeReader that returns new data line-wise

    this means buffering partial line data
    """

    def __init__(self, *args, **kwargs):
        """ forwards all args and kwargs to :py:class:`IterativeReader` """
        super(LineReader, self).__init__(*args, **kwargs)
        self.line_buffers = ['' for _ in range(self.n_sources())]

    def prepare_result(self, description, new_data, idx):
        """ take raw new data and split it into lines

        if line is not complete, then buffer it

        returns lines without their newline characters
        """
        all_data = self.line_buffers[idx] + new_data
        self.line_buffers[idx] = ''
        result = []
        should_be_no_new_lines = False
        for line in all_data.splitlines(True):
            if line[-1] in LINE_SPLITTERS:
                result.append((description, line.rstrip(LINE_SPLITTERS)))
            elif should_be_no_new_lines:
                raise ValueError('line splitters are not compatible with'
                                 'str.splitlines!')
            else:
                self.line_buffers[idx] = line
                should_be_no_new_lines = True  # (this should be the last)

        return result


class LogParser(LineReader):
    """ takes lines from LineReader and parses their contents

    requires a pattern for log lines, auto-detection is not implemented yet

    Iteration returns re.match result or -- if matching failed -- the original
    raw line
    """

    def __init__(self, log_file, pattern=None):
        """ create a LogParser

        :param str log_file: name of log file to parse (required!)
        :param pattern: regexp to split log lines; None (default) to return
                        line as they are
        """
        super(LogParser, self).__init__(log_file)

        self.pattern = pattern

    def prepare_result(self, *args):
        # let super class split data into lines
        for _, raw_line in super(LogParser, self).prepare_result(*args):
            result = re.match(self.pattern, raw_line)
            if result:
                yield result
            else:
                yield raw_line