# Copyright (c) 2013-2016 Intel Corporation # # Permission is hereby granted, free of charge, to any person obtaining a # copy of this software and associated documentation files (the "Software"), # to deal in the Software without restriction, including without limitation # the rights to use, copy, modify, merge, publish, distribute, sublicense, # and/or sell copies of the Software, and to permit persons to whom the # Software is furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice (including the next # paragraph) shall be included in all copies or substantial portions of the # Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL # THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING # FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS # IN THE SOFTWARE. """ Module implementing classes for reading posix dmesg Currently this module only has the default DummyDmesg, and a LinuxDmesg. The method used on Linux requires that timetamps are enabled, and no other posix system has timestamps. On OSX and *BSD one would likely want to implement a system that reads the sysloger, since timestamps can be added by the sysloger, and are not inserted by dmesg. Most users do not want to call a specific dmesg implementation themselves, they will want to use the get_dmesg() helper, which provides the proper dmesg implementation for their OS. """ from __future__ import ( absolute_import, division, print_function, unicode_literals ) import abc import gzip import re import subprocess import sys import warnings import six from framework import exceptions __all__ = [ 'BaseDmesg', 'DummyDmesg', 'LinuxDmesg', 'get_dmesg', ] @six.add_metaclass(abc.ABCMeta) class BaseDmesg(object): """ Abstract base class for Dmesg derived objects This provides the bases of the constructor, and most subclasses should call super() in __init__(). It provides a concrete implementation of the update_result() method, which should be suitible for all subclasses. The update_dmesg() method is the primary method that each subclass needs to override, as this method is used to to actually read the dmesg ringbuffer, and the command used, and the options given are OS dependent. This class is not thread safe, because it does not black between the start of the test and the reading of dmesg, which means that if two tests run at the same time, and test A creates an entri in dmesg, but test B finishes first, test B will be marked as having the dmesg error. """ @abc.abstractmethod def __init__(self): # A list containing all messages since the last time dmesg was read. # This is used by update result, and then emptied self._new_messages = [] # If this is set it should be an re.compile() object, which matches # entries that are to be considered failures. If an entry does not # match the regex it will be ignored. By default (None) all entries are # considered self.regex = None # Populate self.dmesg initially, otherwise the first test will always # be full of false positives. self.update_dmesg() @abc.abstractmethod def update_dmesg(self): """ Update _new_messages list This method should read dmesg, determine which entries are new since the last read of dmesg, and update self._new_messages with those entries. """ pass def update_result(self, result): """ Takes a TestResult object and updates it with dmesg statuses If dmesg is enabled, and if dmesg has been updated, then replace pass with dmesg-warn and warn and fail with dmesg-fail. If dmesg has not been updated, it will return the original result passed in. Arguments: result -- A TestResult instance """ def replace(res): """ helper to replace statuses with the new dmesg status Replaces pass with dmesg-warn, and warn and fail with dmesg-fail, otherwise returns the input """ return { "pass": "dmesg-warn", "warn": "dmesg-fail", "fail": "dmesg-fail" }.get(res, res) # Get a new snapshot of dmesg self.update_dmesg() # if update_dmesg() found new entries replace the results of the test # and subtests if self._new_messages: if self.regex: for line in self._new_messages: if self.regex.search(line): break else: return result result.result = replace(result.result) # Replace the results of any subtests for key, value in six.iteritems(result.subtests): result.subtests[key] = replace(value) # Add the dmesg values to the result result.dmesg = "\n".join(self._new_messages) return result def __repr__(self): return 'BaseDmesg()' class LinuxDmesg(BaseDmesg): """ Read dmesg on posix systems This reads the dmesg ring buffer, stores the contents of the buffer, and then compares it whenever called, returning the difference between the calls. It requires timestamps to be enabled. """ DMESG_COMMAND = ['dmesg', '--level', 'emerg,alert,crit,err,warn,notice'] def __init__(self): """ Create a dmesg instance """ # _last_message store the contents of the last entry in dmesg the last # time it was read. Because dmesg is a ringbuffer old entries can fall # off the end if it becomes too full. To track new entries search for # this entry and take any entries that were added after it. This works # on Linux because each entry is guaranteed unique by timestamps. self._last_message = None super(LinuxDmesg, self).__init__() # First check to see if we can find the live kernel config. try: with gzip.open(b"/proc/config.gz", 'r') as f: for line in f: if line.startswith(b"CONFIG_PRINTK_TIME=y"): return # If there is a malformed or missing file, just ignore it and try the # regex match (which may not exist if dmesg ringbuffer was cleared). except Exception: pass if not self._last_message: # We need to ensure that there is something in dmesg to check # timestamps. warnings.warn("Cannot check dmesg for timestamp support. If you " "do not have timestamps enabled in your kernel you " "get incomplete dmesg captures", RuntimeWarning) elif not re.match(r'\[\s*\d+\.\d+\]', self._last_message): # Do an initial check to ensure that dmesg has timestamps, we need # timestamps to work correctly. A proper linux dmesg timestamp # looks like this: [ 0.00000] raise exceptions.PiglitFatalError( "Your kernel does not seem to support timestamps, which " "are required by the --dmesg option") def update_dmesg(self): """ Call dmesg using subprocess.check_output Get the contents of dmesg, then calculate new messages, finally set self.dmesg equal to the just read contents of dmesg. """ dmesg = subprocess.check_output(self.DMESG_COMMAND).decode('utf-8') dmesg = dmesg.strip().splitlines() # Find all new entries, do this by slicing the list of dmesg to only # returns elements after the last element stored. If there are not # matches a value error is raised, that means all of dmesg is new l = 0 for index, item in enumerate(reversed(dmesg)): if item == self._last_message: l = len(dmesg) - index # don't include the matched element break self._new_messages = dmesg[l:] # Attempt to store the last element of dmesg, unless there was no dmesg self._last_message = dmesg[-1] if dmesg else None def __repr__(self): return 'LinuxDmesg()' class DummyDmesg(BaseDmesg): """ An dummy class for dmesg on non unix-like systems This implements a dummy version of the LinuxDmesg, and can be used anytime it is not desirable to actually read dmesg, such as non-posix systems, or when the contents of dmesg don't matter. """ DMESG_COMMAND = [] def __init__(self): pass def update_dmesg(self): """ Dummy version of update_dmesg """ pass def update_result(self, result): """ Dummy version of update_result """ return result def __repr__(self): return 'DummyDmesg()' def get_dmesg(not_dummy=True): """ Return a Dmesg type instance Normally this does a system check, and returns the type that proper for your system. However, if Dummy is True then it will always return a DummyDmesg instance. """ if sys.platform.startswith('linux') and not_dummy: return LinuxDmesg() return DummyDmesg()