DiskIo

Complexity

Total Complexity

3

Size/Duplication

Total Lines	25
Duplicated Lines	68 %

Importance

Changes	4
Bugs	2	Features	1

1 Method

Rating	Name	Duplication	Size	Complexity
A	poll()	17	17	3

# -*- coding: utf-8 -*-
"""Basic cpu, disk, etc reader using psutils - mainly used for testing."""

import psutil  # pip install psutil

The import psutil could not be resolved.

import plumd

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


class Cpu(plumd.Reader):
    """Plugin to measure cpu utilization using the psutil module."""

    defaults = {
        'poll.interval': 10,
        'percpu': False
    }

    def poll(self):
        """Poll for cpu utilization time metrics.

        :rtype: ResultSet
        """
        metrics = []
        ctimes = psutil.cpu_times()
        for m_name in ctimes._fields:
            val = getattr(ctimes, m_name)
            metrics.append(plumd.Float(m_name, val))
        return plumd.ResultSet([plumd.Result("cpu", metrics)])


class Mem(plumd.Reader):
    """Class to measure memory utilization using the psutil module."""

    defaults = {
        'poll.interval': 10
    }

    def poll(self):
        """Poll for mem memory utilization metrics.

        :rtype: ResultSet
        """
        metrics = []
        meminfo = psutil.virtual_memory()
        for minfo_f in meminfo._fields:
            val = getattr(meminfo, minfo_f)
            metrics.append(plumd.Float(minfo_f, val))
        return plumd.ResultSet([plumd.Result("mem", metrics)])


class Swap(plumd.Reader):
    """Class to measure swap utilization with the psutil module."""

    defaults = {
        'poll.interval': 10
    }

    def poll(self):
        """Poll for swap utilization metrics.

        :rtype: ResultSet
        """
        metrics = []
        swap = psutil.swap_memory()
        for swapf in swap._fields:
            val = getattr(swap, swapf)
            metrics.append(plumd.Float(swapf, val))
        return plumd.ResultSet([plumd.Result("swap", metrics)])


class Net(plumd.Reader):
    """Class to measure network interface metrics using the psutil module."""

    defaults = {
        'poll.interval': 10,
        'nics': None  # all
    }

    def poll(self):
        """Poll for net per nic bandwidth metrics.

        :rtype: ResultSet
        """
        nicios = psutil.net_io_counters(pernic=True)
        rset = plumd.ResultSet()
        for nic in nicios.keys():
            metrics = []
            nic_io = nicios[nic]
            for io_field in nic_io._fields:
                val = getattr(nic_io, io_field)
                metrics.append(plumd.Float(io_field, val))
            robj = plumd.Result("net", metrics)
            robj.meta.add(plumd.String("nic", nic.replace(" ", "_")))
            rset.add(robj)
        return rset


class Disk(plumd.Reader):
    """Class to measure disk utilization using the psutil module."""

    defaults = {
        'poll.interval': 300,
        'devices': None  # all
    }

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

        :rtype: ResultSet
        """
        rset = plumd.ResultSet()
        for disk in psutil.disk_partitions():
            metrics = []
            dev = disk.device.split("/")[-1]
            dev = dev.replace(":", "").replace("\\", "")
            try:
                disk_u = psutil.disk_usage(disk.mountpoint)
            except OSError:
                continue
            for disk_f in disk_u._fields:
                val = getattr(disk_u, disk_f)
                metrics.append(plumd.Float(disk_f, val))
            robj = plumd.Result("disk", metrics)
            robj.meta.add(plumd.String("dev", dev))
            rset.add(robj)
        return rset


class DiskIo(plumd.Reader):
    """Class to measure disk io."""

    defaults = {
        'poll.interval': 10,
        'devices': None  # all
    }

    def poll(self):
        """Poll for diskio metrics.

        :rtype: ResultSet
        """
        rset = plumd.ResultSet()
        diskio = psutil.disk_io_counters(perdisk=True)
        for dev in diskio.keys():
            metrics = []
            dio = diskio[dev]
            for dev_f in dio._fields:
                val = getattr(dio, dev_f)
                metrics.append(plumd.Float(dev_f, val))
            robj = plumd.Result("diskio", metrics)
            robj.meta.add(plumd.String("dev", dev))
            rset.add(robj)
        return rset