DiskSpace

Complexity

Total Complexity

10

Size/Duplication

Total Lines	75
Duplicated Lines	0 %

Importance

Changes	3
Bugs	1	Features	0

3 Methods

Rating	Name	Duplication	Size	Complexity
A	proc_mounts()	0	18	4
A	poll()	0	21	2
B	__init__()	0	23	4

# -*- coding: utf-8 -*-
"""Disk usage metrics reader from /proc on Linux."""
import re
import os
import sys
import plumd
from plumd.util import get_file_list

PY3 = sys.version_info > (3,)

__author__ = 'Kenny Freeman'
__email__ = 'kenny.freeman@gmail.com'
__license__ = "ISCL"
__docformat__ = 'reStructuredText'


class DiskSpace(plumd.Reader):
    """Plugin to measure disk usage."""

    # default config values
    defaults = {
        'poll.interval': 60,
        'fs_types': ['xfs', 'ext2', 'ext3', 'ext4'],
        'mnt_exclude_re': "^\/mnt\/.*",

A suspicious escape sequence \/ was found. Did you maybe forget to add an r prefix?

        'chrs_keep': 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-'
    }

    def __init__(self, log, config):
        """Plugin to measure disk usage.

        :param log: A logger
        :type log: logging.RootLogger
        :param config: a plumd.config.Conf configuration helper instance.
        :type config: plumd.config.Conf
        """
        super(DiskSpace, self).__init__(log, config)
        self.config.defaults(DiskSpace.defaults)
        # skip mounts for devices matching this re
        self.mnt_re = re.compile(config.get('mnt_exclude_re'))

        # characters to remove from device names
        self.filter_chars = "".join(char for char in map(chr, range(256))
                                    if char not in self.config.get('chrs_keep'))
        # args for str.translate() calls
        self.tlate_args = [None, self.filter_chars]
        if PY3:
            self.tlate_args = [dict.fromkeys(
                map(ord, self.filter_chars)), None]
        # get a list of mount points - only on init
        self.mounts = self.proc_mounts()

    def proc_mounts(self):
        """Return a list of mounted file systems.

        :rtype: list
        """
        fs_types = self.config.get('fs_types')
        ret = []
        mounts = get_file_list("/proc/mounts")
        mnt_re = self.mnt_re
        tlate_args = self.tlate_args
        for entry in mounts:
            dev, path, f_sys, _, _, _ = entry.split()
            if mnt_re.match(path) or f_sys not in fs_types:
                continue
            dev = re.sub("^/dev/", "", dev)
            dev = dev.translate(*tlate_args)

Usage of * or ** arguments should usually be done with care.

            ret.append((dev, path))
        return ret

    def poll(self):
        """Poll for disk space usage.

        :rtype: ResultSet
        """
        result = plumd.Result("diskusage")
        for dev, mpoint in self.mounts:
            mp_stat = os.statvfs(mpoint)
            free = (mp_stat.f_bavail * mp_stat.f_frsize)
            total = (mp_stat.f_blocks * mp_stat.f_frsize)
            used = (mp_stat.f_blocks - mp_stat.f_bfree) * mp_stat.f_frsize
            total_inodes = mp_stat.f_files
            free_inodes = mp_stat.f_favail
            used_inodes = total_inodes - free_inodes
            result.add(plumd.Int("{0}.free".format(dev), free))
            result.add(plumd.Int("{0}.total".format(dev), total))
            result.add(plumd.Int("{0}.used".format(dev), used))
            result.add(plumd.Int("{0}.total_inodes".format(dev), total_inodes))
            result.add(plumd.Int("{0}.free_inodes".format(dev), free_inodes))
            result.add(plumd.Int("{0}.used_inodes".format(dev), used_inodes))
        return plumd.ResultSet([result])