tests: add test for prometheus exporter

unlike metrics_test.cc, prometheus_test exercises the exporter server,
so it tests the different query parameters supported by it.

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)