CouchBase

Complexity

Total Complexity

45

Size/Duplication

Total Lines	508
Duplicated Lines	0 %

Importance

Changes	2
Bugs	1	Features	1

10 Methods

Rating	Name	Duplication	Size	Complexity
B	get_bucket_stats()	0	40	6
A	authenticate()	0	18	3
C	get_node_stats_all()	0	32	7
B	request()	0	30	6
B	get_pool_stats_all()	0	26	4
B	get_node_stats_misc()	0	34	5
F	get_pool_stats_misc()	0	36	9
A	__init__()	0	21	1
A	get_index_settings()	0	16	3
A	poll()	0	11	1

# -*- coding: utf-8 -*-
"""Couchbase metrics reader."""
import sys
import time
import json
import urllib2

The import urllib2 could not be resolved.

from collections import deque
from threading import Lock

from plumd import Counter, Gauge, String, Reader, ResultSet, Result

PY3 = sys.version_info > (3,)

__author__ = 'Kenny Freeman'
__email__ = 'kenny.freeman@gmail.com'
__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
    """
    if not arr:
        return 0.0
    ret = 0.0
    items = len(arr)
    if items > 0:
        try:
            ret = sum(arr)/items
        except TypeError as exc:

The variable exc seems to be unused.

            ret = 0.0
    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.URLError, urllib2.HTTPError, ValueError, AttributeError, IOError) 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:

The Instance of str does not seem to have a member named getcode.

            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