7
\$\begingroup\$

This class takes a log file name(which must be of a specific format/order) and returns a list of serial numbers of those drives which contain particular type of error. It uses regular expressions to process the file line by line and all but one of the regular expressions searches using patterns shaped by previous matches.

I have little experience in object orientated programming and would like to better understand how I might structure code to achieve the below task. In particular, I'm unsure about my methods calling each other in sequence as they do.

import re

class SerialNumberFinder:
    """Finds serial numbers of drives with a corrupt filesytem (XFS ERROR)"""

    def __init__(self, filename):
        self.serial_numbers = []
        self.result = None
        self.match = None
        try:
            with open(filename) as f:
                # reverse order of lines otherwise serial number information
                # for a drive is processed before error type is identified
                self.lines = reversed(f.readlines())
        except FileNotFoundError:
            print("Cannot find file {}".format(self.filename))


    def get_next_line(self):
        """Return next line from self.lines if lines, otherwise return None"""
        try:
            line = next(self.lines)
        except StopIteration:
            return None
        else:
            return line

    def get_serial_numbers(self):
        """Return a list of serial numbers"""
        self.find_error()
        return self.serial_numbers

    def find_error(self):
        """Use a regex to find mention of a specific type of error in a line"""
        while not self.match:
            line = self.get_next_line()
            if not line:
                return
            self.match = re.match("XFS ERROR (\[sd[a-z]\])", line)
        self.result = self.match.groups()[0]
        self.match = None
        self.find_bus()

    def find_bus(self):
        """Find mention of bus using identifier from find_error"""
        while not self.match:
            line = self.get_next_line()
            if not line:
                return
            self.match = re.match("(sd \S+) {}.*".format(re.escape
                                                         (self.result)), line)
        self.result = self.match.groups()[0]
        self.match = None
        self.find_serial()

    def find_serial(self):
        """Find a serial number using identifier from find_bus"""
        while not self.match:
            line = self.get_next_line()
            if not line:
                return
            self.match = re.match("{} \(SERIAL=([^)]*)\)".format(self.result),
                                  line)
        self.serial_numbers.append(self.match.groups()[0])
        self.match = None
        self.find_error()

if __name__ == "__main__":
    s = SerialNumberFinder("EXAMPLE_LOG.log")
    nums = s.get_serial_numbers()
    print(nums)

Output:

['WW11111', 'ZZ12345']

Example log file:

unrelated log messages
sd 0:0:0:0 Attached Disk Drive 
unrelated log messages
sd 0:0:0:0 (SERIAL=ZZ12345)
unrelated log messages
sd 0:0:0:0 [sda] Options
unrelated log messages
XFS ERROR [sda]
unrelated log messages
sd 2:0:0:1 Attached Disk Drive 
unrelated log messages
sd 2:0:0:1 (SERIAL=ZZ67890)
unrelated log messages
sd 2:0:0:1 [sdb] Options
unrelated log messages
sd 3:0:1:8 Attached Disk Drive 
unrelated log messages
sd 3:0:1:8 (SERIAL=WW11111)
unrelated log messages
sd 3:0:1:8 [sdc] Options
unrelated log messages
XFS ERROR [sdc]
unrelated log messages
\$\endgroup\$
1
\$\begingroup\$

This looks like a very complex class to me. All you really want to do is read all lines of an error log and keep a dictionary mapping hardware busses to drives, where each drive has a name (like sda), a bus and a serial. If you encounter an xfs error for one of these drives, return its serial number.

This can be accomplished by defining a basically empty Drive class and filling it successively. You don't even need the SerialNumberFinder class, one function is enough:

import collections


class Drive:
    """Object to hold all information about a drive."""
    pass


def get_serial_numbers(file_name):
    """Return a list of serial numbers of drives with XFS errors"""
    drives = collections.defaultdict(Drive)
    try:
        with open(file_name) as f:
            for line in f:
                bus = re.match(r"(sd \S+) (\[sd[a-z]\]).*", line)
                if bus:
                    bus_name, device_name = bus.groups()
                    drives[bus_name].name = device_name
                    continue

                serial = re.match(r"(sd \S+) \(SERIAL=([^)]*)\)", line)
                if serial:
                    bus_name, serial = serial.groups()
                    drives[bus_name].serial = serial
                    continue

                error_disk = re.match(r"XFS ERROR (\[sd[a-z]\])", line)
                if error_disk:
                    device_name, = error_disk.groups()
                    for drive in drives.values():
                        if drive.name == device_name:
                            yield drive.serial
    except FileNotFoundError:
        print("Cannot find file {}".format(file_name))
        raise

if __name__ == "__main__":
    nums = get_serial_numbers("EXAMPLE_LOG.log")
    print(list(nums))

Note that it is often recommended to compile your regexes, i.e do something like:

bus_re = re.compile(r"(sd \S+) (\[sd[a-z]\]).*")
...
bus = bus_re.match(line)

However, Python already does this for you in the background. So whenever a new regex is encountered, it is compiled and this compiled regex is used for all future appearances of this regex. The only time when you should do this is, if you have more than 100 regexes (the default cache size), if you want to take the performance hit of compiling it once way before you run the actual code or if you want to give a regex a nice understandable name.

| improve this answer | |
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.