<commit_before>import glob
import os
import re
from typing import Dict, List, Tuple, Union

import numpy as np

from cbagent.stores import PerfStore
from logger import logger
from perfrunner.settings import CBMONITOR_HOST

Number = Union[float, int]

Metric = Tuple[
    Number,          # Value
    List[str],       # Snapshots
    Dict[str, str],  # Metric info
]

DailyMetric = Tuple[
    str,        # Metric
    Number,     # Value
    List[str],  # Snapshots
]


def s2m(seconds: float) -> float:
    """Convert seconds to minutes."""
    return round(seconds / 60, 1)


def strip(s: str) -> str:
    for c in ' &()':
        s = s.replace(c, '')
    return s.lower()


class MetricHelper:

    def __init__(self, test):
        self.test = test
        self.test_config = test.test_config
        self.cluster_spec = test.cluster_spec

        self.store = PerfStore(CBMONITOR_HOST)

    @property
    def _title(self) -> str:
        return self.test_config.showfast.title

    @property
    def _snapshots(self) -> List[str]:
        return self.test.cbmonitor_snapshots

    @property
    def _num_nodes(self):
        return self.test_config.cluster.initial_nodes[0]

    def _metric_info(self,
                     metric_id: str = None,
                     title: str = None,
                     order_by: str = '') -> Dict[str, str]:
        return {
            'id': metric_id or self.test_config.name,
            'title': title or self._title,
            'orderBy': order_by,
        }

    def ycsb_queries(self, value: float, name: str, title: str) -> Metric:
        metric_id = '{}_{}'.format(self.test_config.name, name)
        title = '{}, {}'.format(title, self.test_config.showfast.title)
        metric_info = self._metric_info(metric_id, title)
        return value, self._snapshots, metric_info

    def avg_n1ql_throughput(self) -> Metric:
        metric_id = '{}_avg_query_requests'.format(self.test_config.name)
        title = 'Avg. Query Throughput (queries/sec), {}'.format(self._title)
        metric_info = self._metric_info(metric_id, title)
        throughput = self._avg_n1ql_throughput()
        return throughput, self._snapshots, metric_info

    def _avg_n1ql_throughput(self) -> int:
        test_time = self.test_config.access_settings.time

        query_node = self.cluster_spec.servers_by_role('n1ql')[0]
        vitals = self.test.rest.get_query_stats(query_node)
        total_requests = vitals['requests.count']

        return int(total_requests / test_time)

    def bulk_n1ql_throughput(self, time_elapsed: float) -> Metric:
        metric_info = self._metric_info()

        items = self.test_config.load_settings.items / 4
        throughput = round(items / time_elapsed)

        return throughput, self._snapshots, metric_info

    def fts_index(self,
                  elapsed_time: float,
                  order_by: str,
                  name: str ='FTS') -> Metric:
        metric_id = self.test_config.name
        title = 'Index build time(sec), {}, {} node, {}'.\
            format(self._title, self._num_nodes, name)
        metric_info = self._metric_info(metric_id, title, order_by)

        index_time = round(elapsed_time, 1)

        return index_time, self._snapshots, metric_info

    def fts_index_size(self,
                       index_size_raw: int,
                       order_by: str,
                       name: str = 'FTS') -> Metric:
        metric_id = "{}_indexsize".format(self.test_config.name)
        title = 'Index size (MB), {}, {} node, {}'.\
                format(self._title, self._num_nodes, name)
        metric_info = self._metric_info(metric_id, title, order_by)

        index_size_mb = int(index_size_raw / (1024 ** 2))

        return index_size_mb, self._snapshots, metric_info

    def fts_rebalance_time(self,
                           reb_time: float,
                           order_by: str,
                           name: str = 'FTS') -> Metric:
        metric_id = "{}_reb".format(self.test_config.name)
        title = 'Rebalance time (min), {}, {}'.format(self._title, name)
        metric_info = self._metric_info(metric_id, title, order_by)

        reb_time = s2m(reb_time)

        return reb_time, self._snapshots, metric_info

    def jts_throughput(self, orderby) -> Metric:
        metric_id = '{}_{}'.format(self.test_config.name, "jts_throughput")
        metric_id = metric_id.replace('.', '')
        title = "Average Throughput (q/sec), {}".format(self._title)
        metric_info = self._metric_info(metric_id, title, orderby)
        timings = self._jts_metric(collector="jts_stats", metric="jts_throughput")
        thr = round(np.average(timings), 2)
        if thr > 100:
            thr = round(thr)
        return thr, self._snapshots, metric_info

    def jts_latency(self, orderby, percentile=50) -> Metric:
        prefix = "Average latency (ms)"
        if percentile != 50:
            prefix = "{}th percentile latency (ms)".format(percentile)
        metric_id = '{}_{}'.format(self.test_config.name, "jts_latency")
        metric_id = metric_id.replace('.', '')
        title = "{}, {}".format(prefix, self._title)
        metric_info = self._metric_info(metric_id, title, orderby)
        timings = self._jts_metric(collector="jts_stats", metric="jts_latency")
        lat = round(np.percentile(timings, percentile), 2)
        if lat > 100:
            lat = round(lat)
        return lat, self._snapshots, metric_info

    def _jts_metric(self, collector, metric):
        timings = []
        for bucket in self.test_config.buckets:
            db = self.store.build_dbname(cluster=self.test.cbmonitor_clusters[0],
                                         collector=collector,
                                         bucket=bucket)
            timings += self.store.get_values(db, metric=metric)
        return timings

    def max_ops(self) -> Metric:
        metric_info = self._metric_info()
        throughput = self._max_ops()

        return throughput, self._snapshots, metric_info

    def _max_ops(self) -> int:
        values = []
        for bucket in self.test_config.buckets:
            db = self.store.build_dbname(cluster=self.test.cbmonitor_clusters[0],
                                         collector='ns_server',
                                         bucket=bucket)
            values += self.store.get_values(db, metric='ops')

        return int(np.percentile(values, 90))

    def get_collector_values(self, collector):
        values = []
        for bucket in self.test_config.buckets:
            db = self.store.build_dbname(cluster=self.test.cbmonitor_clusters[0],
                                         collector=collector,
                                         bucket=bucket)
            values += self.store.get_values(db, metric=collector)
        return values

    def count_overthreshold_value_of_collector(self, collector, threshold):
        values = self.get_collector_values(collector)
        return sum(v >= threshold for v in values)

    def get_percentile_value_of_collector(self, collector, percentile):
        values = self.get_collector_values(collector)
        return np.percentile(values, percentile)

    def lag_collector_metric(self, collector, collector_title,
                             order_by, percentile: Number = 95) -> Metric:
        metric_id = '{}_{}th_{}'.format(self.test_config.name, percentile, collector)
        title = '{}th percentile {} (ms), {}'.format(
            percentile, collector_title, self._title)
        metric_info = self._metric_info(metric_id, title, order_by)
        lag = self.get_percentile_value_of_collector(collector, percentile)
        return round(lag, 1), self._snapshots, metric_info

    def xdcr_lag(self, percentile: Number = 95) -> Metric:
        return self.lag_collector_metric('xdcr_lag', 'replication lag', '', percentile)

    def avg_replication_rate(self, time_elapsed: float) -> Metric:
        metric_info = self._metric_info()

        rate = self._avg_replication_rate(time_elapsed)

        return rate, self._snapshots, metric_info

    def _avg_replication_rate(self, time_elapsed: float) -> float:
        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 max_drain_rate(self, time_elapsed: float) -> Metric:
        metric_info = self._metric_info()

        items_per_node = self.test_config.load_settings.items / self._num_nodes
        drain_rate = round(items_per_node / time_elapsed)

        return drain_rate, self._snapshots, metric_info

    def avg_disk_write_queue(self) -> Metric:
        metric_info = self._metric_info()

        values = []
        for bucket in self.test_config.buckets:
            db = self.store.build_dbname(cluster=self.test.cbmonitor_clusters[0],
                                         collector='ns_server',
                                         bucket=bucket)
            values += self.store.get_values(db, metric='disk_write_queue')

        disk_write_queue = int(np.average(values))

        return disk_write_queue, self._snapshots, metric_info

    def avg_total_queue_age(self) -> Metric:
        metric_info = self._metric_info()

        values = []
        for bucket in self.test_config.buckets:
            db = self.store.build_dbname(cluster=self.test.cbmonitor_clusters[0],
                                         collector='ns_server',
                                         bucket=bucket)
            values += self.store.get_values(db, metric='vb_avg_total_queue_age')

        avg_total_queue_age = int(np.average(values))

        return avg_total_queue_age, self._snapshots, metric_info

    def avg_couch_views_ops(self) -> Metric:
        metric_info = self._metric_info()

        values = []
        for bucket in self.test_config.buckets:
            db = self.store.build_dbname(cluster=self.test.cbmonitor_clusters[0],
                                         collector='ns_server',
                                         bucket=bucket)
            values += self.store.get_values(db, metric='couch_views_ops')

        couch_views_ops = int(np.average(values))

        return couch_views_ops, self._snapshots, metric_info

    def query_latency(self, percentile: Number) -> Metric:
        metric_id = self.test_config.name
        title = '{}th percentile query latency (ms), {}'.format(percentile,
                                                                self._title)

        metric_info = self._metric_info(metric_id, title)

        latency = self._query_latency(percentile)

        return latency, self._snapshots, metric_info

    def _query_latency(self, percentile: Number) -> float:
        values = []
        for bucket in self.test_config.buckets:
            db = self.store.build_dbname(cluster=self.test.cbmonitor_clusters[0],
                                         collector='spring_query_latency',
                                         bucket=bucket)
            values += self.store.get_values(db, metric='latency_query')

        query_latency = np.percentile(values, percentile)
        if query_latency < 100:<commit_msg>Order query results by query id<commit_after>import glob
import os
import re
from typing import Dict, List, Tuple, Union

import numpy as np

from cbagent.stores import PerfStore
from logger import logger
from perfrunner.settings import CBMONITOR_HOST

Number = Union[float, int]

Metric = Tuple[
    Number,          # Value
    List[str],       # Snapshots
    Dict[str, str],  # Metric info
]

DailyMetric = Tuple[
    str,        # Metric
    Number,     # Value
    List[str],  # Snapshots
]


def s2m(seconds: float) -> float:
    """Convert seconds to minutes."""
    return round(seconds / 60, 1)


def strip(s: str) -> str:
    for c in ' &()':
        s = s.replace(c, '')
    return s.lower()


class MetricHelper:

    def __init__(self, test):
        self.test = test
        self.test_config = test.test_config
        self.cluster_spec = test.cluster_spec

        self.store = PerfStore(CBMONITOR_HOST)

    @property
    def _title(self) -> str:
        return self.test_config.showfast.title

    @property
    def _snapshots(self) -> List[str]:
        return self.test.cbmonitor_snapshots

    @property
    def _num_nodes(self):
        return self.test_config.cluster.initial_nodes[0]

    def _metric_info(self,
                     metric_id: str = None,
                     title: str = None,
                     order_by: str = '') -> Dict[str, str]:
        return {
            'id': metric_id or self.test_config.name,
            'title': title or self._title,
            'orderBy': order_by,
        }

    def ycsb_queries(self, value: float, name: str, title: str) -> Metric:
        metric_id = '{}_{}'.format(self.test_config.name, name)
        title = '{}, {}'.format(title, self.test_config.showfast.title)
        metric_info = self._metric_info(metric_id, title)
        return value, self._snapshots, metric_info

    @property
    def n1ql_query_id(self) -> str:
        query_id = self._title.split(',')[0]
        query_id = query_id.replace('CI', '').replace('Q', '')
        return '{:05d}'.format(int(query_id))

    def avg_n1ql_throughput(self) -> Metric:
        metric_id = '{}_avg_query_requests'.format(self.test_config.name)
        title = 'Avg. Query Throughput (queries/sec), {}'.format(self._title)

        metric_info = self._metric_info(metric_id, title,
                                        order_by=self.n1ql_query_id)
        throughput = self._avg_n1ql_throughput()
        return throughput, self._snapshots, metric_info

    def _avg_n1ql_throughput(self) -> int:
        test_time = self.test_config.access_settings.time

        query_node = self.cluster_spec.servers_by_role('n1ql')[0]
        vitals = self.test.rest.get_query_stats(query_node)
        total_requests = vitals['requests.count']

        return int(total_requests / test_time)

    def bulk_n1ql_throughput(self, time_elapsed: float) -> Metric:
        metric_info = self._metric_info()

        items = self.test_config.load_settings.items / 4
        throughput = round(items / time_elapsed)

        return throughput, self._snapshots, metric_info

    def fts_index(self,
                  elapsed_time: float,
                  order_by: str,
                  name: str ='FTS') -> Metric:
        metric_id = self.test_config.name
        title = 'Index build time(sec), {}, {} node, {}'.\
            format(self._title, self._num_nodes, name)
        metric_info = self._metric_info(metric_id, title, order_by)

        index_time = round(elapsed_time, 1)

        return index_time, self._snapshots, metric_info

    def fts_index_size(self,
                       index_size_raw: int,
                       order_by: str,
                       name: str = 'FTS') -> Metric:
        metric_id = "{}_indexsize".format(self.test_config.name)
        title = 'Index size (MB), {}, {} node, {}'.\
                format(self._title, self._num_nodes, name)
        metric_info = self._metric_info(metric_id, title, order_by)

        index_size_mb = int(index_size_raw / (1024 ** 2))

        return index_size_mb, self._snapshots, metric_info

    def fts_rebalance_time(self,
                           reb_time: float,
                           order_by: str,
                           name: str = 'FTS') -> Metric:
        metric_id = "{}_reb".format(self.test_config.name)
        title = 'Rebalance time (min), {}, {}'.format(self._title, name)
        metric_info = self._metric_info(metric_id, title, order_by)

        reb_time = s2m(reb_time)

        return reb_time, self._snapshots, metric_info

    def jts_throughput(self, orderby) -> Metric:
        metric_id = '{}_{}'.format(self.test_config.name, "jts_throughput")
        metric_id = metric_id.replace('.', '')
        title = "Average Throughput (q/sec), {}".format(self._title)
        metric_info = self._metric_info(metric_id, title, orderby)
        timings = self._jts_metric(collector="jts_stats", metric="jts_throughput")
        thr = round(np.average(timings), 2)
        if thr > 100:
            thr = round(thr)
        return thr, self._snapshots, metric_info

    def jts_latency(self, orderby, percentile=50) -> Metric:
        prefix = "Average latency (ms)"
        if percentile != 50:
            prefix = "{}th percentile latency (ms)".format(percentile)
        metric_id = '{}_{}'.format(self.test_config.name, "jts_latency")
        metric_id = metric_id.replace('.', '')
        title = "{}, {}".format(prefix, self._title)
        metric_info = self._metric_info(metric_id, title, orderby)
        timings = self._jts_metric(collector="jts_stats", metric="jts_latency")
        lat = round(np.percentile(timings, percentile), 2)
        if lat > 100:
            lat = round(lat)
        return lat, self._snapshots, metric_info

    def _jts_metric(self, collector, metric):
        timings = []
        for bucket in self.test_config.buckets:
            db = self.store.build_dbname(cluster=self.test.cbmonitor_clusters[0],
                                         collector=collector,
                                         bucket=bucket)
            timings += self.store.get_values(db, metric=metric)
        return timings

    def max_ops(self) -> Metric:
        metric_info = self._metric_info()
        throughput = self._max_ops()

        return throughput, self._snapshots, metric_info

    def _max_ops(self) -> int:
        values = []
        for bucket in self.test_config.buckets:
            db = self.store.build_dbname(cluster=self.test.cbmonitor_clusters[0],
                                         collector='ns_server',
                                         bucket=bucket)
            values += self.store.get_values(db, metric='ops')

        return int(np.percentile(values, 90))

    def get_collector_values(self, collector):
        values = []
        for bucket in self.test_config.buckets:
            db = self.store.build_dbname(cluster=self.test.cbmonitor_clusters[0],
                                         collector=collector,
                                         bucket=bucket)
            values += self.store.get_values(db, metric=collector)
        return values

    def count_overthreshold_value_of_collector(self, collector, threshold):
        values = self.get_collector_values(collector)
        return sum(v >= threshold for v in values)

    def get_percentile_value_of_collector(self, collector, percentile):
        values = self.get_collector_values(collector)
        return np.percentile(values, percentile)

    def lag_collector_metric(self, collector, collector_title,
                             order_by, percentile: Number = 95) -> Metric:
        metric_id = '{}_{}th_{}'.format(self.test_config.name, percentile, collector)
        title = '{}th percentile {} (ms), {}'.format(
            percentile, collector_title, self._title)
        metric_info = self._metric_info(metric_id, title, order_by)
        lag = self.get_percentile_value_of_collector(collector, percentile)
        return round(lag, 1), self._snapshots, metric_info

    def xdcr_lag(self, percentile: Number = 95) -> Metric:
        return self.lag_collector_metric('xdcr_lag', 'replication lag', '', percentile)

    def avg_replication_rate(self, time_elapsed: float) -> Metric:
        metric_info = self._metric_info()

        rate = self._avg_replication_rate(time_elapsed)

        return rate, self._snapshots, metric_info

    def _avg_replication_rate(self, time_elapsed: float) -> float:
        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 max_drain_rate(self, time_elapsed: float) -> Metric:
        metric_info = self._metric_info()

        items_per_node = self.test_config.load_settings.items / self._num_nodes
        drain_rate = round(items_per_node / time_elapsed)

        return drain_rate, self._snapshots, metric_info

    def avg_disk_write_queue(self) -> Metric:
        metric_info = self._metric_info()

        values = []
        for bucket in self.test_config.buckets:
            db = self.store.build_dbname(cluster=self.test.cbmonitor_clusters[0],
                                         collector='ns_server',
                                         bucket=bucket)
            values += self.store.get_values(db, metric='disk_write_queue')

        disk_write_queue = int(np.average(values))

        return disk_write_queue, self._snapshots, metric_info

    def avg_total_queue_age(self) -> Metric:
        metric_info = self._metric_info()

        values = []
        for bucket in self.test_config.buckets:
            db = self.store.build_dbname(cluster=self.test.cbmonitor_clusters[0],
                                         collector='ns_server',
                                         bucket=bucket)
            values += self.store.get_values(db, metric='vb_avg_total_queue_age')

        avg_total_queue_age = int(np.average(values))

        return avg_total_queue_age, self._snapshots, metric_info

    def avg_couch_views_ops(self) -> Metric:
        metric_info = self._metric_info()

        values = []
        for bucket in self.test_config.buckets:
            db = self.store.build_dbname(cluster=self.test.cbmonitor_clusters[0],
                                         collector='ns_server',
                                         bucket=bucket)
            values += self.store.get_values(db, metric='couch_views_ops')

        couch_views_ops = int(np.average(values))

        return couch_views_ops, self._snapshots, metric_info

    def query_latency(self, percentile: Number) -> Metric:
        metric_id = self.test_config.name
        title = '{}th percentile query latency (ms), {}'.format(percentile,
                                                                self._title)
        metric_info = self._metric_info(metric_id, title)

        latency = self._query_latency(percentile)

        return latency, self._snapshots, metric_info

    def _query_latency(self, percentile: Number) -> float:
        values = []
        for bucket in self.test_config.buckets:
            db = self.store.build_dbname(cluster=self.test.cbmonitor_clusters[0],
                                         collector='spring_query_latency',
                                         bucket=bucket)
            values += self.store.get_values(db, metric='latency_query')

        query_latency = np.percentile(values, percentile)
        if query_latency < 100:
{'fraction_non_alphanumeric': 0.10377768128528007, 'fraction_numerical': 0.00438558402084238, 'mean_word_length': 2.2175188600167646, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 62, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 61, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4125682', 'n_tokens_mistral': 6976, 'n_tokens_neox': 6635, 'n_words': 1338}