diff options
32 files changed, 1994 insertions, 121 deletions
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py index 1f0619b..1d4b0f0 100644 --- a/kafka/consumer/fetcher.py +++ b/kafka/consumer/fetcher.py @@ -3,11 +3,13 @@ from __future__ import absolute_import import collections import copy import logging +import time import six import kafka.errors as Errors from kafka.future import Future +from kafka.metrics.stats import Avg, Count, Max, Rate from kafka.protocol.fetch import FetchRequest from kafka.protocol.message import PartialMessage from kafka.protocol.offset import OffsetRequest, OffsetResetStrategy @@ -40,7 +42,8 @@ class Fetcher(six.Iterator): 'api_version': (0, 8, 0), } - def __init__(self, client, subscriptions, **configs): + def __init__(self, client, subscriptions, metrics, metric_group_prefix, + **configs): """Initialize a Kafka Message Fetcher. Keyword Arguments: @@ -68,8 +71,6 @@ class Fetcher(six.Iterator): the messages occurred. This check adds some overhead, so it may be disabled in cases seeking extreme performance. Default: True """ - #metrics=None, - #metric_group_prefix='consumer', self.config = copy.copy(self.DEFAULT_CONFIG) for key in self.config: if key in configs: @@ -83,8 +84,7 @@ class Fetcher(six.Iterator): self._record_too_large_partitions = dict() # {topic_partition: offset} self._iterator = None self._fetch_futures = collections.deque() - - #self.sensors = FetchManagerMetrics(metrics, metric_group_prefix) + self._sensors = FetchManagerMetrics(metrics, metric_group_prefix) def init_fetches(self): """Send FetchRequests asynchronously for all assigned partitions. @@ -109,7 +109,7 @@ class Fetcher(six.Iterator): if self._client.ready(node_id): log.debug("Sending FetchRequest to node %s", node_id) future = self._client.send(node_id, request) - future.add_callback(self._handle_fetch_response, request) + future.add_callback(self._handle_fetch_response, request, time.time()) future.add_errback(log.error, 'Fetch to node %s failed: %s', node_id) futures.append(future) self._fetch_futures.extend(futures) @@ -575,10 +575,11 @@ class Fetcher(six.Iterator): partition_data.items()) return requests - def _handle_fetch_response(self, request, response): + def _handle_fetch_response(self, request, send_time, response): """The callback for fetch completion""" - #total_bytes = 0 - #total_count = 0 + total_bytes = 0 + total_count = 0 + recv_time = time.time() fetch_offsets = {} for topic, partitions in request.topics: @@ -609,6 +610,7 @@ class Fetcher(six.Iterator): position) continue + num_bytes = 0 partial = None if messages and isinstance(messages[-1][-1], PartialMessage): partial = messages.pop() @@ -618,18 +620,18 @@ class Fetcher(six.Iterator): " offset %d to buffered record list", tp, position) self._records.append((fetch_offset, tp, messages)) - #last_offset, _, _ = messages[-1] - #self.sensors.records_fetch_lag.record(highwater - last_offset) + last_offset, _, _ = messages[-1] + self._sensors.records_fetch_lag.record(highwater - last_offset) + num_bytes = sum(msg[1] for msg in messages) elif partial: # we did not read a single message from a non-empty # buffer because that message's size is larger than # fetch size, in this case record this exception self._record_too_large_partitions[tp] = fetch_offset - # TODO: bytes metrics - #self.sensors.record_topic_fetch_metrics(tp.topic, num_bytes, parsed.size()); - #totalBytes += num_bytes; - #totalCount += parsed.size(); + self._sensors.record_topic_fetch_metrics(topic, num_bytes, len(messages)) + total_bytes += num_bytes + total_count += len(messages) elif error_type in (Errors.NotLeaderForPartitionError, Errors.UnknownTopicOrPartitionError): self._client.cluster.request_update() @@ -649,56 +651,82 @@ class Fetcher(six.Iterator): else: raise error_type('Unexpected error while fetching data') - """TOOD - metrics - self.sensors.bytesFetched.record(totalBytes) - self.sensors.recordsFetched.record(totalCount) - self.sensors.fetchThrottleTimeSensor.record(response.getThrottleTime()) - self.sensors.fetchLatency.record(resp.requestLatencyMs()) + self._sensors.bytes_fetched.record(total_bytes) + self._sensors.records_fetched.record(total_count) + self._sensors.fetch_throttle_time_sensor.record(response['throttle_time_ms']) + self._sensors.fetch_latency.record((recv_time - send_time) * 1000) class FetchManagerMetrics(object): def __init__(self, metrics, prefix): self.metrics = metrics - self.group_name = prefix + "-fetch-manager-metrics" - - self.bytes_fetched = metrics.sensor("bytes-fetched") - self.bytes_fetched.add(metrics.metricName("fetch-size-avg", self.group_name, - "The average number of bytes fetched per request"), metrics.Avg()) - self.bytes_fetched.add(metrics.metricName("fetch-size-max", self.group_name, - "The maximum number of bytes fetched per request"), metrics.Max()) - self.bytes_fetched.add(metrics.metricName("bytes-consumed-rate", self.group_name, - "The average number of bytes consumed per second"), metrics.Rate()) - - self.records_fetched = self.metrics.sensor("records-fetched") - self.records_fetched.add(metrics.metricName("records-per-request-avg", self.group_name, - "The average number of records in each request"), metrics.Avg()) - self.records_fetched.add(metrics.metricName("records-consumed-rate", self.group_name, - "The average number of records consumed per second"), metrics.Rate()) - - self.fetch_latency = metrics.sensor("fetch-latency") - self.fetch_latency.add(metrics.metricName("fetch-latency-avg", self.group_name, - "The average time taken for a fetch request."), metrics.Avg()) - self.fetch_latency.add(metrics.metricName("fetch-latency-max", self.group_name, - "The max time taken for any fetch request."), metrics.Max()) - self.fetch_latency.add(metrics.metricName("fetch-rate", self.group_name, - "The number of fetch requests per second."), metrics.Rate(metrics.Count())) - - self.records_fetch_lag = metrics.sensor("records-lag") - self.records_fetch_lag.add(metrics.metricName("records-lag-max", self.group_name, - "The maximum lag in terms of number of records for any partition in self window"), metrics.Max()) - - self.fetch_throttle_time_sensor = metrics.sensor("fetch-throttle-time") - self.fetch_throttle_time_sensor.add(metrics.metricName("fetch-throttle-time-avg", self.group_name, - "The average throttle time in ms"), metrics.Avg()) - self.fetch_throttle_time_sensor.add(metrics.metricName("fetch-throttle-time-max", self.group_name, - "The maximum throttle time in ms"), metrics.Max()) - - def record_topic_fetch_metrics(topic, num_bytes, num_records): - # record bytes fetched - name = '.'.join(["topic", topic, "bytes-fetched"]) - self.metrics[name].record(num_bytes); - - # record records fetched - name = '.'.join(["topic", topic, "records-fetched"]) - self.metrics[name].record(num_records) - """ + self.group_name = '%s-fetch-manager-metrics' % prefix + + self.bytes_fetched = metrics.sensor('bytes-fetched') + self.bytes_fetched.add(metrics.metric_name('fetch-size-avg', self.group_name, + 'The average number of bytes fetched per request'), Avg()) + self.bytes_fetched.add(metrics.metric_name('fetch-size-max', self.group_name, + 'The maximum number of bytes fetched per request'), Max()) + self.bytes_fetched.add(metrics.metric_name('bytes-consumed-rate', self.group_name, + 'The average number of bytes consumed per second'), Rate()) + + self.records_fetched = self.metrics.sensor('records-fetched') + self.records_fetched.add(metrics.metric_name('records-per-request-avg', self.group_name, + 'The average number of records in each request'), Avg()) + self.records_fetched.add(metrics.metric_name('records-consumed-rate', self.group_name, + 'The average number of records consumed per second'), Rate()) + + self.fetch_latency = metrics.sensor('fetch-latency') + self.fetch_latency.add(metrics.metric_name('fetch-latency-avg', self.group_name, + 'The average time taken for a fetch request.'), Avg()) + self.fetch_latency.add(metrics.metric_name('fetch-latency-max', self.group_name, + 'The max time taken for any fetch request.'), Max()) + self.fetch_latency.add(metrics.metric_name('fetch-rate', self.group_name, + 'The number of fetch requests per second.'), Rate(sampled_stat=Count())) + + self.records_fetch_lag = metrics.sensor('records-lag') + self.records_fetch_lag.add(metrics.metric_name('records-lag-max', self.group_name, + 'The maximum lag in terms of number of records for any partition in self window'), Max()) + + self.fetch_throttle_time_sensor = metrics.sensor('fetch-throttle-time') + self.fetch_throttle_time_sensor.add(metrics.metric_name('fetch-throttle-time-avg', self.group_name, + 'The average throttle time in ms'), Avg()) + self.fetch_throttle_time_sensor.add(metrics.metric_name('fetch-throttle-time-max', self.group_name, + 'The maximum throttle time in ms'), Max()) + + def record_topic_fetch_metrics(self, topic, num_bytes, num_records): + metric_tags = {'topic': topic.replace('.', '_')} + + # record bytes fetched + name = '.'.join(['topic', topic, 'bytes-fetched']) + bytes_fetched = self.metrics.get_sensor(name) + if not bytes_fetched: + bytes_fetched = self.metrics.sensor(name) + bytes_fetched.add(self.metrics.metric_name('fetch-size-avg', + self.group_name, + 'The average number of bytes fetched per request for topic %s' % topic, + metric_tags), Avg()) + bytes_fetched.add(self.metrics.metric_name('fetch-size-max', + self.group_name, + 'The maximum number of bytes fetched per request for topic %s' % topic, + metric_tags), Max()) + bytes_fetched.add(self.metrics.metric_name('bytes-consumed-rate', + self.group_name, + 'The average number of bytes consumed per second for topic %s' % topic, + metric_tags), Rate()) + bytes_fetched.record(num_bytes) + + # record records fetched + name = '.'.join(['topic', topic, 'records-fetched']) + records_fetched = self.metrics.get_sensor(name) + if not records_fetched: + records_fetched = self.metrics.sensor(name) + records_fetched.add(self.metrics.metric_name('records-per-request-avg', + self.group_name, + 'The average number of records in each request for topic %s' % topic, + metric_tags), Avg()) + records_fetched.add(self.metrics.metric_name('records-consumed-rate', + self.group_name, + 'The average number of records consumed per second for topic %s' % topic, + metric_tags), Rate()) + records_fetched.record(num_records) diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py index 0a78e7f..abb65ef 100644 --- a/kafka/consumer/group.py +++ b/kafka/consumer/group.py @@ -12,6 +12,7 @@ from kafka.consumer.subscription_state import SubscriptionState from kafka.coordinator.consumer import ConsumerCoordinator from kafka.coordinator.assignors.range import RangePartitionAssignor from kafka.coordinator.assignors.roundrobin import RoundRobinPartitionAssignor +from kafka.metrics import DictReporter, MetricConfig, Metrics from kafka.protocol.offset import OffsetResetStrategy from kafka.structs import TopicPartition from kafka.version import __version__ @@ -143,6 +144,13 @@ class KafkaConsumer(six.Iterator): offset commits; 0.8.0 is what is left. If set to 'auto', will attempt to infer the broker version by probing various APIs. Default: auto + metric_reporters (list): A list of classes to use as metrics reporters. + Implementing the AbstractMetricsReporter interface allows plugging + in classes that will be notified of new metric creation. Default: [] + metrics_num_samples (int): The number of samples maintained to compute + metrics. Default: 2 + metrics_sample_window_ms (int): The number of samples maintained to + compute metrics. Default: 30000 Note: Configuration parameters are described in more detail at @@ -181,9 +189,9 @@ class KafkaConsumer(six.Iterator): 'ssl_keyfile': None, 'api_version': 'auto', 'connections_max_idle_ms': 9 * 60 * 1000, # not implemented yet - #'metric_reporters': None, - #'metrics_num_samples': 2, - #'metrics_sample_window_ms': 30000, + 'metric_reporters': [], + 'metrics_num_samples': 2, + 'metrics_sample_window_ms': 30000, } def __init__(self, *topics, **configs): @@ -202,6 +210,16 @@ class KafkaConsumer(six.Iterator): new_config, self.config['auto_offset_reset']) self.config['auto_offset_reset'] = new_config + metrics_tags = {'client-id': self.config['client_id']} + metric_config = MetricConfig(samples=self.config['metrics_num_samples'], + time_window_ms=self.config['metrics_sample_window_ms'], + tags=metrics_tags) + reporters = [reporter() for reporter in self.config['metric_reporters']] + reporters.append(DictReporter('kafka.consumer')) + self._metrics = Metrics(metric_config, reporters) + metric_group_prefix = 'consumer' + # TODO _metrics likely needs to be passed to KafkaClient, etc. + self._client = KafkaClient(**self.config) # Check Broker Version if not set explicitly @@ -215,16 +233,15 @@ class KafkaConsumer(six.Iterator): self._subscription = SubscriptionState(self.config['auto_offset_reset']) self._fetcher = Fetcher( - self._client, self._subscription, **self.config) + self._client, self._subscription, self._metrics, metric_group_prefix, **self.config) self._coordinator = ConsumerCoordinator( - self._client, self._subscription, + self._client, self._subscription, self._metrics, metric_group_prefix, assignors=self.config['partition_assignment_strategy'], **self.config) self._closed = False self._iterator = None self._consumer_timeout = float('inf') - #self.metrics = None if topics: self._subscription.subscribe(topics=topics) self._client.set_topics(topics) @@ -277,7 +294,7 @@ class KafkaConsumer(six.Iterator): log.debug("Closing the KafkaConsumer.") self._closed = True self._coordinator.close() - #self.metrics.close() + self._metrics.close() self._client.close() try: self.config['key_deserializer'].close() diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py index c75eb7c..a4c25a3 100644 --- a/kafka/coordinator/base.py +++ b/kafka/coordinator/base.py @@ -658,7 +658,7 @@ class GroupCoordinatorMetrics(object): self.heartbeat_latency.add(metrics.metricName( "heartbeat-rate", self.group_name, "The average number of heartbeats per second", - tags), metrics.Rate(metrics.Count())) + tags), metrics.Rate(sampled_stat=metrics.Count())) self.join_latency = metrics.sensor("join-latency") self.join_latency.add(metrics.metricName( @@ -672,7 +672,7 @@ class GroupCoordinatorMetrics(object): self.join_latency.add(metrics.metricName( "join-rate", self.group_name, "The number of group joins per second", - tags), metrics.Rate(metrics.Count())) + tags), metrics.Rate(sampled_stat=metrics.Count())) self.sync_latency = metrics.sensor("sync-latency") self.sync_latency.add(metrics.metricName( @@ -686,7 +686,7 @@ class GroupCoordinatorMetrics(object): self.sync_latency.add(metrics.metricName( "sync-rate", self.group_name, "The number of group syncs per second", - tags), metrics.Rate(metrics.Count())) + tags), metrics.Rate(sampled_stat=metrics.Count())) """ lastHeartbeat = Measurable( diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py index cd3d48a..50d2806 100644 --- a/kafka/coordinator/consumer.py +++ b/kafka/coordinator/consumer.py @@ -14,6 +14,8 @@ from .assignors.roundrobin import RoundRobinPartitionAssignor from .protocol import ConsumerProtocol from .. import errors as Errors from ..future import Future +from ..metrics import AnonMeasurable +from ..metrics.stats import Avg, Count, Max, Rate from ..protocol.commit import OffsetCommitRequest, OffsetFetchRequest from ..structs import OffsetAndMetadata, TopicPartition from ..util import WeakMethod @@ -36,7 +38,8 @@ class ConsumerCoordinator(BaseCoordinator): 'api_version': (0, 9), } - def __init__(self, client, subscription, **configs): + def __init__(self, client, subscription, metrics, metric_group_prefix, + **configs): """Initialize the coordination manager. Keyword Arguments: @@ -97,10 +100,8 @@ class ConsumerCoordinator(BaseCoordinator): interval = self.config['auto_commit_interval_ms'] / 1000.0 self._auto_commit_task = AutoCommitTask(weakref.proxy(self), interval) - # metrics=None, - # metric_group_prefix=None, - # metric_tags=None, - # self.sensors = ConsumerCoordinatorMetrics(metrics, metric_group_prefix, metric_tags) + self._sensors = ConsumerCoordinatorMetrics(metrics, metric_group_prefix, + self._subscription) def __del__(self): if hasattr(self, '_auto_commit_task') and self._auto_commit_task: @@ -470,12 +471,13 @@ class ConsumerCoordinator(BaseCoordinator): future = Future() _f = self._client.send(node_id, request) - _f.add_callback(self._handle_offset_commit_response, offsets, future) + _f.add_callback(self._handle_offset_commit_response, offsets, future, time.time()) _f.add_errback(self._failed_request, node_id, request, future) return future - def _handle_offset_commit_response(self, offsets, future, response): - #self.sensors.commit_latency.record(response.requestLatencyMs()) + def _handle_offset_commit_response(self, offsets, future, send_time, response): + # TODO look at adding request_latency_ms to response (like java kafka) + self._sensors.commit_latency.record((time.time() - send_time) * 1000) unauthorized_topics = set() for topic, partitions in response.topics: @@ -720,38 +722,25 @@ class AutoCommitTask(object): self._reschedule(next_at) -# TODO -""" class ConsumerCoordinatorMetrics(object): - def __init__(self, metrics, prefix, tags): + def __init__(self, metrics, metric_group_prefix, subscription): self.metrics = metrics - self.group_name = prefix + "-coordinator-metrics" - - self.commit_latency = metrics.sensor("commit-latency") - self.commit_latency.add(metrics.MetricName( - "commit-latency-avg", self.group_name, - "The average time taken for a commit request", - tags), metrics.Avg()) - self.commit_latency.add(metrics.MetricName( - "commit-latency-max", self.group_name, - "The max time taken for a commit request", - tags), metrics.Max()) - self.commit_latency.add(metrics.MetricName( - "commit-rate", self.group_name, - "The number of commit calls per second", - tags), metrics.Rate(metrics.Count())) - - ''' - def _num_partitions(config, now): - new Measurable() { - public double measure(MetricConfig config, long now) { - return subscriptions.assignedPartitions().size(); - } - }; - metrics.addMetric(new MetricName("assigned-partitions", - this.metricGrpName, - "The number of partitions currently assigned to this consumer", - tags), - numParts); - ''' -""" + self.metric_group_name = '%s-coordinator-metrics' % metric_group_prefix + + self.commit_latency = metrics.sensor('commit-latency') + self.commit_latency.add(metrics.metric_name( + 'commit-latency-avg', self.metric_group_name, + 'The average time taken for a commit request'), Avg()) + self.commit_latency.add(metrics.metric_name( + 'commit-latency-max', self.metric_group_name, + 'The max time taken for a commit request'), Max()) + self.commit_latency.add(metrics.metric_name( + 'commit-rate', self.metric_group_name, + 'The number of commit calls per second'), Rate(sampled_stat=Count())) + + num_parts = AnonMeasurable(lambda config, now: + len(subscription.assigned_partitions())) + metrics.add_metric(metrics.metric_name( + 'assigned-partitions', self.metric_group_name, + 'The number of partitions currently assigned to this consumer'), + num_parts) diff --git a/kafka/errors.py b/kafka/errors.py index a36ee75..dd64b04 100644 --- a/kafka/errors.py +++ b/kafka/errors.py @@ -361,6 +361,10 @@ class KafkaConfigurationError(KafkaError): pass +class QuotaViolationError(KafkaError): + pass + + class AsyncProducerQueueFull(KafkaError): def __init__(self, failed_msgs, *args): super(AsyncProducerQueueFull, self).__init__(*args) 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) diff --git a/test/test_coordinator.py b/test/test_coordinator.py index 399609d..4b90f30 100644 --- a/test/test_coordinator.py +++ b/test/test_coordinator.py @@ -1,5 +1,6 @@ # pylint: skip-file from __future__ import absolute_import +import time import pytest @@ -14,6 +15,7 @@ from kafka.coordinator.protocol import ( ConsumerProtocolMemberMetadata, ConsumerProtocolMemberAssignment) import kafka.errors as Errors from kafka.future import Future +from kafka.metrics import Metrics from kafka.protocol.commit import ( OffsetCommitRequest, OffsetCommitResponse, OffsetFetchRequest, OffsetFetchResponse) @@ -23,12 +25,14 @@ from kafka.util import WeakMethod @pytest.fixture def coordinator(conn): - return ConsumerCoordinator(KafkaClient(), SubscriptionState()) + return ConsumerCoordinator(KafkaClient(), SubscriptionState(), Metrics(), + 'consumer') def test_init(conn): cli = KafkaClient() - coordinator = ConsumerCoordinator(cli, SubscriptionState()) + coordinator = ConsumerCoordinator(cli, SubscriptionState(), Metrics(), + 'consumer') # metadata update on init assert cli.cluster._need_update is True @@ -38,6 +42,7 @@ def test_init(conn): @pytest.mark.parametrize("api_version", [(0, 8, 0), (0, 8, 1), (0, 8, 2), (0, 9)]) def test_autocommit_enable_api_version(conn, api_version): coordinator = ConsumerCoordinator(KafkaClient(), SubscriptionState(), + Metrics(), 'consumer', enable_auto_commit=True, group_id='foobar', api_version=api_version) @@ -354,6 +359,7 @@ def test_maybe_auto_commit_offsets_sync(mocker, api_version, group_id, enable, mock_warn = mocker.patch('kafka.coordinator.consumer.log.warning') mock_exc = mocker.patch('kafka.coordinator.consumer.log.exception') coordinator = ConsumerCoordinator(KafkaClient(), SubscriptionState(), + Metrics(), 'consumer', api_version=api_version, enable_auto_commit=enable, group_id=group_id) @@ -441,7 +447,7 @@ def test_send_offset_commit_request_failure(patched_coord, offsets): assert future.exception is error -def test_send_offset_commit_request_success(patched_coord, offsets): +def test_send_offset_commit_request_success(mocker, patched_coord, offsets): _f = Future() patched_coord._client.send.return_value = _f future = patched_coord._send_offset_commit_request(offsets) @@ -449,7 +455,7 @@ def test_send_offset_commit_request_success(patched_coord, offsets): response = OffsetCommitResponse[0]([('foobar', [(0, 0), (1, 0)])]) _f.success(response) patched_coord._handle_offset_commit_response.assert_called_with( - offsets, future, response) + offsets, future, mocker.ANY, response) @pytest.mark.parametrize('response,error,dead,reassign', [ @@ -478,10 +484,11 @@ def test_send_offset_commit_request_success(patched_coord, offsets): (OffsetCommitResponse[0]([('foobar', [(0, 29), (1, 29)])]), Errors.TopicAuthorizationFailedError, False, False), ]) -def test_handle_offset_commit_response(patched_coord, offsets, +def test_handle_offset_commit_response(mocker, patched_coord, offsets, response, error, dead, reassign): future = Future() - patched_coord._handle_offset_commit_response(offsets, future, response) + patched_coord._handle_offset_commit_response(offsets, future, time.time(), + response) assert isinstance(future.exception, error) assert patched_coord.coordinator_id is (None if dead else 0) assert patched_coord._subscription.needs_partition_assignment is reassign diff --git a/test/test_fetcher.py b/test/test_fetcher.py index 644adfa..bf4a3a9 100644 --- a/test/test_fetcher.py +++ b/test/test_fetcher.py @@ -8,6 +8,7 @@ from kafka.consumer.fetcher import Fetcher from kafka.consumer.subscription_state import SubscriptionState import kafka.errors as Errors from kafka.future import Future +from kafka.metrics import Metrics from kafka.protocol.fetch import FetchRequest from kafka.structs import TopicPartition, OffsetAndMetadata @@ -29,7 +30,7 @@ def fetcher(client, subscription_state): subscription_state.assign_from_subscribed(assignment) for tp in assignment: subscription_state.seek(tp, 0) - return Fetcher(client, subscription_state) + return Fetcher(client, subscription_state, Metrics(), 'test_fetcher') def test_init_fetches(fetcher, mocker): diff --git a/test/test_metrics.py b/test/test_metrics.py new file mode 100644 index 0000000..e4757d6 --- /dev/null +++ b/test/test_metrics.py @@ -0,0 +1,499 @@ +import sys +import time + +import pytest + +from kafka.errors import QuotaViolationError +from kafka.metrics import DictReporter, MetricConfig, MetricName, Metrics, Quota +from kafka.metrics.measurable import AbstractMeasurable +from kafka.metrics.stats import (Avg, Count, Max, Min, Percentile, Percentiles, + Rate, Total) +from kafka.metrics.stats.percentiles import BucketSizing +from kafka.metrics.stats.rate import TimeUnit + +EPS = 0.000001 + + +@pytest.fixture +def time_keeper(): + return TimeKeeper() + + +@pytest.fixture +def config(): + return MetricConfig() + + +@pytest.fixture +def reporter(): + return DictReporter() + + +@pytest.fixture +def metrics(request, config, reporter): + metrics = Metrics(config, [reporter], enable_expiration=True) + request.addfinalizer(lambda: metrics.close()) + return metrics + + +def test_MetricName(): + # The Java test only cover the differences between the deprecated + # constructors, so I'm skipping them but doing some other basic testing. + + # In short, metrics should be equal IFF their name, group, and tags are + # the same. Descriptions do not matter. + name1 = MetricName('name', 'group', 'A metric.', {'a': 1, 'b': 2}) + name2 = MetricName('name', 'group', 'A description.', {'a': 1, 'b': 2}) + assert name1 == name2 + + name1 = MetricName('name', 'group', tags={'a': 1, 'b': 2}) + name2 = MetricName('name', 'group', tags={'a': 1, 'b': 2}) + assert name1 == name2 + + name1 = MetricName('foo', 'group') + name2 = MetricName('name', 'group') + assert name1 != name2 + + name1 = MetricName('name', 'foo') + name2 = MetricName('name', 'group') + assert name1 != name2 + + # name and group must be non-empty. Everything else is optional. + with pytest.raises(Exception): + MetricName('', 'group') + with pytest.raises(Exception): + MetricName('name', None) + # tags must be a dict if supplied + with pytest.raises(Exception): + MetricName('name', 'group', tags=set()) + + # Because of the implementation of __eq__ and __hash__, the values of + # a MetricName cannot be mutable. + tags = {'a': 1} + name = MetricName('name', 'group', 'description', tags=tags) + with pytest.raises(AttributeError): + name.name = 'new name' + with pytest.raises(AttributeError): + name.group = 'new name' + with pytest.raises(AttributeError): + name.tags = {} + # tags is a copy, so the instance isn't altered + name.tags['b'] = 2 + assert name.tags == tags + + +def test_simple_stats(mocker, time_keeper, config, metrics): + mocker.patch('time.time', side_effect=time_keeper.time) + + measurable = ConstantMeasurable() + + metrics.add_metric(metrics.metric_name('direct.measurable', 'grp1', + 'The fraction of time an appender waits for space allocation.'), + measurable) + sensor = metrics.sensor('test.sensor') + sensor.add(metrics.metric_name('test.avg', 'grp1'), Avg()) + sensor.add(metrics.metric_name('test.max', 'grp1'), Max()) + sensor.add(metrics.metric_name('test.min', 'grp1'), Min()) + sensor.add(metrics.metric_name('test.rate', 'grp1'), Rate(TimeUnit.SECONDS)) + sensor.add(metrics.metric_name('test.occurences', 'grp1'),Rate(TimeUnit.SECONDS, Count())) + sensor.add(metrics.metric_name('test.count', 'grp1'), Count()) + percentiles = [Percentile(metrics.metric_name('test.median', 'grp1'), 50.0), + Percentile(metrics.metric_name('test.perc99_9', 'grp1'), 99.9)] + sensor.add_compound(Percentiles(100, BucketSizing.CONSTANT, 100, -100, + percentiles=percentiles)) + + sensor2 = metrics.sensor('test.sensor2') + sensor2.add(metrics.metric_name('s2.total', 'grp1'), Total()) + sensor2.record(5.0) + + sum_val = 0 + count = 10 + for i in range(count): + sensor.record(i) + sum_val += i + + # prior to any time passing + elapsed_secs = (config.time_window_ms * (config.samples - 1)) / 1000.0 + assert abs(count / elapsed_secs - + metrics.metrics.get(metrics.metric_name('test.occurences', 'grp1')).value()) \ + < EPS, 'Occurrences(0...%d) = %f' % (count, count / elapsed_secs) + + # pretend 2 seconds passed... + sleep_time_seconds = 2.0 + time_keeper.sleep(sleep_time_seconds) + elapsed_secs += sleep_time_seconds + + assert abs(5.0 - metrics.metrics.get(metrics.metric_name('s2.total', 'grp1')).value()) \ + < EPS, 's2 reflects the constant value' + assert abs(4.5 - metrics.metrics.get(metrics.metric_name('test.avg', 'grp1')).value()) \ + < EPS, 'Avg(0...9) = 4.5' + assert abs((count - 1) - metrics.metrics.get(metrics.metric_name('test.max', 'grp1')).value()) \ + < EPS, 'Max(0...9) = 9' + assert abs(0.0 - metrics.metrics.get(metrics.metric_name('test.min', 'grp1')).value()) \ + < EPS, 'Min(0...9) = 0' + assert abs((sum_val / elapsed_secs) - metrics.metrics.get(metrics.metric_name('test.rate', 'grp1')).value()) \ + < EPS, 'Rate(0...9) = 1.40625' + assert abs((count / elapsed_secs) - metrics.metrics.get(metrics.metric_name('test.occurences', 'grp1')).value()) \ + < EPS, 'Occurrences(0...%d) = %f' % (count, count / elapsed_secs) + assert abs(count - metrics.metrics.get(metrics.metric_name('test.count', 'grp1')).value()) \ + < EPS, 'Count(0...9) = 10' + + +def test_hierarchical_sensors(metrics): + parent1 = metrics.sensor('test.parent1') + parent1.add(metrics.metric_name('test.parent1.count', 'grp1'), Count()) + parent2 = metrics.sensor('test.parent2') + parent2.add(metrics.metric_name('test.parent2.count', 'grp1'), Count()) + child1 = metrics.sensor('test.child1', parents=[parent1, parent2]) + child1.add(metrics.metric_name('test.child1.count', 'grp1'), Count()) + child2 = metrics.sensor('test.child2', parents=[parent1]) + child2.add(metrics.metric_name('test.child2.count', 'grp1'), Count()) + grandchild = metrics.sensor('test.grandchild', parents=[child1]) + grandchild.add(metrics.metric_name('test.grandchild.count', 'grp1'), Count()) + + # increment each sensor one time + parent1.record() + parent2.record() + child1.record() + child2.record() + grandchild.record() + + p1 = parent1.metrics[0].value() + p2 = parent2.metrics[0].value() + c1 = child1.metrics[0].value() + c2 = child2.metrics[0].value() + gc = grandchild.metrics[0].value() + + # each metric should have a count equal to one + its children's count + assert 1.0 == gc + assert 1.0 + gc == c1 + assert 1.0 == c2 + assert 1.0 + c1 == p2 + assert 1.0 + c1 + c2 == p1 + assert [child1, child2] == metrics._children_sensors.get(parent1) + assert [child1] == metrics._children_sensors.get(parent2) + assert metrics._children_sensors.get(grandchild) is None + + +def test_bad_sensor_hierarchy(metrics): + parent = metrics.sensor('parent') + child1 = metrics.sensor('child1', parents=[parent]) + child2 = metrics.sensor('child2', parents=[parent]) + + with pytest.raises(ValueError): + metrics.sensor('gc', parents=[child1, child2]) + + +def test_remove_sensor(metrics): + size = len(metrics.metrics) + parent1 = metrics.sensor('test.parent1') + parent1.add(metrics.metric_name('test.parent1.count', 'grp1'), Count()) + parent2 = metrics.sensor('test.parent2') + parent2.add(metrics.metric_name('test.parent2.count', 'grp1'), Count()) + child1 = metrics.sensor('test.child1', parents=[parent1, parent2]) + child1.add(metrics.metric_name('test.child1.count', 'grp1'), Count()) + child2 = metrics.sensor('test.child2', parents=[parent2]) + child2.add(metrics.metric_name('test.child2.count', 'grp1'), Count()) + grandchild1 = metrics.sensor('test.gchild2', parents=[child2]) + grandchild1.add(metrics.metric_name('test.gchild2.count', 'grp1'), Count()) + + sensor = metrics.get_sensor('test.parent1') + assert sensor is not None + metrics.remove_sensor('test.parent1') + assert metrics.get_sensor('test.parent1') is None + assert metrics.metrics.get(metrics.metric_name('test.parent1.count', 'grp1')) is None + assert metrics.get_sensor('test.child1') is None + assert metrics._children_sensors.get(sensor) is None + assert metrics.metrics.get(metrics.metric_name('test.child1.count', 'grp1')) is None + + sensor = metrics.get_sensor('test.gchild2') + assert sensor is not None + metrics.remove_sensor('test.gchild2') + assert metrics.get_sensor('test.gchild2') is None + assert metrics._children_sensors.get(sensor) is None + assert metrics.metrics.get(metrics.metric_name('test.gchild2.count', 'grp1')) is None + + sensor = metrics.get_sensor('test.child2') + assert sensor is not None + metrics.remove_sensor('test.child2') + assert metrics.get_sensor('test.child2') is None + assert metrics._children_sensors.get(sensor) is None + assert metrics.metrics.get(metrics.metric_name('test.child2.count', 'grp1')) is None + + sensor = metrics.get_sensor('test.parent2') + assert sensor is not None + metrics.remove_sensor('test.parent2') + assert metrics.get_sensor('test.parent2') is None + assert metrics._children_sensors.get(sensor) is None + assert metrics.metrics.get(metrics.metric_name('test.parent2.count', 'grp1')) is None + + assert size == len(metrics.metrics) + + +def test_remove_inactive_metrics(mocker, time_keeper, metrics): + mocker.patch('time.time', side_effect=time_keeper.time) + + s1 = metrics.sensor('test.s1', None, 1) + s1.add(metrics.metric_name('test.s1.count', 'grp1'), Count()) + + s2 = metrics.sensor('test.s2', None, 3) + s2.add(metrics.metric_name('test.s2.count', 'grp1'), Count()) + + purger = Metrics.ExpireSensorTask + purger.run(metrics) + assert metrics.get_sensor('test.s1') is not None, \ + 'Sensor test.s1 must be present' + assert metrics.metrics.get(metrics.metric_name('test.s1.count', 'grp1')) is not None, \ + 'MetricName test.s1.count must be present' + assert metrics.get_sensor('test.s2') is not None, \ + 'Sensor test.s2 must be present' + assert metrics.metrics.get(metrics.metric_name('test.s2.count', 'grp1')) is not None, \ + 'MetricName test.s2.count must be present' + + time_keeper.sleep(1.001) + purger.run(metrics) + assert metrics.get_sensor('test.s1') is None, \ + 'Sensor test.s1 should have been purged' + assert metrics.metrics.get(metrics.metric_name('test.s1.count', 'grp1')) is None, \ + 'MetricName test.s1.count should have been purged' + assert metrics.get_sensor('test.s2') is not None, \ + 'Sensor test.s2 must be present' + assert metrics.metrics.get(metrics.metric_name('test.s2.count', 'grp1')) is not None, \ + 'MetricName test.s2.count must be present' + + # record a value in sensor s2. This should reset the clock for that sensor. + # It should not get purged at the 3 second mark after creation + s2.record() + + time_keeper.sleep(2) + purger.run(metrics) + assert metrics.get_sensor('test.s2') is not None, \ + 'Sensor test.s2 must be present' + assert metrics.metrics.get(metrics.metric_name('test.s2.count', 'grp1')) is not None, \ + 'MetricName test.s2.count must be present' + + # After another 1 second sleep, the metric should be purged + time_keeper.sleep(1) + purger.run(metrics) + assert metrics.get_sensor('test.s1') is None, \ + 'Sensor test.s2 should have been purged' + assert metrics.metrics.get(metrics.metric_name('test.s1.count', 'grp1')) is None, \ + 'MetricName test.s2.count should have been purged' + + # After purging, it should be possible to recreate a metric + s1 = metrics.sensor('test.s1', None, 1) + s1.add(metrics.metric_name('test.s1.count', 'grp1'), Count()) + assert metrics.get_sensor('test.s1') is not None, \ + 'Sensor test.s1 must be present' + assert metrics.metrics.get(metrics.metric_name('test.s1.count', 'grp1')) is not None, \ + 'MetricName test.s1.count must be present' + + +def test_remove_metric(metrics): + size = len(metrics.metrics) + metrics.add_metric(metrics.metric_name('test1', 'grp1'), Count()) + metrics.add_metric(metrics.metric_name('test2', 'grp1'), Count()) + + assert metrics.remove_metric(metrics.metric_name('test1', 'grp1')) is not None + assert metrics.metrics.get(metrics.metric_name('test1', 'grp1')) is None + assert metrics.metrics.get(metrics.metric_name('test2', 'grp1')) is not None + + assert metrics.remove_metric(metrics.metric_name('test2', 'grp1')) is not None + assert metrics.metrics.get(metrics.metric_name('test2', 'grp1')) is None + + assert size == len(metrics.metrics) + + +def test_event_windowing(mocker, time_keeper): + mocker.patch('time.time', side_effect=time_keeper.time) + + count = Count() + config = MetricConfig(event_window=1, samples=2) + count.record(config, 1.0, time_keeper.ms()) + count.record(config, 1.0, time_keeper.ms()) + assert 2.0 == count.measure(config, time_keeper.ms()) + count.record(config, 1.0, time_keeper.ms()) # first event times out + assert 2.0 == count.measure(config, time_keeper.ms()) + + +def test_time_windowing(mocker, time_keeper): + mocker.patch('time.time', side_effect=time_keeper.time) + + count = Count() + config = MetricConfig(time_window_ms=1, samples=2) + count.record(config, 1.0, time_keeper.ms()) + time_keeper.sleep(.001) + count.record(config, 1.0, time_keeper.ms()) + assert 2.0 == count.measure(config, time_keeper.ms()) + time_keeper.sleep(.001) + count.record(config, 1.0, time_keeper.ms()) # oldest event times out + assert 2.0 == count.measure(config, time_keeper.ms()) + + +def test_old_data_has_no_effect(mocker, time_keeper): + mocker.patch('time.time', side_effect=time_keeper.time) + + max_stat = Max() + min_stat = Min() + avg_stat = Avg() + count_stat = Count() + window_ms = 100 + samples = 2 + config = MetricConfig(time_window_ms=window_ms, samples=samples) + max_stat.record(config, 50, time_keeper.ms()) + min_stat.record(config, 50, time_keeper.ms()) + avg_stat.record(config, 50, time_keeper.ms()) + count_stat.record(config, 50, time_keeper.ms()) + + time_keeper.sleep(samples * window_ms / 1000.0) + assert float('-inf') == max_stat.measure(config, time_keeper.ms()) + assert float(sys.maxsize) == min_stat.measure(config, time_keeper.ms()) + assert 0.0 == avg_stat.measure(config, time_keeper.ms()) + assert 0 == count_stat.measure(config, time_keeper.ms()) + + +def test_duplicate_MetricName(metrics): + metrics.sensor('test').add(metrics.metric_name('test', 'grp1'), Avg()) + with pytest.raises(ValueError): + metrics.sensor('test2').add(metrics.metric_name('test', 'grp1'), Total()) + + +def test_Quotas(metrics): + sensor = metrics.sensor('test') + sensor.add(metrics.metric_name('test1.total', 'grp1'), Total(), + MetricConfig(quota=Quota.upper_bound(5.0))) + sensor.add(metrics.metric_name('test2.total', 'grp1'), Total(), + MetricConfig(quota=Quota.lower_bound(0.0))) + sensor.record(5.0) + with pytest.raises(QuotaViolationError): + sensor.record(1.0) + + assert abs(6.0 - metrics.metrics.get(metrics.metric_name('test1.total', 'grp1')).value()) \ + < EPS + + sensor.record(-6.0) + with pytest.raises(QuotaViolationError): + sensor.record(-1.0) + + +def test_Quotas_equality(): + quota1 = Quota.upper_bound(10.5) + quota2 = Quota.lower_bound(10.5) + assert quota1 != quota2, 'Quota with different upper values should not be equal' + + quota3 = Quota.lower_bound(10.5) + assert quota2 == quota3, 'Quota with same upper and bound values should be equal' + + +def test_Percentiles(metrics): + buckets = 100 + _percentiles = [ + Percentile(metrics.metric_name('test.p25', 'grp1'), 25), + Percentile(metrics.metric_name('test.p50', 'grp1'), 50), + Percentile(metrics.metric_name('test.p75', 'grp1'), 75), + ] + percs = Percentiles(4 * buckets, BucketSizing.CONSTANT, 100.0, 0.0, + percentiles=_percentiles) + config = MetricConfig(event_window=50, samples=2) + sensor = metrics.sensor('test', config) + sensor.add_compound(percs) + p25 = metrics.metrics.get(metrics.metric_name('test.p25', 'grp1')) + p50 = metrics.metrics.get(metrics.metric_name('test.p50', 'grp1')) + p75 = metrics.metrics.get(metrics.metric_name('test.p75', 'grp1')) + + # record two windows worth of sequential values + for i in range(buckets): + sensor.record(i) + + assert abs(p25.value() - 25) < 1.0 + assert abs(p50.value() - 50) < 1.0 + assert abs(p75.value() - 75) < 1.0 + + for i in range(buckets): + sensor.record(0.0) + + assert p25.value() < 1.0 + assert p50.value() < 1.0 + assert p75.value() < 1.0 + +def test_rate_windowing(mocker, time_keeper, metrics): + mocker.patch('time.time', side_effect=time_keeper.time) + + # Use the default time window. Set 3 samples + config = MetricConfig(samples=3) + sensor = metrics.sensor('test.sensor', config) + sensor.add(metrics.metric_name('test.rate', 'grp1'), Rate(TimeUnit.SECONDS)) + + sum_val = 0 + count = config.samples - 1 + # Advance 1 window after every record + for i in range(count): + sensor.record(100) + sum_val += 100 + time_keeper.sleep(config.time_window_ms / 1000.0) + + # Sleep for half the window. + time_keeper.sleep(config.time_window_ms / 2.0 / 1000.0) + + # prior to any time passing + elapsed_secs = (config.time_window_ms * (config.samples - 1) + config.time_window_ms / 2.0) / 1000.0 + + kafka_metric = metrics.metrics.get(metrics.metric_name('test.rate', 'grp1')) + assert abs((sum_val / elapsed_secs) - kafka_metric.value()) < EPS, \ + 'Rate(0...2) = 2.666' + assert abs(elapsed_secs - (kafka_metric.measurable.window_size(config, time.time() * 1000) / 1000.0)) \ + < EPS, 'Elapsed Time = 75 seconds' + + +def test_reporter(metrics): + reporter = DictReporter() + foo_reporter = DictReporter(prefix='foo') + metrics.add_reporter(reporter) + metrics.add_reporter(foo_reporter) + sensor = metrics.sensor('kafka.requests') + sensor.add(metrics.metric_name('pack.bean1.avg', 'grp1'), Avg()) + sensor.add(metrics.metric_name('pack.bean2.total', 'grp2'), Total()) + sensor2 = metrics.sensor('kafka.blah') + sensor2.add(metrics.metric_name('pack.bean1.some', 'grp1'), Total()) + sensor2.add(metrics.metric_name('pack.bean2.some', 'grp1', + tags={'a': 42, 'b': 'bar'}), Total()) + + # kafka-metrics-count > count is the total number of metrics and automatic + expected = { + 'kafka-metrics-count': {'count': 5.0}, + 'grp2': {'pack.bean2.total': 0.0}, + 'grp1': {'pack.bean1.avg': 0.0, 'pack.bean1.some': 0.0}, + 'grp1.a=42,b=bar': {'pack.bean2.some': 0.0}, + } + assert expected == reporter.snapshot() + + for key in list(expected.keys()): + metrics = expected.pop(key) + expected['foo.%s' % key] = metrics + assert expected == foo_reporter.snapshot() + + +class ConstantMeasurable(AbstractMeasurable): + _value = 0.0 + + def measure(self, config, now): + return self._value + + +class TimeKeeper(object): + """ + A clock that you can manually advance by calling sleep + """ + def __init__(self, auto_tick_ms=0): + self._millis = time.time() * 1000 + self._auto_tick_ms = auto_tick_ms + + def time(self): + return self.ms() / 1000.0 + + def ms(self): + self.sleep(self._auto_tick_ms) + return self._millis + + def sleep(self, seconds): + self._millis += (seconds * 1000) |