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) yield metrics metrics.close() 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)