diff --git a/apollo-router/tests/common.rs b/apollo-router/tests/common.rs index 7dd7636a00..e70cf5f0de 100644 --- a/apollo-router/tests/common.rs +++ b/apollo-router/tests/common.rs @@ -8,7 +8,7 @@ use std::sync::Arc; use std::sync::Mutex; use std::time::Duration; -use buildstructor::{buildstructor}; +use buildstructor::buildstructor; use fred::clients::RedisClient; use fred::interfaces::ClientLike; use fred::interfaces::KeysInterface; @@ -335,7 +335,11 @@ impl Telemetry { } } - pub(crate) fn extract_context(&self, request: &wiremock::Request, context: &Context) -> Context { + pub(crate) fn extract_context( + &self, + request: &wiremock::Request, + context: &Context, + ) -> Context { let headers: HashMap = request .headers .iter() @@ -362,8 +366,7 @@ impl Telemetry { .expect("psr"); let new_trace_id = if original_span_context.is_valid() { original_span_context.trace_id() - } - else { + } else { context.span().span_context().trace_id() }; context = context.with_remote_span_context(SpanContext::new( diff --git a/apollo-router/tests/integration/batching.rs b/apollo-router/tests/integration/batching.rs index f9e7ba6ab8..521e615b30 100644 --- a/apollo-router/tests/integration/batching.rs +++ b/apollo-router/tests/integration/batching.rs @@ -856,7 +856,8 @@ mod helper { use wiremock::ResponseTemplate; use super::test_is_enabled; - use crate::integration::common::{IntegrationTest, Query}; + use crate::integration::common::IntegrationTest; + use crate::integration::common::Query; /// Helper type for specifying a valid handler pub type Handler = fn(&wiremock::Request) -> ResponseTemplate; @@ -916,7 +917,9 @@ mod helper { // Execute the request let request = serde_json::to_value(requests)?; - let (_span, response) = router.execute_query(Query::builder().body(request).build()).await; + let (_span, response) = router + .execute_query(Query::builder().body(request).build()) + .await; serde_json::from_slice::>(&response.bytes().await?).map_err(BoxError::from) } diff --git a/apollo-router/tests/integration/coprocessor.rs b/apollo-router/tests/integration/coprocessor.rs index dd7adcfbbf..d82d15ca7c 100644 --- a/apollo-router/tests/integration/coprocessor.rs +++ b/apollo-router/tests/integration/coprocessor.rs @@ -7,7 +7,8 @@ use wiremock::matchers::path; use wiremock::Mock; use wiremock::ResponseTemplate; -use crate::integration::common::{graph_os_enabled, Query}; +use crate::integration::common::graph_os_enabled; +use crate::integration::common::Query; use crate::integration::IntegrationTest; #[tokio::test(flavor = "multi_thread")] @@ -75,7 +76,9 @@ async fn test_coprocessor_limit_payload() -> Result<(), BoxError> { assert_eq!(response.status(), 200); // This query is huge and will be rejected because it is too large before hitting the coprocessor - let (_trace_id, response) = router.execute_query(Query::default().with_huge_query()).await; + let (_trace_id, response) = router + .execute_query(Query::default().with_huge_query()) + .await; assert_eq!(response.status(), 413); assert_yaml_snapshot!(response.text().await?); diff --git a/apollo-router/tests/integration/introspection.rs b/apollo-router/tests/integration/introspection.rs index 56b2a496cd..8ad142a9cb 100644 --- a/apollo-router/tests/integration/introspection.rs +++ b/apollo-router/tests/integration/introspection.rs @@ -1,10 +1,11 @@ -use crate::integration::common::Query; -use crate::integration::IntegrationTest; use apollo_router::plugin::test::MockSubgraph; use apollo_router::services::supergraph::Request; use serde_json::json; use tower::ServiceExt; +use crate::integration::common::Query; +use crate::integration::IntegrationTest; + #[tokio::test] async fn simple() { let request = Request::fake_builder() diff --git a/apollo-router/tests/integration/operation_limits.rs b/apollo-router/tests/integration/operation_limits.rs index b0c5b25802..1b6b186e41 100644 --- a/apollo-router/tests/integration/operation_limits.rs +++ b/apollo-router/tests/integration/operation_limits.rs @@ -9,6 +9,7 @@ use apollo_router::TestHarness; use serde_json::json; use tower::BoxError; use tower::ServiceExt; + use crate::integration::common::Query; use crate::integration::IntegrationTest; @@ -310,7 +311,9 @@ async fn test_request_bytes_limit_with_coprocessor() -> Result<(), BoxError> { .await; router.start().await; router.assert_started().await; - let (_, resp) = router.execute_query(Query::default().with_huge_query()).await; + let (_, resp) = router + .execute_query(Query::default().with_huge_query()) + .await; assert_eq!(resp.status(), 413); router.graceful_shutdown().await; Ok(()) @@ -324,7 +327,9 @@ async fn test_request_bytes_limit() -> Result<(), BoxError> { .await; router.start().await; router.assert_started().await; - let (_, resp) = router.execute_query(Query::default().with_huge_query()).await; + let (_, resp) = router + .execute_query(Query::default().with_huge_query()) + .await; assert_eq!(resp.status(), 413); router.graceful_shutdown().await; Ok(()) diff --git a/apollo-router/tests/integration/query_planner/max_evaluated_plans.rs b/apollo-router/tests/integration/query_planner/max_evaluated_plans.rs index f3edb84232..6326d600ee 100644 --- a/apollo-router/tests/integration/query_planner/max_evaluated_plans.rs +++ b/apollo-router/tests/integration/query_planner/max_evaluated_plans.rs @@ -1,4 +1,5 @@ use serde_json::json; + use crate::integration::common::Query; use crate::integration::IntegrationTest; @@ -31,10 +32,14 @@ async fn reports_evaluated_plans() { router.start().await; router.assert_started().await; router - .execute_query(Query::builder().body(json!({ - "query": r#"{ t { v1 v2 v3 v4 } }"#, - "variables": {}, - })).build()) + .execute_query( + Query::builder() + .body(json!({ + "query": r#"{ t { v1 v2 v3 v4 } }"#, + "variables": {}, + })) + .build(), + ) .await; let metrics = router @@ -70,10 +75,14 @@ async fn does_not_exceed_max_evaluated_plans() { router.start().await; router.assert_started().await; router - .execute_query(Query::builder().body(json!({ - "query": r#"{ t { v1 v2 v3 v4 } }"#, - "variables": {}, - })).build()) + .execute_query( + Query::builder() + .body(json!({ + "query": r#"{ t { v1 v2 v3 v4 } }"#, + "variables": {}, + })) + .build(), + ) .await; let metrics = router diff --git a/apollo-router/tests/integration/redis.rs b/apollo-router/tests/integration/redis.rs index ea9f37be62..07d16d7b92 100644 --- a/apollo-router/tests/integration/redis.rs +++ b/apollo-router/tests/integration/redis.rs @@ -40,7 +40,8 @@ use serde_json::Value; use tower::BoxError; use tower::ServiceExt; -use crate::integration::common::{graph_os_enabled, Query}; +use crate::integration::common::graph_os_enabled; +use crate::integration::common::Query; use crate::integration::IntegrationTest; #[tokio::test(flavor = "multi_thread")] @@ -1072,11 +1073,15 @@ async fn test_redis_query_plan_config_update(updated_config: &str, new_cache_key ); assert_ne!(starting_key, new_cache_key, "starting_key (cache key for the initial config) and new_cache_key (cache key with the updated config) should not be equal. This either means that the cache key is not being generated correctly, or that the test is not actually checking the updated key."); - router.execute_query(Query::default().with_anonymous()).await; + router + .execute_query(Query::default().with_anonymous()) + .await; router.assert_redis_cache_contains(starting_key, None).await; router.update_config(updated_config).await; router.assert_reloaded().await; - router.execute_query(Query::default().with_anonymous()).await; + router + .execute_query(Query::default().with_anonymous()) + .await; router .assert_redis_cache_contains(new_cache_key, Some(starting_key)) .await; diff --git a/apollo-router/tests/integration/subgraph_response.rs b/apollo-router/tests/integration/subgraph_response.rs index cba16ca370..3f0f194d92 100644 --- a/apollo-router/tests/integration/subgraph_response.rs +++ b/apollo-router/tests/integration/subgraph_response.rs @@ -1,6 +1,7 @@ use serde_json::json; use tower::BoxError; use wiremock::ResponseTemplate; + use crate::integration::common::Query; use crate::integration::IntegrationTest; @@ -21,7 +22,9 @@ async fn test_subgraph_returning_data_null() -> Result<(), BoxError> { router.assert_started().await; let query = "{ __typename topProducts { name } }"; - let (_trace_id, response) = router.execute_query(Query::builder().body(json!({ "query": query })).build()).await; + let (_trace_id, response) = router + .execute_query(Query::builder().body(json!({ "query": query })).build()) + .await; assert_eq!(response.status(), 200); assert_eq!( response.json::().await?, @@ -64,7 +67,9 @@ async fn test_subgraph_returning_different_typename_on_query_root() -> Result<() inside_fragment: __typename } "#; - let (_trace_id, response) = router.execute_query(Query::builder().body(json!({ "query": query })).build()).await; + let (_trace_id, response) = router + .execute_query(Query::builder().body(json!({ "query": query })).build()) + .await; assert_eq!(response.status(), 200); assert_eq!( response.json::().await?, @@ -99,7 +104,11 @@ async fn test_valid_extensions_service_for_subgraph_error() -> Result<(), BoxErr router.assert_started().await; let (_trace_id, response) = router - .execute_query(Query::builder().body(json!({ "query": "{ topProducts { name } }" })).build()) + .execute_query( + Query::builder() + .body(json!({ "query": "{ topProducts { name } }" })) + .build(), + ) .await; assert_eq!(response.status(), 200); assert_eq!( @@ -141,7 +150,11 @@ async fn test_valid_extensions_service_is_preserved_for_subgraph_error() -> Resu router.assert_started().await; let (_trace_id, response) = router - .execute_query(Query::builder().body(json!({ "query": "{ topProducts { name } }" })).build()) + .execute_query( + Query::builder() + .body(json!({ "query": "{ topProducts { name } }" })) + .build(), + ) .await; assert_eq!(response.status(), 200); assert_eq!( @@ -174,7 +187,11 @@ async fn test_valid_extensions_service_for_invalid_subgraph_response() -> Result router.assert_started().await; let (_trace_id, response) = router - .execute_query(Query::builder().body(json!({ "query": "{ topProducts { name } }" })).build()) + .execute_query( + Query::builder() + .body(json!({ "query": "{ topProducts { name } }" })) + .build(), + ) .await; assert_eq!(response.status(), 200); assert_eq!( @@ -222,7 +239,11 @@ async fn test_valid_error_locations() -> Result<(), BoxError> { router.assert_started().await; let (_trace_id, response) = router - .execute_query(Query::builder().body(json!({ "query": "{ topProducts { name } }" })).build()) + .execute_query( + Query::builder() + .body(json!({ "query": "{ topProducts { name } }" })) + .build(), + ) .await; assert_eq!(response.status(), 200); assert_eq!( @@ -264,7 +285,11 @@ async fn test_empty_error_locations() -> Result<(), BoxError> { router.assert_started().await; let (_trace_id, response) = router - .execute_query(Query::builder().body(json!({ "query": "{ topProducts { name } }" })).build()) + .execute_query( + Query::builder() + .body(json!({ "query": "{ topProducts { name } }" })) + .build(), + ) .await; assert_eq!(response.status(), 200); assert_eq!( @@ -302,7 +327,11 @@ async fn test_invalid_error_locations() -> Result<(), BoxError> { router.assert_started().await; let (_trace_id, response) = router - .execute_query(Query::builder().body(json!({ "query": "{ topProducts { name } }" })).build()) + .execute_query( + Query::builder() + .body(json!({ "query": "{ topProducts { name } }" })) + .build(), + ) .await; assert_eq!(response.status(), 200); assert_eq!( @@ -345,7 +374,11 @@ async fn test_invalid_error_locations_with_single_negative_one_location() -> Res router.assert_started().await; let (_trace_id, response) = router - .execute_query(Query::builder().body(json!({ "query": "{ topProducts { name } }" })).build()) + .execute_query( + Query::builder() + .body(json!({ "query": "{ topProducts { name } }" })) + .build(), + ) .await; assert_eq!(response.status(), 200); assert_eq!( @@ -387,7 +420,11 @@ async fn test_invalid_error_locations_contains_negative_one_location() -> Result router.assert_started().await; let (_trace_id, response) = router - .execute_query(Query::builder().body(json!({ "query": "{ topProducts { name } }" })).build()) + .execute_query( + Query::builder() + .body(json!({ "query": "{ topProducts { name } }" })) + .build(), + ) .await; assert_eq!(response.status(), 200); assert_eq!( @@ -427,9 +464,7 @@ async fn test_valid_error_path() -> Result<(), BoxError> { router.start().await; router.assert_started().await; - let (_trace_id, response) = router - .execute_query(Query::default()) - .await; + let (_trace_id, response) = router.execute_query(Query::default()).await; assert_eq!(response.status(), 200); assert_eq!( response.json::().await?, @@ -464,9 +499,7 @@ async fn test_invalid_error_path() -> Result<(), BoxError> { router.start().await; router.assert_started().await; - let (_trace_id, response) = router - .execute_query(Query::default()) - .await; + let (_trace_id, response) = router.execute_query(Query::default()).await; assert_eq!(response.status(), 200); assert_eq!( response.json::().await?, @@ -502,9 +535,7 @@ async fn test_partially_valid_error_path() -> Result<(), BoxError> { router.start().await; router.assert_started().await; - let (_trace_id, response) = router - .execute_query(Query::default()) - .await; + let (_trace_id, response) = router.execute_query(Query::default()).await; assert_eq!(response.status(), 200); assert_eq!( response.json::().await?, diff --git a/apollo-router/tests/integration/subscription.rs b/apollo-router/tests/integration/subscription.rs index 74c42f2034..faad126f8e 100644 --- a/apollo-router/tests/integration/subscription.rs +++ b/apollo-router/tests/integration/subscription.rs @@ -4,7 +4,8 @@ use http::HeaderValue; use serde_json::json; use tower::BoxError; -use super::common::{IntegrationTest, Query}; +use super::common::IntegrationTest; +use super::common::Query; use super::common::Telemetry; const SUBSCRIPTION_CONFIG: &str = include_str!("../fixtures/subscription.router.yaml"); @@ -59,7 +60,10 @@ async fn test_subscription_load() -> Result<(), BoxError> { for _ in 0..100 { let (_id, resp) = router - .execute_query(Query::builder().body(json!({"query":"query ExampleQuery {topProducts{name}}","variables":{}})).build(), + .execute_query( + Query::builder() + .body(json!({"query":"query ExampleQuery {topProducts{name}}","variables":{}})) + .build(), ) .await; assert!(resp.status().is_success()); diff --git a/apollo-router/tests/integration/supergraph.rs b/apollo-router/tests/integration/supergraph.rs index 5732b0921d..8d7ae9727b 100644 --- a/apollo-router/tests/integration/supergraph.rs +++ b/apollo-router/tests/integration/supergraph.rs @@ -1,6 +1,6 @@ - use serde_json::json; use tower::BoxError; + use crate::integration::common::Query; use crate::integration::IntegrationTest; @@ -100,7 +100,12 @@ async fn test_supergraph_errors_on_http1_header_that_does_not_fit_inside_buffer( router.assert_started().await; let (_trace_id, response) = router - .execute_query(Query::builder().body(json!({ "query": "{ __typename }"})).header("test-header", "x".repeat(1048576 + 1)).build()) + .execute_query( + Query::builder() + .body(json!({ "query": "{ __typename }"})) + .header("test-header", "x".repeat(1048576 + 1)) + .build(), + ) .await; assert_eq!(response.status(), 431); Ok(()) @@ -122,7 +127,12 @@ async fn test_supergraph_allow_to_change_http1_max_buf_size() -> Result<(), BoxE router.assert_started().await; let (_trace_id, response) = router - .execute_query(Query::builder().body(json!({ "query": "{ __typename }"})).header("test-header", "x".repeat(1048576 + 1)).build()) + .execute_query( + Query::builder() + .body(json!({ "query": "{ __typename }"})) + .header("test-header", "x".repeat(1048576 + 1)) + .build(), + ) .await; assert_eq!(response.status(), 200); assert_eq!( diff --git a/apollo-router/tests/integration/telemetry/datadog.rs b/apollo-router/tests/integration/telemetry/datadog.rs index 591a2702fc..23bbc32652 100644 --- a/apollo-router/tests/integration/telemetry/datadog.rs +++ b/apollo-router/tests/integration/telemetry/datadog.rs @@ -790,7 +790,6 @@ impl Verifier for DatadogTraceSpec { } fn validate_span_kind(&self, trace: &Value, name: &str, kind: &str) -> Result<(), BoxError> { - let binding1 = trace.select_path(&format!( "$..[?(@.meta.['otel.original_name'] == '{}')].meta.['span.kind']", name diff --git a/apollo-router/tests/integration/telemetry/jaeger.rs b/apollo-router/tests/integration/telemetry/jaeger.rs index b18039bf26..2fc5e8a9f6 100644 --- a/apollo-router/tests/integration/telemetry/jaeger.rs +++ b/apollo-router/tests/integration/telemetry/jaeger.rs @@ -9,7 +9,8 @@ use serde_json::json; use serde_json::Value; use tower::BoxError; -use crate::integration::common::{Query, Telemetry}; +use crate::integration::common::Query; +use crate::integration::common::Telemetry; use crate::integration::telemetry::verifier::Verifier; use crate::integration::telemetry::TraceSpec; use crate::integration::IntegrationTest; @@ -258,31 +259,50 @@ async fn test_span_attributes() -> Result<(), BoxError> { TraceSpec::builder() .services(["client", "router", "subgraph"].into()) .operation_name("ExampleQuery") - .span_attribute("router", [("http.request.method", "POST"), - ("http.response.status_code", "200"), - ("url.path", "/"), - ("http.request.header.x-my-header", "test"), - ("http.request.header.x-not-present", "nope"), - ("http.request.header.x-my-header-condition", "test"), - ("studio.operation.id", "*"), - ].into()) - .span_attribute("supergraph", [ - ("graphql.operation.name", "ExampleQuery"), - ("graphql.operation.type", "query"), - ("graphql.document", "query ExampleQuery {topProducts{name}}"), - ].into()) - .span_attribute("subgraph", [ - ("subgraph.graphql.operation.type", "query"), - ("subgraph.name", "products")].into()) + .span_attribute( + "router", + [ + ("http.request.method", "POST"), + ("http.response.status_code", "200"), + ("url.path", "/"), + ("http.request.header.x-my-header", "test"), + ("http.request.header.x-not-present", "nope"), + ("http.request.header.x-my-header-condition", "test"), + ("studio.operation.id", "*"), + ] + .into(), + ) + .span_attribute( + "supergraph", + [ + ("graphql.operation.name", "ExampleQuery"), + ("graphql.operation.type", "query"), + ("graphql.document", "query ExampleQuery {topProducts{name}}"), + ] + .into(), + ) + .span_attribute( + "subgraph", + [ + ("subgraph.graphql.operation.type", "query"), + ("subgraph.name", "products"), + ] + .into(), + ) .build() - .validate_jaeger_trace(&mut router, Query::builder().header("x-my-header", "test").header("x-my-header-condition", "condition").build()) + .validate_jaeger_trace( + &mut router, + Query::builder() + .header("x-my-header", "test") + .header("x-my-header-condition", "condition") + .build(), + ) .await?; router.graceful_shutdown().await; } Ok(()) } - #[tokio::test(flavor = "multi_thread")] async fn test_decimal_trace_id() -> Result<(), BoxError> { let mut router = IntegrationTest::builder() @@ -327,19 +347,26 @@ impl Verifier for JaegerTraceSpec { fn verify_span_attributes(&self, trace: &Value) -> Result<(), BoxError> { for (span, attributes) in &self.span_attributes { for (key, value) in attributes { - let binding = trace.select_path(&format!("$..spans[?(@.operationName == '{span}')]..tags..[?(@.key == '{key}')].value"))?; + let binding = trace.select_path(&format!( + "$..spans[?(@.operationName == '{span}')]..tags..[?(@.key == '{key}')].value" + ))?; let actual_value = binding .first() - .expect(&format!("could not find attribute {key} on {span}")); + .unwrap_or_else(|| panic!("could not find attribute {key} on {span}")); match actual_value { Value::String(_) if *value == "*" => continue, - Value::String(s) => assert_eq!(s, value, "unexpected attribute {key} on {span}"), + Value::String(s) => { + assert_eq!(s, value, "unexpected attribute {key} on {span}") + } Value::Number(_) if *value == "*" => continue, - Value::Number(n) => assert_eq!(n.to_string(), *value, "unexpected attribute {key} on {span}"), + Value::Number(n) => assert_eq!( + n.to_string(), + *value, + "unexpected attribute {key} on {span}" + ), _ => panic!("unexpected value type"), } - } } Ok(()) diff --git a/apollo-router/tests/integration/telemetry/logging.rs b/apollo-router/tests/integration/telemetry/logging.rs index 21c19a3246..59dc1c7ccd 100644 --- a/apollo-router/tests/integration/telemetry/logging.rs +++ b/apollo-router/tests/integration/telemetry/logging.rs @@ -1,8 +1,9 @@ use tower::BoxError; use uuid::Uuid; -use crate::integration::common::{graph_os_enabled, Query}; +use crate::integration::common::graph_os_enabled; use crate::integration::common::IntegrationTest; +use crate::integration::common::Query; use crate::integration::common::Telemetry; #[tokio::test(flavor = "multi_thread")] diff --git a/apollo-router/tests/integration/telemetry/metrics.rs b/apollo-router/tests/integration/telemetry/metrics.rs index f63613fa00..56a5d6223d 100644 --- a/apollo-router/tests/integration/telemetry/metrics.rs +++ b/apollo-router/tests/integration/telemetry/metrics.rs @@ -2,7 +2,8 @@ use std::time::Duration; use serde_json::json; -use crate::integration::common::{graph_os_enabled, Query}; +use crate::integration::common::graph_os_enabled; +use crate::integration::common::Query; use crate::integration::IntegrationTest; const PROMETHEUS_CONFIG: &str = include_str!("fixtures/prometheus.router.yaml"); @@ -106,9 +107,7 @@ async fn test_subgraph_auth_metrics() { router.update_config(PROMETHEUS_CONFIG).await; router.assert_reloaded().await; // This one will not be signed, counters shouldn't increment. - router - .execute_query(Query::default()) - .await; + router.execute_query(Query::default()).await; // Get Prometheus metrics. let metrics_response = router.get_metrics_response().await.unwrap(); @@ -267,11 +266,7 @@ async fn test_gauges_on_reload() { router.execute_default_query().await; // Introspection query - router - .execute_query(Query::introspection() - - ) - .await; + router.execute_query(Query::introspection()).await; // Persisted query router diff --git a/apollo-router/tests/integration/telemetry/mod.rs b/apollo-router/tests/integration/telemetry/mod.rs index 2769f18d9e..c814faa7ff 100644 --- a/apollo-router/tests/integration/telemetry/mod.rs +++ b/apollo-router/tests/integration/telemetry/mod.rs @@ -1,4 +1,5 @@ -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; +use std::collections::HashSet; #[cfg(any(not(feature = "ci"), all(target_arch = "x86_64", target_os = "linux")))] mod datadog; @@ -8,11 +9,10 @@ mod logging; mod metrics; mod otlp; mod propagation; +mod verifier; #[cfg(any(not(feature = "ci"), all(target_arch = "x86_64", target_os = "linux")))] mod zipkin; -mod verifier; -#[derive(buildstructor::Builder)] struct TraceSpec { operation_name: Option, version: Option, @@ -22,7 +22,34 @@ struct TraceSpec { unmeasured_spans: HashSet<&'static str>, priority_sampled: Option<&'static str>, subgraph_sampled: Option, - span_attributes: HashMap<&'static str, Vec<(&'static str, &'static str)>> + span_attributes: HashMap<&'static str, Vec<(&'static str, &'static str)>>, } - +#[buildstructor::buildstructor] +impl TraceSpec { + #[allow(clippy::too_many_arguments)] + #[builder] + pub fn new( + operation_name: Option, + version: Option, + services: Vec<&'static str>, + span_names: HashSet<&'static str>, + measured_spans: HashSet<&'static str>, + unmeasured_spans: HashSet<&'static str>, + priority_sampled: Option<&'static str>, + subgraph_sampled: Option, + span_attributes: HashMap<&'static str, Vec<(&'static str, &'static str)>>, + ) -> Self { + Self { + operation_name, + version, + services, + span_names, + measured_spans, + unmeasured_spans, + priority_sampled, + subgraph_sampled, + span_attributes, + } + } +} diff --git a/apollo-router/tests/integration/telemetry/otlp.rs b/apollo-router/tests/integration/telemetry/otlp.rs index 6947879116..02a7ce3093 100644 --- a/apollo-router/tests/integration/telemetry/otlp.rs +++ b/apollo-router/tests/integration/telemetry/otlp.rs @@ -16,11 +16,12 @@ use wiremock::Mock; use wiremock::MockServer; use wiremock::ResponseTemplate; +use crate::integration::common::graph_os_enabled; +use crate::integration::common::Query; use crate::integration::common::Telemetry; -use crate::integration::common::{graph_os_enabled, Query}; -use crate::integration::IntegrationTest; -use crate::integration::telemetry::TraceSpec; use crate::integration::telemetry::verifier::Verifier; +use crate::integration::telemetry::TraceSpec; +use crate::integration::IntegrationTest; use crate::integration::ValueExt; #[tokio::test(flavor = "multi_thread")] @@ -62,11 +63,7 @@ async fn test_basic() -> Result<(), BoxError> { ) .subgraph_sampled(true) .build() - .validate_otlp_trace( - &mut router, - &mock_server, - Query::default(), - ) + .validate_otlp_trace(&mut router, &mock_server, Query::default()) .await?; TraceSpec::builder() .service("router") @@ -105,11 +102,7 @@ async fn test_otlp_request_with_datadog_propagator() -> Result<(), BoxError> { .priority_sampled("1") .subgraph_sampled(true) .build() - .validate_otlp_trace( - &mut router, - &mock_server, - Query::default(), - ) + .validate_otlp_trace(&mut router, &mock_server, Query::default()) .await?; router.graceful_shutdown().await; Ok(()) @@ -277,12 +270,14 @@ async fn test_untraced_request_no_sample_datadog_agent() -> Result<(), BoxError> let mock_server = mock_otlp_server().await; let config = include_str!("fixtures/otlp_datadog_agent_no_sample.router.yaml") .replace("", &mock_server.uri()); - let mut router = IntegrationTest::builder().config(&config) + let mut router = IntegrationTest::builder() + .config(&config) .telemetry(Telemetry::Otlp { endpoint: Some(format!("{}/v1/traces", mock_server.uri())), }) .extra_propagator(Telemetry::Datadog) - .build().await; + .build() + .await; router.start().await; router.assert_started().await; @@ -316,7 +311,8 @@ async fn test_untraced_request_sample_datadog_agent() -> Result<(), BoxError> { endpoint: Some(format!("{}/v1/traces", mock_server.uri())), }) .extra_propagator(Telemetry::Datadog) - .build().await; + .build() + .await; router.start().await; router.assert_started().await; @@ -542,7 +538,7 @@ async fn test_priority_sampling_no_parent_propagated() -> Result<(), BoxError> { struct OtlpTraceSpec<'a> { trace_spec: TraceSpec, - mock_server: &'a MockServer + mock_server: &'a MockServer, } impl Deref for OtlpTraceSpec<'_> { type Target = TraceSpec; @@ -552,7 +548,6 @@ impl Deref for OtlpTraceSpec<'_> { } } - impl Verifier for OtlpTraceSpec<'_> { fn verify_span_attributes(&self, _span: &Value) -> Result<(), BoxError> { // TODO @@ -580,7 +575,8 @@ impl Verifier for OtlpTraceSpec<'_> { } async fn find_valid_metrics(&self) -> Result<(), BoxError> { - let requests = self.mock_server + let requests = self + .mock_server .received_requests() .await .expect("Could not get otlp requests"); @@ -596,7 +592,6 @@ impl Verifier for OtlpTraceSpec<'_> { } } - async fn get_trace(&self, trace_id: TraceId) -> Result { let requests = self.mock_server.received_requests().await; let trace = Value::Array(requests.unwrap_or_default().iter().filter(|r| r.url.path().ends_with("/traces")) @@ -642,8 +637,6 @@ impl Verifier for OtlpTraceSpec<'_> { Ok(()) } - - fn verify_services(&self, trace: &Value) -> Result<(), axum::BoxError> { let actual_services: HashSet = trace .select_path("$..resource.attributes..[?(@.key == 'service.name')].value.stringValue")? @@ -713,7 +706,6 @@ impl Verifier for OtlpTraceSpec<'_> { Ok(()) } - fn verify_operation_name(&self, trace: &Value) -> Result<(), BoxError> { if let Some(expected_operation_name) = &self.operation_name { let binding = @@ -752,7 +744,6 @@ impl Verifier for OtlpTraceSpec<'_> { } Ok(()) } - } async fn mock_otlp_server() -> MockServer { @@ -788,18 +779,26 @@ impl DatadogId for TraceId { } } - impl TraceSpec { - async fn validate_otlp_trace(self, router: &mut IntegrationTest, mock_server: &MockServer, query: Query) -> Result<(), BoxError>{ + async fn validate_otlp_trace( + self, + router: &mut IntegrationTest, + mock_server: &MockServer, + query: Query, + ) -> Result<(), BoxError> { OtlpTraceSpec { trace_spec: self, - mock_server - }.validate_trace(router, query).await + mock_server, + } + .validate_trace(router, query) + .await } - async fn validate_otlp_metrics(self, mock_server: &MockServer) -> Result<(), BoxError>{ + async fn validate_otlp_metrics(self, mock_server: &MockServer) -> Result<(), BoxError> { OtlpTraceSpec { trace_spec: self, - mock_server - }.validate_metrics().await + mock_server, + } + .validate_metrics() + .await } -} \ No newline at end of file +} diff --git a/apollo-router/tests/integration/telemetry/propagation.rs b/apollo-router/tests/integration/telemetry/propagation.rs index d1b8258c6f..9505efa558 100644 --- a/apollo-router/tests/integration/telemetry/propagation.rs +++ b/apollo-router/tests/integration/telemetry/propagation.rs @@ -1,9 +1,10 @@ use serde_json::json; use tower::BoxError; +use crate::integration::common::graph_os_enabled; use crate::integration::common::IntegrationTest; +use crate::integration::common::Query; use crate::integration::common::Telemetry; -use crate::integration::common::{graph_os_enabled, Query}; #[tokio::test(flavor = "multi_thread")] async fn test_trace_id_via_header() -> Result<(), BoxError> { diff --git a/apollo-router/tests/integration/telemetry/verifier.rs b/apollo-router/tests/integration/telemetry/verifier.rs index bf4ffe4d70..59cbdf2683 100644 --- a/apollo-router/tests/integration/telemetry/verifier.rs +++ b/apollo-router/tests/integration/telemetry/verifier.rs @@ -1,15 +1,22 @@ -use crate::integration::common::Query; -use crate::integration::telemetry::TraceSpec; -use crate::integration::IntegrationTest; -use opentelemetry_api::trace::{SpanContext, TraceId}; -use serde_json::Value; use std::time::Duration; + use anyhow::anyhow; +use opentelemetry_api::trace::SpanContext; +use opentelemetry_api::trace::TraceId; +use serde_json::Value; use tower::BoxError; +use crate::integration::common::Query; +use crate::integration::telemetry::TraceSpec; +use crate::integration::IntegrationTest; + pub trait Verifier { fn spec(&self) -> &TraceSpec; - async fn validate_trace(&self, router: &mut IntegrationTest, query: Query) -> Result<(), BoxError> { + async fn validate_trace( + &self, + router: &mut IntegrationTest, + query: Query, + ) -> Result<(), BoxError> { let (id, response) = router.execute_query(query).await; for _ in 0..20 { if self.find_valid_trace(id).await.is_ok() { @@ -22,7 +29,6 @@ pub trait Verifier { assert!(response.status().is_success()); self.validate_subgraph(subgraph_context)?; Ok(()) - } async fn validate_metrics(&self) -> Result<(), BoxError> { @@ -40,18 +46,12 @@ pub trait Verifier { unimplemented!("find_valid_metrics") } - fn validate_subgraph( - &self, - subgraph_context: SpanContext, - ) -> Result<(), BoxError> { + fn validate_subgraph(&self, subgraph_context: SpanContext) -> Result<(), BoxError> { self.validate_subgraph_priority_sampled(&subgraph_context)?; self.validate_subgraph_sampled(&subgraph_context)?; Ok(()) } - fn validate_subgraph_sampled( - &self, - subgraph_context: &SpanContext, - ) -> Result<(), BoxError> { + fn validate_subgraph_sampled(&self, subgraph_context: &SpanContext) -> Result<(), BoxError> { if let Some(sampled) = self.spec().priority_sampled { assert_eq!( subgraph_context.trace_state().get("psr"), @@ -60,22 +60,19 @@ pub trait Verifier { ); } - Ok(()) } fn validate_subgraph_priority_sampled( &self, subgraph_context: &SpanContext, - ) -> Result<(), BoxError>{ + ) -> Result<(), BoxError> { if let Some(sampled) = self.spec().subgraph_sampled { assert_eq!(subgraph_context.is_sampled(), sampled, "subgraph sampled"); } Ok(()) } - - #[allow(clippy::too_many_arguments)] async fn find_valid_trace(&self, trace_id: TraceId) -> Result<(), BoxError> { // A valid trace has: @@ -156,5 +153,4 @@ pub trait Verifier { fn verify_operation_name(&self, trace: &Value) -> Result<(), BoxError>; fn verify_priority_sampled(&self, trace: &Value) -> Result<(), BoxError>; - } diff --git a/apollo-router/tests/integration/telemetry/zipkin.rs b/apollo-router/tests/integration/telemetry/zipkin.rs index 460c10add4..45f51620d0 100644 --- a/apollo-router/tests/integration/telemetry/zipkin.rs +++ b/apollo-router/tests/integration/telemetry/zipkin.rs @@ -3,16 +3,18 @@ extern crate core; use std::collections::HashSet; use std::ops::Deref; -use crate::integration::common::{Query, Telemetry}; -use crate::integration::telemetry::verifier::Verifier; -use crate::integration::telemetry::TraceSpec; -use crate::integration::IntegrationTest; -use crate::integration::ValueExt; use anyhow::anyhow; use opentelemetry_api::trace::TraceId; use serde_json::Value; use tower::BoxError; +use crate::integration::common::Query; +use crate::integration::common::Telemetry; +use crate::integration::telemetry::verifier::Verifier; +use crate::integration::telemetry::TraceSpec; +use crate::integration::IntegrationTest; +use crate::integration::ValueExt; + #[tokio::test(flavor = "multi_thread")] async fn test_basic() -> Result<(), BoxError> { let mut router = IntegrationTest::builder() @@ -38,7 +40,6 @@ async fn test_basic() -> Result<(), BoxError> { Ok(()) } - struct ZipkinTraceSpec { trace_spec: TraceSpec, } @@ -55,11 +56,9 @@ impl Verifier for ZipkinTraceSpec { Ok(()) } fn verify_version(&self, _trace: &Value) -> Result<(), BoxError> { - Ok(()) } - fn measured_span(&self, _trace: &Value, _name: &str) -> Result { Ok(true) } @@ -76,7 +75,9 @@ impl Verifier for ZipkinTraceSpec { .collect(); tracing::debug!("found services {:?}", actual_services); - let expected_services = self.trace_spec.services + let expected_services = self + .trace_spec + .services .iter() .map(|s| s.to_string()) .collect::>(); @@ -98,8 +99,8 @@ impl Verifier for ZipkinTraceSpec { fn verify_operation_name(&self, trace: &Value) -> Result<(), BoxError> { if let Some(expected_operation_name) = &self.operation_name { - let binding = - trace.select_path("$..[?(@.name == 'supergraph')].tags..['graphql.operation.name']")?; + let binding = trace + .select_path("$..[?(@.name == 'supergraph')].tags..['graphql.operation.name']")?; let operation_name = binding.first(); assert_eq!( operation_name @@ -116,10 +117,15 @@ impl Verifier for ZipkinTraceSpec { Ok(()) } - async fn get_trace(&self, trace_id: TraceId) -> Result { let params = url::form_urlencoded::Serializer::new(String::new()) - .append_pair("service", self.trace_spec.services.first().expect("expected root service")) + .append_pair( + "service", + self.trace_spec + .services + .first() + .expect("expected root service"), + ) .finish(); let id = trace_id.to_string(); diff --git a/apollo-router/tests/integration/traffic_shaping.rs b/apollo-router/tests/integration/traffic_shaping.rs index 5d8b45e28b..579cb2b2a5 100644 --- a/apollo-router/tests/integration/traffic_shaping.rs +++ b/apollo-router/tests/integration/traffic_shaping.rs @@ -5,7 +5,8 @@ use serde_json::json; use tower::BoxError; use wiremock::ResponseTemplate; -use crate::integration::common::{graph_os_enabled, Query}; +use crate::integration::common::graph_os_enabled; +use crate::integration::common::Query; use crate::integration::common::Telemetry; use crate::integration::IntegrationTest; @@ -99,9 +100,13 @@ async fn test_router_timeout_operation_name_in_tracing() -> Result<(), BoxError> router.assert_started().await; let (_trace_id, response) = router - .execute_query(Query::builder().body(json!({ - "query": "query UniqueName { topProducts { name } }" - })).build()) + .execute_query( + Query::builder() + .body(json!({ + "query": "query UniqueName { topProducts { name } }" + })) + .build(), + ) .await; assert_eq!(response.status(), 504); let response = response.text().await?; diff --git a/apollo-router/tests/samples_tests.rs b/apollo-router/tests/samples_tests.rs index abc45e2ff1..5beba9d4b5 100644 --- a/apollo-router/tests/samples_tests.rs +++ b/apollo-router/tests/samples_tests.rs @@ -30,6 +30,7 @@ use wiremock::ResponseTemplate; #[path = "./common.rs"] pub(crate) mod common; pub(crate) use common::IntegrationTest; + use crate::common::Query; fn main() -> Result> { @@ -498,7 +499,12 @@ impl TestExecution { writeln!(out, "header: {:?}\n", headers).unwrap(); let (_, response) = router - .execute_query(Query::builder().body(request).headers(headers.clone()).build()) + .execute_query( + Query::builder() + .body(request) + .headers(headers.clone()) + .build(), + ) .await; writeln!(out, "response headers: {:?}", response.headers()).unwrap();