feat(router): Subgraph endpoint overrides (#488)

kamilkisiela · web-flow · commit 39784f7b8053 · 2025-10-16T14:51:02.000+02:00
This adds a way to override subgraph URLs at runtime - either statically or dynamically per request using expressions. ```yaml override_subgraph_urls: accounts: url: expression: | if .request.headers."x-region" == "eu" { "https://eu.example.com/accounts" } else { .original_url # represents the subgraph url defined in supergraph sdl } products: url: "https://example.com/products" ``` --- I also slightly changed the e2e testing setup as I was facing problems with race conditions. --- Fixes #461 Closes #471 Ref ROUTER-149
diff --git a/.cargo/config.toml b/.cargo/config.toml
@@ -3,9 +3,9 @@
 router = "run --package hive-router"
 dev = "run --package qp-dev-cli"
 subgraphs = "run --package subgraphs"
-test_all = "test --workspace"
 test_qp = "test --package hive-router-query-planner -- --nocapture"
-test_e2e = "test --package e2e -- --nocapture"
+test_all = "test --workspace --exclude e2e"
+test_e2e = "test --package e2e  --jobs 1 -- --nocapture --test-threads=1"
 test_qpe = "test --package hive-router-plan-executor -- --nocapture"
 "clippy:fix" = "clippy --all --fix --allow-dirty --allow-staged"
 "router-config" = "run --release -p hive-router-config router-config.schema.json"
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
@@ -24,6 +24,7 @@ jobs:
       - uses: actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 # v5
       - uses: actions-rust-lang/setup-rust-toolchain@1780873c7b576612439a134613cc4cc74ce5538c # v1
       - run: cargo test_all
+      - run: cargo test_e2e
 
   build-release:
     name: cargo build
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/bench/subgraphs/lib.rs b/bench/subgraphs/lib.rs
@@ -95,6 +95,7 @@ pub struct RequestLog {
 #[derive(Clone)]
 pub struct SubgraphsServiceState {
     pub request_log: Arc<Mutex<HashMap<String, Vec<RequestLog>>>>,
+    pub health_check_url: String,
 }
 
 pub fn start_subgraphs_server(
@@ -108,6 +109,7 @@ pub fn start_subgraphs_server(
 
     let shared_state = SubgraphsServiceState {
         request_log: Arc::new(Mutex::new(HashMap::new())),
+        health_check_url: format!("http://{}:{}/health", host, port),
     };
 
     let app = Router::new()
diff --git a/bin/router/src/jwt/errors.rs b/bin/router/src/jwt/errors.rs
@@ -87,10 +87,7 @@ impl From<&JwtError> for StatusCode {
 impl From<&JwtError> for GraphQLError {
     fn from(val: &JwtError) -> Self {
         GraphQLError {
-            extensions: GraphQLErrorExtensions {
-                code: Some(val.error_code().to_string()),
-                ..Default::default()
-            },
+            extensions: GraphQLErrorExtensions::new_from_code(val.error_code()),
             message: val.to_string(),
             locations: None,
             path: None,
diff --git a/bin/router/src/shared_state.rs b/bin/router/src/shared_state.rs
@@ -31,7 +31,7 @@ pub struct RouterSharedState {
     pub validate_cache: Cache<u64, Arc<Vec<ValidationError>>>,
     pub parse_cache: Cache<u64, Arc<graphql_parser::query::Document<'static, String>>>,
     pub normalize_cache: Cache<u64, Arc<GraphQLNormalizationPayload>>,
-    pub router_config: HiveRouterConfig,
+    pub router_config: Arc<HiveRouterConfig>,
     pub headers_plan: HeaderRulesPlan,
     pub cors: Option<Cors>,
     pub jwt_auth_runtime: Option<JwtAuthRuntime>,
@@ -47,10 +47,11 @@ impl RouterSharedState {
         let planner =
             Planner::new_from_supergraph(&parsed_supergraph_sdl).expect("failed to create planner");
         let schema_metadata = planner.consumer_schema.schema_metadata();
+        let router_config = Arc::new(router_config);
 
         let subgraph_executor_map = SubgraphExecutorMap::from_http_endpoint_map(
             supergraph_state.subgraph_endpoint_map,
-            router_config.traffic_shaping.clone(),
+            router_config.clone(),
         )
         .expect("Failed to create subgraph executor map");
 
diff --git a/docs/README.md b/docs/README.md
@@ -10,6 +10,7 @@
 |[**http**](#http)|`object`|Configuration for the HTTP server/listener.<br/>Default: `{"host":"0.0.0.0","port":4000}`<br/>||
 |[**jwt**](#jwt)|`object`, `null`|Configuration for JWT authentication plugin.<br/>|yes|
 |[**log**](#log)|`object`|The router logger configuration.<br/>Default: `{"filter":null,"format":"json","level":"info"}`<br/>||
+|[**override\_subgraph\_urls**](#override_subgraph_urls)|`object`|Configuration for overriding subgraph URLs.<br/>Default: `{}`<br/>||
 |[**query\_planner**](#query_planner)|`object`|Query planning configuration.<br/>Default: `{"allow_expose":false,"timeout":"10s"}`<br/>||
 |[**supergraph**](#supergraph)|`object`|Configuration for the Federation supergraph source. By default, the router will use a local file-based supergraph source (`./supergraph.graphql`).<br/>||
 |[**traffic\_shaping**](#traffic_shaping)|`object`|Configuration for the traffic-shaper executor. Use these configurations to control how requests are being executed to subgraphs.<br/>Default: `{"dedupe_enabled":true,"max_connections_per_host":100,"pool_idle_timeout_seconds":50}`<br/>||
@@ -59,6 +60,21 @@ log:
   filter: null
   format: json
   level: info
+override_subgraph_urls:
+  accounts:
+    url: https://accounts.example.com/graphql
+  products:
+    url:
+      expression: |2-
+
+                if .request.headers."x-region" == "us-east" {
+                    "https://products-us-east.example.com/graphql"
+                } else if .request.headers."x-region" == "eu-west" {
+                    "https://products-eu-west.example.com/graphql"
+                } else {
+                  .original_url
+                }
+            
 query_planner:
   allow_expose: false
   timeout: 10s
@@ -1497,6 +1513,47 @@ level: info
 
 ```
 
+<a name="override_subgraph_urls"></a>
+## override\_subgraph\_urls: object
+
+Configuration for overriding subgraph URLs.
+
+
+**Additional Properties**
+
+|Name|Type|Description|Required|
+|----|----|-----------|--------|
+|[**Additional Properties**](#override_subgraph_urlsadditionalproperties)|`object`||yes|
+
+**Example**
+
+```yaml
+accounts:
+  url: https://accounts.example.com/graphql
+products:
+  url:
+    expression: |2-
+
+              if .request.headers."x-region" == "us-east" {
+                  "https://products-us-east.example.com/graphql"
+              } else if .request.headers."x-region" == "eu-west" {
+                  "https://products-eu-west.example.com/graphql"
+              } else {
+                .original_url
+              }
+          
+
+```
+
+<a name="override_subgraph_urlsadditionalproperties"></a>
+### override\_subgraph\_urls\.additionalProperties: object
+
+**Properties**
+
+|Name|Type|Description|Required|
+|----|----|-----------|--------|
+|**url**||Overrides for the URL of the subgraph.<br/><br/>For convenience, a plain string in your configuration will be treated as a static URL.<br/><br/>### Static URL Example<br/>```yaml<br/>url: "https://api.example.com/graphql"<br/>```<br/><br/>### Dynamic Expression Example<br/><br/>The expression has access to the following variables:<br/>- `request`: The incoming HTTP request, including headers and other metadata.<br/>- `original_url`: The original URL of the subgraph (from supergraph sdl).<br/><br/>```yaml<br/>url:<br/>  expression: \|<br/>    if .request.headers."x-region" == "us-east" {<br/>      "https://products-us-east.example.com/graphql"<br/>    } else if .request.headers."x-region" == "eu-west" {<br/>      "https://products-eu-west.example.com/graphql"<br/>    } else {<br/>      .original_url<br/>    }<br/>|yes|
+
 <a name="query_planner"></a>
 ## query\_planner: object
 
diff --git a/e2e/Cargo.toml b/e2e/Cargo.toml
@@ -17,6 +17,8 @@ lazy_static = { workspace = true }
 jsonwebtoken = { workspace = true }
 insta = { workspace = true }
 
+reqwest = "0.12.23"
+
 hive-router = { path = "../bin/router" }
 hive-router-config = { path = "../lib/router-config" }
 subgraphs = { path = "../bench/subgraphs" }
diff --git a/e2e/configs/override_subgraph_urls/override_dynamic_header.router.yaml b/e2e/configs/override_subgraph_urls/override_dynamic_header.router.yaml
@@ -0,0 +1,13 @@
+# yaml-language-server: $schema=../../../router-config.schema.json
+supergraph:
+  source: file
+  path: ../../supergraph.graphql
+override_subgraph_urls:
+  accounts:
+    url:
+      expression: |
+        if .request.headers."x-accounts-port" == "4100" {
+          "http://0.0.0.0:4100/accounts"
+        } else {
+          .original_url
+        }
diff --git a/e2e/configs/override_subgraph_urls/override_static.router.yaml b/e2e/configs/override_subgraph_urls/override_static.router.yaml
@@ -0,0 +1,7 @@
+# yaml-language-server: $schema=../../../router-config.schema.json
+supergraph:
+  source: file
+  path: ../../supergraph.graphql
+override_subgraph_urls:
+  accounts:
+    url: "http://0.0.0.0:4100/accounts"
diff --git a/e2e/src/jwt.rs b/e2e/src/jwt.rs
@@ -24,7 +24,7 @@ mod jwt_e2e_tests {
 
     #[ntex::test]
     async fn should_forward_claims_to_subgraph_via_extensions() {
-        let subgraphs_server = SubgraphsServer::start();
+        let subgraphs_server = SubgraphsServer::start().await;
         let app = init_router_from_config_file("configs/jwt_auth_forward.router.yaml")
             .await
             .unwrap();
diff --git a/e2e/src/lib.rs b/e2e/src/lib.rs
@@ -1,4 +1,6 @@
 #[cfg(test)]
 mod jwt;
 #[cfg(test)]
+mod override_subgraph_urls;
+#[cfg(test)]
 mod testkit;
diff --git a/e2e/src/override_subgraph_urls.rs b/e2e/src/override_subgraph_urls.rs
@@ -0,0 +1,96 @@
+#[cfg(test)]
+mod override_subgraph_urls_e2e_tests {
+    use ntex::web::test;
+    use sonic_rs::{from_slice, Value};
+
+    use crate::testkit::{init_graphql_request, init_router_from_config_file, SubgraphsServer};
+
+    #[ntex::test]
+    /// Test that a static URL override for a subgraph is respected.
+    /// Starts a subgraph server on port 4100, but the supergraph SDL points to 4200.
+    /// The router config overrides the URL for the "accounts" subgraph to point to 4100.
+    /// This way we can verify that the override is applied correctly.
+    async fn should_override_subgraph_url_based_on_static_value() {
+        let subgraphs_server = SubgraphsServer::start_with_port(4100).await;
+        let app = init_router_from_config_file(
+            "configs/override_subgraph_urls/override_static.router.yaml",
+        )
+        .await
+        .unwrap();
+
+        let req = init_graphql_request("{ users { id } }", None);
+        let resp = test::call_service(&app, req.to_request()).await;
+
+        assert!(resp.status().is_success(), "Expected 200 OK");
+
+        let subgraph_requests = subgraphs_server
+            .get_subgraph_requests_log("accounts")
+            .await
+            .expect("expected requests sent to accounts subgraph");
+        assert_eq!(
+            subgraph_requests.len(),
+            1,
+            "expected 1 request to accounts subgraph"
+        );
+    }
+
+    #[ntex::test]
+    /// Test that a dynamic URL override for a subgraph based on a header value is respected.
+    /// Starts a subgraph server on port 4100, but the supergraph SDL points to 4200.
+    /// The router config overrides the URL for the "accounts" subgraph to point to 4100
+    /// when a specific header is present.
+    /// This way we can verify that the override is applied correctly.
+    /// Without the header, the request goes to 4200 and fail (thanks to `.original_url`).
+    async fn should_override_subgraph_url_based_on_header_value() {
+        let subgraphs_server = SubgraphsServer::start_with_port(4100).await;
+        let app = init_router_from_config_file(
+            "configs/override_subgraph_urls/override_dynamic_header.router.yaml",
+        )
+        .await
+        .unwrap();
+
+        // Makes the expression to evaluate to port 4100
+        let req = init_graphql_request("{ users { id } }", None).header("x-accounts-port", "4100");
+        let resp = test::call_service(&app, req.to_request()).await;
+
+        assert!(resp.status().is_success(), "Expected 200 OK");
+
+        let subgraph_requests = subgraphs_server
+            .get_subgraph_requests_log("accounts")
+            .await
+            .expect("expected requests sent to accounts subgraph");
+        assert_eq!(
+            subgraph_requests.len(),
+            1,
+            "expected 1 request to accounts subgraph"
+        );
+
+        // Makes the expression to evaluate to port 4200 (value of .original_url)
+        // which is not running, so the request fails
+        let req = init_graphql_request("{ users { id } }", None);
+        let resp = test::call_service(&app, req.to_request()).await;
+
+        assert!(resp.status().is_success(), "Expected 200 OK");
+        let body = test::read_body(resp).await;
+        let json_body: Value = from_slice(&body).unwrap();
+
+        assert_eq!(
+            json_body["errors"][0]["message"],
+            "Failed to execute request to subgraph"
+        );
+        assert_eq!(
+            json_body["errors"][0]["extensions"]["code"],
+            "SUBGRAPH_REQUEST_FAILURE"
+        );
+
+        let subgraph_requests = subgraphs_server
+            .get_subgraph_requests_log("accounts")
+            .await
+            .expect("expected requests sent to accounts subgraph");
+        assert_eq!(
+            subgraph_requests.len(),
+            1, // still 1, no new request should be made
+            "expected 1 request to accounts subgraph"
+        );
+    }
+}
diff --git a/e2e/src/testkit.rs b/e2e/src/testkit.rs
@@ -1,4 +1,4 @@
-use std::{path::PathBuf, sync::Once};
+use std::{path::PathBuf, sync::Once, time::Duration};
 
 use hive_router::{
     background_tasks::BackgroundTasksManager, configure_app_from_config, configure_ntex_app,
@@ -59,8 +59,28 @@ impl Drop for SubgraphsServer {
 }
 
 impl SubgraphsServer {
-    pub fn start() -> Self {
-        let (_server_handle, shutdown_tx, subgraph_shared_state) = start_subgraphs_server(None);
+    /// Port defaults to 4200 if None (as defined in supergraph.graphql)
+    pub async fn start() -> Self {
+        Self::start_with_port(4200).await
+    }
+
+    pub async fn start_with_port(port: u16) -> Self {
+        let (_server_handle, shutdown_tx, subgraph_shared_state) =
+            start_subgraphs_server(Some(port));
+
+        let health_check_url = subgraph_shared_state.health_check_url.clone();
+        loop {
+            match reqwest::get(&health_check_url).await {
+                Ok(response) if response.status().is_success() => {
+                    // Server is up and running.
+                    break;
+                }
+                _ => {
+                    // Server not ready yet, wait and retry.
+                    tokio::time::sleep(Duration::from_millis(1)).await;
+                }
+            }
+        }
 
         Self {
             shutdown_tx: Some(shutdown_tx),
diff --git a/lib/executor/src/execution/plan.rs b/lib/executor/src/execution/plan.rs
@@ -741,7 +741,7 @@ impl<'exec> Executor<'exec> {
             subgraph_name: node.service_name.clone(),
             response: self
                 .executors
-                .execute(&node.service_name, subgraph_request)
+                .execute(&node.service_name, subgraph_request, self.client_request)
                 .await
                 .into(),
         }))
diff --git a/lib/executor/src/executors/error.rs b/lib/executor/src/executors/error.rs
diff --git a/lib/executor/src/executors/http.rs b/lib/executor/src/executors/http.rs
diff --git a/lib/executor/src/executors/map.rs b/lib/executor/src/executors/map.rs
diff --git a/lib/executor/src/response/graphql_error.rs b/lib/executor/src/response/graphql_error.rs
diff --git a/lib/router-config/src/lib.rs b/lib/router-config/src/lib.rs
diff --git a/lib/router-config/src/override_subgraph_urls.rs b/lib/router-config/src/override_subgraph_urls.rs
