Skip to content

Commit 633fbeb

Browse files
fix(planner): prioritize configured metric schemas (#731)
1 parent 69f0b59 commit 633fbeb

4 files changed

Lines changed: 174 additions & 26 deletions

File tree

‎asap-planner-rs/src/config/input.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,8 @@ pub struct ControllerConfig {
1313
pub windowing: Option<WindowingConfig>,
1414
pub sketch_parameters: Option<SketchParameterOverrides>,
1515
pub aggregate_cleanup: Option<AggregateCleanupConfig>,
16-
/// Optional hint: per-metric label sets used as a fallback when Prometheus
17-
/// returns no series for a metric. Prometheus-inferred labels take priority.
16+
/// Optional authoritative per-metric label sets. Prometheus discovery fills
17+
/// only metrics that have no hint.
1818
#[serde(default)]
1919
pub metrics: Option<Vec<MetricDefinition>>,
2020
/// Current streaming config, passed as context for repeated reconfiguration.

‎asap-planner-rs/src/prometheus_client.rs‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -165,10 +165,18 @@ pub fn build_schema_from_prometheus(
165165
queries: &[String],
166166
) -> Result<PromQLSchema, ControllerError> {
167167
let metric_names = extract_metric_names(queries);
168+
build_schema_from_metric_names(prometheus_url, &metric_names)
169+
}
170+
171+
/// Build a `PromQLSchema` by querying Prometheus for the supplied metric names.
172+
pub(crate) fn build_schema_from_metric_names(
173+
prometheus_url: &str,
174+
metric_names: &HashSet<String>,
175+
) -> Result<PromQLSchema, ControllerError> {
168176
debug!("Inferred metric names from queries: {:?}", metric_names);
169177
let mut schema = PromQLSchema::new();
170178

171-
for metric_name in &metric_names {
179+
for metric_name in metric_names {
172180
match fetch_labels_for_metric(prometheus_url, metric_name)? {
173181
Some(labels) => {
174182
debug!("Inferred labels for metric '{}': {:?}", metric_name, labels);

‎asap-planner-rs/src/promql/controller.rs‎

Lines changed: 15 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,5 @@
1-
use std::path::Path;
2-
use tracing::debug;
3-
41
use asap_types::PromQLSchema;
5-
use promql_utilities::data_model::KeyByLabelNames;
2+
use std::path::Path;
63

74
use super::generator;
85
use crate::config::input::ControllerConfig;
@@ -26,10 +23,10 @@ impl Controller {
2623
}
2724
}
2825

29-
/// Build a `Controller` from a config file, fetching metric labels from Prometheus.
26+
/// Build a `Controller` from a config file, using metric hints before Prometheus discovery.
3027
///
3128
/// `prometheus_url` is queried via `GET /api/v1/series?match[]=<metric>` for each metric
32-
/// name found in the config's PromQL queries.
29+
/// name found in the config's PromQL queries that has no `metrics` hint.
3330
pub fn from_file(
3431
path: &Path,
3532
opts: RuntimeOptions,
@@ -48,23 +45,18 @@ impl Controller {
4845
.iter()
4946
.flat_map(|qg| qg.queries.clone())
5047
.collect();
51-
let mut schema =
52-
prometheus_client::build_schema_from_prometheus(prometheus_url, &all_queries)?;
53-
// For any metric that Prometheus had no series for, fall back to the
54-
// `metrics` hint in the config file (if present).
55-
if let Some(metric_hints) = &config.metrics {
56-
for hint in metric_hints {
57-
if !schema.config.contains_key(&hint.metric) {
58-
debug!(
59-
"Prometheus had no series for '{}'; falling back to config-file hint with labels {:?}",
60-
hint.metric, hint.labels
61-
);
62-
schema = schema.add_metric(
63-
hint.metric.clone(),
64-
KeyByLabelNames::new(hint.labels.clone()),
65-
);
66-
}
67-
}
48+
let mut schema = config.schema_from_hints();
49+
let metric_names = prometheus_client::extract_metric_names(&all_queries);
50+
let missing_metric_names = metric_names
51+
.into_iter()
52+
.filter(|metric_name| !schema.config.contains_key(metric_name))
53+
.collect();
54+
let discovered_schema = prometheus_client::build_schema_from_metric_names(
55+
prometheus_url,
56+
&missing_metric_names,
57+
)?;
58+
for (metric_name, labels) in discovered_schema.config {
59+
schema = schema.add_metric(metric_name, labels);
6860
}
6961
Ok(Self {
7062
config,

‎asap-planner-rs/tests/integration.rs‎

Lines changed: 148 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
use asap_planner::{Controller, ControllerError, PromQLSchema, RuntimeOptions, StreamingEngine};
22
use promql_utilities::data_model::KeyByLabelNames;
33
use promql_utilities::query_logics::enums::AggregationType;
4+
use std::io::{Read, Write};
5+
use std::net::TcpListener;
46
use std::path::Path;
57

68
// ─── helpers ─────────────────────────────────────────────────────────────────
@@ -28,6 +30,29 @@ fn http_requests_schema() -> PromQLSchema {
2830
)
2931
}
3032

33+
fn single_request_prometheus_server(
34+
status: &str,
35+
body: &str,
36+
) -> (String, std::thread::JoinHandle<String>) {
37+
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
38+
let url = format!("http://{}", listener.local_addr().unwrap());
39+
let status = status.to_string();
40+
let body = body.to_string();
41+
let server = std::thread::spawn(move || {
42+
let (mut stream, _) = listener.accept().unwrap();
43+
let mut request = [0u8; 4096];
44+
let bytes_read = stream.read(&mut request).unwrap();
45+
let response = format!(
46+
"HTTP/1.1 {status}\r\nContent-Length: {}\r\nContent-Type: application/json\r\nConnection: close\r\n\r\n{}",
47+
body.len(),
48+
body
49+
);
50+
stream.write_all(response.as_bytes()).unwrap();
51+
String::from_utf8_lossy(&request[..bytes_read]).into_owned()
52+
});
53+
(url, server)
54+
}
55+
3156
#[test]
3257
fn config_file_sliding_window_override_generates_per_query_candidates() {
3358
let controller = Controller::from_file_with_schema(
@@ -806,6 +831,129 @@ metrics:
806831
assert!(result.is_ok());
807832
}
808833

834+
#[test]
835+
fn hinted_metric_skips_prometheus_schema_discovery() {
836+
let config = tempfile::NamedTempFile::new().unwrap();
837+
std::fs::write(
838+
config.path(),
839+
r#"
840+
query_groups:
841+
- id: 1
842+
queries:
843+
- "rate(http_requests_total[5m])"
844+
repetition_delay_ms: 300000
845+
metrics:
846+
- metric: "http_requests_total"
847+
labels: ["instance"]
848+
"#,
849+
)
850+
.unwrap();
851+
852+
let unused_listener = TcpListener::bind("127.0.0.1:0").unwrap();
853+
let unused_port = unused_listener.local_addr().unwrap().port();
854+
drop(unused_listener);
855+
856+
// The port has no listener. A successful construction proves the hinted
857+
// metric did not attempt URL-based discovery.
858+
let controller = Controller::from_file(
859+
config.path(),
860+
default_opts(),
861+
&format!("http://127.0.0.1:{unused_port}"),
862+
)
863+
.unwrap();
864+
865+
assert_eq!(controller.generate().unwrap().inference_query_count(), 1);
866+
}
867+
868+
#[test]
869+
fn unhinted_metric_discovers_schema_from_prometheus() {
870+
let config = tempfile::NamedTempFile::new().unwrap();
871+
std::fs::write(
872+
config.path(),
873+
r#"
874+
query_groups:
875+
- id: 1
876+
queries:
877+
- "rate(http_requests_total[5m])"
878+
repetition_delay_ms: 300000
879+
"#,
880+
)
881+
.unwrap();
882+
883+
let (url, server) = single_request_prometheus_server(
884+
"200 OK",
885+
r#"{"status":"success","data":[{"__name__":"http_requests_total","instance":"one"}]}"#,
886+
);
887+
888+
let controller = Controller::from_file(config.path(), default_opts(), &url).unwrap();
889+
890+
assert_eq!(controller.generate().unwrap().inference_query_count(), 1);
891+
assert!(server
892+
.join()
893+
.unwrap()
894+
.starts_with("GET /api/v1/series?match%5B%5D=http_requests_total"));
895+
}
896+
897+
#[test]
898+
fn hints_remain_authoritative_while_unhinted_metrics_are_discovered() {
899+
let config = tempfile::NamedTempFile::new().unwrap();
900+
std::fs::write(
901+
config.path(),
902+
r#"
903+
query_groups:
904+
- id: 1
905+
queries:
906+
- "sum by (hint_label) (hinted_metric)"
907+
- "rate(unhinted_metric[5m])"
908+
repetition_delay_ms: 300000
909+
metrics:
910+
- metric: "hinted_metric"
911+
labels: ["hint_label"]
912+
"#,
913+
)
914+
.unwrap();
915+
let (url, server) = single_request_prometheus_server(
916+
"200 OK",
917+
r#"{"status":"success","data":[{"__name__":"unhinted_metric","instance":"one"}]}"#,
918+
);
919+
920+
let controller = Controller::from_file(config.path(), default_opts(), &url).unwrap();
921+
922+
assert_eq!(controller.generate().unwrap().inference_query_count(), 2);
923+
assert!(server
924+
.join()
925+
.unwrap()
926+
.starts_with("GET /api/v1/series?match%5B%5D=unhinted_metric"));
927+
}
928+
929+
#[test]
930+
fn unhinted_metric_propagates_prometheus_discovery_error() {
931+
let config = tempfile::NamedTempFile::new().unwrap();
932+
std::fs::write(
933+
config.path(),
934+
r#"
935+
query_groups:
936+
- id: 1
937+
queries:
938+
- "rate(http_requests_total[5m])"
939+
repetition_delay_ms: 300000
940+
"#,
941+
)
942+
.unwrap();
943+
let (url, server) = single_request_prometheus_server(
944+
"422 Unprocessable Entity",
945+
r#"{"status":"error","error":"too many series"}"#,
946+
);
947+
948+
let result = Controller::from_file(config.path(), default_opts(), &url);
949+
950+
assert!(matches!(result, Err(ControllerError::PrometheusClient(_))));
951+
assert!(server
952+
.join()
953+
.unwrap()
954+
.starts_with("GET /api/v1/series?match%5B%5D=http_requests_total"));
955+
}
956+
809957
// --- Overlapping window tests ---
810958
// Queries where range vector > t_repeat: e.g. [5m] repeated every 60s.
811959
// Windows are always tumbling (sliding disabled); the planner emits windowSize=t_repeat

0 commit comments

Comments
 (0)