CouchBase.poll() - Code Metrics - Inspection of "added couchbase reader" - s4z/plumd - Measure and Improve Code Quality continuously with Scrutinizer

Completed
Push — master ( d84ba0...9b96e9 )

by Kenny
created 2017-03-16 13:32 UTC
CouchBase.poll() A

↳ Parent: CouchBase
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	11
rs	9.4285
cc	1
# -*- coding: utf-8 -*-
"""Couchbase metrics reader."""
import sys
import time
import json
import urllib2
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3
from collections import deque
from threading import Lock

from plumd import Counter, Gauge, String, Reader, ResultSet, Result
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3

PY3 = sys.version_info > (3,)

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


def get_sample_i(arr, timestamp):
    """Return the index in arr where arr[i] < ts.

    This is used to reduce the samples returned from the Couchbase rest api
    to the timestamps occouring after our last poll time.

    :param timestamp: the last poll timestamp to compare against
    :type timestamp: long
    :rtype: int
    """
    # arr metrics include a list of timestamps (milliseconds)
    # find the sample number to start recording from, start from the end of the list
    sample_i = len(arr) - 1
    while sample_i > 0:
        if arr[sample_i] < timestamp:
            break
        sample_i -= 1
    return sample_i


def avg(arr):
    """Return average value from the list of values.

    :param arr: A list of int or float values
    :type arr: list
    :rtype: float
    """
    ret = 0.0
    items = len(arr)
    if items > 0:
        ret = sum(arr)/items
    return ret


def stats(arr):
    """Return the min, max and average value from the list of values.

    :param arr: A list of int or float values
    :type arr: list
    :rtype: tuple(float, float, float)
    """
    ret = 0.0
    items = len(arr)
    if items > 0:
        ret = sum(arr)/items
    return (min(arr), max(arr), ret)


class CouchBase(Reader):
    """Plugin to record nginx stub_status metrics."""

    # default config values
    defaults = {
        'poll.interval': 10,
        'host': 'localhost',
        'port': 8091,
        'username': 'Administrator',
        'password': '',
        'proto': 'http',
        'timeout': 10,
        'metrics_index': [
            'indexerThreads',
            'maxRollbackPoints',
            'memorySnapshotInterval',
            'stableSnapshotInterval'
        ],
        'metrics_bucket': [
            "avg_bg_wait_time",
            "avg_disk_commit_time",
            "avg_disk_update_time",
            "bg_wait_count",
            "bg_wait_total",
            "bytes_read",
            "bytes_written",
            "cas_badval",
            "cas_hits",
            "cas_misses",
            "cmd_get",
            "cmd_set",
            "couch_docs_actual_disk_size",
            "couch_docs_data_size",
            "couch_docs_disk_size",
            "couch_docs_fragmentation",
            "couch_spatial_data_size",
            "couch_spatial_disk_size",
            "couch_spatial_ops",
            "couch_total_disk_size",
            "couch_views_actual_disk_size",
            "couch_views_data_size",
            "couch_views_disk_size",
            "couch_views_fragmentation",
            "couch_views_ops",
            "cpu_idle_ms",
            "cpu_local_ms",
            "cpu_utilization_rate",
            "curr_connections",
            "curr_items",
            "curr_items_tot",
            "decr_hits",
            "decr_misses",
            "delete_hits",
            "delete_misses",
            "disk_commit_count",
            "disk_commit_total",
            "disk_update_count",
            "disk_update_total",
            "disk_write_queue",
            "ep_bg_fetched",
            "ep_cache_miss_rate",
            "ep_dcp_2i_backoff",
            "ep_dcp_2i_count",
            "ep_dcp_2i_items_remaining",
            "ep_dcp_2i_items_sent",
            "ep_dcp_2i_producer_count",
            "ep_dcp_2i_total_backlog_size",
            "ep_dcp_2i_total_bytes",
            "ep_dcp_fts_backoff",
            "ep_dcp_fts_count",
            "ep_dcp_fts_items_remaining",
            "ep_dcp_fts_items_sent",
            "ep_dcp_fts_producer_count",
            "ep_dcp_fts_total_backlog_size",
            "ep_dcp_fts_total_bytes",
            "ep_dcp_other_backoff",
            "ep_dcp_other_count",
            "ep_dcp_other_items_remaining",
            "ep_dcp_other_items_sent",
            "ep_dcp_other_producer_count",
            "ep_dcp_other_total_backlog_size",
            "ep_dcp_other_total_bytes",
            "ep_dcp_replica_backoff",
            "ep_dcp_replica_count",
            "ep_dcp_replica_items_remaining",
            "ep_dcp_replica_items_sent",
            "ep_dcp_replica_producer_count",
            "ep_dcp_replica_total_backlog_size",
            "ep_dcp_replica_total_bytes",
            "ep_dcp_views_backoff",
            "ep_dcp_views_count",
            "ep_dcp_views_items_remaining",
            "ep_dcp_views_items_sent",
            "ep_dcp_views_producer_count",
            "ep_dcp_views_total_backlog_size",
            "ep_dcp_views_total_bytes",
            "ep_dcp_views+indexes_backoff",
            "ep_dcp_views+indexes_count",
            "ep_dcp_views+indexes_items_remaining",
            "ep_dcp_views+indexes_items_sent",
            "ep_dcp_views+indexes_producer_count",
            "ep_dcp_views+indexes_total_backlog_size",
            "ep_dcp_views+indexes_total_bytes",
            "ep_dcp_xdcr_backoff",
            "ep_dcp_xdcr_count",
            "ep_dcp_xdcr_items_remaining",
            "ep_dcp_xdcr_items_sent",
            "ep_dcp_xdcr_producer_count",
            "ep_dcp_xdcr_total_backlog_size",
            "ep_dcp_xdcr_total_bytes",
            "ep_diskqueue_drain",
            "ep_diskqueue_fill",
            "ep_diskqueue_items",
            "ep_flusher_todo",
            "ep_item_commit_failed",
            "ep_kv_size",
            "ep_max_size",
            "ep_mem_high_wat",
            "ep_mem_low_wat",
            "ep_meta_data_memory",
            "ep_num_non_resident",
            "ep_num_ops_del_meta",
            "ep_num_ops_del_ret_meta",
            "ep_num_ops_get_meta",
            "ep_num_ops_set_meta",
            "ep_num_ops_set_ret_meta",
            "ep_num_value_ejects",
            "ep_oom_errors",
            "ep_ops_create",
            "ep_ops_update",
            "ep_overhead",
            "ep_queue_size",
            "ep_resident_items_rate",
            "ep_tap_rebalance_count",
            "ep_tap_rebalance_qlen",
            "ep_tap_rebalance_queue_backfillremaining",
            "ep_tap_rebalance_queue_backoff",
            "ep_tap_rebalance_queue_drain",
            "ep_tap_rebalance_queue_fill",
            "ep_tap_rebalance_queue_itemondisk",
            "ep_tap_rebalance_total_backlog_size",
            "ep_tap_replica_count",
            "ep_tap_replica_qlen",
            "ep_tap_replica_queue_backfillremaining",
            "ep_tap_replica_queue_backoff",
            "ep_tap_replica_queue_drain",
            "ep_tap_replica_queue_fill",
            "ep_tap_replica_queue_itemondisk",
            "ep_tap_replica_total_backlog_size",
            "ep_tap_total_count",
            "ep_tap_total_qlen",
            "ep_tap_total_queue_backfillremaining",
            "ep_tap_total_queue_backoff",
            "ep_tap_total_queue_drain",
            "ep_tap_total_queue_fill",
            "ep_tap_total_queue_itemondisk",
            "ep_tap_total_total_backlog_size",
            "ep_tap_user_count",
            "ep_tap_user_qlen",
            "ep_tap_user_queue_backfillremaining",
            "ep_tap_user_queue_backoff",
            "ep_tap_user_queue_drain",
            "ep_tap_user_queue_fill",
            "ep_tap_user_queue_itemondisk",
            "ep_tap_user_total_backlog_size",
            "ep_tmp_oom_errors",
            "ep_vb_total",
            "evictions",
            "get_hits",
            "get_misses",
            "hibernated_requests",
            "hibernated_waked",
            "hit_ratio",
            "incr_hits",
            "incr_misses",
            "mem_actual_free",
            "mem_actual_used",
            "mem_free",
            "mem_total",
            "mem_used",
            "mem_used_sys",
            "misses",
            "ops",
            "rest_requests",
            "swap_total",
            "swap_used",
            "timestamp",
            "vb_active_eject",
            "vb_active_itm_memory",
            "vb_active_meta_data_memory",
            "vb_active_num",
            "vb_active_num_non_resident",
            "vb_active_ops_create",
            "vb_active_ops_update",
            "vb_active_queue_age",
            "vb_active_queue_drain",
            "vb_active_queue_fill",
            "vb_active_queue_size",
            "vb_active_resident_items_ratio",
            "vb_avg_active_queue_age",
            "vb_avg_pending_queue_age",
            "vb_avg_replica_queue_age",
            "vb_avg_total_queue_age",
            "vb_pending_curr_items",
            "vb_pending_eject",
            "vb_pending_itm_memory",
            "vb_pending_meta_data_memory",
            "vb_pending_num",
            "vb_pending_num_non_resident",
            "vb_pending_ops_create",
            "vb_pending_ops_update",
            "vb_pending_queue_age",
            "vb_pending_queue_drain",
            "vb_pending_queue_fill",
            "vb_pending_queue_size",
            "vb_pending_resident_items_ratio",
            "vb_replica_curr_items",
            "vb_replica_eject",
            "vb_replica_itm_memory",
            "vb_replica_meta_data_memory",
            "vb_replica_num",
            "vb_replica_num_non_resident",
            "vb_replica_ops_create",
            "vb_replica_ops_update",
            "vb_replica_queue_age",
            "vb_replica_queue_drain",
            "vb_replica_queue_fill",
            "vb_replica_queue_size",
            "vb_replica_resident_items_ratio",
            "vb_total_queue_age",
            "xdc_ops"
        ],
        'limit_bucket': False
    }

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

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


        # Nginx connection
        self.base_url = "{0}://{1}:{2}".format(self.config.get("proto"),

                                               self.config.get("host"),

                                               self.config.get("port"))

        self.auth = (self.config.get('username'), self.config.get('password'))

        self.http_pm = urllib2.HTTPPasswordMgrWithDefaultRealm()
        self.urls = dict()
        self.timeout = config.get('timeout')
        self.lock = Lock()
        self.last_poll = 0

    def authenticate(self, url):
        """Setup authentication to the given url.

        :param url: The URL to setup authentication for.
        :type url: str
        """
        auth = self.auth
        with self.lock:
            self.http_pm.add_password(None, url, auth[0], auth[1])
            handler = urllib2.HTTPBasicAuthHandler(self.http_pm)
            opener = urllib2.build_opener(handler)
            try:
                opener.open(url)
            except ValueError as exc:
                err = "Couchbase: invalid url: {0}: {1}"
                self.log.error(err.format(url, exc))

            urllib2.install_opener(opener)
            self.urls[url] = True

    def request(self, url):
        """Make an HTTP request and return decoded JSON.

        :param url: The URL to request.
        :type url: str
        :rtype: object
        """
        if url not in self.urls:
            self.authenticate(url)

        resp = ""
        ret = dict()
        try:
            self.lock.acquire(True)
            resp = urllib2.urlopen(url, timeout=self.timeout)
        except urllib2.HTTPError as exc:
            err = "Couchbase: HTTP Error: {0}: {1}"
            self.log.error(err.format(url, exc))

            return ret
        finally:
            self.lock.release()
        if resp is None or resp.getcode() != 200:

            return ret
        try:
            ret = json.load(resp)
        except ValueError as exc:
            err = "Couchbase: JSON decode error: {0}: {1}: {2}"
            self.log.error(err.format(url, resp, exc))

            return ret
        return ret

    def get_index_settings(self, rset):
        """Return index settings.

        :param rset: A ResultSet to add results to
        :type rset: ResultSet
        """
        url = "{0}/settings/indexes".format(self.base_url)
        data = self.request(url)
        metric_names = self.config.get('metrics_index')

        metrics = deque()
        for name in metric_names:
            if name in data:
                metrics.append(Gauge(name, data[name]))
        res = Result("couchbase.index", metrics)
        rset.add(res)
        return

    def get_pool_stats_all(self, rset):
        """Get basic pool metrics, add to the ResultSet.

        todo: reduce cyclo complexity

        :param rset: A ResultSet to add results to
        :type rset: ResultSet
        """
        # get basic metrics for the given pool
        pool_url = "{0}/pools/default/buckets/".format(self.base_url)
        data = self.request(pool_url)

        # response should be an array of dicts - we want basicStats from each
        mnames = None
        if self.config.get("limit_bucket"):

            mnames = deque(self.config.get("metrics_bucket"))

        for bucket in data:
            try:
                # record misc pool metrics
                self.get_pool_stats_misc(data, rset)
                # also record detailed bucket metrics
                self.get_bucket_stats(bucket['name'], rset, mnames)
            except KeyError as exc:
                err = "Couchbase: get_pool_stats_all: server response missing key: {0}"
                self.log.error(err.format(exc))

        return

    def get_pool_stats_misc(self, data, rset):
        """Get basic pool metrics, add to the ResultSet.

        todo: reduce cyclo complexity

        :param data: Couchbase metrics returned from api call
        :type data: dict
        :param rset: A ResultSet to add results to
        :type rset: ResultSet
        """
        # data should be an array of dicts - we want basicStats from each
        for bucket in data:
            cnt_healthy = 0
            cnt_repl = 0
            cnt_members = 0
            metrics = deque()
            for entry in bucket['nodes']:
                if entry["status"] == "healthy":
                    cnt_healthy += 1
                if entry["replication"] == 1:
                    cnt_repl += 1
                if entry["clusterMembership"] == "active":
                    cnt_members += 1
                if "thisNode" in entry and entry["thisNode"]:
                    for key, val in entry["interestingStats"].items():
                        # record metrics
                        metrics.append(Gauge(key, val))
            # other various metrics
            metrics.append(Gauge("nodes_healthy", cnt_healthy))
            metrics.append(Gauge("nodes_replicating", cnt_repl))
            metrics.append(Gauge("nodes_members", cnt_members))
            # record metrics into a Result with meta data for the pool infos
            res = Result("couchbase.bucket_basic", metrics)
            res.meta.add(String("bucket", bucket["name"]))
            rset.add(res)
        return

    def get_bucket_stats(self, bucket, rset, mnames):
        """Get bucket metrics, add to the ResultSet.

        :param bucket: The name of the bucket to query
        :type bucket: str
        :param rset: A ResultSet to add results to
        :param mnames: A deque/iterable of metric names to record
        :type mnames: deque
        :type rset: ResultSet
        """
        bucket_url = "{0}/pools/default/buckets/{1}/stats".format(self.base_url, bucket)
        data = self.request(bucket_url)
        try:
            metrics = deque()
            # arr metrics include a list of timestamps (milliseconds)
            # find the sample number to start recording from, start from the end of the list
            times = data["op"]["samples"]["timestamp"]
            sample_i = get_sample_i(times, self.last_poll)

            if mnames is not None:
                # record average values from the samples
                for metric, val in data["op"]["samples"].items():
                    if metric not in mnames:
                        continue
                    avg_val = avg(val[sample_i:])
                    metrics.append(Gauge(metric, avg_val))
            else:
                # record average values from the samples
                for metric, val in data["op"]["samples"].items():
                    avg_val = avg(val[sample_i:])
                    metrics.append(Gauge(metric, avg_val))

        except KeyError as exc:
            err = "Couchbase: get_bucket_stats: server response missing key: {0}"
            self.log.error(err.format(exc))

        # record metrics into a Result with meta data for the pool infos
        res = Result("couchbase.bucket_basic", metrics)
        res.meta.add(String("bucket", bucket))
        rset.add(res)
        return

    def get_node_stats_all(self, rset):
        """Get node metrics, add to the ResultSet.

        todo:
            - record compaction settings
            - reduce cyclo complexity

        :param rset: A ResultSet to add metrics to.
        :type rset: ResultSet
        """
        # fix me - pool is name: url dict
        node_url = "{0}/pools/nodes/".format(self.base_url)
        data = self.request(node_url)
        try:
            self.get_node_stats_misc(data, rset)

            stat = "nodes"
            metrics = deque()
            for node in data[stat]:
                # only record metrics for the current node
                if "thisNode" in node and node["thisNode"]:
                    # record all key,val in interestingStats
                    for metric, val in node["interestingStats"].items():
                        metrics.append(Gauge(metric, val))
                    for name in ["mcdMemoryAllocated", "mcdMemoryReserved"]:
                        metrics.append(Gauge(name, node[name]))
        except KeyError as exc:
            err = "Couchbase: get_node_stats: server response missing key: {0}"
            self.log.error(err.format(exc))

        res = Result("couchbase.nodes", metrics)
        rset.add(res)
        return

    def get_node_stats_misc(self, data, rset):
        """Get node metrics, add to the ResultSet.

        todo:
            - record compaction settings
            - reduce cyclo complexity

        :param data: A dict from couchbase api call containing metrics.
        :type data: dict
        :param rset: A ResultSet to add metrics to.
        :type rset: ResultSet
        """
        metrics = deque()
        for key, val in data["counters"].items():
            metrics.append(Counter(key, val))
        res = Result("couchbase.counters", metrics)
        rset.add(res)

        metrics = deque()
        for name in ["ftsMemoryQuota", "indexMemoryQuota", "maxBucketCount",
                     "memoryQuota"]:
            metrics.append(Gauge(name, data[name]))
        res = Result("couchbase.quotas", metrics)
        rset.add(res)

        stat = "storageTotals"
        metrics = deque()
        for key in data[stat].keys():
            for metric, val in data[stat][key].items():
                name = "{0}_{1}".format(key, metric)
                metrics.append(Gauge(name, val))
        res = Result("couchbase.storage", metrics)
        rset.add(res)
        return

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

        :rtype: ResultSet
        """
        rset = ResultSet()
        self.get_index_settings(rset)
        self.get_pool_stats_all(rset)
        self.get_node_stats_all(rset)
        self.last_poll = time.time() * 1000
        return rset



Push — master ( d84ba0...9b96e9 )

CouchBase.poll() A

Complexity

Size

Duplication

Importance

1. Missing Dependencies

2. Missing init.py files

1. Missing Dependencies

2. Missing init.py files

s4z / plumd

Push — master ( d84ba0...9b96e9 )

CouchBase.poll() A

Complexity

Size

Duplication

Importance

1. Missing Dependencies

2. Missing __init__.py files

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

2. Missing init.py files

2. Missing init.py files
