check_linux_interface

#!/usr/bin/env python
# SnapServ/monitoring-plugins - Simple and reliable Nagios / Icinga plugins
# Copyright (C) 2016 - 2017 SnapServ Mathis
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

import logging

import nagiosplugin
from shared.common import NagiosPlugin, ExceptionContext, OptionalExactMatchContext

LOGGER = logging.getLogger('nagiosplugin')


class Interface(nagiosplugin.Resource):
    SYSFS_PATH_OPERSTATE = '/sys/class/net/%s/operstate'
    SYSFS_PATH_SPEED = '/sys/class/net/%s/speed'
    SYSFS_PATH_DUPLEX = '/sys/class/net/%s/duplex'

    def __init__(self, name):
        self._name = name

    def probe(self):
        operstate = self.get_operstate()
        if operstate:
            yield nagiosplugin.Metric('state', operstate)
            yield nagiosplugin.Metric('speed', self.get_speed())
            yield nagiosplugin.Metric('duplex', self.get_duplex())
        else:
            yield nagiosplugin.Metric('exception', 'Interface does not exist: %s' % self._name)

    def get_operstate(self):
        try:
            with open(self.SYSFS_PATH_OPERSTATE % self._name) as file:
                return file.read().strip().upper()
        except EnvironmentError:
            return None

    def get_speed(self):
        try:
            with open(self.SYSFS_PATH_SPEED % self._name) as file:
                return int(file.read().strip())
        except EnvironmentError:
            return None

    def get_duplex(self):
        try:
            with open(self.SYSFS_PATH_DUPLEX % self._name) as file:
                return file.read().strip().lower()
        except EnvironmentError:
            return None


class InterfaceStateContext(nagiosplugin.Context):
    def __init__(self, name):
        super(InterfaceStateContext, self).__init__(name, fmt_metric=self.fmt_interface_state)

    @staticmethod
    def fmt_interface_state(metric, _):
        return 'State is %s' % metric.value

    def evaluate(self, metric, resource):
        interface_state = str(metric.value)
        if interface_state == 'UP':
            return self.result_cls(nagiosplugin.Ok, None, metric)
        else:
            return self.result_cls(nagiosplugin.Critical, None, metric)


class InterfaceSummary(nagiosplugin.Summary):
    def ok(self, results):
        fmt = 'State:%s Speed:%s Duplex:%s'
        speed = results['speed'].metric.value or 'N/A'
        duplex = results['duplex'].metric.value or 'N/A'
        return fmt % (results['state'].metric.value, speed, duplex)

    def problem(self, results):
        return super().problem(results)


class InterfacePlugin(NagiosPlugin):
    def declare_arguments(self) -> None:
        self.argument_parser.add_argument(
            '-n', '--name', required=True,
            help='Specifies the name of the network interface which should be checked.'
        )
        self.argument_parser.add_argument(
            '-s', '--speed', metavar='SPEED', type=int, default=None,
            help='Return WARNING if interface speed in Mbps does not match SPEED.'
        )
        self.argument_parser.add_argument(
            '-d', '--duplex', metavar='DUPLEX', default=None,
            help='Return WARNING if interface duplex does not match DUPLEX.'
        )
        self.exclude_from_kwargs += ('speed', 'duplex')

    def instantiate_check(self) -> nagiosplugin.Check:
        return nagiosplugin.Check(
            Interface(**self.keyword_arguments),

            InterfaceStateContext(
                name='state'
            ),

            OptionalExactMatchContext(
                name='speed',
                reference=self.arguments.get('speed'),
                format_string='Speed is %s'
            ),

            OptionalExactMatchContext(
                name='duplex',
                reference=self.arguments.get('duplex'),
                format_string='Duplex is %s'
            ),

            ExceptionContext(),

            InterfaceSummary()
        )


if __name__ == '__main__':
    InterfacePlugin().execute()