<commit_before>""" Aggregation classes used by the collector and monstatsd to batch messages sent to the forwarder.
"""
import logging
from time import time

from monagent.common.metrics import Gauge, BucketGauge, Counter, Histogram, Measurement, Set, Rate


log = logging.getLogger(__name__)


# This is used to ensure that metrics with a timestamp older than
# RECENT_POINT_THRESHOLD_DEFAULT seconds (or the value passed in to
# the MetricsAggregator constructor) get discarded rather than being
# input into the incorrect bucket. Currently, the MetricsAggregator
# does not support submitting values for the past, and all values get
# submitted for the timestamp passed into the flush() function.
# The MetricsBucketAggregator uses times that are aligned to "buckets"
# that are the length of the interval that is passed into the
# MetricsBucketAggregator constructor.
RECENT_POINT_THRESHOLD_DEFAULT = 3600


class Aggregator(object):
    """
    Abstract metric aggregator class.
    """

    def __init__(self, hostname, interval=1.0, expiry_seconds=300, recent_point_threshold=None):
        self.events = []
        self.total_count = 0
        self.count = 0
        self.event_count = 0
        self.hostname = hostname
        self.expiry_seconds = expiry_seconds
        self.interval = float(interval)

        recent_point_threshold = recent_point_threshold or RECENT_POINT_THRESHOLD_DEFAULT
        self.recent_point_threshold = int(recent_point_threshold)
        self.num_discarded_old_points = 0

    @staticmethod
    def formatter(metric, value, timestamp, dimensions, hostname, device_name=None, metric_type=None, interval=None):
        """ Formats metrics, put them into a Measurement class
            (metric, timestamp, value, {"dimensions": {"name1": "value1", "name2": "value2"}, ...})
            dimensions should be a dictionary
        """
        if hostname:
            dimensions['hostname'] = hostname
        if device_name:
            dimensions['device_name'] = device_name

        return Measurement(metric, int(timestamp), value, dimensions)

    def packets_per_second(self, interval):
        if interval == 0:
            return 0
        return round(float(self.count)/interval, 2)

    def submit_metric(self, name, value, mtype, dimensions=None, hostname=None, device_name=None, timestamp=None,
                      sample_rate=1):
        """ Add a metric to be aggregated """
        raise NotImplementedError()

    def event(self, title, text, date_happened=None, alert_type=None, aggregation_key=None, source_type_name=None,
              priority=None, dimensions=None, hostname=None):
        event = {
            'msg_title': title,
            'msg_text': text,
        }
        if date_happened is not None:
            event['timestamp'] = date_happened
        else:
            event['timestamp'] = int(time())
        if alert_type is not None:
            event['alert_type'] = alert_type
        if aggregation_key is not None:
            event['aggregation_key'] = aggregation_key
        if source_type_name is not None:
            event['source_type_name'] = source_type_name
        if priority is not None:
            event['priority'] = priority
        if dimensions is not None:
            event['dimensions'] = dimensions
        if hostname is not None:
            event['host'] = hostname
        else:
            event['host'] = self.hostname

        self.events.append(event)

    def flush(self):
        """ Flush aggreaged metrics """
        raise NotImplementedError()

    def flush_events(self):
        events = self.events
        self.events = []

        self.total_count += self.event_count
        self.event_count = 0

        log.debug("Received %d events since last flush" % len(events))

        return events


class MetricsBucketAggregator(Aggregator):
    """
    A metric aggregator class.
    """

    def __init__(self, hostname, interval=1.0, expiry_seconds=300, recent_point_threshold=None):
        super(MetricsBucketAggregator, self).__init__(hostname, interval, expiry_seconds, recent_point_threshold)
        self.metric_by_bucket = {}
        self.last_sample_time_by_context = {}
        self.current_bucket = None
        self.current_mbc = None
        self.last_flush_cutoff_time = 0
        self.metric_type_to_class = {
            'g': BucketGauge,
            'c': Counter,
            'h': Histogram,
            'ms': Histogram,
            's': Set,
        }

    def calculate_bucket_start(self, timestamp):
        return timestamp - (timestamp % self.interval)

    def submit_metric(self, name, value, mtype, dimensions=None, hostname=None,
                      device_name=None, timestamp=None, sample_rate=1):
        # Avoid calling extra functions to dedupe dimensions if there are none
        # Note: if you change the way that context is created, please also change create_empty_metrics,
        #  which counts on this order
        if dimensions is None:
            context = (name, {}, hostname, device_name)
        else:
            context = (name, dimensions, hostname, device_name)

        cur_time = time()
        # Check to make sure that the timestamp that is passed in (if any) is not older than
        #  recent_point_threshold.  If so, discard the point.
        if timestamp is not None and cur_time - int(timestamp) > self.recent_point_threshold:
            log.debug("Discarding %s - ts = %s , current ts = %s " % (name, timestamp, cur_time))
            self.num_discarded_old_points += 1
        else:
            timestamp = timestamp or cur_time<commit_msg>Fixes for issues encountered now that we use dimensions<commit_after>""" Aggregation classes used by the collector and monstatsd to batch messages sent to the forwarder.
"""
import logging
from time import time

from monagent.common.metrics import Gauge, BucketGauge, Counter, Histogram, Measurement, Set, Rate


log = logging.getLogger(__name__)


# This is used to ensure that metrics with a timestamp older than
# RECENT_POINT_THRESHOLD_DEFAULT seconds (or the value passed in to
# the MetricsAggregator constructor) get discarded rather than being
# input into the incorrect bucket. Currently, the MetricsAggregator
# does not support submitting values for the past, and all values get
# submitted for the timestamp passed into the flush() function.
# The MetricsBucketAggregator uses times that are aligned to "buckets"
# that are the length of the interval that is passed into the
# MetricsBucketAggregator constructor.
RECENT_POINT_THRESHOLD_DEFAULT = 3600


class Aggregator(object):
    """
    Abstract metric aggregator class.
    """

    def __init__(self, hostname, interval=1.0, expiry_seconds=300, recent_point_threshold=None):
        self.events = []
        self.total_count = 0
        self.count = 0
        self.event_count = 0
        self.hostname = hostname
        self.expiry_seconds = expiry_seconds
        self.interval = float(interval)

        recent_point_threshold = recent_point_threshold or RECENT_POINT_THRESHOLD_DEFAULT
        self.recent_point_threshold = int(recent_point_threshold)
        self.num_discarded_old_points = 0

    @staticmethod
    def formatter(metric, value, timestamp, dimensions, hostname, device_name=None, metric_type=None, interval=None):
        """ Formats metrics, put them into a Measurement class
            (metric, timestamp, value, {"dimensions": {"name1": "value1", "name2": "value2"}, ...})
            dimensions should be a dictionary
        """
        if dimensions is None:
            dimensions = {}
        if hostname:
            dimensions['hostname'] = hostname
        if device_name:
            dimensions['device_name'] = device_name

        return Measurement(metric, int(timestamp), value, dimensions)

    def packets_per_second(self, interval):
        if interval == 0:
            return 0
        return round(float(self.count)/interval, 2)

    def submit_metric(self, name, value, mtype, dimensions=None, hostname=None, device_name=None, timestamp=None,
                      sample_rate=1):
        """ Add a metric to be aggregated """
        raise NotImplementedError()

    def event(self, title, text, date_happened=None, alert_type=None, aggregation_key=None, source_type_name=None,
              priority=None, dimensions=None, hostname=None):
        event = {
            'msg_title': title,
            'msg_text': text,
        }
        if date_happened is not None:
            event['timestamp'] = date_happened
        else:
            event['timestamp'] = int(time())
        if alert_type is not None:
            event['alert_type'] = alert_type
        if aggregation_key is not None:
            event['aggregation_key'] = aggregation_key
        if source_type_name is not None:
            event['source_type_name'] = source_type_name
        if priority is not None:
            event['priority'] = priority
        if dimensions is not None:
            event['dimensions'] = dimensions
        if hostname is not None:
            event['host'] = hostname
        else:
            event['host'] = self.hostname

        self.events.append(event)

    def flush(self):
        """ Flush aggreaged metrics """
        raise NotImplementedError()

    def flush_events(self):
        events = self.events
        self.events = []

        self.total_count += self.event_count
        self.event_count = 0

        log.debug("Received %d events since last flush" % len(events))

        return events


class MetricsBucketAggregator(Aggregator):
    """
    A metric aggregator class.
    """

    def __init__(self, hostname, interval=1.0, expiry_seconds=300, recent_point_threshold=None):
        super(MetricsBucketAggregator, self).__init__(hostname, interval, expiry_seconds, recent_point_threshold)
        self.metric_by_bucket = {}
        self.last_sample_time_by_context = {}
        self.current_bucket = None
        self.current_mbc = None
        self.last_flush_cutoff_time = 0
        self.metric_type_to_class = {
            'g': BucketGauge,
            'c': Counter,
            'h': Histogram,
            'ms': Histogram,
            's': Set,
        }

    def calculate_bucket_start(self, timestamp):
        return timestamp - (timestamp % self.interval)

    def submit_metric(self, name, value, mtype, dimensions=None, hostname=None,
                      device_name=None, timestamp=None, sample_rate=1):
        # Avoid calling extra functions to dedupe dimensions if there are none
        # Note: if you change the way that context is created, please also change create_empty_metrics,
        #  which counts on this order
        if dimensions is None:
            context = (name, (), hostname, device_name)
        else:
            context = (name, dimensions.items(), hostname, device_name)

        cur_time = time()
        # Check to make sure that the timestamp that is passed in (if any) is not older than
        #  recent_point_threshold.  If so, discard the point.
        if timestamp is not None and cur_time - int(timestamp) > self.recent_point_threshold:
            log.debug("Discarding %s - ts = %s , current ts = %s " % (name, timestamp, cur_time))
            self.num_discarded_old_points += 1
        else:
            timestamp = timestamp or cur_time
{'fraction_non_alphanumeric': 0.08423554264589385, 'fraction_numerical': 0.0052812252442566675, 'mean_word_length': 2.661617789236223, 'pattern_counts': {'":': 6, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1183015', 'n_tokens_mistral': 3159, 'n_tokens_neox': 3034, 'n_words': 946}