Working example of lambda_http with Opentelemetry and shared client

[guillaumefont]

Hi,

I'm trying to use the new opentelemtry feature with a shared dynamodb client :

use aws_config::BehaviorVersion;

use lambda_http::lambda_runtime::layers::{OpenTelemetryFaasTrigger, OpenTelemetryLayer};
use lambda_http::tower::ServiceBuilder;
use lambda_http::Request;
use lambdas::api::routes::users::get::api_user_list;

use lambdas::utils::telemetry::init_tracer;

#[tokio::main]
async fn main() -> Result<(), lambda_http::Error> {
    let _guard = init_tracer();

    let config = aws_config::load_defaults(BehaviorVersion::latest()).await;
    let db_client = aws_sdk_dynamodb::Client::new(&config);

    // Create a service from the layer and the handler
    let service = ServiceBuilder::new()
        .layer(
            OpenTelemetryLayer::new(|| {
                _guard.tracer_provider.force_flush();
            })
            .with_trigger(OpenTelemetryFaasTrigger::Http),
        )
        .service_fn(|event: Request| async { api_user_list(event, &db_client).await });

    lambda_http::run(service).await?;

    Ok(())
}

And I'm getting the following error :

error[E0277]: the trait bound `layers::otel::OpenTelemetryService<ServiceFn<{closure@packages/lambdas/src/bin/api_user_list.rs:25:21: 25:37}>, {closure@packages/lambdas/src/bin/api_user_list.rs:20:37: 20:39}>: Service<lambda_http::http::Request<lambda_http::Body>>` is not satisfied
   --> packages/lambdas/src/bin/api_user_list.rs:27:5
    |
27  |     lambda_http::run(service).await?;
    |     ^^^^^^^^^^^^^^^^^^^^^^^^^ the trait `Service<lambda_http::http::Request<lambda_http::Body>>` is not implemented for `OpenTelemetryService<ServiceFn<{closure@api_user_list.rs:25:21}>, {closure@api_user_list.rs:20:37}>`
    |
    = help: the trait `Service<LambdaInvocation>` is implemented for `layers::otel::OpenTelemetryService<S, F>`
note: required by a bound in `lambda_http::run`
   --> /Users/guillaume/.cargo/registry/src/index.crates.io-6f17d22bba15001f/lambda_http-0.13.0/src/lib.rs:194:8

I can't find a working example of this setup, is it the right way to do it ?

Thanks

[alessandrobologna]

Hi, just chiming in... I think it may depends on how lambdas::api::routes::users::get::api_user_list is defined. I am not familiar with that crate

Starting from this example something like this should work:

use lambda_runtime::{
    layers::{OpenTelemetryFaasTrigger, OpenTelemetryLayer as OtelLayer},
    LambdaEvent, Runtime,
};
use opentelemetry::trace::TracerProvider;
use opentelemetry_sdk::{runtime, trace};
use tower::{service_fn, BoxError};
use tracing_subscriber::prelude::*;
use aws_sdk_dynamodb::Client as DynamoDbClient;


async fn echo(
    event: LambdaEvent<serde_json::Value>,
    client: DynamoDbClient,
) -> Result<serde_json::Value, &'static str> {
    // use the client here
    // ...
    Ok(event.payload)
}

#[tokio::main]
async fn main() -> Result<(), BoxError> {
    // Set up OpenTelemetry tracer provider that writes spans to stdout for debugging purposes
    let exporter = opentelemetry_stdout::SpanExporter::default();
    let tracer_provider = trace::TracerProvider::builder()
        .with_batch_exporter(exporter, runtime::Tokio)
        .build();

    // Set up link between OpenTelemetry and tracing crate
    tracing_subscriber::registry()
        .with(tracing_opentelemetry::OpenTelemetryLayer::new(
            tracer_provider.tracer("my-app"),
        ))
        .init();

    let config = aws_config::load_from_env().await;
    let dynamodb_client = DynamoDbClient::new(&config);

    let func = service_fn(move |event| {
        let client = dynamodb_client.clone();
        echo(event, client)
    });

    // Initialize the Lambda runtime and add OpenTelemetry tracing
    let runtime = Runtime::new(func).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(())
}

[guillaumefont]

I've managed to make something work (the api_user_list was not the problem) but it feels kind of dirty :

use aws_config::BehaviorVersion;

use lambda_http::lambda_runtime::layers::{OpenTelemetryFaasTrigger, OpenTelemetryLayer};
use lambda_http::{lambda_runtime, service_fn, Adapter, Request};
use lambdas::api::routes::users::get::api_user_list;

use lambdas::utils::telemetry::init_tracer;

#[tokio::main]
async fn main() -> Result<(), lambda_http::Error> {
    let config = aws_config::load_defaults(BehaviorVersion::latest()).await;
    let db_client = aws_sdk_dynamodb::Client::new(&config);

    let _guard = init_tracer();

    let runtime = lambda_runtime::Runtime::new(Adapter::from(service_fn(|event: Request| async {
        api_user_list(event, &db_client).await
    })))
    .layer(
        OpenTelemetryLayer::new(|| {
            _guard.tracer_provider.force_flush();
        })
        .with_trigger(OpenTelemetryFaasTrigger::Http),
    );
    runtime.run().await?;

    Ok(())
}

Thanks for the help !

[maxday]

Thanks! I'll close the issue as it seems that you find a working solution, let us know if you want to re-open it.
An other solution is to extract the closure into a dedicated variable, it might be easier to test, in case you want to swap the implementation during test phase, mocking a DynamoDB client for instance.
Here is a full example: (I will add it in example folder)

use aws_config::BehaviorVersion;
use aws_sdk_dynamodb::Client as DynamoDbClient;
use lambda_http::lambda_runtime::layers::{OpenTelemetryFaasTrigger, OpenTelemetryLayer};
use lambda_http::lambda_runtime::Runtime;
use lambda_http::Adapter;
use lambda_http::{service_fn, Request};
use opentelemetry_sdk::{runtime, trace};

async fn echo(
    _event: Request,
    _client: &DynamoDbClient,
) -> Result<serde_json::Value, &'static str> {
    Ok(serde_json::json!({
        "status": 200,
        "message": "hello world",
    }))
}

#[tokio::main]
async fn main() -> Result<(), lambda_http::Error> {
    let config = aws_config::load_defaults(BehaviorVersion::latest()).await;
    let db_client = aws_sdk_dynamodb::Client::new(&config);

    let exporter = opentelemetry_stdout::SpanExporter::default();
    let tracer_provider = trace::TracerProvider::builder()
        .with_batch_exporter(exporter, runtime::Tokio)
        .build();

    let handler_func_closure = |event: Request| async { echo(event, &db_client).await };

    let runtime = Runtime::new(Adapter::from(service_fn(handler_func_closure))).layer(
        OpenTelemetryLayer::new(|| {
            tracer_provider.force_flush();
        })
        .with_trigger(OpenTelemetryFaasTrigger::Http),
    );

    runtime.run().await?;
    Ok(())
}

[github-actions]

This issue is now closed. Comments on closed issues are hard for our team to see.
If you need more assistance, please either tag a team member or open a new issue that references this one.

[guillaumefont]

Thanks guys, I'll keep this solution for now.

Cheers,