Stat

Complexity

Total Complexity

14

Size/Duplication

Total Lines	97
Duplicated Lines	16.49 %

Importance

Changes	2
Bugs	0	Features	0

3 Methods

Rating	Name	Duplication	Size	Complexity
A	poll()	0	6	1
F	check()	0	42	9
A	proc_stat_cpu()	0	17	3

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

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

import time
from collections import deque

import plumd
import plumd.plugins
from plumd.calc import Differential
from plumd.util import get_file_map


## todo: switch from list with pop(0) to deque
class Stat(plumd.plugins.Reader):
    """Class to read metrics from /proc/stat"""
    defaults = {
        'proc_stat_gauges': ['procs_running', 'procs_blocked'],
        'proc_stat_rates': ['intr', 'ctxt', 'softirq'],
        'cpu_metrics': ["user", "nice", "system", "idle", "iowait", "irq",
                        "softirq","steal", "guest", "guest_nice"],
        'per_cpu': False,
    }

    def __init__(self, log, config):

This code seems to be duplicated in your project.

        """Plugin to measure various kernel metrics from /proc/stat

        :param log: A logger
        :type log: logging.RootLogger
        :param config: a plumd.config.Conf configuration helper instance.
        :type config: plumd.config.Conf
        """
        super(Stat, self).__init__(log, config)
        self.config.defaults(Stat.defaults)
        self.calc = Differential()
        self.proc_file = "{0}/stat".format(config.get('proc_path'))
        self.per_cpu = self.config.get('per_cpu')
        self.cpu_metrics = self.config.get('cpu_metrics')
        self.gauges = self.config.get('proc_stat_gauges')
        self.rates = self.config.get('proc_stat_rates')


    def poll(self):
        """Return cpu utilization and process metrics from proc file stat.

        :rtype: plumd.ResultSet
        """
        return plumd.ResultSet(self.check())


    def check(self):
        """Return cpu utilization and process metrics from proc file stat.

        :rtype: collections.deque
        """
        results = deque()
        result = plumd.Result("stat")

        dat = get_file_map(self.proc_file, 0, 0)
        ts = time.time()

        # record gauges
        for i, metric in enumerate(self.gauges):
            if metric not in dat:
                self.log.warn("stat: unknown metric {0}".format(metric))
                del(self.gauges[i])
                continue
            result.add(plumd.Int(metric, dat[metric][0]))

        # record rates
        for i, metric in enumerate(self.rates):
            if metric not in dat:
                self.log.warn("stat: unknown metric {0}".format(metric))
                del(self.rates[i])
                continue
            mval = self.calc.per_second(metric, float(dat[metric][0]), ts)
            result.add(plumd.Int(metric, mval))

        # record cpu
        if "cpu" in dat:
            results.append(self.proc_stat_cpu("cpu", "cpu", dat["cpu"]))

        # record each cpu if configured
        if self.per_cpu:
            for i in xrange(0, len(dat)):
                mstr = "cpu{0}".format(i)
                if mstr not in dat:
                    break
                results.append(self.proc_stat_cpu("cpus", mstr, dat[mstr]))

        results.append(result)
        return results


    def proc_stat_cpu(self, rname, key, val):
        """Return cpu utilization metrics in percentage.

        :param rname: The Result name (eg. cpu or cpus)
        :type rname: str
        :param val: A deque populated with the metric values from stat
        :type val: deque
        :rtype: list
        """
        result = plumd.Result(rname)
        cpu = self.cpu_metrics
        for map_val in cpu:
            if not val:
                break
            mstr = "{0}_{1}".format(key, map_val)
            result.add(plumd.Float(mstr, val.popleft()))
        return result