Introduce make_fake_batch() to avoid racy caches and redact more otel trace details

.changesets/maint_garypen_modify_batch_for_tracing.md

@@ -0,0 +1,14 @@
+### Improve testing by avoiding cache effects and redacting tracing details ([PR #5638](https://github.com/apollographql/router/pull/5638))
+
+We've had some problems with flaky tests and this PR addresses some of them.
+
+The router executes in parallel and concurrently. Many of our tests use snapshots to try and make assertions that functionality is continuing to work correctly. Unfortunately, concurrent/parallel execution and static snapshots don't co-operate very well. Results may appear in pseudo-random order (compared to snapshot expectations) and so tests become flaky and fail without obvious cause.
+
+The problem becomes particularly acute with features which are specifically designed for highly concurrent operation, such as batching.
+
+This set of changes addresses some of the router testing problems by:
+
+1. Making items in a batch test different enough that caching effects are avoided.
+2. Redacting various details so that sequencing is not as much of an issue in the otel traces tests.
+
+By [@garypen](https://github.com/garypen) in https://github.com/apollographql/router/pull/5638

apollo-router/src/lib.rs

@@ -95,6 +95,7 @@ pub use crate::router::RouterHttpServer;
 pub use crate::router::SchemaSource;
 pub use crate::router::ShutdownSource;
 pub use crate::router_factory::Endpoint;
+pub use crate::test_harness::make_fake_batch;
 pub use crate::test_harness::MockedSubgraphs;
 pub use crate::test_harness::TestHarness;
 pub use crate::uplink::UplinkConfig;

apollo-router/src/services/router/tests.rs

@@ -23,6 +23,7 @@ use crate::services::supergraph;
 use crate::services::SupergraphRequest;
 use crate::services::SupergraphResponse;
 use crate::services::MULTIPART_DEFER_CONTENT_TYPE;
+use crate::test_harness::make_fake_batch;
 use crate::Context;
 
 // Test Vary processing
@@ -229,8 +230,6 @@ async fn test_http_max_request_bytes() {
     assert_eq!(response.status(), http::StatusCode::PAYLOAD_TOO_LARGE);
 }
 
-//  Test query batching
-
 #[tokio::test]
 async fn it_only_accepts_batch_http_link_mode_for_query_batch() {
     let expected_response: serde_json::Value = serde_json::from_str(include_str!(
@@ -239,20 +238,13 @@ async fn it_only_accepts_batch_http_link_mode_for_query_batch() {
     .unwrap();
 
     async fn with_config() -> router::Response {
-        let http_request = supergraph::Request::canned_builder()
-            .build()
-            .unwrap()
-            .supergraph_request
-            .map(|req: graphql::Request| {
-                // Modify the request so that it is a valid array of requests.
-                let mut json_bytes = serde_json::to_vec(&req).unwrap();
-                let mut result = vec![b'['];
-                result.append(&mut json_bytes.clone());
-                result.push(b',');
-                result.append(&mut json_bytes);
-                result.push(b']');
-                hyper::Body::from(result)
-            });
+        let http_request = make_fake_batch(
+            supergraph::Request::canned_builder()
+                .build()
+                .unwrap()
+                .supergraph_request,
+            None,
+        );
         let config = serde_json::json!({});
         crate::TestHarness::builder()
             .configuration_json(config)
@@ -336,20 +328,13 @@ async fn it_will_not_process_a_query_batch_without_enablement() {
     .unwrap();
 
     async fn with_config() -> router::Response {
-        let http_request = supergraph::Request::canned_builder()
-            .build()
-            .unwrap()
-            .supergraph_request
-            .map(|req: graphql::Request| {
-                // Modify the request so that it is a valid array of requests.
-                let mut json_bytes = serde_json::to_vec(&req).unwrap();
-                let mut result = vec![b'['];
-                result.append(&mut json_bytes.clone());
-                result.push(b',');
-                result.append(&mut json_bytes);
-                result.push(b']');
-                hyper::Body::from(result)
-            });
+        let http_request = make_fake_batch(
+            supergraph::Request::canned_builder()
+                .build()
+                .unwrap()
+                .supergraph_request,
+            None,
+        );
         let config = serde_json::json!({});
         crate::TestHarness::builder()
             .configuration_json(config)
@@ -382,7 +367,7 @@ async fn it_will_not_process_a_poorly_formatted_query_batch() {
             .unwrap()
             .supergraph_request
             .map(|req: graphql::Request| {
-                // Modify the request so that it is a valid array of requests.
+                // Modify the request so that it is an invalid array of requests.
                 let mut json_bytes = serde_json::to_vec(&req).unwrap();
                 let mut result = vec![b'['];
                 result.append(&mut json_bytes.clone());
@@ -488,22 +473,15 @@ async fn it_will_not_process_a_batched_deferred_query() {
                 }
             }
         ";
-        let http_request = supergraph::Request::canned_builder()
-            .header(http::header::ACCEPT, MULTIPART_DEFER_CONTENT_TYPE)
-            .query(query)
-            .build()
-            .unwrap()
-            .supergraph_request
-            .map(|req: graphql::Request| {
-                // Modify the request so that it is a valid array of requests.
-                let mut json_bytes = serde_json::to_vec(&req).unwrap();
-                let mut result = vec![b'['];
-                result.append(&mut json_bytes.clone());
-                result.push(b',');
-                result.append(&mut json_bytes);
-                result.push(b']');
-                hyper::Body::from(result)
-            });
+        let http_request = make_fake_batch(
+            supergraph::Request::canned_builder()
+                .header(http::header::ACCEPT, MULTIPART_DEFER_CONTENT_TYPE)
+                .query(query)
+                .build()
+                .unwrap()
+                .supergraph_request,
+            None,
+        );
         let config = serde_json::json!({
             "batching": {
                 "enabled": true,

apollo-router/src/test_harness.rs

@@ -15,6 +15,7 @@ use crate::axum_factory::span_mode;
 use crate::axum_factory::utils::PropagatingMakeSpan;
 use crate::configuration::Configuration;
 use crate::configuration::ConfigurationError;
+use crate::graphql;
 use crate::plugin::test::canned;
 use crate::plugin::test::MockSubgraph;
 use crate::plugin::DynPlugin;
@@ -488,3 +489,62 @@ impl Plugin for MockedSubgraphs {
             .unwrap_or(default)
     }
 }
+
+// This function takes a valid request and duplicates it (optionally, with a new operation
+// name) to create an array (batch) request.
+//
+// Note: It's important to make the operation name different to prevent race conditions in testing
+// where various tests assume the presence (or absence) of a test span.
+//
+// Detailed Explanation
+//
+// A batch sends a series of requests concurrently through a router. If we
+// simply duplicate the request, then there is significant chance that spans such as
+// "parse_query" won't appear because the document has already been parsed and is now in a cache.
+//
+// To explicitly avoid this, we add an operation name which will force the router to re-parse the
+// document since operation name is part of the parsed document cache key.
+//
+// This has been a significant cause of racy/flaky tests in the past.
+
+///
+/// Convert a graphql request into a batch of requests
+///
+/// This is helpful for testing batching functionality.
+/// Given a GraphQL request, generate an array containing the request and it's duplicate.
+///
+/// If an op_from_to is supplied, this will modify the duplicated request so that it uses the new
+/// operation name.
+///
+pub fn make_fake_batch(
+    input: http::Request<graphql::Request>,
+    op_from_to: Option<(&str, &str)>,
+) -> http::Request<crate::services::router::Body> {
+    input.map(|req| {
+        // Modify the request so that it is a valid array of requests.
+        let mut new_req = req.clone();
+
+        // If we were given an op_from_to, then try to modify the query to update the operation
+        // name from -> to.
+        // If our request doesn't have an operation name or we weren't given an op_from_to,
+        // just duplicate the request as is.
+        if let Some((from, to)) = op_from_to {
+            if let Some(operation_name) = &req.operation_name {
+                if operation_name == from {
+                    new_req.query = req.query.clone().map(|q| q.replace(from, to));
+                    new_req.operation_name = Some(to.to_string());
+                }
+            }
+        }
+
+        let mut json_bytes_req = serde_json::to_vec(&req).unwrap();
+        let mut json_bytes_new_req = serde_json::to_vec(&new_req).unwrap();
+
+        let mut result = vec![b'['];
+        result.append(&mut json_bytes_req);
+        result.push(b',');
+        result.append(&mut json_bytes_new_req);
+        result.push(b']');
+        crate::services::router::Body::from(result)
+    })
+}

apollo-router/tests/apollo_otel_traces.rs

@@ -16,6 +16,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::anyhow;
+use apollo_router::make_fake_batch;
 use apollo_router::services::router;
 use apollo_router::services::router::BoxCloneService;
 use apollo_router::services::supergraph;
@@ -153,11 +154,14 @@ async fn get_batch_router_service(
 
 macro_rules! assert_report {
         ($report: expr)=> {
+            assert_report!($report, false)
+        };
+        ($report: expr, $batch: literal)=> {
             insta::with_settings!({sort_maps => true}, {
                     insta::assert_yaml_snapshot!($report, {
                         ".**.attributes" => insta::sorted_redaction(),
                         ".**.attributes[]" => insta::dynamic_redaction(|mut value, _| {
-                            const REDACTED_ATTRIBUTES: [&'static str; 11] = [
+                            let mut redacted_attributes = vec![
                                 "apollo.client.host",
                                 "apollo.client.uname",
                                 "apollo.router.id",
@@ -170,9 +174,15 @@ macro_rules! assert_report {
                                 "apollo_private.sent_time_offset",
                                 "trace_id",
                             ];
+                            if $batch {
+                                redacted_attributes.append(&mut vec![
+                                "apollo_private.operation_signature",
+                                "graphql.operation.name"
+                            ]);
+                            }
                             if let insta::internals::Content::Struct(name, key_value)  = &mut value{
                                 if name == &"KeyValue" {
-                                    if REDACTED_ATTRIBUTES.contains(&key_value[0].1.as_str().unwrap()) {
+                                    if redacted_attributes.contains(&key_value[0].1.as_str().unwrap()) {
                                         key_value[1].1 = insta::internals::Content::NewtypeVariant(
                                             "Value", 0, "stringValue", Box::new(insta::internals::Content::from("[redacted]"))
                                         );
@@ -403,24 +413,18 @@ async fn test_trace_id() {
 #[tokio::test(flavor = "multi_thread")]
 async fn test_batch_trace_id() {
     for use_legacy_request_span in [true, false] {
-        let request = supergraph::Request::fake_builder()
-            .query("query{topProducts{name reviews {author{name}} reviews{author{name}}}}")
-            .build()
-            .unwrap()
-            .supergraph_request
-            .map(|req| {
-                // Modify the request so that it is a valid array of requests.
-                let mut json_bytes = serde_json::to_vec(&req).unwrap();
-                let mut result = vec![b'['];
-                result.append(&mut json_bytes.clone());
-                result.push(b',');
-                result.append(&mut json_bytes);
-                result.push(b']');
-                hyper::Body::from(result)
-            });
+        let request = make_fake_batch(
+            supergraph::Request::fake_builder()
+                .query("query one {topProducts{name reviews {author{name}} reviews{author{name}}}}")
+                .operation_name("one")
+                .build()
+                .unwrap()
+                .supergraph_request,
+            Some(("one", "two")),
+        );
         let reports = Arc::new(Mutex::new(vec![]));
         let report = get_batch_trace_report(reports, request.into(), use_legacy_request_span).await;
-        assert_report!(report);
+        assert_report!(report, true);
     }
 }
 
@@ -473,26 +477,20 @@ async fn test_send_header() {
 #[tokio::test(flavor = "multi_thread")]
 async fn test_batch_send_header() {
     for use_legacy_request_span in [true, false] {
-        let request = supergraph::Request::fake_builder()
-            .query("query{topProducts{name reviews {author{name}} reviews{author{name}}}}")
-            .header("send-header", "Header value")
-            .header("dont-send-header", "Header value")
-            .build()
-            .unwrap()
-            .supergraph_request
-            .map(|req| {
-                // Modify the request so that it is a valid array of requests.
-                let mut json_bytes = serde_json::to_vec(&req).unwrap();
-                let mut result = vec![b'['];
-                result.append(&mut json_bytes.clone());
-                result.push(b',');
-                result.append(&mut json_bytes);
-                result.push(b']');
-                hyper::Body::from(result)
-            });
+        let request = make_fake_batch(
+            supergraph::Request::fake_builder()
+                .query("query one {topProducts{name reviews {author{name}} reviews{author{name}}}}")
+                .operation_name("one")
+                .header("send-header", "Header value")
+                .header("dont-send-header", "Header value")
+                .build()
+                .unwrap()
+                .supergraph_request,
+            Some(("one", "two")),
+        );
         let reports = Arc::new(Mutex::new(vec![]));
         let report = get_batch_trace_report(reports, request.into(), use_legacy_request_span).await;
-        assert_report!(report);
+        assert_report!(report, true);
     }
 }