diff options
Diffstat (limited to 'kafka/metrics')
24 files changed, 1328 insertions, 0 deletions
diff --git a/kafka/metrics/__init__.py b/kafka/metrics/__init__.py new file mode 100644 index 0000000..dd22f53 --- /dev/null +++ b/kafka/metrics/__init__.py @@ -0,0 +1,13 @@ +from .compound_stat import NamedMeasurable +from .dict_reporter import DictReporter +from .kafka_metric import KafkaMetric +from .measurable import AnonMeasurable +from .metric_config import MetricConfig +from .metric_name import MetricName +from .metrics import Metrics +from .quota import Quota + +__all__ = [ + 'AnonMeasurable', 'DictReporter', 'KafkaMetric', 'MetricConfig', + 'MetricName', 'Metrics', 'NamedMeasurable', 'Quota' +] diff --git a/kafka/metrics/compound_stat.py b/kafka/metrics/compound_stat.py new file mode 100644 index 0000000..09bc24a --- /dev/null +++ b/kafka/metrics/compound_stat.py @@ -0,0 +1,32 @@ +import abc + +from kafka.metrics.stat import AbstractStat + + +class AbstractCompoundStat(AbstractStat): + """ + A compound stat is a stat where a single measurement and associated + data structure feeds many metrics. This is the example for a + histogram which has many associated percentiles. + """ + __metaclass__ = abc.ABCMeta + + def stats(self): + """ + Return list of NamedMeasurable + """ + raise NotImplementedError + + +class NamedMeasurable(object): + def __init__(self, metric_name, measurable_stat): + self._name = metric_name + self._stat = measurable_stat + + @property + def name(self): + return self._name + + @property + def stat(self): + return self._stat diff --git a/kafka/metrics/dict_reporter.py b/kafka/metrics/dict_reporter.py new file mode 100644 index 0000000..49af604 --- /dev/null +++ b/kafka/metrics/dict_reporter.py @@ -0,0 +1,81 @@ +import logging +import threading + +from kafka.metrics.metrics_reporter import AbstractMetricsReporter + +logger = logging.getLogger(__name__) + + +class DictReporter(AbstractMetricsReporter): + """A basic dictionary based metrics reporter. + + Store all metrics in a two level dictionary of category > name > metric. + """ + def __init__(self, prefix=''): + self._lock = threading.Lock() + self._prefix = prefix if prefix else '' # never allow None + self._store = {} + + def snapshot(self): + """ + Return a nested dictionary snapshot of all metrics and their + values at this time. Example: + { + 'category': { + 'metric1_name': 42.0, + 'metric2_name': 'foo' + } + } + """ + return dict((category, dict((name, metric.value()) + for name, metric in list(metrics.items()))) + for category, metrics in + list(self._store.items())) + + def init(self, metrics): + for metric in metrics: + self.metric_change(metric) + + def metric_change(self, metric): + with self._lock: + category = self.get_category(metric) + if category not in self._store: + self._store[category] = {} + self._store[category][metric.metric_name.name] = metric + + def metric_removal(self, metric): + with self._lock: + category = self.get_category(metric) + metrics = self._store.get(category, {}) + removed = metrics.pop(metric.metric_name.name, None) + if not metrics: + self._store.pop(category, None) + return removed + + def get_category(self, metric): + """ + Return a string category for the metric. + + The category is made up of this reporter's prefix and the + metric's group and tags. + + Examples: + prefix = 'foo', group = 'bar', tags = {'a': 1, 'b': 2} + returns: 'foo.bar.a=1,b=2' + + prefix = 'foo', group = 'bar', tags = None + returns: 'foo.bar' + + prefix = None, group = 'bar', tags = None + returns: 'bar' + """ + tags = ','.join('%s=%s' % (k, v) for k, v in + sorted(metric.metric_name.tags.items())) + return '.'.join(x for x in + [self._prefix, metric.metric_name.group, tags] if x) + + def configure(self, configs): + pass + + def close(self): + pass diff --git a/kafka/metrics/kafka_metric.py b/kafka/metrics/kafka_metric.py new file mode 100644 index 0000000..75d32a4 --- /dev/null +++ b/kafka/metrics/kafka_metric.py @@ -0,0 +1,34 @@ +import time + + +class KafkaMetric(object): + # NOTE java constructor takes a lock instance + def __init__(self, metric_name, measurable, config): + if not metric_name: + raise ValueError('metric_name must be non-empty') + if not measurable: + raise ValueError('measurable must be non-empty') + self._metric_name = metric_name + self._measurable = measurable + self._config = config + + @property + def metric_name(self): + return self._metric_name + + @property + def measurable(self): + return self._measurable + + @property + def config(self): + return self._config + + @config.setter + def config(self, config): + self._config = config + + def value(self, time_ms=None): + if time_ms is None: + time_ms = time.time() * 1000 + return self.measurable.measure(self.config, time_ms) diff --git a/kafka/metrics/measurable.py b/kafka/metrics/measurable.py new file mode 100644 index 0000000..ef096f3 --- /dev/null +++ b/kafka/metrics/measurable.py @@ -0,0 +1,27 @@ +import abc + + +class AbstractMeasurable(object): + """A measurable quantity that can be registered as a metric""" + @abc.abstractmethod + def measure(self, config, now): + """ + Measure this quantity and return the result + + Arguments: + config (MetricConfig): The configuration for this metric + now (int): The POSIX time in milliseconds the measurement + is being taken + + Returns: + The measured value + """ + raise NotImplementedError + + +class AnonMeasurable(AbstractMeasurable): + def __init__(self, measure_fn): + self._measure_fn = measure_fn + + def measure(self, config, now): + return float(self._measure_fn(config, now)) diff --git a/kafka/metrics/measurable_stat.py b/kafka/metrics/measurable_stat.py new file mode 100644 index 0000000..dba887d --- /dev/null +++ b/kafka/metrics/measurable_stat.py @@ -0,0 +1,14 @@ +import abc + +from kafka.metrics.measurable import AbstractMeasurable +from kafka.metrics.stat import AbstractStat + + +class AbstractMeasurableStat(AbstractStat, AbstractMeasurable): + """ + An AbstractMeasurableStat is an AbstractStat that is also + an AbstractMeasurable (i.e. can produce a single floating point value). + This is the interface used for most of the simple statistics such + as Avg, Max, Count, etc. + """ + __metaclass__ = abc.ABCMeta diff --git a/kafka/metrics/metric_config.py b/kafka/metrics/metric_config.py new file mode 100644 index 0000000..e30c477 --- /dev/null +++ b/kafka/metrics/metric_config.py @@ -0,0 +1,31 @@ +import sys + + +class MetricConfig(object): + """Configuration values for metrics""" + def __init__(self, quota=None, samples=2, event_window=sys.maxsize, + time_window_ms=30 * 1000, tags=None): + """ + Arguments: + quota (Quota, optional): Upper or lower bound of a value. + samples (int, optional): Max number of samples kept per metric. + event_window (int, optional): Max number of values per sample. + time_window_ms (int, optional): Max age of an individual sample. + tags (dict of {str: str}, optional): Tags for each metric. + """ + self.quota = quota + self._samples = samples + self.event_window = event_window + self.time_window_ms = time_window_ms + # tags should be OrderedDict (not supported in py26) + self.tags = tags if tags else {} + + @property + def samples(self): + return self._samples + + @samples.setter + def samples(self, value): + if value < 1: + raise ValueError('The number of samples must be at least 1.') + self._samples = value diff --git a/kafka/metrics/metric_name.py b/kafka/metrics/metric_name.py new file mode 100644 index 0000000..02068f0 --- /dev/null +++ b/kafka/metrics/metric_name.py @@ -0,0 +1,104 @@ +import copy + + +class MetricName(object): + """ + This class encapsulates a metric's name, logical group and its + related attributes (tags). + + group, tags parameters can be used to create unique metric names. + e.g. domainName:type=group,key1=val1,key2=val2 + + Usage looks something like this: + + # set up metrics: + metric_tags = {'client-id': 'producer-1', 'topic': 'topic'} + metric_config = MetricConfig(tags=metric_tags) + + # metrics is the global repository of metrics and sensors + metrics = Metrics(metric_config) + + sensor = metrics.sensor('message-sizes') + metric_name = metrics.metric_name('message-size-avg', + 'producer-metrics', + 'average message size') + sensor.add(metric_name, Avg()) + + metric_name = metrics.metric_name('message-size-max', + sensor.add(metric_name, Max()) + + tags = {'client-id': 'my-client', 'topic': 'my-topic'} + metric_name = metrics.metric_name('message-size-min', + 'producer-metrics', + 'message minimum size', tags) + sensor.add(metric_name, Min()) + + # as messages are sent we record the sizes + sensor.record(message_size) + """ + + def __init__(self, name, group, description=None, tags=None): + """ + Arguments: + name (str): The name of the metric. + group (str): The logical group name of the metrics to which this + metric belongs. + description (str, optional): A human-readable description to + include in the metric. + tags (dict, optional): Additional key/val attributes of the metric. + """ + if not (name and group): + raise Exception('name and group must be non-empty.') + if tags is not None and not isinstance(tags, dict): + raise Exception('tags must be a dict if present.') + + self._name = name + self._group = group + self._description = description + self._tags = copy.copy(tags) + self._hash = 0 + + @property + def name(self): + return self._name + + @property + def group(self): + return self._group + + @property + def description(self): + return self._description + + @property + def tags(self): + return copy.copy(self._tags) + + def __hash__(self): + if self._hash != 0: + return self._hash + prime = 31 + result = 1 + result = prime * result + hash(self.group) + result = prime * result + hash(self.name) + tags_hash = hash(frozenset(self.tags.items())) if self.tags else 0 + result = prime * result + tags_hash + self._hash = result + return result + + def __eq__(self, other): + if self is other: + return True + if other is None: + return False + return (type(self) == type(other) and + self.group == other.group and + self.name == other.name and + self.tags == other.tags) + + def __ne__(self, other): + return not self.__eq__(other) + + def __str__(self): + return 'MetricName(name=%s, group=%s, description=%s, tags=%s)' % ( + self.name, self.group, self.description, self.tags) diff --git a/kafka/metrics/metrics.py b/kafka/metrics/metrics.py new file mode 100644 index 0000000..d02f48d --- /dev/null +++ b/kafka/metrics/metrics.py @@ -0,0 +1,257 @@ +import logging +import sys +import time +import threading + +from kafka.metrics import AnonMeasurable, KafkaMetric, MetricConfig, MetricName +from kafka.metrics.stats import Sensor + +logger = logging.getLogger(__name__) + + +class Metrics(object): + """ + A registry of sensors and metrics. + + A metric is a named, numerical measurement. A sensor is a handle to + record numerical measurements as they occur. Each Sensor has zero or + more associated metrics. For example a Sensor might represent message + sizes and we might associate with this sensor a metric for the average, + maximum, or other statistics computed off the sequence of message sizes + that are recorded by the sensor. + + Usage looks something like this: + # set up metrics: + metrics = Metrics() # the global repository of metrics and sensors + sensor = metrics.sensor('message-sizes') + metric_name = MetricName('message-size-avg', 'producer-metrics') + sensor.add(metric_name, Avg()) + metric_name = MetricName('message-size-max', 'producer-metrics') + sensor.add(metric_name, Max()) + + # as messages are sent we record the sizes + sensor.record(message_size); + """ + def __init__(self, default_config=None, reporters=None, + enable_expiration=False): + """ + Create a metrics repository with a default config, given metric + reporters and the ability to expire eligible sensors + + Arguments: + default_config (MetricConfig, optional): The default config + reporters (list of AbstractMetricsReporter, optional): + The metrics reporters + enable_expiration (bool, optional): true if the metrics instance + can garbage collect inactive sensors, false otherwise + """ + self._lock = threading.RLock() + self._config = default_config or MetricConfig() + self._sensors = {} + self._metrics = {} + self._children_sensors = {} + self._reporters = reporters or [] + for reporter in self._reporters: + reporter.init([]) + + if enable_expiration: + def expire_loop(): + while True: + # delay 30 seconds + time.sleep(30) + self.ExpireSensorTask.run(self) + metrics_scheduler = threading.Thread(target=expire_loop) + # Creating a daemon thread to not block shutdown + metrics_scheduler.daemon = True + metrics_scheduler.start() + + self.add_metric(self.metric_name('count', 'kafka-metrics-count', + 'total number of registered metrics'), + AnonMeasurable(lambda config, now: len(self._metrics))) + + @property + def config(self): + return self._config + + @property + def metrics(self): + """ + Get all the metrics currently maintained and indexed by metricName + """ + return self._metrics + + def metric_name(self, name, group, description='', tags=None): + """ + Create a MetricName with the given name, group, description and tags, + plus default tags specified in the metric configuration. + Tag in tags takes precedence if the same tag key is specified in + the default metric configuration. + + Arguments: + name (str): The name of the metric + group (str): logical group name of the metrics to which this + metric belongs + description (str, optional): A human-readable description to + include in the metric + tags (dict, optionals): additional key/value attributes of + the metric + """ + combined_tags = dict(self.config.tags) + combined_tags.update(tags or {}) + return MetricName(name, group, description, combined_tags) + + def get_sensor(self, name): + """ + Get the sensor with the given name if it exists + + Arguments: + name (str): The name of the sensor + + Returns: + Sensor: The sensor or None if no such sensor exists + """ + if not name: + raise ValueError('name must be non-empty') + return self._sensors.get(name, None) + + def sensor(self, name, config=None, + inactive_sensor_expiration_time_seconds=sys.maxsize, + parents=None): + """ + Get or create a sensor with the given unique name and zero or + more parent sensors. All parent sensors will receive every value + recorded with this sensor. + + Arguments: + name (str): The name of the sensor + config (MetricConfig, optional): A default configuration to use + for this sensor for metrics that don't have their own config + inactive_sensor_expiration_time_seconds (int, optional): + If no value if recorded on the Sensor for this duration of + time, it is eligible for removal + parents (list of Sensor): The parent sensors + + Returns: + Sensor: The sensor that is created + """ + sensor = self.get_sensor(name) + if sensor: + return sensor + + with self._lock: + sensor = self.get_sensor(name) + if not sensor: + sensor = Sensor(self, name, parents, config or self.config, + inactive_sensor_expiration_time_seconds) + self._sensors[name] = sensor + if parents: + for parent in parents: + children = self._children_sensors.get(parent) + if not children: + children = [] + self._children_sensors[parent] = children + children.append(sensor) + logger.debug('Added sensor with name %s', name) + return sensor + + def remove_sensor(self, name): + """ + Remove a sensor (if it exists), associated metrics and its children. + + Arguments: + name (str): The name of the sensor to be removed + """ + sensor = self._sensors.get(name) + if sensor: + child_sensors = None + with sensor._lock: + with self._lock: + val = self._sensors.pop(name, None) + if val and val == sensor: + for metric in sensor.metrics: + self.remove_metric(metric.metric_name) + logger.debug('Removed sensor with name %s', name) + child_sensors = self._children_sensors.pop(sensor, None) + if child_sensors: + for child_sensor in child_sensors: + self.remove_sensor(child_sensor.name) + + def add_metric(self, metric_name, measurable, config=None): + """ + Add a metric to monitor an object that implements measurable. + This metric won't be associated with any sensor. + This is a way to expose existing values as metrics. + + Arguments: + metricName (MetricName): The name of the metric + measurable (AbstractMeasurable): The measurable that will be + measured by this metric + config (MetricConfig, optional): The configuration to use when + measuring this measurable + """ + # NOTE there was a lock here, but i don't think it's needed + metric = KafkaMetric(metric_name, measurable, config or self.config) + self.register_metric(metric) + + def remove_metric(self, metric_name): + """ + Remove a metric if it exists and return it. Return None otherwise. + If a metric is removed, `metric_removal` will be invoked + for each reporter. + + Arguments: + metric_name (MetricName): The name of the metric + + Returns: + KafkaMetric: the removed `KafkaMetric` or None if no such + metric exists + """ + with self._lock: + metric = self._metrics.pop(metric_name, None) + if metric: + for reporter in self._reporters: + reporter.metric_removal(metric) + return metric + + def add_reporter(self, reporter): + """Add a MetricReporter""" + with self._lock: + reporter.init(list(self.metrics.values())) + self._reporters.append(reporter) + + def register_metric(self, metric): + with self._lock: + if metric.metric_name in self.metrics: + raise ValueError('A metric named "%s" already exists, cannot' + ' register another one.' % metric.metric_name) + self.metrics[metric.metric_name] = metric + for reporter in self._reporters: + reporter.metric_change(metric) + + class ExpireSensorTask(object): + """ + This iterates over every Sensor and triggers a remove_sensor + if it has expired. Package private for testing + """ + @staticmethod + def run(metrics): + items = list(metrics._sensors.items()) + for name, sensor in items: + # remove_sensor also locks the sensor object. This is fine + # because synchronized is reentrant. There is however a minor + # race condition here. Assume we have a parent sensor P and + # child sensor C. Calling record on C would cause a record on + # P as well. So expiration time for P == expiration time for C. + # If the record on P happens via C just after P is removed, + # that will cause C to also get removed. Since the expiration + # time is typically high it is not expected to be a significant + # concern and thus not necessary to optimize + with sensor._lock: + if sensor.has_expired(): + logger.debug('Removing expired sensor %s', name) + metrics.remove_sensor(name) + + def close(self): + """Close this metrics repository.""" + for reporter in self._reporters: + reporter.close() diff --git a/kafka/metrics/metrics_reporter.py b/kafka/metrics/metrics_reporter.py new file mode 100644 index 0000000..b48ad0b --- /dev/null +++ b/kafka/metrics/metrics_reporter.py @@ -0,0 +1,55 @@ +import abc + + +class AbstractMetricsReporter(object): + """ + An abstract class to allow things to listen as new metrics + are created so they can be reported. + """ + __metaclass__ = abc.ABCMeta + + @abc.abstractmethod + def init(self, metrics): + """ + This is called when the reporter is first registered + to initially register all existing metrics + + Arguments: + metrics (list of KafkaMetric): All currently existing metrics + """ + raise NotImplementedError + + @abc.abstractmethod + def metric_change(self, metric): + """ + This is called whenever a metric is updated or added + + Arguments: + metric (KafkaMetric) + """ + raise NotImplementedError + + @abc.abstractmethod + def metric_removal(self, metric): + """ + This is called whenever a metric is removed + + Arguments: + metric (KafkaMetric) + """ + raise NotImplementedError + + @abc.abstractmethod + def configure(self, configs): + """ + Configure this class with the given key-value pairs + + Arguments: + configs (dict of {str, ?}) + """ + raise NotImplementedError + + @abc.abstractmethod + def close(self): + """Called when the metrics repository is closed.""" + raise NotImplementedError diff --git a/kafka/metrics/quota.py b/kafka/metrics/quota.py new file mode 100644 index 0000000..0410e37 --- /dev/null +++ b/kafka/metrics/quota.py @@ -0,0 +1,39 @@ +class Quota(object): + """An upper or lower bound for metrics""" + def __init__(self, bound, is_upper): + self._bound = bound + self._upper = is_upper + + @staticmethod + def upper_bound(upper_bound): + return Quota(upper_bound, True) + + @staticmethod + def lower_bound(lower_bound): + return Quota(lower_bound, False) + + def is_upper_bound(self): + return self._upper + + @property + def bound(self): + return self._bound + + def is_acceptable(self, value): + return ((self.is_upper_bound() and value <= self.bound) or + (not self.is_upper_bound() and value >= self.bound)) + + def __hash__(self): + prime = 31 + result = prime + self.bound + return prime * result + self.is_upper_bound() + + def __eq__(self, other): + if self is other: + return True + return (type(self) == type(other) and + self.bound == other.bound and + self.is_upper_bound() == other.is_upper_bound()) + + def __ne__(self, other): + return not self.__eq__(other) diff --git a/kafka/metrics/stat.py b/kafka/metrics/stat.py new file mode 100644 index 0000000..c10f3ce --- /dev/null +++ b/kafka/metrics/stat.py @@ -0,0 +1,21 @@ +import abc + + +class AbstractStat(object): + """ + An AbstractStat is a quantity such as average, max, etc that is computed + off the stream of updates to a sensor + """ + __metaclass__ = abc.ABCMeta + + @abc.abstractmethod + def record(self, config, value, time_ms): + """ + Record the given value + + Arguments: + config (MetricConfig): The configuration to use for this metric + value (float): The value to record + timeMs (int): The POSIX time in milliseconds this value occurred + """ + raise NotImplementedError diff --git a/kafka/metrics/stats/__init__.py b/kafka/metrics/stats/__init__.py new file mode 100644 index 0000000..15eafd9 --- /dev/null +++ b/kafka/metrics/stats/__init__.py @@ -0,0 +1,15 @@ +from .avg import Avg +from .count import Count +from .histogram import Histogram +from .max_stat import Max +from .min_stat import Min +from .percentile import Percentile +from .percentiles import Percentiles +from .rate import Rate +from .sensor import Sensor +from .total import Total + +__all__ = [ + 'Avg', 'Count', 'Histogram', 'Max', 'Min', 'Percentile', 'Percentiles', + 'Rate', 'Sensor', 'Total' +] diff --git a/kafka/metrics/stats/avg.py b/kafka/metrics/stats/avg.py new file mode 100644 index 0000000..4d0be0a --- /dev/null +++ b/kafka/metrics/stats/avg.py @@ -0,0 +1,22 @@ +from kafka.metrics.stats.sampled_stat import AbstractSampledStat + + +class Avg(AbstractSampledStat): + """ + An AbstractSampledStat that maintains a simple average over its samples. + """ + def __init__(self): + super(Avg, self).__init__(0.0) + + def update(self, sample, config, value, now): + sample.value += value + + def combine(self, samples, config, now): + total_sum = 0 + total_count = 0 + for sample in samples: + total_sum += sample.value + total_count += sample.event_count + if not total_count: + return 0 + return float(total_sum) / total_count diff --git a/kafka/metrics/stats/count.py b/kafka/metrics/stats/count.py new file mode 100644 index 0000000..183e4f2 --- /dev/null +++ b/kafka/metrics/stats/count.py @@ -0,0 +1,15 @@ +from kafka.metrics.stats.sampled_stat import AbstractSampledStat + + +class Count(AbstractSampledStat): + """ + An AbstractSampledStat that maintains a simple count of what it has seen. + """ + def __init__(self): + super(Count, self).__init__(0.0) + + def update(self, sample, config, value, now): + sample.value += 1.0 + + def combine(self, samples, config, now): + return float(sum(sample.value for sample in samples)) diff --git a/kafka/metrics/stats/histogram.py b/kafka/metrics/stats/histogram.py new file mode 100644 index 0000000..42aacdb --- /dev/null +++ b/kafka/metrics/stats/histogram.py @@ -0,0 +1,93 @@ +import math + + +class Histogram(object): + def __init__(self, bin_scheme): + self._hist = [0.0] * bin_scheme.bins + self._count = 0.0 + self._bin_scheme = bin_scheme + + def record(self, value): + self._hist[self._bin_scheme.to_bin(value)] += 1.0 + self._count += 1.0 + + def value(self, quantile): + if self._count == 0.0: + return float('NaN') + _sum = 0.0 + quant = float(quantile) + for i, value in enumerate(self._hist[:-1]): + _sum += value + if _sum / self._count > quant: + return self._bin_scheme.from_bin(i) + return float('inf') + + @property + def counts(self): + return self._hist + + def clear(self): + for i in range(self._hist): + self._hist[i] = 0.0 + self._count = 0 + + def __str__(self): + values = ['%.10f:%.0f' % (self._bin_scheme.from_bin(i), value) for + i, value in enumerate(self._hist[:-1])] + values.append('%s:%s' % (float('inf'), self._hist[-1])) + return '{%s}' % ','.join(values) + + class ConstantBinScheme(object): + def __init__(self, bins, min_val, max_val): + if bins < 2: + raise ValueError('Must have at least 2 bins.') + self._min = float(min_val) + self._max = float(max_val) + self._bins = int(bins) + self._bucket_width = (max_val - min_val) / (bins - 2) + + @property + def bins(self): + return self._bins + + def from_bin(self, b): + if b == 0: + return float('-inf') + elif b == self._bins - 1: + return float('inf') + else: + return self._min + (b - 1) * self._bucket_width + + def to_bin(self, x): + if x < self._min: + return 0 + elif x > self._max: + return self._bins - 1 + else: + return int(((x - self._min) / self._bucket_width) + 1) + + class LinearBinScheme(object): + def __init__(self, num_bins, max_val): + self._bins = num_bins + self._max = max_val + self._scale = max_val / (num_bins * (num_bins - 1) / 2) + + @property + def bins(self): + return self._bins + + def from_bin(self, b): + if b == self._bins - 1: + return float('inf') + else: + unscaled = (b * (b + 1.0)) / 2.0 + return unscaled * self._scale + + def to_bin(self, x): + if x < 0.0: + raise ValueError('Values less than 0.0 not accepted.') + elif x > self._max: + return self._bins - 1 + else: + scaled = x / self._scale + return int(-0.5 + math.sqrt(2.0 * scaled + 0.25)) diff --git a/kafka/metrics/stats/max_stat.py b/kafka/metrics/stats/max_stat.py new file mode 100644 index 0000000..8df54d3 --- /dev/null +++ b/kafka/metrics/stats/max_stat.py @@ -0,0 +1,15 @@ +from kafka.metrics.stats.sampled_stat import AbstractSampledStat + + +class Max(AbstractSampledStat): + """An AbstractSampledStat that gives the max over its samples.""" + def __init__(self): + super(Max, self).__init__(float('-inf')) + + def update(self, sample, config, value, now): + sample.value = max(sample.value, value) + + def combine(self, samples, config, now): + if not samples: + return float('-inf') + return float(max(sample.value for sample in samples)) diff --git a/kafka/metrics/stats/min_stat.py b/kafka/metrics/stats/min_stat.py new file mode 100644 index 0000000..a57c2dd --- /dev/null +++ b/kafka/metrics/stats/min_stat.py @@ -0,0 +1,17 @@ +import sys + +from kafka.metrics.stats.sampled_stat import AbstractSampledStat + + +class Min(AbstractSampledStat): + """An AbstractSampledStat that gives the min over its samples.""" + def __init__(self): + super(Min, self).__init__(float(sys.maxsize)) + + def update(self, sample, config, value, now): + sample.value = min(sample.value, value) + + def combine(self, samples, config, now): + if not samples: + return float(sys.maxsize) + return float(min(sample.value for sample in samples)) diff --git a/kafka/metrics/stats/percentile.py b/kafka/metrics/stats/percentile.py new file mode 100644 index 0000000..723b9e6 --- /dev/null +++ b/kafka/metrics/stats/percentile.py @@ -0,0 +1,12 @@ +class Percentile(object): + def __init__(self, metric_name, percentile): + self._metric_name = metric_name + self._percentile = float(percentile) + + @property + def name(self): + return self._metric_name + + @property + def percentile(self): + return self._percentile diff --git a/kafka/metrics/stats/percentiles.py b/kafka/metrics/stats/percentiles.py new file mode 100644 index 0000000..84e7160 --- /dev/null +++ b/kafka/metrics/stats/percentiles.py @@ -0,0 +1,72 @@ +from kafka.metrics import AnonMeasurable, NamedMeasurable +from kafka.metrics.compound_stat import AbstractCompoundStat +from kafka.metrics.stats import Histogram +from kafka.metrics.stats.sampled_stat import AbstractSampledStat + + +class BucketSizing(object): + CONSTANT = 0 + LINEAR = 1 + + +class Percentiles(AbstractSampledStat, AbstractCompoundStat): + """A compound stat that reports one or more percentiles""" + def __init__(self, size_in_bytes, bucketing, max_val, min_val=0.0, + percentiles=None): + super(Percentiles, self).__init__(0.0) + self._percentiles = percentiles or [] + self._buckets = int(size_in_bytes / 4) + if bucketing == BucketSizing.CONSTANT: + self._bin_scheme = Histogram.ConstantBinScheme(self._buckets, + min_val, max_val) + elif bucketing == BucketSizing.LINEAR: + if min_val != 0.0: + raise ValueError('Linear bucket sizing requires min_val' + ' to be 0.0.') + self.bin_scheme = Histogram.LinearBinScheme(self._buckets, max_val) + else: + ValueError('Unknown bucket type: %s' % bucketing) + + def stats(self): + measurables = [] + + def make_measure_fn(pct): + return lambda config, now: self.value(config, now, + pct / 100.0) + + for percentile in self._percentiles: + measure_fn = make_measure_fn(percentile.percentile) + stat = NamedMeasurable(percentile.name, AnonMeasurable(measure_fn)) + measurables.append(stat) + return measurables + + def value(self, config, now, quantile): + self.purge_obsolete_samples(config, now) + count = sum(sample.event_count for sample in self._samples) + if count == 0.0: + return float('NaN') + sum_val = 0.0 + quant = float(quantile) + for b in range(self._buckets): + for sample in self._samples: + assert type(sample) is self.HistogramSample + hist = sample.histogram.counts + sum_val += hist[b] + if sum_val / count > quant: + return self._bin_scheme.from_bin(b) + return float('inf') + + def combine(self, samples, config, now): + return self.value(config, now, 0.5) + + def new_sample(self, time_ms): + return Percentiles.HistogramSample(self._bin_scheme, time_ms) + + def update(self, sample, config, value, time_ms): + assert type(sample) is self.HistogramSample + sample.histogram.record(value) + + class HistogramSample(AbstractSampledStat.Sample): + def __init__(self, scheme, now): + super(Percentiles.HistogramSample, self).__init__(0.0, now) + self.histogram = Histogram(scheme) diff --git a/kafka/metrics/stats/rate.py b/kafka/metrics/stats/rate.py new file mode 100644 index 0000000..3ce2e74 --- /dev/null +++ b/kafka/metrics/stats/rate.py @@ -0,0 +1,115 @@ +from kafka.metrics.measurable_stat import AbstractMeasurableStat +from kafka.metrics.stats.sampled_stat import AbstractSampledStat + + +class TimeUnit(object): + _names = { + 'nanosecond': 0, + 'microsecond': 1, + 'millisecond': 2, + 'second': 3, + 'minute': 4, + 'hour': 5, + 'day': 6, + } + + NANOSECONDS = _names['nanosecond'] + MICROSECONDS = _names['microsecond'] + MILLISECONDS = _names['millisecond'] + SECONDS = _names['second'] + MINUTES = _names['minute'] + HOURS = _names['hour'] + DAYS = _names['day'] + + @staticmethod + def get_name(time_unit): + return TimeUnit._names[time_unit] + + +class Rate(AbstractMeasurableStat): + """ + The rate of the given quantity. By default this is the total observed + over a set of samples from a sampled statistic divided by the elapsed + time over the sample windows. Alternative AbstractSampledStat + implementations can be provided, however, to record the rate of + occurrences (e.g. the count of values measured over the time interval) + or other such values. + """ + def __init__(self, time_unit=TimeUnit.SECONDS, sampled_stat=None): + self._stat = sampled_stat or SampledTotal() + self._unit = time_unit + + def unit_name(self): + return TimeUnit.get_name(self._unit) + + def record(self, config, value, time_ms): + self._stat.record(config, value, time_ms) + + def measure(self, config, now): + value = self._stat.measure(config, now) + return float(value) / self.convert(self.window_size(config, now)) + + def window_size(self, config, now): + # purge old samples before we compute the window size + self._stat.purge_obsolete_samples(config, now) + + """ + Here we check the total amount of time elapsed since the oldest + non-obsolete window. This give the total window_size of the batch + which is the time used for Rate computation. However, there is + an issue if we do not have sufficient data for e.g. if only + 1 second has elapsed in a 30 second window, the measured rate + will be very high. Hence we assume that the elapsed time is + always N-1 complete windows plus whatever fraction of the final + window is complete. + + Note that we could simply count the amount of time elapsed in + the current window and add n-1 windows to get the total time, + but this approach does not account for sleeps. AbstractSampledStat + only creates samples whenever record is called, if no record is + called for a period of time that time is not accounted for in + window_size and produces incorrect results. + """ + total_elapsed_time_ms = now - self._stat.oldest(now).last_window_ms + # Check how many full windows of data we have currently retained + num_full_windows = int(total_elapsed_time_ms / config.time_window_ms) + min_full_windows = config.samples - 1 + + # If the available windows are less than the minimum required, + # add the difference to the totalElapsedTime + if num_full_windows < min_full_windows: + total_elapsed_time_ms += ((min_full_windows - num_full_windows) * + config.time_window_ms) + + return total_elapsed_time_ms + + def convert(self, time_ms): + if self._unit == TimeUnit.NANOSECONDS: + return time_ms * 1000.0 * 1000.0 + elif self._unit == TimeUnit.MICROSECONDS: + return time_ms * 1000.0 + elif self._unit == TimeUnit.MILLISECONDS: + return time_ms + elif self._unit == TimeUnit.SECONDS: + return time_ms / 1000.0 + elif self._unit == TimeUnit.MINUTES: + return time_ms / (60.0 * 1000.0) + elif self._unit == TimeUnit.HOURS: + return time_ms / (60.0 * 60.0 * 1000.0) + elif self._unit == TimeUnit.DAYS: + return time_ms / (24.0 * 60.0 * 60.0 * 1000.0) + else: + raise ValueError('Unknown unit: %s' % self._unit) + + +class SampledTotal(AbstractSampledStat): + def __init__(self, initial_value=None): + if initial_value is not None: + raise ValueError('initial_value cannot be set on SampledTotal') + super(SampledTotal, self).__init__(0.0) + + def update(self, sample, config, value, time_ms): + sample.value += value + + def combine(self, samples, config, now): + return float(sum(sample.value for sample in samples)) diff --git a/kafka/metrics/stats/sampled_stat.py b/kafka/metrics/stats/sampled_stat.py new file mode 100644 index 0000000..ca0db69 --- /dev/null +++ b/kafka/metrics/stats/sampled_stat.py @@ -0,0 +1,99 @@ +import abc + +from kafka.metrics.measurable_stat import AbstractMeasurableStat + + +class AbstractSampledStat(AbstractMeasurableStat): + """ + An AbstractSampledStat records a single scalar value measured over + one or more samples. Each sample is recorded over a configurable + window. The window can be defined by number of events or elapsed + time (or both, if both are given the window is complete when + *either* the event count or elapsed time criterion is met). + + All the samples are combined to produce the measurement. When a + window is complete the oldest sample is cleared and recycled to + begin recording the next sample. + + Subclasses of this class define different statistics measured + using this basic pattern. + """ + __metaclass__ = abc.ABCMeta + + def __init__(self, initial_value): + self._initial_value = initial_value + self._samples = [] + self._current = 0 + + @abc.abstractmethod + def update(self, sample, config, value, time_ms): + raise NotImplementedError + + @abc.abstractmethod + def combine(self, samples, config, now): + raise NotImplementedError + + def record(self, config, value, time_ms): + sample = self.current(time_ms) + if sample.is_complete(time_ms, config): + sample = self._advance(config, time_ms) + self.update(sample, config, float(value), time_ms) + sample.event_count += 1 + + def new_sample(self, time_ms): + return self.Sample(self._initial_value, time_ms) + + def measure(self, config, now): + self.purge_obsolete_samples(config, now) + return float(self.combine(self._samples, config, now)) + + def current(self, time_ms): + if not self._samples: + self._samples.append(self.new_sample(time_ms)) + return self._samples[self._current] + + def oldest(self, now): + if not self._samples: + self._samples.append(self.new_sample(now)) + oldest = self._samples[0] + for sample in self._samples[1:]: + if sample.last_window_ms < oldest.last_window_ms: + oldest = sample + return oldest + + def purge_obsolete_samples(self, config, now): + """ + Timeout any windows that have expired in the absence of any events + """ + expire_age = config.samples * config.time_window_ms + for sample in self._samples: + if now - sample.last_window_ms >= expire_age: + sample.reset(now) + + def _advance(self, config, time_ms): + self._current = (self._current + 1) % config.samples + if self._current >= len(self._samples): + sample = self.new_sample(time_ms) + self._samples.append(sample) + return sample + else: + sample = self.current(time_ms) + sample.reset(time_ms) + return sample + + class Sample(object): + + def __init__(self, initial_value, now): + self.initial_value = initial_value + self.event_count = 0 + self.last_window_ms = now + self.value = initial_value + + def reset(self, now): + self.event_count = 0 + self.last_window_ms = now + self.value = self.initial_value + + def is_complete(self, time_ms, config): + return (time_ms - self.last_window_ms >= config.time_window_ms or + self.event_count >= config.event_window) diff --git a/kafka/metrics/stats/sensor.py b/kafka/metrics/stats/sensor.py new file mode 100644 index 0000000..6878096 --- /dev/null +++ b/kafka/metrics/stats/sensor.py @@ -0,0 +1,132 @@ +import threading +import time + +from kafka.errors import QuotaViolationError +from kafka.metrics import KafkaMetric + + +class Sensor(object): + """ + A sensor applies a continuous sequence of numerical values + to a set of associated metrics. For example a sensor on + message size would record a sequence of message sizes using + the `record(double)` api and would maintain a set + of metrics about request sizes such as the average or max. + """ + def __init__(self, registry, name, parents, config, + inactive_sensor_expiration_time_seconds): + if not name: + raise ValueError('name must be non-empty') + self._lock = threading.RLock() + self._registry = registry + self._name = name + self._parents = parents or [] + self._metrics = [] + self._stats = [] + self._config = config + self._inactive_sensor_expiration_time_ms = ( + inactive_sensor_expiration_time_seconds * 1000) + self._last_record_time = time.time() * 1000 + self._check_forest(set()) + + def _check_forest(self, sensors): + """Validate that this sensor doesn't end up referencing itself.""" + if self in sensors: + raise ValueError('Circular dependency in sensors: %s is its own' + 'parent.' % self.name) + sensors.add(self) + for parent in self._parents: + parent._check_forest(sensors) + + @property + def name(self): + """ + The name this sensor is registered with. + This name will be unique among all registered sensors. + """ + return self._name + + @property + def metrics(self): + return tuple(self._metrics) + + def record(self, value=1.0, time_ms=None): + """ + Record a value at a known time. + Arguments: + value (double): The value we are recording + time_ms (int): The current POSIX time in milliseconds + + Raises: + QuotaViolationException: if recording this value moves a + metric beyond its configured maximum or minimum bound + """ + now = time.time() * 1000 + if time_ms is None: + time_ms = now + self._last_record_time = now + with self._lock: # XXX high volume, might be performance issue + # increment all the stats + for stat in self._stats: + stat.record(self._config, value, time_ms) + self._check_quotas(time_ms) + for parent in self._parents: + parent.record(value, time_ms) + + def _check_quotas(self, time_ms): + """ + Check if we have violated our quota for any metric that + has a configured quota + """ + for metric in self._metrics: + if metric.config and metric.config.quota: + value = metric.value(time_ms) + if not metric.config.quota.is_acceptable(value): + raise QuotaViolationError('(%s) violated quota. Actual: ' + '(%d), Threshold: (%d)' % + (metric.metric_name, + metric.config.quota.bound, + value)) + + def add_compound(self, compound_stat, config=None): + """ + Register a compound statistic with this sensor which + yields multiple measurable quantities (like a histogram) + + Arguments: + stat (AbstractCompoundStat): The stat to register + config (MetricConfig): The configuration for this stat. + If None then the stat will use the default configuration + for this sensor. + """ + if not compound_stat: + raise ValueError('compound stat must be non-empty') + self._stats.append(compound_stat) + for named_measurable in compound_stat.stats(): + metric = KafkaMetric(named_measurable.name, named_measurable.stat, + config or self._config) + self._registry.register_metric(metric) + self._metrics.append(metric) + + def add(self, metric_name, stat, config=None): + """ + Register a metric with this sensor + + Arguments: + metric_name (MetricName): The name of the metric + stat (AbstractMeasurableStat): The statistic to keep + config (MetricConfig): A special configuration for this metric. + If None use the sensor default configuration. + """ + with self._lock: + metric = KafkaMetric(metric_name, stat, config or self._config) + self._registry.register_metric(metric) + self._metrics.append(metric) + self._stats.append(stat) + + def has_expired(self): + """ + Return True if the Sensor is eligible for removal due to inactivity. + """ + return ((time.time() * 1000 - self._last_record_time) > + self._inactive_sensor_expiration_time_ms) diff --git a/kafka/metrics/stats/total.py b/kafka/metrics/stats/total.py new file mode 100644 index 0000000..76a82d8 --- /dev/null +++ b/kafka/metrics/stats/total.py @@ -0,0 +1,13 @@ +from kafka.metrics.measurable_stat import AbstractMeasurableStat + + +class Total(AbstractMeasurableStat): + """An un-windowed cumulative total maintained over all time.""" + def __init__(self, value=0.0): + self._total = value + + def record(self, config, value, now): + self._total += value + + def measure(self, config, now): + return float(self._total) |