Compares NFSv4 mounts defined in /etc/fstab and current mounts on the system

check-nfs-mounts.py

#!/usr/bin/env python
"""
Usage: check-nfs-mounts-v2.py

Compares NFSv4 mounts defined in /etc/fstab and current mounts on the system
"""
from __future__ import print_function

import re
try:
    import subprocess32 as subprocess
except ImportError:
    import subprocess
from collections import Counter
from copy import copy


def log(output_str, *args):
    """For debug logging"""
    if False:  # if True:
        print(output_str.format(*args))


def check_nfs_mounts(expected, current):
    """Compare the list of expected NFS mounts against current mounts"""

    # prepare to store error messages
    errors = []

    # quick check to see if number of mounts are correct
    log("Expecting to have {} mounts", len(expected))
    log("Found {} current mounts", len(current))
    if len(expected) != len(current):
        errors.append(Warning("Expected %d mounts, got %d!\n" %
                              (len(expected), len(current))))

    # quick check to find duplicate mounts
    counter = Counter(mount_point for server, path, mount_point in current)
    for mount_point, count in list(counter.items()):
        if count > 1:
            # expected mount is not found
            errors.append(Warning("Duplicate mount: %s!\n" % mount_point))

    # clone the list of current mounts line. we'll pop correctly matched
    # mounts from the buffer as they are found - any leftovers are presume
    # incorrect.
    buffer = copy(current)

    # compare the current mounts with the expected mounts in /etc/fstab
    for mount in expected:
        log("Checking: {}:{} on {}", mount[0], mount[1], mount[2])
        for current_mount in current:
            if current_mount[0] == mount[0] and \
                    current_mount[1] == mount[1] and \
                    current_mount[2] == mount[2]:
                log("Found.")
                buffer.remove(current_mount)
                break
        else:
            # expected mount is not found
            log("Couldn't find {}:{} on {}", mount[0], mount[1], mount[2])
            errors.append(Warning("Missing mount: %s:%s on %s!\n" %
                                  (mount[0], mount[1], mount[2])))

    # if `buffer` is not empty, some mounts are incorrectly mounted
    for current_server, current_path, current_mount in buffer:
        errors.append(Exception("Incorrect extra mounts: %s:%s on %s!\n" %
                                (current_server, current_path, current_mount)))

    return errors


def parse_fstab():
    """Parse /etc/fstab for a list of expected NFS mounts"""
    pattern = re.compile(
        r"(\w+\d+)\:(\/\w+) (\/nfs[\w\/]+\d+) nfs ([\w\d=,]+) ")

    try:
        with open("/etc/fstab") as fstab:
            return pattern.findall(fstab.read())
    except:
        # unable to read /etc/fstab for some reason! we should immediately
        # return at this point and avoid further processing.
        print("UNKNOWN - could not open /etc/fstab!")
        exit(3)


def parse_mounts():
    """Parse the output of `df` for a list of current NFS mounts"""

    try:
        stdout = subprocess.check_output(
            "df -t nfs4 -h -a | grep -E -v \"(primary|secondary)\" | sort",
            stderr=subprocess.STDOUT, shell=True, timeout=60)  # 60 seconds
        return re.findall(
            r"(\w+\d+)\:(\/\w+)\s+.*?(\/nfs[\w\/]+\d+)", stdout)
    except:
        # unable to run df for some reason! we should immediately return at
        # this point and avoid further processing.
        import traceback
        print("UNKNOWN - couldn't run NFS mount check!\n",
              traceback.format_exc())
        exit(3)


if __name__ == '__main__':
    EXPECTED = parse_fstab()
    CURRENT = parse_mounts()
    RESULTS = check_nfs_mounts(EXPECTED, CURRENT)

    # process and print returned results
    if not len(RESULTS):
        return_code = 0
        print("OK - ", end='')
    elif any(type(result) == Exception for result in RESULTS):
        return_code = 2
        print("CRITICAL - NFS mount check failure: ", end='')
    else:
        return_code = 1
        print("WARNING - NFS mount check warning: ", end='')
    print("%s/%s\n" % (len(CURRENT), len(EXPECTED)))

    for result in RESULTS:
        print(type(result) == Exception and "E: " or "W: ",
              result.message.strip())

    exit(return_code)