Skip to content

Commit e319e11

Browse files
ardatanardatan
committed
Rebase
1 parent c1b0e3d commit e319e11

File tree

4 files changed

+60
-27
lines changed

4 files changed

+60
-27
lines changed

‎lib/executor/src/executors/error.rs‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,10 @@ pub enum SubgraphExecutorError {
2020
RequestFailure(String, String),
2121
#[error("Failed to serialize variable \"{0}\": {1}")]
2222
VariablesSerializationFailure(String, String),
23-
#[error("Failed to resolve VRL expression for timeout. Runtime error: {0}")]
24-
TimeoutExpressionResolutionFailure(String),
23+
#[error("Failed to compile VRL expression for timeout for subgraph '{0}'. Please check your VRL expression for syntax errors. Diagnostic: {1}")]
24+
RequestTimeoutExpressionBuild(String, String),
25+
#[error("Failed to resolve VRL expression for timeout for subgraph '{0}'. Runtime error: {1}")]
26+
TimeoutExpressionResolution(String, String),
2527
#[error("Request to subgraph \"{0}\" timed out after {1} milliseconds")]
2628
RequestTimeout(String, u64),
2729
}
@@ -65,10 +67,13 @@ impl SubgraphExecutorError {
6567
SubgraphExecutorError::VariablesSerializationFailure(_, _) => {
6668
"SUBGRAPH_VARIABLES_SERIALIZATION_FAILURE"
6769
}
68-
SubgraphExecutorError::TimeoutExpressionResolutionFailure(_) => {
70+
SubgraphExecutorError::TimeoutExpressionResolution(_, _) => {
6971
"SUBGRAPH_TIMEOUT_EXPRESSION_RESOLUTION_FAILURE"
7072
}
7173
SubgraphExecutorError::RequestTimeout(_, _) => "SUBGRAPH_REQUEST_TIMEOUT",
74+
SubgraphExecutorError::RequestTimeoutExpressionBuild(_, _) => {
75+
"SUBGRAPH_TIMEOUT_EXPRESSION_BUILD_FAILURE"
76+
}
7277
}
7378
}
7479
}

‎lib/executor/src/executors/http.rs‎

Lines changed: 32 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,12 @@
11
use std::collections::BTreeMap;
22
use std::sync::Arc;
3+
use std::time::Duration;
34

45
use crate::execution::client_request_details::ClientRequestDetails;
56
use crate::executors::common::HttpExecutionResponse;
67
use crate::executors::dedupe::{request_fingerprint, ABuildHasher, SharedResponse};
8+
use crate::utils::expression::execute_expression_with_value;
79
use dashmap::DashMap;
8-
use hive_router_config::traffic_shaping::DurationOrExpression;
910
use tokio::sync::OnceCell;
1011

1112
use async_trait::async_trait;
@@ -20,6 +21,7 @@ use hyper_tls::HttpsConnector;
2021
use hyper_util::client::legacy::{connect::HttpConnector, Client};
2122
use tokio::sync::Semaphore;
2223
use tracing::debug;
24+
use vrl::compiler::Program as VrlProgram;
2325
use vrl::core::Value as VrlValue;
2426

2527
use crate::executors::common::HttpExecutionRequest;
@@ -31,6 +33,12 @@ use crate::utils::consts::COMMA;
3133
use crate::utils::consts::QUOTE;
3234
use crate::{executors::common::SubgraphExecutor, json_writer::write_and_escape_string};
3335

36+
#[derive(Debug)]
37+
pub enum DurationOrProgram {
38+
Duration(Duration),
39+
Program(Box<VrlProgram>),
40+
}
41+
3442
#[derive(Debug)]
3543
pub struct HTTPSubgraphExecutor {
3644
pub subgraph_name: String,
@@ -40,7 +48,7 @@ pub struct HTTPSubgraphExecutor {
4048
pub semaphore: Arc<Semaphore>,
4149
pub dedupe_enabled: bool,
4250
pub in_flight_requests: Arc<DashMap<u64, Arc<OnceCell<SharedResponse>>, ABuildHasher>>,
43-
pub timeout: DurationOrExpression,
51+
pub timeout: DurationOrProgram,
4452
}
4553

4654
const FIRST_VARIABLE_STR: &[u8] = b",\"variables\":{";
@@ -56,7 +64,7 @@ impl HTTPSubgraphExecutor {
5664
semaphore: Arc<Semaphore>,
5765
dedupe_enabled: bool,
5866
in_flight_requests: Arc<DashMap<u64, Arc<OnceCell<SharedResponse>>, ABuildHasher>>,
59-
timeout: DurationOrExpression,
67+
timeout: DurationOrProgram,
6068
) -> Self {
6169
let mut header_map = HeaderMap::new();
6270
header_map.insert(
@@ -159,17 +167,21 @@ impl HTTPSubgraphExecutor {
159167
debug!("making http request to {}", self.endpoint.to_string());
160168

161169
let timeout = match &self.timeout {
162-
DurationOrExpression::Duration(dur) => *dur,
163-
DurationOrExpression::Expression(expr) => {
170+
DurationOrProgram::Duration(dur) => *dur,
171+
DurationOrProgram::Program(program) => {
164172
let value =
165173
VrlValue::Object(BTreeMap::from([("request".into(), client_request.into())]));
166-
let result = expr.execute_with_value(value).map_err(|err| {
167-
SubgraphExecutorError::TimeoutExpressionResolutionFailure(err.to_string())
174+
let result = execute_expression_with_value(program, value).map_err(|err| {
175+
SubgraphExecutorError::TimeoutExpressionResolution(
176+
self.subgraph_name.to_string(),
177+
err.to_string(),
178+
)
168179
})?;
169180
match result {
170181
VrlValue::Integer(i) => {
171182
if i < 0 {
172-
return Err(SubgraphExecutorError::TimeoutExpressionResolutionFailure(
183+
return Err(SubgraphExecutorError::TimeoutExpressionResolution(
184+
self.subgraph_name.to_string(),
173185
"Timeout expression resolved to a negative integer".to_string(),
174186
));
175187
}
@@ -178,28 +190,30 @@ impl HTTPSubgraphExecutor {
178190
VrlValue::Float(f) => {
179191
let f = f.into_inner();
180192
if f < 0.0 {
181-
return Err(SubgraphExecutorError::TimeoutExpressionResolutionFailure(
193+
return Err(SubgraphExecutorError::TimeoutExpressionResolution(
194+
self.subgraph_name.to_string(),
182195
"Timeout expression resolved to a negative float".to_string(),
183196
));
184197
}
185198
std::time::Duration::from_millis(f as u64)
186199
}
187200
VrlValue::Bytes(b) => {
188201
let s = std::str::from_utf8(&b).map_err(|e| {
189-
SubgraphExecutorError::TimeoutExpressionResolutionFailure(format!(
190-
"Failed to parse duration string from bytes: {}",
191-
e
192-
))
202+
SubgraphExecutorError::TimeoutExpressionResolution(
203+
self.subgraph_name.to_string(),
204+
format!("Failed to parse duration string from bytes: {}", e),
205+
)
193206
})?;
194207
humantime::parse_duration(s).map_err(|e| {
195-
SubgraphExecutorError::TimeoutExpressionResolutionFailure(format!(
196-
"Failed to parse duration string '{}': {}",
197-
s, e
198-
))
208+
SubgraphExecutorError::TimeoutExpressionResolution(
209+
self.subgraph_name.to_string(),
210+
format!("Failed to parse duration string '{}': {}", s, e),
211+
)
199212
})?
200213
}
201214
other => {
202-
return Err(SubgraphExecutorError::TimeoutExpressionResolutionFailure(
215+
return Err(SubgraphExecutorError::TimeoutExpressionResolution(
216+
self.subgraph_name.to_string(),
203217
format!(
204218
"Timeout expression resolved to an unexpected type: {}. Expected a non-negative integer/float (ms) or a duration string.",
205219
other.kind()

‎lib/executor/src/executors/map.rs‎

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,10 @@ use std::{
66

77
use bytes::{BufMut, Bytes, BytesMut};
88
use dashmap::DashMap;
9-
use hive_router_config::{override_subgraph_urls::UrlOrExpression, HiveRouterConfig};
9+
use hive_router_config::{
10+
override_subgraph_urls::UrlOrExpression, traffic_shaping::DurationOrExpression,
11+
HiveRouterConfig,
12+
};
1013
use http::Uri;
1114
use hyper_tls::HttpsConnector;
1215
use hyper_util::{
@@ -25,7 +28,7 @@ use crate::{
2528
},
2629
dedupe::{ABuildHasher, SharedResponse},
2730
error::SubgraphExecutorError,
28-
http::{HTTPSubgraphExecutor, HttpClient},
31+
http::{DurationOrProgram, HTTPSubgraphExecutor, HttpClient},
2932
},
3033
response::graphql_error::GraphQLError,
3134
utils::expression::{compile_expression, execute_expression_with_value},
@@ -324,14 +327,27 @@ impl SubgraphExecutorMap {
324327
.unwrap_or(timeout);
325328
}
326329

330+
let timeout = match &timeout {
331+
DurationOrExpression::Duration(dur) => DurationOrProgram::Duration(*dur),
332+
DurationOrExpression::Expression { expression } => {
333+
let program = compile_expression(expression, None).map_err(|err| {
334+
SubgraphExecutorError::RequestTimeoutExpressionBuild(
335+
subgraph_name.to_string(),
336+
err,
337+
)
338+
})?;
339+
DurationOrProgram::Program(Box::new(program))
340+
}
341+
};
342+
327343
let executor = HTTPSubgraphExecutor::new(
328344
subgraph_name.to_string(),
329345
endpoint_uri,
330346
client,
331347
semaphore,
332348
dedupe_enabled,
333349
self.in_flight_requests.clone(),
334-
timeout.clone(),
350+
timeout,
335351
);
336352

337353
self.executors_by_subgraph

‎lib/router-config/src/traffic_shaping.rs‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,8 +3,6 @@ use std::{collections::HashMap, time::Duration};
33
use schemars::JsonSchema;
44
use serde::{Deserialize, Serialize};
55

6-
use crate::primitives::expression::Expression;
7-
86
#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
97
#[serde(deny_unknown_fields)]
108
pub struct TrafficShapingConfig {
@@ -123,7 +121,7 @@ pub enum DurationOrExpression {
123121
)]
124122
Duration(Duration),
125123
/// A VRL expression that evaluates to a duration. The result can be an integer (milliseconds), a float (milliseconds), or a duration string (e.g. "5s").
126-
Expression(Expression),
124+
Expression { expression: String },
127125
}
128126

129127
impl Default for TrafficShapingExecutorConfig {

0 commit comments

Comments
 (0)