feat(otel): allow to configure the faas.trigger attribute of the span (#903)

* feat(otel): allow to configure the faas.trigger attribute of the span

* fix: update opentelemetry-tracing example

* fix: use datasource as default faas.trigger + introduce enum OpenTelemetryFaasTrigger

Author: Oliboy50

examples/opentelemetry-tracing/src/main.rs

@@ -1,4 +1,7 @@
-use lambda_runtime::{layers::OpenTelemetryLayer as OtelLayer, LambdaEvent, Runtime};
+use lambda_runtime::{
+    layers::{OpenTelemetryFaasTrigger, OpenTelemetryLayer as OtelLayer},
+    LambdaEvent, Runtime,
+};
 use opentelemetry::trace::TracerProvider;
 use opentelemetry_sdk::{runtime, trace};
 use tower::{service_fn, BoxError};
@@ -24,10 +27,15 @@ async fn main() -> Result<(), BoxError> {
         .init();
 
     // Initialize the Lambda runtime and add OpenTelemetry tracing
-    let runtime = Runtime::new(service_fn(echo)).layer(OtelLayer::new(|| {
-        // Make sure that the trace is exported before the Lambda runtime is frozen
-        tracer_provider.force_flush();
-    }));
+    let runtime = Runtime::new(service_fn(echo)).layer(
+        // Create a tracing span for each Lambda invocation
+        OtelLayer::new(|| {
+            // Make sure that the trace is exported before the Lambda runtime is frozen
+            tracer_provider.force_flush();
+        })
+        // Set the "faas.trigger" attribute of the span to "pubsub"
+        .with_trigger(OpenTelemetryFaasTrigger::PubSub),
+    );
     runtime.run().await?;
     Ok(())
 }

lambda-runtime/src/layers/mod.rs

@@ -14,4 +14,4 @@ pub use trace::TracingLayer;
 #[cfg(feature = "opentelemetry")]
 mod otel;
 #[cfg(feature = "opentelemetry")]
-pub use otel::OpenTelemetryLayer;
+pub use otel::{OpenTelemetryFaasTrigger, OpenTelemetryLayer};

lambda-runtime/src/layers/otel.rs

@@ -1,4 +1,4 @@
-use std::{future::Future, pin::Pin, task};
+use std::{fmt::Display, future::Future, pin::Pin, task};
 
 use crate::LambdaInvocation;
 use opentelemetry_semantic_conventions::trace as traceconv;
@@ -10,6 +10,7 @@ use tracing::{instrument::Instrumented, Instrument};
 /// a function to flush OpenTelemetry after the end of the invocation.
 pub struct OpenTelemetryLayer<F> {
     flush_fn: F,
+    otel_attribute_trigger: OpenTelemetryFaasTrigger,
 }
 
 impl<F> OpenTelemetryLayer<F>
@@ -18,7 +19,18 @@ where
 {
     /// Create a new [OpenTelemetryLayer] with the provided flush function.
     pub fn new(flush_fn: F) -> Self {
-        Self { flush_fn }
+        Self {
+            flush_fn,
+            otel_attribute_trigger: Default::default(),
+        }
+    }
+
+    /// Configure the `faas.trigger` attribute of the OpenTelemetry span.
+    pub fn with_trigger(self, trigger: OpenTelemetryFaasTrigger) -> Self {
+        Self {
+            otel_attribute_trigger: trigger,
+            ..self
+        }
     }
 }
 
@@ -33,6 +45,7 @@ where
             inner,
             flush_fn: self.flush_fn.clone(),
             coldstart: true,
+            otel_attribute_trigger: self.otel_attribute_trigger.to_string(),
         }
     }
 }
@@ -42,6 +55,7 @@ pub struct OpenTelemetryService<S, F> {
     inner: S,
     flush_fn: F,
     coldstart: bool,
+    otel_attribute_trigger: String,
 }
 
 impl<S, F> Service<LambdaInvocation> for OpenTelemetryService<S, F>
@@ -61,7 +75,7 @@ where
         let span = tracing::info_span!(
             "Lambda function invocation",
             "otel.name" = req.context.env_config.function_name,
-            { traceconv::FAAS_TRIGGER } = "http",
+            { traceconv::FAAS_TRIGGER } = &self.otel_attribute_trigger,
             { traceconv::FAAS_INVOCATION_ID } = req.context.request_id,
             { traceconv::FAAS_COLDSTART } = self.coldstart
         );
@@ -114,3 +128,33 @@ where
         task::Poll::Ready(ready)
     }
 }
+
+/// Represent the possible values for the OpenTelemetry `faas.trigger` attribute.
+/// See https://opentelemetry.io/docs/specs/semconv/attributes-registry/faas/ for more details.
+#[derive(Default, Clone, Copy)]
+#[non_exhaustive]
+pub enum OpenTelemetryFaasTrigger {
+    /// A response to some data source operation such as a database or filesystem read/write
+    #[default]
+    Datasource,
+    /// To provide an answer to an inbound HTTP request
+    Http,
+    /// A function is set to be executed when messages are sent to a messaging system
+    PubSub,
+    /// A function is scheduled to be executed regularly
+    Timer,
+    /// If none of the others apply
+    Other,
+}
+
+impl Display for OpenTelemetryFaasTrigger {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            OpenTelemetryFaasTrigger::Datasource => write!(f, "datasource"),
+            OpenTelemetryFaasTrigger::Http => write!(f, "http"),
+            OpenTelemetryFaasTrigger::PubSub => write!(f, "pubsub"),
+            OpenTelemetryFaasTrigger::Timer => write!(f, "timer"),
+            OpenTelemetryFaasTrigger::Other => write!(f, "other"),
+        }
+    }
+}