Use opentelemetry semantic conventions (#34)

Migrates span field names to use [OpenTelemetry Semantic
Convention](https://opentelemetry.io/docs/specs/semconv/) names. This
ensures maximum consumer discoverability.

Author: akrantz01

src/graphql.rs

@@ -6,8 +6,8 @@ use async_graphql::{
     parser::types::{ExecutableDocument, OperationType, Selection},
     Response, ServerError, ServerResult, ValidationResult, Value, Variables,
 };
-use std::{sync::Arc, time::Instant};
-use tracing::{debug, info, span, warn, Instrument, Level};
+use std::sync::Arc;
+use tracing::{span, warn, Instrument, Level, Span};
 
 /// Adds tracing to the GraphQL flow
 ///
@@ -35,31 +35,24 @@ impl Extension for LoggingExtension {
         next: NextParseQuery<'_>,
     ) -> ServerResult<ExecutableDocument> {
         let span = span!(Level::INFO, "parse");
-        let now = Instant::now();
 
         async move {
-            debug!("started parsing request");
-            let result = next.run(ctx, query, variables).await;
-            debug!(
-                latency = format_args!("{} ms", now.elapsed().as_millis()),
-                "finished parsing request",
-            );
-
-            let document = result?;
+            let document = next.run(ctx, query, variables).await?;
 
             let is_schema = document
                 .operations
                 .iter()
                 .filter(|(_, operation)| operation.node.ty == OperationType::Query)
                 .any(|(_, operation)| operation.node.selection_set.node.items.iter().any(|selection| matches!(&selection.node, Selection::Field(field) if field.node.name.node == "__schema")));
             if !is_schema {
-                info!(document = ctx.stringify_execute_doc(&document, variables));
+                let span = Span::current();
+                span.record("graphql.document", ctx.stringify_execute_doc(&document, variables));
             }
 
             Ok(document)
         }
-        .instrument(span)
-        .await
+            .instrument(span)
+            .await
     }
 
     async fn validation(
@@ -68,7 +61,19 @@ impl Extension for LoggingExtension {
         next: NextValidation<'_>,
     ) -> async_graphql::Result<ValidationResult, Vec<ServerError>> {
         let span = span!(Level::INFO, "validation");
-        next.run(ctx).instrument(span).await
+        async move {
+            let result = next.run(ctx).await;
+
+            if let Ok(result) = result {
+                let span = Span::current();
+                span.record("graphql.complexity", result.complexity);
+                span.record("graphql.depth", result.depth);
+            }
+
+            result
+        }
+        .instrument(span)
+        .await
     }
 
     async fn execute(
@@ -77,21 +82,9 @@ impl Extension for LoggingExtension {
         operation_name: Option<&str>,
         next: NextExecute<'_>,
     ) -> Response {
-        let span = span!(Level::INFO, "execute", operation = %operation_name.unwrap_or_default());
-        let now = Instant::now();
+        let span = span!(Level::INFO, "execute", graphql.operation.name = %operation_name.unwrap_or_default());
 
-        async move {
-            debug!("started execution");
-            let response = next.run(ctx, operation_name).await;
-            debug!(
-                latency = format_args!("{} ms", now.elapsed().as_millis()),
-                "finished execution",
-            );
-
-            response
-        }
-        .instrument(span)
-        .await
+        next.run(ctx, operation_name).instrument(span).await
     }
 
     async fn resolve(
@@ -107,23 +100,20 @@ impl Extension for LoggingExtension {
 
         let span = span!(
             Level::INFO, "field",
-            path = %path_node,
-            parent_type = %info.parent_type,
-            return_type = %info.return_type,
+            graphql.node.path = %path_node,
+            graphql.node.parent_type = %info.parent_type,
+            graphql.node.return_type = %info.return_type,
         );
-        let now = Instant::now();
 
         async move {
-            debug!("started resolution");
             let result = next.run(ctx, info).await;
-            debug!(
-                latency = format_args!("{} ms", now.elapsed().as_millis()),
-                "finished resolution",
-            );
 
             // we do the actual error handling elsewhere, these errors are typically caused by
             // user error.
             if let Err(ref err) = result {
+                let span = Span::current();
+                span.record("error", true);
+                span.record("error.type", "graphql");
                 warn!(error = %err.message, locations = ?err.locations, path = ?err.path);
             }
 

src/http.rs

@@ -1,49 +1,132 @@
-use http::Request;
-#[cfg(feature = "opentelemetry")]
-use opentelemetry_sdk::propagation::TraceContextPropagator;
+use http::{header, HeaderMap, Request, Response};
+use std::time::Duration;
 use tower_http::{
-    classify::{ServerErrorsAsFailures, SharedClassifier},
-    trace::{DefaultOnRequest, DefaultOnResponse, MakeSpan, TraceLayer},
+    classify::{SharedClassifier, StatusInRangeAsFailures, StatusInRangeFailureClass},
+    trace::{self, TraceLayer},
 };
-use tracing::{span, Level, Span};
+use tracing::{info, span, Level, Span};
 use uuid::Uuid;
 
 /// Creates a custom tracing span
-#[derive(Clone, Debug, Default)]
-pub struct MakeSpanWithId {
-    #[cfg(feature = "opentelemetry")]
-    propagator: TraceContextPropagator,
-}
+#[derive(Clone, Copy, Debug)]
+pub struct MakeSpan;
 
-impl<B> MakeSpan<B> for MakeSpanWithId {
+impl<B> trace::MakeSpan<B> for MakeSpan {
     fn make_span(&mut self, request: &Request<B>) -> Span {
+        let uri = request.uri();
         let span = span!(
             Level::INFO,
             "request",
-            method = %request.method(),
-            uri = %request.uri(),
-            version = ?request.version(),
+            span.kind = "server",
+            http.request.method = %request.method(),
+            network.protocol.version = ?request.version(),
+            uri.path = uri.path(),
+            uri.scheme = uri.scheme_str().unwrap_or_default(),
+            uri.query = uri.query().unwrap_or_default(),
             id = %Uuid::new_v4(),
         );
 
+        let headers = request.headers();
+        if let Some(user_agent) = headers
+            .get(header::USER_AGENT)
+            .and_then(|h| h.to_str().ok())
+        {
+            span.record("user_agent.original", user_agent);
+        }
+
+        record_headers("request", headers, &span);
+
         #[cfg(feature = "opentelemetry")]
         {
             use opentelemetry::propagation::TextMapPropagator;
             use opentelemetry_http::HeaderExtractor;
+            use opentelemetry_sdk::propagation::{BaggagePropagator, TraceContextPropagator};
             use tracing_opentelemetry::OpenTelemetrySpanExt;
 
-            let context = self.propagator.extract(&HeaderExtractor(request.headers()));
+            let extractor = HeaderExtractor(headers);
+            let context = TraceContextPropagator::new().extract(&extractor);
+            let context = BaggagePropagator::new().extract_with_context(&context, &extractor);
             span.set_parent(context);
         }
 
         span
     }
 }
 
+/// Adds custom trace metadata from the request
+#[derive(Clone, Copy, Debug)]
+pub struct OnRequest;
+
+impl<B> trace::OnRequest<B> for OnRequest {
+    fn on_request(&mut self, _request: &Request<B>, _span: &Span) {
+        info!("started processing request");
+    }
+}
+
+/// Adds custom trace metadata from the request
+#[derive(Clone, Copy, Debug)]
+pub struct OnResponse;
+
+impl<B> trace::OnResponse<B> for OnResponse {
+    fn on_response(self, response: &Response<B>, _latency: Duration, span: &Span) {
+        span.record("http.response.status", response.status().as_u16());
+        record_headers("response", response.headers(), span);
+        info!("finished processing request");
+    }
+}
+
+/// Adds custom trace metadata when a failure occurs
+#[derive(Clone, Copy, Debug)]
+pub struct OnFailure;
+
+impl trace::OnFailure<StatusInRangeFailureClass> for OnFailure {
+    fn on_failure(&mut self, class: StatusInRangeFailureClass, _latency: Duration, span: &Span) {
+        if span.has_field("error") || span.has_field("error.type") {
+            return;
+        }
+
+        span.record("error", true);
+
+        let kind = match class {
+            StatusInRangeFailureClass::StatusCode(status) => {
+                if status.is_client_error() {
+                    "client_error"
+                } else {
+                    "server_error"
+                }
+            }
+            StatusInRangeFailureClass::Error(_) => "internal_error",
+        };
+        span.record("error.type", kind);
+    }
+}
+
 /// Create a logging middleware layer
-pub fn layer() -> TraceLayer<SharedClassifier<ServerErrorsAsFailures>, MakeSpanWithId> {
-    TraceLayer::new_for_http()
-        .make_span_with(MakeSpanWithId::default())
-        .on_request(DefaultOnRequest::new().level(Level::INFO))
-        .on_response(DefaultOnResponse::new().level(Level::INFO))
+pub fn layer() -> TraceLayer<
+    SharedClassifier<StatusInRangeAsFailures>,
+    MakeSpan,
+    OnRequest,
+    OnResponse,
+    (),
+    (),
+    OnFailure,
+> {
+    let classifier = StatusInRangeAsFailures::new_for_client_and_server_errors();
+    TraceLayer::new(classifier.into_make_classifier())
+        .make_span_with(MakeSpan)
+        .on_request(OnRequest)
+        .on_response(OnResponse)
+        .on_failure(OnFailure)
+        .on_eos(())
+        .on_body_chunk(())
+}
+
+fn record_headers(direction: &'static str, headers: &HeaderMap, span: &Span) {
+    for (header, value) in headers
+        .iter()
+        .filter_map(|(header, value)| value.to_str().ok().map(|value| (header, value)))
+    {
+        let header = header.as_str().to_lowercase();
+        span.record(format!("http.{direction}.header.{header}").as_str(), value);
+    }
 }