Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions asap-planner-rs/src/config/input.rs
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,8 @@ pub struct ControllerConfig {
pub windowing: Option<WindowingConfig>,
pub sketch_parameters: Option<SketchParameterOverrides>,
pub aggregate_cleanup: Option<AggregateCleanupConfig>,
/// 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<Vec<MetricDefinition>>,
/// Current streaming config, passed as context for repeated reconfiguration.
Expand Down
10 changes: 9 additions & 1 deletion asap-planner-rs/src/prometheus_client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -165,10 +165,18 @@ pub fn build_schema_from_prometheus(
queries: &[String],
) -> Result<PromQLSchema, ControllerError> {
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<String>,
) -> Result<PromQLSchema, ControllerError> {
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);
Expand Down
38 changes: 15 additions & 23 deletions asap-planner-rs/src/promql/controller.rs
Original file line number Diff line number Diff line change
@@ -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;
Expand All @@ -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[]=<metric>` 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,
Expand All @@ -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,
Expand Down
148 changes: 148 additions & 0 deletions asap-planner-rs/tests/integration.rs
Original file line number Diff line number Diff line change
@@ -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 ─────────────────────────────────────────────────────────────────
Expand Down Expand Up @@ -28,6 +30,29 @@ fn http_requests_schema() -> PromQLSchema {
)
}

fn single_request_prometheus_server(
status: &str,
body: &str,
) -> (String, std::thread::JoinHandle<String>) {
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(
Expand Down Expand Up @@ -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
Expand Down
Loading