plumd.plugins.readers.Net

Complexity

Total Complexity

3

Size/Duplication

Total Lines	24
Duplicated Lines	0 %

1 Method

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

# -*- coding: utf-8 -*-

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


import psutil # pip install psutil

import plumd
import plumd.plugins

# note: these plugins are currently used for demonstration and testing purposes.


class Cpu(plumd.plugins.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.plugins.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 m in meminfo._fields:
            val = getattr(meminfo, m)
            metrics.append(plumd.Float(m, val))
        return plumd.ResultSet([plumd.Result("mem", metrics)])


class Swap(plumd.plugins.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 = []
        s = psutil.swap_memory()
        for m in s._fields:
            val =getattr(s, m)
            metrics.append(plumd.Float(m, val))
        return plumd.ResultSet([plumd.Result("swap", metrics)])


class Net(plumd.plugins.Reader):

This code seems to be duplicated in your project.

    """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 = []
            io = nicios[nic]
            for f in io._fields:
                val = getattr(io, f)
                metrics.append(plumd.Float(f, val))
            robj = plumd.Result("net", metrics)
            robj.meta.add(plumd.String("nic", nic.replace(" ", "_")))
            rset.add(robj)
        return rset


class Disk(plumd.plugins.Reader):

This code seems to be duplicated in your project.

    """
    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:
                u = psutil.disk_usage(disk.mountpoint)
            except OSError:
                continue
            for f in u._fields:
                val = getattr(u, f)
                metrics.append(plumd.Float(f, val))
            robj = plumd.Result("disk", metrics)
            robj.meta.add(plumd.String("dev", dev))
            rset.add(robj)
        return rset


class DiskIo(plumd.plugins.Reader):

This code seems to be duplicated in your project.

    """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 = []
            io = diskio[dev]
            for f in io._fields:
                val = getattr(io, f)
                metrics.append(plumd.Float(f, val))
            robj = plumd.Result("diskio", metrics)
            robj.meta.add(plumd.String("dev", dev))
            rset.add(robj)
        return rset