from __future__ import unicode_literals import math import sys from prometheus_client.core import ( CollectorRegistry, CounterMetricFamily, GaugeMetricFamily, HistogramMetricFamily, Metric, Sample, SummaryMetricFamily, ) from prometheus_client.exposition import generate_latest from prometheus_client.parser import text_string_to_metric_families if sys.version_info < (2, 7): # We need the skip decorators from unittest2 on Python 2.6. import unittest2 as unittest else: import unittest class TestParse(unittest.TestCase): def assertEqualMetrics(self, first, second, msg=None): super(TestParse, self).assertEqual(first, second, msg) # Test that samples are actually named tuples of type Sample. for a, b in zip(first, second): for sa, sb in zip(a.samples, b.samples): assert sa.name == sb.name def test_simple_counter(self): families = text_string_to_metric_families("""# TYPE a counter # HELP a help a 1 """) self.assertEqualMetrics([CounterMetricFamily("a", "help", value=1)], list(families)) def test_simple_gauge(self): families = text_string_to_metric_families("""# TYPE a gauge # HELP a help a 1 """) self.assertEqualMetrics([GaugeMetricFamily("a", "help", value=1)], list(families)) def test_simple_summary(self): families = text_string_to_metric_families("""# TYPE a summary # HELP a help a_count 1 a_sum 2 """) summary = SummaryMetricFamily("a", "help", count_value=1, sum_value=2) self.assertEqualMetrics([summary], list(families)) def test_summary_quantiles(self): families = text_string_to_metric_families("""# TYPE a summary # HELP a help a_count 1 a_sum 2 a{quantile="0.5"} 0.7 """) # The Python client doesn't support quantiles, but we # still need to be able to parse them. metric_family = SummaryMetricFamily("a", "help", count_value=1, sum_value=2) metric_family.add_sample("a", {"quantile": "0.5"}, 0.7) self.assertEqualMetrics([metric_family], list(families)) def test_simple_histogram(self): families = text_string_to_metric_families("""# TYPE a histogram # HELP a help a_bucket{le="1"} 0 a_bucket{le="+Inf"} 3 a_count 3 a_sum 2 """) self.assertEqualMetrics([HistogramMetricFamily("a", "help", sum_value=2, buckets=[("1", 0.0), ("+Inf", 3.0)])], list(families)) def test_no_metadata(self): families = text_string_to_metric_families("""a 1 """) metric_family = Metric("a", "", "untyped") metric_family.add_sample("a", {}, 1) self.assertEqualMetrics([metric_family], list(families)) def test_untyped(self): # https://github.com/prometheus/client_python/issues/79 families = text_string_to_metric_families("""# HELP redis_connected_clients Redis connected clients # TYPE redis_connected_clients untyped redis_connected_clients{instance="rough-snowflake-web",port="6380"} 10.0 redis_connected_clients{instance="rough-snowflake-web",port="6381"} 12.0 """) m = Metric("redis_connected_clients", "Redis connected clients", "untyped") m.samples = [ Sample("redis_connected_clients", {"instance": "rough-snowflake-web", "port": "6380"}, 10), Sample("redis_connected_clients", {"instance": "rough-snowflake-web", "port": "6381"}, 12), ] self.assertEqualMetrics([m], list(families)) def test_type_help_switched(self): families = text_string_to_metric_families("""# HELP a help # TYPE a counter a 1 """) self.assertEqualMetrics([CounterMetricFamily("a", "help", value=1)], list(families)) def test_blank_lines_and_comments(self): families = text_string_to_metric_families(""" # TYPE a counter # FOO a # BAR b # HELP a help a 1 """) self.assertEqualMetrics([CounterMetricFamily("a", "help", value=1)], list(families)) def test_tabs(self): families = text_string_to_metric_families("""#\tTYPE\ta\tcounter #\tHELP\ta\thelp a\t1 """) self.assertEqualMetrics([CounterMetricFamily("a", "help", value=1)], list(families)) def test_labels_with_curly_braces(self): families = text_string_to_metric_families("""# TYPE a counter # HELP a help a{foo="bar", bar="b{a}z"} 1 """) metric_family = CounterMetricFamily("a", "help", labels=["foo", "bar"]) metric_family.add_metric(["bar", "b{a}z"], 1) self.assertEqualMetrics([metric_family], list(families)) def test_empty_help(self): families = text_string_to_metric_families("""# TYPE a counter # HELP a a 1 """) self.assertEqualMetrics([CounterMetricFamily("a", "", value=1)], list(families)) def test_labels_and_infinite(self): families = text_string_to_metric_families("""# TYPE a counter # HELP a help a{foo="bar"} +Inf a{foo="baz"} -Inf """) metric_family = CounterMetricFamily("a", "help", labels=["foo"]) metric_family.add_metric(["bar"], float('inf')) metric_family.add_metric(["baz"], float('-inf')) self.assertEqualMetrics([metric_family], list(families)) def test_spaces(self): families = text_string_to_metric_families("""# TYPE a counter # HELP a help a{ foo = "bar" } 1 a\t\t{\t\tfoo\t\t=\t\t"baz"\t\t}\t\t2 a { foo = "buz" } 3 a\t { \t foo\t = "biz"\t } \t 4 a \t{\t foo = "boz"\t}\t 5 """) metric_family = CounterMetricFamily("a", "help", labels=["foo"]) metric_family.add_metric(["bar"], 1) metric_family.add_metric(["baz"], 2) metric_family.add_metric(["buz"], 3) metric_family.add_metric(["biz"], 4) metric_family.add_metric(["boz"], 5) self.assertEqualMetrics([metric_family], list(families)) def test_commas(self): families = text_string_to_metric_families("""# TYPE a counter # HELP a help a{foo="bar",} 1 a{foo="baz", } 1 # TYPE b counter # HELP b help b{,} 2 # TYPE c counter # HELP c help c{ ,} 3 # TYPE d counter # HELP d help d{, } 4 """) a = CounterMetricFamily("a", "help", labels=["foo"]) a.add_metric(["bar"], 1) a.add_metric(["baz"], 1) b = CounterMetricFamily("b", "help", value=2) c = CounterMetricFamily("c", "help", value=3) d = CounterMetricFamily("d", "help", value=4) self.assertEqualMetrics([a, b, c, d], list(families)) def test_multiple_trailing_commas(self): text = """# TYPE a counter # HELP a help a{foo="bar",, } 1 """ self.assertRaises(ValueError, lambda: list(text_string_to_metric_families(text))) def test_empty_brackets(self): families = text_string_to_metric_families("""# TYPE a counter # HELP a help a{} 1 """) self.assertEqualMetrics([CounterMetricFamily("a", "help", value=1)], list(families)) def test_nan(self): families = text_string_to_metric_families("""a NaN """) # Can't use a simple comparison as nan != nan. self.assertTrue(math.isnan(list(families)[0].samples[0][2])) def test_empty_label(self): families = text_string_to_metric_families("""# TYPE a counter # HELP a help a{foo="bar"} 1 a{foo=""} 2 """) metric_family = CounterMetricFamily("a", "help", labels=["foo"]) metric_family.add_metric(["bar"], 1) metric_family.add_metric([""], 2) self.assertEqualMetrics([metric_family], list(families)) def test_label_escaping(self): for escaped_val, unescaped_val in [ ('foo', 'foo'), ('\\foo', '\\foo'), ('\\\\foo', '\\foo'), ('foo\\\\', 'foo\\'), ('\\\\', '\\'), ('\\n', '\n'), ('\\\\n', '\\n'), ('\\\\\\n', '\\\n'), ('\\"', '"'), ('\\\\\\"', '\\"')]: families = list(text_string_to_metric_families(""" # TYPE a counter # HELP a help a{foo="%s",bar="baz"} 1 """ % escaped_val)) metric_family = CounterMetricFamily( "a", "help", labels=["foo", "bar"]) metric_family.add_metric([unescaped_val, "baz"], 1) self.assertEqualMetrics([metric_family], list(families)) def test_help_escaping(self): for escaped_val, unescaped_val in [ ('foo', 'foo'), ('\\foo', '\\foo'), ('\\\\foo', '\\foo'), ('foo\\', 'foo\\'), ('foo\\\\', 'foo\\'), ('\\n', '\n'), ('\\\\n', '\\n'), ('\\\\\\n', '\\\n'), ('\\"', '\\"'), ('\\\\"', '\\"'), ('\\\\\\"', '\\\\"')]: families = list(text_string_to_metric_families(""" # TYPE a counter # HELP a %s a{foo="bar"} 1 """ % escaped_val)) metric_family = CounterMetricFamily("a", unescaped_val, labels=["foo"]) metric_family.add_metric(["bar"], 1) self.assertEqualMetrics([metric_family], list(families)) def test_escaping(self): families = text_string_to_metric_families("""# TYPE a counter # HELP a he\\n\\\\l\\tp a{foo="b\\"a\\nr"} 1 a{foo="b\\\\a\\z"} 2 """) metric_family = CounterMetricFamily("a", "he\n\\l\\tp", labels=["foo"]) metric_family.add_metric(["b\"a\nr"], 1) metric_family.add_metric(["b\\a\\z"], 2) self.assertEqualMetrics([metric_family], list(families)) def test_timestamps(self): families = text_string_to_metric_families("""# TYPE a counter # HELP a help a{foo="bar"} 1\t000 # TYPE b counter # HELP b help b 2 1234567890 b 88 1234566000 """) a = CounterMetricFamily("a", "help", labels=["foo"]) a.add_metric(["bar"], 1, timestamp=0) b = CounterMetricFamily("b", "help") b.add_metric([], 2, timestamp=1234567.89) b.add_metric([], 88, timestamp=1234566) self.assertEqualMetrics([a, b], list(families)) @unittest.skipIf(sys.version_info < (2, 7), "Test requires Python 2.7+.") def test_roundtrip(self): text = """# HELP go_gc_duration_seconds A summary of the GC invocation durations. # TYPE go_gc_duration_seconds summary go_gc_duration_seconds{quantile="0"} 0.013300656000000001 go_gc_duration_seconds{quantile="0.25"} 0.013638736 go_gc_duration_seconds{quantile="0.5"} 0.013759906 go_gc_duration_seconds{quantile="0.75"} 0.013962066 go_gc_duration_seconds{quantile="1"} 0.021383540000000003 go_gc_duration_seconds_sum 56.12904785 go_gc_duration_seconds_count 7476.0 # HELP go_goroutines Number of goroutines that currently exist. # TYPE go_goroutines gauge go_goroutines 166.0 # HELP prometheus_local_storage_indexing_batch_duration_milliseconds Quantiles for batch indexing duration in milliseconds. # TYPE prometheus_local_storage_indexing_batch_duration_milliseconds summary prometheus_local_storage_indexing_batch_duration_milliseconds{quantile="0.5"} NaN prometheus_local_storage_indexing_batch_duration_milliseconds{quantile="0.9"} NaN prometheus_local_storage_indexing_batch_duration_milliseconds{quantile="0.99"} NaN prometheus_local_storage_indexing_batch_duration_milliseconds_sum 871.5665949999999 prometheus_local_storage_indexing_batch_duration_milliseconds_count 229.0 # HELP process_cpu_seconds_total Total user and system CPU time spent in seconds. # TYPE process_cpu_seconds_total counter process_cpu_seconds_total 29323.4 # HELP process_virtual_memory_bytes Virtual memory size in bytes. # TYPE process_virtual_memory_bytes gauge process_virtual_memory_bytes 2.478268416e+09 # HELP prometheus_build_info A metric with a constant '1' value labeled by version, revision, and branch from which Prometheus was built. # TYPE prometheus_build_info gauge prometheus_build_info{branch="HEAD",revision="ef176e5",version="0.16.0rc1"} 1.0 # HELP prometheus_local_storage_chunk_ops_total The total number of chunk operations by their type. # TYPE prometheus_local_storage_chunk_ops_total counter prometheus_local_storage_chunk_ops_total{type="clone"} 28.0 prometheus_local_storage_chunk_ops_total{type="create"} 997844.0 prometheus_local_storage_chunk_ops_total{type="drop"} 1.345758e+06 prometheus_local_storage_chunk_ops_total{type="load"} 1641.0 prometheus_local_storage_chunk_ops_total{type="persist"} 981408.0 prometheus_local_storage_chunk_ops_total{type="pin"} 32662.0 prometheus_local_storage_chunk_ops_total{type="transcode"} 980180.0 prometheus_local_storage_chunk_ops_total{type="unpin"} 32662.0 """ families = list(text_string_to_metric_families(text)) class TextCollector(object): def collect(self): return families registry = CollectorRegistry() registry.register(TextCollector()) self.assertEqual(text.encode('utf-8'), generate_latest(registry)) if __name__ == '__main__': unittest.main()