RedisNet.read() - Code Metrics - Inspection of "pep8/flake8 fixes" - s4z/plumd - Measure and Improve Code Quality continuously with Scrutinizer

Completed
Push — master ( 08acac...d85769 )

by Kenny
created 2016-11-13 07:46 UTC
RedisNet.read() A

↳ Parent: RedisNet
Complexity

Conditions
Size

Total Lines
Duplication

Lines	0
Ratio	0 %
Importance

Changes	1
Bugs	0	Features	1
Metric	Value
c	1
b	0
f	1
dl	0
loc	21
rs	9.0534
cc	4
# -*- coding: utf-8 -*-
"""A redis reader plugin with builtin redis client."""
import sys
import time
import socket
from collections import deque
from collections import defaultdict

import plumd
from plumd.util import Differential

__author__ = 'Kenny Freeman'
__email__ = '[email protected]'
__license__ = "ISCL"
__docformat__ = 'reStructuredText'

PY3 = sys.version_info > (3,)


class RedisError(Exception):
    """A generic Redis error."""

    pass


class RedisClient(object):
    """A minimal Redis client."""

    def __init__(self, log, addr, sfamily, timeout):
        """A minimal Redis client.

        :param log: A logger created from loggging.getLogger
        :type log: logging.RootLogger
        :param addr: Either a tuple of ('host', port) or a path to a unix socket
        :type addr: str or tuple(str, int)
        :param sfamily: The socket family eg. socket.AF_INET or AF_UNIX
        :type sfamily: int
        :param timeout: The timeout in seconds for all socket operations
        :type timeout: float or int
        """
        self.log = log
        self.net = RedisNet(log, addr, sfamily, timeout)

    def info(self, section=None):
        """Return redis info.

        :param section: The info section to request from Redis
        :type section: str
        :raises RedisError: for any socket related exceptions
        :raises RedisError: for unexpcted server responses
        :rtype: dict
        """
        ret = {}
        section = "all" if section is None else section
        self.net.send("INFO {0}\r\n".format(section))
        for info_str in RedisResponse(self.net):
            for line in info_str.split("\n"):
                if not line or line[0] == "#" or line == '\r':
                    continue
                if line.find(":") >= 0:
                    key, val = line.split(":")
                ret[key] = val
        return ret

    def config_get_multi(self, globs):
        """Return redis config.

        :param globs: An containing glob search strings
        :type globs: iterable
        :raises RedisError: for any socket related exceptions
        :raises RedisError: for unexpcted server responses
        :rtype: dict
        """
        for glob_str in globs:
            self.net.send("CONFIG GET {0}\r\n".format(glob_str))
            vals = [val.strip() for val in RedisResponse(self.net)]
            yield dict(zip(vals[0::2], vals[1::2]))

    def scan(self, prefix, count=None):
        """Return a deque of key names that match the requested prefix.

        :param prefix: The key prefix/glob to search for
        :type prefix: str
        :param count: The number of keys to request on each iteration
        :type count: int
        :raises RedisError: for any socket related exceptions
        :raises RedisError: for unexpcted server responses
        :rtype: deque
        """
        scan_cmd = "scan {0} match {1} count {2}\r\n"
        count = 10 if count is None else count
        cursor = 0
        while True:
            # send scan request
            self.net.send(scan_cmd.format(cursor, prefix, count))
            # response is the next cursor followed by a list of matches
            resp = RedisResponse(self.net)
            cursor = int(next(resp))
            for key in resp:
                yield key.strip()
            if cursor == 0:
                break

    def llen_multi(self, keys):
        """Return the total length of each key provided.

        :param keys: The iterable of keys to return the total length of.
        :type keys: iterable
        :raises RedisError: for any socket related exceptions
        :raises RedisError: for unexpcted server responses
        :rtype: int
        """
        llen_cmd = "llen {0}\r\n"
        total = 0
        for key in keys:
            # send scan request
            self.net.send(llen_cmd.format(key))
            # response should just be an int
            resp = RedisResponse(self.net)
            total += int(next(resp))
        return total

    def zcard_multi(self, keys):
        """Return the total cardinality of each key provided.

        :param keys: The iterable of keys to return the total cardinality of.
        :type keys: iterable
        :raises RedisError: for any socket related exceptions
        :raises RedisError: for unexpcted server responses
        :rtype: int
        """
        zcard_cmd = "zcard {0}\r\n"
        total = 0
        for key in keys:
            # send scan request
            self.net.send(zcard_cmd.format(key))
            # response should just be an int
            resp = RedisResponse(self.net)
            total += int(next(resp))
        return total

    def scard_multi(self, keys):
        """Return the total cardinality of each key provided.

        :param keys: The iterable of keys to return the total cardinality of.
        :type keys: iterable
        :raises RedisError: for any socket related exceptions
        :raises RedisError: for unexpcted server responses
        :rtype: int
        """
        scard_cmd = "scard {0}\r\n"
        total = 0
        for key in keys:
            # send scan request
            self.net.send(scard_cmd.format(key))
            # response should just be an int
            resp = RedisResponse(self.net)
            total += int(next(resp))
        return total

    def pfcount_multi(self, keys):
        """Return the total cardinality of each key provided.

        :param keys: The iterable of keys to return the total cardinality of.
        :type keys: iterable
        :raises RedisError: for any socket related exceptions
        :raises RedisError: for unexpcted server responses
        :rtype: int
        """
        pfcount_cmd = "pfcount {0}\r\n"
        total = 0
        for key in keys:
            # send scan request
            self.net.send(pfcount_cmd.format(key))
            # response should just be an int
            resp = RedisResponse(self.net)
            total += int(next(resp))
        return total


class RedisResponse(object):
    """An iterable of redis command responses."""

    def __init__(self, reader):
        """An iterable of redis command responses.

        :param reader: A RedisNet reader instance
        :type reader: RedisNet

        :raises RedisError: for any socket related errors
        :raises RedisError: for any unknown responses
        :raises RedisError: for any Redis Errors returned
        :raises RedisError: for any ValueErrors encountered when casting
        """
        self.reader = reader
        # handlers consume responses and add them to self.vals
        self.func = defaultdict(lambda: RedisResponse.h_unknown)
        self.func["*"] = lambda buff: self.parse(int(buff))
        self.func["+"] = lambda buff: self.vals.append(str(buff))
        # remove the \r from the string
        self.func["$"] = lambda buff: \
            self.vals.append(self.reader.readnbytes(int(buff) + 2))
        self.func[":"] = lambda buff: self.vals.append(int(buff))
        self.func["-"] = RedisResponse.h_error
        self.vals = deque()
        self.parse()

    def parse(self, nitems=None):
        """Read the full response from self.sock.

        :raises RedisError: for any socket related Exceptions
        :raises RedisError: for any unknown types read
        :raises RedisError: for any redis protocol errors
        :rtype: varies
        """
        nitems = 1 if nitems is None else nitems
        for i in xrange(nitems):

            try:
                buff = self.reader.readline()
                self.func[buff[0]](buff[1:])
            except (ValueError, IndexError) as exc:
                msg = "could not parse response: {0}: {1}"
                raise RedisError(msg.format(buff, exc))

    @staticmethod
    def h_unknown(buff):
        """Uknown response handler.

        :param buff: A response buffer read from Redis
        :type buff: str
        :raises RedisError: this function always raises a RedisError
        """
        raise RedisError("unknown command: {0}".format(buff))

    @staticmethod
    def h_error(buff):
        """Raise a RedisError with unknown command buff.

        :param buff: A response buffer read from Redis
        :type buff: str
        :raises RedisError: on any socket related exceptions
        :rtype: str
        """
        msg = "RedisResponse: h_error({0})"
        raise RedisError(msg.format(buff))

    def __iter__(self):
        """A Redis command response iterator.

        :rtype: iterator
        """
        return self

    def __next__(self):
        """Return the next response, if any.

        :rtype: object
        """
        if not self.vals:
            raise StopIteration()
        return self.vals.popleft()

    def next(self):
        """Return the next response, if any.

        :rtype: object
        """
        if not self.vals:
            raise StopIteration()
        return self.vals.popleft()


class RedisNet(object):
    """A helper class that talks to Redis on a unix/tcp socket."""

    BUFF_LEN = 8192

    def __init__(self, log, addr, sfamily, timeout):
        """A helper class that talks to Redis on a unix/tcp socket.

        :param log: A logger created from loggging.getLogger
        :type log: logging.RootLogger
        :param addr: Either a tuple of ('host', port) or a path to a unix socket
        :type addr: str or tuple(str, int)
        :param sfamily: The socket family eg. socket.AF_INET or AF_UNIX
        :type sfamily: int
        :param timeout: The timeout in seconds for all socket operations
        :type timeout: float or int
        """
        self.log = log
        # addr can be unix socket or (host, port) tuple
        self.addr = addr
        # socket.AF_INET or socket.AF_UNIX
        self.sfamily = sfamily
        # all socket operations timeout
        self.timeout = timeout
        self.sock = None
        # read from our socket into this buffer
        # keep an index in the buffer that we've read up to
        # and record the total number of bytes in the buffer
        self.buff = ""
        self.buff_end = -1
        self.buff_i = -1

    def connect(self):
        """Connect to Redis.

        :raises RedisError: for any socket related exceptions
        :rtype: Exception or None
        """
        if self.sock:
            self.disconnect()
        try:
            # create the socket
            self.sock = socket.socket(self.sfamily, socket.SOCK_STREAM)
            # set timeout for socket operations
            self.sock.settimeout(self.timeout)
            self.sock.connect(self.addr)
            msg = "RedisNet: connected: {0}"
            self.log.info(msg.format(self.addr))
        except Exception as exc:
            msg = "RedisNet: Exception during connect: {0}"
            self.log.error(msg.format(exc))
            raise RedisError(msg.format(exc))
        return True

    def disconnect(self):
        """Disconnect from Redis.

        :raises RedisError: for any socket related exceptions
        """
        self.log.debug("RedisNet: disconnect")
        if self.sock:
            try:
                self.sock.close()
                self.sock = None
            except Exception as exc:
                msg = "RedisNet: exception during disconnect: {0}"
                self.log.error(msg.format(exc))
                raise RedisError(msg.format(exc))

    def read(self):
        """Read RedisNet.BUFF_LEN bytes from our socket into self.buff.

        Calls here overwrite self.buff and reset self.buff_i and
        self.buff_end.

        :raises RedisError: for any socket related exceptions
        """
        if not self.sock and not self.connect():
            msg = "RedisNet: unable to connect to: {0}"
            raise RedisError(msg.format(self.addr))

        try:
            self.buff = self.sock.recv(RedisNet.BUFF_LEN)
            self.buff_end = len(self.buff)
            self.buff_i = 0
        except Exception as exc:
            msg = "RedisNet: Exception during readline: {0}"
            self.log.error(msg.format(exc))
            self.disconnect()
            raise RedisError(msg.format(exc))

    def recv(self, nbytes):
        """Read nbytes from our socket and return it.

        :param nbytes: The number of bytes to read
        :type nbytes: int
        :raises RedisError: for any socket related exceptions
        :rytpe: str
        """
        if not self.sock and not self.connect():
            msg = "RedisNet: unable to connect to: {0}"
            raise RedisError(msg.format(self.addr))

        ret = ""
        try:
            ret = self.sock.recv(nbytes)
        except Exception as exc:
            msg = "RedisNet: Exception during recv: {0}"
            self.log.error(msg.format(exc))
            self.disconnect()
            raise RedisError(msg.format(exc))
        return ret

    def readline(self):
        """Get the next available line.

        :raises RedisError: for any socket related exceptions
        :rytpe: str
        """
        if not self.sock and not self.connect():
            msg = "RedisNet: unable to connect to: {0}"
            raise RedisError(msg.format(self.addr))

        buffs = deque()
        while True:
            # do we have any data available?
            if self.buff_end < 0 or self.buff_i >= self.buff_end:
                # read data, reset buffer state
                while self.buff_end < 1:
                    self.read()
            # now we have data, do we have a newline?
            i = self.buff[self.buff_i:].find("\n")
            if i > -1:
                # return line, advance buffer past it
                # move i past the newline
                # also need to find
                buff_i = self.buff_i
                buffs.append(self.buff[buff_i:buff_i+i])
                # advance beyond i
                self.buff_i = buff_i + i + 1
                # reset if we have no buffer left
                if self.buff_i >= self.buff_end:
                    self.buff_i = -1
                    self.buff_end = -1
                break
            # no newline yet, record and keep reading
            buffs.append(self.buff[self.buff_i:])
            self.buff_end = -1
            self.buff_i = -1
        ret = "".join(buffs)
        return ret

    def readnbytes(self, nbytes):
        """Read nbytes from our socket.

        :param nbytes: The number of bytes to read
        :type nbytes: int
        :raises RedisError: for any socket related exceptions
        :rytpe: str
        """

        # any bytes in our buffer?
        ret = ""
        buffs = deque()
        if self.buff_end and self.buff_i < self.buff_end:
            # do we have enough buffer to fullfill the request?
            nbytes_left = self.buff_end - self.buff_i
            if nbytes_left >= nbytes:
                # yes, advance our pointer
                buffi = self.buff_i
                buffs.append(self.buff[buffi:buffi+nbytes])
                self.buff_i += nbytes
                nbytes = 0
            else:
                # no, consume all of the buffer and then get remaining
                buffs.append(self.buff[self.buff_i:])
                # reset so next access on buffer forces a read
                self.buff_i = -1
                self.buff_end = -1
                nbytes -= nbytes_left
        # do we need more bytes?
        if nbytes:
            # just do a recv - don't use our buffer
            buffs.append(self.recv(nbytes))

        # join the buffers
        ret = "".join(buffs)
        return ret

    def send(self, cmd):
        """Send the supplied string to the redis server.

        :param cmd: The string to send to the redis server
        :type cmd: str
        :raises RedisError: for any socket related exceptions
        """
        if not self.sock and not self.connect():
            msg = "RedisNet: unable to connect to: {0}"
            raise RedisError(msg.format(self.addr))
        # send info request
        try:
            self.sock.sendall(cmd)
        except Exception as exc:
            msg = "RedisNet: exception sending to server: {0}"
            self.log.error(msg.format(exc))
            self.disconnect()
            raise RedisError(msg.format(exc))


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

    # 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': [],
        'configs': [
            'maxmemory'
        ],
        'keys': {
            # 'type': { metric_prefix: [key_prefix*, ...] }
            'lists': {},
            'zsets': {},
            'sets': {},
            'hlls': {}
        },
        'addr': '127.0.0.1:6379',
        'addr_type': 'inet',
        'timeout': 10
    }

    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')
        self.configs = self.config.get('configs')
        self.keys = self.config.get('keys')

        # Redis connection - either unix socket or tcp
        addr = self.config.get('addr')
        addr_type = self.config.get('addr_type').lower()
        if addr_type == "unix":
            sfamily = socket.AF_UNIX
        elif addr_type == "inet":
            try:
                host, port = addr.split(":")
            except AttributeError:
                msg = "Redis: invalid address: {0}, (host:port)"
                raise plumd.ConfigError(msg.format(addr))
            addr = (host, int(port))
            sfamily = socket.AF_INET
        else:
            msg = "Redis: unsupported connection type: {0} (unix, inet)"
            raise plumd.ConfigError(msg.format(addr_type))
        timeout = config.get('timeout')
        self.client = RedisClient(self.log, addr, sfamily, timeout)
        self.calc = Differential()

    def poll(self):
        """Query Redis for metrics.

        :rtype: ResultSet
        """
        # catch exceptions - simply skip the poll on error
        try:
            result = plumd.Result("redis")

            # config values
            self.record_configs(result)

            # key sizes
            self.record_sizes(result)

            # get server metrics
            stats = self.client.info()

            # record gauges, rates
            self.record_metrics(stats, result)

            # replication, if any slaves are connected
            if "slave0" in stats:
                self.record_slaves(stats, result)

            # db metrics, maxmem
            self.record_dbs(stats, result)

            # record lists, zsets, sets and hll sizes
            self.record_sizes(result)

            # and finally command stats - if available
            self.record_cmdstats(result)

        except RedisError as exc:
            msg = "Redis: exception during poll: {0}"
            self.log.error(msg.format(exc))
        return plumd.ResultSet([result])

    def record_cmdstats(self, result):
        """Record the stats from info commandstats.

        :param result: A result object to add metrics to
        :type result: ResultSet
        """
        name = self.name
        infos = self.client.info("commandstats")
        for key in sorted(infos.keys()):
            vals = infos[key].split(",")
            cstat, cname = key.split("_")
            for val in vals:
                mname, mval = val.split("=")
                metric = "{0}.{1}.{2}.{3}".format(name, cstat, cname, mname)
                result.add(plumd.Float(metric, mval))

    def record_metrics(self, stats, result):
        """Record the configured gauges and metrics.

        :param stats: Dictionary returned from info command
        :type stats: dict
        :param result: A result object to add metrics to
        :type result: ResultSet
        """
        timest = 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]), timest)
                result.add(plumd.Float(mname, mval))

    def record_dbs(self, stats, result):
        """Record per database metrics into result.

        :param stats: Dictionary returned from info command
        :type stats: dict
        :param result: A result object to add metrics to
        :type result: ResultSet
        """
        # 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 key, val in dbmetrics.items():
                    metric_str = metric_fmt.format(name, i, key)
                    result.add(plumd.Int(metric_str, val))
            except KeyError:
                self.log.error("Redis: invalid db entry: {0}".format(dbname))

    def record_slaves(self, stats, result):
        """Record slave metrics into result.

        :param stats: A dictionary returned from info command
        :type stats: dict
        :param result: A ResultSet object to add metrics to
        :type result: ResultSet
        """
        # 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'
        moffset = 0
        try:
            moffset = int(stats[moffstr])
        except(TypeError, KeyError):
            self.log.error("Redis: no {0} value".format(moffstr))

        # for each slave entry
        for i in xrange(0, len(stats.keys())):

            sname = slave_str.format(i)
            if sname not in stats:
                break
            try:
                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, smname)
                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):
                self.log.error("Redis: invalid slave entry: {0}".format(sname))

    def record_configs(self, result):
        """Record the configured configuration values.

        :param result: A ResultSet to record max mem to.
        :type result: plumd.ResultSet
        """
        configs = self.configs
        if not configs:
            return
        name = self.name
        for config in self.client.config_get_multi(configs):
            for key, val in config.items():
                mstr = "{0}.configs.{1}".format(name, key)
                result.add(plumd.Float(mstr, val))

    def record_sizes(self, result):
        """Record the total sizes of the configured keys.

        For each type of key (list, zset, set, hyperloglog)
        scan for a list of keys matching the prefix and record the
        total number of items for all keys matching that prefix.

        :param result: A ResultSet to record into.
        :type result: plumd.ResultSet
        """
        if not self.keys:
            return
        keys = self.config.get("keys")
        if "lists" in keys:
            self.record_lists(keys['lists'], result)
        if "zsets" in keys:
            self.record_zsets(keys['zsets'], result)
        if "sets" in keys:
            self.record_sets(keys['sets'], result)
        if "hlls" in keys:
            self.record_hlls(keys['hlls'], result)

    def record_lists(self, lconfig, result):
        """Record the total length of the configured lists.

        eg. lconfig: {"metric_name": [ "list*", "of*", "globs*"]}

        :param lconfig: A dict of metric name => globs
        :type lconfig: dict
        :param result: A ResultSet to record into.
        :type result: plumd.ResultSet
        """
        name = self.name
        for mprefix, kprefixes in lconfig.items():
            for prefix in kprefixes:
                # get the total for this prefix
                total = self.client.llen_multi(self.client.scan(prefix))
                mstr = "{0}.sizes.lists.{1}".format(name, mprefix)
                result.add(plumd.Int(mstr, total))

    def record_zsets(self, zconfig, result):
        """Record the total length of the configured zsets.

        eg. zconfig: {"metric_name": [ "list*", "of*", "globs*"]}

        :param zconfig: A dict of metric name => globs
        :type zconfig: dict
        :param result: A ResultSet to record into.
        :type result: plumd.ResultSet
        """
        name = self.name
        for mprefix, kprefixes in zconfig.items():
            for prefix in kprefixes:
                # get the total for this prefix
                total = self.client.zcard_multi(self.client.scan(prefix))
                mstr = "{0}.sizes.zset.{1}".format(name, mprefix)
                result.add(plumd.Int(mstr, total))

    def record_sets(self, sconfig, result):
        """Record the total length of the configured sets.

        eg. sconfig: {"metric_name": [ "list*", "of*", "globs*"]}

        :param sconfig: A dict of metric name => globs
        :type sconfig: dict
        :param result: A ResultSet to record into.
        :type result: plumd.ResultSet
        """
        name = self.name
        for mprefix, kprefixes in sconfig.items():
            for prefix in kprefixes:
                # get the total for this prefix
                total = self.client.scard_multi(self.client.scan(prefix))
                mstr = "{0}.sizes.set.{1}".format(name, mprefix)
                result.add(plumd.Int(mstr, total))

    def record_hlls(self, hllconfig, result):
        """Record the total length of the configured hlls.

        eg. sconfig: {"metric_name": [ "list*", "of*", "globs*"]}

        :param hllconfig: A dict of metric name => globs
        :type hllconfig: dict
        :param result: A ResultSet to record into.
        :type result: plumd.ResultSet
        """
        name = self.name
        for mprefix, kprefixes in hllconfig.items():
            for prefix in kprefixes:
                # get the total for this prefix
                total = self.client.pfcount_multi(self.client.scan(prefix))
                mstr = "{0}.sizes.hll.{1}".format(name, mprefix)
                result.add(plumd.Int(mstr, total))

s4z / plumd

Push — master ( 08acac...d85769 )

RedisNet.read() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like
