Redis

Complexity

Total Complexity

38

Size/Duplication

Total Lines	269
Duplicated Lines	23.42 %

Importance

Changes	1
Bugs	1	Features	1

7 Methods

Rating	Name	Duplication	Size	Complexity
F	get_metrics()	10	31	9
B	record_keys()	0	18	7
B	connect()	0	29	4
B	poll()	32	32	6
B	__init__()	0	24	1
D	record_slave()	19	33	8
A	disconnect()	0	10	3

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

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

import sys

PY3 = sys.version_info > (3,)

import time
import socket

import plumd
import plumd.plugins
from plumd.calc import Differential


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

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

import re
import os
import sys

PY3 = sys.version_info > (3,)

import plumd
import plumd.plugins
from plumd.util import get_file_list


class Redis(plumd.plugins.Reader):
    """Plugin to record redis metrics."""

    INFO_CMD = "info\r\n"
    RECV_SIZE = 4096

    # default config values
    defaults = {
        'poll.interval': 10,
        'gauges': [
            "aof_current_rewrite_time_sec",
            "aof_enabled",
            "aof_last_rewrite_time_sec",
            "aof_rewrite_in_progress",
            "aof_rewrite_scheduled",
            "blocked_clients",
            "client_biggest_input_buf",
            "client_longest_output_list",
            "connected_clients",
            "connected_slaves",
            "evicted_keys",
            "expired_keys",
            "instantaneous_input_kbps",
            "instantaneous_ops_per_sec",
            "instantaneous_output_kbps",
            "keyspace_hits",
            "keyspace_misses",
            "latest_fork_usec",
            "loading",
            "master_repl_offset",
            "mem_fragmentation_ratio",
            "pubsub_channels",
            "pubsub_patterns",
            "rdb_bgsave_in_progress",
            "rdb_changes_since_last_save",
            "rdb_current_bgsave_time_sec",
            "rdb_last_bgsave_time_sec",
            "rdb_last_save_time",
            "rejected_connections",
            "repl_backlog_active",
            "repl_backlog_first_byte_offset",
            "repl_backlog_histlen",
            "repl_backlog_size",
            "sync_full",
            "sync_partial_err",
            "sync_partial_ok",
            "total_commands_processed",
            "total_connections_received",
            "total_net_input_bytes",
            "total_net_output_bytes",
            "uptime_in_days",
            "uptime_in_seconds",
            "used_cpu_sys",
            "used_cpu_sys_children",
            "used_cpu_user",
            "used_cpu_user_children",
            "used_memory",
            "used_memory_lua",
            "used_memory_peak",
            "used_memory_rss",
            "master_last_io_seconds_ago",
            "master_sync_in_progress",
            "slave_repl_offset",
            "slave_priority",
            "slave_read_only",
            "connected_slaves",
            "master_repl_offset",
            "repl_backlog_active",
            "repl_backlog_size",
            "repl_backlog_first_byte_offset",
            "repl_backlog_histlen"
            "connected_slaves"        ],
        'rates': [],
        'host': '127.0.0.1', # Redis server hostname/ip
        'port': 6379,        # Redis server port
        'timeout': 10,       # connection timeouts
        'retry_time': 30,    # pause n seconds between reconnects
        'retry_cnt': 3       # retry connection n times
    }

    def __init__(self, log, config):
        """Plugin to record redis metrics.

        :param log: A logger
        :type log: logging.RootLogger
        :param config: a plumd.config.Conf configuration helper instance.
        :type config: plumd.config.Conf
        """
        super(Redis, self).__init__(log, config)
        self.config.defaults(Redis.defaults)

        # metrics to record
        self.gauges = self.config.get('gauges')
        self.rates = self.config.get('rates')

        # Redis connection
        host = self.config.get('host')
        port = self.config.get('port')
        self.addr = (host, port)
        self.socket = None
        self.timeout = config.get('timeout')
        self.retry_time = config.get('retry_time')
        self.retry_cnt = config.get('retry_cnt')
        self.calc = Differential()


    def poll(self):

This code seems to be duplicated in your project.

        """Query Redis for metrics.

        :rtype: ResultSet
        """
        result = plumd.Result("redis")

        stats = self.get_metrics()
        ts = time.time()
        name = self.name

        # record gauges
        for stat in self.gauges:
            if stat in stats:
                mname = "{0}.{1}".format(name, stat)
                result.add(plumd.Float(mname, stats[stat]))

        # record rates
        for stat in self.rates:
            if stat in stats:
                mname = "{0}.{1}".format(name, stat)
                mval = self.calc.per_second(mname, float(stats[stat]), ts)
                result.add(plumd.Float(mname, mval))

        # replication
        if "slave0" in stats:
            self.record_slave(stats, result)

        # record db metrics
        self.record_keys(stats, result)

        return plumd.ResultSet([result])


    def record_keys(self, stats, result):
        #db0:keys=1,expires=0,avg_ttl=0
        name = self.name
        db_fmt = "db{0}"
        metric_fmt = "{0}.db.{1}.{2}"

        for i in xrange(0, len(stats.keys())):
            dbname = db_fmt.format(i)
            if dbname not in stats:
                break
            try:
                vals = stats[dbname].split(",")
                dbmetrics = dict((k,v) for k,v in (v.split('=') for v in vals))
                for k, v in dbmetrics.items():
                    metric_str = metric_fmt.format(name, i,k)
                    result.add(plumd.Int(metric_str, v))
            except KeyError as E:
                self.log.error("Redis: invalid db entry: {0}".format(dbname))


    def record_slave(self, stats, result):
        #slave0:ip=127.0.0.1,port=6399,state=online,offset=239,lag=1
        name = self.name
        slave_str = "slave{0}"
        moffstr = 'master_repl_offset'
        try:
            moffset = int(stats[moffstr])
        except(TypeError, KeyError) as e:
            self.log.error("Redis: invalid slave entry: {0}".format(sname))

        for i in xrange(0, len(stats.keys())):
            sname = slave_str.format(i)
            if sname not in stats:
                break
            try:

This code seems to be duplicated in your project.

                vals = stats[sname].split(",")
                smetrics = dict((k,v) for k,v in (v.split('=') for v in vals))
                sip = smetrics['ip'].replace(".", "_")
                smname = "{0}_{1}".format(sip, smetrics['port'])

                # record offset and lag
                mname = "{0}.slave.{1}.offset".format(name, sname)
                soffset = moffset - int(smetrics['offset'])
                result.add(plumd.Int(mname, soffset))
                mname = "{0}.slave.{1}.lag".format(name, sname)
                result.add(plumd.Int(mname, smetrics['lag']))

                # if slave is online set online = 1, otherwise 0
                sonline = 1 if smetrics['state'] == "online" else 0
                mname = "{0}.slave.{1}.online".format(name, sname)
                result.add(plumd.Int(mname, sonline))
            except(TypeError, KeyError, ValueError) as e:
                self.log.error("Redis: invalid slave entry: {0}".format(sname))


    def get_metrics(self):
        """Request and read stats from Redis socket."""
        stats = {}

        if not self.socket and not self.connect():
            return {}

        try:
            if PY3:
                self.socket.sendall(bytes(Redis.INFO_CMD, 'utf8'))
            else:
                self.socket.sendall(Redis.INFO_CMD)

            st_str = self.socket.recv(Redis.RECV_SIZE)

            for line in st_str.split("\n"):
                if line.startswith("#") or not line:
                    continue
                vals = line.split(":")
                if len(vals) != 2:
                    continue
                sname, sval = vals
                msg = "Redis: {0} = {1}"
                self.log.debug(msg.format(sname, sval))
                stats[sname] = sval
        except Exception as e:
            msg = "Redis: {0}: poll: exception: {1}"
            self.log.error(msg.format(self.addr, e))
            self.disconnect()

        return stats


    def connect(self):
        """Connect to Redis, returns True if sucessful, False otherwise.

        :rtype: bool
        """
        self.log.debug("Redis: connecting: {0}".format(self.addr))
        for i in xrange(0, self.retry_cnt):
            if self.socket:
                self.disconnect()
            try:
                # create the socket
                self.socket = socket.socket(socket.AF_INET,
                                            socket.SOCK_STREAM)
                # set timeout for socket operations
                self.socket.settimeout(self.timeout)
                # connect
                self.socket.connect(self.addr)
                # log and return
                msg = "Redis: connected: {0}"
                self.log.info(msg.format(self.addr))
                return True
            except Exception as e:
                # log exception, ensure cleanup is done (disconnect)
                msg = "Redis: {0}: connect: excception: {1}"
                self.log.error(msg.format(self.addr, e))
                # pause before reconnecting
                time.sleep(self.retry_time)
                self.disconnect()
        return False


    def disconnect(self):
        """Severe the Redis connection."""
        if self.socket:
            try:
                self.socket.close()
                self.socket = None
            except Exception as e:
                msg = "Redis: dicsonnect: exception {0}".format(e)
                self.log.error(msg)
                self.socket = None