add opentracing layer

apollo-router-core/src/layer.rs

@@ -72,6 +72,16 @@ pub trait ConfigurableLayer: Send + Sync + 'static + Sized {
 /// Layers will appear in the configuration as a layer property called: {group}_{name}
 #[macro_export]
 macro_rules! register_layer {
+    ($name: literal, $value: ident) => {
+        startup::on_startup! {
+            let qualified_name = $name.to_string();
+
+            $crate::register_layer(qualified_name, $crate::LayerFactory::new(|configuration| {
+                let layer = $value::new(serde_json::from_value(configuration.clone())?)?;
+                Ok(tower::util::BoxLayer::new(layer))
+            }, |gen| gen.subschema_for::<<$value as $crate::ConfigurableLayer>::Config>()));
+        }
+    };
     ($group: literal, $name: literal, $value: ident) => {
         startup::on_startup! {
             let qualified_name = if $group == "" {

apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__schema_generation.snap

@@ -1,6 +1,6 @@
 ---
 source: apollo-router/src/configuration/mod.rs
-assertion_line: 645
+assertion_line: 474
 expression: "&schema"
 
 ---

apollo-router/src/layers/mod.rs

@@ -1 +1,2 @@
 mod hello;
+pub mod opentracing;

apollo-router/src/layers/opentracing.rs

@@ -0,0 +1,131 @@
+use std::fmt::Display;
+use std::task::{Context, Poll};
+
+use apollo_router_core::{ConfigurableLayer, SubgraphRequest};
+use http::HeaderValue;
+use opentelemetry::trace::TraceContextExt;
+use schemars::JsonSchema;
+use serde::Deserialize;
+use tower::{BoxError, Layer, Service};
+use tracing::instrument::Instrumented;
+use tracing::{span, Instrument, Level, Span};
+use tracing_opentelemetry::OpenTelemetrySpanExt;
+
+#[derive(Clone, JsonSchema, Deserialize, Debug)]
+#[serde(rename_all = "snake_case")]
+pub enum PropagationFormat {
+    Jaeger,
+    ZipkinB3,
+}
+
+impl Display for PropagationFormat {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            PropagationFormat::Jaeger => write!(f, "jaeger"),
+            PropagationFormat::ZipkinB3 => write!(f, "zipkin_b3"),
+        }
+    }
+}
+
+#[derive(Clone, JsonSchema, Deserialize, Debug)]
+pub struct OpenTracingConfig {
+    format: PropagationFormat,
+}
+
+#[derive(Debug)]
+pub struct OpenTracingLayer {
+    format: PropagationFormat,
+}
+
+impl ConfigurableLayer for OpenTracingLayer {
+    type Config = OpenTracingConfig;
+    fn new(configuration: Self::Config) -> Result<Self, BoxError> {
+        Ok(Self {
+            format: configuration.format,
+        })
+    }
+}
+
+impl<S> Layer<S> for OpenTracingLayer {
+    type Service = OpenTracingService<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        OpenTracingService {
+            inner,
+            format: self.format.clone(),
+        }
+    }
+}
+
+pub struct OpenTracingService<S> {
+    inner: S,
+    format: PropagationFormat,
+}
+
+impl<S> Service<SubgraphRequest> for OpenTracingService<S>
+where
+    S: Service<SubgraphRequest>,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = Instrumented<<S as tower::Service<SubgraphRequest>>::Future>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, mut req: SubgraphRequest) -> Self::Future {
+        let current_span = Span::current();
+        let span_context = current_span.context();
+        let span_ref = span_context.span();
+        let current_span_ctx = span_ref.span_context();
+        let (trace_id, parent_span_id, trace_flags) = (
+            current_span_ctx.trace_id(),
+            current_span_ctx.span_id(),
+            current_span_ctx.trace_flags(),
+        );
+
+        let new_span = span!(parent: current_span, Level::INFO, "subgraph_request");
+        let new_span_context = new_span.context();
+        let new_span_ref = new_span_context.span();
+        let span_id = new_span_ref.span_context().span_id();
+
+        match self.format {
+            PropagationFormat::Jaeger => {
+                req.http_request.headers_mut().insert(
+                    "uber-trace-id",
+                    HeaderValue::from_str(&format!(
+                        "{}:{}:{}:{}",
+                        trace_id,
+                        parent_span_id,
+                        span_id,
+                        trace_flags.to_u8()
+                    ))
+                    .unwrap(),
+                );
+            }
+            PropagationFormat::ZipkinB3 => {
+                req.http_request.headers_mut().insert(
+                    "X-B3-TraceId",
+                    HeaderValue::from_str(&trace_id.to_string()).unwrap(),
+                );
+                req.http_request.headers_mut().insert(
+                    "X-B3-SpanId",
+                    HeaderValue::from_str(&span_id.to_string()).unwrap(),
+                );
+                req.http_request.headers_mut().insert(
+                    "X-B3-ParentSpanId",
+                    HeaderValue::from_str(&parent_span_id.to_string()).unwrap(),
+                );
+                req.http_request.headers_mut().insert(
+                    "X-B3-Sampled",
+                    HeaderValue::from_static(
+                        current_span_ctx.is_sampled().then(|| "1").unwrap_or("0"),
+                    ),
+                );
+            }
+        }
+
+        self.inner.call(req).instrument(new_span)
+    }
+}

apollo-router/src/plugins/reporting.rs

@@ -5,8 +5,14 @@ use crate::apollo_telemetry::new_pipeline;
 use crate::apollo_telemetry::SpaceportConfig;
 use crate::apollo_telemetry::StudioGraph;
 use crate::configuration::{default_service_name, default_service_namespace};
+use crate::layers::opentracing::OpenTracingConfig;
+use crate::layers::opentracing::OpenTracingLayer;
 use crate::set_subscriber;
 use crate::GLOBAL_ENV_FILTER;
+
+use apollo_router_core::ConfigurableLayer;
+use apollo_router_core::SubgraphRequest;
+use apollo_router_core::SubgraphResponse;
 use apollo_router_core::{register_plugin, Plugin};
 use apollo_spaceport::server::ReportSpaceport;
 use derivative::Derivative;
@@ -28,7 +34,9 @@ use std::net::SocketAddr;
 use std::pin::Pin;
 use std::str::FromStr;
 use std::sync::Arc;
-use tower::BoxError;
+use tower::util::BoxService;
+use tower::Layer;
+use tower::{BoxError, ServiceExt};
 use tracing_subscriber::prelude::*;
 use tracing_subscriber::EnvFilter;
 
@@ -159,6 +167,7 @@ impl std::error::Error for ReportingError {}
 struct Reporting {
     config: Conf,
     tx: tokio::sync::mpsc::Sender<SpaceportConfig>,
+    opentracing_layer: Option<OpenTracingLayer>,
 }
 
 #[derive(Debug, Deserialize, JsonSchema)]
@@ -168,6 +177,8 @@ struct Conf {
     pub graph: Option<StudioGraph>,
 
     pub opentelemetry: Option<OpenTelemetry>,
+
+    pub opentracing: Option<OpenTracingConfig>,
 }
 
 fn studio_graph() -> Option<StudioGraph> {
@@ -254,11 +265,29 @@ impl Plugin for Reporting {
             }
             tracing::debug!("terminating spaceport loop");
         });
+
+        let mut opentracing_layer = None;
+        if let Some(opentracing_conf) = &configuration.opentracing {
+            opentracing_layer = OpenTracingLayer::new(opentracing_conf.clone())?.into();
+        }
+
         Ok(Reporting {
             config: configuration,
             tx,
+            opentracing_layer,
         })
     }
+
+    fn subgraph_service(
+        &mut self,
+        _name: &str,
+        service: BoxService<SubgraphRequest, SubgraphResponse, BoxError>,
+    ) -> BoxService<SubgraphRequest, SubgraphResponse, BoxError> {
+        match &self.opentracing_layer {
+            Some(opentracing_layer) => opentracing_layer.layer(service).boxed(),
+            None => service,
+        }
+    }
 }
 
 impl Reporting {