Add Prometheus metrics test

Author: rdettai

quickwit/Cargo.lock

@@ -24,6 +24,7 @@ aws-sdk-sqs = { workspace = true }
 futures-util = { workspace = true }
 hyper = { workspace = true }
 itertools = { workspace = true }
+regex = { workspace = true }
 reqwest = { workspace = true }
 serde_json = { workspace = true }
 tempfile = { workspace = true }

quickwit/quickwit-integration-tests/Cargo.toml

@@ -13,6 +13,8 @@
 // limitations under the License.
 
 mod cluster_sandbox;
+mod prometheus_parser;
 mod shutdown;
 
 pub(crate) use cluster_sandbox::{ingest, ClusterSandbox, ClusterSandboxBuilder};
+pub(crate) use prometheus_parser::{filter_metrics, parse_prometheus_metrics};

quickwit/quickwit-integration-tests/src/test_utils/mod.rs

@@ -0,0 +1,236 @@
+// Copyright 2021-Present Datadog, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// 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.
+
+use std::collections::HashMap;
+
+use regex::Regex;
+
+#[derive(Debug, PartialEq, Clone)]
+pub struct PrometheusMetric {
+    pub name: String,
+    pub labels: HashMap<String, String>,
+    pub metric_value: f64,
+}
+
+/// Parse Prometheus metrics serialized with prometheus::TextEncoder
+///
+/// Unfortunately, the prometheus crate does not provide a way to parse metrics
+pub fn parse_prometheus_metrics(input: &str) -> Vec<PrometheusMetric> {
+    let mut metrics = Vec::new();
+    let re = Regex::new(r"(?P<name>[^{]+)(?:\{(?P<labels>[^\}]*)\})? (?P<value>.+)").unwrap();
+
+    for line in input.lines() {
+        if line.starts_with('#') {
+            continue;
+        }
+
+        if let Some(caps) = re.captures(line) {
+            let name = caps.name("name").unwrap().as_str().to_string();
+            let metric_value: f64 = caps
+                .name("value")
+                .unwrap()
+                .as_str()
+                .parse()
+                .expect("Failed to parse value");
+
+            let labels = caps.name("labels").map_or(HashMap::new(), |m| {
+                m.as_str()
+                    .split(',')
+                    .map(|label| {
+                        let mut parts = label.splitn(2, '=');
+                        let key = parts.next().unwrap().to_string();
+                        let value = parts.next().unwrap().trim_matches('"').to_string();
+                        (key, value)
+                    })
+                    .collect()
+            });
+
+            metrics.push(PrometheusMetric {
+                name,
+                labels,
+                metric_value,
+            });
+        }
+    }
+
+    metrics
+}
+
+/// Filter metrics by name and a subset of the available labels
+///
+/// Specify an empty Vec of labels to return all metrics with the specified name
+pub fn filter_metrics(
+    metrics: &[PrometheusMetric],
+    name: &str,
+    labels: Vec<(&'static str, &'static str)>,
+) -> Vec<PrometheusMetric> {
+    metrics
+        .iter()
+        .filter(|metric| metric.name == name)
+        .filter(|metric| {
+            labels
+                .iter()
+                .all(|(key, value)| metric.labels.get(*key).map(String::as_str) == Some(*value))
+        })
+        .cloned()
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const TEST_INPUT: &str = r#"
+quickwit_search_leaf_search_single_split_warmup_num_bytes_sum 0
+# HELP quickwit_storage_object_storage_request_duration_seconds Duration of object storage requests in seconds.
+# TYPE quickwit_storage_object_storage_request_duration_seconds histogram
+quickwit_storage_object_storage_request_duration_seconds_bucket{action="delete_objects",le="30"} 0
+quickwit_storage_object_storage_request_duration_seconds_bucket{action="delete_objects",le="+Inf"} 0
+quickwit_storage_object_storage_request_duration_seconds_sum{action="delete_objects"} 0
+quickwit_search_root_search_request_duration_seconds_sum{kind="server",status="success"} 0.004093958
+quickwit_storage_object_storage_requests_total{action="delete_object"} 0
+quickwit_storage_object_storage_requests_total{action="delete_objects"} 0
+"#;
+
+    #[test]
+    fn test_parse_prometheus_metrics() {
+        let metrics = parse_prometheus_metrics(TEST_INPUT);
+        assert_eq!(metrics.len(), 7);
+        assert_eq!(
+            metrics[0],
+            PrometheusMetric {
+                name: "quickwit_search_leaf_search_single_split_warmup_num_bytes_sum".to_string(),
+                labels: HashMap::new(),
+                metric_value: 0.0,
+            }
+        );
+        assert_eq!(
+            metrics[1],
+            PrometheusMetric {
+                name: "quickwit_storage_object_storage_request_duration_seconds_bucket".to_string(),
+                labels: [
+                    ("action".to_string(), "delete_objects".to_string()),
+                    ("le".to_string(), "30".to_string())
+                ]
+                .iter()
+                .cloned()
+                .collect(),
+                metric_value: 0.0,
+            }
+        );
+        assert_eq!(
+            metrics[2],
+            PrometheusMetric {
+                name: "quickwit_storage_object_storage_request_duration_seconds_bucket".to_string(),
+                labels: [
+                    ("action".to_string(), "delete_objects".to_string()),
+                    ("le".to_string(), "+Inf".to_string())
+                ]
+                .iter()
+                .cloned()
+                .collect(),
+                metric_value: 0.0,
+            }
+        );
+        assert_eq!(
+            metrics[3],
+            PrometheusMetric {
+                name: "quickwit_storage_object_storage_request_duration_seconds_sum".to_string(),
+                labels: [("action".to_string(), "delete_objects".to_string())]
+                    .iter()
+                    .cloned()
+                    .collect(),
+                metric_value: 0.0,
+            }
+        );
+        assert_eq!(
+            metrics[4],
+            PrometheusMetric {
+                name: "quickwit_search_root_search_request_duration_seconds_sum".to_string(),
+                labels: [
+                    ("kind".to_string(), "server".to_string()),
+                    ("status".to_string(), "success".to_string())
+                ]
+                .iter()
+                .cloned()
+                .collect(),
+                metric_value: 0.004093958,
+            }
+        );
+        assert_eq!(
+            metrics[5],
+            PrometheusMetric {
+                name: "quickwit_storage_object_storage_requests_total".to_string(),
+                labels: [("action".to_string(), "delete_object".to_string())]
+                    .iter()
+                    .cloned()
+                    .collect(),
+                metric_value: 0.0,
+            }
+        );
+        assert_eq!(
+            metrics[6],
+            PrometheusMetric {
+                name: "quickwit_storage_object_storage_requests_total".to_string(),
+                labels: [("action".to_string(), "delete_objects".to_string())]
+                    .iter()
+                    .cloned()
+                    .collect(),
+                metric_value: 0.0,
+            }
+        );
+    }
+
+    #[test]
+    fn test_filter_prometheus_metrics() {
+        let metrics = parse_prometheus_metrics(TEST_INPUT);
+        {
+            let filtered_metric = filter_metrics(
+                &metrics,
+                "quickwit_storage_object_storage_request_duration_seconds_bucket",
+                vec![],
+            );
+            assert_eq!(filtered_metric.len(), 2);
+        }
+        {
+            let filtered_metric = filter_metrics(
+                &metrics,
+                "quickwit_search_root_search_request_duration_seconds_sum",
+                vec![("status", "success")],
+            );
+            assert_eq!(filtered_metric.len(), 1);
+        }
+        {
+            let filtered_metric =
+                filter_metrics(&metrics, "quickwit_doest_not_exist_metric", vec![]);
+            assert_eq!(filtered_metric.len(), 0);
+        }
+        {
+            let filtered_metric = filter_metrics(
+                &metrics,
+                "quickwit_storage_object_storage_requests_total",
+                vec![("does_not_exist_label", "value")],
+            );
+            assert_eq!(filtered_metric.len(), 0);
+        }
+        {
+            let filtered_metric = filter_metrics(
+                &metrics,
+                "quickwit_storage_object_storage_requests_total",
+                vec![("action", "does_not_exist_value")],
+            );
+            assert_eq!(filtered_metric.len(), 0);
+        }
+    }
+}

quickwit/quickwit-integration-tests/src/test_utils/prometheus_parser.rs

@@ -17,6 +17,7 @@ mod ingest_v1_tests;
 mod ingest_v2_tests;
 mod no_cp_tests;
 mod otlp_tests;
+mod prometheus_tests;
 #[cfg(feature = "sqs-localstack-tests")]
 mod sqs_tests;
 mod tls_tests;

quickwit/quickwit-integration-tests/src/tests/mod.rs

@@ -0,0 +1,101 @@
+// Copyright 2021-Present Datadog, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// 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.
+
+use quickwit_config::service::QuickwitService;
+use quickwit_serve::SearchRequestQueryString;
+
+use crate::test_utils::{filter_metrics, parse_prometheus_metrics, ClusterSandboxBuilder};
+
+#[tokio::test]
+async fn test_metrics_standalone_server() {
+    quickwit_common::setup_logging_for_tests();
+    let sandbox = ClusterSandboxBuilder::build_and_start_standalone().await;
+    let client = sandbox.rest_client(QuickwitService::Indexer);
+
+    client
+        .indexes()
+        .create(
+            r#"
+                version: 0.8
+                index_id: my-new-index
+                doc_mapping:
+                  field_mappings:
+                  - name: body
+                    type: text
+                "#,
+            quickwit_config::ConfigFormat::Yaml,
+            false,
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(
+        client
+            .search(
+                "my-new-index",
+                SearchRequestQueryString {
+                    query: "body:test".to_string(),
+                    max_hits: 10,
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap()
+            .num_hits,
+        0
+    );
+
+    let prometheus_url = format!("{}metrics", client.base_url());
+    let response = reqwest::Client::new()
+        .get(&prometheus_url)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert!(
+        response.status().is_success(),
+        "Request failed with status {}",
+        response.status(),
+    );
+
+    let body = response.text().await.expect("Failed to read response body");
+    // println!("Prometheus metrics:\n{}", body);
+    let metrics = parse_prometheus_metrics(&body);
+    // The assertions validate some very specific metrics. Feel free to add more as needed.
+    {
+        let quickwit_http_requests_total_get_metrics = filter_metrics(
+            &metrics,
+            "quickwit_http_requests_total",
+            vec![("method", "GET")],
+        );
+        assert_eq!(quickwit_http_requests_total_get_metrics.len(), 1);
+        // we don't know exactly how many GET requests to expect as they are used to
+        // poll the node state
+        assert!(quickwit_http_requests_total_get_metrics[0].metric_value > 0.0);
+    }
+    {
+        let quickwit_http_requests_total_post_metrics = filter_metrics(
+            &metrics,
+            "quickwit_http_requests_total",
+            vec![("method", "POST")],
+        );
+        assert_eq!(quickwit_http_requests_total_post_metrics.len(), 1);
+        // 2 POST requests: create index + search
+        assert_eq!(
+            quickwit_http_requests_total_post_metrics[0].metric_value,
+            2.0
+        );
+    }
+    sandbox.shutdown().await.unwrap();
+}

quickwit/quickwit-integration-tests/src/tests/prometheus_tests.rs

@@ -342,6 +342,10 @@ impl QuickwitClient {
 
         Ok(cumulated_resp)
     }
+
+    pub fn base_url(&self) -> &Url {
+        &self.transport.base_url
+    }
 }
 
 pub enum IngestEvent {