metrics.py

# Unless explicitly stated otherwise all files in this repository are licensed under the BSD-3-Clause License.
# This product includes software developed at Datadog (https://www.datadoghq.com/).
# Copyright 2015-Present Datadog, Inc
"""
Metric roll-up classes.
"""
from collections import defaultdict
import random
import itertools
import threading

from datadog.util.compat import iternext
from datadog.threadstats.constants import MetricType


class Metric(object):
    """
    A base metric class that accepts points, slices them into time intervals
    and performs roll-ups within those intervals.
    """

    def add_point(self, value):
        """ Add a point to the given metric. """
        raise NotImplementedError()

    def flush(self, timestamp, interval):
        """ Flush all metrics up to the given timestamp. """
        raise NotImplementedError()


class Set(Metric):
    """ A set metric. """

    stats_tag = "g"

    def __init__(self, name, tags, host):
        self.name = name
        self.tags = tags
        self.host = host
        self.set = set()

    def add_point(self, value):
        self.set.add(value)

    def flush(self, timestamp, interval):
        return [(timestamp, len(self.set), self.name, self.tags, self.host, MetricType.Gauge, interval)]


class Gauge(Metric):
    """ A gauge metric. """

    stats_tag = "g"

    def __init__(self, name, tags, host):
        self.name = name
        self.tags = tags
        self.host = host
        self.value = None

    def add_point(self, value):
        self.value = value

    def flush(self, timestamp, interval):
        return [(timestamp, self.value, self.name, self.tags, self.host, MetricType.Gauge, interval)]


class Counter(Metric):
    """ A metric that tracks a counter value. """

    stats_tag = "c"

    def __init__(self, name, tags, host):
        self.name = name
        self.tags = tags
        self.host = host
        self.count = []

    def add_point(self, value):
        self.count.append(value)

    def flush(self, timestamp, interval):
        count = sum(self.count, 0)
        return [(timestamp, count / float(interval), self.name, self.tags, self.host, MetricType.Rate, interval)]


class Distribution(Metric):
    """ A distribution metric. """

    stats_tag = "d"

    def __init__(self, name, tags, host):
        self.name = name
        self.tags = tags
        self.host = host
        self.value = []

    def add_point(self, value):
        self.value.append(value)

    def flush(self, timestamp, interval):
        return [(timestamp, self.value, self.name, self.tags, self.host, MetricType.Distribution, interval)]


class Histogram(Metric):
    """ A histogram metric. """

    stats_tag = "h"

    def __init__(self, name, tags, host):
        self.name = name
        self.tags = tags
        self.host = host
        self.max = float("-inf")
        self.min = float("inf")
        self.sum = []
        self.iter_counter = itertools.count()
        self.count = iternext(self.iter_counter)
        self.sample_size = 1000
        self.samples = []
        self.percentiles = [0.75, 0.85, 0.95, 0.99]

    def add_point(self, value):
        self.max = self.max if self.max > value else value
        self.min = self.min if self.min < value else value
        self.sum.append(value)
        if self.count < self.sample_size:
            self.samples.append(value)
        else:
            self.samples[random.randrange(0, self.sample_size)] = value
        self.count = iternext(self.iter_counter)

    def flush(self, timestamp, interval):
        if not self.count:
            return []
        metrics = [
            (timestamp, self.min, "%s.min" % self.name, self.tags, self.host, MetricType.Gauge, interval),
            (timestamp, self.max, "%s.max" % self.name, self.tags, self.host, MetricType.Gauge, interval),
            (
                timestamp,
                self.count / float(interval),
                "%s.count" % self.name,
                self.tags,
                self.host,
                MetricType.Rate,
                interval,
            ),
            (timestamp, self.average(), "%s.avg" % self.name, self.tags, self.host, MetricType.Gauge, interval),
        ]
        length = len(self.samples)
        self.samples.sort()
        for p in self.percentiles:
            val = self.samples[int(round(p * length - 1))]
            name = "%s.%spercentile" % (self.name, int(p * 100))
            metrics.append((timestamp, val, name, self.tags, self.host, MetricType.Gauge, interval))
        return metrics

    def average(self):
        sum_metrics = sum(self.sum, 0)
        return float(sum_metrics) / self.count


class Timing(Histogram):
    """
    A timing metric.
    Inherit from Histogram to workaround and support it in API mode
    """

    stats_tag = "ms"


class MetricsAggregator(object):
    """
    A small class to handle the roll-ups of multiple metrics at once.
    """

    def __init__(self, roll_up_interval=10):
        self._lock = threading.RLock()
        self._metrics = defaultdict(lambda: {})
        self._roll_up_interval = roll_up_interval

    def add_point(self, metric, tags, timestamp, value, metric_class, sample_rate=1, host=None):
        # The sample rate is currently ignored for in process stuff
        interval = timestamp - timestamp % self._roll_up_interval
        key = (metric, host, tuple(sorted(tags)) if tags else None)
        with self._lock:
            if key not in self._metrics[interval]:
                self._metrics[interval][key] = metric_class(metric, tags, host)
            self._metrics[interval][key].add_point(value)

    def flush(self, timestamp):
        """ Flush all metrics up to the given timestamp. """
        if timestamp == float("inf"):
            interval = float("inf")
        else:
            interval = timestamp - timestamp % self._roll_up_interval

        with self._lock:
            past_intervals = [i for i in self._metrics.keys() if i < interval]
            metrics = []
            for i in past_intervals:
                for m in list(self._metrics.pop(i).values()):
                    metrics += m.flush(i, self._roll_up_interval)
        return metrics