Cpu.poll() - Code Metrics - s4z/plumd - Measure and Improve Code Quality continuously with Scrutinizer

Cpu.poll() A
last analyzed 2017-05-25 01:53 UTC

↳ Parent: Cpu

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	4
Bugs	2	Features	1

Metric	Value
c	4
b	2
f	1
dl	0
loc	11
rs	9.4285
cc	2

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

import psutil  # pip install psutil
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3

import plumd

__author__ = 'Kenny Freeman'
__email__ = '[email protected]'
__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


1		# -- coding: utf-8 --
2		"""Basic cpu, disk, etc reader using psutils - mainly used for testing."""
3
4		import psutil # pip install psutil
		0 ignored issues – show Configuration introduced 2016-11-13 07:47 UTC by Report Bug Copy Issue Report The import `psutil` could not be resolved. This can be caused by one of the following: 1. Missing Dependencies This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands. # .scrutinizer.yml before_commands: - sudo pip install abc # Python2 - sudo pip3 install abc # Python3 Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version. 2. Missing __init__.py files This error could also result from missing `__init__.py` files in your module folders. Make sure that you place one file in each sub-folder. Loading history...
5
6		import plumd
7
8		__author__ = 'Kenny Freeman'
9		__email__ = '[email protected]'
10		__license__ = "ISCL"
11		__docformat__ = 'reStructuredText'
12
13
14		class Cpu(plumd.Reader):
15		"""Plugin to measure cpu utilization using the psutil module."""
16
17		defaults = {
18		'poll.interval': 10,
19		'percpu': False
20		}
21
22		def poll(self):
23		"""Poll for cpu utilization time metrics.
24
25		:rtype: ResultSet
26		"""
27		metrics = []
28		ctimes = psutil.cpu_times()
29		for m_name in ctimes._fields:
30		val = getattr(ctimes, m_name)
31		metrics.append(plumd.Float(m_name, val))
32		return plumd.ResultSet([plumd.Result("cpu", metrics)])
33
34
35		class Mem(plumd.Reader):
36		"""Class to measure memory utilization using the psutil module."""
37
38		defaults = {
39		'poll.interval': 10
40		}
41
42		def poll(self):
43		"""Poll for mem memory utilization metrics.
44
45		:rtype: ResultSet
46		"""
47		metrics = []
48		meminfo = psutil.virtual_memory()
49		for minfo_f in meminfo._fields:
50		val = getattr(meminfo, minfo_f)
51		metrics.append(plumd.Float(minfo_f, val))
52		return plumd.ResultSet([plumd.Result("mem", metrics)])
53
54
55		class Swap(plumd.Reader):
56		"""Class to measure swap utilization with the psutil module."""
57
58		defaults = {
59		'poll.interval': 10
60		}
61
62		def poll(self):
63		"""Poll for swap utilization metrics.
64
65		:rtype: ResultSet
66		"""
67		metrics = []
68		swap = psutil.swap_memory()
69		for swapf in swap._fields:
70		val = getattr(swap, swapf)
71		metrics.append(plumd.Float(swapf, val))
72		return plumd.ResultSet([plumd.Result("swap", metrics)])
73
74
75		class Net(plumd.Reader):
76		"""Class to measure network interface metrics using the psutil module."""
77
78		defaults = {
79		'poll.interval': 10,
80		'nics': None # all
81		}
82
83	View Code Duplication	def poll(self):
84		"""Poll for net per nic bandwidth metrics.
85
86		:rtype: ResultSet
87		"""
88		nicios = psutil.net_io_counters(pernic=True)
89		rset = plumd.ResultSet()
90		for nic in nicios.keys():
91		metrics = []
92		nic_io = nicios[nic]
93		for io_field in nic_io._fields:
94		val = getattr(nic_io, io_field)
95		metrics.append(plumd.Float(io_field, val))
96		robj = plumd.Result("net", metrics)
97		robj.meta.add(plumd.String("nic", nic.replace(" ", "_")))
98		rset.add(robj)
99		return rset
100
101
102		class Disk(plumd.Reader):
103		"""Class to measure disk utilization using the psutil module."""
104
105		defaults = {
106		'poll.interval': 300,
107		'devices': None # all
108		}
109
110		def poll(self):
111		"""Poll for disk space utilization metrics.
112
113		:rtype: ResultSet
114		"""
115		rset = plumd.ResultSet()
116		for disk in psutil.disk_partitions():
117		metrics = []
118		dev = disk.device.split("/")[-1]
119		dev = dev.replace(":", "").replace("\\", "")
120		try:
121		disk_u = psutil.disk_usage(disk.mountpoint)
122		except OSError:
123		continue
124		for disk_f in disk_u._fields:
125		val = getattr(disk_u, disk_f)
126		metrics.append(plumd.Float(disk_f, val))
127		robj = plumd.Result("disk", metrics)
128		robj.meta.add(plumd.String("dev", dev))
129		rset.add(robj)
130		return rset
131
132
133		class DiskIo(plumd.Reader):
134		"""Class to measure disk io."""
135
136		defaults = {
137		'poll.interval': 10,
138		'devices': None # all
139		}
140
141	View Code Duplication	def poll(self):
142		"""Poll for diskio metrics.
143
144		:rtype: ResultSet
145		"""
146		rset = plumd.ResultSet()
147		diskio = psutil.disk_io_counters(perdisk=True)
148		for dev in diskio.keys():
149		metrics = []
150		dio = diskio[dev]
151		for dev_f in dio._fields:
152		val = getattr(dio, dev_f)
153		metrics.append(plumd.Float(dev_f, val))
154		robj = plumd.Result("diskio", metrics)
155		robj.meta.add(plumd.String("dev", dev))
156		rset.add(robj)
157		return rset
158

s4z / plumd

Cpu.poll() A last analyzed 2017-05-25 01:53 UTC

Complexity

Size

Duplication

Importance

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

Cpu.poll() A
last analyzed 2017-05-25 01:53 UTC

2. Missing init.py files