<commit_before>import numpy as np
from logger import logger
from seriesly import Seriesly

from perfrunner.settings import StatsSettings


class MetricHelper(object):

    def __init__(self, test):
        self.test = test
        self.seriesly = Seriesly(StatsSettings.SERIESLY)
        self.test_config = test.test_config
        self.title = test.test_config.test_case.title
        self.cluster_spec = test.cluster_spec
        self.build = test.build

    @staticmethod
    def _get_query_params(metric):
        """Convert metric definition to Seriesly query params. E.g.:

            'avg_xdc_ops' -> {'ptr': '/xdc_ops',
                              'group': 1000000000000, 'reducer': 'avg'}

        Where group is constant."""
        return {'ptr': '/{}'.format(metric[4:]),
                'reducer': metric[:3],
                'group': 1000000000000}

    def _get_metric_info(self, title, order_by=''):
        return {
            'title': title,
            'orderBy': order_by,
        }

    def calc_ycsb_queries(self, value, name, title):
        metric = '{}_{}'.format(self.test_config.name, name)
        title = '{} , {}'.format(title, self.title)
        metric_info = self._get_metric_info(title)
        return value, metric, metric_info

    def calc_avg_n1ql_queries(self):
        metric = '{}_avg_query_requests'.format(self.test_config.name)
        title = 'Avg. Query Throughput (queries/sec), {}'.format(self.title)
        metric_info = self._get_metric_info(title)

        queries = self._calc_avg_n1ql_queries()
        return queries, metric, metric_info

    def _calc_avg_n1ql_queries(self):
        query_params = self._get_query_params('avg_query_requests')
        db = 'n1ql_stats{}'.format(self.test.cbagent.cluster_ids[0])
        data = self.seriesly[db].query(query_params)
        queries = data.values()[0][0]

        return int(queries)

    def cal_avg_n1ql_queries_perfdaily(self):
        return {
            "name": "avg_query_throughput",
            "description": "Avg. Query Throughput (queries/sec)",
            "value": self._calc_avg_n1ql_queries(),
            "larger_is_better": True,
            "threshold": self.test.test_config.dailyp_settings.threshold,
        }

    def parse_log(self, test_config, name):
        if name.find('Elasticsearch') != -1:
            self.test.cbagent.add_elastic_stats(test_config)

        fts_es = self.test.cbagent.fts_stats
        fts_es.collect_stats()
        total = fts_es.cbft_query_total()
        return total

    def calc_avg_fts_queries(self, order_by, name='FTS'):
        metric = '{}_avg_query_requests'.format(self.test_config.name)
        title = 'Query Throughput (queries/sec), {}, {} node, {}'.\
                format(self.title,
                       self.test_config.cluster.initial_nodes[0],
                       name)
        metric_info = self._get_metric_info(title, order_by=order_by)
        total_queries = self.parse_log(self.test_config, name)
        time_taken = self.test_config.access_settings.time
        qps = total_queries / float(time_taken)
        if qps < 100:
            qps = round(qps, 2)
        else:
            qps = round(qps)
        return qps, metric, metric_info

    def calc_latency_ftses_queries(self, percentile, dbname,
                                   metrics, order_by, name='FTS'):
        if percentile == 0:
            metric = '{}_average'.format(self.test_config.name)
            title = 'Average query latency (ms), {}, {} node, {}'.\
                    format(self.title,
                           self.test_config.cluster.initial_nodes[0],
                           name)
        else:
            metric = self.test_config.name
            title = '{}th percentile query latency (ms), {}, {} node, {}'. \
                    format(percentile,
                           self.title,
                           self.test_config.cluster.initial_nodes[0],
                           name)

        metric_info = self._get_metric_info(title, order_by=order_by)
        timings = []
        db = '{}{}'.format(dbname, self.test.cbagent.cluster_ids[0])
        data = self.seriesly[db].get_all()
        timings += [v[metrics] for v in data.values()]
        if percentile == 0:
            fts_latency = np.average(timings)
        else:
            fts_latency = np.percentile(timings, percentile)
        return round(fts_latency), metric, metric_info

    def calc_ftses_index(self, elapsedtime, order_by, name='FTS'):
        metric = self.test_config.name
        title = 'Index Throughput(sec), {}, {} node, {}'.\
                format(self.title,
                       self.test_config.cluster.initial_nodes[0],
                       name)
        metric_info = self._get_metric_info(title, order_by=order_by)
        return round(elapsedtime, 1), metric, metric_info

    def calc_ops_perfdaily(self):
        return {
            "name": 'average_throughput',
            "description": 'Max Throughput (ops/sec)',
            "value": self.calc_max_ops(),
            "larger_is_better": True,
            "threshold": self.test.test_config.dailyp_settings.threshold,
        }

    def calc_max_ops(self):
        values = []
        for bucket in self.test_config.buckets:
            db = 'ns_server{}{}'.format(self.test.cbagent.cluster_ids[0], bucket)
            data = self.seriesly[db].get_all()
            values += [v['ops'] for v in data.values()]
        return int(np.percentile(values, 90))

    def calc_xdcr_lag(self, percentile=90):
        metric = '{}_{}th_xdc_lag'.format(self.test_config.name, percentile)
        title = '{}th percentile replication lag (ms), {}'.format(
            percentile, self.title)
        metric_info = self._get_metric_info(title)

        timings = []
        for bucket in self.test_config.buckets:
            db = 'xdcr_lag{}{}'.format(self.test.cbagent.cluster_ids[0], bucket)
            data = self.seriesly[db].get_all()
            timings += [v['xdcr_lag'] for v in data.values()]
        lag = round(np.percentile(timings, percentile))

        return lag, metric, metric_info

    def calc_avg_replication_rate(self, time_elapsed):
        initial_items = self.test_config.load_settings.ops or \
            self.test_config.load_settings.items
        num_buckets = self.test_config.cluster.num_buckets
        avg_replication_rate = num_buckets * initial_items / time_elapsed

        return round(avg_replication_rate)

    def calc_avg_replication_rate_perfdaily(self, time_elapsed):
        initial_items = self.test_config.load_settings.ops or \<commit_msg>Track 95th percentile of XDCR latency<commit_after>import numpy as np
from logger import logger
from seriesly import Seriesly

from perfrunner.settings import StatsSettings


class MetricHelper(object):

    def __init__(self, test):
        self.test = test
        self.seriesly = Seriesly(StatsSettings.SERIESLY)
        self.test_config = test.test_config
        self.title = test.test_config.test_case.title
        self.cluster_spec = test.cluster_spec
        self.build = test.build

    @staticmethod
    def _get_query_params(metric):
        """Convert metric definition to Seriesly query params. E.g.:

            'avg_xdc_ops' -> {'ptr': '/xdc_ops',
                              'group': 1000000000000, 'reducer': 'avg'}

        Where group is constant."""
        return {'ptr': '/{}'.format(metric[4:]),
                'reducer': metric[:3],
                'group': 1000000000000}

    def _get_metric_info(self, title, order_by=''):
        return {
            'title': title,
            'orderBy': order_by,
        }

    def calc_ycsb_queries(self, value, name, title):
        metric = '{}_{}'.format(self.test_config.name, name)
        title = '{} , {}'.format(title, self.title)
        metric_info = self._get_metric_info(title)
        return value, metric, metric_info

    def calc_avg_n1ql_queries(self):
        metric = '{}_avg_query_requests'.format(self.test_config.name)
        title = 'Avg. Query Throughput (queries/sec), {}'.format(self.title)
        metric_info = self._get_metric_info(title)

        queries = self._calc_avg_n1ql_queries()
        return queries, metric, metric_info

    def _calc_avg_n1ql_queries(self):
        query_params = self._get_query_params('avg_query_requests')
        db = 'n1ql_stats{}'.format(self.test.cbagent.cluster_ids[0])
        data = self.seriesly[db].query(query_params)
        queries = data.values()[0][0]

        return int(queries)

    def cal_avg_n1ql_queries_perfdaily(self):
        return {
            "name": "avg_query_throughput",
            "description": "Avg. Query Throughput (queries/sec)",
            "value": self._calc_avg_n1ql_queries(),
            "larger_is_better": True,
            "threshold": self.test.test_config.dailyp_settings.threshold,
        }

    def parse_log(self, test_config, name):
        if name.find('Elasticsearch') != -1:
            self.test.cbagent.add_elastic_stats(test_config)

        fts_es = self.test.cbagent.fts_stats
        fts_es.collect_stats()
        total = fts_es.cbft_query_total()
        return total

    def calc_avg_fts_queries(self, order_by, name='FTS'):
        metric = '{}_avg_query_requests'.format(self.test_config.name)
        title = 'Query Throughput (queries/sec), {}, {} node, {}'.\
                format(self.title,
                       self.test_config.cluster.initial_nodes[0],
                       name)
        metric_info = self._get_metric_info(title, order_by=order_by)
        total_queries = self.parse_log(self.test_config, name)
        time_taken = self.test_config.access_settings.time
        qps = total_queries / float(time_taken)
        if qps < 100:
            qps = round(qps, 2)
        else:
            qps = round(qps)
        return qps, metric, metric_info

    def calc_latency_ftses_queries(self, percentile, dbname,
                                   metrics, order_by, name='FTS'):
        if percentile == 0:
            metric = '{}_average'.format(self.test_config.name)
            title = 'Average query latency (ms), {}, {} node, {}'.\
                    format(self.title,
                           self.test_config.cluster.initial_nodes[0],
                           name)
        else:
            metric = self.test_config.name
            title = '{}th percentile query latency (ms), {}, {} node, {}'. \
                    format(percentile,
                           self.title,
                           self.test_config.cluster.initial_nodes[0],
                           name)

        metric_info = self._get_metric_info(title, order_by=order_by)
        timings = []
        db = '{}{}'.format(dbname, self.test.cbagent.cluster_ids[0])
        data = self.seriesly[db].get_all()
        timings += [v[metrics] for v in data.values()]
        if percentile == 0:
            fts_latency = np.average(timings)
        else:
            fts_latency = np.percentile(timings, percentile)
        return round(fts_latency), metric, metric_info

    def calc_ftses_index(self, elapsedtime, order_by, name='FTS'):
        metric = self.test_config.name
        title = 'Index Throughput(sec), {}, {} node, {}'.\
                format(self.title,
                       self.test_config.cluster.initial_nodes[0],
                       name)
        metric_info = self._get_metric_info(title, order_by=order_by)
        return round(elapsedtime, 1), metric, metric_info

    def calc_ops_perfdaily(self):
        return {
            "name": 'average_throughput',
            "description": 'Max Throughput (ops/sec)',
            "value": self.calc_max_ops(),
            "larger_is_better": True,
            "threshold": self.test.test_config.dailyp_settings.threshold,
        }

    def calc_max_ops(self):
        values = []
        for bucket in self.test_config.buckets:
            db = 'ns_server{}{}'.format(self.test.cbagent.cluster_ids[0], bucket)
            data = self.seriesly[db].get_all()
            values += [v['ops'] for v in data.values()]
        return int(np.percentile(values, 90))

    def calc_xdcr_lag(self, percentile=95):
        metric = '{}_{}th_xdc_lag'.format(self.test_config.name, percentile)
        title = '{}th percentile replication lag (ms), {}'.format(
            percentile, self.title)
        metric_info = self._get_metric_info(title)

        timings = []
        for bucket in self.test_config.buckets:
            db = 'xdcr_lag{}{}'.format(self.test.cbagent.cluster_ids[0], bucket)
            data = self.seriesly[db].get_all()
            timings += [v['xdcr_lag'] for v in data.values()]
        lag = round(np.percentile(timings, percentile), 1)

        return lag, metric, metric_info

    def calc_avg_replication_rate(self, time_elapsed):
        initial_items = self.test_config.load_settings.ops or \
            self.test_config.load_settings.items
        num_buckets = self.test_config.cluster.num_buckets
        avg_replication_rate = num_buckets * initial_items / time_elapsed

        return round(avg_replication_rate)

    def calc_avg_replication_rate_perfdaily(self, time_elapsed):
        initial_items = self.test_config.load_settings.ops or \
{'fraction_non_alphanumeric': 0.11581006847768831, 'fraction_numerical': 0.008653773797877944, 'mean_word_length': 2.2375152253349575, 'pattern_counts': {'":': 20, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 4, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5260484', 'n_tokens_mistral': 4176, 'n_tokens_neox': 3967, 'n_words': 744}