tests: add test for prometheus exporter

Fixes scylladb#2233
Signed-off-by: Kefu Chai <[email protected]>

tests/unit/CMakeLists.txt

@@ -757,3 +757,20 @@ add_test (
 set_tests_properties (Seastar.unit.json2code
  PROPERTIES
  TIMEOUT ${Seastar_TEST_TIMEOUT})
+
+add_executable (exporter_httpd
+ exporter_httpd.cc)
+target_link_libraries (exporter_httpd
+ PRIVATE seastar_private)
+
+add_dependencies (unit_tests exporter_httpd)
+add_custom_target (test_unit_prometheus_run
+ COMMAND ${CMAKE_COMMAND} -E env ${Seastar_TEST_ENVIRONMENT} ${CMAKE_CURRENT_SOURCE_DIR}/prometheus_test.py --exporter $<TARGET_FILE:exporter_httpd>
+ USES_TERMINAL)
+add_dependencies (test_unit_prometheus_run exporter_httpd)
+add_test (
+ NAME Seastar.unit.prometheus
+ COMMAND ${CMAKE_COMMAND} --build ${Seastar_BINARY_DIR} --target test_unit_prometheus_run)
+set_tests_properties (Seastar.unit.prometheus
+ PROPERTIES
+ TIMEOUT ${Seastar_TEST_TIMEOUT})

tests/unit/exporter_httpd.cc

@@ -0,0 +1,110 @@
+/*
+ * This file is open source software, licensed to you under the terms
+ * of the Apache License, Version 2.0 (the "License"). See the NOTICE file
+ * distributed with this work for additional information regarding copyright
+ * ownership. You may not use this file except in compliance with the License.
+ *
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+/*
+ * Copyright (C) 2024 Scylladb, Ltd.
+ */
+
+#include <fmt/core.h>
+
+#include <seastar/core/metrics_registration.hh>
+#include <seastar/core/metrics.hh>
+#include <seastar/core/seastar.hh>
+#include <seastar/core/reactor.hh>
+#include <seastar/core/relabel_config.hh>
+#include <seastar/core/prometheus.hh>
+#include <seastar/core/sstring.hh>
+#include <seastar/core/thread.hh>
+#include <seastar/net/inet_address.hh>
+#include <seastar/util/closeable.hh>
+#include "../../apps/lib/stop_signal.hh"
+
+namespace bpo = boost::program_options;
+
+using namespace seastar;
+using namespace httpd;
+
+namespace sm = metrics;
+
+class aggregate_by_name {
+ sm::metric_groups _metrics;
+public:
+ aggregate_by_name() {
+ // aggregation
+ sm::label lb("lb");
+ _metrics.add_group("test", {
+ sm::make_counter("counter_1", sm::description("counter 1"), [] { return 1; })(lb("1")),
+ sm::make_counter("counter_1", sm::description("counter 1"), [] { return 2; })(lb("2"))
+ });
+ std::vector<sm::metric_family_config> fc{
+ // aggregate the metrics named "test_counter_1", and group them by the "lb" label
+ sm::metric_family_config{
+ .name = "test_counter_1",
+ .aggregate_labels = {"lb"}
+ },
+ };
+ sm::set_metric_family_configs(fc);
+ }
+
+ ~aggregate_by_name() {
+ _metrics.clear();
+ }
+};
+
+static_assert(stoppable<httpd::http_server_control>);
+
+int main(int ac, char** av) {
+ app_template app;
+
+ app.add_options()("port", bpo::value<uint16_t>()->default_value(9180), "Prometheus port.");
+ app.add_options()("address", bpo::value<sstring>()->default_value("0.0.0.0"), "Prometheus address");
+ app.add_options()("prefix", bpo::value<sstring>()->default_value("seastar"), "Prometheus metrics prefix");
+
+ return app.run(ac, av, [&] {
+ return seastar::async([&] {
+ seastar_apps_lib::stop_signal stop_signal;
+ auto&& config = app.configuration();
+
+ httpd::http_server_control server;
+
+ aggregate_by_name aggregation;
+
+ uint16_t port = config["port"].as<uint16_t>();
+ server.start("prometheus").get();
+ auto stop_server = defer([&] () noexcept {
+ server.stop().get();
+ });
+ prometheus::config prometheus_config;
+ prometheus_config.prefix = config["prefix"].as<sstring>();
+ prometheus::start(server, prometheus_config).get();
+
+ const net::inet_address iaddr(config["address"].as<sstring>());
+ socket_address saddr{iaddr, port};
+ server.listen(saddr).handle_exception([saddr] (auto ep) {
+ fmt::print(std::cerr, "Could not start exporter on {}: {}\n", saddr, ep);
+ return make_exception_future<>(ep);
+ }).get();
+
+
+ fmt::print("{}\n", port);
+ fflush(stdout);
+
+ stop_signal.wait().get();
+ return 0;
+ });
+ });
+}

tests/unit/prometheus_test.py

@@ -0,0 +1,222 @@
+#!/usr/bin/env python3
+#
+# This file is open source software, licensed to you under the terms
+# of the Apache License, Version 2.0 (the "License"). See the NOTICE file
+# distributed with this work for additional information regarding copyright
+# ownership. You may not use this file except in compliance with the License.
+#
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+#
+# Copyright (C) 2024 Scylladb, Ltd.
+#
+
+import argparse
+import re
+import subprocess
+import sys
+import unittest
+import urllib.request
+import urllib.parse
+
+from typing import Optional
+from collections import namedtuple
+
+
+class Exposition:
+ # we don't verify histogram yet
+ def __init__(self,
+ name: str,
+ value: float,
+ labels: Optional[dict[str, str]] = None) -> None:
+ self.name = name
+ self.value = int(value)
+ self.labels = labels
+
+
+class Metrics:
+ prefix = 'seastar'
+ group = 'test'
+ # parse lines like:
+ # rest_api_scheduler_queue_length{group="main",shard="0"} 0.000000
+ # where:
+ # - "rest_api" is the prometheus prefix
+ # - "scheduler" is the metric group name
+ # - "queue_length" is the name of the metric
+ # - the kv pairs in "{}" are labels"
+ # - "0.000000" is the value of the metric
+ # this format is compatible with
+ # https://github.com/prometheus/docs/blob/main/content/docs/instrumenting/exposition_formats.md
+ # NOTE: scylla does not include timestamp in the exported metrics
+ pattern = re.compile(r'''(?P<metric_name>\w+) # rest_api_scheduler_queue_length
+ \{(?P<labels>[^\}]+)\} # {group="main",shard="0"}
+ \s+ # <space>
+ (?P<value>[^\s])+ # 0.000000''', re.X)
+
+ def __init__(self, lines: list[str]) -> None:
+ self.lines: list[str] = lines
+
+ @classmethod
+ def full_name(cls, name: str) -> str:
+ '''return the full name of a metrics
+ '''
+ return f'{cls.group}_{name}'
+
+ @staticmethod
+ def _parse_labels(s: str) -> dict[str, str]:
+ return dict(name_value.split('=', 1) for name_value in s.split(','))
+
+ def get(self,
+ name: Optional[str] = None,
+ labels: Optional[dict[str, str]] = None) -> list[Exposition]:
+ '''Return all expositions matching the given name and labels
+ '''
+ full_name = ''
+ if name is not None:
+ full_name = f'{self.prefix}_{self.group}_{name}'
+ results: list[Exposition] = []
+
+ for line in self.lines:
+ if line.startswith('#'):
+ # HELP or TYPE
+ continue
+ if not line:
+ continue
+ matched = self.pattern.match(line)
+ assert matched, f'malformed metric line: {line}'
+
+ metric_name = matched.group('metric_name')
+ if full_name and metric_name != full_name:
+ continue
+
+ metric_labels = self._parse_labels(matched.group('labels'))
+ if labels is not None and metric_labels != labels:
+ continue
+
+ metric_value = matched.group('value')
+ results.append(Exposition(metric_name,
+ metric_value,
+ metric_labels))
+ return results
+
+ def _get_meta(self, name: str, meta: str) -> Optional[str]:
+ full_name = f'{self.prefix}_{self.group}_{name}'
+ header = f'# {meta} {full_name}'
+ for line in self.lines:
+ if line.startswith(header):
+ return line
+ return None
+
+ def get_help(self, name: str) -> Optional[str]:
+ return self._get_meta(name, 'HELP')
+
+ def get_type(self, name: str) -> Optional[str]:
+ return self._get_meta(name, 'TYPE')
+
+
+class TestPrometheus(unittest.TestCase):
+ exporter_path = None
+ exporter_process = None
+ port = 10001
+
+
+ @classmethod
+ def setUpClass(cls) -> None:
+ args = [cls.exporter_path,
+ '--port', f'{cls.port}',
+ '--prefix', f'{Metrics.prefix}',
+ '--smp=2']
+ cls.exporter_process = subprocess.Popen(args,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.DEVNULL,
+ bufsize=0, text=True)
+ # wait until the server is ready for serve
+ cls.exporter_process.stdout.readline()
+
+ @classmethod
+ def tearDownClass(cls) -> None:
+ cls.exporter_process.terminate()
+
+ @classmethod
+ def _get_metrics(cls,
+ name: Optional[str] = None,
+ labels: Optional[dict[str, str]] = None,
+ with_help: bool = True,
+ aggregate: bool = True) -> Metrics:
+ query: dict[str, str] = {}
+ if name is not None:
+ query['__name__'] = name
+ if labels is not None:
+ query.update(labels)
+ if not with_help:
+ query['__help__'] = 'false'
+ if not aggregate:
+ query['__aggregate__'] = 'false'
+ params = urllib.parse.urlencode(query)
+ host = 'localhost'
+ url = f'http://{host}:{cls.port}/metrics?{params}'
+ with urllib.request.urlopen(url) as f:
+ body = f.read().decode('utf-8')
+ return Metrics(body.rstrip().split('\n'))
+
+ def test_filtering_by_label(self) -> None:
+ TestCase = namedtuple('TestCase', ['label', 'regex', 'found'])
+ tests = [
+ # TestCase(label='lb', regex='1', found=1),
+ # aggregated
+ # TestCase(label='lb', regex='1|2', found=1),
+ TestCase(label='lb', regex='dne', found=0),
+ TestCase(label='lb', regex='404', found=0),
+ ]
+ for test in tests:
+ with self.subTest(regex=test.regex, found=test.found):
+ metrics = self._get_metrics(labels={test.label: test.regex})
+ self.assertEqual(len(metrics.get()), test.found)
+
+ def test_aggregated(self) -> None:
+ name = 'counter_1'
+ # see also rest_api_httpd.cc::aggregate_by_name
+ TestCase = namedtuple('TestCase', ['aggregate', 'expected_values'])
+ tests = [
+ TestCase(aggregate=False, expected_values=[1, 2]),
+ TestCase(aggregate=True, expected_values=[3])
+ ]
+ for test in tests:
+ with self.subTest(aggregate=test.aggregate,
+ values=test.expected_values):
+ metrics = self._get_metrics(Metrics.full_name(name), aggregate=test.aggregate)
+ expositions = metrics.get(name)
+ actual_values = sorted(e.value for e in expositions)
+ self.assertEqual(actual_values, test.expected_values)
+
+ def test_help(self) -> None:
+ name = 'counter_1'
+ tests = [True, False]
+ for with_help in tests:
+ with self.subTest(with_help=with_help):
+ metrics = self._get_metrics(Metrics.full_name(name), with_help=with_help)
+ msg = metrics.get_help(name)
+ if with_help:
+ self.assertIsNotNone(msg)
+ else:
+ self.assertIsNone(msg)
+
+
+if __name__ == '__main__':
+ parser = argparse.ArgumentParser()
+ parser.add_argument('--exporter',
+ required=True,
+ help='Path of the exporter executable')
+ opts, remaining = parser.parse_known_args()
+ remaining.insert(0, sys.argv[0])
+ TestPrometheus.exporter_path = opts.exporter
+ unittest.main(argv=remaining)