Skip to content

Commit a9df958

Browse files
nullrendsmith3197StephenWakely
authored
enhancement(prometheus_scrape source): run requests in parallel with timeouts (vectordotdev#18021)
<!-- **Your PR title must conform to the conventional commit spec!** <type>(<scope>)!: <description> * `type` = chore, enhancement, feat, fix, docs * `!` = OPTIONAL: signals a breaking change * `scope` = Optional when `type` is "chore" or "docs", available scopes https://github.com/vectordotdev/vector/blob/master/.github/semantic.yml#L20 * `description` = short description of the change Examples: * enhancement(file source): Add `sort` option to sort discovered files * feat(new source): Initial `statsd` source * fix(file source): Fix a bug discovering new files * chore(external docs): Clarify `batch_size` option --> fixes vectordotdev#14087 fixes vectordotdev#14132 fixes vectordotdev#17659 - [x] make target timeout configurable this builds on what @wjordan did in vectordotdev#17660 ### what's changed - prometheus scrapes happen concurrently - requests to targets can timeout - the timeout can be configured (user facing change) - small change in how the http was instantiated --------- Co-authored-by: Doug Smith <[email protected]> Co-authored-by: Stephen Wakely <[email protected]>
1 parent 684e43f commit a9df958

File tree

7 files changed

+124
-17
lines changed

7 files changed

+124
-17
lines changed

‎src/sources/http_client/client.rs

+16-2
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,8 @@ use crate::{
2020
sources::util::{
2121
http::HttpMethod,
2222
http_client::{
23-
build_url, call, default_interval, GenericHttpClientInputs, HttpClientBuilder,
23+
build_url, call, default_interval, default_timeout, warn_if_interval_too_low,
24+
GenericHttpClientInputs, HttpClientBuilder,
2425
},
2526
},
2627
tls::{TlsConfig, TlsSettings},
@@ -51,13 +52,22 @@ pub struct HttpClientConfig {
5152
#[configurable(metadata(docs::examples = "http://127.0.0.1:9898/logs"))]
5253
pub endpoint: String,
5354

54-
/// The interval between calls.
55+
/// The interval between scrapes. Requests are run concurrently so if a scrape takes longer
56+
/// than the interval a new scrape will be started. This can take extra resources, set the timeout
57+
/// to a value lower than the scrape interval to prevent this from happening.
5558
#[serde(default = "default_interval")]
5659
#[serde_as(as = "serde_with::DurationSeconds<u64>")]
5760
#[serde(rename = "scrape_interval_secs")]
5861
#[configurable(metadata(docs::human_name = "Scrape Interval"))]
5962
pub interval: Duration,
6063

64+
/// The timeout for each scrape request.
65+
#[serde(default = "default_timeout")]
66+
#[serde_as(as = "serde_with:: DurationSecondsWithFrac<f64>")]
67+
#[serde(rename = "scrape_timeout_secs")]
68+
#[configurable(metadata(docs::human_name = "Scrape Timeout"))]
69+
pub timeout: Duration,
70+
6171
/// Custom parameters for the HTTP request query string.
6272
///
6373
/// One or more values for the same parameter key can be provided.
@@ -153,6 +163,7 @@ impl Default for HttpClientConfig {
153163
endpoint: "http://localhost:9898/logs".to_string(),
154164
query: HashMap::new(),
155165
interval: default_interval(),
166+
timeout: default_timeout(),
156167
decoding: default_decoding(),
157168
framing: default_framing_message_based(),
158169
headers: HashMap::new(),
@@ -193,9 +204,12 @@ impl SourceConfig for HttpClientConfig {
193204
log_namespace,
194205
};
195206

207+
warn_if_interval_too_low(self.timeout, self.interval);
208+
196209
let inputs = GenericHttpClientInputs {
197210
urls,
198211
interval: self.interval,
212+
timeout: self.timeout,
199213
headers: self.headers.clone(),
200214
content_type,
201215
auth: self.auth.clone(),

‎src/sources/http_client/integration_tests.rs

+12-1
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ use codecs::decoding::DeserializerConfig;
1919
use vector_core::config::log_schema;
2020

2121
use super::{
22-
tests::{run_compliance, INTERVAL},
22+
tests::{run_compliance, INTERVAL, TIMEOUT},
2323
HttpClientConfig,
2424
};
2525

@@ -53,6 +53,7 @@ async fn invalid_endpoint() {
5353
run_error(HttpClientConfig {
5454
endpoint: "http://nope".to_string(),
5555
interval: INTERVAL,
56+
timeout: TIMEOUT,
5657
query: HashMap::new(),
5758
decoding: default_decoding(),
5859
framing: default_framing_message_based(),
@@ -71,6 +72,7 @@ async fn collected_logs_bytes() {
7172
let events = run_compliance(HttpClientConfig {
7273
endpoint: format!("{}/logs/bytes", dufs_address()),
7374
interval: INTERVAL,
75+
timeout: TIMEOUT,
7476
query: HashMap::new(),
7577
decoding: DeserializerConfig::Bytes,
7678
framing: default_framing_message_based(),
@@ -95,6 +97,7 @@ async fn collected_logs_json() {
9597
let events = run_compliance(HttpClientConfig {
9698
endpoint: format!("{}/logs/json.json", dufs_address()),
9799
interval: INTERVAL,
100+
timeout: TIMEOUT,
98101
query: HashMap::new(),
99102
decoding: DeserializerConfig::Json(Default::default()),
100103
framing: default_framing_message_based(),
@@ -119,6 +122,7 @@ async fn collected_metrics_native_json() {
119122
let events = run_compliance(HttpClientConfig {
120123
endpoint: format!("{}/metrics/native.json", dufs_address()),
121124
interval: INTERVAL,
125+
timeout: TIMEOUT,
122126
query: HashMap::new(),
123127
decoding: DeserializerConfig::NativeJson(Default::default()),
124128
framing: default_framing_message_based(),
@@ -148,6 +152,7 @@ async fn collected_trace_native_json() {
148152
let events = run_compliance(HttpClientConfig {
149153
endpoint: format!("{}/traces/native.json", dufs_address()),
150154
interval: INTERVAL,
155+
timeout: TIMEOUT,
151156
query: HashMap::new(),
152157
decoding: DeserializerConfig::NativeJson(Default::default()),
153158
framing: default_framing_message_based(),
@@ -172,6 +177,7 @@ async fn unauthorized_no_auth() {
172177
run_error(HttpClientConfig {
173178
endpoint: format!("{}/logs/json.json", dufs_auth_address()),
174179
interval: INTERVAL,
180+
timeout: TIMEOUT,
175181
query: HashMap::new(),
176182
decoding: DeserializerConfig::Json(Default::default()),
177183
framing: default_framing_message_based(),
@@ -190,6 +196,7 @@ async fn unauthorized_wrong_auth() {
190196
run_error(HttpClientConfig {
191197
endpoint: format!("{}/logs/json.json", dufs_auth_address()),
192198
interval: INTERVAL,
199+
timeout: TIMEOUT,
193200
query: HashMap::new(),
194201
decoding: DeserializerConfig::Json(Default::default()),
195202
framing: default_framing_message_based(),
@@ -211,6 +218,7 @@ async fn authorized() {
211218
run_compliance(HttpClientConfig {
212219
endpoint: format!("{}/logs/json.json", dufs_auth_address()),
213220
interval: INTERVAL,
221+
timeout: TIMEOUT,
214222
query: HashMap::new(),
215223
decoding: DeserializerConfig::Json(Default::default()),
216224
framing: default_framing_message_based(),
@@ -232,6 +240,7 @@ async fn tls_invalid_ca() {
232240
run_error(HttpClientConfig {
233241
endpoint: format!("{}/logs/json.json", dufs_https_address()),
234242
interval: INTERVAL,
243+
timeout: TIMEOUT,
235244
query: HashMap::new(),
236245
decoding: DeserializerConfig::Json(Default::default()),
237246
framing: default_framing_message_based(),
@@ -253,6 +262,7 @@ async fn tls_valid() {
253262
run_compliance(HttpClientConfig {
254263
endpoint: format!("{}/logs/json.json", dufs_https_address()),
255264
interval: INTERVAL,
265+
timeout: TIMEOUT,
256266
query: HashMap::new(),
257267
decoding: DeserializerConfig::Json(Default::default()),
258268
framing: default_framing_message_based(),
@@ -275,6 +285,7 @@ async fn shutdown() {
275285
let source = HttpClientConfig {
276286
endpoint: format!("{}/logs/json.json", dufs_address()),
277287
interval: INTERVAL,
288+
timeout: TIMEOUT,
278289
query: HashMap::new(),
279290
decoding: DeserializerConfig::Json(Default::default()),
280291
framing: default_framing_message_based(),

‎src/sources/http_client/tests.rs

+8
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@ use crate::test_util::{
1616

1717
pub(crate) const INTERVAL: Duration = Duration::from_secs(1);
1818

19+
pub(crate) const TIMEOUT: Duration = Duration::from_secs(1);
20+
1921
/// The happy path should yield at least one event and must emit the required internal events for sources.
2022
pub(crate) async fn run_compliance(config: HttpClientConfig) -> Vec<Event> {
2123
let events =
@@ -47,6 +49,7 @@ async fn bytes_decoding() {
4749
run_compliance(HttpClientConfig {
4850
endpoint: format!("http://{}/endpoint", in_addr),
4951
interval: INTERVAL,
52+
timeout: TIMEOUT,
5053
query: HashMap::new(),
5154
decoding: default_decoding(),
5255
framing: default_framing_message_based(),
@@ -75,6 +78,7 @@ async fn json_decoding_newline_delimited() {
7578
run_compliance(HttpClientConfig {
7679
endpoint: format!("http://{}/endpoint", in_addr),
7780
interval: INTERVAL,
81+
timeout: TIMEOUT,
7882
query: HashMap::new(),
7983
decoding: DeserializerConfig::Json(Default::default()),
8084
framing: FramingConfig::NewlineDelimited(Default::default()),
@@ -103,6 +107,7 @@ async fn json_decoding_character_delimited() {
103107
run_compliance(HttpClientConfig {
104108
endpoint: format!("http://{}/endpoint", in_addr),
105109
interval: INTERVAL,
110+
timeout: TIMEOUT,
106111
query: HashMap::new(),
107112
decoding: DeserializerConfig::Json(Default::default()),
108113
framing: FramingConfig::CharacterDelimited(CharacterDelimitedDecoderConfig {
@@ -135,6 +140,7 @@ async fn request_query_applied() {
135140
let events = run_compliance(HttpClientConfig {
136141
endpoint: format!("http://{}/endpoint?key1=val1", in_addr),
137142
interval: INTERVAL,
143+
timeout: TIMEOUT,
138144
query: HashMap::from([
139145
("key1".to_string(), vec!["val2".to_string()]),
140146
(
@@ -203,6 +209,7 @@ async fn headers_applied() {
203209
run_compliance(HttpClientConfig {
204210
endpoint: format!("http://{}/endpoint", in_addr),
205211
interval: INTERVAL,
212+
timeout: TIMEOUT,
206213
query: HashMap::new(),
207214
decoding: default_decoding(),
208215
framing: default_framing_message_based(),
@@ -234,6 +241,7 @@ async fn accept_header_override() {
234241
run_compliance(HttpClientConfig {
235242
endpoint: format!("http://{}/endpoint", in_addr),
236243
interval: INTERVAL,
244+
timeout: TIMEOUT,
237245
query: HashMap::new(),
238246
decoding: DeserializerConfig::Bytes,
239247
framing: default_framing_message_based(),

‎src/sources/prometheus/scrape.rs

+22-1
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ use vector_core::{config::LogNamespace, event::Event};
1111

1212
use super::parser;
1313
use crate::sources::util::http::HttpMethod;
14+
use crate::sources::util::http_client::{default_timeout, warn_if_interval_too_low};
1415
use crate::{
1516
config::{GenerateConfig, SourceConfig, SourceContext, SourceOutput},
1617
http::Auth,
@@ -53,13 +54,22 @@ pub struct PrometheusScrapeConfig {
5354
#[serde(alias = "hosts")]
5455
endpoints: Vec<String>,
5556

56-
/// The interval between scrapes, in seconds.
57+
/// The interval between scrapes. Requests are run concurrently so if a scrape takes longer
58+
/// than the interval a new scrape will be started. This can take extra resources, set the timeout
59+
/// to a value lower than the scrape interval to prevent this from happening.
5760
#[serde(default = "default_interval")]
5861
#[serde_as(as = "serde_with::DurationSeconds<u64>")]
5962
#[serde(rename = "scrape_interval_secs")]
6063
#[configurable(metadata(docs::human_name = "Scrape Interval"))]
6164
interval: Duration,
6265

66+
/// The timeout for each scrape request.
67+
#[serde(default = "default_timeout")]
68+
#[serde_as(as = "serde_with:: DurationSecondsWithFrac<f64>")]
69+
#[serde(rename = "scrape_timeout_secs")]
70+
#[configurable(metadata(docs::human_name = "Scrape Timeout"))]
71+
timeout: Duration,
72+
6373
/// The tag name added to each event representing the scraped instance's `host:port`.
6474
///
6575
/// The tag value is the host and port of the scraped instance.
@@ -114,6 +124,7 @@ impl GenerateConfig for PrometheusScrapeConfig {
114124
toml::Value::try_from(Self {
115125
endpoints: vec!["http://localhost:9090/metrics".to_string()],
116126
interval: default_interval(),
127+
timeout: default_timeout(),
117128
instance_tag: Some("instance".to_string()),
118129
endpoint_tag: Some("endpoint".to_string()),
119130
honor_labels: false,
@@ -143,9 +154,12 @@ impl SourceConfig for PrometheusScrapeConfig {
143154
endpoint_tag: self.endpoint_tag.clone(),
144155
};
145156

157+
warn_if_interval_too_low(self.timeout, self.interval);
158+
146159
let inputs = GenericHttpClientInputs {
147160
urls,
148161
interval: self.interval,
162+
timeout: self.timeout,
149163
headers: HashMap::new(),
150164
content_type: "text/plain".to_string(),
151165
auth: self.auth.clone(),
@@ -351,6 +365,7 @@ mod test {
351365
let config = PrometheusScrapeConfig {
352366
endpoints: vec![format!("http://{}/metrics", in_addr)],
353367
interval: Duration::from_secs(1),
368+
timeout: default_timeout(),
354369
instance_tag: Some("instance".to_string()),
355370
endpoint_tag: Some("endpoint".to_string()),
356371
honor_labels: true,
@@ -384,6 +399,7 @@ mod test {
384399
let config = PrometheusScrapeConfig {
385400
endpoints: vec![format!("http://{}/metrics", in_addr)],
386401
interval: Duration::from_secs(1),
402+
timeout: default_timeout(),
387403
instance_tag: Some("instance".to_string()),
388404
endpoint_tag: Some("endpoint".to_string()),
389405
honor_labels: true,
@@ -435,6 +451,7 @@ mod test {
435451
let config = PrometheusScrapeConfig {
436452
endpoints: vec![format!("http://{}/metrics", in_addr)],
437453
interval: Duration::from_secs(1),
454+
timeout: default_timeout(),
438455
instance_tag: Some("instance".to_string()),
439456
endpoint_tag: Some("endpoint".to_string()),
440457
honor_labels: false,
@@ -500,6 +517,7 @@ mod test {
500517
let config = PrometheusScrapeConfig {
501518
endpoints: vec![format!("http://{}/metrics", in_addr)],
502519
interval: Duration::from_secs(1),
520+
timeout: default_timeout(),
503521
instance_tag: Some("instance".to_string()),
504522
endpoint_tag: Some("endpoint".to_string()),
505523
honor_labels: true,
@@ -555,6 +573,7 @@ mod test {
555573
let config = PrometheusScrapeConfig {
556574
endpoints: vec![format!("http://{}/metrics?key1=val1", in_addr)],
557575
interval: Duration::from_secs(1),
576+
timeout: default_timeout(),
558577
instance_tag: Some("instance".to_string()),
559578
endpoint_tag: Some("endpoint".to_string()),
560579
honor_labels: false,
@@ -668,6 +687,7 @@ mod test {
668687
honor_labels: false,
669688
query: HashMap::new(),
670689
interval: Duration::from_secs(1),
690+
timeout: default_timeout(),
671691
tls: None,
672692
auth: None,
673693
},
@@ -753,6 +773,7 @@ mod integration_tests {
753773
let config = PrometheusScrapeConfig {
754774
endpoints: vec!["http://prometheus:9090/metrics".into()],
755775
interval: Duration::from_secs(1),
776+
timeout: Duration::from_secs(1),
756777
instance_tag: Some("instance".to_string()),
757778
endpoint_tag: Some("endpoint".to_string()),
758779
honor_labels: false,

0 commit comments

Comments
 (0)