From 7417b28e1ef98f16ac8b84affc28eed50fd111fc Mon Sep 17 00:00:00 2001 From: "g.chouet" Date: Fri, 23 Feb 2024 16:59:07 +0100 Subject: [PATCH] Prometheus configurable queries allow externalized prometheus queries in a properties file This resolves #1540 --- config/prometheus-query.properties | 10 ++++ .../ConfigurablePrometheusQuerySupplier.java | 57 +++++++++++++++++++ .../prometheus/PrometheusMetricSampler.java | 23 +++++--- .../PrometheusMetricSamplerTest.java | 38 +++++++++++++ .../prometheusQueriesTest.properties | 1 + .../prometheusQueriesTestFailing.properties | 2 + docs/wiki/User Guide/Configurations.md | 1 + 7 files changed, 125 insertions(+), 7 deletions(-) create mode 100644 config/prometheus-query.properties create mode 100644 cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/monitor/sampling/prometheus/ConfigurablePrometheusQuerySupplier.java create mode 100644 cruise-control/src/test/resources/prometheusQueriesTest.properties create mode 100644 cruise-control/src/test/resources/prometheusQueriesTestFailing.properties diff --git a/config/prometheus-query.properties b/config/prometheus-query.properties new file mode 100644 index 0000000000..3e1b820d74 --- /dev/null +++ b/config/prometheus-query.properties @@ -0,0 +1,10 @@ +# This file contains all the custom prometheus queries +# This is an example property file for Kafka Cruise Control configurable prometheus queries. + + +# ======================================= +# This must define all the queries for the different RawMetricType needed. +# based on the schema =query + +BROKER_CPU_UTIL=1 - avg by (instance) (irate(node_cpu_seconds_total{mode="idle"}[2m])) +# .. \ No newline at end of file diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/monitor/sampling/prometheus/ConfigurablePrometheusQuerySupplier.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/monitor/sampling/prometheus/ConfigurablePrometheusQuerySupplier.java new file mode 100644 index 0000000000..f33d23a206 --- /dev/null +++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/monitor/sampling/prometheus/ConfigurablePrometheusQuerySupplier.java @@ -0,0 +1,57 @@ +package com.linkedin.kafka.cruisecontrol.monitor.sampling.prometheus; + +import com.linkedin.cruisecontrol.common.CruiseControlConfigurable; +import com.linkedin.kafka.cruisecontrol.metricsreporter.metric.RawMetricType; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; + +import static com.linkedin.cruisecontrol.common.utils.Utils.validateNotNull; +import static com.linkedin.kafka.cruisecontrol.monitor.sampling.prometheus.PrometheusMetricSampler.PROMETHEUS_QUERY_FILE_CONFIG; + +/** + * Configurable prometheus query supplier. This needs a configuration file to specify the different + * prometheus metrics. + *

+ * See {@link PrometheusQuerySupplier} + */ +public class ConfigurablePrometheusQuerySupplier implements CruiseControlConfigurable, PrometheusQuerySupplier { + private static final Map TYPE_TO_QUERY = new HashMap<>(); + + @Override + public Map get() { + return TYPE_TO_QUERY; + } + + @Override + public void configure(Map configs) { + String configFileName = validateNotNull((String) configs.get(PROMETHEUS_QUERY_FILE_CONFIG), + "Prometheus configuration file is missing."); + + internalParse(configFileName); + } + + /** + * Parse the config file to fill in the map + * @param configFileName the name of the input config file + */ + private void internalParse(String configFileName) { + Properties props = new Properties(); + try (InputStream propStream = new FileInputStream(configFileName)) { + props.load(propStream); + } catch (IOException e) { + throw new IllegalArgumentException(e); + } + + // load each entry of the properties into the internal map + props.forEach((key, value) -> loadEntry((String) key, (String) value)); + } + + private void loadEntry(String rawMetricTypeName, String query) { + // will throw a IllegalArgumentException if the name is unknown in the RawMetricType enum + TYPE_TO_QUERY.put(RawMetricType.valueOf(rawMetricTypeName), query); + } +} diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/monitor/sampling/prometheus/PrometheusMetricSampler.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/monitor/sampling/prometheus/PrometheusMetricSampler.java index 34ed02b65a..94462c2bd5 100644 --- a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/monitor/sampling/prometheus/PrometheusMetricSampler.java +++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/monitor/sampling/prometheus/PrometheusMetricSampler.java @@ -59,7 +59,10 @@ public class PrometheusMetricSampler extends AbstractMetricSampler { // Config name visible to tests static final String PROMETHEUS_QUERY_SUPPLIER_CONFIG = "prometheus.query.supplier"; + static final String PROMETHEUS_QUERY_FILE_CONFIG = "prometheus.query.file"; + private static final Class DEFAULT_PROMETHEUS_QUERY_SUPPLIER = DefaultPrometheusQuerySupplier.class; + private static final Class PROMETHEUS_QUERY_FILE_SUPPLIER = ConfigurablePrometheusQuerySupplier.class; private static final Logger LOG = LoggerFactory.getLogger(PrometheusMetricSampler.class); @@ -120,15 +123,21 @@ private void configurePrometheusAdapter(Map configs) { } private void configureQueryMap(Map configs) { + String prometheusQueryFileName = (String) configs.get(PROMETHEUS_QUERY_FILE_CONFIG); String prometheusQuerySupplierClassName = (String) configs.get(PROMETHEUS_QUERY_SUPPLIER_CONFIG); Class prometheusQuerySupplierClass = DEFAULT_PROMETHEUS_QUERY_SUPPLIER; - if (prometheusQuerySupplierClassName != null) { - prometheusQuerySupplierClass = (Class) ConfigDef.parseType(PROMETHEUS_QUERY_SUPPLIER_CONFIG, - prometheusQuerySupplierClassName, CLASS); - if (!PrometheusQuerySupplier.class.isAssignableFrom(prometheusQuerySupplierClass)) { - throw new ConfigException(String.format( - "Invalid %s is provided to prometheus metric sampler, provided %s", - PROMETHEUS_QUERY_SUPPLIER_CONFIG, prometheusQuerySupplierClass)); + // prometheus configuration file is first over other custom or default query supplier class + if (null != prometheusQueryFileName) { + prometheusQuerySupplierClass = PROMETHEUS_QUERY_FILE_SUPPLIER; + } else { + if (prometheusQuerySupplierClassName != null) { + prometheusQuerySupplierClass = (Class) ConfigDef.parseType(PROMETHEUS_QUERY_SUPPLIER_CONFIG, + prometheusQuerySupplierClassName, CLASS); + if (!PrometheusQuerySupplier.class.isAssignableFrom(prometheusQuerySupplierClass)) { + throw new ConfigException(String.format( + "Invalid %s is provided to prometheus metric sampler, provided %s", + PROMETHEUS_QUERY_SUPPLIER_CONFIG, prometheusQuerySupplierClass)); + } } } PrometheusQuerySupplier prometheusQuerySupplier = KafkaCruiseControlConfigUtils.getConfiguredInstance( diff --git a/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/monitor/sampling/prometheus/PrometheusMetricSamplerTest.java b/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/monitor/sampling/prometheus/PrometheusMetricSamplerTest.java index 7e73f03f57..e34b5d058d 100644 --- a/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/monitor/sampling/prometheus/PrometheusMetricSamplerTest.java +++ b/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/monitor/sampling/prometheus/PrometheusMetricSamplerTest.java @@ -169,6 +169,44 @@ public void testGetSamplesPrometheusQuerySupplierInvalidClass() throws Exception _prometheusMetricSampler.configure(config); } + @Test + public void testGetSamplesCustomPrometheusQueryFile() throws Exception { + Map config = new HashMap<>(); + config.put(PROMETHEUS_SERVER_ENDPOINT_CONFIG, "http://kafka-cluster-1.org:9090"); + addCapacityConfig(config); + config.put(PROMETHEUS_QUERY_FILE_CONFIG, this.getClass().getClassLoader().getResource("prometheusQueriesTest.properties").getFile()); + _prometheusMetricSampler.configure(config); + + MetricSamplerOptions metricSamplerOptions = buildMetricSamplerOptions(TEST_TOPIC); + _prometheusMetricSampler._prometheusAdapter = _prometheusAdapter; + + expect(_prometheusAdapter.queryMetric(eq(TestQuerySupplier.TEST_QUERY), anyLong(), anyLong())) + .andReturn(buildBrokerResults()); + replay(_prometheusAdapter); + + _prometheusMetricSampler.getSamples(metricSamplerOptions); + + verify(_prometheusAdapter); + } + + @Test(expected = IllegalArgumentException.class) + public void testGetSamplesCustomPrometheusQueryFileNotFoundFile() throws Exception { + Map config = new HashMap<>(); + config.put(PROMETHEUS_SERVER_ENDPOINT_CONFIG, "http://kafka-cluster-1.org:9090"); + addCapacityConfig(config); + config.put(PROMETHEUS_QUERY_FILE_CONFIG, "/a/b/file.properties"); + _prometheusMetricSampler.configure(config); + } + + @Test(expected = IllegalArgumentException.class) + public void testGetSamplesCustomPrometheusQueryFileUnknownMetricType() throws Exception { + Map config = new HashMap<>(); + config.put(PROMETHEUS_SERVER_ENDPOINT_CONFIG, "http://kafka-cluster-1.org:9090"); + addCapacityConfig(config); + config.put(PROMETHEUS_QUERY_FILE_CONFIG, this.getClass().getClassLoader().getResource("prometheusQueriesTestFailing.properties").getFile()); + _prometheusMetricSampler.configure(config); + } + private static MetricSamplerOptions buildMetricSamplerOptions(String topic) { return new MetricSamplerOptions( diff --git a/cruise-control/src/test/resources/prometheusQueriesTest.properties b/cruise-control/src/test/resources/prometheusQueriesTest.properties new file mode 100644 index 0000000000..3e19c2bb5c --- /dev/null +++ b/cruise-control/src/test/resources/prometheusQueriesTest.properties @@ -0,0 +1 @@ +ALL_TOPIC_BYTES_IN=test_query \ No newline at end of file diff --git a/cruise-control/src/test/resources/prometheusQueriesTestFailing.properties b/cruise-control/src/test/resources/prometheusQueriesTestFailing.properties new file mode 100644 index 0000000000..669136dd87 --- /dev/null +++ b/cruise-control/src/test/resources/prometheusQueriesTestFailing.properties @@ -0,0 +1,2 @@ +ALL_TOPIC_BYTES_IN=test_query +Unknown="I don't know" \ No newline at end of file diff --git a/docs/wiki/User Guide/Configurations.md b/docs/wiki/User Guide/Configurations.md index 5100b9cca3..91faafe9dd 100644 --- a/docs/wiki/User Guide/Configurations.md +++ b/docs/wiki/User Guide/Configurations.md @@ -298,6 +298,7 @@ We are still trying to improve cruise control. And following are some configurat | prometheus.server.endpoint | String | Y | | The HTTP endpoint of the Prometheus server which is to be used as a source for sampling metrics. | | prometheus.query.resolution.step.ms | Integer | N | 60,000 | The resolution of the Prometheus query made to the server. If this is set to 30 seconds for a 2 minutes query interval, the query returns with 4 values, which are then aggregated into the metric sample. | | prometheus.query.supplier | Class | N | com.linkedin.kafka.cruisecontrol.monitor.sampling.prometheus.DefaultPrometheusQuerySupplier | The class that supplies the Prometheus queries corresponding to Kafka raw metrics. If there are no customizations done when configuring Prometheus node exporter, the default class should work fine. | +| prometheus.query.file | String | N | | The configuration file supplying the custom Prometheus queries corresponding to Kafka raw metrics. Takes precedence over prometheus.query.supplier. | | prometheus.broker.metrics.scraping.frequency.seconds | Integer | N | 60 | The scraping frequency with which Prometheus scrapes metrics from Kafka brokers. This value is used by DefaultPrometheusQuerySupplier to construct the iRate query that is used to get broker cpu metrics. | ### KafkaSampleStore configurations | Name | Type | Required? | Default Value | Description |