Integration test for jaeger

.circleci/config.yml

@@ -41,12 +41,33 @@ commands:
           command: |
             sudo apt-get update
             sudo apt-get install -y libssl-dev cmake
+      - run:
+          name: Download jaeger
+          command: |
+            curl -L https://github.com/jaegertracing/jaeger/releases/download/v1.33.0/jaeger-1.33.0-linux-amd64.tar.gz --output jaeger.tar.gz
+            tar -xf jaeger.tar.gz
+            mv jaeger-1.33.0-linux-amd64 jaeger
   macos_install_baseline:
     steps:
       - run: echo "HOMEBREW_NO_AUTO_UPDATE=1" >> $BASH_ENV
       - run: echo "export OPENSSL_ROOT_DIR=/usr/local/opt/openssl@1.1" >> $BASH_ENV
       - run: test -e "$OPENSSL_ROOT_DIR"
       - run: brew install cmake
+      - run:
+          name: Download jaeger
+          command: |
+            curl -L https://github.com/jaegertracing/jaeger/releases/download/v1.33.0/jaeger-1.33.0-darwin-amd64.tar.gz --output jaeger.tar.gz
+            tar -xf jaeger.tar.gz
+            mv jaeger-1.33.0-darwin-amd64 jaeger
+  windows_install_baseline:
+    steps:
+      - run:
+          name: Download jaeger
+          shell: bash.exe
+          command: |
+            curl -L https://github.com/jaegertracing/jaeger/releases/download/v1.33.0/jaeger-1.33.0-windows-amd64.tar.gz --output jaeger.tar.gz
+            tar -xf jaeger.tar.gz
+            mv jaeger-1.33.0-windows-amd64 jaeger
 
   # This job sets up our nodejs dependencies,
   # and makes sure everything is ready to run integration tests

.gitignore

@@ -29,3 +29,4 @@ apollo-spaceport/proto/reports.proto
 #
 # Prevent accidental commits of router.yaml in the root
 /router.yaml
+/jaeger/

apollo-router-core/src/query_cache.rs

@@ -40,8 +40,8 @@ impl QueryCache {
     }
 
     /// Attempt to parse a string to a [`Query`] using cache if possible.
-    #[tracing::instrument(skip_all, level = "info" name = "get_query")]
-    pub async fn get_query(&self, query: impl AsRef<str>) -> Option<Arc<Query>> {
+    #[tracing::instrument(skip_all, level = "info" name = "parse_query")]
+    pub async fn parse_query(&self, query: impl AsRef<str>) -> Option<Arc<Query>> {
         let key = query.as_ref().to_string();
 
         match self.cm.get(key).await {

apollo-router-core/src/query_planner/bridge_query_planner.rs

@@ -69,7 +69,6 @@ impl Service<QueryPlannerRequest> for BridgeQueryPlanner {
 
 #[async_trait]
 impl QueryPlanner for BridgeQueryPlanner {
-    #[tracing::instrument(skip_all, level = "info", name = "plan")]
     async fn get(
         &self,
         query: String,

apollo-router-core/src/services/execution_service.rs

@@ -35,7 +35,6 @@ impl Service<ExecutionRequest> for ExecutionService {
         Poll::Ready(Ok(()))
     }
 
-    #[tracing::instrument(skip_all, level = "info", name = "execute")]
     fn call(&mut self, req: ExecutionRequest) -> Self::Future {
         let this = self.clone();
         let fut = async move {

apollo-router-core/src/services/router_service.rs

@@ -102,7 +102,7 @@ where
                 let body = context.request.body();
                 let variables = body.variables.clone();
                 let query = query_cache
-                    .get_query(
+                    .parse_query(
                         body.query
                             .as_ref()
                             .expect("apollo.ensure-query-is-present has checked this already; qed")

apollo-router/Cargo.toml

@@ -104,6 +104,7 @@ yaml-rust = "0.4.5"
 
 [dev-dependencies]
 insta = "1.12.0"
+jsonpath_lib = "0.3.0"
 maplit = "1.0.2"
 mockall = "0.11.0"
 reqwest = { version = "0.11.10", default-features = false, features = [

apollo-router/src/configuration/mod.rs

@@ -545,15 +545,6 @@ mod tests {
     use std::fs;
     use walkdir::WalkDir;
 
-    macro_rules! assert_config_snapshot {
-        ($file:expr) => {{
-            let config = serde_yaml::from_str::<Configuration>(include_str!($file)).unwrap();
-            insta::with_settings!({sort_maps => true}, {
-                insta::assert_yaml_snapshot!(config);
-            });
-        }};
-    }
-
     #[cfg(unix)]
     #[test]
     fn schema_generation() {

apollo-router/src/lib.rs

@@ -1,5 +1,7 @@
 //! Starts a server that will handle http graphql requests.
 
+extern crate core;
+
 #[cfg(feature = "axum-server")]
 mod axum_http_server_factory;
 pub mod configuration;

apollo-router/src/plugins/telemetry/mod.rs

@@ -8,7 +8,8 @@ use crate::plugins::telemetry::tracing::TracingConfigurator;
 use crate::subscriber::replace_layer;
 use ::tracing::{info_span, Span};
 use apollo_router_core::{
-    http_compat, register_plugin, Handler, Plugin, ResponseBody, RouterRequest, RouterResponse,
+    http_compat, register_plugin, ExecutionRequest, ExecutionResponse, Handler, Plugin,
+    QueryPlannerRequest, QueryPlannerResponse, ResponseBody, RouterRequest, RouterResponse,
     ServiceBuilderExt, SubgraphRequest, SubgraphResponse,
 };
 use apollo_spaceport::server::ReportSpaceport;
@@ -107,6 +108,7 @@ impl Drop for Telemetry {
                         opentelemetry::trace::TracerProvider::force_flush(&tracer_provider);
                     });
                     futures::executor::block_on(jh).expect("failed to flush tracer provider");
+                    global::shutdown_tracer_provider();
                 });
             });
         }
@@ -237,7 +239,9 @@ impl Plugin for Telemetry {
     ) -> BoxService<RouterRequest, RouterResponse, BoxError> {
         let metrics = BasicMetrics::new(&self.meter_provider);
         ServiceBuilder::new()
-            .instrument(self.router_service_span())
+            .instrument(Self::router_service_span(
+                self.config.apollo.clone().unwrap_or_default(),
+            ))
             .service(service)
             .map_future(move |f| {
                 let metrics = metrics.clone();
@@ -268,14 +272,37 @@ impl Plugin for Telemetry {
             .boxed()
     }
 
+    fn query_planning_service(
+        &mut self,
+        service: BoxService<QueryPlannerRequest, QueryPlannerResponse, BoxError>,
+    ) -> BoxService<QueryPlannerRequest, QueryPlannerResponse, BoxError> {
+        ServiceBuilder::new()
+            .instrument(move |_| info_span!("query_planning"))
+            .service(service)
+            .boxed()
+    }
+
+    fn execution_service(
+        &mut self,
+        service: BoxService<ExecutionRequest, ExecutionResponse, BoxError>,
+    ) -> BoxService<ExecutionRequest, ExecutionResponse, BoxError> {
+        ServiceBuilder::new()
+            .instrument(move |_| info_span!("execution"))
+            .service(service)
+            .boxed()
+    }
+
     fn subgraph_service(
         &mut self,
         name: &str,
         service: BoxService<SubgraphRequest, SubgraphResponse, BoxError>,
     ) -> BoxService<SubgraphRequest, SubgraphResponse, BoxError> {
         let metrics = BasicMetrics::new(&self.meter_provider);
         let subgraph_attribute = KeyValue::new("subgraph", name.to_string());
-        service
+        let name = name.to_owned();
+        ServiceBuilder::new()
+            .instrument(move |_| info_span!("subgraph", name = name.as_str()))
+            .service(service)
             .map_future(move |f| {
                 let metrics = metrics.clone();
                 let subgraph_attribute = subgraph_attribute.clone();
@@ -423,42 +450,29 @@ impl Telemetry {
         futures::executor::block_on(jh).expect("failed to replace tracer provider");
     }
 
-    fn router_service_span(&self) -> impl Fn(&RouterRequest) -> Span + Clone {
-        let config = self.config.apollo.clone().unwrap_or_default();
+    fn router_service_span(config: apollo::Config) -> impl Fn(&RouterRequest) -> Span + Clone {
         let client_name_header = config.client_name_header;
         let client_version_header = config.client_version_header;
 
         move |request: &RouterRequest| {
-            let query = request
-                .context
-                .request
-                .body()
-                .query
-                .clone()
-                .unwrap_or_default();
-            let operation_name = request
-                .context
-                .request
+            let http_request = &request.context.request;
+            let headers = http_request.headers();
+            let query = http_request.body().query.clone().unwrap_or_default();
+            let operation_name = http_request
                 .body()
                 .operation_name
                 .clone()
                 .unwrap_or_else(|| "".to_string());
-            let client_name = request
-                .context
-                .request
-                .headers()
+            let client_name = headers
                 .get(&client_name_header)
                 .cloned()
                 .unwrap_or_else(|| HeaderValue::from_static(""));
-            let client_version = request
-                .context
-                .request
-                .headers()
+            let client_version = headers
                 .get(&client_version_header)
                 .cloned()
                 .unwrap_or_else(|| HeaderValue::from_static(""));
             let span = info_span!(
-                "graphql_request",
+                "router",
                 query = query.as_str(),
                 operation_name = operation_name.as_str(),
                 client_name = client_name.to_str().unwrap_or_default(),
@@ -487,6 +501,7 @@ register_plugin!("apollo", "telemetry", Telemetry);
 
 #[cfg(test)]
 mod tests {
+
     #[tokio::test]
     async fn plugin_registered() {
         apollo_router_core::plugins()

apollo-router/src/plugins/telemetry/tracing/apollo_telemetry.rs

@@ -333,7 +333,7 @@ impl SpanExporter for Exporter {
         /*
          * Process the batch
          */
-        for span in batch.iter().filter(|span| span.name == "graphql_request") {
+        for span in batch.iter().filter(|span| span.name == "router") {
             // We can't process a span if we don't have a query
             if let Some(query) = span
                 .attributes

apollo-router/src/plugins/telemetry/tracing/datadog.rs

@@ -42,39 +42,3 @@ impl TracingConfigurator for Config {
         Ok(builder.with_batch_exporter(exporter, opentelemetry::runtime::Tokio))
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use crate::plugins::telemetry::tracing::test::run_query;
-    use opentelemetry::global;
-    use tower::BoxError;
-    use tracing::instrument::WithSubscriber;
-    use tracing_subscriber::layer::SubscriberExt;
-    use tracing_subscriber::Registry;
-
-    // This test can be run manually from your IDE to help with testing otel
-    // It is set to ignore by default as Datadog may not be set up
-    #[ignore]
-    #[tokio::test(flavor = "multi_thread")]
-    async fn test_tracing() -> Result<(), BoxError> {
-        tracing_subscriber::fmt().init();
-
-        global::set_text_map_propagator(opentelemetry_datadog::DatadogPropagator::new());
-        let tracer = opentelemetry_datadog::new_pipeline()
-            .with_service_name("my_app")
-            .install_batch(opentelemetry::runtime::Tokio)?;
-
-        // Create a tracing layer with the configured tracer
-        let telemetry = tracing_opentelemetry::layer().with_tracer(tracer);
-
-        // Use the tracing subscriber `Registry`, or any other subscriber
-        // that impls `LookupSpan`
-        let subscriber = Registry::default().with(telemetry);
-
-        // Trace executed code
-        run_query().with_subscriber(subscriber).await;
-        global::shutdown_tracer_provider();
-
-        Ok(())
-    }
-}

apollo-router/src/plugins/telemetry/tracing/jaeger.rs

@@ -76,39 +76,3 @@ impl TracingConfigurator for Config {
         Ok(builder.with_batch_exporter(exporter, opentelemetry::runtime::Tokio))
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use crate::plugins::telemetry::tracing::test::run_query;
-    use opentelemetry::global;
-    use tower::BoxError;
-    use tracing::instrument::WithSubscriber;
-    use tracing_subscriber::layer::SubscriberExt;
-    use tracing_subscriber::Registry;
-
-    // This test can be run manually from your IDE to help with testing otel
-    // It is set to ignore by default as jaeger may not be set up
-    #[ignore]
-    #[tokio::test(flavor = "multi_thread")]
-    async fn test_tracing() -> Result<(), BoxError> {
-        tracing_subscriber::fmt().init();
-
-        global::set_text_map_propagator(opentelemetry_jaeger::Propagator::new());
-        let tracer = opentelemetry_jaeger::new_pipeline()
-            .with_service_name("my_app")
-            .install_batch(opentelemetry::runtime::Tokio)?;
-
-        // Create a tracing layer with the configured tracer
-        let telemetry = tracing_opentelemetry::layer().with_tracer(tracer);
-
-        // Use the tracing subscriber `Registry`, or any other subscriber
-        // that impls `LookupSpan`
-        let subscriber = Registry::default().with(telemetry);
-
-        // Trace executed code
-        run_query().with_subscriber(subscriber).await;
-        global::shutdown_tracer_provider();
-
-        Ok(())
-    }
-}

apollo-router/src/plugins/telemetry/tracing/mod.rs

@@ -12,31 +12,3 @@ pub mod zipkin;
 pub trait TracingConfigurator {
     fn apply(&self, builder: Builder, trace_config: &Trace) -> Result<Builder, BoxError>;
 }
-
-#[cfg(test)]
-mod test {
-    use http::{Method, Request, Uri};
-    use opentelemetry::global;
-    use opentelemetry_http::HttpClient;
-    use tracing::{info_span, Instrument};
-    use tracing_opentelemetry::OpenTelemetrySpanExt;
-    pub async fn run_query() {
-        let span = info_span!("client_request");
-        let client = reqwest::Client::new();
-
-        let mut request = Request::builder()
-            .method(Method::POST)
-            .uri(Uri::from_static("http://localhost:4000"))
-            .body(r#"{"query":"query {\n  topProducts {\n    name\n  }\n}","variables":{}}"#.into())
-            .unwrap();
-
-        global::get_text_map_propagator(|propagator| {
-            propagator.inject_context(
-                &span.context(),
-                &mut opentelemetry_http::HeaderInjector(request.headers_mut()),
-            )
-        });
-
-        client.send(request).instrument(span).await.unwrap();
-    }
-}