Merge pull request #2441 from calebschoepp/trigger-tracing

Trigger tracing

Author: calebschoepp

crates/trigger-http/Cargo.toml

@@ -54,3 +54,6 @@ harness = false
 llm = ["spin-trigger/llm"]
 llm-metal = ["llm", "spin-trigger/llm-metal"]
 llm-cublas = ["llm", "spin-trigger/llm-cublas"]
+
+[lints]
+workspace = true

crates/trigger-http/src/handler.rs

@@ -17,15 +17,15 @@ use spin_trigger::TriggerAppEngine;
 use spin_world::v1::http_types;
 use std::sync::Arc;
 use tokio::{sync::oneshot, task};
-use tracing::{instrument, Instrument};
+use tracing::{instrument, Instrument, Level};
 use wasmtime_wasi_http::{proxy::Proxy, WasiHttpView};
 
 #[derive(Clone)]
 pub struct HttpHandlerExecutor;
 
 #[async_trait]
 impl HttpExecutor for HttpHandlerExecutor {
-    #[instrument(name = "execute_wasm", skip_all, fields(otel.kind = "server"))]
+    #[instrument(name = "spin_trigger_http.execute_wasm", skip_all, err(level = Level::INFO), fields(otel.name = format!("execute_wasm_component {}", component_id)))]
     async fn execute(
         &self,
         engine: Arc<TriggerAppEngine<HttpTrigger>>,

crates/trigger-http/src/instrument.rs

@@ -0,0 +1,93 @@
+use anyhow::Result;
+use http::Response;
+use tracing::Level;
+use wasmtime_wasi_http::body::HyperIncomingBody;
+
+/// Create a span for an HTTP request.
+macro_rules! http_span {
+    ($request:tt, $addr:tt) => {
+        tracing::info_span!(
+            "spin_trigger_http.handle_http_request",
+            "otel.kind" = "server",
+            "http.request.method" = %$request.method(),
+            "network.peer.address" = %$addr.ip(),
+            "network.peer.port" = %$addr.port(),
+            "network.protocol.name" = "http",
+            "url.path" = $request.uri().path(),
+            "url.query" = $request.uri().query().unwrap_or(""),
+            "url.scheme" = $request.uri().scheme_str().unwrap_or(""),
+            "client.address" = $request.headers().get("x-forwarded-for").and_then(|val| val.to_str().ok()),
+            // Recorded later
+            "error.type" = Empty,
+            "http.response.status_code" = Empty,
+            "http.route" = Empty,
+            "otel.name" = Empty,
+        )
+    };
+}
+
+pub(crate) use http_span;
+
+/// Finish setting attributes on the HTTP span.
+pub(crate) fn finalize_http_span(
+    response: Result<Response<HyperIncomingBody>>,
+    method: String,
+) -> Result<Response<HyperIncomingBody>> {
+    let span = tracing::Span::current();
+    match response {
+        Ok(response) => {
+            let matched_route = response.extensions().get::<MatchedRoute>();
+            // Set otel.name and http.route
+            if let Some(MatchedRoute { route }) = matched_route {
+                span.record("http.route", route);
+                span.record("otel.name", format!("{method} {route}"));
+            } else {
+                span.record("otel.name", method);
+            }
+
+            // Set status code
+            span.record("http.response.status_code", response.status().as_u16());
+
+            Ok(response)
+        }
+        Err(err) => {
+            instrument_error(&err);
+            span.record("http.response.status_code", 500);
+            span.record("otel.name", method);
+            Err(err)
+        }
+    }
+}
+
+/// Marks the current span as errored.
+pub(crate) fn instrument_error(err: &anyhow::Error) {
+    let span = tracing::Span::current();
+    tracing::event!(target:module_path!(), Level::INFO, error = %err);
+    span.record("error.type", format!("{:?}", err));
+}
+
+/// MatchedRoute is used as a response extension to track the route that was matched for OTel
+/// tracing purposes.
+#[derive(Clone)]
+pub struct MatchedRoute {
+    pub route: String,
+}
+
+impl MatchedRoute {
+    pub fn set_response_extension(
+        resp: &mut Response<HyperIncomingBody>,
+        route: impl Into<String>,
+    ) {
+        resp.extensions_mut().insert(MatchedRoute {
+            route: route.into(),
+        });
+    }
+
+    pub fn with_response_extension(
+        mut resp: Response<HyperIncomingBody>,
+        route: impl Into<String>,
+    ) -> Response<HyperIncomingBody> {
+        Self::set_response_extension(&mut resp, route);
+        resp
+    }
+}

crates/trigger-http/src/lib.rs

@@ -1,6 +1,7 @@
 //! Implementation for the Spin HTTP engine.
 
 mod handler;
+mod instrument;
 mod tls;
 mod wagi;
 
@@ -25,6 +26,7 @@ use hyper::{
     Request, Response,
 };
 use hyper_util::rt::tokio::TokioIo;
+use instrument::{finalize_http_span, http_span};
 use spin_app::{AppComponent, APP_DESCRIPTION_KEY};
 use spin_core::{Engine, OutboundWasiHttpHandler};
 use spin_http::{
@@ -42,10 +44,17 @@ use tokio::{
     net::TcpListener,
     task,
 };
-use tracing::{log, Instrument};
-use wasmtime_wasi_http::{body::HyperIncomingBody as Body, WasiHttpView};
+use tracing::{field::Empty, log, Instrument};
+use wasmtime_wasi_http::{
+    body::{HyperIncomingBody as Body, HyperOutgoingBody},
+    WasiHttpView,
+};
 
-use crate::{handler::HttpHandlerExecutor, wagi::WagiHttpExecutor};
+use crate::{
+    handler::HttpHandlerExecutor,
+    instrument::{instrument_error, MatchedRoute},
+    wagi::WagiHttpExecutor,
+};
 
 pub use tls::TlsConfig;
 
@@ -240,19 +249,22 @@ impl HttpTrigger {
             req.uri()
         );
 
-        let path = req.uri().path();
+        let path = req.uri().path().to_string();
 
         // Handle well-known spin paths
         if let Some(well_known) = path.strip_prefix(spin_http::WELL_KNOWN_PREFIX) {
             return match well_known {
-                "health" => Ok(Response::new(body::full(Bytes::from_static(b"OK")))),
-                "info" => self.app_info(),
+                "health" => Ok(MatchedRoute::with_response_extension(
+                    Response::new(body::full(Bytes::from_static(b"OK"))),
+                    path,
+                )),
+                "info" => self.app_info(path),
                 _ => Self::not_found(NotFoundRouteKind::WellKnown),
             };
         }
 
         // Route to app component
-        match self.router.route(path) {
+        match self.router.route(&path) {
             Ok(component_id) => {
                 let trigger = self.component_trigger_configs.get(component_id).unwrap();
 
@@ -293,10 +305,14 @@ impl HttpTrigger {
                     }
                 };
                 match res {
-                    Ok(res) => Ok(res),
+                    Ok(res) => Ok(MatchedRoute::with_response_extension(
+                        res,
+                        raw_route.to_string(),
+                    )),
                     Err(e) => {
                         log::error!("Error processing request: {:?}", e);
-                        Self::internal_error(None)
+                        instrument_error(&e);
+                        Self::internal_error(None, raw_route.to_string())
                     }
                 }
             }
@@ -305,24 +321,30 @@ impl HttpTrigger {
     }
 
     /// Returns spin status information.
-    fn app_info(&self) -> Result<Response<Body>> {
+    fn app_info(&self, route: String) -> Result<Response<Body>> {
         let info = AppInfo::new(self.engine.app());
         let body = serde_json::to_vec_pretty(&info)?;
-        Ok(Response::builder()
-            .header("content-type", "application/json")
-            .body(body::full(body.into()))?)
+        Ok(MatchedRoute::with_response_extension(
+            Response::builder()
+                .header("content-type", "application/json")
+                .body(body::full(body.into()))?,
+            route,
+        ))
     }
 
     /// Creates an HTTP 500 response.
-    fn internal_error(body: Option<&str>) -> Result<Response<Body>> {
+    fn internal_error(body: Option<&str>, route: String) -> Result<Response<Body>> {
         let body = match body {
             Some(body) => body::full(Bytes::copy_from_slice(body.as_bytes())),
             None => body::empty(),
         };
 
-        Ok(Response::builder()
-            .status(StatusCode::INTERNAL_SERVER_ERROR)
-            .body(body)?)
+        Ok(MatchedRoute::with_response_extension(
+            Response::builder()
+                .status(StatusCode::INTERNAL_SERVER_ERROR)
+                .body(body)?,
+            route,
+        ))
     }
 
     /// Creates an HTTP 404 response.
@@ -351,38 +373,7 @@ impl HttpTrigger {
                 .keep_alive(true)
                 .serve_connection(
                     TokioIo::new(stream),
-                    service_fn(move |request| {
-                        let self_ = self_.clone();
-                        let span = tracing::info_span!(
-                            "handle_http_request",
-                            "otel.kind" = "server",
-                            "http.request.method" = %request.method(),
-                            "network.peer.address" = %addr.ip(),
-                            "network.peer.port" = %addr.port(),
-                            "network.protocol.name" = "http",
-                            "url.path" = request.uri().path(),
-                            "url.query" = request.uri().query().unwrap_or(""),
-                            "url.scheme" = request.uri().scheme_str().unwrap_or(""),
-                            "client.address" = request.headers().get("x-forwarded-for").and_then(|val| val.to_str().ok()),
-                            // TODO(Caleb): Recorded later
-                            // "error.type" = Empty,
-                            // "http.response.status_code" = Empty,
-                            // "http.route" = Empty,
-                        );
-                        async move {
-                            self_
-                                .handle(
-                                    request.map(|body: Incoming| {
-                                        body.map_err(wasmtime_wasi_http::hyper_response_error)
-                                            .boxed()
-                                    }),
-                                    Scheme::HTTP,
-                                    addr,
-                                )
-                                .instrument(span)
-                                .await
-                        }
-                    }),
+                    service_fn(move |request| self_.clone().instrumented_service_fn(addr, request)),
                 )
                 .await
             {
@@ -391,6 +382,30 @@ impl HttpTrigger {
         });
     }
 
+    async fn instrumented_service_fn(
+        self: Arc<Self>,
+        addr: SocketAddr,
+        request: Request<Incoming>,
+    ) -> Result<Response<HyperOutgoingBody>> {
+        let span = http_span!(request, addr);
+        let method = request.method().to_string();
+        async {
+            let result = self
+                .handle(
+                    request.map(|body: Incoming| {
+                        body.map_err(wasmtime_wasi_http::hyper_response_error)
+                            .boxed()
+                    }),
+                    Scheme::HTTP,
+                    addr,
+                )
+                .await;
+            finalize_http_span(result, method)
+        }
+        .instrument(span)
+        .await
+    }
+
     async fn serve(self, listen_addr: SocketAddr) -> Result<()> {
         let self_ = Arc::new(self);
 

crates/trigger-http/src/wagi.rs

@@ -8,6 +8,7 @@ use hyper::{Request, Response};
 use spin_core::WasiVersion;
 use spin_http::{config::WagiTriggerConfig, routes::RoutePattern, wagi};
 use spin_trigger::TriggerAppEngine;
+use tracing::{instrument, Level};
 use wasi_common_preview1::{pipe::WritePipe, I32Exit};
 
 use crate::{Body, HttpExecutor, HttpTrigger};
@@ -19,6 +20,7 @@ pub struct WagiHttpExecutor {
 
 #[async_trait]
 impl HttpExecutor for WagiHttpExecutor {
+    #[instrument(name = "spin_trigger_http.execute_wagi", skip_all, err(level = Level::INFO), fields(otel.name = format!("execute_wagi_component {}", component)))]
     async fn execute(
         &self,
         engine: Arc<TriggerAppEngine<HttpTrigger>>,

crates/trigger-redis/Cargo.toml

@@ -24,3 +24,6 @@ tokio = { version = "1.23", features = ["full"] }
 
 [dev-dependencies]
 spin-testing = { path = "../testing" }
+
+[lints]
+workspace = true

crates/trigger-redis/src/lib.rs

@@ -11,6 +11,7 @@ use spin_core::{async_trait, InstancePre};
 use spin_trigger::{cli::NoArgs, TriggerAppEngine, TriggerExecutor};
 use std::collections::HashMap;
 use std::sync::Arc;
+use tracing::{instrument, Level};
 
 use crate::spin::SpinRedisExecutor;
 
@@ -110,6 +111,9 @@ impl TriggerExecutor for RedisTrigger {
 
 impl RedisTrigger {
     // Handle the message.
+    #[instrument(name = "spin_trigger_redis.handle_message", skip(self, channel_components, msg),
+        err(level = Level::INFO), fields(otel.name = format!("{} receive", msg.get_channel_name()),
+        otel.kind = "consumer", messaging.operation = "receive", messaging.system = "redis"))]
     async fn handle(
         &self,
         address: &str,

crates/trigger-redis/src/spin.rs

@@ -3,6 +3,7 @@ use async_trait::async_trait;
 use spin_core::Instance;
 use spin_trigger::TriggerAppEngine;
 use spin_world::v1::redis_types::{Error, Payload};
+use tracing::{instrument, Level};
 
 use crate::{RedisExecutor, RedisTrigger, Store};
 
@@ -11,6 +12,7 @@ pub struct SpinRedisExecutor;
 
 #[async_trait]
 impl RedisExecutor for SpinRedisExecutor {
+    #[instrument(name = "spin_trigger_redis.execute_wasm", skip(self, engine, payload), err(level = Level::INFO), fields(otel.name = format!("execute_wasm_component {}", component_id)))]
     async fn execute(
         &self,
         engine: &TriggerAppEngine<RedisTrigger>,