diff --git a/asap-planner-rs/src/config/input.rs b/asap-planner-rs/src/config/input.rs index 66626c18..18fa1a7f 100644 --- a/asap-planner-rs/src/config/input.rs +++ b/asap-planner-rs/src/config/input.rs @@ -13,8 +13,8 @@ pub struct ControllerConfig { pub windowing: Option, pub sketch_parameters: Option, pub aggregate_cleanup: Option, - /// Optional hint: per-metric label sets used as a fallback when Prometheus - /// returns no series for a metric. Prometheus-inferred labels take priority. + /// Optional authoritative per-metric label sets. Prometheus discovery fills + /// only metrics that have no hint. #[serde(default)] pub metrics: Option>, /// Current streaming config, passed as context for repeated reconfiguration. diff --git a/asap-planner-rs/src/prometheus_client.rs b/asap-planner-rs/src/prometheus_client.rs index 76958fd8..b8b34994 100644 --- a/asap-planner-rs/src/prometheus_client.rs +++ b/asap-planner-rs/src/prometheus_client.rs @@ -165,10 +165,18 @@ pub fn build_schema_from_prometheus( queries: &[String], ) -> Result { let metric_names = extract_metric_names(queries); + build_schema_from_metric_names(prometheus_url, &metric_names) +} + +/// Build a `PromQLSchema` by querying Prometheus for the supplied metric names. +pub(crate) fn build_schema_from_metric_names( + prometheus_url: &str, + metric_names: &HashSet, +) -> Result { debug!("Inferred metric names from queries: {:?}", metric_names); let mut schema = PromQLSchema::new(); - for metric_name in &metric_names { + for metric_name in metric_names { match fetch_labels_for_metric(prometheus_url, metric_name)? { Some(labels) => { debug!("Inferred labels for metric '{}': {:?}", metric_name, labels); diff --git a/asap-planner-rs/src/promql/controller.rs b/asap-planner-rs/src/promql/controller.rs index be08fd36..a24d6edc 100644 --- a/asap-planner-rs/src/promql/controller.rs +++ b/asap-planner-rs/src/promql/controller.rs @@ -1,8 +1,5 @@ -use std::path::Path; -use tracing::debug; - use asap_types::PromQLSchema; -use promql_utilities::data_model::KeyByLabelNames; +use std::path::Path; use super::generator; use crate::config::input::ControllerConfig; @@ -26,10 +23,10 @@ impl Controller { } } - /// Build a `Controller` from a config file, fetching metric labels from Prometheus. + /// Build a `Controller` from a config file, using metric hints before Prometheus discovery. /// /// `prometheus_url` is queried via `GET /api/v1/series?match[]=` for each metric - /// name found in the config's PromQL queries. + /// name found in the config's PromQL queries that has no `metrics` hint. pub fn from_file( path: &Path, opts: RuntimeOptions, @@ -48,23 +45,18 @@ impl Controller { .iter() .flat_map(|qg| qg.queries.clone()) .collect(); - let mut schema = - prometheus_client::build_schema_from_prometheus(prometheus_url, &all_queries)?; - // For any metric that Prometheus had no series for, fall back to the - // `metrics` hint in the config file (if present). - if let Some(metric_hints) = &config.metrics { - for hint in metric_hints { - if !schema.config.contains_key(&hint.metric) { - debug!( - "Prometheus had no series for '{}'; falling back to config-file hint with labels {:?}", - hint.metric, hint.labels - ); - schema = schema.add_metric( - hint.metric.clone(), - KeyByLabelNames::new(hint.labels.clone()), - ); - } - } + let mut schema = config.schema_from_hints(); + let metric_names = prometheus_client::extract_metric_names(&all_queries); + let missing_metric_names = metric_names + .into_iter() + .filter(|metric_name| !schema.config.contains_key(metric_name)) + .collect(); + let discovered_schema = prometheus_client::build_schema_from_metric_names( + prometheus_url, + &missing_metric_names, + )?; + for (metric_name, labels) in discovered_schema.config { + schema = schema.add_metric(metric_name, labels); } Ok(Self { config, diff --git a/asap-planner-rs/tests/integration.rs b/asap-planner-rs/tests/integration.rs index 67b2d8af..7c864429 100644 --- a/asap-planner-rs/tests/integration.rs +++ b/asap-planner-rs/tests/integration.rs @@ -1,6 +1,8 @@ use asap_planner::{Controller, ControllerError, PromQLSchema, RuntimeOptions, StreamingEngine}; use promql_utilities::data_model::KeyByLabelNames; use promql_utilities::query_logics::enums::AggregationType; +use std::io::{Read, Write}; +use std::net::TcpListener; use std::path::Path; // ─── helpers ───────────────────────────────────────────────────────────────── @@ -28,6 +30,29 @@ fn http_requests_schema() -> PromQLSchema { ) } +fn single_request_prometheus_server( + status: &str, + body: &str, +) -> (String, std::thread::JoinHandle) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let status = status.to_string(); + let body = body.to_string(); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut request = [0u8; 4096]; + let bytes_read = stream.read(&mut request).unwrap(); + let response = format!( + "HTTP/1.1 {status}\r\nContent-Length: {}\r\nContent-Type: application/json\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ); + stream.write_all(response.as_bytes()).unwrap(); + String::from_utf8_lossy(&request[..bytes_read]).into_owned() + }); + (url, server) +} + #[test] fn config_file_sliding_window_override_generates_per_query_candidates() { let controller = Controller::from_file_with_schema( @@ -806,6 +831,129 @@ metrics: assert!(result.is_ok()); } +#[test] +fn hinted_metric_skips_prometheus_schema_discovery() { + let config = tempfile::NamedTempFile::new().unwrap(); + std::fs::write( + config.path(), + r#" +query_groups: + - id: 1 + queries: + - "rate(http_requests_total[5m])" + repetition_delay_ms: 300000 +metrics: + - metric: "http_requests_total" + labels: ["instance"] +"#, + ) + .unwrap(); + + let unused_listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let unused_port = unused_listener.local_addr().unwrap().port(); + drop(unused_listener); + + // The port has no listener. A successful construction proves the hinted + // metric did not attempt URL-based discovery. + let controller = Controller::from_file( + config.path(), + default_opts(), + &format!("http://127.0.0.1:{unused_port}"), + ) + .unwrap(); + + assert_eq!(controller.generate().unwrap().inference_query_count(), 1); +} + +#[test] +fn unhinted_metric_discovers_schema_from_prometheus() { + let config = tempfile::NamedTempFile::new().unwrap(); + std::fs::write( + config.path(), + r#" +query_groups: + - id: 1 + queries: + - "rate(http_requests_total[5m])" + repetition_delay_ms: 300000 +"#, + ) + .unwrap(); + + let (url, server) = single_request_prometheus_server( + "200 OK", + r#"{"status":"success","data":[{"__name__":"http_requests_total","instance":"one"}]}"#, + ); + + let controller = Controller::from_file(config.path(), default_opts(), &url).unwrap(); + + assert_eq!(controller.generate().unwrap().inference_query_count(), 1); + assert!(server + .join() + .unwrap() + .starts_with("GET /api/v1/series?match%5B%5D=http_requests_total")); +} + +#[test] +fn hints_remain_authoritative_while_unhinted_metrics_are_discovered() { + let config = tempfile::NamedTempFile::new().unwrap(); + std::fs::write( + config.path(), + r#" +query_groups: + - id: 1 + queries: + - "sum by (hint_label) (hinted_metric)" + - "rate(unhinted_metric[5m])" + repetition_delay_ms: 300000 +metrics: + - metric: "hinted_metric" + labels: ["hint_label"] +"#, + ) + .unwrap(); + let (url, server) = single_request_prometheus_server( + "200 OK", + r#"{"status":"success","data":[{"__name__":"unhinted_metric","instance":"one"}]}"#, + ); + + let controller = Controller::from_file(config.path(), default_opts(), &url).unwrap(); + + assert_eq!(controller.generate().unwrap().inference_query_count(), 2); + assert!(server + .join() + .unwrap() + .starts_with("GET /api/v1/series?match%5B%5D=unhinted_metric")); +} + +#[test] +fn unhinted_metric_propagates_prometheus_discovery_error() { + let config = tempfile::NamedTempFile::new().unwrap(); + std::fs::write( + config.path(), + r#" +query_groups: + - id: 1 + queries: + - "rate(http_requests_total[5m])" + repetition_delay_ms: 300000 +"#, + ) + .unwrap(); + let (url, server) = single_request_prometheus_server( + "422 Unprocessable Entity", + r#"{"status":"error","error":"too many series"}"#, + ); + + let result = Controller::from_file(config.path(), default_opts(), &url); + + assert!(matches!(result, Err(ControllerError::PrometheusClient(_)))); + assert!(server + .join() + .unwrap() + .starts_with("GET /api/v1/series?match%5B%5D=http_requests_total")); +} + // --- Overlapping window tests --- // Queries where range vector > t_repeat: e.g. [5m] repeated every 60s. // Windows are always tumbling (sliding disabled); the planner emits windowSize=t_repeat