xref: /aosp_15_r20/external/autotest/client/cros/bluetooth/logger_helper.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

# Lint as: python2, python3
# Copyright 2021 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Helper class to manage bluetooth logs"""

from datetime import datetime
import logging
import os
import re
import subprocess
import time


SYSLOG_PATH = '/var/log/messages'


class LogManager(object):
    """The LogManager class helps to collect logs without a listening thread"""

    DEFAULT_ENCODING = 'utf-8'

    class LoggingException(Exception):
        """A stub exception class for LogManager class."""
        pass

    def __init__(self, log_path=SYSLOG_PATH, raise_missing=False):
        """Initialize log manager object

        @param log_path: string path to log file to manage
        @param raise_missing: raise an exception if the log file is missing

        @raises: LogManager.LoggingException on non-existent log file
        """
        if not os.path.isfile(log_path):
            msg = 'Requested log file {} does not exist'.format(log_path)
            if raise_missing:
                raise LogManager.LoggingException(msg)
            else:
                self._LogErrorToSyslog(msg)

        self.log_path = log_path

        self.ResetLogMarker()
        self._bin_log_contents = []

    def _LogErrorToSyslog(self, message):
        """Create a new syslog file and add a message to syslog."""
        subprocess.call(['reload', 'syslog'])
        subprocess.call(['logger', message])

    def _GetSize(self):
        """Get the size of the log"""
        try:
            return os.path.getsize(self.log_path)
        except Exception as e:
            logging.error('Failed to get log size: {}'.format(e))
            return 0

    def ResetLogMarker(self, now_size=None):
        """Reset the start-of-log marker for later comparison"""
        if now_size is None:
            now_size = self._GetSize()
        self.initial_log_size = now_size

    def StartRecording(self):
        """Mark initial log size for later comparison"""

        self._bin_log_contents = []

    def StopRecording(self):
        """Gather the logs since StartRecording was called

        @raises: LogManager.LoggingException if:
                - Log file disappeared since StartRecording was called
                - Log file is smaller than when logging began
                - StartRecording was never called
        """
        initial_size = self.initial_log_size
        now_size = self._GetSize()

        if not os.path.isfile(self.log_path):
            msg = 'File {} disappeared unexpectedly'.format(self.log_path)
            raise LogManager.LoggingException(msg)

        if now_size < initial_size:
            msg = 'Log became smaller unexpectedly'
            raise LogManager.LoggingException(msg)

        with open(self.log_path, 'rb') as mf:
            # Skip to the point where we started recording
            mf.seek(self.initial_log_size)

            readsize = now_size - self.initial_log_size
            self._bin_log_contents = mf.read(readsize).split(b'\n')

        # Re-set start of log marker
        self.ResetLogMarker(now_size)

    def LogContains(self, search_str):
        """Performs simple string checking on each line from the collected log

        @param search_str: string to be located within log contents. This arg
                is expected to not span between lines in the logs

        @returns: True if search_str was located in the collected log contents,
                False otherwise
        """
        pattern = re.compile(search_str.encode(self.DEFAULT_ENCODING))
        for line in self._bin_log_contents:
            if pattern.search(line):
                return True

        return False

    def FilterOut(self, rm_reg_exp):
        """Remove lines with specified pattern from the log file

        @param rm_reg_exp: regular expression of the lines to be removed
        """
        # If log_path doesn't exist, there's nothing to do
        if not os.path.isfile(self.log_path):
            return

        rm_line_cnt = 0
        initial_size = self._GetSize()
        rm_pattern = re.compile(rm_reg_exp.encode(self.DEFAULT_ENCODING))

        with open(self.log_path, 'rb+') as mf:
            lines = mf.readlines()
            mf.seek(0)
            for line in lines:
                if rm_pattern.search(line):
                    rm_line_cnt += 1
                else:
                    mf.write(line)
            mf.truncate()

        # Some tracebacks point out here causing /var/log/messages missing but
        # we don't have many clues. Adding a check and logs here.
        if not os.path.isfile(self.log_path):
            msg = '{} does not exist after FilterOut'.format(self.log_path)
            logging.warning(msg)
            self._LogErrorToSyslog(msg)

        new_size = self._GetSize()
        rm_byte = initial_size - new_size
        logging.info('Removed number of line: %d, Reduced log size: %d byte',
                     rm_line_cnt, rm_byte)

        # Note the new size of the log
        self.ResetLogMarker(new_size)


class InterleaveLogger(LogManager):
    """LogManager class that focus on interleave scan"""

    # Example bluetooth kernel log:
    # "2020-11-23T07:52:31.395941Z DEBUG kernel: [ 6469.811135] Bluetooth: "
    # "cancel_interleave_scan() hci0: cancelling interleave scan"
    KERNEL_LOG_PATTERN = ('([^ ]+) DEBUG kernel: \[.*\] Bluetooth: '
                          '{FUNCTION}\(\) hci0: {LOG_STR}')
    STATE_PATTERN = KERNEL_LOG_PATTERN.format(
            FUNCTION='hci_req_add_le_interleaved_scan',
            LOG_STR='next state: (.+)')
    CANCEL_PATTERN = KERNEL_LOG_PATTERN.format(
            FUNCTION='cancel_interleave_scan',
            LOG_STR='cancelling interleave scan')
    SYSTIME_LENGTH = len('2020-12-18T00:11:22.345678')

    def __init__(self):
        """ Initialize object
        """
        self.reset()
        self.state_pattern = re.compile(
                self.STATE_PATTERN.encode(self.DEFAULT_ENCODING))
        self.cancel_pattern = re.compile(
                self.CANCEL_PATTERN.encode(self.DEFAULT_ENCODING))
        super(InterleaveLogger, self).__init__()

    def reset(self):
        """ Clear data between each log collection attempt
        """
        self.records = []
        self.cancel_events = []

    def StartRecording(self):
        """ Reset the previous data and start recording.
        """
        self.reset()
        super(InterleaveLogger, self).ResetLogMarker()
        super(InterleaveLogger, self).StartRecording()

    def StopRecording(self):
        """ Stop recording and parse logs
            The following data will be set after this call

            - self.records: a dictionary where each item is a record of
                            interleave |state| and the |time| the state starts.
                            |state| could be {'no filter', 'allowlist'}
                            |time| is system time in sec

            - self.cancel_events: a list of |time| when a interleave cancel
                                  event log was found
                                  |time| is system time in sec

            @returns: True if StopRecording success, False otherwise

        """
        try:
            super(InterleaveLogger, self).StopRecording()
        except Exception as e:
            logging.error(e)
            return False

        success = True

        def sys_time_to_timestamp(time_str):
            """ Return timestamp of time_str """

            # This is to remove the suffix of time string, in some cases the
            # time string ends with an extra 'Z', in other cases, the string
            # ends with time zone (ex. '+08:00')
            time_str = time_str[:self.SYSTIME_LENGTH]

            try:
                dt = datetime.strptime(time_str, "%Y-%m-%dT%H:%M:%S.%f")
            except Exception as e:
                logging.error(e)
                success = False
                return 0

            return time.mktime(dt.timetuple()) + dt.microsecond * (10**-6)

        for line in self._bin_log_contents:
            line = line.strip().replace(b'\\r\\n', b'')
            state_pattern = self.state_pattern.search(line)
            cancel_pattern = self.cancel_pattern.search(line)

            if cancel_pattern:
                time_str = cancel_pattern.groups()[0].decode(
                        self.DEFAULT_ENCODING)
                time_sec = sys_time_to_timestamp(time_str)
                self.cancel_events.append(time_sec)

            if state_pattern:
                time_str, state = [
                        x.decode(self.DEFAULT_ENCODING)
                        for x in state_pattern.groups()
                ]
                time_sec = sys_time_to_timestamp(time_str)
                self.records.append({'time': time_sec, 'state': state})

        return success